diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py index 2a37cc2b0..f833c9d02 100644 --- a/app/api/dependencies/subscription.py +++ b/app/api/dependencies/subscription.py @@ -1,6 +1,7 @@ """订阅领域的请求级 command/query 依赖。""" -from collections.abc import Mapping +from collections.abc import AsyncIterator, Mapping +from contextlib import asynccontextmanager from typing import cast from fastapi import Depends @@ -27,6 +28,7 @@ from app.application.servarr import ServarrSubscriptionService from app.application.subscription.contract import ( SessionSubscriptionPort, SubscriptionHistoryStagingPort, + SubscriptionQueryPort, SubscriptionStagingPort, ) from app.application.subscription.delete import ( @@ -272,3 +274,15 @@ def get_servarr_subscription_service( sync_repository=runtime.subscription.repository(db), unit_of_work=cast(DeleteUnitOfWork, unit_of_work), ) + + +@asynccontextmanager +async def subscription_log_query_scope(runtime: HostRuntime) -> AsyncIterator[SubscriptionQueryService]: + """日志权限轮询使用短会话,每次检查后释放连接,避免 SSE 占用数据库连接池。""" + sessions = runtime.subscription.async_session() + try: + session = await anext(sessions) + repository = cast(SubscriptionQueryPort, runtime.subscription.repository(session)) + yield SubscriptionQueryService(repository, async_repository=repository) + finally: + await sessions.aclose() diff --git a/app/api/endpoints/subexecution.py b/app/api/endpoints/subexecution.py index 12a1b7aa8..3980edb2e 100644 --- a/app/api/endpoints/subexecution.py +++ b/app/api/endpoints/subexecution.py @@ -1,20 +1,31 @@ """订阅执行批次状态与取消端点。""" +import time +from collections.abc import AsyncIterator from typing import Any, List, Optional -from fastapi import Depends, HTTPException +from fastapi import Depends, HTTPException, Path, Query, Request +from starlette.responses import Response, StreamingResponse +from app.adapters.web.security.access import verify_resource_token +from app.api.context import get_host_runtime 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, + subscription_log_query_scope, ) +from app.api.endpoints.system import _build_log_zip_response from app.api.principal import ApiPrincipal from app.api.response import ResponseAPIRouter from app.application.subscription.query import SubscriptionQueryService from app.application.subscription.status import SubscriptionExecutionStatusService +from app.application.system import LogNotFoundError +from app.runtime.log import redact_scoped_log from app.schemas.response import Response as _SchemaResponse from app.schemas.subscribe import SubscriptionBatchStatus as _SchemaSubscriptionBatchStatus +from app.schemas.token import TokenPayload as _SchemaTokenPayload +from app.startup.composition.context import HostRuntime router = ResponseAPIRouter() @@ -96,3 +107,90 @@ async def cancel_subscription_execution_batch( success=bool(cancelled), message="" if cancelled else "这次搜索已经结束,暂时无法停止", ) + + +async def _require_subscription_log_access( + subscription_id: int, + query: SubscriptionQueryService, + user: _SchemaTokenPayload, +) -> None: + """日志仅对当前订阅 owner 或管理员开放,未知和越权 ID 均返回 404。""" + subscribe = await query.get_public(subscription_id) + if subscribe is None or (not user.super_user and subscribe.username != user.username): + raise HTTPException(status_code=404, detail="订阅不存在或无权查看") + + +async def _authorize_subscription_logs(subscription_id: int, user: _SchemaTokenPayload, runtime: HostRuntime) -> None: + """在独立短会话中确认归属,返回后不持有数据库事务。""" + async with subscription_log_query_scope(runtime) as query: + await _require_subscription_log_access(subscription_id, query, user) + + +@router.get( # type: ignore[misc] + "/{subscription_id}/logs", + summary="查看单个订阅的执行日志", + response_model=None, + response_class=StreamingResponse, + responses={200: {"content": {"text/event-stream": {}, "text/plain": {}}}}, +) +async def get_subscription_logs( + request: Request, + subscription_id: int = Path(gt=0), + length: int = Query(default=100, ge=-1, le=600), + user: _SchemaTokenPayload = Depends(verify_resource_token), + runtime: HostRuntime = Depends(get_host_runtime), +) -> Response: + """按订阅权限输出脱敏文本或实时流;首次执行前保持空流等待日志生成。""" + await _authorize_subscription_logs(subscription_id, user, runtime) + logfile = f"subscriptions/{subscription_id}.log" + if length == -1: + try: + content = await runtime.system.read_log(logfile) + except LogNotFoundError: + content = "" + return Response(redact_scoped_log(content), media_type="text/plain", headers={"Cache-Control": "no-store"}) + last_check = time.monotonic() + + async def disconnected() -> bool: + """断开或归属变更时停止流;周期重读归属并验证资源令牌的有效性。""" + nonlocal last_check + if await request.is_disconnected(): + return True + if time.monotonic() - last_check >= 5: + try: + verify_resource_token(request.cookies.get(runtime.system.settings.get("PROJECT_NAME"))) + await _authorize_subscription_logs(subscription_id, user, runtime) + except HTTPException: + return True + last_check = time.monotonic() + return False + + source = await runtime.system.follow_log(logfile, max(1, length), disconnected) + + async def events() -> AsyncIterator[str]: + """逐行编码 SSE,心跳不进入前端日志列表。""" + async for line in source: + content = redact_scoped_log(line) + yield "".join(f"data: {part}\n" for part in content.splitlines()) + "\n" if content else ": keepalive\n\n" + + return StreamingResponse(events(), media_type="text/event-stream", headers={"Cache-Control": "no-store", "X-Accel-Buffering": "no"}) + + +@router.get( # type: ignore[misc] + "/{subscription_id}/logs/download", + summary="下载单个订阅的执行日志", + response_model=None, + response_class=Response, + responses={200: {"content": {"application/zip": {}}}}, +) +async def download_subscription_logs( + subscription_id: int = Path(gt=0), + user: _SchemaTokenPayload = Depends(verify_resource_token), + runtime: HostRuntime = Depends(get_host_runtime), +) -> Response: + """在归属检查后下载当前订阅的日志及滚动备份,客户端不能指定路径。""" + await _authorize_subscription_logs(subscription_id, user, runtime) + return await _build_log_zip_response( + f"subscriptions/{subscription_id}", runtime, + transform=redact_scoped_log, allow_empty=True, + ) diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index 0dd68f8ed..9bb9e6e34 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -3,7 +3,7 @@ import io import json import zipfile from datetime import datetime -from typing import Annotated, Any, Optional +from typing import Annotated, Any, Callable, Optional from urllib.parse import urlsplit import anyio @@ -182,14 +182,21 @@ def _verify_log_resource_superuser( return token_payload -async def _build_log_zip_response(name: str, runtime: HostRuntime) -> StreamingResponse: +async def _build_log_zip_response( + name: str, + runtime: HostRuntime, + transform: Optional[Callable[[str], str]] = None, + allow_empty: bool = False, +) -> StreamingResponse: """读取受限日志条目并生成兼容 ZIP 响应。""" try: entries = await runtime.system.collect_logs(name) except LogNotFoundError as error: - raise HTTPException(status_code=404, detail="Not Found") from error - zip_data, zip_stem = await anyio.to_thread.run_sync(_build_log_zip_data, name, entries) - headers = {"Content-Disposition": f'attachment; filename="{zip_stem}.zip"'} + if not allow_empty: + raise HTTPException(status_code=404, detail="Not Found") from error + entries = [LogFileData(name="README.txt", content="暂无执行日志,下一次搜索或 RSS 匹配后开始记录。".encode("utf-8"))] + zip_data, zip_stem = await anyio.to_thread.run_sync(_build_log_zip_data, name, entries, transform) + headers = {"Content-Disposition": f'attachment; filename="{zip_stem}.zip"', "Cache-Control": "no-store"} return StreamingResponse( iter([zip_data]), media_type="application/zip", @@ -197,15 +204,20 @@ async def _build_log_zip_response(name: str, runtime: HostRuntime) -> StreamingR ) -def _build_log_zip_data(name: str, entries: list[LogFileData]) -> tuple[bytes, str]: +def _build_log_zip_data( + name: str, + entries: list[LogFileData], + transform: Optional[Callable[[str], str]] = None, +) -> tuple[bytes, str]: """只负责把应用服务提供的日志内容编码为 ZIP framing。""" zip_buffer = io.BytesIO() filename_time = datetime.now().strftime("%Y%m%d-%H%M%S") - safe_name = (name or "logs").strip().lower() or "logs" + safe_name = (name or "logs").strip().lower().replace("/", "-") or "logs" zip_stem = f"{safe_name}-logs-{filename_time}" with zipfile.ZipFile(zip_buffer, mode="w", compression=zipfile.ZIP_DEFLATED) as archive: for entry in entries: - archive.writestr(f"{zip_stem}/{entry.name}", entry.content) + content = transform(entry.content.decode("utf-8", errors="replace")) if transform else entry.content + archive.writestr(f"{zip_stem}/{entry.name}", content) zip_buffer.seek(0) return zip_buffer.getvalue(), zip_stem diff --git a/app/application/subscription/mutation.py b/app/application/subscription/mutation.py index 9348b5eeb..475cb5508 100644 --- a/app/application/subscription/mutation.py +++ b/app/application/subscription/mutation.py @@ -29,6 +29,8 @@ from app.application.subscription.contract import ( SubscriptionSnapshot, ) from app.application.subscription.delete import AsyncUnitOfWork +from app.application.subscription.observability import subscription_log_scope +from app.runtime.log import logger from app.schemas.common import JsonData from app.schemas.event import SubscribeModifiedEventData from app.schemas.types import MUSIC_ENTITY_ALBUM @@ -213,13 +215,16 @@ class SyncSubscriptionMutationService: subscribe = self.get_accessible(subscribe_id, actor) if not subscribe: return None - return self.update( - subscribe_id, - _reset_payload(subscribe), - actor, - existing=subscribe, - scene="reset", - ) + with subscription_log_scope(subscribe_id, "reset"): + result = self.update( + subscribe_id, + _reset_payload(subscribe), + actor, + existing=subscribe, + scene="reset", + ) + logger.scoped_info("订阅进度已重置,下次搜索将重新检查缺失内容" if result is not None else "本次重置未完成,订阅记录可能已变更") + return result class SubscriptionMutationService: @@ -334,13 +339,16 @@ class SubscriptionMutationService: subscribe = await self.get_accessible(subscribe_id, actor) if not subscribe: return None - return await self.update( - subscribe_id, - _reset_payload(subscribe), - actor, - existing=subscribe, - scene="reset", - ) + with subscription_log_scope(subscribe_id, "reset"): + result = await self.update( + subscribe_id, + _reset_payload(subscribe), + actor, + existing=subscribe, + scene="reset", + ) + logger.scoped_info("订阅进度已重置,下次搜索将重新检查缺失内容" if result is not None else "本次重置未完成,订阅记录可能已变更") + return result async def delete_history( self, diff --git a/app/application/subscription/observability.py b/app/application/subscription/observability.py index a6bb4a0b8..0ccf49347 100644 --- a/app/application/subscription/observability.py +++ b/app/application/subscription/observability.py @@ -3,6 +3,8 @@ from __future__ import annotations import time +from collections.abc import Iterator +from contextlib import contextmanager from dataclasses import dataclass, field from typing import Collection, Literal, Mapping, Optional from uuid import uuid4 @@ -13,6 +15,7 @@ from app.application.subscription.execution import ( SubscriptionSearchRepository, ) from app.application.subscription.sitebudget import SubscriptionSiteBudgetMetrics +from app.runtime.log import bind_log_scope, logger SearchTaskOutcome = Literal["completed", "skipped", "failed", "cancelled", "requeued"] @@ -344,3 +347,20 @@ class SearchExecutionSummary: if unfinished_cleanup: message += f"另有 {unfinished_cleanup} 个搜索状态没有正常恢复,系统稍后会继续处理。" return message + + +@contextmanager +def subscription_log_scope( + subscription_id: int, + source: str, + operation_id: Optional[str] = None, +) -> Iterator[None]: + """隔离单订阅执行日志,并在所有退出路径汇总被限流的诊断详情。""" + identity = operation_id or uuid4().hex + with bind_log_scope("subscriptions", subscription_id, identity, source) as scope: + before = scope.suppressed + try: + yield + finally: + if scope.operation_id == identity and scope.suppressed > before: + logger.scoped_info(f"本轮合并或省略重复诊断详情 {scope.suppressed - before} 条") diff --git a/app/chain/search/result.py b/app/chain/search/result.py index 27b78b0c7..0ccc35242 100644 --- a/app/chain/search/result.py +++ b/app/chain/search/result.py @@ -56,7 +56,10 @@ def _filter_site_torrents( if filter_params: helper = cast(Callable[[], TorrentHelper], TorrentHelper)() filtered = [torrent for torrent in filtered if helper.filter_torrent(torrent, filter_params)] - diagnostics["filter_params"] += len(torrents) - len(filtered) + rejected = len(torrents) - len(filtered) + diagnostics["filter_params"] += rejected + if rejected: + logger.scoped_info(f"附加参数过滤排除 {rejected} 个资源(质量、分辨率等订阅要求)") if rule_groups and filtered: count = len(filtered) filtered = ( @@ -67,7 +70,10 @@ def _filter_site_torrents( ) or [] ) - diagnostics["filter_rules"] += count - len(filtered) + rejected = count - len(filtered) + diagnostics["filter_rules"] += rejected + if rejected: + logger.scoped_info(f"优先级过滤规则排除 {rejected} 个资源") return filtered diff --git a/app/chain/subscribe/completion.py b/app/chain/subscribe/completion.py index 6681499fe..6022aa270 100644 --- a/app/chain/subscribe/completion.py +++ b/app/chain/subscribe/completion.py @@ -11,6 +11,7 @@ from app.application.subscription.contract import ( subscribe_media_keys, ) from app.application.subscription.mutation import SubscriptionActor +from app.application.subscription.observability import subscription_log_scope from app.chain.subscribe.contract import _SubscribeOwnerBase from app.chain.subscribe.policy import SubscribePolicyOwner from app.domain.context import ( @@ -111,13 +112,26 @@ class SubscribeCompletionOwner(_SubscribeOwnerBase): subscribe: SubscriptionSnapshot, meta: MetaBase, mediainfo: MediaInfo, - downloads: List[Context] = None, - lefts: Dict[Union[int | str], Dict[int, _SchemaNotExistMediaInfo]] = None, + downloads: Optional[List[Context]] = None, + lefts: Optional[Dict[Union[int | str], Dict[int, _SchemaNotExistMediaInfo]]] = None, force: Optional[bool] = False, ) -> None: """ 判断是否应完成订阅 """ + with subscription_log_scope(subscribe.id, "completion"): + return self._evaluate_subscription_completion(subscribe, meta, mediainfo, downloads, lefts, force) + + def _evaluate_subscription_completion( + self, + subscribe: SubscriptionSnapshot, + meta: MetaBase, + mediainfo: MediaInfo, + downloads: Optional[List[Context]] = None, + lefts: Optional[Dict[Union[int | str], Dict[int, _SchemaNotExistMediaInfo]]] = None, + force: Optional[bool] = False, + ) -> None: + """在所属订阅日志上下文内检查剩余内容并执行原有完成策略。""" media_keys = subscribe_media_keys(subscribe) # 是否有剩余集 no_lefts = not lefts or not any(lefts.get(media_key) for media_key in media_keys) diff --git a/app/chain/subscribe/match.py b/app/chain/subscribe/match.py index 5b7715725..c7f1a1a6d 100644 --- a/app/chain/subscribe/match.py +++ b/app/chain/subscribe/match.py @@ -17,7 +17,7 @@ from app.application.subscription.execution import ( SubscriptionExecutionLease, ) from app.application.subscription.facts import FreshFactLease -from app.application.subscription.observability import MatchExecutionSummary +from app.application.subscription.observability import MatchExecutionSummary, subscription_log_scope from app.application.subscription.sitebudget import SubscriptionSearchCancelled from app.application.torrent.download import TorrentHelper from app.chain.media import MediaChain @@ -411,65 +411,9 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): text=(f"正在检查订阅({index}/{total_num}){listed_subscribe.name} ..."), current=listed_subscribe.id, ) - outcome: Optional[str] = "skipped" - reason: Optional[str] = "not_eligible" - lease = self._subscription_execution_admission.try_acquire( - subscription_id=listed_subscribe.id, - operation="match", - ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, - ) - if lease is None: - reason = "admission_conflict" - logger.debug(f"订阅 {listed_subscribe.name} 正在处理中,本次不再重复检查资源") - else: - current_subscribe = None - execution_context = SubscriptionExecutionContext( - lease=lease, - admission=self._subscription_execution_admission, - cancel_requested=lambda: runtime_stop_state.is_system_stopped, - ) - try: - current_subscribe = self.subscription_repository.get(listed_subscribe.id) - if current_subscribe is None: - logger.debug(f"订阅 {listed_subscribe.id} 已删除,本次不再检查资源") - elif current_subscribe.state not in {"R", "P"}: - logger.debug(f"订阅 {current_subscribe.name} 当前不需要检查资源,本次跳过") - elif execution_context.should_stop(): - reason = _match_stop_reason(execution_context) - logger.debug(f"订阅 {current_subscribe.name} 的资源检查已停止") - else: - outcome = self._match_subscription( - subscribe=current_subscribe, - processed_torrents=processed_torrents, - candidate_index=candidate_index, - fresh_fact_lease=fresh_fact_lease, - execution_context=execution_context, - ) - reason = _match_result_reason(outcome, execution_context) - except SubscriptionSearchCancelled as err: - outcome = "skipped" - reason = _match_stop_reason(execution_context) or "cancelled" - logger.debug(f"订阅 {listed_subscribe.name} 的资源检查已停止:{str(err)}") - except Exception as err: - outcome = "failed" - reason = "error" - subscribe_name = ( - current_subscribe.name if current_subscribe is not None else listed_subscribe.name - ) - logger.error(f"订阅 {subscribe_name} 检查资源时出错:{str(err)}", exc_info=True) - finally: - _release_match_admission( - self._subscription_execution_admission, - lease, - listed_subscribe.id, - summary, - ) - summary.record(outcome, reason) - _report_match_progress( - progress_callback, - summary, - value=20 + (summary.finished / total_num * 80 if total_num else 80), - text=(f"已检查订阅({index}/{total_num})"), + self._check_match_subscription( + listed_subscribe, processed_torrents, candidate_index, + fresh_fact_lease, summary, progress_callback, total_num, ) finally: processed_torrents.clear() @@ -484,6 +428,81 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): self._match_lock.release() logger.debug(f"订阅资源检查已结束:{datetime.now()}") + def _check_match_subscription( + self, + listed_subscribe: SubscriptionSnapshot, + processed_torrents: Dict[str, List[Context]], + candidate_index: CandidateIndex, + fresh_fact_lease: FreshFactLease, + summary: MatchExecutionSummary, + progress_callback: Optional[Callable[..., None]], + total_num: int, + ) -> None: + """在独立日志作用域内检查一个订阅并收口准入、异常和进度。""" + with subscription_log_scope(listed_subscribe.id, "rss", summary.run_id): + logger.scoped_info(f"开始 RSS 资源检查:{listed_subscribe.name},本轮共享候选 {summary.candidate_count} 个") + outcome: Optional[str] = "skipped" + reason: Optional[str] = "not_eligible" + lease = self._subscription_execution_admission.try_acquire( + subscription_id=listed_subscribe.id, + operation="match", + ttl_seconds=self._SUBSCRIPTION_EXECUTION_TTL, + ) + if lease is None: + reason = "admission_conflict" + logger.debug(f"订阅 {listed_subscribe.name} 正在处理中,本次不再重复检查资源") + else: + current_subscribe = None + execution_context = SubscriptionExecutionContext( + lease=lease, + admission=self._subscription_execution_admission, + cancel_requested=lambda: runtime_stop_state.is_system_stopped, + ) + try: + current_subscribe = self.subscription_repository.get(listed_subscribe.id) + if current_subscribe is None: + logger.debug(f"订阅 {listed_subscribe.id} 已删除,本次不再检查资源") + elif current_subscribe.state not in {"R", "P"}: + logger.debug(f"订阅 {current_subscribe.name} 当前不需要检查资源,本次跳过") + elif execution_context.should_stop(): + reason = _match_stop_reason(execution_context) + logger.debug(f"订阅 {current_subscribe.name} 的资源检查已停止") + else: + outcome = self._match_subscription( + subscribe=current_subscribe, + processed_torrents=processed_torrents, + candidate_index=candidate_index, + fresh_fact_lease=fresh_fact_lease, + execution_context=execution_context, + ) + reason = _match_result_reason(outcome, execution_context) + except SubscriptionSearchCancelled as err: + outcome = "skipped" + reason = _match_stop_reason(execution_context) or "cancelled" + logger.debug(f"订阅 {listed_subscribe.name} 的资源检查已停止:{str(err)}") + except Exception as err: + outcome = "failed" + reason = "error" + subscribe_name = ( + current_subscribe.name if current_subscribe is not None else listed_subscribe.name + ) + logger.error(f"订阅 {subscribe_name} 检查资源时出错:{str(err)}", exc_info=True) + finally: + _release_match_admission( + self._subscription_execution_admission, + lease, + listed_subscribe.id, + summary, + ) + summary.record(outcome, reason) + logger.scoped_info(f"本轮 RSS 检查结束:{outcome},原因:{reason or '正常结束'};检查完成不代表订阅完成") + _report_match_progress( + progress_callback, + summary, + value=20 + (summary.finished / total_num * 80 if total_num else 80), + text=(f"已检查订阅({summary.finished}/{total_num})"), + ) + def _match_subscription( self, *, @@ -815,6 +834,7 @@ class SubscribeMatchOwner(_SubscribeOwnerBase): execution_context=execution_context, ) + logger.scoped_info(f"本轮新增下载 {len(downloads or [])} 个") # 同步外部修改,更新订阅信息 updated_subscribe = self.subscription_repository.get(subscribe.id) diff --git a/app/chain/subscribe/refresh.py b/app/chain/subscribe/refresh.py index 61b9eb384..a64ba4a3b 100644 --- a/app/chain/subscribe/refresh.py +++ b/app/chain/subscribe/refresh.py @@ -12,6 +12,7 @@ from app.application.subscription.contract import ( subscribe_media_keys, ) from app.application.subscription.facts import FreshFactLease +from app.application.subscription.observability import subscription_log_scope from app.chain.download import DownloadChain from app.chain.media import MediaChain from app.chain.subscribe.identity import subscribe_recognize_kwargs @@ -128,32 +129,34 @@ class SubscribeRefreshOwner(SubscribeMetadataOwner): for index, subscribe in enumerate(subscribes, start=1): if runtime_stop_state.is_system_stopped: break - logger.info(f"开始更新订阅元数据:{subscribe.name} ...") - if progress_callback: - progress_callback( - value=(index - 1) / total_num * 100 if total_num else 100, - text=f"正在更新订阅元数据({index}/{total_num}){subscribe.name} ...", - data={ - "total": total_num, - "finished": index - 1, - "current": subscribe.id, - }, - ) - updated_subscribe = self._check_subscription( - subscribe, - fresh_fact_lease, - reconcile_completion=reconcile_completion, - media_chain_factory=MediaChain, - ) - 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, - text=f"订阅元数据({index}/{total_num})更新完成", - data={"total": total_num, "finished": index}, + with subscription_log_scope(subscribe.id, "refresh"): + logger.info(f"开始更新订阅元数据:{subscribe.name} ...") + if progress_callback: + progress_callback( + value=(index - 1) / total_num * 100 if total_num else 100, + text=f"正在更新订阅元数据({index}/{total_num}){subscribe.name} ...", + data={ + "total": total_num, + "finished": index - 1, + "current": subscribe.id, + }, + ) + updated_subscribe = self._check_subscription( + subscribe, + fresh_fact_lease, + reconcile_completion=reconcile_completion, + media_chain_factory=MediaChain, ) + 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, + text=f"订阅元数据({index}/{total_num})更新完成", + data={"total": total_num, "finished": index}, + ) + if progress_callback: progress_callback(value=100, text="订阅元数据更新完成") diff --git a/app/chain/subscribe/search.py b/app/chain/subscribe/search.py index a87060c2b..20482f4a1 100644 --- a/app/chain/subscribe/search.py +++ b/app/chain/subscribe/search.py @@ -29,6 +29,7 @@ from app.application.subscription.observability import ( batch_finished_count, batch_progress_text, inline_search_result, + subscription_log_scope, ) from app.application.subscription.query import SubscriptionQueryService, subscription_search_due from app.application.subscription.sitebudget import ( @@ -115,9 +116,11 @@ def _process_paged_subscription(owner: _SubscribeOwnerBase, subscribe: Subscript execution_context.incremental_search = True def submit(contexts: list[Context], ready: set[str]) -> set[str]: + """仅下载已就绪目标,并记录本页候选和实际提交结果。""" _ensure_execution_active(execution_context) if execution_context: execution_context.report_phase("preparing") + logger.scoped_info(f"本页可选资源 {len(contexts)} 个,开始检查下载条件") downloads, lefts = owner._SubscribeChain__download_best_version_with_full_pack_first( contexts=contexts, no_exists=deepcopy(target.missing), subscribe=subscribe, mediakey=target.media_key, username=subscribe.username, @@ -125,6 +128,7 @@ def _process_paged_subscription(owner: _SubscribeOwnerBase, subscribe: Subscript source=owner.get_subscribe_source_keyword(subscribe), execution_context=execution_context, eligible_targets=ready, ) + logger.scoped_info(f"本页新增下载 {len(downloads or [])} 个") if target.media.type == MediaType.MOVIE: settled = ready if downloads else set() else: @@ -818,6 +822,22 @@ class SubscribeSearchOwner(_SubscribeSearchQueueOwner): execution_context: Optional[SubscriptionExecutionContext] = None, ) -> Optional[SubscriptionSnapshot]: """处理单个订阅,并返回下载后重新读取的状态快照。""" + with subscription_log_scope(subscribe.id, "inline"): + logger.scoped_info(f"开始搜索订阅:{subscribe.name}") + result = self._search_subscription(subscribe, searchchain, execution_context) + if result is not None: + logger.scoped_info("本轮搜索结束,订阅继续等待") + else: + logger.scoped_info("本轮搜索结束,订阅状态已更新;请以完成记录为准") + return result + + def _search_subscription( + self, + subscribe: SubscriptionSnapshot, + searchchain: Optional[SearchChain], + execution_context: Optional[SubscriptionExecutionContext] = None, + ) -> Optional[SubscriptionSnapshot]: + """在订阅日志作用域内执行既有搜索、下载和完成判定。""" _ensure_execution_active(execution_context) # 站点补查属于同一轮搜索,不能反复推迟下一轮健康站点的完整检查。 if execution_context is None or not execution_context.resuming_sites: diff --git a/app/chain/subscribe/searchtask.py b/app/chain/subscribe/searchtask.py index 5b4a0351d..e80cf6164 100644 --- a/app/chain/subscribe/searchtask.py +++ b/app/chain/subscribe/searchtask.py @@ -21,6 +21,7 @@ from app.application.subscription.execution import ( from app.application.subscription.observability import ( SearchExecutionSummary, finish_returned_search_task, + subscription_log_scope, ) from app.application.subscription.sitebudget import ( SubscriptionSearchCancelled, @@ -96,6 +97,12 @@ class SubscriptionSearchTaskRunner: def execute(self) -> Optional[int]: """执行当前任务,返回实际进入搜索处理的订阅 ID。""" + with subscription_log_scope(self.task.subscription_id, self.task.source, str(self.task.task_id)): + logger.scoped_info("开始处理订阅搜索任务") + return self._execute() + + def _execute(self) -> Optional[int]: + """在已绑定日志的任务作用域内执行搜索及等待判断。""" lease_token = self.task.lease_token if lease_token is None: logger.error("这次订阅搜索暂时无法开始,系统稍后会重新处理") @@ -107,6 +114,7 @@ class SubscriptionSearchTaskRunner: if cancelled(): self.queue.release_task(task_id=task_id, lease_token=lease_token, cancelled=True) self.summary.record("cancelled", "cancelled") + logger.scoped_info("本轮搜索已取消") return None subscribe = self.subscription_repository.get(self.task.subscription_id) @@ -132,6 +140,7 @@ class SubscriptionSearchTaskRunner: message="订阅刚刚创建,保存好设置后会自动开始搜索", ) self.summary.record("requeued", "recent_subscription") + logger.scoped_info(f"订阅刚刚创建,搜索已安排在 {recent_retry_at} 继续") return None execution_lease = self.execution_admission.try_acquire( @@ -166,6 +175,7 @@ class SubscriptionSearchTaskRunner: return False skip_search_task(self.queue, self.task, "订阅已暂停,这次没有自动搜索") self.summary.record("skipped", "paused") + logger.scoped_info("订阅已暂停,本轮不执行自动搜索") return True def _handle_active_subscription( @@ -197,6 +207,7 @@ class SubscriptionSearchTaskRunner: "这个订阅正在处理,本次自动检查无需重复执行", ) self.summary.record("skipped", "admission_conflict") + logger.scoped_info("订阅正在处理,本轮自动检查无需重复执行") def _execute_owned_task( self, @@ -262,10 +273,12 @@ class SubscriptionSearchTaskRunner: cancel_requested=cancel_requested, ) self.summary.record(outcome, reason) + logger.scoped_info(f"本轮搜索任务结束:{outcome},原因:{reason or '正常结束'};订阅是否完成以缺失内容检查为准") return subscription_id except SubscriptionSearchCancelled: self._handle_cancelled_search(task_id, lease_token, execution_context) except SubscriptionSearchDeferred as deferred: + logger.scoped_info(f"站点暂时忙,本轮搜索延后继续:{deferred}") handle_subscription_search_deferred( self.queue, task_id, @@ -297,6 +310,7 @@ class SubscriptionSearchTaskRunner: current_site_id: Optional[int] = None, ) -> None: """保存当前任务的用户可见阶段和正在访问的站点。""" + logger.debug(f"执行阶段:{phase},当前站点:{current_site_id or '-'}") self.queue.update_task_phase( task_id=task_id, lease_token=lease_token, @@ -311,6 +325,7 @@ class SubscriptionSearchTaskRunner: execution_context: SubscriptionExecutionContext, ) -> None: """区分搜索超时、用户停止和系统关闭。""" + logger.scoped_info("本轮搜索已停止,将根据取消、超时或系统关闭状态收尾") if execution_context.is_expired() and not execution_context.is_cancel_requested(): self.queue.finish_task( task_id=task_id, diff --git a/app/locales/en-US.json b/app/locales/en-US.json index 7f9a57e52..d73497813 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -305,6 +305,7 @@ "请输入认证站点和认证参数": "Please enter the authentication site and authentication parameters", "新名称为空": "New name is empty", "订阅不存在": "Subscription does not exist", + "订阅不存在或无权查看": "Subscription does not exist or access is denied", "订阅历史不存在": "Subscription history does not exist", "无权删除该订阅历史": "You do not have permission to delete this subscription history", "无权删除该订阅": "You do not have permission to delete this subscription", diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index 3490950e5..a237cea07 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -297,6 +297,7 @@ "请输入认证站点和认证参数": "請輸入認證站點和認證參數", "新名称为空": "新名稱為空", "订阅不存在": "訂閱不存在", + "订阅不存在或无权查看": "訂閱不存在或無權查看", "订阅历史不存在": "訂閱歷史不存在", "无权删除该订阅历史": "無權刪除該訂閱歷史", "无权删除该订阅": "無權刪除該訂閱", diff --git a/app/runtime/log.py b/app/runtime/log.py index 7b21f94ca..1d4c54967 100644 --- a/app/runtime/log.py +++ b/app/runtime/log.py @@ -8,13 +8,14 @@ import inspect import logging import os import queue +import re import sys import threading import time from collections import deque from contextlib import contextmanager from contextvars import ContextVar -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone from logging.handlers import RotatingFileHandler from pathlib import Path @@ -68,6 +69,77 @@ class LogSettings(LogConfigModel): """保存当前进程已经生效的日志策略。""" +_SCOPED_LOG_MAX_BYTES = 1024 * 1024 +_SCOPED_LOG_BACKUPS = 2 +_SCOPED_LOG_MAX_HANDLERS = 32 +_SCOPED_LOG_IDLE_SECONDS = 60.0 +_SCOPED_LOG_TOTAL_BYTES = 64 * 1024 * 1024 +_SCOPED_LOG_DETAIL_LIMIT = 300 +_LOG_URL_PATTERN = re.compile(r"(?:https?://|magnet:\?)[^\s<>]+", re.IGNORECASE) +_LOG_SECRET_PATTERN = re.compile( + r"(?i)(?:[\"']?(?:cookie|authorization|password|passwd|passkey|api[_-]?key|" + r"api[_-]?token|access[_-]?token|secret|token)[\"']?\s*[:=]\s*).+" +) + + +def redact_scoped_log(message: str) -> str: + """订阅日志在落盘与读取时隐藏链接和凭据,避免普通用户取得站点密钥。""" + message = _LOG_URL_PATTERN.sub("[链接已隐藏]", message) + return "\n".join(_LOG_SECRET_PATTERN.sub("[凭据已隐藏]", line) for line in message.splitlines()) + + +@dataclass(slots=True) +class LogScope: + """携带一个业务资源的日志路由和本轮有界详情计数。""" + + category: str + resource_id: int + operation_id: str + source: str + details: int = 0 + suppressed: int = 0 + repetitions: dict[str, int] = field(default_factory=dict) + lock: Any = field(default_factory=threading.Lock, repr=False) + + def accept(self, level: int, message: str) -> bool: + """限制 DEBUG 详情及重复内容,保留运行结果和错误。""" + if level > logging.DEBUG: + return True + with self.lock: + count = self.repetitions.get(message, 0) + if self.details >= _SCOPED_LOG_DETAIL_LIMIT or count >= 3: + self.suppressed += 1 + return False + self.repetitions[message] = count + 1 + self.details += 1 + return True + + +_current_log_scope: ContextVar[Optional[LogScope]] = ContextVar("current_log_scope", default=None) + + +@contextmanager +def bind_log_scope( + category: str, + resource_id: int, + operation_id: str, + source: str, +) -> Iterator[LogScope]: + """绑定资源日志;同资源嵌套复用本轮身份,退出时恢复原上下文。""" + if not re.fullmatch(r"[a-z]+", category) or resource_id <= 0: + raise ValueError("日志资源标识无效") + current = _current_log_scope.get() + if current and (current.category, current.resource_id) == (category, resource_id): + yield current + return + scope = LogScope(category, resource_id, operation_id, source) + token = _current_log_scope.set(scope) + try: + yield scope + finally: + _current_log_scope.reset(token) + + class LogEntry: """表示一条等待基础设施写入的文件日志。""" @@ -407,8 +479,10 @@ class NonBlockingFileHandler: return self._initialized = True self._state_lock = threading.RLock() - self._handlers_lock = threading.Lock() + self._handlers_lock = threading.RLock() self._rotating_handlers: dict[Path, RotatingFileHandler] = {} + self._scoped_handler_times: dict[Path, float] = {} + self._scoped_cleanup_at = 0.0 self._write_queue: queue.Queue[Optional[LogEntry]] = queue.Queue( maxsize=log_settings.ASYNC_FILE_QUEUE_SIZE, ) @@ -433,8 +507,8 @@ class NonBlockingFileHandler: file_path.parent.mkdir(parents=True, exist_ok=True) handler = RotatingFileHandler( filename=str(file_path), - maxBytes=log_settings.LOG_MAX_FILE_SIZE_BYTES, - backupCount=log_settings.LOG_BACKUP_COUNT, + maxBytes=_SCOPED_LOG_MAX_BYTES if file_path.parent.name == "subscriptions" else log_settings.LOG_MAX_FILE_SIZE_BYTES, + backupCount=_SCOPED_LOG_BACKUPS if file_path.parent.name == "subscriptions" else log_settings.LOG_BACKUP_COUNT, encoding="utf-8", ) handler.setFormatter(logging.Formatter(log_settings.LOG_FILE_FORMAT)) @@ -472,14 +546,56 @@ class NonBlockingFileHandler: return True def _write_sync(self, entry: LogEntry) -> None: - """立即将单条日志写入滚动文件。""" + """串行完成写入和处理器回收,避免回收与 emit 并发导致重新打开已淘汰文件。""" try: - handler = self._get_rotating_handler(entry.file_path) - handler.handle(self._to_record(entry)) + with self._handlers_lock: + if entry.file_path.parent.name == "subscriptions": + self._maintain_scoped_logs(entry.file_path) + handler = self._get_rotating_handler(entry.file_path) + handler.handle(self._to_record(entry)) except Exception as err: print(f"日志写入失败 {entry.file_path}: {err}") print(f"【{entry.level.upper()}】{entry.timestamp} - {entry.message}") + def _close_scoped_handler(self, path: Path) -> None: + """在写入锁内释放一个订阅文件句柄并移除其空闲计时。""" + handler = self._rotating_handlers.pop(path, None) + if handler: + handler.close() + self._scoped_handler_times.pop(path, None) + + def _maintain_scoped_logs(self, active_path: Path) -> None: + """按空闲时间和数量回收句柄,周期清理最旧归档以限制目录总容量。""" + now = time.monotonic() + for path, last_used in tuple(self._scoped_handler_times.items()): + if now - last_used >= _SCOPED_LOG_IDLE_SECONDS: + self._close_scoped_handler(path) + self._scoped_handler_times[active_path] = now + while len(self._scoped_handler_times) > _SCOPED_LOG_MAX_HANDLERS: + self._close_scoped_handler(min(self._scoped_handler_times, key=self._scoped_handler_times.__getitem__)) + if now < self._scoped_cleanup_at: + return + self._scoped_cleanup_at = now + 30.0 + self._trim_scoped_directory(active_path) + + def _trim_scoped_directory(self, active_path: Path) -> None: + """只清理数字订阅 ID 的日志及滚动备份,不跟随符号链接或删除当前写入文件。""" + directory = active_path.parent + if not directory.is_dir(): + return + files = [(path, path.stat()) for path in directory.iterdir() + if re.fullmatch(r"[1-9][0-9]*\.log(?:\.[12])?", path.name) + and not path.is_symlink() and path.is_file()] + total = sum(stat.st_size for _, stat in files) + for path, stat in sorted(files, key=lambda item: item[1].st_mtime): + if total <= _SCOPED_LOG_TOTAL_BYTES: + break + if path == active_path: + continue + self._close_scoped_handler(directory / f"{path.name.split('.log')[0]}.log") + path.unlink(missing_ok=True) + total -= stat.st_size + @staticmethod def _to_record(entry: LogEntry) -> logging.LogRecord: """把日志条目转换为标准库日志记录。""" @@ -811,7 +927,8 @@ class LoggerManager: def logger(self, method: str, msg: str, *args: Any, **kwargs: Any) -> None: """按调用来源路由并输出一条日志。 - 等级过滤只看 `current_plugin_instance_id()`:命中受控调用点绑定的插件 + 主日志等级过滤看 `current_plugin_instance_id()`;资源日志独立接收 DEBUG, + 在落盘前脱敏并限制本轮详情数量。命中受控调用点绑定的插件 实例时按该实例的覆盖等级过滤(未设置覆盖时等同全局等级),未绑定任何 实例时直接按全局等级过滤。这一步必须排在全局等级判定之前——先按全局 等级短路就会把实例调低的等级挡在门外,再往下的栈帧内省也就永远看不到 @@ -823,7 +940,10 @@ class LoggerManager: 由 `_ContextLevelFilter` 用同一个判据再判一次,两处结论因而一致。 """ method_level = getattr(logging, method.upper(), logging.INFO) - if method_level < _effective_context_log_level(): + scope = _current_log_scope.get() + scope_only = kwargs.pop("scope_only", False) + ordinary_enabled = not scope_only and method_level >= _effective_context_log_level() + if not ordinary_enabled and scope is None: return caller_name, plugin_name = self._get_caller() @@ -834,6 +954,15 @@ class LoggerManager: except (TypeError, ValueError): formatted_msg = f"{formatted_msg} {' '.join(str(arg) for arg in args)}" + if scope: + scoped_message = redact_scoped_log(formatted_msg)[:8192] + if scope.accept(method_level, scoped_message): + prefix = f"[来源={scope.source} 轮次={scope.operation_id}]" + self._write_file_log(method.upper(), f"{prefix} {scoped_message}", + Path(scope.category) / f"{scope.resource_id}.log") + if not ordinary_enabled: + return + logfile = ( Path("plugins") / f"{plugin_name}.log" if plugin_name @@ -845,6 +974,10 @@ class LoggerManager: log_method = getattr(configured_logger, method, configured_logger.info) log_method(formatted_msg, **kwargs) + def scoped_info(self, msg: str) -> None: + """仅向已绑定资源输出 INFO 摘要,保持主日志的批次摘要密度。""" + self.logger("info", msg, scope_only=True) + def info(self, msg: str, *args: Any, **kwargs: Any) -> None: """输出 INFO 日志。""" self.logger("info", msg, *args, **kwargs) diff --git a/app/startup/composition/system.py b/app/startup/composition/system.py index f51d8e572..9b9137ca9 100644 --- a/app/startup/composition/system.py +++ b/app/startup/composition/system.py @@ -3,8 +3,8 @@ from __future__ import annotations import asyncio +import os import re -from collections import deque from collections.abc import AsyncIterator, Awaitable, Callable from pathlib import Path from typing import Any, cast @@ -77,6 +77,9 @@ class _FileLogAdapter(SystemLogPort): disconnected: Callable[[], Awaitable[bool]], ) -> AsyncIterator[str]: """先返回历史尾部,再轮询并返回新增日志行。""" + if re.fullmatch(r"subscriptions/[1-9][0-9]*\.log", logfile): + path = Path(self._settings.get("LOG_PATH")) / logfile + return self._follow_path(path, length, disconnected) path = await self._resolve_logfile(logfile) return self._follow_path(path, length, disconnected) @@ -93,7 +96,8 @@ class _FileLogAdapter(SystemLogPort): user_path=AsyncPath(path), allowed_suffixes={".log"}, ) - if not safe or not await asyncio.to_thread(path.is_file): + subscription = str(logfile).startswith("subscriptions/") + if (subscription and (await AsyncPath(path).is_symlink() or await AsyncPath(path.parent).is_symlink())) or not safe or not await asyncio.to_thread(path.is_file): raise LogNotFoundError(logfile) return path @@ -103,59 +107,56 @@ class _FileLogAdapter(SystemLogPort): length: int, disconnected: Callable[[], Awaitable[bool]], ) -> AsyncIterator[str]: - """对一个已准入文件执行有界历史读取和新增内容轮询。""" - lines_queue: deque[str] = deque(maxlen=max(length, 50)) - file_size = (await asyncio.to_thread(path.stat)).st_size - async with aiofiles.open(path, mode="r", encoding="utf-8", errors="replace") as file: - if file_size > 100 * 1024: - position = file_size - min(file_size, 100 * 1024) - await file.seek(position) - content = await file.read() - first_newline = content.find("\n") - if first_newline != -1: - content = content[first_newline + 1 :] - else: - content = await file.read() - for line in (item.strip() for item in content.splitlines() if item.strip()): - lines_queue.append(line) - for line in lines_queue: - yield line - async with aiofiles.open(path, mode="r", encoding="utf-8", errors="replace") as file: - await file.seek(0, 2) - previous_size = (await asyncio.to_thread(path.stat)).st_size - pending = "" - while not runtime_stop_state.is_system_stopped: - if await disconnected(): - break - current_size = (await asyncio.to_thread(path.stat)).st_size - if current_size < previous_size: - await file.seek(0) - previous_size = 0 - pending = "" - if current_size > previous_size: - pending += await file.read() - previous_size = current_size - fragments = pending.splitlines(keepends=True) - if fragments and not fragments[-1].endswith(("\n", "\r")): - pending = fragments.pop() - else: - pending = "" - for fragment in fragments: - line = fragment.strip() - if line: - yield line - else: - await asyncio.sleep(0.5) + """有界读取尾部及完整新增行;重开文件以追踪滚动,并等待订阅首次产生日志。""" + cursor: tuple[int, int] | None = None + initial = True + while not runtime_stop_state.is_system_stopped and not await disconnected(): + lines, cursor = await asyncio.to_thread(self._log_chunk, path, cursor, length if initial else 0) + if cursor is not None: + initial = False + for line in lines: + yield line + if path.parent.name == "subscriptions": + yield "" + await asyncio.sleep(0.5) + + def _log_chunk( + self, + path: Path, + cursor: tuple[int, int] | None, + length: int, + ) -> tuple[list[str], tuple[int, int] | None]: + """在一个工作线程内校验路径并读取完整行,保留末尾未写完的行供下次读取。""" + root = Path(self._settings.get("LOG_PATH")) + subscription = path.parent.name == "subscriptions" + if (subscription and (path.is_symlink() or path.parent.is_symlink())) or not SecurityUtils.is_safe_path(root, path, {".log"}): + raise LogNotFoundError(path.name) + try: + with path.open("rb") as file: + stat = os.fstat(file.fileno()) + position = cursor[1] if cursor and cursor[0] == stat.st_ino and cursor[1] <= stat.st_size else 0 + tail_start = max(0, stat.st_size - 100 * 1024) if length else position + file.seek(tail_start) + content = file.read(1024 * 1024) + end = content.rfind(b"\n") + 1 + complete = content[:end] + if length and tail_start: + complete = complete.partition(b"\n")[2] + lines = complete.decode("utf-8", errors="replace").splitlines() + return (lines[-length:] if length else lines), (stat.st_ino, tail_start + end) + except FileNotFoundError: + return [], None def _collect_sync(self, name: str) -> list[LogFileData]: """同步发现并读取最多十个主程序或指定插件日志。""" normalized = str(name or "").strip().lower() - if not normalized or not _LOG_DOWNLOAD_NAME_PATTERN.fullmatch(normalized): + subscription = re.fullmatch(r"subscriptions/([1-9][0-9]*)", normalized) + if not normalized or not (subscription or _LOG_DOWNLOAD_NAME_PATTERN.fullmatch(normalized)): raise LogNotFoundError(name) root = Path(self._settings.get("LOG_PATH")) - directory = root if normalized == "moviepilot" else root / "plugins" - prefix = "moviepilot.log" if normalized == "moviepilot" else f"{normalized}.log" - if not directory.is_dir(): + directory = root / "subscriptions" if subscription else root if normalized == "moviepilot" else root / "plugins" + prefix = f"{subscription[1]}.log" if subscription else "moviepilot.log" if normalized == "moviepilot" else f"{normalized}.log" + if not directory.is_dir() or (subscription and directory.is_symlink()): raise LogNotFoundError(name) current = directory / prefix backups = [ @@ -167,7 +168,7 @@ class _FileLogAdapter(SystemLogPort): paths = ([current] if current.is_file() else []) + backups entries: list[LogFileData] = [] for path in paths[:_LOG_DOWNLOAD_LIMIT]: - if not SecurityUtils.is_safe_path(base_path=root, user_path=path): + if (subscription and path.is_symlink()) or not SecurityUtils.is_safe_path(base_path=root, user_path=path): raise LogNotFoundError(name) entries.append(LogFileData(name=path.name, content=path.read_bytes())) if not entries: diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index ebeffdfa9..a0cd49a31 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -763,7 +763,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1114 | -| 内部导入边 | 9,437 | +| 内部导入边 | 9,458 | | 非平凡 SCC | 1(精确 containment 的 TMDB 移植包环) | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 49(债务已清零,49 条精确 containment) | @@ -773,7 +773,7 @@ flowchart LR | Model/Oper 自动事务与自建 Session | 0 | | 组合根外 `SystemConfigOper()` | 0 | -整理失败反馈由 `app.application.transfer.feedback` 集中投影;已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列;网络连通性检测按 `app.application.nettest` 归档,并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理;音乐标签准入与批次共识由 `app.chain.transfer.music` 负责,预览证据由 `app.application.transfer.projection` 只读投影;宿主依赖边现为 9,437;插件实例校验复用 `app.domain.plugin` 的物理身份规则,兼容下划线 ID,新增 2 条 Schema 到 Domain 的合法依赖边(含父包展开);种子下载限流窗口由 `app.domain.torrent` 定义,失败冷却 owner 复用同一语义,新增 1 条合法 Chain 到 Domain 的导入边;工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文,新增 5 条模块导入边;资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链,新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验,音乐交互包未新增 SCC,插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器,依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`,文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端,经 `RequestUtils` 出站,宿主不再导入 pyTelegramBotAPI,启用 Telegram 的实例也不会因该 SDK 加载 redis;该依赖暂时保留给插件使用。Web Push 协议(aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递)由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现,消息端点与 `WebPushModule` 共用,宿主不再导入 pywebpush 及其连带的 aiohttp;该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端(经 `RequestUtils` 出站)与网络适配器 `app.adapters.network.feishu` 事件长连接(websocket-client 同步收发 pbbp2 帧),宿主不再导入 lark-oapi 及其上万个生成模块,长连接也不再依赖 SDK 的模块级事件循环;插件经 `app.sdk.feishu` 复用同一长连接,lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放,默认关闭,未开放时不生成文档。当前 `app.startup.lifecycle` 为 582、`app.factory` 为 595、`app.main` 为 597。音乐调用观察 `app.application.music.observation` 新增一个轻量模块,记录来源结果及请求预算;AcoustID及路径识别复用此上下文传递有界指纹候选,保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`,各启动入口增加一个模块;性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。 +整理失败反馈由 `app.application.transfer.feedback` 集中投影;已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列;网络连通性检测按 `app.application.nettest` 归档,并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理;音乐标签准入与批次共识由 `app.chain.transfer.music` 负责,预览证据由 `app.application.transfer.projection` 只读投影;宿主依赖边现为 9,458;订阅独立日志复用运行时上下文、系统日志端口与资源鉴权,新增 21 条合法依赖边(含父包展开),未新增模块或 SCC;插件实例校验复用 `app.domain.plugin` 的物理身份规则,兼容下划线 ID,新增 2 条 Schema 到 Domain 的合法依赖边(含父包展开);种子下载限流窗口由 `app.domain.torrent` 定义,失败冷却 owner 复用同一语义,新增 1 条合法 Chain 到 Domain 的导入边;工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文,新增 5 条模块导入边;资源过滤动作 `app.workflow.actions.filter_torrents` 与动作包共用的领域还原函数同样把 Schema 资源还原为领域对象后再交给过滤链,新增 6 条模块导入边。音乐来源链与识别缓存复用领域身份核验,音乐交互包未新增 SCC,插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器,依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`,文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。Telegram 渠道改用模块内 `app.modules.telegram.botapi` 精简 Bot API 客户端,经 `RequestUtils` 出站,宿主不再导入 pyTelegramBotAPI,启用 Telegram 的实例也不会因该 SDK 加载 redis;该依赖暂时保留给插件使用。Web Push 协议(aes128gcm 加密、VAPID 签名与经 `RequestUtils` 投递)由网络适配器 `app.adapters.network.webpush` 直接基于 http-ece 与 py-vapid 实现,消息端点与 `WebPushModule` 共用,宿主不再导入 pywebpush 及其连带的 aiohttp;该依赖同样暂时保留给插件使用。飞书渠道改用模块内 `app.modules.feishu.openapi` OpenAPI 客户端(经 `RequestUtils` 出站)与网络适配器 `app.adapters.network.feishu` 事件长连接(websocket-client 同步收发 pbbp2 帧),宿主不再导入 lark-oapi 及其上万个生成模块,长连接也不再依赖 SDK 的模块级事件循环;插件经 `app.sdk.feishu` 复用同一长连接,lark-oapi 暂时保留给尚未迁移的插件使用。API 文档路由由 `app.adapters.web.docs` 按 `API_DOCS_ENABLE` 逐请求开放,默认关闭,未开放时不生成文档。当前 `app.startup.lifecycle` 为 582、`app.factory` 为 595、`app.main` 为 597。音乐调用观察 `app.application.music.observation` 新增一个轻量模块,记录来源结果及请求预算;AcoustID及路径识别复用此上下文传递有界指纹候选,保留旧单ID模块合同。站点图片域名快照新增 `app.application.security.image`,各启动入口增加一个模块;性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。 架构专项验证分为两个 CI 投影:`Check event semantic policy` 先运行依赖、Adapter、出口和 Event 语义门禁,`Check host architecture snapshot` 再执行快照测试及一次 diff --git a/docs/mcp-api.md b/docs/mcp-api.md index db79dcb08..d38987eb1 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -31,6 +31,17 @@ MCP 使用系统配置中的 `API_TOKEN` 作为认证密钥,文档中的 API K 域名放行后仍执行 DNS/私网校验, Fake-IP 等非公网解析地址需要通过 `IMAGE_PROXY_ALLOWED_PRIVATE_RANGES` 明确允许。 +### 单订阅执行日志(浏览器资源接口) + +- `GET /api/v1/subscribe/{subscription_id}/logs`:默认返回 SSE,先读取最近 100 行,随后实时输出;`length=1..600` 调整历史行数,`length=-1` 返回当前文件的脱敏文本。 +- `GET /api/v1/subscribe/{subscription_id}/logs/download`:下载该订阅当前日志及滚动备份 ZIP。 +- 使用现有资源 Cookie 验证当前有效账号,普通用户仅能查看自己创建的订阅,管理员可查看所有现存订阅。未知或越权 ID 返回 404,接口不接受文件路径。 +- SSE 每 5 秒重新确认账号和订阅归属,每次数据库检查使用独立短会话;账号失效、订阅删除或归属改变后停止输出。首次无日志时保持空流等待,不触发搜索。 +- 日志按订阅 ID 隔离,搜索任务复用 task ID,RSS 按轮次关联;过滤详情与每轮结果摘要可用于解释未完成原因。共享 RSS 抓取日志不归入某一条订阅,任务执行完成不等于订阅完成。 +- 订阅日志独立接收 DEBUG,并在写入及读取时隐藏 HTTP(S)/磁力链接和常见凭据字段;每轮最多 300 条 DEBUG,每条重复内容最多 3 次,省略数量写入摘要。 +- 每份日志 1 MiB,保留 2 个滚动备份,最多缓存 32 个订阅文件处理器;后续写入时回收空闲 60 秒的句柄。目录总容量以 64 MiB 为清理目标,定期淘汰最旧订阅日志。日志只从升级后的执行开始记录,清理后历史不可回溯。 +- 这些文本、SSE 和 ZIP 接口供前端使用,不属于 JSON envelope 的 Agent/MCP 工具。 + ## 2. 标准 MCP 协议 (JSON-RPC 2.0) ### 端点 diff --git a/docs/refactor/agent-api-surface-audit.json b/docs/refactor/agent-api-surface-audit.json index 0d340b006..45cb9e171 100644 --- a/docs/refactor/agent-api-surface-audit.json +++ b/docs/refactor/agent-api-surface-audit.json @@ -4,7 +4,7 @@ "consolidated": 71, "gateway": 228, "provider-skill": 12, - "stream_or_binary": 11, + "stream_or_binary": 13, "transport_or_identity": 76, "ui_presentation": 15 }, @@ -21,7 +21,7 @@ "gateway_http_route_count": 229, "gateway_operation_count": 231, "matched_gateway_http_route_count": 228, - "openapi_operation_count": 424, + "openapi_operation_count": 426, "operations": [ { "disposition": "consolidated", @@ -4263,6 +4263,30 @@ "subscribe" ] }, + { + "disposition": "stream_or_binary", + "method": "GET", + "operation_ids": [], + "owner": "host-transport", + "path": "/api/v1/subscribe/{subscription_id}/logs", + "reason": "Streaming, image, archive, or file response consumed by a direct client; the structured JSON Agent gateway does not proxy binary or unbounded streams.", + "summary": "查看单个订阅的执行日志", + "tags": [ + "subscribe" + ] + }, + { + "disposition": "stream_or_binary", + "method": "GET", + "operation_ids": [], + "owner": "host-transport", + "path": "/api/v1/subscribe/{subscription_id}/logs/download", + "reason": "Streaming, image, archive, or file response consumed by a direct client; the structured JSON Agent gateway does not proxy binary or unbounded streams.", + "summary": "下载单个订阅的执行日志", + "tags": [ + "subscribe" + ] + }, { "disposition": "stream_or_binary", "method": "GET", diff --git a/docs/refactor/agent-api-surface-audit.md b/docs/refactor/agent-api-surface-audit.md index 8faf27a39..3e40d1fba 100644 --- a/docs/refactor/agent-api-surface-audit.md +++ b/docs/refactor/agent-api-surface-audit.md @@ -5,7 +5,7 @@ ## Result -- OpenAPI HTTP operations: **424** +- OpenAPI HTTP operations: **426** - Stable `moviepilot_api` operations: **231** - Exact HTTP routes used by the gateway: **229** - OpenAPI routes matched directly by the gateway: **228** @@ -21,7 +21,7 @@ | `consolidated` | 71 | Source/UI route represented by a stable aggregate Agent operation. | | `gateway` | 228 | Approved structured MoviePilot Agent operation. | | `provider-skill` | 12 | Low-level downloader or media-server capability owned by a provider Skill. | -| `stream_or_binary` | 11 | Streaming or binary response owned by a direct client transport. | +| `stream_or_binary` | 13 | Streaming or binary response owned by a direct client transport. | | `transport_or_identity` | 76 | Authentication, protocol, callback, account, or conversation transport boundary. | | `ui_presentation` | 15 | Frontend or plugin-rendered presentation contract. | @@ -358,6 +358,8 @@ | `GET` | `/api/v1/subscribe/user/{username}` | subscribe | `gateway` | subscription.user.list | 用户订阅 | | `DELETE` | `/api/v1/subscribe/{subscribe_id}` | subscribe | `gateway` | subscription.delete | 删除订阅 | | `GET` | `/api/v1/subscribe/{subscribe_id}` | subscribe | `gateway` | subscription.get | 订阅详情 | +| `GET` | `/api/v1/subscribe/{subscription_id}/logs` | subscribe | `stream_or_binary` | host-transport | 查看单个订阅的执行日志 | +| `GET` | `/api/v1/subscribe/{subscription_id}/logs/download` | subscribe | `stream_or_binary` | host-transport | 下载单个订阅的执行日志 | | `GET` | `/api/v1/system/cache/image` | system | `stream_or_binary` | host-transport | 图片缓存 | | `GET` | `/api/v1/system/database/backups` | system | `gateway` | database.backups.list | 查询受管数据库备份 | | `POST` | `/api/v1/system/database/backups` | system | `gateway` | database.backups.create | 立即创建数据库备份 | diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 2daccecfd..82bdb2bcb 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -98,7 +98,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1114 / 9,437 | `dependency-baseline.json` 当前快照;插件实例校验复用 `app.domain.plugin` 的物理身份规则,兼容下划线 ID,新增 2 条合法依赖边(含父包展开);种子失败冷却复用 `app.domain.torrent` 的限流语义,新增 1 条合法导入边;系统设置合同按 `app.application.settings` 包组织,系统设置 API 端点独立按 `app.api.endpoints.settings` 归档;消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理;已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档;网络连通性检测按 `app.application.nettest` 归档;工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文,新增 5 条合法模块导入边,资源过滤动作复用同一还原再新增 6 条;LLM 独占连接关闭复用宿主线程池;歌词字形转换复用 `app.foundation.text`;订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口,将非 TMDB 影视身份投影为 TMDB 身份 | +| 宿主 Python 模块 / 内部依赖边 | 1114 / 9,458 | `dependency-baseline.json` 当前快照;插件实例校验复用 `app.domain.plugin` 的物理身份规则,兼容下划线 ID,新增 2 条合法依赖边(含父包展开);种子失败冷却复用 `app.domain.torrent` 的限流语义,新增 1 条合法导入边;系统设置合同按 `app.application.settings` 包组织,系统设置 API 端点独立按 `app.api.endpoints.settings` 归档;消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理;已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档;网络连通性检测按 `app.application.nettest` 归档;工作流添加下载动作由 `app.workflow.actions.add_download` 转换工作流 Schema 为下载链领域上下文,新增 5 条合法模块导入边,资源过滤动作复用同一还原再新增 6 条;LLM 独占连接关闭复用宿主线程池;歌词字形转换复用 `app.foundation.text`;订阅创建使用 `app.domain.context.MediaInfo` 和运行时配置端口,将非 TMDB 影视身份投影为 TMDB 身份 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -106,7 +106,7 @@ 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,169 / 508 文件 | 工作流资源过滤改用领域对象调用过滤链,恢复的节点输出在消费前归一化,并收紧可空资源列表的类型低水位;消息媒体交互类型债务已清零,canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 | +| 全量 mypy 历史债务 | 9,167 / 508 文件 | 工作流资源过滤改用领域对象调用过滤链,恢复的节点输出在消费前归一化,并收紧可空资源列表的类型低水位;消息媒体交互类型债务已清零,canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 | | Ruff 历史诊断 | 100 | 低水位门禁通过,但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%,Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index c043d00ef..c93a09cfc 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -81,6 +81,13 @@ to make the directory tree look symmetrical. | `app/application/messaging/` | Message rendering/routing, interactions and the Agent-to-message bridge: `ingress.py` owns the single channel-to-host loopback boundary; `interaction/` owns generic interaction state plus Agent choice contracts; `channel/admin.py` owns channel administrator resolution; `router.py` unified interaction priority and callback dispatch; `site.py`/`subscribe.py`/`skill.py` per-command sessions, input parsing and views; `media.py` media interaction state while the business workflow stays in `MediaInteractionChain`; `plugin.py` plugin input capture and plugin button callbacks; `webagent/` owns WebAgent notification events and stream transport; `agent.py` owns WebAgent application orchestration, display projection, session identity/persistence coordination, temporary attachment registry and audio preparation/transcription; `message.py` notification rendering, templates and queue. Not a public SDK recommended for direct plugin use | | `app/application/security/` | Authentication, authorization, frozen user/auth projections, atomic user aggregate commands, per-user configuration publication, cookies, passkeys, OTP/two-factor, path/URL safety, SSRF and signing policy | +订阅日志仍由 `runtime/log.py` 的通用资源作用域负责路由、等级、脱敏和有界滚动写入; +`application/subscription/observability.py` 只绑定订阅身份并汇总诊断省略数量。 +搜索、RSS 匹配、元数据刷新、重置和完成判定在单订阅边界绑定上下文。 +`api/endpoints/subexecution.py` 用资源 Cookie 校验归属并返回文本/SSE/ZIP, +权限检查通过 `api/dependencies/subscription.py` 的短会话作用域完成,流读取不能占用请求级数据库事务。 +文件读取、滚动追踪和归档复用现有 SystemLogPort 及启动装配,ZIP 压缩复用已有受控工作线程。 + ### Agent runtime boundaries | Path | Ownership | diff --git a/scripts/generate_agent_api_surface_audit.py b/scripts/generate_agent_api_surface_audit.py index 73e00ad9d..8861b3cb7 100644 --- a/scripts/generate_agent_api_surface_audit.py +++ b/scripts/generate_agent_api_surface_audit.py @@ -86,6 +86,8 @@ STREAM_OR_BINARY_PATHS = frozenset( "/api/v1/storage/image", "/api/v1/system/cache/image", "/api/v1/system/img/{proxy}", + "/api/v1/subscribe/{subscription_id}/logs", + "/api/v1/subscribe/{subscription_id}/logs/download", "/api/v1/system/logging", "/api/v1/system/logging/download/{name}", "/api/v1/system/message", diff --git a/scripts/validation/subscription_governance_scale.py b/scripts/validation/subscription_governance_scale.py index 9454aa019..875c2c5c9 100644 --- a/scripts/validation/subscription_governance_scale.py +++ b/scripts/validation/subscription_governance_scale.py @@ -790,6 +790,10 @@ def _run_match_execution_case(case: ScaleCase) -> dict[str, Any]: def debug(*_args: Any, **_kwargs: Any) -> None: """忽略受控样本的逐项诊断日志。""" + @staticmethod + def scoped_info(*_args: Any, **_kwargs: Any) -> None: + """独立订阅日志不计入主日志的批次 INFO 预算。""" + @staticmethod def warning(*_args: Any, **_kwargs: Any) -> None: """忽略受控样本中预期不会出现的警告。""" diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 064126818..031ea9bb4 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -265,6 +265,14 @@ Select the operation for the task first, then send only fields declared by that | `download.tasks.active` | none | optional `page`, `count`, `name` | | `site.list` | none | optional `page`, `count`, `name`, `status=all\|active\|inactive` | +Subscription execution logs are browser resource endpoints: +`GET /api/v1/subscribe/{subscription_id}/logs` (SSE, or text with `length=-1`) +and `/logs/download` (ZIP). They require the existing resource Cookie and enforce +subscription ownership; they are not JSON gateway or MCP tools. Do not pass a +logfile path or try to call them through `call_api`. Read subscription status +through the supported subscription operations; completion of a search task does +not prove completion of the subscription. + The `download.add` body must contain `torrent_in` (at least `title` and `enclosure`) plus sibling `media_source` and `media_id`; do not put a magnet URI in `url`, or move media identity and filters into `query`. When a write returns `unknown`, never retry it; verify the actual state with a supported read operation first. Preserve the candidate's `site` ID and download URL. For direct HTTP(S) torrents, diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index 0733fb435..9d9e2826f 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -55,7 +55,6 @@ "app/chain/subscribe/completion.py:SubscribeCompletionOwner._SubscribeChain__get_subscribe_no_exits": 16, "app/chain/subscribe/completion.py:SubscribeCompletionOwner._SubscribeChain__record_subscribe_download_facts": 19, "app/chain/subscribe/match.py:SubscribeMatchOwner._execute_match": 47, - "app/chain/subscribe/match.py:SubscribeMatchOwner._run_match": 18, "app/chain/subscribe/query.py:SubscribeQueryOwner.follow": 19, "app/chain/subscribe/query.py:SubscribeQueryOwner.subscribe_files_info": 24, "app/chain/subscribe/refresh.py:SubscribeRefreshOwner.backfill_existing_episodes": 16, @@ -148,7 +147,6 @@ "app/chain/scraping.py:ScrapingChain._download_and_save_image": 7, "app/chain/scraping.py:ScrapingChain.scrape_metadata_event": 6, "app/chain/subscribe/match.py:SubscribeMatchOwner._execute_match": 7, - "app/chain/subscribe/match.py:SubscribeMatchOwner._run_match": 6, "app/chain/subscribe/query.py:SubscribeQueryOwner.subscribe_files_info": 6, "app/chain/transfer/queue.py:TransferQueueOwner._TransferChain__start_transfer": 6, "app/chain/transfer/workflow.py:TransferWorkflowOwner._build_transfer_tasks": 8, diff --git a/tests/fixtures/architecture/concurrency-baseline.json b/tests/fixtures/architecture/concurrency-baseline.json index 3be86493a..a7d81c615 100644 --- a/tests/fixtures/architecture/concurrency-baseline.json +++ b/tests/fixtures/architecture/concurrency-baseline.json @@ -654,7 +654,7 @@ }, "app/startup/composition/system.py:asyncio.to_thread": { "owners": { - "_FileLogAdapter._follow_path": 3, + "_FileLogAdapter._follow_path": 1, "_FileLogAdapter._resolve_logfile": 1, "_FileLogAdapter.collect": 1 }, diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 4ced99d22..719465162 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -941,8 +941,8 @@ "runtime_only": true } }, - "edge_count": 9437, - "edge_sha256": "5ec570cd4243faee56bbc810f7fa74b2d14cc35b29b56a6424f491f6a73191a5", + "edge_count": 9458, + "edge_sha256": "ebdaf426be6696641961d4b3fda2081b36f812d18c86b313ece2bd122253f1c0", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3025,19 +3025,33 @@ "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.adapters", + "app.api.endpoints.subexecution -> app.adapters.web", + "app.api.endpoints.subexecution -> app.adapters.web.security", + "app.api.endpoints.subexecution -> app.adapters.web.security.access", "app.api.endpoints.subexecution -> app.api", + "app.api.endpoints.subexecution -> app.api.context", "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.endpoints", + "app.api.endpoints.subexecution -> app.api.endpoints.system", "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.query", "app.api.endpoints.subexecution -> app.application.subscription.status", + "app.api.endpoints.subexecution -> app.application.system", + "app.api.endpoints.subexecution -> app.runtime", + "app.api.endpoints.subexecution -> app.runtime.log", "app.api.endpoints.subexecution -> app.schemas", "app.api.endpoints.subexecution -> app.schemas.response", "app.api.endpoints.subexecution -> app.schemas.subscribe", + "app.api.endpoints.subexecution -> app.schemas.token", + "app.api.endpoints.subexecution -> app.startup", + "app.api.endpoints.subexecution -> app.startup.composition", + "app.api.endpoints.subexecution -> app.startup.composition.context", "app.api.endpoints.submaintenance -> app.api", "app.api.endpoints.submaintenance -> app.api.dependencies", "app.api.endpoints.submaintenance -> app.api.dependencies.auth", @@ -4174,6 +4188,9 @@ "app.application.subscription.mutation -> app.application.subscription", "app.application.subscription.mutation -> app.application.subscription.contract", "app.application.subscription.mutation -> app.application.subscription.delete", + "app.application.subscription.mutation -> app.application.subscription.observability", + "app.application.subscription.mutation -> app.runtime", + "app.application.subscription.mutation -> app.runtime.log", "app.application.subscription.mutation -> app.schemas", "app.application.subscription.mutation -> app.schemas.common", "app.application.subscription.mutation -> app.schemas.event", @@ -4182,6 +4199,8 @@ "app.application.subscription.observability -> app.application.subscription", "app.application.subscription.observability -> app.application.subscription.execution", "app.application.subscription.observability -> app.application.subscription.sitebudget", + "app.application.subscription.observability -> app.runtime", + "app.application.subscription.observability -> app.runtime.log", "app.application.subscription.priority -> app.application", "app.application.subscription.priority -> app.application.subscription", "app.application.subscription.priority -> app.application.subscription.contract", @@ -5310,6 +5329,7 @@ "app.chain.subscribe.completion -> app.application.subscription", "app.chain.subscribe.completion -> app.application.subscription.contract", "app.chain.subscribe.completion -> app.application.subscription.mutation", + "app.chain.subscribe.completion -> app.application.subscription.observability", "app.chain.subscribe.completion -> app.chain", "app.chain.subscribe.completion -> app.chain.subscribe", "app.chain.subscribe.completion -> app.chain.subscribe.contract", @@ -5528,6 +5548,7 @@ "app.chain.subscribe.refresh -> app.application.subscription", "app.chain.subscribe.refresh -> app.application.subscription.contract", "app.chain.subscribe.refresh -> app.application.subscription.facts", + "app.chain.subscribe.refresh -> app.application.subscription.observability", "app.chain.subscribe.refresh -> app.application.subscription.priority", "app.chain.subscribe.refresh -> app.chain", "app.chain.subscribe.refresh -> app.chain.download", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index c252f7759..8f867a81b 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -901,7 +901,7 @@ }, "app/chain/subscribe/completion.py": { "arg-type": 2, - "assignment": 3, + "assignment": 1, "no-untyped-call": 1, "operator": 3, "type-var": 1, diff --git a/tests/test_api_response.py b/tests/test_api_response.py index 8ca23356d..6c4389461 100644 --- a/tests/test_api_response.py +++ b/tests/test_api_response.py @@ -680,6 +680,8 @@ def test_v1_model_free_routes_match_audited_native_allowlist(): ("/system/notification-image", "notification_image"), ("/system/progress/{process_type}", "get_progress"), ("/system/message", "get_message"), + ("/subscribe/{subscription_id}/logs", "get_subscription_logs"), + ("/subscribe/{subscription_id}/logs/download", "download_subscription_logs"), ("/system/logging", "get_logging"), ("/system/logging/download/{name}", "download_logging"), ( diff --git a/tests/test_subscription_logging.py b/tests/test_subscription_logging.py new file mode 100644 index 000000000..645b9f6af --- /dev/null +++ b/tests/test_subscription_logging.py @@ -0,0 +1,323 @@ +"""订阅日志的上下文隔离、权限、脱敏及文件生命周期回归。""" + +import asyncio +import io +import logging +import threading +import zipfile +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +import pytest +from fastapi import HTTPException + +from app.api.endpoints import subexecution +from app.application.system import LogFileData, LogNotFoundError +from app.runtime import log as log_module +from app.runtime.execution import submit_with_context +from app.runtime.log import LoggerManager, NonBlockingFileHandler, bind_log_scope, logger +from app.startup.composition.system import _FileLogAdapter + + +@pytest.fixture(name="captured") +def fixture_captured(monkeypatch, tmp_path): + """在内存中捕获文件路由,隔离全局等级及控制台输出。""" + entries = [] + monkeypatch.setattr(LoggerManager, "_writer", SimpleNamespace(write_log=lambda *args: entries.append(args))) + monkeypatch.setattr(LoggerManager, "_log_path", tmp_path) + monkeypatch.setattr(log_module.log_settings, "DEBUG", False) + monkeypatch.setattr(log_module.log_settings, "LOG_LEVEL", "INFO") + monkeypatch.setattr(LoggerManager, "_get_console_logger", classmethod(lambda _cls, _path: logging.getLogger("subscription-test"))) + return entries + + +def test_debug_routes_to_subscription_without_enabling_main_log(captured): + """单订阅 DEBUG 独立落盘;主日志仍只接收原等级,专属摘要不污染主日志。""" + logger.debug("outside") + with bind_log_scope("subscriptions", 7, "run-7", "manual"): + logger.debug("分辨率不符合要求") + logger.info("正常业务日志") + logger.scoped_info("本轮新增下载 0 个") + logger.debug("outside-again") + assert [path.name for _, _, path in captured] == ["7.log", "7.log", "moviepilot.log", "7.log"] + assert all("run-7" in message for _, message, path in captured if path.name == "7.log") + + +def test_nested_context_and_exception_restore_previous_subscription(captured): + """不同资源嵌套和异常退出恢复外层 ID,同资源嵌套复用原轮次。""" + with bind_log_scope("subscriptions", 1, "outer", "rss"): + with bind_log_scope("subscriptions", 1, "ignored", "inline"): + logger.debug("same") + with pytest.raises(RuntimeError): + with bind_log_scope("subscriptions", 2, "inner", "manual"): + logger.debug("inner") + raise RuntimeError("stop") + logger.debug("outer") + logger.info("outside") + assert [path.name for _, _, path in captured] == ["1.log", "2.log", "1.log", "moviepilot.log"] + assert "outer" in captured[0][1] and "ignored" not in captured[0][1] + + +def test_parallel_tasks_and_context_aware_threads_do_not_mix_logs(captured): + """交错协程及受控线程池继承各自 ID,写入队列前已固定日志路由。""" + async def run(): + """制造协程交错和线程边界,保留真实上下文传播。""" + with ThreadPoolExecutor(max_workers=2) as pool: + async def task(subscription_id): + """绑定每条订阅后跨越任务切换和线程提交。""" + with bind_log_scope("subscriptions", subscription_id, f"run-{subscription_id}", "manual"): + await asyncio.sleep(0) + await asyncio.wrap_future(submit_with_context(pool, logger.debug, f"订阅 {subscription_id}")) + await asyncio.gather(task(3), task(4)) + asyncio.run(run()) + assert {path.name for _, _, path in captured} == {"3.log", "4.log"} + for _, message, path in captured: + assert f"订阅 {path.stem}" in message + + +@pytest.mark.parametrize("message", [ + "下载 https://user:password@example.com/download?passkey=SECRET", + "Cookie: session=SECRET; uid=9", + "Authorization: Bearer SECRET", + '{"api_token": "SECRET"}', + "magnet:?xt=urn:btih:SECRET", +]) +def test_credentials_are_redacted_before_subscription_write(captured, message): + """原始消息中的常见凭据和下载链接不能进入订阅文件。""" + with bind_log_scope("subscriptions", 1, "run", "manual"): + logger.debug(message) + assert "SECRET" not in captured[0][1] + + +def test_detail_limits_preserve_error_and_summary(captured): + """详情数量和重复次数都有上限,但错误及结束摘要始终保留。""" + with bind_log_scope("subscriptions", 1, "run", "rss") as scope: + for _ in range(10): + logger.debug("重复原因") + for index in range(400): + logger.debug(f"资源 {index} 不满足规则") + logger.error("下载失败") + logger.scoped_info("新增下载 0 个") + scoped = [message for _, message, path in captured if path.name == "1.log"] + assert len(scoped) == 302 + assert scope.suppressed == 110 + assert sum("重复原因" in item for item in scoped) == 3 + assert "下载失败" in scoped[-2] and "新增下载 0 个" in scoped[-1] + + +@pytest.fixture(name="file_writer") +def fixture_file_writer(): + """只构造同步文件输出状态,避免本组文件回归创建后台线程或修改 singleton。""" + writer = object.__new__(NonBlockingFileHandler) + writer._handlers_lock = threading.RLock() + writer._state_lock = threading.RLock() + writer._rotating_handlers = {} + writer._scoped_handler_times = {} + writer._scoped_cleanup_at = 0.0 + yield writer + writer._close_handlers() + + +def test_subscription_handlers_are_bounded_and_rotation_is_small(file_writer, tmp_path, monkeypatch): + """大量订阅有界复用句柄,每份日志采用独立的小容量滚动策略。""" + monkeypatch.setattr(log_module, "_SCOPED_LOG_MAX_BYTES", 200) + for subscription_id in range(1, 45): + path = tmp_path / "subscriptions" / f"{subscription_id}.log" + file_writer._write_sync(log_module.LogEntry("DEBUG", "x" * 100, path)) + assert len(file_writer._scoped_handler_times) == 32 + assert len(file_writer._rotating_handlers) == 32 + path = tmp_path / "subscriptions" / "44.log" + for _ in range(8): + file_writer._write_sync(log_module.LogEntry("DEBUG", "x" * 100, path)) + assert (path.parent / "44.log.2").is_file() + assert not (path.parent / "44.log.3").exists() + + +def test_directory_quota_preserves_active_file_and_other_files(file_writer, tmp_path, monkeypatch): + """总容量清理只删除旧订阅日志,保留当前日志和目录内其他文件。""" + directory = tmp_path / "subscriptions" + directory.mkdir() + for name in ("1.log", "1.log.1", "2.log", "notes.txt"): + (directory / name).write_text("x" * 100) + monkeypatch.setattr(log_module, "_SCOPED_LOG_TOTAL_BYTES", 100) + file_writer._trim_scoped_directory(directory / "2.log") + assert (directory / "2.log").exists() + assert (directory / "notes.txt").exists() + assert not (directory / "1.log").exists() + assert not (directory / "1.log.1").exists() + + +@pytest.fixture(name="adapter") +def fixture_adapter(tmp_path): + """提供只指向临时日志目录的真实文件读取适配器。""" + return _FileLogAdapter(SimpleNamespace(get=lambda _key: tmp_path)) + + +def test_file_cursor_handles_first_write_partial_line_and_rotation(adapter, tmp_path): + """首次无文件、分段写行和换 inode 的轮转都不遗漏或重复已完成的行。""" + path = tmp_path / "subscriptions" / "1.log" + assert adapter._log_chunk(path, None, 50) == ([], None) + path.parent.mkdir() + path.write_text("first\npartial") + lines, cursor = adapter._log_chunk(path, None, 50) + assert lines == ["first"] + with path.open("a") as file: + file.write(" done\n") + lines, cursor = adapter._log_chunk(path, cursor, 0) + assert lines == ["partial done"] + assert adapter._log_chunk(path, cursor, 0)[0] == [] + path.rename(path.with_suffix(".log.1")) + path.write_text("rotated\n") + assert adapter._log_chunk(path, cursor, 0)[0] == ["rotated"] + + +def test_subscription_symlink_cannot_read_another_log(adapter, tmp_path): + """即使符号链接仍在日志根内,也不能绕过订阅 ID 读取其他日志。""" + (tmp_path / "subscriptions").mkdir() + secret = tmp_path / "moviepilot.log" + secret.write_text("private") + path = tmp_path / "subscriptions" / "1.log" + path.symlink_to(secret) + with pytest.raises(LogNotFoundError): + adapter._log_chunk(path, None, 50) + with pytest.raises(LogNotFoundError): + asyncio.run(adapter.read("subscriptions/1.log")) + with pytest.raises(LogNotFoundError): + adapter._collect_sync("subscriptions/1") + + +@pytest.mark.parametrize("owner,username,admin,allowed", [ + ("alice", "alice", False, True), ("alice", "bob", False, False), + ("alice", "bob", True, True), (None, "alice", True, False), +]) +def test_subscription_log_access_matches_owner_scope(owner, username, admin, allowed): + """普通用户仅能看自己的订阅,管理员也不能查询不存在的 ID。""" + async def get_public(_id): + """只返回本用例设置的订阅归属。""" + return SimpleNamespace(username=owner) if owner else None + query = SimpleNamespace(get_public=get_public) + user = SimpleNamespace(username=username, super_user=admin) + if allowed: + asyncio.run(subexecution._require_subscription_log_access(1, query, user)) + else: + with pytest.raises(HTTPException) as error: + asyncio.run(subexecution._require_subscription_log_access(1, query, user)) + assert error.value.status_code == 404 + + +def test_download_only_collects_requested_id_and_redacts_archive(monkeypatch): + """下载路径由服务端 ID 构造,ZIP 内容再次脱敏且不混入其他订阅。""" + collected = [] + async def get_public(_id): + """返回已授权 owner。""" + return SimpleNamespace(username="alice") + async def collect(name): + """记录服务端构造的日志集合并返回含旧凭据的文本。""" + collected.append(name) + return [LogFileData("1.log", b"Cookie: SECRET\nreason\n")] + async def authorize(subscription_id, user, _runtime): + """复用真实归属校验,仅替换短会话装配。""" + await subexecution._require_subscription_log_access(subscription_id, SimpleNamespace(get_public=get_public), user) + monkeypatch.setattr(subexecution, "_authorize_subscription_logs", authorize) + response = asyncio.run(subexecution.download_subscription_logs( + 1, SimpleNamespace(username="alice", super_user=False), + SimpleNamespace(system=SimpleNamespace(collect_logs=collect)), + )) + async def body(): + """消费流响应以检查真实归档内容。""" + return b"".join([chunk async for chunk in response.body_iterator]) + archive = zipfile.ZipFile(io.BytesIO(asyncio.run(body()))) + assert collected == ["subscriptions/1"] + assert all("SECRET" not in archive.read(name).decode() for name in archive.namelist()) + assert len(archive.namelist()) == 1 + + +def test_log_permission_checks_release_session_before_streaming(): + """每次 SSE 授权仅短暂打开一个会话,成功和拒绝路径都立即关闭。""" + from app.application.subscription.contract import SubscriptionSnapshot + states = [] + async def sessions(): + """记录权限检查所持有会话的完整生命周期。""" + states.append("open") + try: + yield object() + finally: + states.append("close") + async def async_get(_id): + """提供可投影的真实订阅快照。""" + return SubscriptionSnapshot(id=1, name="测试", type="电影", username="alice") + runtime = SimpleNamespace(subscription=SimpleNamespace( + async_session=sessions, repository=lambda _session: SimpleNamespace(async_get=async_get), + )) + asyncio.run(subexecution._authorize_subscription_logs(1, SimpleNamespace(username="alice", super_user=False), runtime)) + assert states == ["open", "close"] + with pytest.raises(HTTPException): + asyncio.run(subexecution._authorize_subscription_logs(1, SimpleNamespace(username="bob", super_user=False), runtime)) + assert states == ["open", "close", "open", "close"] + + +def test_sse_waits_for_first_log_and_stops_after_owner_changes(monkeypatch): + """保留真实 SSE 编码,验证空流心跳、首次日志与权限撤销停止。""" + clock = [0.0] + checks = [] + async def authorize(_id, _user, _runtime): + """第二次归属检查模拟订阅已被删除或转移。""" + checks.append(True) + if len(checks) > 1: + raise HTTPException(status_code=404) + async def is_disconnected(): + """模拟连接始终在线。""" + return False + async def source(disconnected): + """在空文件心跳后输出一次日志,再触发权限轮询。""" + assert not await disconnected() + yield "" + yield "下载 https://example.com/?token=SECRET" + clock[0] = 6.0 + assert await disconnected() + async def follow(_path, _length, disconnected): + """遵守 SystemLogPort 的异步返回迭代器契约。""" + return source(disconnected) + monkeypatch.setattr(subexecution, "_authorize_subscription_logs", authorize) + monkeypatch.setattr(subexecution, "verify_resource_token", lambda _token: SimpleNamespace()) + monkeypatch.setattr(subexecution.time, "monotonic", lambda: clock[0]) + async def run(): + """消费路由输出,确认 SSE framing 和二次脱敏。""" + response = await subexecution.get_subscription_logs( + SimpleNamespace(is_disconnected=is_disconnected, cookies={"MoviePilot": "resource-token"}), + 1, 100, SimpleNamespace(username="alice", super_user=False), + SimpleNamespace(system=SimpleNamespace(follow_log=follow, settings=SimpleNamespace(get=lambda _key: "MoviePilot"))), + ) + return [chunk async for chunk in response.body_iterator] + content = asyncio.run(run()) + assert content[0] == ": keepalive\n\n" + assert "SECRET" not in content[1] + assert len(checks) == 2 + + +def test_log_routes_require_resource_cookie_and_positive_subscription_id(): + """两个新路由必须使用浏览器资源身份,不能通过日志路径参数绕过归属。""" + from app.adapters.web.security.access import verify_resource_token + routes = {route.path: route for route in subexecution.router.routes} + for path in ("/{subscription_id}/logs", "/{subscription_id}/logs/download"): + route = routes[path] + assert verify_resource_token in {dependency.call for dependency in route.dependant.dependencies} + assert {field.name for field in route.dependant.path_params} == {"subscription_id"} + assert "logfile" not in {field.name for field in route.dependant.query_params} + + +@pytest.mark.parametrize("asynchronous", [False, True]) +def test_unsuccessful_reset_does_not_report_success(captured, monkeypatch, asynchronous): + """更新阶段未取得快照时,日志不能把未提交的重置描述为成功。""" + from unittest.mock import AsyncMock + from app.application.subscription.contract import SubscriptionSnapshot + from app.application.subscription.mutation import SubscriptionMutationService, SyncSubscriptionMutationService + snapshot = SubscriptionSnapshot(id=7, name="测试", type="电影", username="alice") + service_type = SubscriptionMutationService if asynchronous else SyncSubscriptionMutationService + service = object.__new__(service_type) + monkeypatch.setattr(service, "get_accessible", AsyncMock(return_value=snapshot) if asynchronous else lambda *_args: snapshot) + monkeypatch.setattr(service, "update", AsyncMock(return_value=None) if asynchronous else lambda *_args, **_kwargs: None) + result = asyncio.run(service.reset(7, None)) if asynchronous else service.reset(7, None) + assert result is None + assert any("本次重置未完成" in message for _, message, _ in captured) + assert all("订阅进度已重置" not in message for _, message, _ in captured)