fix: backfill transfer download history matching

Fixes #5702

app/chain/transfer.py

@@ -1584,6 +1584,45 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
             )
         ]
 
+    @staticmethod
+    def _resolve_download_history(
+        downloadhis: DownloadHistoryOper,
+        file_path: Path,
+        bluray_dir: bool = False,
+        download_hash: Optional[str] = None,
+    ) -> Optional[DownloadHistory]:
+        """
+        根据显式 hash、文件路径或种子根目录回查下载历史。
+        """
+        if download_hash:
+            return downloadhis.get_by_hash(download_hash)
+
+        if bluray_dir:
+            return downloadhis.get_by_path(file_path.as_posix())
+
+        download_file = downloadhis.get_file_by_fullpath(file_path.as_posix())
+        if download_file:
+            return downloadhis.get_by_hash(download_file.download_hash)
+
+        # 多文件种子里的字幕/附加文件可能没有稳定的 fullpath 记录，
+        # 退回到父目录和 savepath 继续查找，尽量补齐同一种子的关联信息。
+        for parent_path in file_path.parents:
+            parent_posix = parent_path.as_posix()
+            download_history = downloadhis.get_by_path(parent_posix)
+            if download_history:
+                return download_history
+
+            download_files = downloadhis.get_files_by_savepath(parent_posix) or []
+            download_hashes = {
+                download_file.download_hash
+                for download_file in download_files
+                if download_file.download_hash
+            }
+            if len(download_hashes) == 1:
+                return downloadhis.get_by_hash(next(iter(download_hashes)))
+
+        return None
+
     def do_transfer(
         self,
         fileitem: FileItem,
@@ -1725,23 +1764,13 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton):
                         continue
 
                 # 提前获取下载历史，以便获取自定义识别词
-                download_history = None
                 downloadhis = DownloadHistoryOper()
-                if download_hash:
-                    # 先按hash查询
-                    download_history = downloadhis.get_by_hash(download_hash)
-                elif bluray_dir:
-                    # 蓝光原盘，按目录名查询
-                    download_history = downloadhis.get_by_path(file_path.as_posix())
-                else:
-                    # 按文件全路径查询
-                    download_file = downloadhis.get_file_by_fullpath(
-                        file_path.as_posix()
-                    )
-                    if download_file:
-                        download_history = downloadhis.get_by_hash(
-                            download_file.download_hash
-                        )
+                download_history = self._resolve_download_history(
+                    downloadhis=downloadhis,
+                    file_path=file_path,
+                    bluray_dir=bluray_dir,
+                    download_hash=download_hash,
+                )
 
                 if not meta:
                     subscribe_custom_words = None

tests/test_transfer_download_history_lookup.py

@@ -0,0 +1,91 @@
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+from app.chain.transfer import TransferChain
+
+
+class FakeDownloadHistoryOper:
+    def __init__(
+        self,
+        histories_by_hash=None,
+        histories_by_path=None,
+        files_by_fullpath=None,
+        files_by_savepath=None,
+    ):
+        self.histories_by_hash = histories_by_hash or {}
+        self.histories_by_path = histories_by_path or {}
+        self.files_by_fullpath = files_by_fullpath or {}
+        self.files_by_savepath = files_by_savepath or {}
+
+    def get_by_hash(self, download_hash: str):
+        return self.histories_by_hash.get(download_hash)
+
+    def get_by_path(self, path: str):
+        return self.histories_by_path.get(path)
+
+    def get_file_by_fullpath(self, fullpath: str):
+        return self.files_by_fullpath.get(fullpath)
+
+    def get_files_by_savepath(self, savepath: str):
+        return self.files_by_savepath.get(savepath, [])
+
+
+class TransferDownloadHistoryLookupTest(unittest.TestCase):
+    def test_resolve_download_history_falls_back_to_parent_download_path(self):
+        expected = SimpleNamespace(download_hash="hash1", downloader="qb")
+        oper = FakeDownloadHistoryOper(
+            histories_by_hash={"hash1": expected},
+            histories_by_path={"/downloads/season-pack": expected},
+        )
+
+        history = TransferChain._resolve_download_history(
+            downloadhis=oper,
+            file_path=Path("/downloads/season-pack/Test.Show.S01E01.mkv"),
+        )
+
+        self.assertIs(history, expected)
+
+    def test_resolve_download_history_falls_back_to_unique_savepath_hash(self):
+        expected = SimpleNamespace(download_hash="hash1", downloader="qb")
+        oper = FakeDownloadHistoryOper(
+            histories_by_hash={"hash1": expected},
+            files_by_savepath={
+                "/downloads/season-pack": [
+                    SimpleNamespace(download_hash="hash1"),
+                    SimpleNamespace(download_hash="hash1"),
+                ]
+            },
+        )
+
+        history = TransferChain._resolve_download_history(
+            downloadhis=oper,
+            file_path=Path("/downloads/season-pack/subs/Test.Show.S01E01.zh.ass"),
+        )
+
+        self.assertIs(history, expected)
+
+    def test_resolve_download_history_skips_ambiguous_savepath_hashes(self):
+        oper = FakeDownloadHistoryOper(
+            histories_by_hash={
+                "hash1": SimpleNamespace(download_hash="hash1", downloader="qb"),
+                "hash2": SimpleNamespace(download_hash="hash2", downloader="tr"),
+            },
+            files_by_savepath={
+                "/downloads/shared": [
+                    SimpleNamespace(download_hash="hash1"),
+                    SimpleNamespace(download_hash="hash2"),
+                ]
+            },
+        )
+
+        history = TransferChain._resolve_download_history(
+            downloadhis=oper,
+            file_path=Path("/downloads/shared/Test.Show.S01E01.mkv"),
+        )
+
+        self.assertIsNone(history)
+
+
+if __name__ == "__main__":
+    unittest.main()