Add torrent objects
- Init torrent module - Init torrent.types - Init torrent.abc - Init torrent.torrents - Add TorrentTypes Enum - Add TorrentObj absctract class - Add TorrentFile, TorrentMagnet and TorrentURL classes
This commit is contained in:
parent
8656f1e68e
commit
ed324557c8
|
@ -1,3 +1,4 @@
|
||||||
|
aiofiles>=24.1.0
|
||||||
aiohappyeyeballs>=2.3.4
|
aiohappyeyeballs>=2.3.4
|
||||||
aiohttp>=3.10.0
|
aiohttp>=3.10.0
|
||||||
aiosignal>=1.3.1
|
aiosignal>=1.3.1
|
||||||
|
@ -5,4 +6,5 @@ attrs>=23.2.0
|
||||||
frozenlist>=1.4.1
|
frozenlist>=1.4.1
|
||||||
idna>=3.7
|
idna>=3.7
|
||||||
multidict>=6.0.5
|
multidict>=6.0.5
|
||||||
|
python-magic>=0.4.27
|
||||||
yarl>=1.9.4
|
yarl>=1.9.4
|
||||||
|
|
8
setup.py
8
setup.py
|
@ -7,6 +7,10 @@ setup(
|
||||||
author="ORUDO",
|
author="ORUDO",
|
||||||
author_email="root@orudo.ru",
|
author_email="root@orudo.ru",
|
||||||
description="A simple Telegram bot that will allow you to upload torrent files / magnet links to a remote Torrent server (qBitTorrent, Transmission, etc.)",
|
description="A simple Telegram bot that will allow you to upload torrent files / magnet links to a remote Torrent server (qBitTorrent, Transmission, etc.)",
|
||||||
install_requires=["aiohttp>=3.10.0"],
|
install_requires=[
|
||||||
packages=["tubot", "tubot.static"],
|
"aiohttp>=3.10.0",
|
||||||
|
"aiofiles>=24.1.0",
|
||||||
|
"aiofiles>=24.1.0"
|
||||||
|
],
|
||||||
|
packages=["tubot", "tubot.static", "tubot.torrent"],
|
||||||
)
|
)
|
||||||
|
|
|
@ -0,0 +1,36 @@
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
######################################################
|
||||||
|
# Abstract methods and interfaces for torrent module #
|
||||||
|
######################################################
|
||||||
|
|
||||||
|
from abc import ABC, abstractmethod
|
||||||
|
from tubot.static.abc import IValidatable
|
||||||
|
from tubot.torrent.types import TorrentTypes
|
||||||
|
|
||||||
|
|
||||||
|
class TorrentObj(IValidatable, ABC):
|
||||||
|
"""
|
||||||
|
Abstract class of torrent object
|
||||||
|
"""
|
||||||
|
|
||||||
|
_ttype: TorrentTypes # Torrent type property
|
||||||
|
dest: str
|
||||||
|
content: str
|
||||||
|
|
||||||
|
@property
|
||||||
|
def torrent_type(self) -> TorrentTypes:
|
||||||
|
"""
|
||||||
|
:return: Torrent type
|
||||||
|
"""
|
||||||
|
if self._ttype is None:
|
||||||
|
raise NotImplementedError("Torrent type not implemented")
|
||||||
|
return self._ttype
|
||||||
|
|
||||||
|
def __init__(self, content: str, destination: str) -> None:
|
||||||
|
"""
|
||||||
|
:param content: Torrent content (link, file path)
|
||||||
|
:param destination: Download directory
|
||||||
|
"""
|
||||||
|
self.content = content
|
||||||
|
self.dest = destination
|
|
@ -0,0 +1,61 @@
|
||||||
|
# -*- 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
|
|
@ -0,0 +1,18 @@
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
|
||||||
|
############################
|
||||||
|
# Types for torrent module #
|
||||||
|
############################
|
||||||
|
|
||||||
|
# Imports
|
||||||
|
from enum import Enum
|
||||||
|
|
||||||
|
|
||||||
|
class TorrentTypes(Enum):
|
||||||
|
"""
|
||||||
|
Types of torrents
|
||||||
|
"""
|
||||||
|
|
||||||
|
File = ".torrent file"
|
||||||
|
Magnet = "torrent magnet link"
|
||||||
|
URL = "http(s) link to .torrent file"
|
Reference in New Issue