97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
|
||
|
|
########################################
|
||
|
|
# Torrent Server API's implementations #
|
||
|
|
########################################
|
||
|
|
|
||
|
|
# Imports
|
||
|
|
from http.cookies import SimpleCookie
|
||
|
|
from aiohttp import ClientSession, ClientResponse
|
||
|
|
from tubot.torrent.abc import TorrentAPI
|
||
|
|
from tubot.torrent.torrents import TorrentFile, TorrentMagnet, TorrentURL
|
||
|
|
from tubot.torrent.types import ServerTypes, TorrentListBuilder
|
||
|
|
|
||
|
|
|
||
|
|
class qBitTorrent(TorrentAPI):
|
||
|
|
"""
|
||
|
|
qBitTorrent API implementation
|
||
|
|
"""
|
||
|
|
|
||
|
|
host: str
|
||
|
|
username: str
|
||
|
|
password: str
|
||
|
|
cookie: SimpleCookie | None
|
||
|
|
_atype = ServerTypes.qBitTorrent
|
||
|
|
|
||
|
|
def __init__(self, host: str, username: str, password: str) -> None:
|
||
|
|
"""
|
||
|
|
:param host: qBitTorrent remote server adress
|
||
|
|
:param username: qBitTorrent remote username
|
||
|
|
:param password: qBitTorrent remote password
|
||
|
|
"""
|
||
|
|
super().__init__()
|
||
|
|
self.cookie = None
|
||
|
|
self.host = host
|
||
|
|
self.username = username
|
||
|
|
self.password = password
|
||
|
|
|
||
|
|
async def _get(
|
||
|
|
self, api: str, cookie: SimpleCookie | None = None
|
||
|
|
) -> ClientResponse:
|
||
|
|
"""
|
||
|
|
Send get request to Torrent server
|
||
|
|
|
||
|
|
:param api: API schema
|
||
|
|
:param cookie: Cookies for auth
|
||
|
|
"""
|
||
|
|
async with ClientSession() as session:
|
||
|
|
return await session.get(url=f"{self.host}/{api}", cookies=cookie)
|
||
|
|
|
||
|
|
async def _post(
|
||
|
|
self, api: str, cookie: SimpleCookie | None = None, data: dict | None = None
|
||
|
|
) -> ClientResponse:
|
||
|
|
"""
|
||
|
|
Send post request to Torrent server
|
||
|
|
|
||
|
|
:param api: API schema
|
||
|
|
:param cookie: Cookies for auth
|
||
|
|
:param data: Request data
|
||
|
|
"""
|
||
|
|
async with ClientSession() as session:
|
||
|
|
return await session.get(
|
||
|
|
url=f"{self.host}/{api}", cookies=cookie, data=data
|
||
|
|
)
|
||
|
|
|
||
|
|
async def auth(self) -> bool:
|
||
|
|
"""
|
||
|
|
Generates cookies for auth
|
||
|
|
"""
|
||
|
|
creds = {"username": self.username, "password": self.password}
|
||
|
|
resp = await self._post(api="api/v2/auth/login", data=creds)
|
||
|
|
if resp.status == 200:
|
||
|
|
cookies = resp.cookies
|
||
|
|
resp = await self._get(api="api/v2/app/version", cookie=cookies)
|
||
|
|
if resp.status != 200:
|
||
|
|
raise ValueError("Auth error")
|
||
|
|
self.cookie = cookies
|
||
|
|
return True
|
||
|
|
raise ConnectionError()
|
||
|
|
|
||
|
|
async def upload_file(self, torrent: TorrentFile) -> None:
|
||
|
|
...
|
||
|
|
|
||
|
|
async def upload_magnet(self, torrent: TorrentMagnet) -> None:
|
||
|
|
...
|
||
|
|
|
||
|
|
async def upload_url(self, torrent: TorrentURL) -> None:
|
||
|
|
...
|
||
|
|
|
||
|
|
@property
|
||
|
|
async def torrent_list(self) -> str:
|
||
|
|
lb = TorrentListBuilder()
|
||
|
|
...
|
||
|
|
return str(lb)
|
||
|
|
|
||
|
|
async def __validate__(self) -> bool:
|
||
|
|
...
|