nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-01 21:07:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(download): 重复接受且零新增交付的种子按资源退避，订阅改试其他资源 (#7107)

Browse Source 
* fix(download): back off resources re-submitted with no new files

A subscription that re-selects an already-held package gets accepted by the
downloader again every search round, so the same resource keeps being
submitted while nothing new can be delivered. Record a resource-scoped
backoff when a reused task contributes no file rows: it suppresses only that
one resource, so candidates keep trying other torrents for the same media and
the subscription is neither completed nor failed. The cooldown ledger is no
longer failure-only, so the skip log reports the recorded reason instead of
claiming the add failed.

* fix(download): keep duplicate backoff stable across selection

---------

Co-authored-by: jianlin6 <jianlin6@jianlin6-MacBookPro.local>
Co-authored-by: jxxghp <jxxghp@gmail.com>
This commit is contained in:
JontyLee
2026-10-20 18:35:14 +08:00
committed by GitHub
parent 6766a0516e
commit d57af7979e
6 changed files with 284 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
app/chain/download/contract.py
View File

@@ -30,6 +30,7 @@ if TYPE_CHECKING:
_matches_media_identity: Callable[..., bool]
_prepare_batch_download_contexts: Callable[..., Any]
_record_download_failure: Callable[..., Any]
_record_duplicate_download_backoff: Callable[..., Any]
_record_music_album_track_keys: Callable[..., Any]
_resolve_media_download_dir: Callable[..., Any]
_settle_download_success: Callable[..., None]

117
app/chain/download/failure.py
View File

@@ -1,5 +1,5 @@
"""
下载失败指纹、冷却和持久化 owner。
下载资源级失败与退避指纹、冷却和持久化 owner。
"""
import hashlib
@@ -27,6 +27,10 @@ from app.schemas.media import build_media_key, resolve_media_identity
DOWNLOAD_FAILURE_RESOURCE_TTL_SECONDS = 24 * 60 * 60
DOWNLOAD_FAILURE_TRANSIENT_TTL_SECONDS = 60 * 60
# 重复提交退避只压制单个资源指纹，必须长于订阅重新选源的周期，否则同一个整包仍会被反复接受。
DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS = 7 * 24 * 60 * 60
# 退避记录写入台账的成因说明，候选跳过日志直接引用它，不能再假设这是一次失败。
DOWNLOAD_FAILURE_DUPLICATE_MESSAGE = "下载器已存在同一任务且本次无新增文件明细"
DOWNLOAD_FAILURE_RESOURCE_ERROR_KEYWORDS = (
"无法读取种子文件",
"无法获取下载地址",
@@ -44,7 +48,7 @@ DOWNLOAD_FAILURE_RESOURCE_ERROR_KEYWORDS = (
class DownloadFailureOwner(_DownloadOwnerBase):
"""
下载失败指纹、冷却和持久化 owner。
下载资源级失败与退避指纹、冷却和持久化 owner。
"""
@staticmethod
@@ -101,9 +105,15 @@ class DownloadFailureOwner(_DownloadOwnerBase):
return f"title={title}|size={size}"
@classmethod
def _build_download_failure_fingerprint(cls, context: Context) -> Optional[str]:
def _build_download_failure_fingerprint(
cls, context: Context, *, resource_backoff: bool = False,
) -> Optional[str]:
"""
根据媒体和种子资源信息生成失败冷却指纹。
根据媒体和种子资源信息生成冷却指纹，整包退避不依赖解析后补齐的集数。
:param context: 识别出的媒体与资源上下文
:param resource_backoff: 是否为整包均已记录的资源退避；与普通失败指纹隔离
:return: 稳定资源指纹；缺少身份时返回 None
"""
media = getattr(context, "media_info", None)
torrent = getattr(context, "torrent_info", None)
@@ -124,7 +134,7 @@ class DownloadFailureOwner(_DownloadOwnerBase):
"media_type": str(media_type or ""),
"media_key": str(media_key or ""),
"season": str(season) if season is not None else "",
"episodes": cls._format_failure_episodes(meta) or "",
"episodes": None if resource_backoff else cls._format_failure_episodes(meta) or "",
"site": str(site or ""),
"resource": cls._torrent_resource_key(torrent),
}
@@ -153,11 +163,25 @@ class DownloadFailureOwner(_DownloadOwnerBase):
downloader: Optional[str] = None,
source: Optional[str] = None,
episodes: Optional[Set[int]] = None,
ttl_seconds: Optional[int] = None,
resource_backoff: bool = False,
) -> Optional[str]:
"""
记录资源级下载失败，并返回本次失败指纹。
记录资源级冷却事实，并返回本次资源指纹。
该台账是资源级退避记录，既承载真实失败，也承载"重复提交但无新增交付"这类
没有失败、但不值得立刻再试一次的事实，因此 ``error_msg`` 只是跳过原因说明。
:param context: 本次提交的上下文，用于计算资源指纹
:param error_msg: 写入台账的跳过原因说明
:param downloader: 本次使用的下载器
:param source: 下载来源，仅订阅来源的冷却会被读取
:param episodes: 目标集数，缺省时按识别元数据记录
:param ttl_seconds: 显式冷却秒数，缺省时按原因文本推断
:param resource_backoff: 是否使用不含运行时集数的整包退避指纹
:return: 本次记录的资源指纹；缺少身份时返回 None
"""
fingerprint = self._build_download_failure_fingerprint(context)
fingerprint = self._build_download_failure_fingerprint(context, resource_backoff=resource_backoff)
if not fingerprint:
return None
@@ -165,7 +189,9 @@ class DownloadFailureOwner(_DownloadOwnerBase):
now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(now_timestamp))
next_retry_at = time.strftime(
"%Y-%m-%d %H:%M:%S",
time.localtime(now_timestamp + self._download_failure_ttl(error_msg)),
time.localtime(now_timestamp + (
ttl_seconds if ttl_seconds is not None else self._download_failure_ttl(error_msg)
)),
)
media = context.media_info
media_source, media_id = resolve_media_identity(media=media)
@@ -212,32 +238,74 @@ class DownloadFailureOwner(_DownloadOwnerBase):
logger.error(f"记录下载失败冷却失败：{str(err)}")
return fingerprint
def _record_duplicate_download_backoff(
self,
context: Context,
downloader: Optional[str] = None,
source: Optional[str] = None,
episodes: Optional[Set[int]] = None,
) -> Optional[str]:
"""
为重复接受且未带来任何新增文件的资源记录退避，并返回该资源指纹。
调用方须确认整包全部媒体文件均已记录。退避只压制这一个资源，订阅候选会继续
尝试同媒体的其他资源，订阅本身既不会被判定完成也不会被判失败，冷却到期后仍可重试。
仅订阅来源需要记录，因为候选过滤只在订阅来源读取该台账。
:param context: 本次结算的上下文，用于计算资源指纹
:param downloader: 本次使用的下载器
:param source: 下载来源
:param episodes: 目标集数
:return: 本次记录的整包资源指纹；非订阅来源返回 None
"""
if not self._is_subscribe_source(source):
return None
return self._record_download_failure(
context=context,
error_msg=DOWNLOAD_FAILURE_DUPLICATE_MESSAGE,
downloader=downloader,
source=source,
episodes=episodes,
ttl_seconds=DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS,
resource_backoff=True,
)
def _active_download_failure_fingerprints(
self,
contexts: List[Context],
source: Optional[str],
) -> Dict[str, DownloadFailureSnapshot]:
"""
查询当前订阅候选中仍处于冷却期的失败记录，返回指纹到失败记录的映射。
查询候选的普通失败与整包退避，并映射回当前候选指纹供所有选源阶段消费。
:param contexts: 本轮待筛选的资源上下文
:param source: 本次批量下载的业务来源
:return: 当前候选指纹到有效冷却记录的映射；非订阅来源不读取台账
"""
if not self._is_subscribe_source(source):
return {}
fingerprints = [
fingerprint
for fingerprint in [
self._build_download_failure_fingerprint(context)
for context in contexts or []
]
if fingerprint
]
backoff_fingerprints = {
fingerprint: self._build_download_failure_fingerprint(context, resource_backoff=True) or fingerprint
for context in contexts or []
if (fingerprint := self._build_download_failure_fingerprint(context))
}
fingerprints = list(dict.fromkeys([
*backoff_fingerprints,
*backoff_fingerprints.values(),
]))
if not fingerprints:
return {}
now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
try:
repository: DownloadFailureRepository = self.download_failure_repository
return repository.get_active_by_fingerprints(
records = repository.get_active_by_fingerprints(
fingerprints=fingerprints, now_time=now_time,
)
return {
fingerprint: records.get(fingerprint) or records[backoff]
for fingerprint, backoff in backoff_fingerprints.items()
if fingerprint in records or backoff in records
}
except Exception as err:
logger.error(f"查询下载失败冷却失败：{str(err)}")
return {}
@@ -248,17 +316,20 @@ class DownloadFailureOwner(_DownloadOwnerBase):
failure: Optional[DownloadFailureSnapshot],
) -> None:
"""
记录候选资源处于失败冷却期时的跳过原因和下次重试时间。
记录候选资源处于冷却期时的跳过原因和下次重试时间。
台账里的成因可能是真实失败，也可能是重复提交退避，所以这里只转述记录中的原因，
不把它断言成"添加下载失败"。
"""
reason = getattr(failure, "error_message", None) or "未知原因"
retry_at = getattr(failure, "next_retry_at", None)
if retry_at:
logger.info(
f"{context.torrent_info.title} 近期添加下载失败（失败原因：{reason}），"
f"暂时跳过该资源，将于 {retry_at} 后重试"
f"{context.torrent_info.title} 暂时跳过该资源（原因：{reason}），"
f"将于 {retry_at} 后重试，本次继续尝试其他资源"
)
else:
logger.info(
f"{context.torrent_info.title} 近期添加下载失败（失败原因：{reason}），"
"暂时跳过该资源"
f"{context.torrent_info.title} 暂时跳过该资源（原因：{reason}），"
"本次继续尝试其他资源"
)

27
app/chain/download/history.py
View File

@@ -82,7 +82,8 @@ class DownloadHistoryOwner(_DownloadOwnerBase):
"""
提交下载历史、文件明细和 durable 下载事件。
:param already_exists: 下载器中已存在同一任务时为 True，仅补记历史，不再重复通知与文件明细
:param already_exists: 下载器中已存在同一任务时为 True，仅补记历史，不再重复通知与文件明细；
若整包全部媒体文件已记录过，同时为这个资源登记提交退避
"""
if layout == "NoSubfolder" or not folder_name:
download_path = download_dir / file_list[0] if file_list else download_dir
@@ -127,11 +128,6 @@ class DownloadHistoryOwner(_DownloadOwnerBase):
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, 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:
continue
files_to_add.append(
@@ -145,7 +141,24 @@ class DownloadHistoryOwner(_DownloadOwnerBase):
)
)
if already_exists:
files_to_add = self._filter_unrecorded_download_files(files_to_add, download_hash)
retained_files = self._filter_unrecorded_download_files(files_to_add, download_hash)
if files_to_add and not retained_files:
# 必须核对整包文件，不能因本次只选了已记录的一集就阻止以后补下包内其他集。
self._record_duplicate_download_backoff(
context=context,
downloader=downloader,
source=source,
episodes=episodes,
)
files_to_add = retained_files
if episodes:
selected_files = []
for item in files_to_add:
# 与订阅选集使用相同识别词，避免偏移前的文件集数被目标集数过滤。
file_meta = MetaInfo(Path(cast(str, item.filepath)).stem, custom_words=custom_word_list)
if file_meta.begin_episode and file_meta.begin_episode in episodes:
selected_files.append(item)
files_to_add = selected_files
frozen_files = tuple(files_to_add)
event_payload = {
"hash": download_hash, "context": context, "username": username,

7
docs/testing.md
View File

@@ -163,6 +163,13 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)):
短期签名或换票资源在下载链中屏蔽原始 enclosure，不将临时请求地址写入无效缓存。
订阅持久化保留具体失败原因，无效种子使用 24 小时资源冷却，避免重启后每小时重复通知。
网络异常、空响应、403/5xx 和限流仍沿用瞬时重试与流控规则，不因本次修改加入内容无效黑名单。
下载器报告同一任务已存在且整包全部媒体文件明细已记录时，按资源指纹登记 7 天提交退避，
使订阅不再重投同一个已持有的整包；退避只压制该资源，候选仍继续尝试同媒体的其他种子，
订阅不会被判定完成或失败，且仅订阅来源写台账（手动补投不留冷却记录）。冷却跳过日志转述台账
中的原因，不再把它断言成"添加下载失败"。见 `tests/test_torrent_download_retry.py` 与
`tests/test_downloader_submission.py`。
整包退避指纹不包含解析种子后补齐的集数，下一轮搜索仍能命中；普通失败仍按原有集数指纹隔离。
只重复选择了已记录的部分集数、但包内仍有未记录文件时，不登记整包退避，也不把未选文件写入历史。
换票资源的跨批次复用见 `tests/test_download_resource_cache.py`：种子成功解析后，
本地 `torrents` 缓存以站点资源身份、原始换票请求和 Cookie 的摘要保存内容，命中后

45
tests/test_downloader_submission.py
View File

@@ -381,15 +381,17 @@ def test_downloader_task_identity_decides_reuse_flag(downloader, existing):
@pytest.mark.parametrize("durable", [False, True])
@pytest.mark.parametrize("source", ["Subscribe|1", "MoviePilot|手动下载"])
@pytest.mark.parametrize("existing,expected_files", [
("matched", [_FILES[1]]),
("all", []),
("empty", _FILES),
("unrelated", _FILES),
], ids=["matched", "all", "empty", "unrelated"])
def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, existing, expected_files):
def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, source, existing, expected_files):
"""
复用已有任务时仍记录历史并触发整理与事件，但不生成通知意图，也不重复写入已记录的文件明细。
复用已有任务时仍记录历史并触发整理与事件，但不生成通知意图、不重复写入已记录的文件明细；
文件明细全部已记录时才为该资源登记退避，仍有新增文件则不退避。
"""
recorded_file, new_file = _FILES
records = {
@@ -402,6 +404,7 @@ def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, exi
media = MediaInfo(type=MediaType.TV, title="一念永恒", tmdb_id=107371)
torrent = TorrentInfo(title="A.Will.Eternal.S04.2160p")
chain = DownloadChain.__new__(DownloadChain)
chain.download_failure_repository = MagicMock()
chain.download_history_repository = MagicMock()
chain.download_history_repository.get_files_by_hash.return_value = [
SimpleNamespace(fullpath=path) for path in records
@@ -418,7 +421,7 @@ def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, exi
folder_name="shared-show", file_list=[recorded_file, new_file],
download_dir=Path("/downloads"), layout="Original", downloader="qb",
download_hash="reuse-hash", download_episodes=None, episodes=None,
channel=None, source="subscribe", userid=None, username=None,
channel=None, source=source, userid=None, username=None,
torrent_content=b"torrent", custom_words=None, already_exists=True,
)
@@ -435,6 +438,42 @@ def test_reuse_settlement_keeps_history_and_skips_duplicate_effects(durable, exi
assert [item.filepath for item in files] == list(expected_files)
chain._build_download_notification.assert_not_called()
chain.download_history_repository.get_files_by_hash.assert_called_once_with("reuse-hash", state=1)
record_backoff = chain.download_failure_repository.record_failure
assert record_backoff.called is (existing == "all" and source == "Subscribe|1")
if record_backoff.called:
write = record_backoff.call_args.args[0]
assert write.source == source
assert write.downloader == "qb"
def test_reused_episode_subset_keeps_other_torrent_files_available():
"""
仅重投已记录的一集时，整包仍含未记录的另一集，不能对整个资源登记退避。
"""
recorded_file, _new_file = _FILES
meta = MetaInfo(recorded_file)
media = MediaInfo(type=MediaType.TV, title="一念永恒", tmdb_id=107371)
torrent = TorrentInfo(title="A.Will.Eternal.S04.2160p")
chain = DownloadChain.__new__(DownloadChain)
chain.download_history_repository = MagicMock()
chain.download_history_repository.get_files_by_hash.return_value = [
SimpleNamespace(fullpath=f"/downloads/shared-show/{recorded_file}"),
]
chain.download_failure_repository = MagicMock()
chain.durable_event_writer = MagicMock()
chain._settle_download_success(
context=Context(meta_info=meta, media_info=media, torrent_info=torrent),
media=media, meta=meta, torrent=torrent,
folder_name="shared-show", file_list=_FILES,
download_dir=Path("/downloads"), layout="Original", downloader="qb",
download_hash="reuse-hash", download_episodes=None, episodes={5},
channel=None, source="Subscribe|1", userid=None, username=None,
torrent_content=b"torrent", custom_words=None, already_exists=True,
)
chain.download_failure_repository.record_failure.assert_not_called()
assert chain.durable_event_writer.download_added.call_args.kwargs["files"] == ()
def test_reuse_settlement_without_media_files_skips_file_lookup():

121
tests/test_torrent_download_retry.py
View File

@@ -5,15 +5,24 @@
import json
from datetime import datetime
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock
import pytest
from requests import Response
import app.application.torrent.download as download
import app.chain.download.failure as failure_owner
import app.chain.download.submission as submission
from app.chain.download import DownloadChain
from app.chain.download.failure import DownloadFailureOwner
from app.chain.download.failure import (
DOWNLOAD_FAILURE_DUPLICATE_MESSAGE,
DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS,
DownloadFailureOwner,
)
from app.db.adapters.download import TransactionalDownloadFailureRepository
from app.db.models.downloadfailure import DownloadFailure
from app.db.session import SessionFactory
from app.domain.context import Context, MediaInfo, TorrentInfo
from app.domain.metainfo import MetaInfo
from app.schemas.types import MediaType
@@ -278,6 +287,116 @@ def test_invalid_content_preparation_preserves_persisted_resource_cooldown(bound
- datetime.fromisoformat(failure.failed_at)).total_seconds() == 86400
def _duplicate_backoff_context():
"""
构造可生成稳定资源指纹的重复提交上下文。
"""
return Context(
torrent_info=TorrentInfo(site=1, site_name="test", title="Series S01", enclosure=URL),
media_info=MediaInfo(title="Series", type=MediaType.TV),
meta_info=MetaInfo("Series S01"),
)
@pytest.mark.parametrize("source,recorded", [("Subscribe|1", True), ("MoviePilot|手动下载", False)])
def test_duplicate_submission_backoff_is_subscribe_only_and_long(source, recorded):
"""
重复提交退避只为订阅来源登记台账，并使用覆盖重新选源周期的长冷却与中性成因。
"""
chain = object.__new__(DownloadChain)
chain.download_failure_repository = Mock()
context = _duplicate_backoff_context()
fingerprint = chain._record_duplicate_download_backoff(
context=context, downloader="qb", source=source,
)
if not recorded:
assert fingerprint is None
chain.download_failure_repository.record_failure.assert_not_called()
return
write = chain.download_failure_repository.record_failure.call_args.args[0]
assert write.error_message == DOWNLOAD_FAILURE_DUPLICATE_MESSAGE
assert write.downloader == "qb"
assert (datetime.fromisoformat(write.next_retry_at)
- datetime.fromisoformat(write.failed_at)).total_seconds() == DOWNLOAD_FAILURE_DUPLICATE_TTL_SECONDS
assert fingerprint == chain._build_download_failure_fingerprint(context, resource_backoff=True)
def test_duplicate_backoff_cools_only_that_resource_for_subscription_candidates(db, monkeypatch):
"""
退避按资源指纹生效：同一媒体的其他种子仍可被候选选中。
"""
db.watermark(DownloadFailure)
chain = object.__new__(DownloadChain)
chain.download_failure_repository = TransactionalDownloadFailureRepository(SessionFactory)
cooled = _duplicate_backoff_context()
cooled.media_info = MediaInfo(title="Series", type=MediaType.MOVIE)
other = Context(
torrent_info=TorrentInfo(site=1, site_name="test", title="Series S01 1080p",
enclosure="https://tracker.example/download.php?id=40"),
media_info=cooled.media_info,
meta_info=MetaInfo("Series S01"),
)
fingerprint = chain._record_duplicate_download_backoff(
context=cooled, downloader="qb", source="Subscribe|1",
)
chain.eventmanager = Mock()
chain.eventmanager.send_event.return_value = None
chain.download_single = Mock(return_value="new-hash")
monkeypatch.setattr(download.TorrentHelper, "sort_torrents", lambda _self, contexts: contexts)
downloads, _lefts = chain.batch_download(contexts=[cooled, other], source="Subscribe|1")
assert fingerprint
assert downloads == [other]
chain.download_single.assert_called_once()
assert chain.download_single.call_args.args[0] is other
def test_duplicate_backoff_survives_whole_season_episode_expansion(db):
"""
整季种子解析后的退避，在下一轮尚未解析文件集数时仍须命中同一资源。
"""
db.watermark(DownloadFailure)
chain = object.__new__(DownloadChain)
chain.download_failure_repository = TransactionalDownloadFailureRepository(SessionFactory)
submitted = _duplicate_backoff_context()
submitted.meta_info.set_episodes(begin=1, end=12)
chain._record_duplicate_download_backoff(context=submitted, source="Subscribe|1")
next_round = _duplicate_backoff_context()
active = chain._active_download_failure_fingerprints(contexts=[next_round], source="Subscribe|1")
assert chain._build_download_failure_fingerprint(next_round) in active
def test_cooldown_log_keeps_backoff_reason_neutral(monkeypatch):
"""
候选跳过日志转述台账成因，不把重复提交退避说成添加下载失败。
"""
logger = Mock()
monkeypatch.setattr(failure_owner, "logger", logger)
DownloadFailureOwner._log_download_failure_cooldown(
Context(torrent_info=TorrentInfo(title="Series S01")),
SimpleNamespace(error_message=DOWNLOAD_FAILURE_DUPLICATE_MESSAGE,
next_retry_at="2026-10-26 12:00:00"),
)
text = logger.info.call_args.args[0]
assert DOWNLOAD_FAILURE_DUPLICATE_MESSAGE in text
assert "将于 2026-10-26 12:00:00 后重试" in text
assert "添加下载失败" not in text
DownloadFailureOwner._log_download_failure_cooldown(
Context(torrent_info=TorrentInfo(title="Series S01")),
SimpleNamespace(error_message=None, next_retry_at=None),
)
assert logger.info.call_args.args[0] == "Series S01 暂时跳过该资源（原因：未知原因），本次继续尝试其他资源"
def test_redirected_get_retries_at_destination(boundary):
"""
重定向目标上的瞬时错误也重试，并保留原始 URL 的成功缓存键。