Merge pull request #98 from shinonomeow/shino_aio

refactor: improve download system architecture and reliability

Author: shinonomeow

backend/src/module/core/services/download_service.py

@@ -3,7 +3,7 @@
 from typing import TYPE_CHECKING, Any
 from typing_extensions import override
 
-from module.downloader.download_queue import DownloadController
+from module.downloader import DownloadController
 from module.utils.events import ServiceException
 from .base_services import BaseService
 

backend/src/module/core/services/rss_service.py

@@ -43,13 +43,20 @@ async def execute(self) -> None:
             await self._engine.refresh_all()
 
             if self._eps_complete_enabled:
-                from module.downloader import DownloadQueue
+                from module.downloader import download_queue
                 from module.manager import eps_complete
 
                 # 要等到 download queue 空了后再做这个,不然会重复下载
                 # 等太久了就说明现在挺重的, 就先不 eps 了
-                await asyncio.wait_for(DownloadQueue().queue.join(), timeout=60)
-                await eps_complete()
+                try:
+                    await asyncio.wait_for(
+                        download_queue.join(), timeout=30
+                    )
+                    await eps_complete()
+                except asyncio.TimeoutError:
+                    logger.warning(
+                        "[RSSService] 当前任务过多，跳过Eps Complete处理"
+                    )
 
             logger.debug("[RSSService] RSS刷新完成")
         except Exception as e:

backend/src/module/downloader/__init__.py

@@ -1,9 +1,11 @@
 from .download_client import Client, DownloadClient
-from .download_queue import DownloadController, DownloadQueue
+from .download_queue import  download_queue, DownloadQueue
+from .download_controller import DownloadController
 
 __all__ = [
     "DownloadClient",
     "DownloadQueue",
     "DownloadController",
     "Client",
+    "download_queue",
 ]

backend/src/module/downloader/client/qbittorrent/qbittorrent.py

@@ -53,7 +53,7 @@ def initialize(self) -> None:
         )
 
     @override
-    async def auth(self):
+    async def auth(self)->bool:
         try:
             resp = await self._client.post(
                 url=QB_API_URL["login"],
@@ -98,14 +98,16 @@ async def logout(self):
         return False
 
     @override
-    async def check_host(self):
+    async def check_host(self)->bool:
         try:
+            logger.debug(f"[qbittorrent] Check host: {self.config.host}")
             resp = await self._client.get(url=QB_API_URL["version"], timeout=5)
-            resp.raise_for_status()
-            if resp.status_code == 200:
+            if resp.status_code == 200 or resp.status_code == 403:
+                logger.debug(f"[qbittorrent] Check host success: {self.config.host}")
+                # 检查
                 return True
-            return False
-        except (httpx.ConnectError, httpx.TimeoutException, httpx.ReadTimeout) as e:
+            resp.raise_for_status()
+        except (httpx.RemoteProtocolError,httpx.ConnectError, httpx.TimeoutException, httpx.ReadTimeout) as e:
             logger.error(
                 f"[qbittorrent] Check host error,please check your host {self.config.host}"
             )
@@ -239,23 +241,31 @@ async def add(self, torrent_url, save_path, category) -> str | None:
                     logger.debug(f"[QbDownloader] Got torrent hashes: {torrent_hashes}")
                     logger.debug(f"[QbDownloader] Using hash: {torrent_link}")
                     file = {"torrents": torrent_file}
+                    data.pop("urls", None)  # 移除urls字段
                 else:
+                    # 没有拿到的话就拿 hash
+                    torrent_link = get_hash(torrent_url)
+                    if not torrent_link:
+                        logger.error(
+                            f"[QbDownloader] Failed to get torrent hash from {torrent_url}"
+                        )
+                        return None
                     logger.warning(
                         f"[QbDownloader] Failed to get torrent content from {torrent_url}"
                     )
         else:
             # 如果是 magnet 链接，直接使用
             torrent_link = get_hash(torrent_url)
-            if torrent_link:
-                # 判断是否为32字符的Base32格式（DMHY等站点使用）
-                if len(torrent_link) == 32:
-                    # 转换Base32格式为40字符小写十六进制
-                    hex_hash = base32_to_hex(torrent_link)
-                    if hex_hash:
-                        torrent_link = hex_hash
-                        logger.debug(
-                            f"[QbDownloader] 转换Base32 hash为十六进制: {torrent_link}"
-                        )
+            # if torrent_link:
+            #     # 判断是否为32字符的Base32格式（DMHY等站点使用）
+            #     if len(torrent_link) == 32:
+            #         # 转换Base32格式为40字符小写十六进制
+            #         hex_hash = base32_to_hex(torrent_link)
+            #         if hex_hash:
+            #             torrent_link = hex_hash
+            #             logger.debug(
+            #                 f"[QbDownloader] 转换Base32 hash为十六进制: {torrent_link}"
+            #             )
             logger.debug(f"[QbDownloader] Using magnet link: {torrent_url}")
         try:
             resp = await self._client.post(

backend/src/module/downloader/download_client.py

@@ -10,6 +10,7 @@
 from module.utils import gen_save_path
 
 from .client import AuthorizationError, BaseDownloader
+from .download_queue import  download_queue
 
 logger = logging.getLogger(__name__)
 
@@ -48,7 +49,6 @@ async def wrapper(self, *args, **kwargs):
                     if lock_task.done() and not lock_task.cancelled():
                         self._api_lock.release()
                     raise asyncio.CancelledError("API call cancelled")
-
                 # 成功获取锁
                 break
 
@@ -145,19 +145,19 @@ async def login(self):
             self.login_success_event.clear()  # 重置事件状态
             self.is_authenticating = True  # 设置正在认证状态
             logger.debug("[Downloader Client] attempting authentication")
-
+            if not await self.check_host():
+                logger.error("[Downloader Client] Downloader host check failed")
+                return
+            logger.debug("[Downloader Client] Downloader host check passed")
             if await self.downloader.auth():
                 self.login_success_event.set()  # 设置认证成功事件
-                logger.info("[Downloader Client] authentication success")
-            else:
-                # 保持 login_success_event 为 clear 状态，表示认证失败
-                logger.warning("[Downloader Client] authentication failed")
 
         except Exception as e:
             logger.error(f"[Downloader Client] authentication error: {e}")
             # 保持 login_success_event 为 clear 状态，表示认证失败
         finally:
             self.is_authenticating = False
+            self.login_task = None
 
     async def wait_for_login(self) -> bool:
         """等待认证完成，返回是否可以继续API调用"""
@@ -238,14 +238,18 @@ async def add_torrent(self, torrent: Torrent, bangumi) -> bool:
                 )
         except AuthorizationError:
             self.start_login()
+            #TODO: 重试太多了怎么办?
+            # https://mikanani.me/Home/Episode/4294fd53bcd1bfe2ff3b5796004ee3ccb1ba0d0e  这是个死种
+            download_queue.add(torrent, bangumi)  # 重新放回队列
+            logger.debug( f"[Downloader] Add torrent failed, re-adding to queue: {torrent.name}")
         return False
 
     @api_rate_limit
     async def move_torrent(self, hashes: list[str] | str, location: str) -> bool:
         if not await self.wait_for_login():
             return False  # 登录失败时返回False
-
         try:
+            #TODO: 好像是用 | 连起来就行,但现在好像用不上了
             result = await self.downloader.move(hashes=hashes, new_location=location)
             if result:
                 logger.info(f"[Downloader] Move torrents {hashes} to {location}")
@@ -333,7 +337,7 @@ async def get_torrent_files(self, _hash: str) -> list[str] | None:
         return []
 
     def start_login(self):
-        if not self.is_authenticating:
+        if not self.is_authenticating and self.login_task is None:
             self.is_authenticating = True  # 设置认证状态
             self.login_task = asyncio.create_task(self.login(), name="login")
 
@@ -371,7 +375,8 @@ def get_waiting_api_count(self) -> int:
     async def stop(self):
         logger.info("[Download Client] Stopping download client")
         self.cancel_all_api_calls()  # 先取消所有API调用
-        await self.downloader.logout()
+        if self.login_success_event.is_set():
+            await self.downloader.logout()
         if self.login_task:
             self.login_task.cancel()
 

backend/src/module/downloader/download_controller.py

@@ -0,0 +1,91 @@
+import asyncio
+import logging
+
+from module.database import Database, engine
+from module.downloader.download_client import Client as client
+from module.utils import event_bus
+from module.utils.events import Event, EventBus, EventType
+from module.models import Bangumi, Torrent
+
+from .download_queue import download_queue
+
+logger = logging.getLogger(__name__)
+class DownloadController:
+    def __init__(self):
+        self._event_bus:EventBus = event_bus
+
+    # 10秒拿5个
+    async def download(self):
+        if client.login_success_event.is_set() is False:
+            if client.login_task is None:
+                logger.warning(
+                    "[Download Controller] 正在尝试登录下载客户端，请稍候..."
+                )
+                client.login_task = asyncio.create_task(client.login())
+            return
+        queue_size = download_queue.qsize()
+        if queue_size == 0:
+            return
+
+        tasks = []
+        torrents = []
+        torrent_bangumi_pairs = []
+
+        # 一次取五个torrent
+        batch_size = min(queue_size, 5)
+
+        for i in range(batch_size):
+            torrent, bangumi = download_queue.get_nowait()
+
+            download_queue.task_done()
+            logging.debug(f"[Download Controller] start download {torrent.name}")
+            torrents.append(torrent)
+            torrent_bangumi_pairs.append((torrent, bangumi))
+            tasks.append(client.add_torrent(torrent, bangumi))
+            # 更新种子信息
+
+        # 执行下载任务
+        results = await asyncio.gather(*tasks)
+
+        # 处理下载结果并发布事件
+        # 保存到数据库
+
+        with Database(engine) as database:
+            for torrent in torrents:
+                database.torrent.add(torrent)
+        for i, result in enumerate(results):
+            torrent, bangumi = torrent_bangumi_pairs[i]
+
+            if result is True:  # 下载成功
+                if torrent.download_uid:  # 有下载哈希
+                    # 发布下载开始事件
+                    await self._publish_download_started(torrent, bangumi)
+            elif isinstance(result, Exception):
+                logger.error(
+                    f"[Download Controller] 下载失败: {torrent.name} - {result}"
+                )
+
+    async def _publish_download_started(
+        self, torrent: Torrent, bangumi: Bangumi
+    ) -> None:
+        """发布下载开始事件
+
+        Args:
+            torrent: 种子信息
+            bangumi: 番剧信息
+        """
+        if not self._event_bus:
+            logger.warning("[Download Controller] EventBus 未设置，无法发布事件")
+            return
+
+        try:
+            event = Event(
+                type=EventType.DOWNLOAD_STARTED,
+                data={"torrent": torrent, "bangumi": bangumi},
+            )
+
+            asyncio.create_task(self._event_bus.publish(event))
+            logger.debug(f"[Download Controller] 已发布下载开始事件: {torrent.name}")
+
+        except Exception as e:
+            logger.error(f"[Download Controller] 发布下载开始事件失败: {e}")