delete bangumi update

Author: shinonomeow

backend/src/module/core/aiocore.py

@@ -3,7 +3,7 @@
 from contextlib import asynccontextmanager
 from typing import Any
 
-from module.utils.events import EventBus
+from module.utils.events import event_bus
 from .task_manager import TaskManager
 
 logger = logging.getLogger(__name__)
@@ -17,7 +17,7 @@ class AsyncApplicationCore:
 
     def __init__(self):
         self.task_manager = TaskManager()
-        self.event_bus = EventBus()
+        self.event_bus = event_bus
         self.services = []
         self._download_monitor = None
         self._rename_monitor = None

backend/src/module/database/combine.py

@@ -4,11 +4,11 @@
 from module.models.rss import RSSItem
 from module.models.torrent import Torrent
 
-from .bangumi import BangumiDatabase
-from .engine import engine as e
-from .rss import RSSDatabase
-from .torrent import TorrentDatabase
-from .user import UserDatabase
+from module.database.bangumi import BangumiDatabase
+from module.database.engine import engine as e
+from module.database.rss import RSSDatabase
+from module.database.torrent import TorrentDatabase
+from module.database.user import UserDatabase
 
 
 class Database(Session):
@@ -27,6 +27,17 @@ def __init__(self, engine=e):
     def bangumi_to_rss(self, bangumi: Bangumi) -> RSSItem | None:
         return self.rss.search_url(bangumi.rss_link)
 
+    def add_bangumi(self, bangumi: Bangumi):
+        pass
+
+    def find_torrent_by_bangumi(self, bangumi: Bangumi) -> list[Torrent]:
+        """根据 Bangumi 查找相关的 Torrent"""
+        "依据 official_title, seasion, rss_link"
+        return self.torrent.filter_by_bangumi(bangumi.official_title, bangumi.season, bangumi.rss_link)
+
+    def find_bangumi_by_torrent(self, torrent: Torrent) -> Bangumi | None:
+        pass
+
     def create_table(self):
         SQLModel.metadata.create_all(self.engine)
 
@@ -202,3 +213,12 @@ def migrate(self):
 
         self.commit()
         print("数据库迁移完成!")
+
+if __name__ == "__main__":
+    url = "https://mikanani.me/Download/20250531/fb338d0c51c01c2494a9fb1642dd97769416b5c2.torrent"
+    with Database() as db:
+        torrent = db.torrent.search_by_url(url)
+        if torrent:
+            print(type(torrent))
+        else:
+            print("未找到种子")

backend/src/module/database/rss.py

@@ -13,6 +13,7 @@ class RSSDatabase:
     rss 的主码是 rss_link, id
     现在是 rss 的 id 变动时, bangumi 和 torrent 的 rss_id 也要变动
     """
+
     def __init__(self, session: Session):
         self.session = session
 
@@ -110,6 +111,7 @@ def delete_all(self):
         self.session.exec(condition)
         self.session.commit()
 
+
 if __name__ == "__main__":
     from module.database import Database
 

backend/src/module/database/torrent.py

@@ -14,30 +14,14 @@ def __init__(self, session: Session):
         self.session = session
 
     def add(self, data: Torrent):
-        # 找一下是否已经存在相同的种子
-        existing_torrent = self.session.query(Torrent).filter_by(url=data.url).first()
-        # 有的话就更新一下
-        if existing_torrent:
-            data.id = existing_torrent.id
-        ins = self.session.merge(data)
+        self.session.merge(data)
         self.session.commit()
-        self.session.refresh(ins)  # 刷新对象以获取最新的数据库状态
-        data.id = ins.id  # 确保 data.id 被更新为数据库中的 ID
-        logger.debug(f"[TorrentDatabase] {ins.id=}, {ins.name=}, {ins.url=}")
-        logger.debug(f"[TorrentDatabase] Insert {data.name=} {data.id=} in database.")
-        logger.debug(f"Insert {data.name} in database.")
+        logger.debug(f"[TorrentDatabase] Insert {data.name=} {data.url=} in database.")
 
     def add_all(self, datas: list[Torrent]):
-        unique_torrent = [
-            data
-            for data in datas
-            if not self.session.query(Torrent).filter_by(url=data.url).first()
-        ]
-        self.session.add_all(unique_torrent)
+        for data in datas:
+            self.session.merge(data)
         self.session.commit()
-        # 对所有的 Torrent 对象进行刷新
-        for data in unique_torrent:
-            self.session.refresh(data)
         logger.debug(f"Insert {len(datas)} torrents in database.")
 
     def update(self, data: Torrent):
@@ -48,79 +32,92 @@ def update(self, data: Torrent):
         logger.debug(f"[TorrentDatabase] Update {data.name} in database success.")
         logger.debug(f"Update {data.name} in database.")
 
-    def update_all(self, datas: list[Torrent]):
-        self.session.add_all(datas)
-        self.session.commit()
-
-    def update_one_user(self, data: Torrent):
-        self.session.add(data)
-        self.session.commit()
-        self.session.refresh(data)
-        logger.debug(f"Update {data.name} in database.")
-
-    def search(self, _id: int) -> Torrent:
-        return self.session.exec(select(Torrent).where(Torrent.id == _id)).first()
-
-    def search_hash(self, _hash: str) -> Torrent | None:
-        statement = select(Torrent).where(func.instr(Torrent.url, _hash) > 0)
+    # def update_all(self, datas: list[Torrent]):
+    #     self.session.add_all(datas)
+    #     self.session.commit()
+    #
+    # def update_one_user(self, data: Torrent):
+    #     self.session.add(data)
+    #     self.session.commit()
+    #     self.session.refresh(data)
+    #     logger.debug(f"Update {data.name} in database.")
+    #
+    def filter_by_bangumi(self, official_title: str, season: int, rss_link: str) -> list[Torrent]:
+        """根据 Bangumi 的官方标题、季节和 RSS 链接过滤种子"""
+        statement = select(Torrent).where(
+            Torrent.official_title == official_title,
+            Torrent.season == season,
+            Torrent.rss_link == rss_link,
+        )
+        return list(self.session.exec(statement).all())
+
+    def search_by_url(self, url: str) -> Torrent | None:
+        return self.session.get(Torrent, ident=url)
+        # return self.session.exec(select(Torrent).where(Torrent.url == url)).first()
+
+    def search_by_duid(self, duid: str) -> Torrent | None:
+        statement = select(Torrent).where(Torrent.download_guid == duid)
         return self.session.exec(statement).first()
 
-    def search_torrent(self, _hash):
-        # 之前由于 hash 可能不一致, 所以需要搜索 name
-        # 现在会更新种子的 hash,所以只需要搜索 hash 即可
-        if plain_hash := get_hash(_hash):
-            logger.debug(f"[TorrentDatabase] search_torrent {plain_hash}")
-            _hash = plain_hash
-
-        torrent_item = self.search_hash(_hash)
-        logger.debug(f"[TorrentDatabase] search_torrent result {torrent_item}")
-        # if not torrent_item and _name:
-        #     torrent_item = self.search_name(_name)
-        return torrent_item
-
-    # def search_name(self, name: str):
-    #     statement = select(Torrent).where(Torrent.name == name)
-    #     return self.session.exec(statement).first()
-
     def search_bangumi(self, bangumi_id: int):
         return self.session.exec(
             select(Torrent).where(Torrent.bangumi_id == bangumi_id)
         ).all()
 
     def search_all(self) -> list[Torrent]:
-        return self.session.exec(select(Torrent)).all()
+        return list(self.session.exec(select(Torrent)).all())
 
     def search_all_unrenamed(self) -> list[Torrent]:
-        return list(self.session.exec( select(Torrent).where(Torrent.renamed == false())).all())
+        return list(
+            self.session.exec(select(Torrent).where(Torrent.renamed == false())).all()
+        )
 
     def search_all_downloaded(self) -> list[Torrent]:
-        return self.session.exec( select(Torrent).where(Torrent.downloaded == true())).all()
+        torrents = self.session.exec(
+            select(Torrent).where(Torrent.downloaded == true())
+        ).all()
+        return list(torrents)
 
-    def search_rss(self, rss_id: int) -> list[Torrent]:
-        return self.session.exec(select(Torrent).where(Torrent.rss_id == rss_id)).all()
+    # def search_rss(self, rss_url: int) -> list[Torrent]:
+    #     """根据RSS url查询所有种子"""
+    #     torrents = self.session.exec(select(Torrent).where(Torrent.rss_link == rss_url)).all()
+    #     return list(torrents)
 
     def check_new(self, torrents_list: list[Torrent]) -> list[Torrent]:
         new_torrents = []
         for torrent in torrents_list:
-            torrent_item = self.search_torrent(torrent.url)
+            torrent_item = self.search_by_url(torrent.url)
             if not torrent_item or not torrent_item.downloaded:
                 new_torrents.append(torrent)
         return new_torrents
 
-    def delete(self, _id: int) -> bool:
+    def delete_by_url(self, url: str) -> bool:
         # 思考什么时候删除种子
         # 1. 当bangumi已经删除时, 如果删除,会在重命名的时候再次添加
         # 2. 当种子也删除时, 不会再次添加
         # 3. bangumi 删除有几种情况: 1. 有一个全清, 会刷新一次 2. 用户自已删除, 如果是聚合的, 会在下次
         # 刷新时再次添加, 如果是单独的, 会连着rss 一起删除
-        condition = delete(Torrent).where(Torrent.id == _id)
-        try:
-            self.session.exec(condition)
+        stmt = select(Torrent).where(Torrent.url == url)
+        torrent_item = self.session.exec(stmt).first()
+        if torrent_item:
+            self.session.delete(torrent_item)
+            logger.debug(
+                f"[TorrentDatabase] Delete torrent {torrent_item.name} by url: {url}."
+            )
+            self.session.commit()
+            return True
+
+    def delete_by_duid(self, duid: str) -> bool:
+        stmt = select(Torrent).where(Torrent.download_guid == duid)
+        torrent_item = self.session.exec(stmt).first()
+        if torrent_item:
+            self.session.delete(torrent_item)
+            logger.debug(
+                f"[TorrentDatabase] Delete torrent {torrent_item.name} by duid: {duid}."
+            )
             self.session.commit()
             return True
-        except Exception as e:
-            logger.error(f"Delete RSS Item failed. Because: {e}")
-            return False
+        return False
 
 
 if __name__ == "__main__":
@@ -130,8 +127,9 @@ def delete(self, _id: int) -> bool:
     name = "[ANi] 物语系列 第外季＆第怪季 - 06 [1080P][Baha][WEB-DL][AAC AVC][CHT].mp4"
     hash = "1ae27b047005e097b74b66e27c37610aa5a0f5a2"
     with Database() as db:
-        t_name = db.torrent.search_name(name)
-        t_hash = db.torrent.search_hash(hash)
-        db.torrent.delete(t_hash.id)
+        # t_name = db.torrent.search_name(name)
+        t_hash = db.torrent.search_by_duid(hash)
+        if t_hash:
+            db.torrent.delete_by_url(t_hash.url)
         # print(f"{t_name=}")
         # print(f"{t_hash=}")

backend/src/module/downloader/download_monitor.py

@@ -114,8 +114,7 @@ async def monitor_torrent(
                     # 从数据库删除对应的torrent记录
                     try:
                         with Database() as db:
-                            if torrent_item := db.torrent.search_hash(torrent_hash):
-                                db.torrent.delete(torrent_item.id)
+                                db.torrent.delete_by_url(torrent.url)
                                 logger.info(f"[DownloadMonitor] 已从数据库删除种子记录: {torrent.name}")
                     except Exception as e:
                         logger.error(f"[DownloadMonitor] 删除数据库记录失败: {e}")
@@ -126,7 +125,7 @@ async def monitor_torrent(
                     logger.debug(f"[DownloadMonitor] 种子 {torrent.name} 下载状态: 未下载")
                     try:
                         with Database() as db:
-                            if torrent_item := db.torrent.search_hash(torrent_hash):
+                            if torrent_item := db.torrent.search_by_duid(torrent_hash):
                                 if not torrent_item.downloaded:
                                     torrent_item.downloaded = True
                                     db.torrent.add(torrent_item)

backend/src/module/downloader/download_queue.py

@@ -59,7 +59,7 @@ async def download(self):
             for torrent in torrents:
                 database.torrent.add(torrent)
                 logger.debug(
-                    f"[Download Controller] Torrent {torrent.name} {torrent.id=} 已保存到数据库"
+                    f"[Download Controller] Torrent {torrent.name} {torrent.url=} 已保存到数据库"
                 )
         for i, result in enumerate(results):
             torrent, bangumi = torrent_bangumi_pairs[i]