TorrentUplouderBot/tubot/torrent/abc.py

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