fix(transfer): preserve episode-mapped download history associations

app/chain/download/history.py

@@ -113,9 +113,11 @@ class DownloadHistoryOwner(_DownloadOwnerBase):
             custom_words=custom_words,
         )
         files_to_add: list[DownloadFileWrite] = []
+        custom_word_list = custom_words.splitlines() if custom_words else []
         for file in file_list:
             if episodes:
-                file_meta = MetaInfo(Path(file).stem)
+                # 与订阅选集使用相同识别词，避免偏移前的文件集数被目标集数过滤。
+                file_meta = MetaInfo(Path(file).stem, custom_words=custom_word_list)
                 if not file_meta.begin_episode or file_meta.begin_episode not in episodes:
                     continue
             if not Path(file).suffix or Path(file).suffix.lower() not in self.runtime_config.media_extensions:

app/chain/transfer/settlement.py

@@ -666,13 +666,13 @@ class TransferSettlementOwner(_TransferOwnerBase):
     def _TransferChain__is_torrent_download_completed(
             self, download_hash: str, downloader: Optional[str]
     ) -> bool:
-        """
-        检查种子在下载器中是否已完成下载；查询不到或查询失败时视为未完成，
-        留待下载器定时轮询兜底，避免误打已整理标签。
-        """
+        """确认种子下载完成；查询失败或缺失时记录原因，不把未知状态当作完成。"""
         try:
             torrents = self.list_torrents(hashs=download_hash, downloader=downloader)
             if not torrents:
+                logger.warning(
+                    f"下载器 {downloader} 中未查询到种子 {download_hash}，无法回写已整理标签，请检查种子是否已移除或历史关联是否正确"
+                )
                 return False
             return all((torrent.progress or 0) >= 100 for torrent in torrents)
         except Exception as e:

app/db/models/downloadhistory.py

@@ -191,8 +191,10 @@ class DownloadHistory(Base):
 
     @classmethod
     def get_by_path(cls, db: Session, path: str):
+        """保存目录被多次下载复用时，按登记顺序返回最新历史。"""
         return db.execute(
             select(DownloadHistory).where(DownloadHistory.path == path)
+            .order_by(DownloadHistory.id.desc()).limit(1)
         ).scalars().first()
 
     @classmethod

tests/test_db_downloadhistory_queries.py

@@ -115,6 +115,15 @@ def test_list_by_page_is_newest_first_and_paged(db):
     )] == ["p-3", "p-2"]
 
 
+def test_get_by_path_returns_latest_history_for_reused_directory(db):
+    """同目录周更下载应返回最新 hash，不依赖数据库碰巧返回的第一行。"""
+    older = _history("旧集", download_hash="old-hash", path="/downloads/reused")
+    latest = _history("新集", download_hash="new-hash", path="/downloads/reused")
+    db.add(older, latest)
+
+    assert DownloadHistory.get_by_path(db.session, "/downloads/reused").download_hash == "new-hash"
+
+
 def test_get_by_path_finds_the_download_directory(db):
     """
     按保存路径查询用于把落地文件反查回下载任务，查不到时返回 None。

tests/test_download_history_episode_mapping.py

@@ -0,0 +1,71 @@
+"""订阅集数偏移后的文件明细与目录监控关联回归。"""
+
+from pathlib import Path
+from unittest.mock import MagicMock
+
+import pytest
+
+from app.chain.download.facade import DownloadChain
+from app.chain.transfer.facade import TransferChain
+from app.db.models.downloadhistory import DownloadFiles, DownloadHistory
+from app.db.oper.downloadhistory import DownloadHistoryOper
+from app.domain.context import Context, MediaInfo, TorrentInfo
+from app.domain.metainfo import MetaInfo
+from app.schemas.types import MediaType
+
+
+@pytest.mark.parametrize("offset", [0, 165])
+@pytest.mark.parametrize("durable", [False, True])
+def test_selected_episode_file_resolves_its_own_hash(db, offset, durable):
+    """映射后的文件须落库并精确关联 hash，即使同目录还有更新的下载历史。"""
+    db.watermark(DownloadHistory, DownloadFiles)
+    custom_words = (
+        "A.Will.Eternal.S04 => 一念永恒{[tmdbid=107371;type=tv]}S01 "
+        f"&& S01 <> 2160p >> EP+{offset}"
+    ) if offset else None
+    selected_file = "A.Will.Eternal.S04E10.2026.2160p.mkv"
+    unselected_file = "A.Will.Eternal.S04E11.2026.2160p.mkv"
+    meta = MetaInfo(selected_file, custom_words=[custom_words] if custom_words else None)
+    assert meta.begin_episode == 10 + offset
+    media = MediaInfo(type=MediaType.TV, title="一念永恒", tmdb_id=107371)
+    torrent = TorrentInfo(title="A.Will.Eternal.S04.2026.2160p")
+    context = Context(meta_info=meta, media_info=media, torrent_info=torrent)
+    chain = DownloadChain.__new__(DownloadChain)
+    chain.download_history_repository = MagicMock()
+    chain.durable_event_writer = MagicMock() if durable else None
+    chain._build_download_notification = MagicMock(return_value=None)
+    chain._submit_download_added_task = MagicMock()
+    chain.eventmanager = MagicMock()
+
+    chain._settle_download_success(
+        context=context, media=media, meta=meta, torrent=torrent,
+        folder_name="shared-show", file_list=[selected_file, unselected_file, "readme.txt"],
+        download_dir=Path("/downloads"), layout="Original", downloader="qb",
+        download_hash="episode-10-hash", download_episodes=f"E{10 + offset}",
+        episodes={10 + offset}, channel=None, source="subscribe", userid=None,
+        username=None, torrent_content=b"torrent", custom_words=custom_words,
+    )
+
+    if durable:
+        payload = chain.durable_event_writer.download_added.call_args.kwargs
+        history, files = payload["history"], payload["files"]
+    else:
+        history, files = chain.download_history_repository.add.call_args.args
+    assert [file.filepath for file in files] == [selected_file]
+    assert files[0].fullpath == f"/downloads/shared-show/{selected_file}"
+    assert files[0].download_hash == "episode-10-hash"
+    assert history.custom_words == custom_words
+
+    db.add(DownloadHistory(**history.to_payload()))
+    db.add(*(DownloadFiles(**file.to_payload()) for file in files))
+    later_history = history.to_payload()
+    later_history["download_hash"] = "episode-11-hash"
+    db.add(DownloadHistory(**later_history))
+    repository = DownloadHistoryOper(db.session)
+    assert repository.get_by_path(history.path).download_hash == "episode-11-hash"
+    transfer = TransferChain.__new__(TransferChain)
+    resolved = transfer._resolve_download_history(
+        repository=repository, file_path=Path(files[0].fullpath),
+    )
+    assert resolved.download_hash == "episode-10-hash"
+    assert resolved.downloader == "qb"

tests/test_transfer_mark_torrent_completed.py

@@ -1,8 +1,9 @@
 # -*- coding: utf-8 -*-
 from types import SimpleNamespace
+from unittest.mock import MagicMock
 
 from app.application.transfer.workflow import JobManager, TransferTask
-from app.chain.transfer import TransferChain
+from app.chain.transfer.facade import TransferChain
 from app.domain.meta.metabase import MetaBase
 from app.runtime.config import settings
 from app.schemas.file import FileItem
@@ -155,8 +156,10 @@ def test_mark_short_circuits_downloader_query_when_jobview_not_done():
     assert queries == []
 
 
-def test_mark_skips_tag_when_torrent_not_found():
-    """下载器中查不到种子时不打标签，留待定时轮询兜底。"""
+def test_mark_skips_tag_when_torrent_not_found(monkeypatch):
+    """查不到种子时不打标签，并告警提供下载器与 hash 供排查关联。"""
+    warning = MagicMock()
+    monkeypatch.setattr("app.chain.transfer.settlement.logger.warning", warning)
     chain = _make_chain()
     completed = []
     chain.transfer_completed = lambda **kwargs: completed.append(kwargs)
@@ -166,6 +169,11 @@ def test_mark_skips_tag_when_torrent_not_found():
     _mark(chain, "hash1", "qbittorrent")
 
     assert completed == []
+    warning.assert_called_once()
+    message = warning.call_args.args[0]
+    assert "qbittorrent" in message
+    assert "hash1" in message
+    assert "未查询到种子" in message
 
 
 def test_mark_skips_tag_when_list_torrents_raises():
@@ -175,6 +183,7 @@ def test_mark_skips_tag_when_list_torrents_raises():
     chain.transfer_completed = lambda **kwargs: completed.append(kwargs)
 
     def _raise(**_kwargs):
+        """模拟下载器查询失败。"""
         raise RuntimeError("downloader unreachable")
 
     chain.list_torrents = _raise