From e57288b5eaa54180c179c04f29970f4db31fbde2 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Tue, 1 Sep 2026 16:59:52 +0800 Subject: [PATCH] refactor(subscribe): satisfy global governance gates --- app/api/dependencies/subscription.py | 9 +- app/api/endpoints/subexecution.py | 110 ++++ app/api/endpoints/subscribe.py | 91 --- app/api/routers.py | 2 + .../{search_observation.py => observation.py} | 0 app/application/subscription/sitebudget.py | 2 +- app/chain/base.py | 2 - app/chain/download/batch.py | 27 +- app/chain/download/submission.py | 517 +++++++++++------- app/chain/search/provider.py | 10 +- app/chain/subscribe/contract.py | 8 + app/chain/subscribe/match.py | 88 +-- app/chain/subscribe/metadata.py | 144 +++++ app/chain/subscribe/policy.py | 8 +- app/chain/subscribe/refresh.py | 130 +---- app/chain/subscribe/search.py | 8 +- app/db/oper/subscriptiondownload.py | 6 +- app/locales/en-US.json | 2 + app/locales/zh-TW.json | 2 + app/modules/indexer/__init__.py | 2 +- app/scheduler/catalog.py | 64 ++- app/schemas/exports.py | 2 + app/schemas/subscribe.py | 4 +- docs/architecture-overview.md | 4 +- .../architecture/agent-api-surface-audit.json | 40 +- docs/architecture/agent-api-surface-audit.md | 7 +- docs/architecture/optimization-checklist.md | 6 +- scripts/generate_agent_api_surface_audit.py | 8 + scripts/generate_agent_skill_docs.py | 20 + skills/database-operation/SKILL.md | 24 + .../architecture/dependency-baseline.json | 73 ++- .../fixtures/architecture/mypy-baseline.json | 2 +- .../fixtures/architecture/ruff-baseline.json | 3 - .../runtime-contract-baseline.json | 6 +- tests/test_subscription_architecture.py | 1 + tests/test_subscription_governance_joint.py | 2 +- tests/test_subscription_site_budget.py | 2 +- 37 files changed, 908 insertions(+), 528 deletions(-) create mode 100644 app/api/endpoints/subexecution.py rename app/application/site/{search_observation.py => observation.py} (100%) create mode 100644 app/chain/subscribe/metadata.py diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index 497be02d6..9faa5d308 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -43,7 +43,10 @@ from app.application.subscription.mutation import ( ) from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.search import SearchSubscriptionsCommand -from app.application.subscription.status import SubscriptionExecutionStatusService +from app.application.subscription.status import ( + SubscriptionExecutionReadRepository, + SubscriptionExecutionStatusService, +) from app.application.subscription.write import ( SubscriptionBatchWritePort, ) @@ -187,7 +190,9 @@ def get_subscription_execution_status_service( if factory is None: raise RuntimeError("订阅执行状态仓储未注册") repository = factory(db) - return SubscriptionExecutionStatusService(repository=repository) # type: ignore[arg-type] + return SubscriptionExecutionStatusService( + repository=cast(SubscriptionExecutionReadRepository, repository) + ) def get_subscription_search_repository( diff --git a/app/api/endpoints/subexecution.py b/app/api/endpoints/subexecution.py new file mode 100644 index 000000000..9aace7321 --- /dev/null +++ b/app/api/endpoints/subexecution.py @@ -0,0 +1,110 @@ +"""订阅执行批次状态与取消端点。""" + +from typing import Any, List, Optional + +from fastapi import Depends, HTTPException + +from app.api.dependencies.auth import get_current_active_user_async +from app.api.dependencies.subscription import ( + get_subscription_execution_status_service, + get_subscription_query_service, + get_subscription_search_repository, +) +from app.api.principal import ApiPrincipal +from app.api.response import ResponseAPIRouter +from app.application.subscription.execution import SubscriptionSearchRepository +from app.application.subscription.query import SubscriptionQueryService +from app.application.subscription.status import SubscriptionExecutionStatusService +from app.runtime.execution import run_in_threadpool +from app.schemas.response import Response as _SchemaResponse +from app.schemas.subscribe import SubscriptionBatchStatus as _SchemaSubscriptionBatchStatus + +router = ResponseAPIRouter() + + +async def _accessible_subscription_ids( + query: SubscriptionQueryService, + current_user: ApiPrincipal, +) -> Optional[set[int]]: + """返回普通用户可访问订阅 ID；超级用户以 None 表示不限制。""" + if current_user.is_superuser: + return None + subscribes = await query.list_public(current_user.name) + return {item.id for item in subscribes if item.id is not None} + + +@router.get( # type: ignore[misc] + "/execution/batches", + summary="查询订阅搜索批次状态", + response_model=List[_SchemaSubscriptionBatchStatus], +) +async def list_subscription_execution_batches( + limit: int = 10, + status_service: SubscriptionExecutionStatusService = Depends( + get_subscription_execution_status_service + ), + query: SubscriptionQueryService = Depends(get_subscription_query_service), + current_user: ApiPrincipal = Depends(get_current_active_user_async), +) -> Any: + """返回当前用户完整可见的最近搜索批次。""" + accessible_ids = await _accessible_subscription_ids(query, current_user) + batches = await status_service.list_batches( + accessible_subscription_ids=accessible_ids, + limit=limit, + ) + return [_SchemaSubscriptionBatchStatus.model_validate(batch) for batch in batches] + + +@router.get( # type: ignore[misc] + "/execution/batches/{batch_id}", + summary="查询订阅搜索批次", + response_model=_SchemaSubscriptionBatchStatus, +) +async def get_subscription_execution_batch( + batch_id: str, + status_service: SubscriptionExecutionStatusService = Depends( + get_subscription_execution_status_service + ), + query: SubscriptionQueryService = Depends(get_subscription_query_service), + current_user: ApiPrincipal = Depends(get_current_active_user_async), +) -> Any: + """按稳定 ID 返回当前用户可访问的搜索批次。""" + accessible_ids = await _accessible_subscription_ids(query, current_user) + batch = await status_service.get_batch( + batch_id, + accessible_subscription_ids=accessible_ids, + ) + if batch is None: + raise HTTPException(status_code=404, detail="订阅搜索批次不存在") + return _SchemaSubscriptionBatchStatus.model_validate(batch) + + +@router.put( # type: ignore[misc] + "/execution/batches/{batch_id}/cancel", + summary="取消订阅搜索批次", + response_model=_SchemaResponse[None], +) +async def cancel_subscription_execution_batch( + batch_id: str, + status_service: SubscriptionExecutionStatusService = Depends( + get_subscription_execution_status_service + ), + query: SubscriptionQueryService = Depends(get_subscription_query_service), + search_repository: SubscriptionSearchRepository = Depends( + get_subscription_search_repository + ), + current_user: ApiPrincipal = Depends(get_current_active_user_async), +) -> Any: + """在权限校验后请求取消尚未越过下载副作用边界的任务。""" + accessible_ids = await _accessible_subscription_ids(query, current_user) + batch = await status_service.get_batch( + batch_id, + accessible_subscription_ids=accessible_ids, + ) + if batch is None: + return _SchemaResponse(success=False, message="订阅搜索批次不存在") + cancelled = await run_in_threadpool(search_repository.request_cancel, batch_id) + return _SchemaResponse( + success=bool(cancelled), + message="" if cancelled else "订阅搜索批次已结束或无法取消", + ) diff --git a/app/api/endpoints/subscribe.py b/app/api/endpoints/subscribe.py index eb258325b..783f31a04 100644 --- a/app/api/endpoints/subscribe.py +++ b/app/api/endpoints/subscribe.py @@ -25,7 +25,6 @@ from app.api.dependencies.subscription import ( get_subscription_execution_status_service, get_subscription_mutation_service, get_subscription_query_service, - get_subscription_search_repository, ) from app.api.principal import ApiPrincipal from app.api.response import ( @@ -46,7 +45,6 @@ from app.application.subscription.delete import ( DeleteSubscribeCommand, SubscribeDeletionActor, ) -from app.application.subscription.execution import SubscriptionSearchRepository from app.application.subscription.identity import ( DeleteSubscriptionsByIdentityCommand, ) @@ -72,7 +70,6 @@ from app.schemas.subscribe import SubscrbieInfo as _SchemaSubscrbieInfo from app.schemas.subscribe import SubscribeDeletionResult as _SchemaSubscribeDeletionResult from app.schemas.subscribe import SubscribeShare as _SchemaSubscribeShare from app.schemas.subscribe import SubscribeShareStatistics as _SchemaSubscribeShareStatistics -from app.schemas.subscribe import SubscriptionBatchStatus as _SchemaSubscriptionBatchStatus from app.schemas.subscribe import SubscriptionExecutionStatus as _SchemaSubscriptionExecutionStatus from app.schemas.token import TokenPayload as _SchemaTokenPayload from app.schemas.types import ( @@ -105,17 +102,6 @@ async def _attach_execution_status( return subscribes -async def _accessible_subscription_ids( - query: SubscriptionQueryService, - current_user: ApiPrincipal, -) -> Optional[set[int]]: - """返回普通用户可访问订阅 ID；超级用户以 None 表示不限制。""" - if current_user.is_superuser: - return None - subscribes = await query.list_public(current_user.name) - return {item.id for item in subscribes if item.id is not None} - - def start_subscribe_add( title: str, year: str, @@ -502,83 +488,6 @@ async def search_subscribe( return _SchemaResponse(success=True) -@router.get( - "/execution/batches", - summary="查询订阅搜索批次状态", - response_model=List[_SchemaSubscriptionBatchStatus], -) -async def list_subscription_execution_batches( - limit: int = 10, - status_service: SubscriptionExecutionStatusService = Depends( - get_subscription_execution_status_service - ), - query: SubscriptionQueryService = Depends(get_subscription_query_service), - current_user: ApiPrincipal = Depends(get_current_active_user_async), -) -> Any: - """返回当前用户完整可见的最近搜索批次。""" - accessible_ids = await _accessible_subscription_ids(query, current_user) - batches = await status_service.list_batches( - accessible_subscription_ids=accessible_ids, - limit=limit, - ) - return [_SchemaSubscriptionBatchStatus.model_validate(batch) for batch in batches] - - -@router.get( - "/execution/batches/{batch_id}", - summary="查询订阅搜索批次", - response_model=_SchemaSubscriptionBatchStatus, -) -async def get_subscription_execution_batch( - batch_id: str, - status_service: SubscriptionExecutionStatusService = Depends( - get_subscription_execution_status_service - ), - query: SubscriptionQueryService = Depends(get_subscription_query_service), - current_user: ApiPrincipal = Depends(get_current_active_user_async), -) -> Any: - """按稳定 ID 返回当前用户可访问的搜索批次。""" - accessible_ids = await _accessible_subscription_ids(query, current_user) - batch = await status_service.get_batch( - batch_id, - accessible_subscription_ids=accessible_ids, - ) - if batch is None: - raise HTTPException(status_code=404, detail="订阅搜索批次不存在") - return _SchemaSubscriptionBatchStatus.model_validate(batch) - - -@router.put( - "/execution/batches/{batch_id}/cancel", - summary="取消订阅搜索批次", - response_model=_SchemaResponse[None], -) -async def cancel_subscription_execution_batch( - batch_id: str, - status_service: SubscriptionExecutionStatusService = Depends( - get_subscription_execution_status_service - ), - query: SubscriptionQueryService = Depends(get_subscription_query_service), - search_repository: SubscriptionSearchRepository = Depends( - get_subscription_search_repository - ), - current_user: ApiPrincipal = Depends(get_current_active_user_async), -) -> Any: - """在权限校验后请求取消尚未越过下载副作用边界的任务。""" - accessible_ids = await _accessible_subscription_ids(query, current_user) - batch = await status_service.get_batch( - batch_id, - accessible_subscription_ids=accessible_ids, - ) - if batch is None: - return _SchemaResponse(success=False, message="订阅搜索批次不存在") - cancelled = await run_in_threadpool(search_repository.request_cancel, batch_id) - return _SchemaResponse( - success=bool(cancelled), - message="" if cancelled else "订阅搜索批次已结束或无法取消", - ) - - @router.delete("/media/{media_id}", summary="删除订阅", response_model=_SchemaResponse[None]) async def delete_subscribe_by_media_identity( media_id: str, diff --git a/app/api/routers.py b/app/api/routers.py index 243e617b5..871f62341 100644 --- a/app/api/routers.py +++ b/app/api/routers.py @@ -29,6 +29,7 @@ from app.api.endpoints import ( search, site, storage, + subexecution, subscribe, system, tmdb, @@ -58,6 +59,7 @@ API_V1_ROUTER_SPECS: tuple[RouterSpec, ...] = ( RouterSpec(agent.router, "/message/agent", ("agent",)), RouterSpec(webhook.router, "/webhook", ("webhook",)), RouterSpec(subscribe.router, "/subscribe", ("subscribe",)), + RouterSpec(subexecution.router, "/subscribe", ("subscribe",)), RouterSpec(music.router, "/music", ("music",)), RouterSpec(media.router, "/media", ("media",)), RouterSpec(search.router, "/search", ("search",)), diff --git a/app/application/site/search_observation.py b/app/application/site/observation.py similarity index 100% rename from app/application/site/search_observation.py rename to app/application/site/observation.py diff --git a/app/application/subscription/sitebudget.py b/app/application/subscription/sitebudget.py index 1b27c7e72..f3d15d24a 100644 --- a/app/application/subscription/sitebudget.py +++ b/app/application/subscription/sitebudget.py @@ -8,7 +8,7 @@ from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Callable, Optional, Protocol -from app.application.site.search_observation import SiteSearchObservation +from app.application.site.observation import SiteSearchObservation from app.runtime.stop import StopState diff --git a/app/chain/base.py b/app/chain/base.py index c92724073..573676b5c 100644 --- a/app/chain/base.py +++ b/app/chain/base.py @@ -43,7 +43,6 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, """ 处理链基类 """ - def __init__(self, runtime_context: Optional[ChainRuntimeContext] = None): """ 公共初始化；未显式传入上下文时继续使用兼容运行时 provider。 @@ -86,7 +85,6 @@ class ChainBase(RecognitionMixin, MessageProcessingMixin, NotificationMixin, ) self._legacy_transfer_command = context.legacy_transfer_command self.messagequeue = context.message_queue.bind(self.run_module) - @property def runtime_config(self) -> ChainRuntimeConfig: """返回实例快照；兼容绕过构造器的旧调用并按需取得当前快照。""" diff --git a/app/chain/download/batch.py b/app/chain/download/batch.py index 2a860f9ff..b0fb8ca80 100644 --- a/app/chain/download/batch.py +++ b/app/chain/download/batch.py @@ -40,8 +40,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid: Optional[str] = None, username: Optional[str] = None, downloader: Optional[str] = None, - custom_words: Optional[str] = None, - governance: Optional[SubscriptionDownloadGovernance] = None, + custom_words: Optional[str] = None, governance: Optional[SubscriptionDownloadGovernance] = None, ) -> Tuple[ List[Context], Optional[Dict[str, Dict[int, NotExistMediaInfo]]], @@ -61,8 +60,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid=userid, username=username, downloader=downloader, - custom_words=custom_words, - governance=governance, + custom_words=custom_words, governance=governance, ) def _execute_batch_download(self, @@ -74,8 +72,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid: Optional[str] = None, username: Optional[str] = None, downloader: Optional[str] = None, - custom_words: Optional[str] = None, - governance: Optional[SubscriptionDownloadGovernance] = None, + custom_words: Optional[str] = None, governance: Optional[SubscriptionDownloadGovernance] = None, ) -> Tuple[ List[Context], Optional[Dict[str, Dict[int, NotExistMediaInfo]]], @@ -90,8 +87,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): :param userid: 用户ID :param username: 调用下载的用户名/插件名 :param downloader: 下载器 - :param custom_words: 下载来源（如订阅）的完整自定义识别词文本，随下载记录存档，供整理时原样复现识别 - :param governance: 订阅下载幂等、入口任务和取消边界；非订阅调用保持为空 + :param custom_words: 下载来源自定义词；governance: 订阅幂等、入口任务和取消边界 :return: 已下载资源列表及剩余缺集，键格式为 no_exists[source:id] """ no_exists_was_none = no_exists is None @@ -190,8 +186,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid=userid, username=username, downloader=downloader, - custom_words=custom_words, - governance=governance, + custom_words=custom_words, governance=governance, ) # 电视剧整季匹配 @@ -309,8 +304,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid=userid, username=username, downloader=downloader, - custom_words=custom_words, - governance=governance, + custom_words=custom_words, governance=governance, ) else: # 下载 @@ -319,8 +313,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): channel=channel, source=source, userid=userid, username=username, downloader=downloader, - custom_words=custom_words, - governance=governance) + custom_words=custom_words, governance=governance) if download_id: # 下载成功 @@ -409,8 +402,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): channel=channel, source=source, userid=userid, username=username, downloader=downloader, - custom_words=custom_words, - governance=governance) + custom_words=custom_words, governance=governance) if download_id: # 下载成功 if __requires_complete_coverage(missing_info): @@ -527,8 +519,7 @@ class DownloadBatchOwner(_DownloadOwnerBase): userid=userid, username=username, downloader=downloader, - custom_words=custom_words, - governance=governance, + custom_words=custom_words, governance=governance, ) if not download_id: __remember_context_failure(context) diff --git a/app/chain/download/submission.py b/app/chain/download/submission.py index 221dbc2c8..d73d96749 100644 --- a/app/chain/download/submission.py +++ b/app/chain/download/submission.py @@ -3,6 +3,7 @@ import base64 import json import re +from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Optional, Set, Tuple, Union, cast from urllib.parse import urlencode, urljoin, urlparse @@ -27,6 +28,7 @@ from app.domain.context import ( MusicInfo, TorrentInfo, ) +from app.domain.meta.metabase import MetaBase from app.runtime.cache import FileCache from app.runtime.events import eventmanager from app.runtime.log import logger @@ -40,8 +42,24 @@ from app.schemas.types import ( ) -class DownloadSubmissionOwner(_DownloadOwnerBase): - """种子获取与单任务提交 owner。""" +@dataclass(frozen=True, slots=True) +class _PreparedDownload: + """下载器调用前已经验证并规范化的本地提交事实。""" + + torrent: TorrentInfo + media: Union[MediaInfo, MusicInfo] + meta: MetaBase + torrent_content: Union[str, bytes] + folder_name: str + file_list: list[str] + download_dir: Path + download_uri: str + download_episodes: Optional[str] + site_downloader: Optional[str] + + +class _DownloadResourceOwner(_DownloadOwnerBase): + """种子获取、间接地址解析与资源下载事件 owner。""" @staticmethod @@ -262,6 +280,10 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): logger.warn(str(err)) return save_path, str(err) + +class DownloadSubmissionOwner(_DownloadResourceOwner): + """单任务下载准备、提交与结算 owner。""" + def download_single(self, context: Context, torrent_file: Optional[Path] = None, torrent_content: Optional[Union[str, bytes]] = None, @@ -299,146 +321,183 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): governance=governance, ) - def _execute_download_single(self, context: Context, - torrent_file: Optional[Path] = None, - torrent_content: Optional[Union[str, bytes]] = None, - episodes: Optional[Set[int]] = None, - channel: Optional[NotificationChannel] = None, - source: Optional[str] = None, - downloader: Optional[str] = None, - save_path: Optional[str] = None, - userid: Union[str, int, None] = None, - username: Optional[str] = None, - label: Optional[str] = None, - return_detail: bool = False, - custom_words: Optional[str] = None, - governance: Optional[SubscriptionDownloadGovernance] = None, - ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: - """ - 下载及发送通知 - :param context: 资源上下文 - :param torrent_file: 种子文件路径 - :param torrent_content: 种子内容（磁力链或种子文件内容） - :param episodes: 需要下载的集数 - :param channel: 通知渠道 - :param source: 来源（消息通知、Subscribe、Manual等） - :param downloader: 下载器 - :param save_path: 保存路径, 支持:, 如rclone:/MP, smb:/server/share/Movies等 - :param userid: 用户ID - :param username: 调用下载的用户名/插件名 - :param label: 自定义标签 - :param return_detail: 是否返回详细结果；False 时返回下载任务 hash 或 None，True 时返回 (hash, error_msg) - :param custom_words: 下载来源（如订阅）的完整自定义识别词文本，随下载记录存档，供整理时原样复现识别 - :param governance: 订阅级提交幂等、入口任务与取消检查；非订阅调用保持为空 - :return: return_detail=False 时返回下载任务 hash 或 None；return_detail=True 时返回 (hash, error_msg) - """ - _torrent = context.torrent_info - _media = context.media_info - _meta = context.meta_info - if _torrent is None or _media is None or _meta is None: - error_message = "下载上下文缺少媒体、元数据或种子信息" - return (None, error_message) if return_detail else None - _site_downloader = _torrent.site_downloader + def _execute_download_single( + self, + context: Context, + torrent_file: Optional[Path] = None, + torrent_content: Optional[Union[str, bytes]] = None, + episodes: Optional[Set[int]] = None, + channel: Optional[NotificationChannel] = None, + source: Optional[str] = None, + downloader: Optional[str] = None, + save_path: Optional[str] = None, + userid: Union[str, int, None] = None, + username: Optional[str] = None, + label: Optional[str] = None, + return_detail: bool = False, + custom_words: Optional[str] = None, + governance: Optional[SubscriptionDownloadGovernance] = None, + ) -> Union[Optional[str], Tuple[Optional[str], Optional[str]]]: + """准备下载事实，提交下载器并按订阅治理合同结算结果。""" + prepared, error_msg = self._prepare_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, + ) + if prepared is None: + return (None, error_msg) if return_detail else None + download_hash, error_msg = self._submit_prepared_download( + prepared=prepared, + context=context, + episodes=episodes, + channel=channel, + source=source, + downloader=downloader, + userid=userid, + username=username, + label=label, + custom_words=custom_words, + governance=governance, + ) + return (download_hash, error_msg) if return_detail else download_hash - # 下载目录和下载器分类依赖 TMDB 辅助分类，但媒体主身份保持不变。 - supplemented_media = MediaChain().supplement_tmdb_info(_media, _meta) + def _prepare_download_single( + self, + *, + context: Context, + torrent_file: Optional[Path], + torrent_content: Optional[Union[str, bytes]], + episodes: Optional[Set[int]], + channel: Optional[NotificationChannel], + source: Optional[str], + downloader: Optional[str], + save_path: Optional[str], + userid: Union[str, int, None], + username: Optional[str], + ) -> tuple[Optional[_PreparedDownload], Optional[str]]: + """补全媒体、读取种子并解析出可信下载目录。""" + torrent, media, meta = context.torrent_info, context.media_info, context.meta_info + if torrent is None or media is None or meta is None: + return None, "下载上下文缺少媒体、元数据或种子信息" + site_downloader = torrent.site_downloader + supplemented_media = MediaChain().supplement_tmdb_info(media, meta) if not isinstance(supplemented_media, (MediaInfo, MusicInfo)): - error_message = "媒体信息补全失败" - return (None, error_message) if return_detail else None - _media = supplemented_media - context.media_info = _media - + return None, "媒体信息补全失败" + media = supplemented_media + context.media_info = media save_path, event_error = self._apply_resource_download_event( - context, episodes, channel, source, downloader, save_path, - userid, username, + context, episodes, channel, source, downloader, save_path, userid, username ) if event_error: - return (None, event_error) if return_detail else None - - # 实际下载的集数 + return None, event_error download_episodes = episode_rules.format_ranges(list(episodes)) if episodes else None if episodes is not None: context.selected_episodes = sorted(set(episodes)) - elif _meta and _meta.episode_list: - context.selected_episodes = sorted(set(_meta.episode_list)) + elif meta.episode_list: + context.selected_episodes = sorted(set(meta.episode_list)) else: context.selected_episodes = [] - _folder_name = "" if not torrent_file and not torrent_content: - # 下载种子文件，得到的可能是文件也可能是磁力链 - torrent_content, _folder_name, _file_list = self.download_torrent(_torrent, - channel=channel, - source=source, - userid=userid) + torrent_content, _, _ = self.download_torrent( + torrent, channel=channel, source=source, userid=userid + ) elif torrent_file: - if torrent_file.exists(): - torrent_content = torrent_file.read_bytes() - else: - # 缓存处理器 - cache_backend = FileCache() - # 读取缓存的种子文件 - torrent_content = cache_backend.get(torrent_file.as_posix(), region="torrents") - + torrent_content = ( + torrent_file.read_bytes() + if torrent_file.exists() + else FileCache().get(torrent_file.as_posix(), region="torrents") + ) if not torrent_content: self._record_download_failure( context=context, error_msg="下载种子内容为空", - downloader=downloader or _site_downloader, + downloader=downloader or site_downloader, source=source, episodes=episodes, ) - return (None, "下载种子内容为空") if return_detail else None - - # 获取种子文件的文件夹名和文件清单 - _folder_name, _file_list = cast( - Any, TorrentHelper - )().get_fileinfo_from_torrent_content(torrent_content) - - album_validation_error = self._validate_music_album_resource(context, _file_list) - if album_validation_error: - logger.info(f"{_torrent.title} {album_validation_error}，跳过该资源") + return None, "下载种子内容为空" + folder_name, file_list = cast(Any, TorrentHelper)().get_fileinfo_from_torrent_content( + torrent_content + ) + album_error = self._validate_music_album_resource(context, file_list) + if album_error: + logger.info(f"{torrent.title} {album_error}，跳过该资源") self._record_download_failure( context=context, - error_msg=album_validation_error, - downloader=downloader or _site_downloader, + error_msg=album_error, + downloader=downloader or site_downloader, source=source, episodes=episodes, ) - return (None, album_validation_error) if return_detail else None - + return None, album_error storage, download_dir, error_msg = self._resolve_media_download_dir( - media_info=_media, + media_info=media, save_path=save_path, ) if not download_dir or not storage: if error_msg == "未找到下载目录": - self.messagehelper.put(f"{_media.type.value} {_media.title_year} 未找到下载目录！", - title="下载失败", role="system") - return (None, error_msg or "未找到下载目录") if return_detail else None - file_uri = FileURI(storage=storage, path=download_dir.as_posix()) - download_dir = Path(file_uri.uri) + self.messagehelper.put( + f"{media.type.value} {media.title_year} 未找到下载目录！", + title="下载失败", + role="system", + ) + return None, error_msg or "未找到下载目录" + download_uri = FileURI(storage=storage, path=download_dir.as_posix()).uri + return _PreparedDownload( + torrent=torrent, + media=media, + meta=meta, + torrent_content=torrent_content, + folder_name=folder_name, + file_list=file_list, + download_dir=Path(download_uri), + download_uri=download_uri, + download_episodes=download_episodes, + site_downloader=site_downloader, + ), None + def _submit_prepared_download( + self, + *, + prepared: _PreparedDownload, + context: Context, + episodes: Optional[Set[int]], + channel: Optional[NotificationChannel], + source: Optional[str], + downloader: Optional[str], + userid: Union[str, int, None], + username: Optional[str], + label: Optional[str], + custom_words: Optional[str], + governance: Optional[SubscriptionDownloadGovernance], + ) -> tuple[Optional[str], Optional[str]]: + """认领幂等提交权，调用下载器并分派成功或拒绝结算。""" if self._subscription_download_cancelled(governance): - cancel_error = "订阅下载在提交前已取消" - return (None, cancel_error) if return_detail else None + return None, "订阅下载在提交前已取消" admission, duplicate_hash = self._claim_subscription_download( context=context, episodes=episodes, governance=governance, - downloader=downloader or _site_downloader, - download_uri=file_uri.uri, + downloader=downloader or prepared.site_downloader, + download_uri=prepared.download_uri, ) if duplicate_hash: if governance and governance.mark_started: governance.mark_started() - logger.info(f"{_torrent.title} 已由重叠订阅入口提交，复用任务 {duplicate_hash}") - return (duplicate_hash, "下载任务已由重叠入口提交") if return_detail else duplicate_hash + logger.info(f"{prepared.torrent.title} 已由重叠订阅入口提交，复用任务 {duplicate_hash}") + return duplicate_hash, "下载任务已由重叠入口提交" if admission is not None and not admission.acquired: - wait_error = ( + return None, ( f"订阅下载提交当前为 {admission.snapshot.state}，" f"最早可重试：{admission.snapshot.available_at or '待下一轮'}" ) - return (None, wait_error) if return_detail else None attempt_token = admission.snapshot.attempt_token if admission is not None else None admission_key = admission.snapshot.idempotency_key if admission is not None else None if admission is not None and self._subscription_download_cancelled(governance): @@ -446,21 +505,18 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): idempotency_key=admission.snapshot.idempotency_key, attempt_token=admission.snapshot.attempt_token or "", ) - cancel_error = "订阅下载在下载器调用前已取消" - return (None, cancel_error) if return_detail else None - - # 添加下载 + return None, "订阅下载在下载器调用前已取消" if governance and governance.mark_started: governance.mark_started() try: - result: Optional[Tuple[Optional[str], Optional[str], Optional[str], str]] = self.download( - content=torrent_content, - cookie=_torrent.site_cookie, + result = self.download( + content=prepared.torrent_content, + cookie=prepared.torrent.site_cookie, episodes=cast(Set[int], episodes), - download_dir=download_dir, - category=_media.category, + download_dir=prepared.download_dir, + category=prepared.media.category, label=label, - downloader=downloader or _site_downloader, + downloader=downloader or prepared.site_downloader, ) except Exception as err: if admission_key and attempt_token: @@ -470,110 +526,173 @@ class DownloadSubmissionOwner(_DownloadOwnerBase): error=f"下载器调用异常：{str(err)}", ) raise DownloadReconciliationRequired( - f"{_torrent.title} 下载器结果不确定，已冻结自动重试" + f"{prepared.torrent.title} 下载器结果不确定，已冻结自动重试" ) from err raise - if result: - _downloader, _hash, _layout, error_msg = result + actual_downloader, download_hash, layout, error_msg = ( + result if result else (None, None, None, "未找到下载器") + ) + if download_hash: + self._settle_accepted_download( + prepared=prepared, + context=context, + episodes=episodes, + channel=channel, + source=source, + userid=userid, + username=username, + custom_words=custom_words, + actual_downloader=actual_downloader, + download_hash=download_hash, + layout=layout, + admission_key=admission_key, + attempt_token=attempt_token, + ) else: - _downloader, _hash, _layout, error_msg = None, None, None, "未找到下载器" + self._record_rejected_download( + prepared=prepared, + context=context, + episodes=episodes, + channel=channel, + source=source, + downloader=downloader, + userid=userid, + actual_downloader=actual_downloader, + error_msg=error_msg, + admission_key=admission_key, + attempt_token=attempt_token, + ) + return download_hash, error_msg - if _hash: + def _settle_accepted_download( + self, + *, + prepared: _PreparedDownload, + context: Context, + episodes: Optional[Set[int]], + channel: Optional[NotificationChannel], + source: Optional[str], + userid: Union[str, int, None], + username: Optional[str], + custom_words: Optional[str], + actual_downloader: Optional[str], + download_hash: str, + layout: Optional[str], + admission_key: Optional[str], + attempt_token: Optional[str], + ) -> None: + """持久化下载器接受事实，执行本地结算并确认成功终态。""" + if admission_key and attempt_token: + accepted = self._subscription_download_repository().mark_accepted( + idempotency_key=admission_key, + attempt_token=attempt_token, + downloader=actual_downloader, + download_hash=download_hash, + ) + if not accepted: + raise DownloadReconciliationRequired( + f"{prepared.torrent.title} 已被下载器接受，但本地接受状态写入失败" + ) + try: + self._settle_download_success( + context=context, + media=prepared.media, + meta=prepared.meta, + torrent=prepared.torrent, + folder_name=prepared.folder_name, + file_list=prepared.file_list, + download_dir=prepared.download_dir, + layout=layout, + downloader=actual_downloader, + download_hash=download_hash, + download_episodes=prepared.download_episodes, + episodes=episodes, + channel=channel, + source=source, + userid=userid, + username=username, + torrent_content=prepared.torrent_content, + custom_words=custom_words, + ) + except Exception as err: if admission_key and attempt_token: - accepted = self._subscription_download_repository().mark_accepted( + self._subscription_download_repository().mark_reconcile_required( idempotency_key=admission_key, attempt_token=attempt_token, - downloader=_downloader, - download_hash=_hash, + error=f"下载器已接受但本地结算失败：{str(err)}", + downloader=actual_downloader, + download_hash=download_hash, ) - if not accepted: - raise DownloadReconciliationRequired( - f"{_torrent.title} 已被下载器接受，但本地接受状态写入失败" - ) - try: - self._settle_download_success( - context=context, - media=_media, - meta=_meta, - torrent=_torrent, - folder_name=_folder_name, - file_list=_file_list, - download_dir=download_dir, - layout=_layout, - downloader=_downloader, - download_hash=_hash, - download_episodes=download_episodes, - episodes=episodes, - channel=channel, - source=source, - userid=userid, - username=username, - torrent_content=torrent_content, - custom_words=custom_words, - ) - except Exception as err: - if admission_key and attempt_token: - self._subscription_download_repository().mark_reconcile_required( - idempotency_key=admission_key, - attempt_token=attempt_token, - error=f"下载器已接受但本地结算失败：{str(err)}", - downloader=_downloader, - download_hash=_hash, - ) - raise DownloadReconciliationRequired( - f"{_torrent.title} 已被下载器接受但本地结算失败，已转待对账" - ) from err - raise - if admission_key and attempt_token: - succeeded = self._subscription_download_repository().mark_succeeded( + raise DownloadReconciliationRequired( + f"{prepared.torrent.title} 已被下载器接受但本地结算失败，已转待对账" + ) from err + raise + if admission_key and attempt_token: + succeeded = self._subscription_download_repository().mark_succeeded( + idempotency_key=admission_key, + attempt_token=attempt_token, + ) + if not succeeded: + self._subscription_download_repository().mark_reconcile_required( idempotency_key=admission_key, attempt_token=attempt_token, + error="本地结算完成但幂等成功终态写入失败", + downloader=actual_downloader, + download_hash=download_hash, ) - if not succeeded: - self._subscription_download_repository().mark_reconcile_required( - idempotency_key=admission_key, - attempt_token=attempt_token, - error="本地结算完成但幂等成功终态写入失败", - downloader=_downloader, - download_hash=_hash, - ) - raise DownloadReconciliationRequired( - f"{_torrent.title} 本地结算完成但提交终态未确认，已转待对账" - ) - else: - if admission_key and attempt_token: - retry_at = self._subscription_download_retry_at( + raise DownloadReconciliationRequired( + f"{prepared.torrent.title} 本地结算完成但提交终态未确认，已转待对账" + ) + + def _record_rejected_download( + self, + *, + prepared: _PreparedDownload, + context: Context, + episodes: Optional[Set[int]], + channel: Optional[NotificationChannel], + source: Optional[str], + downloader: Optional[str], + userid: Union[str, int, None], + actual_downloader: Optional[str], + error_msg: str, + admission_key: Optional[str], + attempt_token: Optional[str], + ) -> None: + """记录无外部副作用的明确拒绝，并通知原调用渠道。""" + if admission_key and attempt_token: + self._subscription_download_repository().mark_retryable( + idempotency_key=admission_key, + attempt_token=attempt_token, + available_at=self._subscription_download_retry_at( error_msg, self._download_failure_ttl(error_msg), - ) - self._subscription_download_repository().mark_retryable( - idempotency_key=admission_key, - attempt_token=attempt_token, - available_at=retry_at, - error=error_msg, - ) - # 下载失败 - logger.error(f"{_media.title_year} 添加下载任务失败：" - f"{_torrent.title} - {_torrent.enclosure}，{error_msg}") - self._record_download_failure( - context=context, - error_msg=error_msg, - downloader=_downloader or downloader or _site_downloader, - source=source, - episodes=episodes, + ), + error=error_msg, ) - # 只发送给对应渠道和用户 - self.post_message(Message( + logger.error( + f"{prepared.media.title_year} 添加下载任务失败：" + f"{prepared.torrent.title} - {prepared.torrent.enclosure}，{error_msg}" + ) + self._record_download_failure( + context=context, + error_msg=error_msg, + downloader=actual_downloader or downloader or prepared.site_downloader, + source=source, + episodes=episodes, + ) + self.post_message( + Message( channel=channel, source=source if channel else None, mtype=MessageType.Manual, - title="添加下载任务失败：%s %s" - % (_media.title_year, _meta.season_episode), - text=f"站点：{_torrent.site_name}\n" - f"种子名称：{_meta.org_string}\n" - f"错误信息：{error_msg}", - image=_media.get_message_image(), - userid=userid)) - if return_detail: - return _hash, error_msg - return _hash + title=f"添加下载任务失败：{prepared.media.title_year} {prepared.meta.season_episode}", + text=( + f"站点：{prepared.torrent.site_name}\n" + f"种子名称：{prepared.meta.org_string}\n" + f"错误信息：{error_msg}" + ), + image=prepared.media.get_message_image(), + userid=userid, + ) + ) diff --git a/app/chain/search/provider.py b/app/chain/search/provider.py index deb60cf3a..a95bd98c1 100644 --- a/app/chain/search/provider.py +++ b/app/chain/search/provider.py @@ -10,7 +10,7 @@ from datetime import datetime from typing import Any, AsyncGenerator, Awaitable, Callable, Dict, Iterator, List, Optional from app.application.configuration import get_configured_system_config -from app.application.site.search_observation import ( +from app.application.site.observation import ( capture_site_search_observation, report_site_search_outcome, ) @@ -107,8 +107,8 @@ class ProviderBatch: continued: bool -class SearchProviderOwner(_SearchOwnerBase): - """站点与插件资源 provider fan-out owner。""" +class _SearchProviderSyncOwner(_SearchOwnerBase): + """同步站点 provider 选择、预算与 fan-out owner。""" @staticmethod def _selected_site_ids(sites: Optional[List[int]]) -> List[int]: @@ -404,6 +404,10 @@ class SearchProviderOwner(_SearchOwnerBase): finally: progress.end() + +class SearchProviderOwner(_SearchProviderSyncOwner): + """异步站点与插件资源 provider fan-out owner。""" + async def _iter_provider_batches( self, *, diff --git a/app/chain/subscribe/contract.py b/app/chain/subscribe/contract.py index 6df68a62e..4675478e5 100644 --- a/app/chain/subscribe/contract.py +++ b/app/chain/subscribe/contract.py @@ -49,20 +49,28 @@ if TYPE_CHECKING: _SubscribeChain__notify_subscribe_create_failure: Callable[..., Any] _SubscribeChain__post_subscribe_added: Callable[..., Any] _SubscribeChain__prepare_best_version_tv_candidate: Callable[..., Any] + _SubscribeChain__prepare_subscribe_progress_fields: Callable[..., Any] _SubscribeChain__prepare_total_episode_change_fields: Callable[..., Any] _SubscribeChain__refresh_subscribe_progress_with_no_exists: Callable[..., Any] _SubscribeChain__refresh_total_episode_before_completion: Callable[..., Any] _SubscribeChain__report_completed: Callable[..., Any] + _SubscribeChain__resolve_total_episode_decrease: Callable[..., Any] _acquire_run_lock: Callable[..., Any] _async_recognize_music_subscribe: Callable[..., Awaitable[Any]] _get_pending_best_version_episodes: Callable[..., Any] _is_episode_range_covered: Callable[..., Any] _is_music_download_complete: Callable[..., Any] + _load_search_subscriptions: Callable[..., Any] _match_music_subscribe: Callable[..., Any] + _notify_manual_search: Callable[..., Any] + _process_search_subscription: Callable[..., Any] _recognize_music_subscribe: Callable[..., Any] + _report_search_progress: Callable[..., Any] _search_music_subscribe: Callable[..., Any] _subscription_query: Callable[..., Any] + _defer_recent_subscription: Callable[..., Any] _validate_music_subscribe_target: Callable[..., Any] + _wait_before_scheduled_search: Callable[..., Any] add: Callable[..., Any] async def async_obtain_images(self, mediainfo: MediaInfo) -> Optional[MediaInfo]: """异步补全媒体图片。""" diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index 77786eeb0..0a9f371fe 100644 --- a/app/chain/subscribe/match.py +++ b/app/chain/subscribe/match.py @@ -2,7 +2,7 @@ import copy from datetime import datetime -from typing import Callable, Dict, List, Optional +from typing import Any, Callable, Dict, List, Optional from app.application.configuration import get_configured_system_config from app.application.subscription.candidates import CandidateBatch, CandidateIndex @@ -29,8 +29,51 @@ from app.schemas.types import ( ) +def _prepare_subscription_match( + owner: Any, + subscribe: Any, + candidate_index: CandidateIndex, + fresh_fact_lease: FreshFactLease, +) -> Optional[ + tuple[MetaBase, MediaInfo, Dict[str, List[Context]], List[str], List[int]] +]: + """为单个订阅路由候选，并在本轮新鲜事实租约中加载媒体信息。""" + try: + meta = build_subscribe_meta(subscribe) + except ValueError: + logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}") + return None + domains = owner.site_repository.get_domains_by_ids(subscribe.sites) if subscribe.sites else [] + sub_sites = owner.get_sub_sites(subscribe) + routed_torrents = candidate_index.route_for_match( + subscribe, + domains=set(domains) if domains else None, + site_ids=set(sub_sites) if sub_sites else None, + ) + if not routed_torrents: + logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备") + return None + mediainfo = fresh_fact_lease.get_or_load( + subscribe, + lambda: MediaChain().recognize_media( + meta=meta, + mtype=meta.type, + **subscribe_recognize_kwargs(subscribe), + episode_group=subscribe.episode_group, + cache=False, + ), + ) + if not mediainfo: + logger.warning( + f"未识别到媒体信息，标题：{subscribe.name}，" + f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}" + ) + return None + return meta, mediainfo, routed_torrents, domains, sub_sites + + class SubscribeMatchOwner(_SubscribeOwnerBase): - """订阅候选准备、身份复核与资源匹配，作为 SubscribeChain 的单一职责实现 owner。""" + """订阅候选准备、身份复核与资源匹配 owner。""" def _prepare_match_torrents( self, @@ -142,8 +185,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): return processed_torrents = self._prepare_match_torrents(torrents) - candidate_index = CandidateIndex(processed_torrents) - fresh_fact_lease = FreshFactLease() + candidate_index, fresh_fact_lease = CandidateIndex(processed_torrents), FreshFactLease() # 所有订阅 subscribes = self.subscription_repository.list(self.get_states_for_search("R")) @@ -174,41 +216,12 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): self._match_music_subscribe(subscribe, music_contexts) continue mediakey = subscribe_media_key(subscribe) - try: - meta = build_subscribe_meta(subscribe) - except ValueError: - logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}") - continue - # 订阅的站点域名列表 - domains = [] - if subscribe.sites: - domains = self.site_repository.get_domains_by_ids(subscribe.sites) - sub_sites = self.get_sub_sites(subscribe) - routed_torrents = candidate_index.route_for_match( - subscribe, - domains=set(domains) if domains else None, - site_ids=set(sub_sites) if sub_sites else None, + prepared = _prepare_subscription_match( + self, subscribe, candidate_index, fresh_fact_lease ) - if not routed_torrents: - logger.info(f"订阅 {subscribe.name} 本轮没有可能相关的资源，跳过资源匹配准备") - continue - # 识别媒体信息 - mediainfo = fresh_fact_lease.get_or_load( - subscribe, - lambda: MediaChain().recognize_media( - meta=meta, - mtype=meta.type, - **subscribe_recognize_kwargs(subscribe), - episode_group=subscribe.episode_group, - cache=False, - ), - ) - if not mediainfo: - logger.warn( - f"未识别到媒体信息，标题：{subscribe.name}，" - f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}" - ) + if prepared is None: continue + meta, mediainfo, routed_torrents, domains, sub_sites = prepared # 如果媒体已存在或已下载完毕，跳过当前订阅处理 exist_flag, no_exists = self.check_and_handle_existing_media( @@ -505,7 +518,6 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): if lock_acquired: self._rlock.release() logger.debug(f"match Lock released at {datetime.now()}") - @staticmethod def _SubscribeChain__get_media_id_match_source(mediainfo: Optional[MediaInfo]) -> str: """ diff --git a/app/chain/subscribe/metadata.py b/app/chain/subscribe/metadata.py new file mode 100644 index 000000000..65fa853f3 --- /dev/null +++ b/app/chain/subscribe/metadata.py @@ -0,0 +1,144 @@ +"""订阅单条元数据刷新与完成对账协作者。""" + +from collections.abc import Callable +from typing import Any, Optional, cast + +from app.application.subscription.contract import ( + SubscriptionSnapshot, + build_subscribe_meta, + subscribe_media_key, +) +from app.application.subscription.facts import FreshFactLease +from app.chain.subscribe.contract import _SubscribeOwnerBase +from app.chain.subscribe.identity import subscribe_recognize_kwargs +from app.runtime.log import logger +from app.schemas.media import resolve_media_identity +from app.schemas.types import MUSIC_ENTITY_ALBUM, MediaType + + +class SubscribeMetadataOwner(_SubscribeOwnerBase): + """刷新一条订阅的新鲜媒体事实、持久字段与可选完成状态。""" + + def _check_subscription( + self, + subscribe: SubscriptionSnapshot, + fresh_fact_lease: FreshFactLease, + *, + reconcile_completion: bool, + media_chain_factory: Callable[[], Any], + ) -> Optional[SubscriptionSnapshot]: + """刷新单条订阅元数据，并按需复用同一事实执行完成对账。""" + try: + meta = build_subscribe_meta(subscribe) + except ValueError: + logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}") + return None + if meta.type == MediaType.MUSIC: + mediainfo = self._recognize_music_subscribe(subscribe) + else: + mediainfo = fresh_fact_lease.get_or_load( + subscribe, + lambda: media_chain_factory().recognize_media( + meta=meta, + mtype=meta.type, + **subscribe_recognize_kwargs(subscribe), + episode_group=subscribe.episode_group, + cache=False, + ), + ) + if not mediainfo: + logger.warning( + f"未识别到媒体信息，标题：{subscribe.name}，" + f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}" + ) + return None + episodes = ( + mediainfo.seasons.get(subscribe.season) or [] + if meta.type == MediaType.TV + else [] + ) + progress_update: dict[str, Any] + if ( + subscribe.type == MediaType.TV.value + and not subscribe.manual_total_episode + and episodes + ): + current_total_episode = len(episodes) + total_episode = self._SubscribeChain__apply_episodes_refresh( + current_total_episode, + season=subscribe.season, + mediainfo=mediainfo, + media_source=subscribe.media_source, + media_id=subscribe.media_id, + subscribe_id=subscribe.id, + scene="refresh", + ) + old_total_episode = subscribe.total_episode or 0 + if total_episode and total_episode < old_total_episode: + total_episode = self._SubscribeChain__resolve_total_episode_decrease( + subscribe=subscribe, + candidate_total=total_episode, + meta=meta, + mediainfo=mediainfo, + mediakey=subscribe_media_key(subscribe), + ) + if total_episode and total_episode != old_total_episode: + progress_update = self._SubscribeChain__prepare_total_episode_change_fields( + subscribe=subscribe, + total_episode=total_episode, + old_total_episode=old_total_episode, + ) + else: + total_episode = subscribe.total_episode + progress_update = {"lack_episode": subscribe.lack_episode} + if subscribe.best_version: + progress_update = self._SubscribeChain__prepare_subscribe_progress_fields( + subscribe=subscribe, + no_exists={}, + ) + logger.info( + f"订阅 {subscribe.name} 总集数变化，更新总集数为{total_episode}，" + f"缺失集数为{progress_update.get('lack_episode', subscribe.lack_episode)} ..." + ) + else: + total_episode = subscribe.total_episode + progress_update = {"lack_episode": subscribe.lack_episode} + if subscribe.best_version and subscribe.type == MediaType.TV.value: + progress_update = self._SubscribeChain__prepare_subscribe_progress_fields( + subscribe=subscribe, + no_exists={}, + ) + update_data = { + "name": mediainfo.title, + "year": str(mediainfo.year) if mediainfo.year is not None else None, + "vote": mediainfo.vote_average, + "poster": mediainfo.get_poster_image(), + "backdrop": mediainfo.get_backdrop_image(), + "description": mediainfo.overview, + "media_source": resolve_media_identity(media=mediainfo)[0], + "media_id": resolve_media_identity(media=mediainfo)[1], + "total_episode": total_episode, + } + if meta.type == MediaType.MUSIC: + music_type = getattr(mediainfo, "music_type", None) + update_data.update( + { + "music_type": music_type, + "total_tracks": getattr(mediainfo, "total_tracks", None) + if music_type == MUSIC_ENTITY_ALBUM + else None, + } + ) + update_data.update(progress_update) + updated = self._SubscribeChain__apply_subscribe_update( + subscribe, + update_data, + scene="metadata_refresh", + ) + if reconcile_completion and updated.state in self.get_states_for_search("R"): + self.reconcile_subscription_completion( + subscribe=updated, + meta=meta, + mediainfo=mediainfo, + ) + return cast(SubscriptionSnapshot, updated) diff --git a/app/chain/subscribe/policy.py b/app/chain/subscribe/policy.py index 253b3e606..4d32462a2 100644 --- a/app/chain/subscribe/policy.py +++ b/app/chain/subscribe/policy.py @@ -21,8 +21,8 @@ from app.schemas.types import ( ) -class SubscribePolicyOwner(_SubscribeOwnerBase): - """订阅优先级、剧集范围与来源编码策略，作为 SubscribeChain 的单一职责实现 owner。""" +class _SubscribePriorityPolicyOwner(_SubscribeOwnerBase): + """订阅优先级、剧集范围与来源编码策略 owner。""" @staticmethod def _SubscribeChain__normalize_episode_priority( @@ -179,6 +179,10 @@ class SubscribePolicyOwner(_SubscribeOwnerBase): ) return result + +class SubscribePolicyOwner(_SubscribePriorityPolicyOwner): + """订阅提交前复核与下载治理策略 owner。""" + @staticmethod def _SubscribeChain__get_downloaded_episodes(downloads: Optional[List[Context]]) -> List[int]: """获取本次下载实际涉及的剧集。""" diff --git a/app/chain/subscribe/refresh.py b/app/chain/subscribe/refresh.py index df7186846..eb85db836 100644 --- a/app/chain/subscribe/refresh.py +++ b/app/chain/subscribe/refresh.py @@ -5,6 +5,7 @@ from datetime import datetime from typing import Any, Callable, Dict, List, Optional, Tuple, Union, cast from app.application.subscription import priority as _priority +from app.application.subscription.candidates import CandidateBatch from app.application.subscription.contract import ( SubscriptionSnapshot, build_subscribe_meta, @@ -14,8 +15,8 @@ from app.application.subscription.contract import ( from app.application.subscription.facts import FreshFactLease from app.chain.download import DownloadChain from app.chain.media import MediaChain -from app.chain.subscribe.contract import _SubscribeOwnerBase from app.chain.subscribe.identity import subscribe_recognize_kwargs +from app.chain.subscribe.metadata import SubscribeMetadataOwner from app.chain.tmdb import TmdbChain from app.chain.torrents import TorrentsChain from app.domain.context import ( @@ -26,17 +27,15 @@ from app.runtime.events import eventmanager from app.runtime.log import logger from app.runtime.stop import runtime_stop_state from app.schemas.event import SubscribeEpisodesRefreshEventData -from app.schemas.media import resolve_media_identity from app.schemas.mediaserver import NotExistMediaInfo as _SchemaNotExistMediaInfo from app.schemas.types import ( - MUSIC_ENTITY_ALBUM, ChainEventType, MediaSource, MediaType, ) -class SubscribeRefreshOwner(_SubscribeOwnerBase): +class SubscribeRefreshOwner(SubscribeMetadataOwner): """订阅元数据、进度与剧集范围刷新，作为 SubscribeChain 的单一职责实现 owner。""" def refresh(self, progress_callback: Optional[Callable[..., None]] = None) -> None: @@ -78,12 +77,20 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase): data=data, ) - candidate_batch = TorrentsChain().refresh_batch( + torrents_chain = TorrentsChain() + candidate_batch = torrents_chain.refresh_batch( sites=sites, progress_callback=_update_refresh_progress if progress_callback else None, # 存在音乐订阅时额外抓取站点音乐专用入口，音乐不一定在默认种子首页 include_music=self.has_music_subscribe(), ) + if not isinstance(candidate_batch, CandidateBatch): + legacy_candidates = torrents_chain.refresh( + sites=sites, + progress_callback=_update_refresh_progress if progress_callback else None, + include_music=self.has_music_subscribe(), + ) + candidate_batch = CandidateBatch.from_legacy(legacy_candidates or {}, source="refresh") self.match_batch( candidate_batch, progress_callback=_update_match_progress if progress_callback else None, @@ -128,114 +135,15 @@ class SubscribeRefreshOwner(_SubscribeOwnerBase): "current": subscribe.id, }, ) - try: - meta = build_subscribe_meta(subscribe) - except ValueError: - logger.error(f"订阅 {subscribe.name} 类型错误：{subscribe.type}") - continue - # 识别媒体信息 - if meta.type == MediaType.MUSIC: - mediainfo = self._recognize_music_subscribe(subscribe) - else: - mediainfo = fresh_fact_lease.get_or_load( - subscribe, - lambda: MediaChain().recognize_media( - meta=meta, - mtype=meta.type, - **subscribe_recognize_kwargs(subscribe), - episode_group=subscribe.episode_group, - cache=False, - ), - ) - if not mediainfo: - logger.warn( - f"未识别到媒体信息，标题：{subscribe.name}，" - f"媒体来源：{subscribe.media_source}，媒体 ID：{subscribe.media_id}" - ) - continue - # 对于电视剧，获取当前季的总集数 - episodes = (mediainfo.seasons.get(subscribe.season) or []) if meta.type == MediaType.TV else [] - progress_update = {} - if subscribe.type == MediaType.TV.value and not subscribe.manual_total_episode and len(episodes): - current_total_episode = len(episodes) - # 外部事件只能向上覆盖主程序本次识别到的 TMDB 当前季总集数，已有订阅按最终 total 跟随持久化。 - total_episode = self._SubscribeChain__apply_episodes_refresh( - current_total_episode, - season=subscribe.season, - mediainfo=mediainfo, - media_source=subscribe.media_source, - media_id=subscribe.media_id, - subscribe_id=subscribe.id, - scene="refresh", - ) - old_total_episode = subscribe.total_episode or 0 - if total_episode and total_episode < old_total_episode: - total_episode = self._SubscribeChain__resolve_total_episode_decrease( - subscribe=subscribe, - candidate_total=total_episode, - meta=meta, - mediainfo=mediainfo, - mediakey=subscribe_media_key(subscribe), - ) - if total_episode and total_episode != old_total_episode: - progress_update = self._SubscribeChain__prepare_total_episode_change_fields( - subscribe=subscribe, - total_episode=total_episode, - old_total_episode=old_total_episode, - ) - else: - total_episode = subscribe.total_episode - progress_update = {"lack_episode": subscribe.lack_episode} - if subscribe.best_version and subscribe.type == MediaType.TV.value: - progress_update = self._SubscribeChain__prepare_subscribe_progress_fields( - subscribe=subscribe, no_exists={} - ) - logger.info( - f"订阅 {subscribe.name} 总集数变化，更新总集数为{total_episode}，" - f"缺失集数为{progress_update.get('lack_episode', subscribe.lack_episode)} ..." - ) - else: - total_episode = subscribe.total_episode - progress_update = {"lack_episode": subscribe.lack_episode} - if subscribe.best_version and subscribe.type == MediaType.TV.value: - progress_update = self._SubscribeChain__prepare_subscribe_progress_fields( - subscribe=subscribe, no_exists={} - ) - # 更新TMDB信息 - update_data = { - "name": mediainfo.title, - "year": str(mediainfo.year) if mediainfo.year is not None else None, - "vote": mediainfo.vote_average, - "poster": mediainfo.get_poster_image(), - "backdrop": mediainfo.get_backdrop_image(), - "description": mediainfo.overview, - "media_source": resolve_media_identity(media=mediainfo)[0], - "media_id": resolve_media_identity(media=mediainfo)[1], - "total_episode": total_episode, - } - if meta.type == MediaType.MUSIC: - music_type = getattr(mediainfo, "music_type", None) - update_data.update( - { - "music_type": music_type, - "total_tracks": getattr(mediainfo, "total_tracks", None) - if music_type == MUSIC_ENTITY_ALBUM - else None, - } - ) - update_data.update(progress_update) - subscribe = self._SubscribeChain__apply_subscribe_update( + updated_subscribe = self._check_subscription( subscribe, - update_data, - scene="metadata_refresh", + fresh_fact_lease, + reconcile_completion=reconcile_completion, + media_chain_factory=MediaChain, ) - if reconcile_completion and subscribe.state in self.get_states_for_search("R"): - self.reconcile_subscription_completion( - subscribe=subscribe, - meta=meta, - mediainfo=mediainfo, - ) - logger.info(f"{subscribe.name} 订阅元数据更新完成") + if updated_subscribe is None: + continue + logger.info(f"{updated_subscribe.name} 订阅元数据更新完成") if progress_callback: progress_callback( value=index / total_num * 100 if total_num else 100, diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index ad928da24..f728ca8fb 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -36,8 +36,8 @@ from app.schemas.types import ( ) -class SubscribeSearchOwner(_SubscribeOwnerBase): - """订阅主动搜索编排，作为 SubscribeChain 的单一职责实现 owner。""" +class _SubscribeSearchQueueOwner(_SubscribeOwnerBase): + """订阅主动搜索入口与持久队列消费 owner。""" def _subscription_query(self) -> SubscriptionQueryService: """构造绑定订阅 Oper 的查询应用服务。""" @@ -481,6 +481,10 @@ class SubscribeSearchOwner(_SubscribeOwnerBase): ) return queue.get_batch(batch_id) if queue else None + +class SubscribeSearchOwner(_SubscribeSearchQueueOwner): + """订阅搜索加载、单项处理与结果通知 owner。""" + def _load_search_subscriptions( self, sid: Optional[int], diff --git a/app/db/oper/subscriptiondownload.py b/app/db/oper/subscriptiondownload.py index 1d5e32272..bbfd9efad 100644 --- a/app/db/oper/subscriptiondownload.py +++ b/app/db/oper/subscriptiondownload.py @@ -1,7 +1,7 @@ """订阅下载提交账本的事务内状态转换。""" from datetime import datetime, timezone -from typing import Optional +from typing import Optional, cast from uuid import uuid4 from sqlalchemy import and_, or_, select, update @@ -106,11 +106,11 @@ class SubscriptionDownloadOper(DbOper): """按稳定幂等键读取提交记录。""" if not isinstance(self._db, Session): raise RuntimeError("订阅下载提交查询需要调用方提供同步 Session") - return self._db.execute( + return cast(Optional[SubscriptionDownloadSubmission], self._db.execute( select(SubscriptionDownloadSubmission).where( SubscriptionDownloadSubmission.idempotency_key == idempotency_key ) - ).scalars().first() + ).scalars().first()) def mark_accepted( self, diff --git a/app/locales/en-US.json b/app/locales/en-US.json index c44044f95..768e7980f 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -346,6 +346,8 @@ "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道": "WeChat ClawBot notification is not enabled or the configuration has not been saved. Please save and enable this channel first", "请输入至少一个有效的站点 ID": "Enter at least one valid site ID", "所有订阅搜索完成": "All subscription searches are complete", + "订阅搜索批次不存在": "The subscription search batch does not exist", + "订阅搜索批次已结束或无法取消": "The subscription search batch has ended or cannot be cancelled", "订阅搜索锁等待超时，已跳过本轮": "Subscription search lock timed out, this round was skipped", "订阅匹配锁等待超时，已跳过本轮": "Subscription matching lock timed out, this round was skipped", "请输入订阅 ID，多个 ID 用空格分隔，或输入 all": "Enter subscription IDs separated by spaces, or enter all", diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index 7e1fad86d..e0d5792dc 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -342,6 +342,8 @@ "微信 ClawBot 通知未启用或配置尚未保存，请先保存并启用当前渠道": "微信 ClawBot 通知未啟用或設定尚未儲存，請先儲存並啟用目前渠道", "请输入至少一个有效的站点 ID": "請輸入至少一個有效的站點 ID", "所有订阅搜索完成": "所有訂閱搜尋完成", + "订阅搜索批次不存在": "訂閱搜尋批次不存在", + "订阅搜索批次已结束或无法取消": "訂閱搜尋批次已結束或無法取消", "请输入订阅 ID，多个 ID 用空格分隔，或输入 all": "請輸入訂閱 ID，多個 ID 以空格分隔，或輸入 all", "请输入至少一个有效的订阅 ID": "請輸入至少一個有效的訂閱 ID", "格式错误，请输入：cookie [2fa_code/secret]": "格式錯誤，請輸入：cookie [2fa_code/secret]", diff --git a/app/modules/indexer/__init__.py b/app/modules/indexer/__init__.py index 9a0b29660..f7c320c81 100644 --- a/app/modules/indexer/__init__.py +++ b/app/modules/indexer/__init__.py @@ -4,8 +4,8 @@ from types import MappingProxyType from typing import Any, Callable, List, Mapping, Optional, Tuple, Union, cast from app.application.site.health import get_configured_site_health_service +from app.application.site.observation import report_site_search_outcome from app.application.site.query import get_configured_site_query_service -from app.application.site.search_observation import report_site_search_outcome from app.application.site.sites import SitesHelper # pylint: disable=import-error,no-name-in-module from app.domain import site as site_rules from app.domain.context import Context, SubtitleInfo, TorrentInfo diff --git a/app/scheduler/catalog.py b/app/scheduler/catalog.py index c3d4b0887..11b2ca0eb 100644 --- a/app/scheduler/catalog.py +++ b/app/scheduler/catalog.py @@ -23,6 +23,7 @@ from app.application.scheduling import ( # noqa: E402 ) from app.runtime.scheduling import TimerUtils from app.scheduler.contract import _SchedulerOwnerBase +from app.scheduler.services import SchedulerServices from app.schemas.system import MediaServerConf as _SchemaMediaServerConf @@ -35,6 +36,29 @@ class _MediaServerSchedule(TypedDict): interval: int +def _subscription_search_job_specs(services: SchedulerServices) -> tuple[JobSpec, ...]: + """构造订阅搜索、新增搜索与持久队列恢复任务目录。""" + return ( + JobSpec( + "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription", kwargs={"state": "R"} + ), + JobSpec( + "new_subscribe_search", + "新增订阅搜索", + services.search_subscribe, + "subscription", + kwargs={"state": "N"}, + ), + JobSpec( + "subscribe_search_queue", + "恢复订阅搜索队列", + services.resume_subscribe_search, + "subscription", + recovery=JobRecoveryPolicy.DURABLE_QUEUE, + ), + ) + + class SchedulerCatalogOwner(_SchedulerOwnerBase): """调度器静态任务目录与 APScheduler 投影。""" @@ -123,6 +147,18 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): replace_existing=True, ) + def _register_subscription_search_queue_job(self, config: SchedulerRuntimeConfig) -> None: + """注册短周期持久搜索队列恢复任务。""" + self._scheduler.add_job( + self.start, + "interval", + id="subscribe_search_queue", + name="恢复订阅搜索队列", + minutes=1, + next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=10), + kwargs={"job_id": "subscribe_search_queue"}, + ) + def _initialize_catalog(self, config: SchedulerRuntimeConfig) -> None: """构建完整任务目录并投影到尚未启动的 APScheduler。""" services = self._scheduler_services() @@ -132,23 +168,7 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): JobSpec("cookiecloud", "同步CookieCloud站点", services.sync_cookies, "site"), JobSpec("mediaserver_sync", "同步媒体服务器", services.sync_mediaserver, "mediaserver"), JobSpec("subscribe_tmdb", "订阅元数据更新", services.check_subscribe, "subscription"), - JobSpec( - "subscribe_search", "订阅搜索补全", services.search_subscribe, "subscription", kwargs={"state": "R"} - ), - JobSpec( - "new_subscribe_search", - "新增订阅搜索", - services.search_subscribe, - "subscription", - kwargs={"state": "N"}, - ), - JobSpec( - "subscribe_search_queue", - "恢复订阅搜索队列", - services.resume_subscribe_search, - "subscription", - recovery=JobRecoveryPolicy.DURABLE_QUEUE, - ), + *_subscription_search_job_specs(services), JobSpec("subscribe_refresh", "订阅刷新", services.refresh_subscribe, "subscription"), JobSpec("subscribe_follow", "关注的订阅分享", services.follow_subscribe, "subscription"), JobSpec( @@ -253,15 +273,7 @@ class SchedulerCatalogOwner(_SchedulerOwnerBase): ) # 新增订阅时搜索（5分钟检查一次） - self._scheduler.add_job( - self.start, - "interval", - id="subscribe_search_queue", - name="恢复订阅搜索队列", - minutes=1, - next_run_time=datetime.now(pytz.timezone(config.timezone)) + timedelta(seconds=10), - kwargs={"job_id": "subscribe_search_queue"}, - ) + self._register_subscription_search_queue_job(config) self._scheduler.add_job( self.start, diff --git a/app/schemas/exports.py b/app/schemas/exports.py index 7216e543c..a9741953b 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -396,6 +396,8 @@ SCHEMA_EXPORTS = { 'SubscribeShare': ('app.schemas.subscribe', 'SubscribeShare'), 'SubscribeShareStatistics': ('app.schemas.subscribe', 'SubscribeShareStatistics'), 'Subscription': ('app.schemas.message', 'Subscription'), + 'SubscriptionBatchStatus': ('app.schemas.subscribe', 'SubscriptionBatchStatus'), + 'SubscriptionExecutionStatus': ('app.schemas.subscribe', 'SubscriptionExecutionStatus'), 'SubscriptionMessage': ('app.schemas.message', 'SubscriptionMessage'), 'SubtitleDownloadData': ('app.schemas.download', 'SubtitleDownloadData'), 'SubtitleInfo': ('app.schemas.search', 'SubtitleInfo'), diff --git a/app/schemas/subscribe.py b/app/schemas/subscribe.py index de5c5bcbb..44c2e06ff 100644 --- a/app/schemas/subscribe.py +++ b/app/schemas/subscribe.py @@ -45,7 +45,7 @@ def compute_subscribe_completed_episode(subscribe: "Subscribe") -> Optional[int] return min(max(start_episode - 1, 0), total_episode) + priority_completed -class SubscriptionExecutionStatus(BaseModel): +class SubscriptionExecutionStatus(BaseModel): # type: ignore[misc] """订阅列表可见的当前业务执行状态。""" state: str @@ -63,7 +63,7 @@ class SubscriptionExecutionStatus(BaseModel): model_config = ConfigDict(from_attributes=True) -class SubscriptionBatchStatus(BaseModel): +class SubscriptionBatchStatus(BaseModel): # type: ignore[misc] """订阅搜索批次的用户可见进度和操作能力。""" batch_id: str diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index b556fbd6c..0db30930d 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -754,8 +754,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 932 | -| 内部导入边 | 7,784 | +| Python 模块 | 936 | +| 内部导入边 | 7,835 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/agent-api-surface-audit.json b/docs/architecture/agent-api-surface-audit.json index d83a65e60..0760bf429 100644 --- a/docs/architecture/agent-api-surface-audit.json +++ b/docs/architecture/agent-api-surface-audit.json @@ -6,7 +6,7 @@ "provider-skill": 11, "stream_or_binary": 10, "transport_or_identity": 66, - "ui_presentation": 5 + "ui_presentation": 8 }, "dynamic_gateway_routes": [ { @@ -21,7 +21,7 @@ "gateway_http_route_count": 201, "gateway_operation_count": 203, "matched_gateway_http_route_count": 200, - "openapi_operation_count": 375, + "openapi_operation_count": 378, "operations": [ { "disposition": "consolidated", @@ -3367,6 +3367,42 @@ "subscribe" ] }, + { + "disposition": "ui_presentation", + "method": "GET", + "operation_ids": [], + "owner": "host-ui", + "path": "/api/v1/subscribe/execution/batches", + "reason": "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.", + "summary": "查询订阅搜索批次状态", + "tags": [ + "subscribe" + ] + }, + { + "disposition": "ui_presentation", + "method": "GET", + "operation_ids": [], + "owner": "host-ui", + "path": "/api/v1/subscribe/execution/batches/{batch_id}", + "reason": "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.", + "summary": "查询订阅搜索批次", + "tags": [ + "subscribe" + ] + }, + { + "disposition": "ui_presentation", + "method": "PUT", + "operation_ids": [], + "owner": "host-ui", + "path": "/api/v1/subscribe/execution/batches/{batch_id}/cancel", + "reason": "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.", + "summary": "取消订阅搜索批次", + "tags": [ + "subscribe" + ] + }, { "disposition": "gateway", "method": "GET", diff --git a/docs/architecture/agent-api-surface-audit.md b/docs/architecture/agent-api-surface-audit.md index 7c2affa13..657d3f20f 100644 --- a/docs/architecture/agent-api-surface-audit.md +++ b/docs/architecture/agent-api-surface-audit.md @@ -5,7 +5,7 @@ ## Result -- OpenAPI HTTP operations: **375** +- OpenAPI HTTP operations: **378** - Stable `moviepilot_api` operations: **203** - Exact HTTP routes used by the gateway: **201** - OpenAPI routes matched directly by the gateway: **200** @@ -23,7 +23,7 @@ | `provider-skill` | 11 | Low-level downloader or media-server capability owned by a provider Skill. | | `stream_or_binary` | 10 | Streaming or binary response owned by a direct client transport. | | `transport_or_identity` | 66 | Authentication, protocol, callback, account, or conversation transport boundary. | -| `ui_presentation` | 5 | Frontend or plugin-rendered presentation contract. | +| `ui_presentation` | 8 | Frontend or plugin-rendered presentation contract. | ## Bounded Dynamic Routes @@ -291,6 +291,9 @@ | `POST` | `/api/v1/subscribe/` | subscribe | `gateway` | subscription.add | 新增订阅 | | `PUT` | `/api/v1/subscribe/` | subscribe | `gateway` | subscription.update | 更新订阅 | | `GET` | `/api/v1/subscribe/check` | subscribe | `gateway` | subscription.metadata.refresh | 刷新订阅 TMDB 信息 | +| `GET` | `/api/v1/subscribe/execution/batches` | subscribe | `ui_presentation` | host-ui | 查询订阅搜索批次状态 | +| `GET` | `/api/v1/subscribe/execution/batches/{batch_id}` | subscribe | `ui_presentation` | host-ui | 查询订阅搜索批次 | +| `PUT` | `/api/v1/subscribe/execution/batches/{batch_id}/cancel` | subscribe | `ui_presentation` | host-ui | 取消订阅搜索批次 | | `GET` | `/api/v1/subscribe/files/{subscribe_id}` | subscribe | `gateway` | subscription.files | 订阅相关文件信息 | | `DELETE` | `/api/v1/subscribe/follow` | subscribe | `gateway` | subscription.follow.delete | 取消Follow订阅分享人 | | `GET` | `/api/v1/subscribe/follow` | subscribe | `gateway` | subscription.follow.list | 查询已Follow的订阅分享人 | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 86064ddb4..267444baa 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 932 / 7,784 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 936 / 7,835 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -102,8 +102,8 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 | -| 全量 mypy 历史债务 | 9,589 / 517 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | -| Ruff 历史诊断 | 568 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | +| 全量 mypy 历史债务 | 9,588 / 517 文件 | Agent API 重构后的现状基线；canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | +| Ruff 历史诊断 | 567 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | ### 3.3 热点文件 diff --git a/scripts/generate_agent_api_surface_audit.py b/scripts/generate_agent_api_surface_audit.py index 1ddba8850..dc61f796f 100644 --- a/scripts/generate_agent_api_surface_audit.py +++ b/scripts/generate_agent_api_surface_audit.py @@ -106,6 +106,7 @@ EXPLICIT_TRANSPORT_PATHS = frozenset( "/api/v1/system/ping", } ) +SUBSCRIPTION_EXECUTION_UI_PREFIX = "/api/v1/subscribe/execution/" def _gateway_routes() -> dict[tuple[str, str], list[str]]: @@ -162,6 +163,13 @@ def _classify( "Plugin-rendered page, dashboard, or navigation metadata owned by the frontend presentation contract rather than an Agent business action.", [], ) + if path.startswith(SUBSCRIPTION_EXECUTION_UI_PREFIX): + return ( + "ui_presentation", + "host-ui", + "Background subscription execution status and cancellation are owned by the authenticated frontend workflow; they are not yet a stable Agent gateway contract.", + [], + ) if path in EXPLICIT_TRANSPORT_PATHS: return ( "transport_or_identity", diff --git a/scripts/generate_agent_skill_docs.py b/scripts/generate_agent_skill_docs.py index 661598241..efb089354 100644 --- a/scripts/generate_agent_skill_docs.py +++ b/scripts/generate_agent_skill_docs.py @@ -116,6 +116,26 @@ DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = { "Auditing historical subscriptions, media identity, completion criteria, and filter configuration.", "Generated by subscription completion and archival; restore or delete through its business API.", ), + "subscriptiondownloadsubmission": ( + "Stores subscription download idempotency claims, downloader acceptance facts, retries, and reconciliation freezes.", + "Diagnosing duplicate suppression, uncertain downloader outcomes, retry timing, and task ownership.", + "Owned by the subscription download submission state machine; never force accepted, succeeded, or retryable states.", + ), + "subscriptionsearchbatch": ( + "Stores durable subscription search batches, source, aggregate state, counts, and cancellation requests.", + "Inspecting user-visible search progress, recovery state, cancellation, and terminal outcomes.", + "Owned by subscription search orchestration; create and cancel batches through the subscription API.", + ), + "subscriptionsearchtask": ( + "Stores one durable subscription search task per batch and subscription with leases and execution phases.", + "Diagnosing queued, running, failed, cancelled, or recovered work and its current site.", + "Advanced only by the search queue lease state machine; never rewrite leases or terminal states manually.", + ), + "subscriptionsitebudget": ( + "Stores per-site subscription search concurrency, cooldown, health, and fairness state.", + "Diagnosing site pressure, cooldown deferrals, recent failures, and active search ownership.", + "Owned by the subscription site-budget coordinator; do not clear cooldowns or counters by direct SQL.", + ), "systemconfig": ( "Stores JSON business configuration values keyed by SystemConfigKey.", "Verifying the physical value only when the managed settings API behaves unexpectedly.", diff --git a/skills/database-operation/SKILL.md b/skills/database-operation/SKILL.md index 7e0ad6c59..06f834d3c 100644 --- a/skills/database-operation/SKILL.md +++ b/skills/database-operation/SKILL.md @@ -236,6 +236,30 @@ python scripts/mp-db.py write "UPDATE subscribe SET state = 'S' WHERE id = 123" - Write boundary: Generated by subscription completion and archival; restore or delete through its business API. - Columns: `id`, `name`, `year`, `type`, `keyword`, `media_source`, `media_id`, `music_type`, `total_tracks`, `season`, `poster`, `backdrop`, `vote`, `description`, `filter`, `include`, `exclude`, `quality`, `resolution`, `effect`, `audio_quality`, `audio_format`, `min_bitrate`, `min_bit_depth`, `min_sample_rate`, `total_episode`, `start_episode`, `date`, `username`, `sites`, `best_version`, `best_version_full`, `current_priority`, `current_audio_format`, `current_bitrate`, `current_bit_depth`, `current_sample_rate`, `episode_priority`, `save_path`, `search_imdbid`, `custom_words`, `media_category`, `filter_groups`, `episode_group` +### `subscriptiondownloadsubmission` +- Purpose: Stores subscription download idempotency claims, downloader acceptance facts, retries, and reconciliation freezes. +- Useful queries: Diagnosing duplicate suppression, uncertain downloader outcomes, retry timing, and task ownership. +- Write boundary: Owned by the subscription download submission state machine; never force accepted, succeeded, or retryable states. +- Columns: `id`, `idempotency_key`, `subscription_id`, `task_id`, `logical_identity`, `resource_key`, `coverage`, `mode`, `delivery_scope`, `state`, `attempt_count`, `attempt_token`, `downloader`, `download_hash`, `available_at`, `last_error`, `created_at`, `updated_at`, `started_at`, `finished_at` + +### `subscriptionsearchbatch` +- Purpose: Stores durable subscription search batches, source, aggregate state, counts, and cancellation requests. +- Useful queries: Inspecting user-visible search progress, recovery state, cancellation, and terminal outcomes. +- Write boundary: Owned by subscription search orchestration; create and cancel batches through the subscription API. +- Columns: `id`, `batch_id`, `source`, `state`, `priority`, `total_count`, `finished_count`, `failed_count`, `cancelled_count`, `cancel_requested`, `created_at`, `updated_at`, `started_at`, `finished_at`, `last_error` + +### `subscriptionsearchtask` +- Purpose: Stores one durable subscription search task per batch and subscription with leases and execution phases. +- Useful queries: Diagnosing queued, running, failed, cancelled, or recovered work and its current site. +- Write boundary: Advanced only by the search queue lease state machine; never rewrite leases or terminal states manually. +- Columns: `id`, `task_id`, `batch_id`, `subscription_id`, `active_key`, `source`, `priority`, `position`, `state`, `phase`, `current_site_id`, `attempt_count`, `cancel_requested`, `lease_owner`, `lease_token`, `lease_expires_at`, `available_at`, `created_at`, `updated_at`, `started_at`, `finished_at`, `last_error` + +### `subscriptionsitebudget` +- Purpose: Stores per-site subscription search concurrency, cooldown, health, and fairness state. +- Useful queries: Diagnosing site pressure, cooldown deferrals, recent failures, and active search ownership. +- Write boundary: Owned by the subscription site-budget coordinator; do not clear cooldowns or counters by direct SQL. +- Columns: `id`, `site_id`, `lease_owner`, `lease_token`, `lease_expires_at`, `next_allowed_at`, `consecutive_failures`, `success_streak`, `last_outcome`, `last_error`, `updated_at` + ### `systemconfig` - Purpose: Stores JSON business configuration values keyed by SystemConfigKey. - Useful queries: Verifying the physical value only when the managed settings API behaves unexpectedly. diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 1b8d20386..02c65b25e 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 7784, - "edge_sha256": "602f1b401b10d5a26a0782fee6ca31b1276dd9861ec9006be35d1372c4974e59", + "edge_count": 7835, + "edge_sha256": "491127de88d2727e18db050dba310fd4df5a0fe32c67f4976c9f6d1145e2e53f", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -1997,10 +1997,12 @@ "app.api.dependencies.subscription -> app.application.subscription", "app.api.dependencies.subscription -> app.application.subscription.contract", "app.api.dependencies.subscription -> app.application.subscription.delete", + "app.api.dependencies.subscription -> app.application.subscription.execution", "app.api.dependencies.subscription -> app.application.subscription.identity", "app.api.dependencies.subscription -> app.application.subscription.mutation", "app.api.dependencies.subscription -> app.application.subscription.query", "app.api.dependencies.subscription -> app.application.subscription.search", + "app.api.dependencies.subscription -> app.application.subscription.status", "app.api.dependencies.subscription -> app.application.subscription.write", "app.api.dependencies.subscription -> app.runtime", "app.api.dependencies.subscription -> app.runtime.events", @@ -2631,6 +2633,22 @@ "app.api.endpoints.storage -> app.schemas.system", "app.api.endpoints.storage -> app.schemas.types", "app.api.endpoints.storage -> app.schemas.workflow", + "app.api.endpoints.subexecution -> app.api", + "app.api.endpoints.subexecution -> app.api.dependencies", + "app.api.endpoints.subexecution -> app.api.dependencies.auth", + "app.api.endpoints.subexecution -> app.api.dependencies.subscription", + "app.api.endpoints.subexecution -> app.api.principal", + "app.api.endpoints.subexecution -> app.api.response", + "app.api.endpoints.subexecution -> app.application", + "app.api.endpoints.subexecution -> app.application.subscription", + "app.api.endpoints.subexecution -> app.application.subscription.execution", + "app.api.endpoints.subexecution -> app.application.subscription.query", + "app.api.endpoints.subexecution -> app.application.subscription.status", + "app.api.endpoints.subexecution -> app.runtime", + "app.api.endpoints.subexecution -> app.runtime.execution", + "app.api.endpoints.subexecution -> app.schemas", + "app.api.endpoints.subexecution -> app.schemas.response", + "app.api.endpoints.subexecution -> app.schemas.subscribe", "app.api.endpoints.subscribe -> app.adapters", "app.api.endpoints.subscribe -> app.adapters.external", "app.api.endpoints.subscribe -> app.adapters.external.server", @@ -2654,6 +2672,7 @@ "app.api.endpoints.subscribe -> app.application.subscription.mutation", "app.api.endpoints.subscribe -> app.application.subscription.query", "app.api.endpoints.subscribe -> app.application.subscription.search", + "app.api.endpoints.subscribe -> app.application.subscription.status", "app.api.endpoints.subscribe -> app.chain", "app.api.endpoints.subscribe -> app.chain.subscribe", "app.api.endpoints.subscribe -> app.chain.subscribe.facade", @@ -2867,6 +2886,7 @@ "app.api.routers -> app.api.endpoints.search", "app.api.routers -> app.api.endpoints.site", "app.api.routers -> app.api.endpoints.storage", + "app.api.routers -> app.api.endpoints.subexecution", "app.api.routers -> app.api.endpoints.subscribe", "app.api.routers -> app.api.endpoints.system", "app.api.routers -> app.api.endpoints.tmdb", @@ -3497,9 +3517,14 @@ "app.application.subscription.search -> app.application.subscription.contract", "app.application.subscription.sitebudget -> app.application", "app.application.subscription.sitebudget -> app.application.site", - "app.application.subscription.sitebudget -> app.application.site.search_observation", + "app.application.subscription.sitebudget -> app.application.site.observation", "app.application.subscription.sitebudget -> app.runtime", "app.application.subscription.sitebudget -> app.runtime.stop", + "app.application.subscription.status -> app.application", + "app.application.subscription.status -> app.application.download", + "app.application.subscription.status -> app.application.download.admission", + "app.application.subscription.status -> app.application.subscription", + "app.application.subscription.status -> app.application.subscription.execution", "app.application.subscription.write -> app.application", "app.application.subscription.write -> app.application.outbox", "app.application.subscription.write -> app.application.subscription", @@ -3842,6 +3867,8 @@ "app.chain.download.submission -> app.domain", "app.chain.download.submission -> app.domain.context", "app.chain.download.submission -> app.domain.episode", + "app.chain.download.submission -> app.domain.meta", + "app.chain.download.submission -> app.domain.meta.metabase", "app.chain.download.submission -> app.runtime", "app.chain.download.submission -> app.runtime.cache", "app.chain.download.submission -> app.runtime.events", @@ -4274,7 +4301,7 @@ "app.chain.search.provider -> app.application", "app.chain.search.provider -> app.application.configuration", "app.chain.search.provider -> app.application.site", - "app.chain.search.provider -> app.application.site.search_observation", + "app.chain.search.provider -> app.application.site.observation", "app.chain.search.provider -> app.application.subscription", "app.chain.search.provider -> app.application.subscription.sitebudget", "app.chain.search.provider -> app.chain", @@ -4517,6 +4544,19 @@ "app.chain.subscribe.match -> app.schemas", "app.chain.subscribe.match -> app.schemas.media", "app.chain.subscribe.match -> app.schemas.types", + "app.chain.subscribe.metadata -> app.application", + "app.chain.subscribe.metadata -> app.application.subscription", + "app.chain.subscribe.metadata -> app.application.subscription.contract", + "app.chain.subscribe.metadata -> app.application.subscription.facts", + "app.chain.subscribe.metadata -> app.chain", + "app.chain.subscribe.metadata -> app.chain.subscribe", + "app.chain.subscribe.metadata -> app.chain.subscribe.contract", + "app.chain.subscribe.metadata -> app.chain.subscribe.identity", + "app.chain.subscribe.metadata -> app.runtime", + "app.chain.subscribe.metadata -> app.runtime.log", + "app.chain.subscribe.metadata -> app.schemas", + "app.chain.subscribe.metadata -> app.schemas.media", + "app.chain.subscribe.metadata -> app.schemas.types", "app.chain.subscribe.notify -> app.application", "app.chain.subscribe.notify -> app.application.configuration", "app.chain.subscribe.notify -> app.application.messaging", @@ -4591,6 +4631,7 @@ "app.chain.subscribe.reconcile -> app.schemas.types", "app.chain.subscribe.refresh -> app.application", "app.chain.subscribe.refresh -> app.application.subscription", + "app.chain.subscribe.refresh -> app.application.subscription.candidates", "app.chain.subscribe.refresh -> app.application.subscription.contract", "app.chain.subscribe.refresh -> app.application.subscription.facts", "app.chain.subscribe.refresh -> app.application.subscription.priority", @@ -4598,8 +4639,8 @@ "app.chain.subscribe.refresh -> app.chain.download", "app.chain.subscribe.refresh -> app.chain.media", "app.chain.subscribe.refresh -> app.chain.subscribe", - "app.chain.subscribe.refresh -> app.chain.subscribe.contract", "app.chain.subscribe.refresh -> app.chain.subscribe.identity", + "app.chain.subscribe.refresh -> app.chain.subscribe.metadata", "app.chain.subscribe.refresh -> app.chain.tmdb", "app.chain.subscribe.refresh -> app.chain.torrents", "app.chain.subscribe.refresh -> app.domain", @@ -4612,7 +4653,6 @@ "app.chain.subscribe.refresh -> app.runtime.stop", "app.chain.subscribe.refresh -> app.schemas", "app.chain.subscribe.refresh -> app.schemas.event", - "app.chain.subscribe.refresh -> app.schemas.media", "app.chain.subscribe.refresh -> app.schemas.mediaserver", "app.chain.subscribe.refresh -> app.schemas.types", "app.chain.subscribe.search -> app.application", @@ -5185,6 +5225,15 @@ "app.db.adapters.subscriptionsearch -> app.db.oper", "app.db.adapters.subscriptionsearch -> app.db.oper.subscriptionsearch", "app.db.adapters.subscriptionsearch -> app.db.uow", + "app.db.adapters.subscriptionstatus -> app.application", + "app.db.adapters.subscriptionstatus -> app.application.download", + "app.db.adapters.subscriptionstatus -> app.application.download.admission", + "app.db.adapters.subscriptionstatus -> app.application.subscription", + "app.db.adapters.subscriptionstatus -> app.application.subscription.execution", + "app.db.adapters.subscriptionstatus -> app.db", + "app.db.adapters.subscriptionstatus -> app.db.models", + "app.db.adapters.subscriptionstatus -> app.db.models.subscriptiondownload", + "app.db.adapters.subscriptionstatus -> app.db.models.subscriptionsearch", "app.db.adapters.transaction -> app.db", "app.db.adapters.transaction -> app.db.uow", "app.db.adapters.transfer.admission -> app.application", @@ -6186,8 +6235,8 @@ "app.modules.indexer -> app.application", "app.modules.indexer -> app.application.site", "app.modules.indexer -> app.application.site.health", + "app.modules.indexer -> app.application.site.observation", "app.modules.indexer -> app.application.site.query", - "app.modules.indexer -> app.application.site.search_observation", "app.modules.indexer -> app.domain", "app.modules.indexer -> app.domain.context", "app.modules.indexer -> app.domain.site", @@ -7694,6 +7743,7 @@ "app.scheduler.catalog -> app.runtime.scheduling", "app.scheduler.catalog -> app.scheduler", "app.scheduler.catalog -> app.scheduler.contract", + "app.scheduler.catalog -> app.scheduler.services", "app.scheduler.catalog -> app.schemas", "app.scheduler.catalog -> app.schemas.system", "app.scheduler.chain -> app.application", @@ -8289,6 +8339,7 @@ "app.startup.composition.runtime -> app.db.adapters.site", "app.startup.composition.runtime -> app.db.adapters.subscription", "app.startup.composition.runtime -> app.db.adapters.subscriptionsearch", + "app.startup.composition.runtime -> app.db.adapters.subscriptionstatus", "app.startup.composition.runtime -> app.db.adapters.transfer", "app.startup.composition.runtime -> app.db.adapters.transfer.execution", "app.startup.composition.runtime -> app.db.oper", @@ -8877,7 +8928,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 932, + "module_count": 936, "modules": [ "app", "app.adapters", @@ -9051,6 +9102,7 @@ "app.api.endpoints.search", "app.api.endpoints.site", "app.api.endpoints.storage", + "app.api.endpoints.subexecution", "app.api.endpoints.subscribe", "app.api.endpoints.system", "app.api.endpoints.tmdb", @@ -9162,8 +9214,8 @@ "app.application.site.contract", "app.application.site.health", "app.application.site.mutation", + "app.application.site.observation", "app.application.site.query", - "app.application.site.search_observation", "app.application.storage", "app.application.subscription", "app.application.subscription.candidates", @@ -9178,6 +9230,7 @@ "app.application.subscription.query", "app.application.subscription.search", "app.application.subscription.sitebudget", + "app.application.subscription.status", "app.application.subscription.write", "app.application.system", "app.application.torrent", @@ -9260,6 +9313,7 @@ "app.chain.subscribe.identity", "app.chain.subscribe.interaction", "app.chain.subscribe.match", + "app.chain.subscribe.metadata", "app.chain.subscribe.notify", "app.chain.subscribe.policy", "app.chain.subscribe.query", @@ -9309,6 +9363,7 @@ "app.db.adapters.subscription", "app.db.adapters.subscriptiondownload", "app.db.adapters.subscriptionsearch", + "app.db.adapters.subscriptionstatus", "app.db.adapters.transaction", "app.db.adapters.transfer", "app.db.adapters.transfer.admission", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 8c244a64b..b2554a8a1 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -994,7 +994,7 @@ "var-annotated": 2 }, "app/chain/subscribe/search.py": { - "assignment": 3 + "assignment": 2 }, "app/chain/system.py": { "attr-defined": 1, diff --git a/tests/fixtures/architecture/ruff-baseline.json b/tests/fixtures/architecture/ruff-baseline.json index d8b491700..def2b0a06 100644 --- a/tests/fixtures/architecture/ruff-baseline.json +++ b/tests/fixtures/architecture/ruff-baseline.json @@ -570,9 +570,6 @@ "app/schemas/response.py": { "I001": 1 }, - "app/schemas/subscribe.py": { - "I001": 1 - }, "app/schemas/types.py": { "I001": 1 }, diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 3755efef5..e4c47a74d 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -1669,7 +1669,7 @@ "ChainEventType.ResourceDownload": { "consumer_fingerprints": [], "producer_fingerprints": [ - "aac2e9ac03b43fd89c42f3efba186365152568edc05881f24204a88fbd74bdbf" + "26b919353eed404b502c7fdf899981f372b9a25ca75a90ba205f61c5ff90c41a" ] }, "ChainEventType.ResourceSelection": { @@ -2237,10 +2237,10 @@ "events": [ "ChainEventType.ResourceDownload" ], - "fingerprint": "aac2e9ac03b43fd89c42f3efba186365152568edc05881f24204a88fbd74bdbf", + "fingerprint": "26b919353eed404b502c7fdf899981f372b9a25ca75a90ba205f61c5ff90c41a", "invalid": false, "method": "send_event", - "qualname": "DownloadSubmissionOwner._apply_resource_download_event", + "qualname": "_DownloadResourceOwner._apply_resource_download_event", "receiver_kind": "canonical_singleton" }, { diff --git a/tests/test_subscription_architecture.py b/tests/test_subscription_architecture.py index 6654b1366..4e32321b0 100644 --- a/tests/test_subscription_architecture.py +++ b/tests/test_subscription_architecture.py @@ -234,6 +234,7 @@ def test_subscribe_chain_package_has_single_responsibility_owners() -> None: "identity.py", "interaction.py", "match.py", + "metadata.py", "notify.py", "policy.py", "query.py", diff --git a/tests/test_subscription_governance_joint.py b/tests/test_subscription_governance_joint.py index 1d8db8e06..e83463296 100644 --- a/tests/test_subscription_governance_joint.py +++ b/tests/test_subscription_governance_joint.py @@ -3,7 +3,7 @@ from datetime import datetime, timedelta, timezone from types import SimpleNamespace -from app.application.site.search_observation import report_site_search_outcome +from app.application.site.observation import report_site_search_outcome from app.application.subscription.sitebudget import SiteBudgetClaim, SubscriptionSiteBudget from app.chain.search.facade import SearchChain from app.chain.search.provider import SearchProviderOwner diff --git a/tests/test_subscription_site_budget.py b/tests/test_subscription_site_budget.py index f783fd426..3e76326f8 100644 --- a/tests/test_subscription_site_budget.py +++ b/tests/test_subscription_site_budget.py @@ -6,7 +6,7 @@ import pytest from sqlalchemy import create_engine, select from sqlalchemy.orm import Session, sessionmaker -from app.application.site.search_observation import ( +from app.application.site.observation import ( SiteSearchObservation, report_site_search_outcome, )