62 lines
1.4 KiB
Python
62 lines
1.4 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
|
||
|
|
#################################
|
||
|
|
# Torrent types implementations #
|
||
|
|
#################################
|
||
|
|
|
||
|
|
# Imports
|
||
|
|
from tubot.torrent.types import TorrentTypes
|
||
|
|
from tubot.torrent.abc import TorrentObj
|
||
|
|
from aiofiles import open, ospath
|
||
|
|
from magic import Magic
|
||
|
|
from re import match
|
||
|
|
from urllib.parse import urlparse
|
||
|
|
|
||
|
|
|
||
|
|
class TorrentFile(TorrentObj):
|
||
|
|
"""
|
||
|
|
.torrent file
|
||
|
|
"""
|
||
|
|
|
||
|
|
_ttype = TorrentTypes.File
|
||
|
|
|
||
|
|
async def __validate__(self) -> bool:
|
||
|
|
if await ospath.isfile(self.content):
|
||
|
|
mime = Magic(mime=True).from_file(self.content)
|
||
|
|
if mime == "application/x-bittorrent":
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
async def getbytes(self) -> bytes:
|
||
|
|
async with open(self.content, "rb") as dottorrent:
|
||
|
|
return await dottorrent.read()
|
||
|
|
|
||
|
|
|
||
|
|
class TorrentMagnet(TorrentObj):
|
||
|
|
"""
|
||
|
|
Torrent magnet link
|
||
|
|
"""
|
||
|
|
|
||
|
|
_ttype = TorrentTypes.Magnet
|
||
|
|
|
||
|
|
async def __validate__(self) -> bool:
|
||
|
|
pattern = r"^magnet:\?xt=urn:btih:[a-fA-F0-9]{40}.*$"
|
||
|
|
if match(pattern, self.content):
|
||
|
|
return True
|
||
|
|
return False
|
||
|
|
|
||
|
|
|
||
|
|
class TorrentURL(TorrentObj):
|
||
|
|
"""
|
||
|
|
Http(s) link to .torrent file
|
||
|
|
"""
|
||
|
|
|
||
|
|
_ttype = TorrentTypes.URL
|
||
|
|
|
||
|
|
async def __validate__(self) -> bool:
|
||
|
|
try:
|
||
|
|
parse = urlparse(self.content)
|
||
|
|
return all([parse.scheme, parse.netloc])
|
||
|
|
except (TypeError, AttributeError):
|
||
|
|
return False
|