Refactor: Migrating backend from synchronous functions to asynchronous functions

backend/requirements-dev.txt

@@ -2,4 +2,5 @@
 ruff
 black
 pre-commit
-pytest
+pytest
+pytest-asyncio

backend/requirements.txt

@@ -7,9 +7,7 @@ fastapi==0.97.0
 h11==0.14.0
 idna==3.4
 pydantic~=1.10
-PySocks==1.7.1
-qbittorrent-api==2023.9.53
-requests==2.31.0
+httpx[http2,socks]==0.25.0
 six==1.16.0
 sniffio==1.3.0
 soupsieve==2.4.1

backend/src/main.py

@@ -57,7 +57,9 @@ def html(request: Request, path: str):
         else:
             context = {"request": request}
             return templates.TemplateResponse("index.html", context)
+
 else:
+
     @app.get("/", status_code=302, tags=["html"])
     def index():
         return RedirectResponse("/docs")

backend/src/module/api/bangumi.py

@@ -116,6 +116,7 @@ async def refresh_poster():
         resp = manager.refresh_poster()
     return u_response(resp)
 
+
 @router.get(
     path="/refresh/poster/{bangumi_id}",
     response_model=APIResponse,

backend/src/module/api/rss.py

@@ -4,28 +4,30 @@
 from module.downloader import DownloadClient
 from module.manager import SeasonCollector
 from module.models import APIResponse, Bangumi, RSSItem, RSSUpdate, Torrent
-from module.rss import RSSAnalyser, RSSEngine
+from module.rss import RSSAnalyser, RSSEngine, RSSManager
 from module.security.api import UNAUTHORIZED, get_current_user
 
 from .response import u_response
 
 router = APIRouter(prefix="/rss", tags=["rss"])
+engine = RSSEngine()
+analyser = RSSAnalyser()
 
 
 @router.get(
     path="", response_model=list[RSSItem], dependencies=[Depends(get_current_user)]
 )
 async def get_rss():
-    with RSSEngine() as engine:
-        return engine.rss.search_all()
+    with RSSManager() as manager:
+        return manager.rss.search_all()
 
 
 @router.post(
     path="/add", response_model=APIResponse, dependencies=[Depends(get_current_user)]
 )
 async def add_rss(rss: RSSItem):
-    with RSSEngine() as engine:
-        result = engine.add_rss(rss.url, rss.name, rss.aggregate, rss.parser)
+    with RSSManager() as manager:
+        result = await manager.add_rss(rss.url, rss.name, rss.aggregate, rss.parser)
     return u_response(result)
 
 
@@ -37,8 +39,8 @@ async def add_rss(rss: RSSItem):
 async def enable_many_rss(
     rss_ids: list[int],
 ):
-    with RSSEngine() as engine:
-        result = engine.enable_list(rss_ids)
+    with RSSManager() as manager:
+        result = manager.enable_list(rss_ids)
     return u_response(result)
 
 
@@ -48,8 +50,8 @@ async def enable_many_rss(
     dependencies=[Depends(get_current_user)],
 )
 async def delete_rss(rss_id: int):
-    with RSSEngine() as engine:
-        if engine.rss.delete(rss_id):
+    with RSSManager() as manager:
+        if manager.rss.delete(rss_id):
             return JSONResponse(
                 status_code=200,
                 content={"msg_en": "Delete RSS successfully.", "msg_zh": "删除 RSS 成功。"},
@@ -69,8 +71,8 @@ async def delete_rss(rss_id: int):
 async def delete_many_rss(
     rss_ids: list[int],
 ):
-    with RSSEngine() as engine:
-        result = engine.delete_list(rss_ids)
+    with RSSManager() as manager:
+        result = manager.delete_list(rss_ids)
     return u_response(result)
 
 
@@ -80,8 +82,8 @@ async def delete_many_rss(
     dependencies=[Depends(get_current_user)],
 )
 async def disable_rss(rss_id: int):
-    with RSSEngine() as engine:
-        if engine.rss.disable(rss_id):
+    with RSSManager() as manager:
+        if manager.rss.disable(rss_id):
             return JSONResponse(
                 status_code=200,
                 content={"msg_en": "Disable RSS successfully.", "msg_zh": "禁用 RSS 成功。"},
@@ -99,8 +101,8 @@ async def disable_rss(rss_id: int):
     dependencies=[Depends(get_current_user)],
 )
 async def disable_many_rss(rss_ids: list[int]):
-    with RSSEngine() as engine:
-        result = engine.disable_list(rss_ids)
+    with RSSManager() as manager:
+        result = manager.disable_list(rss_ids)
     return u_response(result)
 
 
@@ -114,8 +116,8 @@ async def update_rss(
 ):
     if not current_user:
         raise UNAUTHORIZED
-    with RSSEngine() as engine:
-        if engine.rss.update(rss_id, data):
+    with RSSManager() as manager:
+        if manager.rss.update(rss_id, data):
             return JSONResponse(
                 status_code=200,
                 content={"msg_en": "Update RSS successfully.", "msg_zh": "更新 RSS 成功。"},
@@ -133,8 +135,8 @@ async def update_rss(
     dependencies=[Depends(get_current_user)],
 )
 async def refresh_all():
-    with RSSEngine() as engine, DownloadClient() as client:
-        engine.refresh_rss(client)
+    async with DownloadClient() as client:
+        await engine.refresh_rss(client)
         return JSONResponse(
             status_code=200,
             content={"msg_en": "Refresh all RSS successfully.", "msg_zh": "刷新 RSS 成功。"},
@@ -147,8 +149,8 @@ async def refresh_all():
     dependencies=[Depends(get_current_user)],
 )
 async def refresh_rss(rss_id: int):
-    with RSSEngine() as engine, DownloadClient() as client:
-        engine.refresh_rss(client, rss_id)
+    async with DownloadClient() as client:
+        await engine.refresh_rss(client=client, rss_id=rss_id)
         return JSONResponse(
             status_code=200,
             content={"msg_en": "Refresh RSS successfully.", "msg_zh": "刷新 RSS 成功。"},
@@ -163,12 +165,8 @@ async def refresh_rss(rss_id: int):
 async def get_torrent(
     rss_id: int,
 ):
-    with RSSEngine() as engine:
-        return engine.get_rss_torrents(rss_id)
-
-
-# Old API
-analyser = RSSAnalyser()
+    with RSSManager() as manager:
+        return manager.get_rss_torrents(rss_id)
 
 
 @router.post(

backend/src/module/core/aiocore.py

@@ -0,0 +1,64 @@
+import asyncio
+
+from module.downloader import DownloadClient
+from module.manager import Renamer
+from module.conf import settings
+from module.rss import RSSEngine
+from module.database import Database
+from module.models import Bangumi, RSSItem, Torrent
+
+
+
+rss_item_pool = []
+torrent_pool: list[tuple[Bangumi, list[Torrent]]] = []
+
+
+class AsyncProgram:
+    def __init__(self):
+        self.renamer = Renamer()
+        self.engine = RSSEngine()
+        self.event = asyncio.Event()
+
+    async def run(self):
+        self.event.clear()
+        task = []
+        if settings.bangumi_manage.enable:
+            task.append(self.rename_task())
+        if settings.rss_parser.enable:
+            task.append(self.rss_task())
+        await asyncio.gather(*task)
+
+    async def rename_task(self):
+        while not self.event.is_set():
+            async with DownloadClient() as client:
+                await self.check_downloader(client)
+                await self.renamer.rename(client)
+                await asyncio.sleep(settings.program.rename_time)
+
+    async def rss_task(self):
+        while not self.event.is_set():
+            await self.engine.rss_poller(process_rss)
+            await asyncio.sleep(settings.program.rss_time)
+
+
+async def rename_task():
+    connected = False
+    renamer = Renamer()
+    async with DownloadClient() as client:
+        while not connected:
+            connected = await client.auth()
+            if not connected:
+                await asyncio.sleep(30)
+        for bangumi, torrents in torrent_pool:
+            client.add_torrent(torrents, bangumi)
+        renamer.rename(client)
+        await asyncio.sleep(settings.program.rename_time)
+
+
+async def rss_task():
+    # GET RSS FROM DATABASE
+    with Database() as db:
+        rss_items = db.rss.search_active()
+        for rss_item in rss_items:
+            rss_item_pool.append(rss_item)
+    pass

backend/src/module/core/program.py

@@ -2,7 +2,13 @@
 
 from module.conf import VERSION, settings
 from module.models import ResponseModel
-from module.update import data_migration, first_run, from_30_to_31, start_up, cache_image
+from module.update import (
+    data_migration,
+    first_run,
+    from_30_to_31,
+    start_up,
+    cache_image,
+)
 
 from .sub_thread import RenameThread, RSSThread
 

backend/src/module/core/sub_thread.py

@@ -1,5 +1,6 @@
 import threading
 import time
+import asyncio
 
 from module.conf import settings
 from module.downloader import DownloadClient
@@ -13,38 +14,11 @@
 class RSSThread(ProgramStatus):
     def __init__(self):
         super().__init__()
-        self._rss_thread = threading.Thread(
-            target=self.rss_loop,
-        )
+        self._rss_loop = asyncio.new_event_loop()
         self.analyser = RSSAnalyser()
 
-    def rss_loop(self):
-        while not self.stop_event.is_set():
-            with DownloadClient() as client, RSSEngine() as engine:
-                # Analyse RSS
-                rss_list = engine.rss.search_aggregate()
-                for rss in rss_list:
-                    self.analyser.rss_to_data(rss, engine)
-                # Run RSS Engine
-                engine.refresh_rss(client)
-            if settings.bangumi_manage.eps_complete:
-                eps_complete()
-            self.stop_event.wait(settings.program.rss_time)
-
-    def rss_start(self):
-        self.rss_thread.start()
-
-    def rss_stop(self):
-        if self._rss_thread.is_alive():
-            self._rss_thread.join()
-
-    @property
-    def rss_thread(self):
-        if not self._rss_thread.is_alive():
-            self._rss_thread = threading.Thread(
-                target=self.rss_loop,
-            )
-        return self._rss_thread
+    async def rss_loop(self):
+        pass
 
 
 class RenameThread(ProgramStatus):

backend/src/module/database/bangumi.py

@@ -129,10 +129,11 @@ def match_list(self, torrent_list: list, rss_link: str) -> list:
                 i += 1
         return torrent_list
 
-    def match_torrent(self, torrent_name: str) -> Optional[Bangumi]:
+    def match_torrent(self, torrent_name: str, rss_link: str) -> Optional[Bangumi]:
         statement = select(Bangumi).where(
             and_(
                 func.instr(torrent_name, Bangumi.title_raw) > 0,
+                func.instr(Bangumi.rss_link, rss_link),
                 # use `false()` to avoid E712 checking
                 # see: https://docs.astral.sh/ruff/rules/true-false-comparison/
                 Bangumi.deleted == false(),