From 0b6314e722832e73690b3f4e1966cc2b42faf5e0 Mon Sep 17 00:00:00 2001 From: EstrellaXD Date: Tue, 10 Oct 2023 19:31:00 +0800 Subject: [PATCH] refactor: trans qb client to aio. --- .../module/downloader/client/qb_downloader.py | 261 +++++++++--------- .../src/module/downloader/download_client.py | 156 +++-------- 2 files changed, 164 insertions(+), 253 deletions(-) diff --git a/backend/src/module/downloader/client/qb_downloader.py b/backend/src/module/downloader/client/qb_downloader.py index fe6805f5..94fb9d48 100644 --- a/backend/src/module/downloader/client/qb_downloader.py +++ b/backend/src/module/downloader/client/qb_downloader.py @@ -1,151 +1,148 @@ import logging -import time - -from qbittorrentapi import Client, LoginFailed -from qbittorrentapi.exceptions import ( - APIConnectionError, - Conflict409Error, - Forbidden403Error, -) - -from module.ab_decorator import qb_connect_failed_wait +import httpx logger = logging.getLogger(__name__) class QbDownloader: def __init__(self, host: str, username: str, password: str, ssl: bool): - self._client: Client = Client( - host=host, - username=username, - password=password, - VERIFY_WEBUI_CERTIFICATE=ssl, - DISABLE_LOGGING_DEBUG_OUTPUT=True, - REQUESTS_ARGS={"timeout": (3.1, 10)}, - ) - self.host = host + self.host = host if "://" in host else "http://" + host self.username = username + self.password = password + self.ssl = ssl - def auth(self, retry=3): - times = 0 - while times < retry: - try: - self._client.auth_log_in() - return True - except LoginFailed: - logger.error( - f"Can't login qBittorrent Server {self.host} by {self.username}, retry in {5} seconds." - ) - time.sleep(5) - times += 1 - except Forbidden403Error: - logger.error("Login refused by qBittorrent Server") - logger.info("Please release the IP in qBittorrent Server") - break - except APIConnectionError: - logger.error("Cannot connect to qBittorrent Server") - logger.info("Please check the IP and port in WebUI settings") - time.sleep(10) - times += 1 - except Exception as e: - logger.error(f"Unknown error: {e}") - break - return False - - def logout(self): - self._client.auth_log_out() - - def check_host(self): - try: - self._client.app_version() - return True - except APIConnectionError: - return False - - def check_rss(self, rss_link: str): - pass - - @qb_connect_failed_wait - def prefs_init(self, prefs): - return self._client.app_set_preferences(prefs=prefs) - - @qb_connect_failed_wait - def get_app_prefs(self): - return self._client.app_preferences() - - def add_category(self, category): - return self._client.torrents_createCategory(name=category) - - @qb_connect_failed_wait - def torrents_info(self, status_filter, category, tag=None): - return self._client.torrents_info( - status_filter=status_filter, category=category, tag=tag + async def auth(self): + resp = await self._client.post( + url="/api/v2/auth/login", + data={"username": self.username, "password": self.password}, + timeout=5, ) + return resp.text - def add_torrents(self, torrent_urls, torrent_files, save_path, category): - resp = self._client.torrents_add( - is_paused=False, - urls=torrent_urls, - torrent_files=torrent_files, - save_path=save_path, - category=category, - use_auto_torrent_management=False, - ) - return resp == "Ok." + async def logout(self): + logout_api = "/api/v2/auth/logout" + await self._client.post(url=logout_api, timeout=5) - def torrents_delete(self, hash): - return self._client.torrents_delete(delete_files=True, torrent_hashes=hash) - - def torrents_rename_file(self, torrent_hash, old_path, new_path) -> bool: + async def check_host(self): try: - self._client.torrents_rename_file( - torrent_hash=torrent_hash, old_path=old_path, new_path=new_path + await self._client.get( + url="/api/v2/app/version", + timeout=5 ) return True - except Conflict409Error: - logger.debug(f"Conflict409Error: {old_path} >> {new_path}") + except httpx.RequestError: return False - def rss_add_feed(self, url, item_path): + async def prefs_init(self, prefs): + prefs_api = "/api/v2/app/setPreferences" + await self._client.post(url=prefs_api, data=prefs) + + async def add_category(self, category): + await self._client.post( + url="/api/v2/torrents/createCategory", + data={"category": category}, + timeout=5, + ) + + async def torrents_info(self, status_filter, category, tag=None): + data = { + "filter": status_filter, + "category": category, + "tag": tag, + } + torrent_info = await self._client.get( + url="/api/v2/torrents/info", + params=data, + ) + return torrent_info.json() + + async def add(self, torrent_urls, torrent_files, save_path, category): + data = { + "urls": torrent_urls, + "torrent_files": torrent_files, + "save_path": save_path, + "category": category, + "is_paused": False, + "use_auto_torrent_management": False, + } + resp = await self._client.post( + url="/api/v2/torrents/add", + data=data, + ) + return resp.status_code == 200 + + async def delete(self, _hash): + data = { + "hashes": _hash, + "deleteFiles": True, + } + resp = await self._client.post( + url="/api/v2/torrents/delete", + data=data, + ) + return resp.status_code == 200 + + async def rename(self, torrent_hash, old_path, new_path) -> bool: + data = { + "hash": torrent_hash, + "oldPath": old_path, + "newPath": new_path, + } + resp = await self._client.post( + url="/api/v2/torrents/renameFile", + data=data, + ) + return resp.status_code == 200 + + async def move(self, hashes, new_location): + data = { + "hashes": hashes, + "location": new_location, + } + resp = await self._client.post( + url="/api/v2/torrents/setLocation", + data=data, + ) + return resp.status_code == 200 + + async def set_category(self, _hash, category): + data = { + "category": category, + "hashes": _hash, + } + resp = await self._client.post( + url="/api/v2/torrents/setCategory", + data=data, + ) + return resp.status_code == 200 + + async def add_tag(self, _hash, tag): + data = { + "hashes": _hash, + "tags": tag, + } + resp = await self._client.post( + url="/api/v2/torrents/addTags", + data=data, + ) + return resp.status_code == 200 + + async def __aenter__(self): + self._client = httpx.AsyncClient( + base_url=self.host, + ) try: - self._client.rss_add_feed(url, item_path) - except Conflict409Error: - logger.warning(f"[Downloader] RSS feed {url} already exists") + authed = await self.auth() + if not authed == "Ok.": + logger.error("[Downloader] Failed authing to qbittorrent.") + logger.warning("[Downloader] Please check username/password in settings.") + raise RuntimeError(authed) + return self + except httpx.ReadTimeout: + logger.error("[Downloader] Failed connecting to qbittorrent.") + logger.warning("[Downloader] Please check host in settings.") + raise RuntimeError("Failed connecting to qbittorrent.") - def rss_remove_item(self, item_path): - try: - self._client.rss_remove_item(item_path) - except Conflict409Error: - logger.warning(f"[Downloader] RSS item {item_path} does not exist") - - def rss_get_feeds(self): - return self._client.rss_items() - - def rss_set_rule(self, rule_name, rule_def): - self._client.rss_set_rule(rule_name, rule_def) - - def move_torrent(self, hashes, new_location): - self._client.torrents_set_location(new_location, hashes) - - def get_download_rule(self): - return self._client.rss_rules() - - def get_torrent_path(self, _hash): - return self._client.torrents_info(hashes=_hash)[0].save_path - - def set_category(self, _hash, category): - try: - self._client.torrents_set_category(category, hashes=_hash) - except Conflict409Error: - logger.warning(f"[Downloader] Category {category} does not exist") - self.add_category(category) - self._client.torrents_set_category(category, hashes=_hash) - - def check_connection(self): - return self._client.app_version() - - def remove_rule(self, rule_name): - self._client.rss_remove_rule(rule_name) - - def add_tag(self, _hash, tag): - self._client.torrents_add_tags(tags=tag, hashes=_hash) + async def __aexit__(self, exc_type, exc_val, exc_tb): + await self.logout() + await self._client.aclose() diff --git a/backend/src/module/downloader/download_client.py b/backend/src/module/downloader/download_client.py index d01d4fa3..52650396 100644 --- a/backend/src/module/downloader/download_client.py +++ b/backend/src/module/downloader/download_client.py @@ -9,112 +9,44 @@ from .path import TorrentPath logger = logging.getLogger(__name__) -class DownloadClient(TorrentPath): +def getClient(): + # TODO 多下载器支持 + type = settings.downloader.type + if type == "qbittorrent": + from .client.qb_downloader import QbDownloader + + return QbDownloader + else: + logger.error(f"[Downloader] Unsupported downloader type: {type}") + raise Exception(f"Unsupported downloader type: {type}") + + +class DownloadClient(getClient(), TorrentPath): def __init__(self): - super().__init__() - self.client = self.__getClient() - self.authed = False - - @staticmethod - def __getClient(): - # TODO 多下载器支持 - type = settings.downloader.type - host = settings.downloader.host - username = settings.downloader.username - password = settings.downloader.password - ssl = settings.downloader.ssl - if type == "qbittorrent": - from .client.qb_downloader import QbDownloader - - return QbDownloader(host, username, password, ssl) - else: - logger.error(f"[Downloader] Unsupported downloader type: {type}") - raise Exception(f"Unsupported downloader type: {type}") - - def __enter__(self): - if not self.authed: - self.auth() - else: - logger.error("[Downloader] Already authed.") - return self - - def __exit__(self, exc_type, exc_val, exc_tb): - if self.authed: - self.client.logout() - self.authed = False - - def auth(self): - self.authed = self.client.auth() - if self.authed: - logger.debug("[Downloader] Authed.") - else: - logger.error("[Downloader] Auth failed.") - - def check_host(self): - return self.client.check_host() - - def init_downloader(self): - prefs = { - "rss_auto_downloading_enabled": True, - "rss_max_articles_per_feed": 500, - "rss_processing_enabled": True, - "rss_refresh_interval": 30, - } - self.client.prefs_init(prefs=prefs) - try: - self.client.add_category("BangumiCollection") - except Exception: - logger.debug("[Downloader] Cannot add new category, maybe already exists.") - if settings.downloader.path == "": - prefs = self.client.get_app_prefs() - settings.downloader.path = self._join_path(prefs["save_path"], "Bangumi") - - def set_rule(self, data: Bangumi): - data.rule_name = self._rule_name(data) - data.save_path = self._gen_save_path(data) - rule = { - "enable": True, - "mustContain": data.title_raw, - "mustNotContain": "|".join(data.filter), - "useRegex": True, - "episodeFilter": "", - "smartFilter": False, - "previouslyMatchedEpisodes": [], - "affectedFeeds": data.rss_link, - "ignoreDays": 0, - "lastMatch": "", - "addPaused": False, - "assignedCategory": "Bangumi", - "savePath": data.save_path, - } - self.client.rss_set_rule(rule_name=data.rule_name, rule_def=rule) - data.added = True - logger.info( - f"[Downloader] Add {data.official_title} Season {data.season} to auto download rules." + super().__init__( + host=settings.downloader.host, + username=settings.downloader.username, + password=settings.downloader.password, + ssl=settings.downloader.ssl ) - def set_rules(self, bangumi_info: list[Bangumi]): - logger.debug("[Downloader] Start adding rules.") - for info in bangumi_info: - self.set_rule(info) - logger.debug("[Downloader] Finished.") - - def get_torrent_info(self, category="Bangumi", status_filter="completed", tag=None): - return self.client.torrents_info( + async def get_torrent_info(self, category="Bangumi", status_filter="completed", tag=None): + return await self.torrents_info( status_filter=status_filter, category=category, tag=tag ) - def rename_torrent_file(self, _hash, old_path, new_path) -> bool: + async def rename_torrent_file(self, _hash, old_path, new_path) -> bool: logger.info(f"{old_path} >> {new_path}") - return self.client.torrents_rename_file( + return await self.rename( torrent_hash=_hash, old_path=old_path, new_path=new_path ) - def delete_torrent(self, hashes): - self.client.torrents_delete(hashes) + async def delete_torrent(self, hashes): + resp = await self.delete(hashes) logger.info("[Downloader] Remove torrents.") + return resp - def add_torrent(self, torrent: Torrent | list, bangumi: Bangumi) -> bool: + async def add_torrent(self, torrent: Torrent | list, bangumi: Bangumi) -> bool: if not bangumi.save_path: bangumi.save_path = self._gen_save_path(bangumi) with RequestContent() as req: @@ -126,49 +58,31 @@ class DownloadClient(TorrentPath): torrent_url = [t.url for t in torrent] torrent_file = None else: - torrent_file = [req.get_content(t.url) for t in torrent] + torrent_file = [await req.get_content(t.url) for t in torrent] torrent_url = None else: if "magnet" in torrent.url: torrent_url = torrent.url torrent_file = None else: - torrent_file = req.get_content(torrent.url) + torrent_file = await req.get_content(torrent.url) torrent_url = None - if self.client.add_torrents( + result = await self.add( torrent_urls=torrent_url, torrent_files=torrent_file, save_path=bangumi.save_path, category="Bangumi", - ): + ) + if result: logger.debug(f"[Downloader] Add torrent: {bangumi.official_title}") return True else: logger.debug(f"[Downloader] Torrent added before: {bangumi.official_title}") return False - def move_torrent(self, hashes, location): - self.client.move_torrent(hashes=hashes, new_location=location) + async def move_torrent(self, hashes, location): + await self.move(hashes=hashes, new_location=location) - # RSS Parts - def add_rss_feed(self, rss_link, item_path="Mikan_RSS"): - self.client.rss_add_feed(url=rss_link, item_path=item_path) + async def set_category(self, hashes, category): + await self.set_category(hashes, category) - def remove_rss_feed(self, item_path): - self.client.rss_remove_item(item_path=item_path) - - def get_rss_feed(self): - return self.client.rss_get_feeds() - - def get_download_rules(self): - return self.client.get_download_rule() - - def get_torrent_path(self, hashes): - return self.client.get_torrent_path(hashes) - - def set_category(self, hashes, category): - self.client.set_category(hashes, category) - - def remove_rule(self, rule_name): - self.client.remove_rule(rule_name) - logger.info(f"[Downloader] Delete rule: {rule_name}")