refactor: complete chain entrypoint slicing

This commit is contained in:
jxxghp
2026-08-22 22:31:24 +08:00
parent cc967741d8
commit 16f7d65640
6 changed files with 206 additions and 63 deletions
+61 -9
View File
@@ -925,6 +925,25 @@ class DownloadChain(ChainBase):
logger.error(f"查询下载失败冷却失败:{str(err)}")
return {}
@staticmethod
def _log_download_failure_cooldown(
context: Context,
failure: Optional["DownloadFailure"],
) -> 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} 后重试"
)
else:
logger.info(
f"{context.torrent_info.title} 近期添加下载失败(失败原因:{reason}),"
"暂时跳过该资源"
)
def _prepare_batch_download_contexts(
self,
contexts: List[Context],
@@ -987,6 +1006,10 @@ class DownloadChain(ChainBase):
continue
fingerprint = self._build_download_failure_fingerprint(context)
if fingerprint and fingerprint in active_failure_records:
self._log_download_failure_cooldown(
context,
active_failure_records[fingerprint],
)
continue
if media_type == MediaType.MOVIE:
download_key = context.media_info.title_year
@@ -1289,6 +1312,40 @@ class DownloadChain(ChainBase):
return_detail: bool = False,
custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
"""
下载单个资源并发送结果通知。
保持下载链、消息入口和插件使用的公开签名,实际流程委托给内部执行阶段。
"""
return self._execute_download_single(
context=context,
torrent_file=torrent_file,
torrent_content=torrent_content,
episodes=episodes,
channel=channel,
source=source,
downloader=downloader,
save_path=save_path,
userid=userid,
username=username,
label=label,
return_detail=return_detail,
custom_words=custom_words,
)
def _execute_download_single(self, context: Context,
torrent_file: Path = None,
torrent_content: Optional[Union[str, bytes]] = None,
episodes: Set[int] = None,
channel: NotificationChannel = None,
source: Optional[str] = None,
downloader: Optional[str] = None,
save_path: Optional[str] = None,
userid: Union[str, int] = None,
username: Optional[str] = None,
label: Optional[str] = None,
return_detail: bool = False,
custom_words: Optional[str] = None) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]:
"""
下载及发送通知
:param context: 资源上下文
:param torrent_file: 种子文件路径
@@ -1623,15 +1680,10 @@ class DownloadChain(ChainBase):
"""
fingerprint = self._build_download_failure_fingerprint(_context)
if fingerprint and fingerprint in active_failure_records:
_failure = active_failure_records[fingerprint]
_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} 后重试")
else:
logger.info(f"{_context.torrent_info.title} 近期添加下载失败(失败原因:{_reason}),"
f"暂时跳过该资源")
self._log_download_failure_cooldown(
_context,
active_failure_records[fingerprint],
)
return True
return False
+2
View File
@@ -554,6 +554,8 @@ class MediaServerChain(ChainBase):
global_media_total=global_media_total,
global_media_finished=global_media_finished,
)
if global_vars.is_system_stopped:
return
total_count += server_count
logger.info(f"媒体服务器 {server_name} 数据同步完成,总同步数量:{total_count}")
if progress_callback:
+95 -49
View File
@@ -1498,6 +1498,25 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
state: Optional[str] = 'N',
manual: Optional[bool] = False,
progress_callback: Optional[Callable[..., None]] = None,
) -> None:
"""
执行订阅搜索。
保持定时任务、API 和插件使用的公开签名,搜索实现委托给内部执行阶段。
"""
return self._execute_search(
sid=sid,
state=state,
manual=manual,
progress_callback=progress_callback,
)
def _execute_search(
self,
sid: Optional[int] = None,
state: Optional[str] = 'N',
manual: Optional[bool] = False,
progress_callback: Optional[Callable[..., None]] = None,
) -> None:
"""
订阅搜索
@@ -1928,10 +1947,85 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
self.get_states_for_search('R')
)
def _prepare_match_torrents(
self,
torrents: Dict[str, List[Context]],
) -> Dict[str, List[Context]]:
"""预识别待匹配资源,并保留原上下文供后续订阅复用。"""
processed_torrents: Dict[str, List[Context]] = {}
for domain, contexts in torrents.items():
if global_vars.is_system_stopped:
break
processed_torrents[domain] = []
for context in contexts:
if global_vars.is_system_stopped:
break
if context.torrent_info and getattr(context.torrent_info, "category", None) in (
MediaType.MUSIC,
MediaType.MUSIC.value,
):
# 音乐 RSS 使用订阅目标做实体匹配,不应进入影视识别并累计失败次数。
processed_torrents[domain].append(context)
continue
if (
not context.media_info
or not resolve_media_identity(media=context.media_info)[1]
) and context.media_recognize_fail_count < 3:
logger.debug(
f'尝试重新识别种子:{context.torrent_info.title},当前失败次数:'
f'{context.media_recognize_fail_count}/3'
)
re_mediainfo = MediaChain().recognize_by_meta(
context.meta_info,
obtain_images=False,
)
if re_mediainfo:
re_mediainfo.clear()
context.media_info = re_mediainfo
context.match_source = self.__get_media_id_match_source(re_mediainfo)
context.candidate_recognized = bool(
resolve_media_identity(media=re_mediainfo)[1]
)
context.media_info_is_target = False
context.media_recognize_fail_count = 0
logger.debug(f'种子 {context.torrent_info.title} 重新识别成功')
else:
context.media_recognize_fail_count += 1
logger.debug(
f'种子 {context.torrent_info.title} 媒体识别失败,失败次数:'
f'{context.media_recognize_fail_count}/3'
)
elif context.media_recognize_fail_count >= 3:
logger.debug(f'种子 {context.torrent_info.title} 已达到最大识别失败次数(3次),跳过识别')
processed_torrents[domain].append(context)
return processed_torrents
def match(
self,
torrents: Dict[str, List[Context]],
progress_callback: Optional[Callable[..., None]] = None,
) -> None:
"""
从缓存中匹配订阅,并自动下载。
该入口保持订阅刷新、定时任务和插件调用的稳定签名,具体匹配流程由内部阶段执行。
"""
if not torrents:
logger.warn('没有缓存资源,无法匹配订阅')
if progress_callback:
progress_callback(value=100, text="没有缓存资源,跳过订阅匹配")
return
if progress_callback:
progress_callback(value=0, text="正在预处理订阅资源 ...")
return self._execute_match(
torrents=torrents,
progress_callback=progress_callback,
)
def _execute_match(
self,
torrents: Dict[str, List[Context]],
progress_callback: Optional[Callable[..., None]] = None,
) -> None:
"""
从缓存中匹配订阅,并自动下载
@@ -1954,55 +2048,7 @@ class SubscribeChain(MusicSubscribeMixin, InteractionChainMixin, ChainBase):
if not lock_acquired:
return
# 预识别所有未识别的种子
processed_torrents: Dict[str, List[Context]] = {}
for domain, contexts in torrents.items():
if global_vars.is_system_stopped:
break
processed_torrents[domain] = []
for context in contexts:
if global_vars.is_system_stopped:
break
if context.torrent_info and getattr(context.torrent_info, "category", None) in (
MediaType.MUSIC,
MediaType.MUSIC.value,
):
# 音乐 RSS 使用订阅目标做实体匹配,不应进入影视识别并累计失败次数。
processed_torrents[domain].append(context)
continue
# 如果种子未识别且失败次数未超过3次,尝试识别
if (
not context.media_info
or not resolve_media_identity(media=context.media_info)[1]
) and context.media_recognize_fail_count < 3:
logger.debug(
f'尝试重新识别种子:{context.torrent_info.title},当前失败次数:{context.media_recognize_fail_count}/3')
re_mediainfo = MediaChain().recognize_by_meta(
context.meta_info,
obtain_images=False,
)
if re_mediainfo:
# 清理多余信息
re_mediainfo.clear()
# 更新种子缓存
context.media_info = re_mediainfo
context.match_source = self.__get_media_id_match_source(re_mediainfo)
context.candidate_recognized = bool(
resolve_media_identity(media=re_mediainfo)[1]
)
context.media_info_is_target = False
# 重置失败次数
context.media_recognize_fail_count = 0
logger.debug(f'种子 {context.torrent_info.title} 重新识别成功')
else:
# 识别失败,增加失败次数
context.media_recognize_fail_count += 1
logger.debug(
f'种子 {context.torrent_info.title} 媒体识别失败,失败次数:{context.media_recognize_fail_count}/3')
elif context.media_recognize_fail_count >= 3:
logger.debug(f'种子 {context.torrent_info.title} 已达到最大识别失败次数(3次),跳过识别')
# 添加已预处理
processed_torrents[domain].append(context)
processed_torrents = self._prepare_match_torrents(torrents)
# 所有订阅
subscribes = SubscribeOper().list(self.get_states_for_search('R'))
@@ -995,6 +995,13 @@ MFA/Passkey 专项测试与架构门禁通过,密钥类配置仍保留在安
随后将 `TransferChain.do_transfer` 的公开入口收口为稳定兼容 Facade,先提取媒体身份规范化阶段,保留显式
`media_source/media_id` 校验、识别失败文案和所有原有调用参数;整理专项 80 项测试通过,复杂度基线移除该入口,
后续继续拆分其批次规划与执行阶段。
2026-08-22 继续完成入口垂直切片:`DownloadChain.download_single``SubscribeChain.search`
`SubscribeChain.match` 均改为稳定兼容 Facade,分别委托下载执行、搜索执行、资源预处理和订阅匹配阶段;
保留原参数、对象类型、锁、进度回调、停止信号、候选过滤、失败冷却日志和下载结算语义。
`MediaServerChain.sync` 补回停止信号后的立即退出,避免系统停止后继续发送服务器/全局完成进度。
下载、订阅、媒体服务器及 durable/outbox 专项共 370 项测试通过,复杂度基线移除上述三个订阅/下载入口。
当前仍不把普通用户通知和 MoviePilot Server 外部统计标记为 durable:它们尚未与业务写入和 outbox intent
绑定在同一事务,继续保持 post-commit 的准确边界。
#### ARCH-272:异步阻塞检测
+1 -5
View File
@@ -16,9 +16,5 @@
"app/application/rss.py:RssHelper.parse": 206,
"app/application/security/cookie.py:CookieHelper.get_site_cookie_ua": 221
},
"chain_public": {
"app/chain/download.py:DownloadChain.download_single": 167,
"app/chain/subscribe.py:SubscribeChain.match": 415,
"app/chain/subscribe.py:SubscribeChain.search": 246
}
"chain_public": {}
}
@@ -11,6 +11,7 @@ from app.chain.mediaserver import MediaServerChain
from app.db import Base
from app.db.oper.mediaserver import MediaServerOper
from app.db.models.mediaserver import MediaServerItem
from app.runtime.config import global_vars
@pytest.fixture
@@ -329,3 +330,42 @@ def test_sync_targets_one_server_without_excluding_other_enabled_servers(monkeyp
assert library_calls == ["plex-a"]
assert excluded_server_calls == [["plex-a", "plex-b"]]
def test_sync_stops_without_emitting_completion_after_stop_signal(monkeypatch):
"""系统停止发生在逐库同步期间时,不应再发送服务器或全局完成进度。"""
chain = object.__new__(MediaServerChain)
server = SimpleNamespace(name="plex", enabled=True)
progress = []
class FakeMediaServerOper:
"""提供同步阶段所需的最小数据库端口。"""
def delete_excluded_servers(self, _servers):
"""忽略测试中的媒体服务器清理。"""
def stop_during_sync(**_kwargs):
"""模拟读取媒体条目时收到系统停止信号。"""
global_vars.stop_system()
return 0, 0
monkeypatch.setattr(MEDIA_SERVER_CHAIN_MODULE, "MediaServerOper", FakeMediaServerOper)
monkeypatch.setattr(
chain,
"_prepare_sync_contexts",
lambda _servers, _server: ([server], 1, {"plex": ([], {})}, 0),
)
monkeypatch.setattr(chain, "_sync_server_libraries", stop_during_sync)
monkeypatch.setattr(
MEDIA_SERVER_CHAIN_MODULE.ServiceConfigHelper,
"get_mediaserver_configs",
lambda: [server],
)
global_vars.STOP_EVENT.clear()
try:
chain.sync(progress_callback=lambda **kwargs: progress.append(kwargs))
finally:
global_vars.STOP_EVENT.clear()
texts = [item.get("text") for item in progress]
assert not any(text and "同步完成" in text for text in texts)