forked from OrudoCA/qBitDownload-Bot
83 lines
2.2 KiB
Python
83 lines
2.2 KiB
Python
# -*- 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, ServerTypes
|
|
|
|
|
|
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
|
|
|
|
|
|
class TorrentAPI(IValidatable, ABC):
|
|
"""
|
|
Abstract class of torrent-server API's
|
|
"""
|
|
|
|
_atype: ServerTypes # Server type propery
|
|
|
|
def __init__(self) -> None:
|
|
if self._atype is None:
|
|
raise NotImplementedError("Torrent Server type not implemented")
|
|
|
|
async def upload(self, torrent: TorrentObj) -> None:
|
|
"""
|
|
Adds the torrent to a queue on the server
|
|
|
|
:param torrent: TorrenObject type (file, magnet, etc.)
|
|
"""
|
|
match torrent.torrent_type:
|
|
case TorrentTypes.File:
|
|
await self.upload_file(torrent)
|
|
case TorrentTypes.Magnet:
|
|
await self.upload_magnet(torrent)
|
|
case TorrentTypes.URL:
|
|
await self.upload_url(torrent)
|
|
|
|
@abstractmethod
|
|
async def upload_file(self, torrent) -> None:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def upload_magnet(self, torrent) -> None:
|
|
raise NotImplementedError
|
|
|
|
@abstractmethod
|
|
async def upload_url(self, torrent) -> None:
|
|
raise NotImplementedError
|
|
|
|
@property
|
|
@abstractmethod
|
|
async def torrent_list(self) -> str:
|
|
"""
|
|
Returns PlainString with current torrent queue
|
|
"""
|
|
raise NotImplementedError
|