From 68cfc377ee26bdde1db82e532ba58256f2145079 Mon Sep 17 00:00:00 2001 From: freeman Date: Wed, 4 Nov 2026 06:24:38 +0800 Subject: [PATCH] feat(history): register externally completed transfers (#7293) --- app/adapters/system/fsproxy.py | 49 +- app/adapters/system/fsworker.py | 27 + app/adapters/system/registration.py | 63 ++ app/agent/policy/api.py | 2 + app/agent/policy/mcp.py | 5 + .../policy/resources/api_mcp_schema.json | 226 +++++ app/api/dependencies/history.py | 13 + app/api/endpoints/history.py | 57 +- app/application/history/__init__.py | 7 +- app/application/history/contracts.py | 14 +- app/application/history/registration.py | 405 ++++++++ app/db/adapters/history/registration.py | 79 ++ app/db/adapters/history/transfer.py | 12 +- app/db/models/transferhistory.py | 53 +- app/db/oper/transferhistory.py | 41 +- app/locales/en-US.json | 23 +- app/locales/zh-TW.json | 23 +- app/schemas/exports.py | 3 + app/schemas/history.py | 90 ++ app/startup/composition/context.py | 5 + app/startup/composition/runtime.py | 5 + ...9e613a72f_external_history_registration.py | 37 + docs/architecture-overview.md | 6 +- docs/mcp-api.md | 24 + scripts/generate_agent_skill_docs.py | 21 +- skills/database-operation/SKILL.md | 6 +- skills/moviepilot-api/SKILL.md | 12 +- skills/moviepilot-api/api/transfer.md | 15 +- .../architecture/dependency-baseline.json | 40 +- .../startup-performance-baseline.json | 6 +- tests/test_agent_api_bodies.py | 6 + tests/test_agent_api_gateway.py | 4 +- tests/test_agent_api_identity.py | 2 +- tests/test_agent_skills_middleware.py | 2 +- tests/test_builtin_skill_boundaries.py | 4 +- tests/test_external_transfer_registration.py | 879 ++++++++++++++++++ tests/test_host_runtime_context.py | 2 + tests/test_runtime_composition.py | 21 + ..._subscription_identity_review_migration.py | 3 +- 39 files changed, 2220 insertions(+), 72 deletions(-) create mode 100644 app/adapters/system/registration.py create mode 100644 app/application/history/registration.py create mode 100644 app/db/adapters/history/registration.py create mode 100644 database/versions/b849e613a72f_external_history_registration.py create mode 100644 tests/test_external_transfer_registration.py diff --git a/app/adapters/system/fsproxy.py b/app/adapters/system/fsproxy.py index 868622557..085cc8fc6 100644 --- a/app/adapters/system/fsproxy.py +++ b/app/adapters/system/fsproxy.py @@ -21,6 +21,7 @@ import json import os import selectors import shutil +import stat import subprocess import sys import threading @@ -86,6 +87,15 @@ class FileSystemProxy: """ return cast(dict[str, Any], self._call("stat", path=str(path))) + def lstat(self, path: Path) -> Dict[str, Any]: + """ + 不跟随最终符号链接读取目录项，供外部整理登记核验。 + + :param path: 待核验的原始文件地址 + :return: 文件身份、纳秒时间、类型和符号链接标志 + """ + return cast(dict[str, Any], self._call("lstat", path=str(path))) + def exists(self, path: Path) -> bool: """ 判断路径是否存在。 @@ -297,6 +307,32 @@ class FileSystemProxy: """ return bool(get_runtime_setting("FS_PROXY_ENABLED", True)) + @staticmethod + def _direct_stat(op: str, path: str) -> Dict[str, Any]: + """ + 读取直连模式下的路径或目录项属性，保留既有 stat 跟随链接的语义。 + + :param op: stat 或 lstat + :param path: 目标路径 + :return: 对应操作的固定属性集合 + """ + if op == "lstat": + info = os.lstat(path) + return { + "size": info.st_size, "mtime": info.st_mtime, + "mtime_ns": info.st_mtime_ns, "ctime_ns": info.st_ctime_ns, + "dev": info.st_dev, "ino": info.st_ino, + "is_dir": stat.S_ISDIR(info.st_mode), + "is_file": stat.S_ISREG(info.st_mode), + "is_symlink": stat.S_ISLNK(info.st_mode), + } + info = os.stat(path) + return { + "size": info.st_size, "mtime": info.st_mtime, + "dev": info.st_dev, "ino": info.st_ino, + "is_dir": os.path.isdir(path), "is_file": os.path.isfile(path), + } + @staticmethod def _direct(op: str, payload: Dict[str, Any]) -> Any: """ @@ -306,17 +342,8 @@ class FileSystemProxy: :param payload: 操作参数 :return: 操作结果 """ - if op == "stat": - path = payload["path"] - info = os.stat(path) - return { - "size": info.st_size, - "mtime": info.st_mtime, - "dev": info.st_dev, - "ino": info.st_ino, - "is_dir": os.path.isdir(path), - "is_file": os.path.isfile(path), - } + if op in ("stat", "lstat"): + return FileSystemProxy._direct_stat(op, payload["path"]) if op == "realpath": return str(Path(payload["path"]).resolve()) if op == "exists": diff --git a/app/adapters/system/fsworker.py b/app/adapters/system/fsworker.py index fec4b4108..79b61818e 100644 --- a/app/adapters/system/fsworker.py +++ b/app/adapters/system/fsworker.py @@ -20,6 +20,7 @@ import json import os import shutil +import stat import sys import time from pathlib import Path @@ -74,6 +75,16 @@ class _StatResult(TypedDict): is_file: bool +class _EntryStatResult(_StatResult): + """ + 非跟随查询在普通属性上补充最终目录项的符号链接事实。 + + :ivar is_symlink: 最终目录项是否为符号链接，包括悬空链接 + """ + + is_symlink: bool + + class _CopyProgress(TypedDict): """ 复制期间的进度心跳或结束时的结果，所有键均存在。 @@ -140,6 +151,21 @@ def _stat(payload: _Request, _emit: _Emit) -> _StatResult: } +def _lstat(payload: _Request, _emit: _Emit) -> _EntryStatResult: + """ + 只查询最终目录项，不跟随符号链接，所有类型均来自同一次 lstat。 + """ + info = os.lstat(payload["path"]) + return { + "size": info.st_size, "mtime": info.st_mtime, + "mtime_ns": info.st_mtime_ns, "ctime_ns": info.st_ctime_ns, + "dev": info.st_dev, "ino": info.st_ino, + "is_dir": stat.S_ISDIR(info.st_mode), + "is_file": stat.S_ISREG(info.st_mode), + "is_symlink": stat.S_ISLNK(info.st_mode), + } + + def _exists(payload: _Request, _emit: _Emit) -> bool: """ 判断路径是否存在。 @@ -279,6 +305,7 @@ def _rmtree(payload: _Request, _emit: _Emit) -> bool: _HANDLERS: dict[str, Callable[[_Request, _Emit], _Result]] = { "stat": _stat, + "lstat": _lstat, "realpath": _realpath, "exists": _exists, "listdir": _listdir, diff --git a/app/adapters/system/registration.py b/app/adapters/system/registration.py new file mode 100644 index 000000000..01843ea88 --- /dev/null +++ b/app/adapters/system/registration.py @@ -0,0 +1,63 @@ +""" +为外部历史登记提供可超时且不跟随最终符号链接的本地目录项证明。 +""" + +from pathlib import Path +from typing import Protocol, TypeVar + +from app.adapters.system.fsproxy import fsproxy + +EntryT = TypeVar("EntryT", covariant=True) + + +class LocalEntryFactory(Protocol[EntryT]): + """ + 组合根注入的有界目录项构造能力，系统适配器不依赖消费方 DTO。 + """ + + def __call__( + self, *, dev: int, ino: int, size: int, mtime_ns: int, ctime_ns: int, + is_file: bool, is_symlink: bool, parent_dev: int, parent_ino: int, + canonical_parent: str, name: str, + ) -> EntryT: + """ + 从固定字段构造消费方的冻结事实对象。 + + :param dev: 文件设备号 + :param ino: 文件节点号 + :param size: 文件字节数 + :param mtime_ns: 纳秒修改时间 + :param ctime_ns: 纳秒状态变化时间 + :param is_file: 是否为普通文件 + :param is_symlink: 最终目录项是否为符号链接 + :param parent_dev: 父目录设备号 + :param parent_ino: 父目录节点号 + :param canonical_parent: 仅解析父目录后的绝对地址 + :param name: 原请求最终目录项名称 + :return: 消费方定义的目录项事实 + """ + ... + + +def probe_local_entry(path: str, *, entry_factory: LocalEntryFactory[EntryT]) -> EntryT | None: + """ + 读取文件和真实父目录身份，保留原目录项名称供大小写别名判定。 + + :param path: 原请求中的规范绝对地址 + :param entry_factory: 组合根提供的固定字段 DTO 构造器 + :return: 普通或符号链接目录项事实；仅确认不存在时返回 None + """ + target = Path(path) + try: + entry = fsproxy.lstat(target) + except FileNotFoundError: + return None + parent = fsproxy.realpath(target.parent) + parent_stat = fsproxy.stat(parent) + return entry_factory( + dev=entry["dev"], ino=entry["ino"], size=entry["size"], + mtime_ns=entry["mtime_ns"], ctime_ns=entry["ctime_ns"], + is_file=entry["is_file"], is_symlink=entry["is_symlink"], + parent_dev=parent_stat["dev"], parent_ino=parent_stat["ino"], + canonical_parent=parent.as_posix(), name=target.name, + ) diff --git a/app/agent/policy/api.py b/app/agent/policy/api.py index 9f181fbf2..0a3a65505 100644 --- a/app/agent/policy/api.py +++ b/app/agent/policy/api.py @@ -206,6 +206,7 @@ API_FIRST_BATCH_OPERATION_SPECS: tuple[ApiOperationSpec, ...] = ( _admin_read("storage.settings"), _spec("storage.list", result_sensitivity=ResultSensitivity.PRIVATE), _spec("transfer.history", result_sensitivity=ResultSensitivity.PRIVATE), + _write("transfer.history.register", sensitivity=ResultSensitivity.PRIVATE), _spec( "transfer.file", effect=ActionEffect.EXTERNAL_SIDE_EFFECT, @@ -716,6 +717,7 @@ API_OPERATION_ROUTES: dict[str, ApiOperationRoute] = { "storage.settings": ApiOperationRoute("GET", "/api/v1/storage/directories"), "storage.list": ApiOperationRoute("POST", "/api/v1/storage/agent/list"), "transfer.history": ApiOperationRoute("GET", "/api/v1/history/transfer"), + "transfer.history.register": ApiOperationRoute("POST", "/api/v1/history/transfer"), "transfer.file": ApiOperationRoute("POST", "/api/v1/transfer/manual"), "scheduler.list": ApiOperationRoute("GET", "/api/v1/dashboard/schedule"), "scheduler.run": ApiOperationRoute("GET", "/api/v1/system/runscheduler"), diff --git a/app/agent/policy/mcp.py b/app/agent/policy/mcp.py index 1e98bfddb..6665ee0cf 100644 --- a/app/agent/policy/mcp.py +++ b/app/agent/policy/mcp.py @@ -99,6 +99,7 @@ OPERATION_DESCRIPTIONS = { "system.versions": "列出 GitHub 中可用的 MoviePilot 发布版本。", "transfer.file": "执行手动文件转移与整理流程。", "transfer.history": "按条件和分页读取文件整理历史。", + "transfer.history.register": "管理员登记已完成的外部复制、原生硬链接或移动；严格核验文件事实并记录审计，不操作文件或下载器，同一声明重试返回原回执。", "transfer.history.delete": "删除一条整理历史，并可选择同时删除文件。", "workflow.list": "列出已配置工作流及执行状态。", "workflow.run": "从起点或恢复点执行一个已配置工作流。", @@ -247,6 +248,7 @@ OPERATION_DESCRIPTIONS = { FIELD_DESCRIPTIONS = { + "origin": "整理历史来源；external 仅返回管理员登记的外部整理记录，省略时保留全部历史。", "include_value": "是否在完整设置契约之外返回当前值。", "offset": "结果偏移量，从 0 开始。", "rule_ids": "要返回的精确规则标识；在查询串中重复传入 rule_ids，省略时列出该操作对应的全部规则。", @@ -736,6 +738,9 @@ MODEL_DESCRIPTIONS.update( "EpisodeFormatRecommendItem": "用于推断集号提取模板的文件样本。", "BatchTransferHistoryRedoRequest": "AI 辅助批量重新整理中明确指定的历史记录标识。", "TransferHistoryReplanRequest": "对零文件副作用的识别拒绝提交已核实的媒体身份及季集修正。", + "TransferHistoryRegisterRequest": "管理员声明已完成的外部整理路径、模式和媒体身份，文件事实由服务器核验。", + "TransferRegistrationAudit": "服务器记录的外部登记操作者、UTC 时间和文件核验证据。", + "TransferHistoryRegisterResult": "已提交的外部登记回执，重复声明返回原历史和 created=false。", "TransferManualReviewRequest": "持久化整理操作的已授权决定和可选结果。", "Workflow-Input": "创建或更新操作接受的完整工作流定义。", "Action-Input": "工作流中的一个可执行动作节点。", diff --git a/app/agent/policy/resources/api_mcp_schema.json b/app/agent/policy/resources/api_mcp_schema.json index cc1f92673..ee1008c83 100644 --- a/app/agent/policy/resources/api_mcp_schema.json +++ b/app/agent/policy/resources/api_mcp_schema.json @@ -5950,6 +5950,144 @@ "title": "TransferHistory", "type": "object" }, + "TransferHistoryRegisterRequest": { + "additionalProperties": false, + "description": "管理员声明已完成的外部整理路径、模式和媒体身份，文件事实由服务器核验。", + "properties": { + "dest": { + "description": "整理后目标文件的绝对路径", + "maxLength": 4096, + "minLength": 1, + "title": "Dest", + "type": "string" + }, + "dest_storage": { + "description": "目标存储配置标识", + "maxLength": 64, + "minLength": 1, + "title": "Dest Storage", + "type": "string" + }, + "download_hash": { + "anyOf": [ + { + "maxLength": 128, + "type": "string" + }, + { + "type": "null" + } + ], + "description": "已知下载任务标识，仅作关联声明", + "title": "Download Hash" + }, + "episodes": { + "anyOf": [ + { + "maxLength": 32, + "pattern": "^E(?:0[1-9]|[1-9]\\d{1,3})(?:-E(?:0[1-9]|[1-9]\\d{1,3}))?$", + "type": "string" + }, + { + "type": "null" + } + ], + "description": "集号或集范围，例如 E01 或 E01-E12", + "title": "Episodes" + }, + "media_id": { + "description": "来源中的规范原生媒体标识", + "maxLength": 128, + "minLength": 1, + "title": "Media Id", + "type": "string" + }, + "media_source": { + "$ref": "#/$defs/MediaSource", + "description": "元数据来源标识，保留与 media_id 一同返回的精确值。" + }, + "mode": { + "description": "已完成的复制、本地硬链接或移动", + "enum": [ + "copy", + "link", + "move" + ], + "title": "Mode", + "type": "string" + }, + "seasons": { + "anyOf": [ + { + "maxLength": 32, + "pattern": "^S\\d{2,4}(?:-S\\d{2,4})?$", + "type": "string" + }, + { + "type": "null" + } + ], + "description": "季号或季范围，例如 S01", + "title": "Seasons" + }, + "src": { + "description": "整理前源文件的绝对路径", + "maxLength": 4096, + "minLength": 1, + "title": "Src", + "type": "string" + }, + "src_storage": { + "description": "源存储配置标识", + "maxLength": 64, + "minLength": 1, + "title": "Src Storage", + "type": "string" + }, + "title": { + "description": "管理员确认的媒体标题", + "maxLength": 512, + "minLength": 1, + "title": "Title", + "type": "string" + }, + "type": { + "description": "管理员确认的电影或电视剧类型", + "enum": [ + "电影", + "电视剧" + ], + "title": "Type", + "type": "string" + }, + "year": { + "anyOf": [ + { + "pattern": "^\\d{4}$", + "type": "string" + }, + { + "type": "null" + } + ], + "description": "四位发行年份", + "title": "Year" + } + }, + "required": [ + "src_storage", + "dest_storage", + "src", + "dest", + "mode", + "type", + "title", + "media_source", + "media_id" + ], + "title": "TransferHistoryRegisterRequest", + "type": "object" + }, "TransferHistoryReplanRequest": { "additionalProperties": false, "description": "对零文件副作用的识别拒绝提交已核实的媒体身份及季集修正。", @@ -6071,6 +6209,59 @@ "title": "TransferManualReviewRequest", "type": "object" }, + "TransferRegistrationAudit": { + "additionalProperties": false, + "description": "服务器记录的外部登记操作者、UTC 时间和文件核验证据。", + "properties": { + "actor_id": { + "description": "登记管理员的完整用户主键，身份以此字段为准", + "minimum": 1.0, + "title": "Actor Id", + "type": "integer" + }, + "actor_name": { + "description": "管理员名称显示快照，最多 128 字符；超长时取前 127 字符加省略号，空名称显示 #用户ID", + "maxLength": 128, + "minLength": 1, + "title": "Actor Name", + "type": "string" + }, + "registered_at": { + "description": "服务器生成的 UTC ISO 登记时间", + "maxLength": 40, + "title": "Registered At", + "type": "string" + }, + "source_state": { + "description": "核验时源文件存在或确认不存在", + "enum": [ + "present", + "absent" + ], + "title": "Source State", + "type": "string" + }, + "verification": { + "description": "两端大小匹配、本地硬链接或仅目标存在；均不代表内容校验和", + "enum": [ + "matching_sizes", + "native_hardlink", + "destination_only" + ], + "title": "Verification", + "type": "string" + } + }, + "required": [ + "actor_id", + "actor_name", + "registered_at", + "verification", + "source_state" + ], + "title": "TransferRegistrationAudit", + "type": "object" + }, "Workflow-Input": { "description": "创建或更新操作接受的完整工作流定义。", "properties": { @@ -15526,6 +15717,19 @@ "description": "当前页最多返回的记录数。", "title": "Count" }, + "origin": { + "anyOf": [ + { + "const": "external", + "type": "string" + }, + { + "type": "null" + } + ], + "description": "整理历史来源；external 仅返回管理员登记的外部整理记录，省略时保留全部历史。", + "title": "Origin" + }, "page": { "anyOf": [ { @@ -15702,6 +15906,27 @@ "title": "transfer.history.redo_batch", "type": "object" }, + { + "additionalProperties": false, + "description": "管理员登记已完成的外部复制、原生硬链接或移动；严格核验文件事实并记录审计，不操作文件或下载器，同一声明重试返回原回执。请求方法：POST；路径：/api/v1/history/transfer。操作效果：reversible_write；身份范围：admin。所需角色：system_admin。", + "properties": { + "body": { + "$ref": "#/$defs/TransferHistoryRegisterRequest", + "description": "transfer.history.register 的请求内容。管理员登记已完成的外部复制、原生硬链接或移动；严格核验文件事实并记录审计，不操作文件或下载器，同一声明重试返回原回执。严格使用下列类型和字段。" + }, + "operation_id": { + "const": "transfer.history.register", + "description": "白名单中的精确操作标识，用于选择对应 oneOf 分支。", + "type": "string" + } + }, + "required": [ + "operation_id", + "body" + ], + "title": "transfer.history.register", + "type": "object" + }, { "additionalProperties": false, "description": "为没有文件副作用的识别失败历史提交已核实的电影或电视剧身份和季集修正，保留原任务及失败证据。请求方法：POST；路径：/api/v1/history/transfer/{history_id}/replan。操作效果：external_side_effect；身份范围：admin。所需角色：system_admin。", @@ -16919,6 +17144,7 @@ "transfer.history.delete", "transfer.history.redo", "transfer.history.redo_batch", + "transfer.history.register", "transfer.history.replan", "transfer.manual_history", "transfer.manual_review", diff --git a/app/api/dependencies/history.py b/app/api/dependencies/history.py index 898bf8997..0dccd0767 100644 --- a/app/api/dependencies/history.py +++ b/app/api/dependencies/history.py @@ -22,6 +22,7 @@ from app.application.history.mutation import ( DownloadHistoryMutationRepository, HistoryUnitOfWork, ) +from app.application.history.registration import TransferRegistrationCommand from app.application.history.retry import clear_transfer_failures from app.application.mediaserver import ( AsyncMediaServerQueryRepository, @@ -108,6 +109,18 @@ def get_transfer_history_lookup_service( runtime.history.transfer_repository ) +def get_transfer_registration_command( + runtime: HostRuntime = Depends(get_host_runtime), +) -> TransferRegistrationCommand: + """ + 仅注入只读文件能力和独立短事务，不复用认证请求 Session。 + """ + return TransferRegistrationCommand( + scope_factory=runtime.history.registration_scope, + query_file=StorageChain().get_file_item_strict, + local_entry=runtime.history.registration_local_entry, + ) + def get_transfer_execution_repository( runtime: HostRuntime = Depends(get_host_runtime), diff --git a/app/api/endpoints/history.py b/app/api/endpoints/history.py index 59fb93ed3..2d479ac6b 100644 --- a/app/api/endpoints/history.py +++ b/app/api/endpoints/history.py @@ -5,9 +5,9 @@ import asyncio import time from collections.abc import Coroutine -from typing import Any, Callable, List, Optional +from typing import Any, Callable, List, Literal, Optional -from fastapi import Depends, Response +from fastapi import Depends, HTTPException, Response from app.adapters.web.security.access import verify_token from app.agent.contracts import ReplyMode @@ -30,6 +30,7 @@ from app.api.dependencies.history import ( get_history_query_service, get_transfer_execution_repository, get_transfer_history_mutation_command, + get_transfer_registration_command, ) from app.api.response import ( COLLECTION_TOTAL_HEADER, @@ -43,6 +44,12 @@ from app.application.history import ( HistoryQueryService, TransferHistoryMutationCommand, ) +from app.application.history.registration import ( + RegistrationActor, + RegistrationConflictError, + RegistrationValidationError, + TransferRegistrationCommand, +) from app.application.transfer.execution import ( TransferExecutionCommand, TransferExecutionRepository, @@ -60,6 +67,8 @@ from app.schemas.history import TransferHistory as _SchemaTransferHistory from app.schemas.history import TransferHistoryDeleteResult as _SchemaTransferHistoryDeleteResult from app.schemas.history import TransferHistoryDiscardResult as _SchemaTransferHistoryDiscardResult from app.schemas.history import TransferHistoryPage as _SchemaTransferHistoryPage +from app.schemas.history import TransferHistoryRegisterRequest as _SchemaTransferHistoryRegisterRequest +from app.schemas.history import TransferHistoryRegisterResult as _SchemaTransferHistoryRegisterResult from app.schemas.history import TransferHistoryReplanRequest as _SchemaTransferHistoryReplanRequest from app.schemas.response import Response as _SchemaResponse from app.schemas.token import TokenPayload as _SchemaTokenPayload @@ -303,6 +312,7 @@ async def transfer_history( page: Optional[int] = 1, count: Optional[int] = 30, status: Optional[bool] = None, + origin: Optional[Literal["external"]] = None, query: HistoryQueryService = Depends(get_history_query_service), _: _SchemaTokenPayload = Depends(verify_token), ) -> Any: @@ -311,15 +321,46 @@ async def transfer_history( """ page = page if page is not None else 1 count = count if count is not None else 30 - result = await query.list_transfer( - title=title, - page=page, - count=count, - status=status, - ) + if origin is None: + result = await query.list_transfer(title=title, page=page, count=count, status=status) + else: + result = await query.list_transfer( + title=title, + page=page, + count=count, + status=status, + origin=origin, + ) return _SchemaResponse(success=True, data=result) +@router.post( + "/transfer", + summary="登记已完成的外部整理", + response_model=_SchemaResponse[_SchemaTransferHistoryRegisterResult], +) +def register_transfer_history( + payload: _SchemaTransferHistoryRegisterRequest, + command: TransferRegistrationCommand = Depends(get_transfer_registration_command), + user: Any = Depends(get_current_active_superuser), +) -> Any: + """ + 管理员登记外部复制、硬链接或移动事实，只写历史，不操作文件、下载器或订阅。 + """ + try: + result = command.execute(payload, RegistrationActor(id=user.id, name=user.name)) + except RegistrationConflictError as error: + raise HTTPException(status_code=409, detail=str(error)) from error + except RegistrationValidationError as error: + raise HTTPException(status_code=422, detail=str(error)) from error + return _SchemaResponse( + success=True, + data=_SchemaTransferHistoryRegisterResult( + history=_SchemaTransferHistory.model_validate(result.history), created=result.created, + ), + ) + + def _clear_transfer_history( command: TransferHistoryMutationCommand, ) -> _SchemaResponse[None]: diff --git a/app/application/history/__init__.py b/app/application/history/__init__.py index 6eddb1b8e..c29c8527f 100644 --- a/app/application/history/__init__.py +++ b/app/application/history/__init__.py @@ -598,10 +598,12 @@ class HistoryQueryService: page: int = 1, count: int = 30, status: Optional[bool] = None, + origin: Optional[str] = None, ) -> TransferHistoryPage: """ 应用历史筛选规则并返回整理历史分页 DTO。 """ + origin_filter = {"origin": origin} if origin is not None else {} if title: wildcard = "*" in title or "?" in title if wildcard: @@ -612,6 +614,7 @@ class HistoryQueryService: pattern, status=status, wildcard=wildcard, + **origin_filter, ) records = await self._transfer_repository.async_list_by_title( pattern, @@ -619,14 +622,16 @@ class HistoryQueryService: count=count, status=status, wildcard=wildcard, + **origin_filter, ) else: records = await self._transfer_repository.async_list_by_page( page=page, count=count, status=status, + **origin_filter, ) - total = await self._transfer_repository.async_count(status=status) + total = await self._transfer_repository.async_count(status=status, **origin_filter) return TransferHistoryPage( list=[TransferHistory.model_validate(record) for record in records], diff --git a/app/application/history/contracts.py b/app/application/history/contracts.py index 7ef530564..cfc185c28 100644 --- a/app/application/history/contracts.py +++ b/app/application/history/contracts.py @@ -74,6 +74,8 @@ class TransferHistorySnapshot: 脱离数据库会话后供整理、Agent 和历史用例读取的完整历史快照。 :param id: 整理历史记录的数据库主键 + :param origin: external 为外部登记；旧行未记录来源 + :param registration: 服务端登记审计与文件核验证据的冻结 JSON :param transfer_task_id: 关联的持久整理任务标识；旧历史可能为 None :param transfer_settlement_revision: 对应持久任务终态回执的结算版本 :param src: 整理前的源文件或目录路径 @@ -120,6 +122,8 @@ class TransferHistorySnapshot: """ id: int + origin: Optional[str] = None + registration: Optional[JsonData] = None transfer_task_id: Optional[str] = None transfer_settlement_revision: Optional[int] = None src: Optional[str] = None @@ -171,6 +175,7 @@ class TransferHistorySnapshot: object.__setattr__(self, "src_fileitem", _freeze_json(self.src_fileitem)) object.__setattr__(self, "dest_fileitem", _freeze_json(self.dest_fileitem)) object.__setattr__(self, "files", _freeze_json(self.files)) + object.__setattr__(self, "registration", _freeze_json(self.registration)) @dataclass(frozen=True, slots=True) @@ -457,6 +462,7 @@ class TransferHistoryQueryPort(Protocol): count: int = 30, status: Optional[bool] = None, wildcard: bool = False, + origin: Optional[str] = None, ) -> list[TransferHistorySnapshot]: """ 异步按标题或路径分页返回历史快照。 @@ -466,6 +472,7 @@ class TransferHistoryQueryPort(Protocol): :param count: 每页最多返回的记录数 :param status: 整理成功状态筛选；None 表示不限状态 :param wildcard: 是否将查询文本按通配规则匹配 + :param origin: 外部登记来源筛选；None 保留所有历史 :return: 匹配查询文本与状态的当前页整理历史 """ ... @@ -475,6 +482,7 @@ class TransferHistoryQueryPort(Protocol): page: int = 1, count: int = 30, status: Optional[bool] = None, + origin: Optional[str] = None, ) -> list[TransferHistorySnapshot]: """ 异步按时间倒序分页返回历史快照。 @@ -482,15 +490,17 @@ class TransferHistoryQueryPort(Protocol): :param page: 页码，从 1 开始 :param count: 每页最多返回的记录数 :param status: 整理成功状态筛选；None 表示不限状态 + :param origin: 外部登记来源筛选；None 保留所有历史 :return: 按时间倒序排列且满足状态条件的当前页历史 """ ... - async def async_count(self, status: Optional[bool] = None) -> int: + async def async_count(self, status: Optional[bool] = None, origin: Optional[str] = None) -> int: """ 异步统计指定状态的整理历史数量。 :param status: 整理成功状态筛选；None 表示不限状态 + :param origin: 外部登记来源筛选；None 保留所有历史 :return: 满足成功状态筛选的整理历史数量 """ ... @@ -500,6 +510,7 @@ class TransferHistoryQueryPort(Protocol): title: str, status: Optional[bool] = None, wildcard: bool = False, + origin: Optional[str] = None, ) -> int: """ 异步统计匹配标题或路径的整理历史数量。 @@ -507,6 +518,7 @@ class TransferHistoryQueryPort(Protocol): :param title: 用于匹配媒体标题或文件路径的查询文本 :param status: 整理成功状态筛选；None 表示不限状态 :param wildcard: 是否将查询文本按通配规则匹配 + :param origin: 外部登记来源筛选；None 保留所有历史 :return: 满足查询文本和状态筛选的整理历史数量 """ ... diff --git a/app/application/history/registration.py b/app/application/history/registration.py new file mode 100644 index 000000000..4f095d244 --- /dev/null +++ b/app/application/history/registration.py @@ -0,0 +1,405 @@ +""" +外部已完成整理的只读事实核验和不可覆盖登记用例。 +""" + +from collections.abc import Callable +from contextlib import AbstractContextManager +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from math import isfinite +from pathlib import Path +from typing import Literal, Protocol +from unicodedata import normalize + +from app.application.history.contracts import TransferHistorySnapshot +from app.application.history.mutation import HistoryUnitOfWork +from app.schemas.exception import StorageQueryError +from app.schemas.file import FileItem +from app.schemas.history import TransferHistoryRegisterRequest, TransferRegistrationAudit +from app.schemas.types import MediaSource + + +class RegistrationValidationError(ValueError): + """ + 存储未提供足够一致的文件事实，不能登记成功历史。 + """ + + +class RegistrationConflictError(ValueError): + """ + 相同源地址已经关联不同或非外部登记的历史，禁止覆盖。 + """ + + +@dataclass(frozen=True, slots=True) +class RegistrationActor: + """ + 认证依赖提供的管理员身份。 + + :param id: 已认证用户主键 + :param name: 已认证用户名 + """ + + id: int + name: str + + +@dataclass(frozen=True, slots=True) +class LocalRegistrationEntry: + """ + 不跟随最终符号链接的本地文件和父目录事实。 + + :param dev: 文件设备号 + :param ino: 文件节点号 + :param size: 文件字节数 + :param mtime_ns: 纳秒修改时间 + :param ctime_ns: 纳秒状态变化时间 + :param is_file: 是否为普通文件 + :param is_symlink: 最终目录项是否为符号链接 + :param parent_dev: 父目录设备号 + :param parent_ino: 父目录节点号 + :param canonical_parent: 仅解析父目录后的绝对地址 + :param name: 原请求的最终目录项名称 + """ + + dev: int + ino: int + size: int + mtime_ns: int + ctime_ns: int + is_file: bool + is_symlink: bool + parent_dev: int + parent_ino: int + canonical_parent: str + name: str + + +@dataclass(frozen=True, slots=True) +class RegistrationFileSnapshot: + """ + 核验后允许持久化的文件最小快照，不携带访问地址和 provider 私有字段。 + + :param path: 原声明的规范地址 + :param storage: 存储标识 + :param type: 确认存在时为 file，已移动的源保持 None + :param size: 确认存在时的字节数 + :param fileid: 可选的有界存储文件标识 + :param modify_time: 可选的有限修改时间 + """ + + path: str + storage: str + type: Literal["file"] | None = None + size: int | None = None + fileid: str | None = None + modify_time: float | None = None + + def to_dict(self) -> dict[str, str | int | float]: + """ + 返回独立的有界 JSON 数据，源不存在时只包含地址。 + """ + return {key: value for key, value in asdict(self).items() if value is not None} + + +@dataclass(frozen=True, slots=True) +class TransferRegistrationDeclaration: + """ + 在任何 I/O 之前冻结管理员的完整声明，避免持久化阶段依赖可变请求模型。 + + :param src: 规范源地址 + :param src_storage: 源存储 + :param dest: 规范目标地址 + :param dest_storage: 目标存储 + :param mode: 已完成的整理模式 + :param type: 电影或电视剧 + :param title: 管理员确认的媒体标题 + :param year: 可选发行年份 + :param media_source: 规范来源 + :param media_id: 来源原生标识 + :param seasons: 可选季号 + :param episodes: 可选集号 + :param download_hash: 可选下载任务标识 + """ + + src: str + src_storage: str + dest: str + dest_storage: str + mode: Literal["copy", "link", "move"] + type: Literal["电影", "电视剧"] + title: str + year: str | None + media_source: MediaSource + media_id: str + seasons: str | None + episodes: str | None + download_hash: str | None + + @classmethod + def from_request(cls, request: TransferHistoryRegisterRequest) -> "TransferRegistrationDeclaration": + """ + 显式复制已校验请求的标量，不保留可变请求引用。 + """ + return cls( + src=request.src, src_storage=request.src_storage, + dest=request.dest, dest_storage=request.dest_storage, + mode=request.mode, type=request.type, title=request.title, year=request.year, + media_source=request.media_source, media_id=request.media_id, + seasons=request.seasons, episodes=request.episodes, download_hash=request.download_hash, + ) + + +@dataclass(frozen=True, slots=True) +class TransferRegistrationWrite: + """ + 通过文件事实核验后提交给原子持久端口的数据。 + + :param request: 不可变的管理员媒体与地址声明 + :param source: 源文件已核验快照 + :param destination: 目标文件已核验快照 + :param audit: 服务端生成的不可变审计 + :param date: 兼容历史页面的本地时间文本 + """ + + request: TransferRegistrationDeclaration + source: RegistrationFileSnapshot + destination: RegistrationFileSnapshot + audit: TransferRegistrationAudit + date: str + + +@dataclass(frozen=True, slots=True) +class TransferRegistrationResult: + """ + 已提交的外部登记回执。 + + :param history: 脱离 Session 的不可变历史快照 + :param created: 是否由此次请求新增记录 + """ + + history: TransferHistorySnapshot + created: bool + + +class TransferRegistrationRepository(Protocol): + """ + 绑定短 Session 的外部登记原子读写端口。 + """ + + def get_by_source(self, src: str, storage: str) -> TransferHistorySnapshot | None: + """ + 读取相同源地址的历史，供幂等和不可覆盖判断。 + + :param src: 已规范化的绝对源地址 + :param storage: 对应存储配置标识 + :return: 脱离 Session 的冻结快照；没有记录时返回 None + """ + ... + + def insert_if_absent(self, write: TransferRegistrationWrite) -> TransferRegistrationResult: + """ + 原子插入或返回唯一约束已保留的原历史，不提交事务。 + + :param write: 应用用例已核验的声明、文件事实和服务端审计 + :return: 当前事务内的冻结记录及是否成功插入，由用例提交后才作为成功回执 + """ + ... + + +@dataclass(frozen=True, slots=True) +class TransferRegistrationScope: + """ + 单次短事务需要的仓储与工作单元。 + + :param repository: 当前 Session 的登记仓储 + :param unit_of_work: 由用例决定提交或回滚的工作单元 + """ + + repository: TransferRegistrationRepository + unit_of_work: HistoryUnitOfWork + + +RegistrationScopeFactory = Callable[[], AbstractContextManager[TransferRegistrationScope]] +LocalEntryProbe = Callable[[str], LocalRegistrationEntry | None] +StrictFileQuery = Callable[[str, Path], FileItem | None] + + +def _actor_display_name(actor: RegistrationActor) -> str: + """ + 生成有界管理员名称显示快照，认证身份仍以完整 actor.id 为准。 + + :param actor: 认证依赖提供的管理员，不改变既有账号名称规则 + :return: 普通名称原样保留；超长名称为前 127 字符加省略号，空名称为 #用户ID + """ + if not actor.name: + return f"#{actor.id}" + if len(actor.name) > 128: + return f"{actor.name[:127]}…" + return actor.name + + +def _matching_receipt( + history: TransferHistorySnapshot, request: TransferRegistrationDeclaration, +) -> TransferRegistrationResult: + """ + 只允许完整声明相同的既有外部成功记录作为重复回执。 + """ + if history.origin != "external" or not history.status or history.transfer_task_id: + raise RegistrationConflictError("该源地址已有内部整理或失败历史，不能覆盖") + if history.transfer_settlement_revision is not None or history.registration is None: + raise RegistrationConflictError("该源地址的历史不属于可重复的外部登记") + for key, value in asdict(request).items(): + if getattr(history, key) != value: + raise RegistrationConflictError("该源地址已有不同目标或媒体声明的历史，不能覆盖") + return TransferRegistrationResult(history, False) + + +def _snapshot(item: FileItem, storage: str, path: str) -> RegistrationFileSnapshot: + """ + 从严格 provider 回执投影已确认的正规文件，拒绝不完整或不一致的事实。 + """ + if item.storage != storage or item.path != path or item.type != "file": + raise RegistrationValidationError("存储返回的文件地址或类型与登记声明不一致") + if item.size is None or item.size < 0: + raise RegistrationValidationError("存储未返回有效文件大小") + if item.fileid is not None and len(item.fileid) > 512: + raise RegistrationValidationError("存储返回的文件标识过长") + if item.modify_time is not None and not isfinite(item.modify_time): + raise RegistrationValidationError("存储返回的文件时间无效") + return RegistrationFileSnapshot( + path=path, storage=storage, type="file", size=item.size, + fileid=item.fileid, modify_time=item.modify_time, + ) + + +def _check_local_item(entry: LocalRegistrationEntry | None, item: FileItem | None) -> None: + """ + 对照不跟随链接的事实和 provider 回执，拒绝链接或观察期间已变化的文件。 + """ + if entry is None: + if item is not None: + raise RegistrationValidationError("文件状态在核验期间发生变化，请重试") + return + if entry.is_symlink or not entry.is_file: + raise RegistrationValidationError("外部登记只接受普通文件，不接受符号链接或目录") + if item is None or item.size != entry.size: + raise RegistrationValidationError("文件状态在核验期间发生变化，请重试") + + +def _distinct_entries(source: LocalRegistrationEntry, dest: LocalRegistrationEntry) -> None: + """ + 拒绝同一父目录的同名、大小写及 Unicode 规范等价别名，不把拼写差异当成硬链接。 + """ + same_parent = (source.parent_dev, source.parent_ino) == (dest.parent_dev, dest.parent_ino) + source_name = normalize("NFC", normalize("NFC", source.name).casefold()) + dest_name = normalize("NFC", normalize("NFC", dest.name).casefold()) + if same_parent and source_name == dest_name: + raise RegistrationValidationError("源与目标指向同一或大小写含糊的目录项") + + +class TransferRegistrationCommand: + """ + 核验外部整理事实并登记历史，完全不拥有文件变更或下载器副作用能力。 + """ + + def __init__( + self, *, scope_factory: RegistrationScopeFactory, query_file: StrictFileQuery, + local_entry: LocalEntryProbe, + ) -> None: + """ + 保存显式注入的短事务、严格存储查询和本地目录项只读能力。 + """ + self._scope_factory = scope_factory + self._query_file = query_file + self._local_entry = local_entry + + def _probe(self, storage: str, path: str) -> tuple[FileItem | None, LocalRegistrationEntry | None]: + """ + 用本地非跟随证明包围严格查询，只把已确认不存在投影为 None。 + """ + try: + before = self._local_entry(path) if storage == "local" else None + item = self._query_file(storage, Path(path)) + if storage == "local": + after = self._local_entry(path) + if before != after: + raise RegistrationValidationError("文件状态在核验期间发生变化，请重试") + _check_local_item(after, item) + return item, before + except (OSError, StorageQueryError) as error: + raise RegistrationValidationError("无法确认存储文件状态，请检查存储后重试") from error + + def _verify_files( + self, request: TransferRegistrationDeclaration, + ) -> tuple[RegistrationFileSnapshot, RegistrationFileSnapshot, Literal["matching_sizes", "native_hardlink", "destination_only"]]: + """ + 按复制、硬链接或移动分别验证事实；不推断内容校验和。 + """ + source, source_entry = self._probe(request.src_storage, request.src) + dest, dest_entry = self._probe(request.dest_storage, request.dest) + if dest is None: + raise RegistrationValidationError("目标文件不存在，不能登记成功历史") + destination = _snapshot(dest, request.dest_storage, request.dest) + if request.mode == "move": + if source is not None: + raise RegistrationValidationError("移动登记要求源文件已经不存在") + return RegistrationFileSnapshot(request.src, request.src_storage), destination, "destination_only" + if source is None: + raise RegistrationValidationError("源文件不存在，不能核验复制或硬链接") + original = _snapshot(source, request.src_storage, request.src) + if original.size != destination.size: + raise RegistrationValidationError("源文件与目标文件大小不一致") + if source_entry is not None and dest_entry is not None: + _distinct_entries(source_entry, dest_entry) + if request.mode == "link": + self._verify_hardlink(source_entry, dest_entry) + return original, destination, "native_hardlink" + return original, destination, "matching_sizes" + + @staticmethod + def _verify_hardlink( + source: LocalRegistrationEntry | None, dest: LocalRegistrationEntry | None, + ) -> None: + """ + 只有本地相同设备和节点的不同目录项才能证明 native 硬链接。 + """ + if source is None or dest is None: + raise RegistrationValidationError("当前仅支持核验本地存储的原生硬链接") + if (source.dev, source.ino) != (dest.dev, dest.ino): + raise RegistrationValidationError("源文件与目标文件不是同一原生硬链接") + + def execute( + self, request: TransferHistoryRegisterRequest, actor: RegistrationActor, + ) -> TransferRegistrationResult: + """ + 先短读精确回执，再核验文件，最后用独立短写事务原子登记。 + + :param request: 已通过入口校验的管理员声明，用例在 I/O 前冻结副本 + :param actor: 认证依赖提供的管理员身份 + :return: 事务提交后的新建或既有回执 + """ + declaration = TransferRegistrationDeclaration.from_request(request) + with self._scope_factory() as scope: + previous = scope.repository.get_by_source(declaration.src, declaration.src_storage) + if previous is not None: + return _matching_receipt(previous, declaration) + source, destination, verification = self._verify_files(declaration) + audit = TransferRegistrationAudit( + actor_id=actor.id, actor_name=_actor_display_name(actor), + registered_at=datetime.now(timezone.utc).isoformat(), + verification=verification, source_state="absent" if declaration.mode == "move" else "present", + ) + write = TransferRegistrationWrite( + declaration, source, destination, audit, datetime.now().strftime("%Y-%m-%d %H:%M:%S"), + ) + with self._scope_factory() as scope: + try: + result = scope.repository.insert_if_absent(write) + if not result.created: + result = _matching_receipt(result.history, declaration) + scope.unit_of_work.commit() + return result + except Exception: + scope.unit_of_work.rollback() + raise diff --git a/app/db/adapters/history/registration.py b/app/db/adapters/history/registration.py new file mode 100644 index 000000000..04bfaad9e --- /dev/null +++ b/app/db/adapters/history/registration.py @@ -0,0 +1,79 @@ +""" +外部历史登记的独立短 Session 与原子插入适配器。 +""" + +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from dataclasses import asdict + +from sqlalchemy.orm import Session + +from app.application.history.contracts import TransferHistorySnapshot +from app.application.history.registration import ( + TransferRegistrationResult, + TransferRegistrationScope, + TransferRegistrationWrite, +) +from app.db.adapters.history.transfer import project_transfer_history +from app.db.oper.transferhistory import TransferHistoryOper +from app.db.uow import SqlAlchemyUnitOfWork + + +class SessionTransferRegistrationRepository: + """ + 将外部登记端口绑定到调用方短 Session，不持有提交权限。 + """ + + def __init__(self, session: Session) -> None: + """ + 保存当前 Session 对应的表级访问对象。 + """ + self._oper = TransferHistoryOper(session) + + def get_by_source(self, src: str, storage: str) -> TransferHistorySnapshot | None: + """ + 在 Session 内完成现有源地址历史的冻结投影。 + """ + record = self._oper.get_by_src(src, storage) + return project_transfer_history(record) if record is not None else None + + def insert_if_absent(self, write: TransferRegistrationWrite) -> TransferRegistrationResult: + """ + 将有界声明和服务器事实映射为列，再由唯一约束返回获胜历史。 + """ + values = asdict(write.request) + values["media_source"] = write.request.media_source.value + values.update( + src_fileitem=write.source.to_dict(), + dest_fileitem=write.destination.to_dict(), + origin="external", registration=write.audit.model_dump(mode="json"), + date=write.date, status=True, transfer_task_id=None, + transfer_settlement_revision=None, + ) + record, created = self._oper.stage_insert_if_source_absent(values) + return TransferRegistrationResult(project_transfer_history(record), created) + + +class TransferRegistrationScopeFactory: + """ + 为每个查询或写入阶段单独创建短 Session，避免持有认证请求的旧读事务。 + """ + + def __init__(self, session_factory: Callable[[], Session]) -> None: + """ + 保存组合根提供的 Session 工厂。 + """ + self._session_factory = session_factory + + @contextmanager + def __call__(self) -> Iterator[TransferRegistrationScope]: + """ + 保证 Session 关闭；提交和显式回滚由应用用例决定。 + """ + session = self._session_factory() + try: + yield TransferRegistrationScope( + SessionTransferRegistrationRepository(session), SqlAlchemyUnitOfWork(session), + ) + finally: + session.close() diff --git a/app/db/adapters/history/transfer.py b/app/db/adapters/history/transfer.py index 4c626c058..5f12963ec 100644 --- a/app/db/adapters/history/transfer.py +++ b/app/db/adapters/history/transfer.py @@ -101,6 +101,8 @@ def project_transfer_history(record: object) -> TransferHistorySnapshot: recovery_action = None return TransferHistorySnapshot( id=history_id, + origin=getattr(record, "origin", None), + registration=deepcopy(getattr(record, "registration", None)), transfer_task_id=getattr(record, "transfer_task_id", None), transfer_settlement_revision=getattr( record, @@ -382,6 +384,7 @@ class TransactionalTransferHistoryRepository: count: int = 30, status: Optional[bool] = None, wildcard: bool = False, + origin: Optional[str] = None, ) -> list[TransferHistorySnapshot]: """ 异步按标题或路径分页返回历史快照。 @@ -393,6 +396,7 @@ class TransactionalTransferHistoryRepository: count, status, wildcard, + origin=origin, ) return [project_transfer_history(record) for record in records] @@ -401,6 +405,7 @@ class TransactionalTransferHistoryRepository: page: int = 1, count: int = 30, status: Optional[bool] = None, + origin: Optional[str] = None, ) -> list[TransferHistorySnapshot]: """ 异步按时间倒序分页返回历史快照。 @@ -410,15 +415,16 @@ class TransactionalTransferHistoryRepository: page, count, status, + origin=origin, ) return [project_transfer_history(record) for record in records] - async def async_count(self, status: Optional[bool] = None) -> int: + async def async_count(self, status: Optional[bool] = None, origin: Optional[str] = None) -> int: """ 异步统计指定状态的整理历史数量。 """ async with self._async_session() as session: - count = await TransferHistoryOper(session).async_count(status) + count = await TransferHistoryOper(session).async_count(status, origin=origin) return int(count or 0) async def async_count_by_title( @@ -426,6 +432,7 @@ class TransactionalTransferHistoryRepository: title: str, status: Optional[bool] = None, wildcard: bool = False, + origin: Optional[str] = None, ) -> int: """ 异步统计匹配标题或路径的整理历史数量。 @@ -435,6 +442,7 @@ class TransactionalTransferHistoryRepository: title, status, wildcard, + origin=origin, ) return int(count or 0) diff --git a/app/db/models/transferhistory.py b/app/db/models/transferhistory.py index f7e6ac074..c3737f5db 100644 --- a/app/db/models/transferhistory.py +++ b/app/db/models/transferhistory.py @@ -51,6 +51,10 @@ class TransferHistory(Base): transfer_task_id: Mapped[Optional[str]] = mapped_column(String(64)) # 失败 pending 当前映射对应的结算版本 transfer_settlement_revision: Mapped[Optional[int]] = mapped_column(Integer) + # external 表示外部登记；旧记录未记录来源 + origin: Mapped[Optional[str]] = mapped_column(String(16)) + # 外部整理登记的服务端操作者、时间和核验证据；旧行不回填来源 + registration: Mapped[Optional[Any]] = mapped_column(JSON) # 源路径 src: Mapped[Optional[str]] = mapped_column(String, index=True) # 源存储 @@ -142,7 +146,7 @@ class TransferHistory(Base): @classmethod def list_by_title(cls, db: Session, title: str, page: int = 1, count: int = 30, - status: Optional[bool] = None, wildcard: bool = False) -> list[Self]: + status: Optional[bool] = None, wildcard: bool = False, origin: Optional[str] = None) -> list[Self]: """ 按标题或路径分页查询整理历史。 """ @@ -163,6 +167,9 @@ class TransferHistory(Base): statement = statement.where(cls.status == status) statement = statement.order_by(cls.date.desc()) + if origin is not None: + statement = statement.where(cls.origin == origin) + # 当count为负数时，不限制页数查询所有 if count >= 0: statement = statement.offset((page - 1) * count).limit(count) @@ -171,7 +178,7 @@ class TransferHistory(Base): @classmethod async def async_list_by_title(cls, db: AsyncSession, title: str, page: int = 1, count: int = 30, - status: Optional[bool] = None, wildcard: bool = False) -> list[Self]: + status: Optional[bool] = None, wildcard: bool = False, origin: Optional[str] = None) -> list[Self]: """ 异步按标题或路径分页查询整理历史。 """ @@ -192,6 +199,9 @@ class TransferHistory(Base): query = query.filter(cls.status == status) query = query.order_by(cls.date.desc()) + if origin is not None: + query = query.where(cls.origin == origin) + # 当count为负数时，不限制页数查询所有 if count >= 0: query = query.offset((page - 1) * count).limit(count) @@ -200,7 +210,7 @@ class TransferHistory(Base): return list(result.scalars().all()) @classmethod - def list_by_page(cls, db: Session, page: int = 1, count: int = 30, status: Optional[bool] = None) -> list[Self]: + def list_by_page(cls, db: Session, page: int = 1, count: int = 30, status: Optional[bool] = None, origin: Optional[str] = None) -> list[Self]: """ 按时间倒序分页读取整理历史。 """ @@ -209,6 +219,9 @@ class TransferHistory(Base): statement = statement.where(cls.status == status) statement = statement.order_by(cls.date.desc()) + if origin is not None: + statement = statement.where(cls.origin == origin) + # 当count为负数时，不限制页数查询所有 if count >= 0: statement = statement.offset((page - 1) * count).limit(count) @@ -217,7 +230,7 @@ class TransferHistory(Base): @classmethod async def async_list_by_page(cls, db: AsyncSession, page: int = 1, count: int = 30, - status: Optional[bool] = None) -> list[Self]: + status: Optional[bool] = None, origin: Optional[str] = None) -> list[Self]: """ 异步按时间倒序分页读取整理历史。 """ @@ -232,6 +245,9 @@ class TransferHistory(Base): cls.date.desc() ) + if origin is not None: + query = query.where(cls.origin == origin) + # 当count为负数时，不限制页数查询所有 if count >= 0: query = query.offset((page - 1) * count).limit(count) @@ -605,34 +621,35 @@ class TransferHistory(Base): return rows @classmethod - def count(cls, db: Session, status: Optional[bool] = None) -> int: + def count(cls, db: Session, status: Optional[bool] = None, origin: Optional[str] = None) -> int: """ 统计指定状态的整理历史数量。 """ statement = select(func.count(cls.id)) if status is not None: statement = statement.where(cls.status == status) + if origin is not None: + statement = statement.where(cls.origin == origin) + count: int = db.execute(statement).scalar() return count @classmethod - async def async_count(cls, db: AsyncSession, status: Optional[bool] = None) -> int: + async def async_count(cls, db: AsyncSession, status: Optional[bool] = None, origin: Optional[str] = None) -> int: """ 异步统计指定状态的整理历史数量。 """ + statement = select(func.count(cls.id)) if status is not None: - result = await db.execute( - select(func.count(cls.id)).filter(cls.status == status) - ) - else: - result = await db.execute( - select(func.count(cls.id)) - ) + statement = statement.where(cls.status == status) + if origin is not None: + statement = statement.where(cls.origin == origin) + result = await db.execute(statement) count: int = result.scalar() return count @classmethod - def count_by_title(cls, db: Session, title: str, status: Optional[bool] = None, wildcard: bool = False) -> int: + def count_by_title(cls, db: Session, title: str, status: Optional[bool] = None, wildcard: bool = False, origin: Optional[str] = None) -> int: """ 统计与标题或路径匹配的整理历史数量。 """ @@ -651,11 +668,14 @@ class TransferHistory(Base): statement = select(func.count(cls.id)).where(text_filter) if status is not None: statement = statement.where(cls.status == status) + if origin is not None: + statement = statement.where(cls.origin == origin) + count: int = db.execute(statement).scalar() return count @classmethod - async def async_count_by_title(cls, db: AsyncSession, title: str, status: Optional[bool] = None, wildcard: bool = False) -> int: + async def async_count_by_title(cls, db: AsyncSession, title: str, status: Optional[bool] = None, wildcard: bool = False, origin: Optional[str] = None) -> int: """ 异步统计与标题或路径匹配的整理历史数量。 """ @@ -674,6 +694,9 @@ class TransferHistory(Base): stmt = select(func.count(cls.id)).filter(text_filter) if status is not None: stmt = stmt.filter(cls.status == status) + if origin is not None: + stmt = stmt.where(cls.origin == origin) + result = await db.execute(stmt) count: int = result.scalar() return count diff --git a/app/db/oper/transferhistory.py b/app/db/oper/transferhistory.py index 7fc0c54d8..42bc0bf0c 100644 --- a/app/db/oper/transferhistory.py +++ b/app/db/oper/transferhistory.py @@ -8,6 +8,8 @@ from typing import Any, List, Optional, cast from sqlalchemy import delete as sqlalchemy_delete from sqlalchemy import func, or_, select +from sqlalchemy.dialects.postgresql import insert as postgresql_insert +from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import Session @@ -35,6 +37,34 @@ class TransferHistoryOper(DbOper): 转移历史管理 """ + def stage_insert_if_source_absent(self, values: dict[str, Any]) -> tuple[TransferHistory, bool]: + """ + 由源地址唯一约束原子仲裁插入，永不替换已有行且不提交调用方事务。 + + :param values: 已由用例投影的完整历史列值 + :return: 插入或冲突保留的历史行，以及是否新建 + """ + if not isinstance(self._db, Session): + raise RuntimeError("登记整理历史需要调用方提供同步 Session") + dialect = self._db.get_bind().dialect.name + if dialect == "sqlite": + insert = sqlite_insert + elif dialect == "postgresql": + insert = postgresql_insert + else: + raise RuntimeError("当前数据库不支持原子整理历史登记") + statement = insert(TransferHistory).values(**values).on_conflict_do_nothing( + index_elements=["src", "src_storage"], + ).returning(TransferHistory.id) + inserted_id = self._db.execute(statement).scalar_one_or_none() + record = self._db.execute( + select(TransferHistory).where( + TransferHistory.src == values["src"], + TransferHistory.src_storage == values["src_storage"], + ) + ).scalar_one() + return record, inserted_id is not None + def get(self, historyid: int) -> Optional[TransferHistory]: """ 获取转移历史 @@ -167,6 +197,7 @@ class TransferHistoryOper(DbOper): count: int = 30, status: Optional[bool] = None, wildcard: bool = False, + origin: Optional[str] = None, ) -> List[TransferHistory]: """ 异步按标题分页查询转移记录。 @@ -179,6 +210,7 @@ class TransferHistoryOper(DbOper): count=count, status=status, wildcard=wildcard, + origin=origin, ) ) @@ -187,22 +219,23 @@ class TransferHistoryOper(DbOper): page: int = 1, count: int = 30, status: Optional[bool] = None, + origin: Optional[str] = None, ) -> List[TransferHistory]: """ 异步分页查询转移记录。 """ return await self._execute_async_query( lambda session: TransferHistory.async_list_by_page( - session, page=page, count=count, status=status + session, page=page, count=count, status=status, origin=origin ) ) - async def async_count(self, status: Optional[bool] = None) -> Optional[int]: + async def async_count(self, status: Optional[bool] = None, origin: Optional[str] = None) -> Optional[int]: """ 异步统计转移记录数量。 """ return await self._execute_async_query( - lambda session: TransferHistory.async_count(session, status=status) + lambda session: TransferHistory.async_count(session, status=status, origin=origin) ) async def async_count_by_title( @@ -210,6 +243,7 @@ class TransferHistoryOper(DbOper): title: str, status: Optional[bool] = None, wildcard: bool = False, + origin: Optional[str] = None, ) -> Optional[int]: """ 异步按标题统计转移记录数量。 @@ -220,6 +254,7 @@ class TransferHistoryOper(DbOper): title=title, status=status, wildcard=wildcard, + origin=origin, ) ) diff --git a/app/locales/en-US.json b/app/locales/en-US.json index c874cd5ae..8cd87b77b 100644 --- a/app/locales/en-US.json +++ b/app/locales/en-US.json @@ -601,7 +601,28 @@ "未识别到媒体类型": "Media type could not be identified", "站点域名未配置": "Site domain is not configured", "站点配置不完整": "Site configuration is incomplete", - "插件后台服务尚未停止，已保留实例和数据供重试": "Plugin background services have not stopped. The instance and data were retained so you can retry." + "插件后台服务尚未停止，已保留实例和数据供重试": "Plugin background services have not stopped. The instance and data were retained so you can retry.", + "该源地址已有内部整理或失败历史，不能覆盖": "This source already has internal or failed transfer history and cannot be overwritten.", + "该源地址的历史不属于可重复的外部登记": "The history for this source is not a repeatable external registration.", + "该源地址已有不同目标或媒体声明的历史，不能覆盖": "This source already has history with a different destination or media declaration.", + "存储返回的文件地址或类型与登记声明不一致": "The storage returned a different file path or type from the registration.", + "存储未返回有效文件大小": "The storage did not return a valid file size.", + "存储返回的文件标识过长": "The file identifier returned by the storage is too long.", + "存储返回的文件时间无效": "The storage returned an invalid file timestamp.", + "文件状态在核验期间发生变化，请重试": "The file changed during verification. Please retry.", + "外部登记只接受普通文件，不接受符号链接或目录": "External registration requires regular files; symbolic links and directories are not accepted.", + "源与目标指向同一或大小写含糊的目录项": "The source and destination refer to the same directory entry or an ambiguous case alias.", + "无法确认存储文件状态，请检查存储后重试": "Cannot confirm the file state. Check the storage and retry.", + "目标文件不存在，不能登记成功历史": "The destination file does not exist; successful history cannot be registered.", + "移动登记要求源文件已经不存在": "Move registration requires the source file to be absent.", + "源文件不存在，不能核验复制或硬链接": "The source file is absent; a copy or hard link cannot be verified.", + "源文件与目标文件大小不一致": "The source and destination file sizes do not match.", + "当前仅支持核验本地存储的原生硬链接": "Only native hard links on local storage can currently be verified.", + "源文件与目标文件不是同一原生硬链接": "The source and destination are not native hard links to the same file.", + "登记路径必须为不含父目录跳转的绝对路径": "Registration paths must be absolute and cannot contain parent-directory traversal.", + "登记字段不能包含控制字符": "Registration fields cannot contain control characters.", + "源文件与目标文件不能是同一地址": "The source and destination cannot have the same address.", + "电影登记不能包含季集信息": "Movie registration cannot include season or episode information." }, "message_patterns": [ { diff --git a/app/locales/zh-TW.json b/app/locales/zh-TW.json index e10479158..94361f21d 100644 --- a/app/locales/zh-TW.json +++ b/app/locales/zh-TW.json @@ -576,7 +576,28 @@ "未配置 Follow 订阅用户，跳过刷新": "未設定 Follow 訂閱使用者，跳過重新整理", "订阅日历预缓存完成": "訂閱日曆預快取完成", "订阅活动暂时不可用": "訂閱活動暫時無法使用", - "活动不存在或无权查看": "活動不存在或無權查看" + "活动不存在或无权查看": "活動不存在或無權查看", + "该源地址已有内部整理或失败历史，不能覆盖": "此來源位址已有內部整理或失敗歷史，不能覆寫", + "该源地址的历史不属于可重复的外部登记": "此來源位址的歷史不屬於可重試的外部登記", + "该源地址已有不同目标或媒体声明的历史，不能覆盖": "此來源位址已有不同目標或媒體聲明的歷史，不能覆寫", + "存储返回的文件地址或类型与登记声明不一致": "儲存空間傳回的檔案位址或類型與登記聲明不一致", + "存储未返回有效文件大小": "儲存空間未傳回有效檔案大小", + "存储返回的文件标识过长": "儲存空間傳回的檔案識別碼過長", + "存储返回的文件时间无效": "儲存空間傳回的檔案時間無效", + "文件状态在核验期间发生变化，请重试": "檔案狀態在驗證期間發生變化，請重試", + "外部登记只接受普通文件，不接受符号链接或目录": "外部登記只接受一般檔案，不接受符號連結或目錄", + "源与目标指向同一或大小写含糊的目录项": "來源與目標指向同一或大小寫不明確的目錄項目", + "无法确认存储文件状态，请检查存储后重试": "無法確認儲存檔案狀態，請檢查儲存空間後重試", + "目标文件不存在，不能登记成功历史": "目標檔案不存在，不能登記成功歷史", + "移动登记要求源文件已经不存在": "移動登記要求來源檔案已不存在", + "源文件不存在，不能核验复制或硬链接": "來源檔案不存在，無法驗證複製或硬連結", + "源文件与目标文件大小不一致": "來源檔案與目標檔案大小不一致", + "当前仅支持核验本地存储的原生硬链接": "目前僅支援驗證本機儲存的原生硬連結", + "源文件与目标文件不是同一原生硬链接": "來源檔案與目標檔案不是同一原生硬連結", + "登记路径必须为不含父目录跳转的绝对路径": "登記路徑必須是不含上層目錄跳轉的絕對路徑", + "登记字段不能包含控制字符": "登記欄位不能包含控制字元", + "源文件与目标文件不能是同一地址": "來源檔案與目標檔案不能是同一位址", + "电影登记不能包含季集信息": "電影登記不能包含季集資訊" }, "message_patterns": [ { diff --git a/app/schemas/exports.py b/app/schemas/exports.py index cddb8917c..b890f8e31 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -548,6 +548,8 @@ SCHEMA_EXPORTS = { 'TransferHistoryDeleteStep': ('app.schemas.history', 'TransferHistoryDeleteStep'), 'TransferHistoryDiscardResult': ('app.schemas.history', 'TransferHistoryDiscardResult'), 'TransferHistoryPage': ('app.schemas.history', 'TransferHistoryPage'), + 'TransferHistoryRegisterRequest': ('app.schemas.history', 'TransferHistoryRegisterRequest'), + 'TransferHistoryRegisterResult': ('app.schemas.history', 'TransferHistoryRegisterResult'), 'TransferHistoryReplanRequest': ('app.schemas.history', 'TransferHistoryReplanRequest'), 'TransferInfo': ('app.schemas.transfer', 'TransferInfo'), 'TransferInterceptEventData': ('app.schemas.event', 'TransferInterceptEventData'), @@ -567,6 +569,7 @@ SCHEMA_EXPORTS = { 'TransferQueueMediaData': ('app.schemas.transfer', 'TransferQueueMediaData'), 'TransferQueuePageData': ('app.schemas.transfer', 'TransferQueuePageData'), 'TransferQueueTaskData': ('app.schemas.transfer', 'TransferQueueTaskData'), + 'TransferRegistrationAudit': ('app.schemas.history', 'TransferRegistrationAudit'), 'TransferRenameBuildEventData': ('app.schemas.event', 'TransferRenameBuildEventData'), 'TransferRenameEventData': ('app.schemas.event', 'TransferRenameEventData'), 'TransferResultContractData': ('app.schemas.event', 'TransferResultContractData'), diff --git a/app/schemas/history.py b/app/schemas/history.py index 9909c570a..c4c2ccb44 100644 --- a/app/schemas/history.py +++ b/app/schemas/history.py @@ -2,16 +2,95 @@ 下载和整理历史的接口投影及批量操作结果。 """ +from pathlib import Path as _Path +from pathlib import PurePosixPath as _PurePosixPath from typing import List, Literal, Optional +from typing import Self as _Self +from unicodedata import category as _category from pydantic import BaseModel, ConfigDict, Field from pydantic import field_validator as _field_validator +from pydantic import model_validator as _model_validator from app.schemas.common import JsonData from app.schemas.media import OptionalMediaIdentityMixin, RequiredMediaIdentityMixin from app.schemas.types import MediaSource +class TransferRegistrationAudit(BaseModel): + """ + 外部整理登记时由服务器固定的操作者和文件核验说明。 + """ + + actor_id: int = Field(ge=1, description="登记管理员的完整用户主键，身份以此字段为准") + actor_name: str = Field( + min_length=1, max_length=128, + description="管理员名称显示快照，最多 128 字符；超长时取前 127 字符加省略号，空名称显示 #用户ID", + ) + registered_at: str = Field(max_length=40, description="服务器生成的 UTC ISO 登记时间") + verification: Literal["matching_sizes", "native_hardlink", "destination_only"] = Field( + description="两端大小匹配、本地硬链接或仅目标存在；均不代表内容校验和", + ) + source_state: Literal["present", "absent"] = Field(description="核验时源文件存在或确认不存在") + + model_config = ConfigDict(extra="forbid", frozen=True) + + +class TransferHistoryRegisterRequest(RequiredMediaIdentityMixin, BaseModel): + """ + 管理员对已完成的外部整理提交路径和媒体声明，不允许伪造文件快照或审计。 + """ + + src_storage: str = Field(min_length=1, max_length=64, description="源存储配置标识") + dest_storage: str = Field(min_length=1, max_length=64, description="目标存储配置标识") + src: str = Field(min_length=1, max_length=4096, description="整理前源文件的绝对路径") + dest: str = Field(min_length=1, max_length=4096, description="整理后目标文件的绝对路径") + mode: Literal["copy", "link", "move"] = Field(description="已完成的复制、本地硬链接或移动") + type: Literal["电影", "电视剧"] = Field(description="管理员确认的电影或电视剧类型") + title: str = Field(min_length=1, max_length=512, description="管理员确认的媒体标题") + year: Optional[str] = Field(default=None, pattern=r"^\d{4}$", description="四位发行年份") + media_source: MediaSource = Field(description="管理员确认的媒体数据源") + media_id: str = Field(min_length=1, max_length=128, description="来源中的规范原生媒体标识") + seasons: Optional[str] = Field( + default=None, max_length=32, pattern=r"^S\d{2,4}(?:-S\d{2,4})?$", description="季号或季范围，例如 S01", + ) + episodes: Optional[str] = Field( + default=None, max_length=32, pattern=r"^E(?:0[1-9]|[1-9]\d{1,3})(?:-E(?:0[1-9]|[1-9]\d{1,3}))?$", + description="集号或集范围，例如 E01 或 E01-E12", + ) + download_hash: Optional[str] = Field(default=None, max_length=128, description="已知下载任务标识，仅作关联声明") + + model_config = ConfigDict(extra="forbid", str_strip_whitespace=True) + + @_field_validator("*", mode="before") + @classmethod + def _reject_control_characters(cls, value: object) -> object: + """ + 在去除首尾空白之前拒绝 Unicode 控制字符，避免异常地址被静默改写。 + """ + if isinstance(value, str) and any(_category(char) == "Cc" for char in value): + raise ValueError("登记字段不能包含控制字符") + return value + + @_model_validator(mode="after") + def _normalize_addresses(self) -> _Self: + """ + 只规范地址拼写，不解析符号链接；拒绝相对路径、父跳转和同地址登记。 + """ + for field in ("src", "dest"): + value = getattr(self, field) + storage = getattr(self, f"{field}_storage") + path = _Path(value) if storage == "local" else _PurePosixPath(value) + if not path.is_absolute() or ".." in path.parts: + raise ValueError("登记路径必须为不含父目录跳转的绝对路径") + object.__setattr__(self, field, path.as_posix()) + if self.src_storage == self.dest_storage and self.src == self.dest: + raise ValueError("源文件与目标文件不能是同一地址") + if self.type == "电影" and (self.seasons or self.episodes): + raise ValueError("电影登记不能包含季集信息") + return self + + class DownloadHistory(OptionalMediaIdentityMixin, BaseModel): """ 下载历史记录。 @@ -52,6 +131,8 @@ class TransferHistory(OptionalMediaIdentityMixin, BaseModel): 文件整理历史记录。 """ id: int = Field(description="记录标识") + origin: Optional[Literal["external"]] = Field(default=None, description="external 为外部登记；旧行未记录来源") + registration: Optional[TransferRegistrationAudit] = Field(default=None, description="外部登记审计与核验说明") transfer_task_id: Optional[str] = Field( default=None, exclude=True, @@ -116,6 +197,15 @@ class TransferHistory(OptionalMediaIdentityMixin, BaseModel): return public_error_message(value, context="transfer") +class TransferHistoryRegisterResult(BaseModel): + """ + 已提交的登记回执，重复请求返回原历史及 created=false。 + """ + + history: TransferHistory = Field(description="已提交或先前登记的历史") + created: bool = Field(description="此次请求是否新建历史记录") + + class BatchTransferHistoryRedoRequest(BaseModel): """ 批量重新整理历史请求。 diff --git a/app/startup/composition/context.py b/app/startup/composition/context.py index 1f9949679..006512a0e 100644 --- a/app/startup/composition/context.py +++ b/app/startup/composition/context.py @@ -17,6 +17,7 @@ from app.application.history import ( TransferHistoryMutationRepository, TransferHistoryRepository, ) +from app.application.history.registration import LocalEntryProbe, RegistrationScopeFactory from app.application.messaging.chat import ( AgentChatPersistenceService, AsyncAgentChatRepository, @@ -388,6 +389,8 @@ class HistoryRuntime: :param transfer_mutation_repository: 按请求会话构造整理历史写仓储的工厂 :param media_server_repository: 构造媒体服务器条目仓储的工厂 :param transfer_execution_repository: 整理执行状态仓储 + :param registration_scope: 外部登记的独立短事务工厂 + :param registration_local_entry: 外部登记的本地非跟随文件证明 """ download_repository: RepositoryFactory @@ -395,6 +398,8 @@ class HistoryRuntime: transfer_mutation_repository: TransferHistoryRepositoryFactory media_server_repository: RepositoryFactory transfer_execution_repository: TransferExecutionRepository + registration_scope: RegistrationScopeFactory + registration_local_entry: LocalEntryProbe @dataclass(frozen=True, slots=True) diff --git a/app/startup/composition/runtime.py b/app/startup/composition/runtime.py index b97c5b1e6..6e9bcd7fd 100644 --- a/app/startup/composition/runtime.py +++ b/app/startup/composition/runtime.py @@ -19,6 +19,7 @@ from app.application.history import ( configure_transfer_history_repository, reset_transfer_history_repository, ) +from app.application.history.registration import LocalRegistrationEntry from app.application.site.contract import SiteRepository from app.application.site.health import ( SiteHealthService, @@ -189,7 +190,9 @@ def compose_runtime(inputs: RuntimeInputs) -> RuntimeComposition: 从同一批组合输入构造 HostRuntime 及旧 API 数据投影。 """ # 领域 Runtime 的具体持久化投影只在组合阶段解析，保持模块冷导入合同。 + from app.adapters.system.registration import probe_local_entry from app.db.adapters.history.download import SessionDownloadHistoryRepository + from app.db.adapters.history.registration import TransferRegistrationScopeFactory from app.db.adapters.history.transfer import SessionTransferHistoryRepository from app.db.adapters.outbox import ( SqlAlchemyAsyncOutboxDispatchStore, @@ -257,6 +260,8 @@ def compose_runtime(inputs: RuntimeInputs) -> RuntimeComposition: transfer_mutation_repository=SessionTransferHistoryRepository, media_server_repository=cast(RepositoryFactory, MediaServerOper), transfer_execution_repository=dependencies.transfer_execution, + registration_scope=TransferRegistrationScopeFactory(SessionFactory), + registration_local_entry=partial(probe_local_entry, entry_factory=LocalRegistrationEntry), ), site=SiteRuntime( repository=SessionSiteRepository, diff --git a/database/versions/b849e613a72f_external_history_registration.py b/database/versions/b849e613a72f_external_history_registration.py new file mode 100644 index 000000000..4cc405690 --- /dev/null +++ b/database/versions/b849e613a72f_external_history_registration.py @@ -0,0 +1,37 @@ +""" +为外部整理登记增加可空来源和审计列，旧记录保持未记录来源。 +""" + +import sqlalchemy as sa +from alembic import op + +revision = "b849e613a72f" +down_revision = "97ad7239c1e0" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + """ + 幂等添加登记列，不重建或覆盖既有历史与源地址唯一索引。 + """ + inspector = sa.inspect(op.get_bind()) + if "transferhistory" not in inspector.get_table_names(): + return + columns = {column["name"] for column in inspector.get_columns("transferhistory")} + for name, kind in (("origin", sa.String(16)), ("registration", sa.JSON())): + if name not in columns: + op.add_column("transferhistory", sa.Column(name, kind, nullable=True)) + + +def downgrade() -> None: + """ + 删除外部登记附加列，保留所有历史行及原有唯一约束。 + """ + inspector = sa.inspect(op.get_bind()) + if "transferhistory" not in inspector.get_table_names(): + return + columns = {column["name"] for column in inspector.get_columns("transferhistory")} + for name in ("registration", "origin"): + if name in columns: + op.drop_column("transferhistory", name) diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index c194967be..803c8eceb 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -7,7 +7,7 @@ > [`docs/rules/04-design-patterns.md`](rules/04-design-patterns.md) 为准，本文与其保持一致； > 如出现差异，以规则文档为准。 > -> *最后更新：2026-10-19* +> *最后更新：2026-11-03* --- @@ -820,8 +820,8 @@ flowchart LR | 指标 | 当前值 | |---|---:| -| Python 模块 | 1147 | -| 内部导入边 | 9,930 | +| Python 模块 | 1150 | +| 内部导入边 | 9,961 | | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） | diff --git a/docs/mcp-api.md b/docs/mcp-api.md index 75ab14c6f..d939aa9e8 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -2,6 +2,30 @@ 本文面向通过 MCP 调用 MoviePilot 的外部 Agent 和工具开发者。接入后，客户端通过工具目录发现能力，按 JSON Schema 构造参数，并根据业务结果判断操作是否完成。 +### 外部整理历史登记 + +管理员可通过 `transfer.history.register`（`POST /api/v1/history/transfer`）登记已经完成的 +外部复制、硬链接或移动。请求提供源与目标存储、绝对路径、整理模式，以及管理员确认的 +电影或电视剧身份、标题和可选季集。接口不移动文件、不刮削、不操作下载器，也不会触发 +后续整理事件。媒体声明不代替媒体识别结果，调用前应确认所填信息。 + +复制要求两个普通文件存在且大小相等；硬链接还要求本地存储中两个不同目录项对应相同 +设备和 inode；移动要求源已确认不存在，只核验目标。符号链接（含悬空源链接）、目录、 +存储查询失败或不一致回执均拒绝。上述核验不等于内容校验和，也不保证文件以后不变。 +超级管理员可登记当前存储提供方可访问的绝对路径，不限于配置的下载或媒体库目录。 + +成功返回 `history` 和 `created`。相同源、目标及媒体声明的重试返回最初的回执，保留原 +登记人和时间；不同声明、内部整理、失败历史或持久任务绑定均返回冲突，不覆盖原行。 +登记审计由服务器生成，旧历史来源保持未记录。`GET /api/v1/history/transfer` 可增加 +`origin=external`，列表和总数同时筛选；历史页面提供对应筛选和登记审计显示。 + +审计身份以完整的 `actor_id` 为准，`actor_name` 是最多 128 字符的名称显示快照： +非空且未超长时原样保留，超长时取前 127 字符加 `…`，空名称显示 `#`。 +该显示处理不改变账号名称规则，重复登记仍返回最初保存的身份和名称快照。 + +数据库升级只增加可空 `origin` 与 `registration` 列，保留既有源地址唯一约束和旧记录。 + + **最短调用路径：**配置连接与密钥 → `initialize` → `notifications/initialized` → `tools/list` → `tools/call`。已有 MCP 客户端通常会自动完成初始化和工具发现，只需配置下表中的连接信息。 ### 插件安装前说明入口 diff --git a/scripts/generate_agent_skill_docs.py b/scripts/generate_agent_skill_docs.py index c1771eddd..c1a35a90e 100644 --- a/scripts/generate_agent_skill_docs.py +++ b/scripts/generate_agent_skill_docs.py @@ -17,8 +17,8 @@ PROJECT_ROOT = Path(__file__).resolve().parents[1] if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) -MOVIEPILOT_API_SKILL_VERSION = "71" -DATABASE_OPERATION_SKILL_VERSION = "17" +MOVIEPILOT_API_SKILL_VERSION = "72" +DATABASE_OPERATION_SKILL_VERSION = "18" DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = { @@ -183,7 +183,9 @@ DATABASE_TABLE_GUIDES: dict[str, tuple[str, str, str]] = { "transferhistory": ( "Stores transfer source, destination, mode, media identity, download linkage, and outcome.", "Reviewing success/failure history, destination paths, media classification, and download linkage.", - "Written by transfer settlement; delete or retry through transfer-history business APIs.", + "Written by transfer settlement or the external registration API; never insert history rows directly. " + "External rows preserve administrator audit and file verification evidence; actor_id is the authoritative " + "identity and actor_name is a bounded display snapshot. Old rows have unknown origin.", ), "transferpending": ( "Durably stores pending transfer input, plans, checkpoints, leases, retries, and manual review state.", @@ -302,6 +304,19 @@ _MEDIA_TYPE_FIELD_VALUES: dict[tuple[str, str, str], str] = { # OpenAPI 无法表达的操作级 Agent 指引，在生成时追加到操作参数组之后， # 保证运行中的 Skill 包含这些指引，无需手动修改生成的 api/*.md。 _OPERATION_NOTES: dict[str, list[str]] = { + "transfer.history.register": [ + "- **Completed external work only:** Register a completed copy, native local hard link, or move; " + "this operation only writes history and never moves files, scrapes metadata, or changes downloads.", + "- Supply administrator-confirmed media identity and absolute paths. The server queries file facts. " + "Copy requires equal file sizes; link requires distinct local directory entries with the same inode; " + "move requires a confirmed absent source and verifies only the destination. These are not content checksums.", + "- Retrying the same source, destination and metadata returns the original receipt. Conflicting or " + "internal history returns a conflict and is never overwritten. Actor and registration time come from the server.", + "- The complete `actor_id` is authoritative. `actor_name` is a display snapshot: nonempty names up to " + "128 characters stay unchanged; longer names use the first 127 characters plus `…`; empty names use `#`.", + "- Use transfer.history with query origin=external to page through external registrations; " + "older records have no recorded origin.", + ], "plugin.instance.purge": [ "- **Historical invalid-ID recovery:** If the exact instance ID from a warning is " "rejected by normal validation, send only `config: true` (all other scope flags false). " diff --git a/skills/database-operation/SKILL.md b/skills/database-operation/SKILL.md index a9d9d523d..6c57ef92a 100644 --- a/skills/database-operation/SKILL.md +++ b/skills/database-operation/SKILL.md @@ -1,6 +1,6 @@ --- name: database-operation -version: 17 +version: 18 description: >- Use this skill when you need to inspect, query, maintain, or carefully modify the MoviePilot database. This skill uses the bundled scripts/mp-db.py helper, @@ -340,8 +340,8 @@ and pass the flag only for an explicitly authorized repair. ### `transferhistory` - Purpose: Stores transfer source, destination, mode, media identity, download linkage, and outcome. - Useful queries: Reviewing success/failure history, destination paths, media classification, and download linkage. -- Write boundary: Written by transfer settlement; delete or retry through transfer-history business APIs. -- Columns: `id`, `transfer_task_id`, `transfer_settlement_revision`, `src`, `src_storage`, `src_fileitem`, `dest`, `dest_storage`, `dest_fileitem`, `mode`, `type`, `media_category_id`, `category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `title`, `year`, `media_source`, `media_id`, `music_type`, `total_tracks`, `audio_format`, `audio_lossless`, `bit_depth`, `sample_rate`, `bitrate`, `seasons`, `episodes`, `image`, `downloader`, `download_hash`, `status`, `errmsg`, `retry_count`, `auto_paused`, `failure_stage`, `recovery_action`, `cleanup_status`, `cleanup_error`, `date`, `files`, `episode_group` +- Write boundary: Written by transfer settlement or the external registration API; never insert history rows directly. External rows preserve administrator audit and file verification evidence; actor_id is the authoritative identity and actor_name is a bounded display snapshot. Old rows have unknown origin. +- Columns: `id`, `transfer_task_id`, `transfer_settlement_revision`, `origin`, `registration`, `src`, `src_storage`, `src_fileitem`, `dest`, `dest_storage`, `dest_fileitem`, `mode`, `type`, `media_category_id`, `category`, `classification_rule_id`, `classification_policy_revision`, `classification_source`, `title`, `year`, `media_source`, `media_id`, `music_type`, `total_tracks`, `audio_format`, `audio_lossless`, `bit_depth`, `sample_rate`, `bitrate`, `seasons`, `episodes`, `image`, `downloader`, `download_hash`, `status`, `errmsg`, `retry_count`, `auto_paused`, `failure_stage`, `recovery_action`, `cleanup_status`, `cleanup_error`, `date`, `files`, `episode_group` ### `transferpending` - Purpose: Durably stores pending transfer input, plans, checkpoints, leases, retries, and manual review state. diff --git a/skills/moviepilot-api/SKILL.md b/skills/moviepilot-api/SKILL.md index 82648d57c..c557ae9f0 100644 --- a/skills/moviepilot-api/SKILL.md +++ b/skills/moviepilot-api/SKILL.md @@ -1,6 +1,6 @@ --- name: moviepilot-api -version: 71 +version: 72 description: >- Use this skill for MoviePilot product operations such as media search, torrent search, downloads, subscriptions, library checks, sites, storage, workflows, @@ -17,11 +17,11 @@ allowed-api-operations: >- download.clients download.paths download.history.list download.history.delete transfer.history.delete site.update site.list site.userdata site.test site.cookie.update recommendation.list library.exists library.latest storage.settings storage.list transfer.history - transfer.file scheduler.list scheduler.run workflow.list workflow.run plugin.installed - plugin.market plugin.capabilities plugin.config.get plugin.config.update plugin.source.options - plugin.source.install plugin.source.change plugin.reload plugin.install plugin.uninstall - slash.list config.identifiers.get config.identifiers.update search.torrents search.results - filter.builtin filter.custom filter.groups filter.custom.add filter.custom.update + transfer.history.register transfer.file scheduler.list scheduler.run workflow.list workflow.run + plugin.installed plugin.market plugin.capabilities plugin.config.get plugin.config.update + plugin.source.options plugin.source.install plugin.source.change plugin.reload plugin.install + plugin.uninstall slash.list config.identifiers.get config.identifiers.update search.torrents + search.results filter.builtin filter.custom filter.groups filter.custom.add filter.custom.update filter.custom.delete filter.group.add filter.group.update filter.group.delete plugin.data config.system.list config.system.describe config.system.get config.system.update slash.run music.recognize music.explore music.album.get music.album.related music.artist.get diff --git a/skills/moviepilot-api/api/transfer.md b/skills/moviepilot-api/api/transfer.md index bf03ceda2..842cb73af 100644 --- a/skills/moviepilot-api/api/transfer.md +++ b/skills/moviepilot-api/api/transfer.md @@ -65,7 +65,7 @@ Purpose: 按条件和分页读取文件整理历史。 Identity scope: `bound_user`; required role: `user`. - `response`: structured page object; items stay in `data.list` and the exact total stays in `data.total`. - `path_params`: none -- `query`: `count` (integer|null; default `30`): 当前页最多返回的记录数。; `page` (integer|null; default `1`): 从 1 开始的结果页码。; `status` (boolean|null): 用于筛选历史或描述记录的整理成功状态。; `title` (string|null): 操作使用的媒体、种子、订阅或历史标题。 +- `query`: `count` (integer|null; default `30`): 当前页最多返回的记录数。; `origin` (string=external|null): 整理历史来源；external 仅返回管理员登记的外部整理记录，省略时保留全部历史。; `page` (integer|null; default `1`): 从 1 开始的结果页码。; `status` (boolean|null): 用于筛选历史或描述记录的整理成功状态。; `title` (string|null): 操作使用的媒体、种子、订阅或历史标题。 - `body`: none ### `transfer.history.clear` @@ -100,6 +100,19 @@ Identity scope: `admin`; required role: `system_admin`. - `query`: none - `body`: `history_ids` (array): 历史记录 ID 列表 +### `transfer.history.register` +`POST /api/v1/history/transfer`; policy effect: `reversible_write`. +Purpose: 管理员登记已完成的外部复制、原生硬链接或移动；严格核验文件事实并记录审计，不操作文件或下载器，同一声明重试返回原回执。 +Identity scope: `admin`; required role: `system_admin`. +- `path_params`: none +- `query`: none +- `body`: `dest*` (string; minimum length `1`): 整理后目标文件的绝对路径; `dest_storage*` (string; minimum length `1`): 目标存储配置标识; `download_hash` (string|null): 已知下载任务标识，仅作关联声明; `episodes` (string|null): 集号或集范围，例如 E01 或 E01-E12; `media_id*` (string; minimum length `1`): 来源中的规范原生媒体标识; `media_source*` (MediaSource): 元数据来源标识，保留与 media_id 一同返回的精确值。; `mode*` (string(copy,link,move)): 已完成的复制、本地硬链接或移动; `seasons` (string|null): 季号或季范围，例如 S01; `src*` (string; minimum length `1`): 整理前源文件的绝对路径; `src_storage*` (string; minimum length `1`): 源存储配置标识; `title*` (string; minimum length `1`): 管理员确认的媒体标题; `type*` (string): 管理员确认的电影或电视剧类型; `year` (string|null): 四位发行年份 +- **Completed external work only:** Register a completed copy, native local hard link, or move; this operation only writes history and never moves files, scrapes metadata, or changes downloads. +- Supply administrator-confirmed media identity and absolute paths. The server queries file facts. Copy requires equal file sizes; link requires distinct local directory entries with the same inode; move requires a confirmed absent source and verifies only the destination. These are not content checksums. +- Retrying the same source, destination and metadata returns the original receipt. Conflicting or internal history returns a conflict and is never overwritten. Actor and registration time come from the server. +- The complete `actor_id` is authoritative. `actor_name` is a display snapshot: nonempty names up to 128 characters stay unchanged; longer names use the first 127 characters plus `…`; empty names use `#`. +- Use transfer.history with query origin=external to page through external registrations; older records have no recorded origin. + ### `transfer.history.replan` `POST /api/v1/history/transfer/{history_id}/replan`; policy effect: `external_side_effect`. Purpose: 为没有文件副作用的识别失败历史提交已核实的电影或电视剧身份和季集修正，保留原任务及失败证据。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index e14d41267..3120ae766 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -959,8 +959,8 @@ "runtime_only": true } }, - "edge_count": 9930, - "edge_sha256": "723708b08703c8b8a3121676a00792baf0fedfff2342aee706b1c7b52d7cf3a2", + "edge_count": 9961, + "edge_sha256": "9df270930880d2f8cbe39f55eb842d884b1b7e19294a71e2269b5b2e4def83bb", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -1162,6 +1162,9 @@ "app.adapters.system.plugin.package -> app.runtime.log", "app.adapters.system.plugin.package -> app.runtime.settings", "app.adapters.system.plugin.package -> app.runtime.version", + "app.adapters.system.registration -> app.adapters", + "app.adapters.system.registration -> app.adapters.system", + "app.adapters.system.registration -> app.adapters.system.fsproxy", "app.adapters.system.resource -> app.adapters", "app.adapters.system.resource -> app.adapters.network", "app.adapters.system.resource -> app.adapters.network.http", @@ -2239,6 +2242,7 @@ "app.api.dependencies.history -> app.application.dashboard", "app.api.dependencies.history -> app.application.history", "app.api.dependencies.history -> app.application.history.mutation", + "app.api.dependencies.history -> app.application.history.registration", "app.api.dependencies.history -> app.application.history.retry", "app.api.dependencies.history -> app.application.mediaserver", "app.api.dependencies.history -> app.application.transfer", @@ -2596,6 +2600,7 @@ "app.api.endpoints.history -> app.application.agent", "app.api.endpoints.history -> app.application.configuration", "app.api.endpoints.history -> app.application.history", + "app.api.endpoints.history -> app.application.history.registration", "app.api.endpoints.history -> app.application.transfer", "app.api.endpoints.history -> app.application.transfer.execution", "app.api.endpoints.history -> app.application.transfer.recovery", @@ -3781,6 +3786,15 @@ "app.application.history.mutation -> app.schemas", "app.application.history.mutation -> app.schemas.common", "app.application.history.mutation -> app.schemas.history", + "app.application.history.registration -> app.application", + "app.application.history.registration -> app.application.history", + "app.application.history.registration -> app.application.history.contracts", + "app.application.history.registration -> app.application.history.mutation", + "app.application.history.registration -> app.schemas", + "app.application.history.registration -> app.schemas.exception", + "app.application.history.registration -> app.schemas.file", + "app.application.history.registration -> app.schemas.history", + "app.application.history.registration -> app.schemas.types", "app.application.history.retry -> app.application", "app.application.history.retry -> app.application.configuration", "app.application.history.retry -> app.application.history", @@ -6459,6 +6473,17 @@ "app.db.adapters.history.download -> app.schemas", "app.db.adapters.history.download -> app.schemas.media", "app.db.adapters.history.download -> app.schemas.types", + "app.db.adapters.history.registration -> app.application", + "app.db.adapters.history.registration -> app.application.history", + "app.db.adapters.history.registration -> app.application.history.contracts", + "app.db.adapters.history.registration -> app.application.history.registration", + "app.db.adapters.history.registration -> app.db", + "app.db.adapters.history.registration -> app.db.adapters", + "app.db.adapters.history.registration -> app.db.adapters.history", + "app.db.adapters.history.registration -> app.db.adapters.history.transfer", + "app.db.adapters.history.registration -> app.db.oper", + "app.db.adapters.history.registration -> app.db.oper.transferhistory", + "app.db.adapters.history.registration -> app.db.uow", "app.db.adapters.history.transfer -> app.application", "app.db.adapters.history.transfer -> app.application.history", "app.db.adapters.history.transfer -> app.application.history.retry", @@ -10054,6 +10079,7 @@ "app.startup.composition.context -> app.application.classification.runtime", "app.startup.composition.context -> app.application.configuration", "app.startup.composition.context -> app.application.history", + "app.startup.composition.context -> app.application.history.registration", "app.startup.composition.context -> app.application.messaging", "app.startup.composition.context -> app.application.messaging.chat", "app.startup.composition.context -> app.application.outbox", @@ -10210,10 +10236,14 @@ "app.startup.composition.resource -> app.runtime.extensions", "app.startup.composition.resource -> app.runtime.extensions.resource", "app.startup.composition.resource -> app.runtime.resources", + "app.startup.composition.runtime -> app.adapters", + "app.startup.composition.runtime -> app.adapters.system", + "app.startup.composition.runtime -> app.adapters.system.registration", "app.startup.composition.runtime -> app.api", "app.startup.composition.runtime -> app.api.data", "app.startup.composition.runtime -> app.application", "app.startup.composition.runtime -> app.application.history", + "app.startup.composition.runtime -> app.application.history.registration", "app.startup.composition.runtime -> app.application.messaging", "app.startup.composition.runtime -> app.application.messaging.message", "app.startup.composition.runtime -> app.application.site", @@ -10228,6 +10258,7 @@ "app.startup.composition.runtime -> app.db.adapters", "app.startup.composition.runtime -> app.db.adapters.history", "app.startup.composition.runtime -> app.db.adapters.history.download", + "app.startup.composition.runtime -> app.db.adapters.history.registration", "app.startup.composition.runtime -> app.db.adapters.history.transfer", "app.startup.composition.runtime -> app.db.adapters.outbox", "app.startup.composition.runtime -> app.db.adapters.site", @@ -10893,7 +10924,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 1147, + "module_count": 1150, "modules": [ "app", "app.adapters", @@ -10943,6 +10974,7 @@ "app.adapters.system.plugin.health", "app.adapters.system.plugin.manifest", "app.adapters.system.plugin.package", + "app.adapters.system.registration", "app.adapters.system.resource", "app.adapters.system.rust", "app.adapters.system.stdio", @@ -11205,6 +11237,7 @@ "app.application.history", "app.application.history.contracts", "app.application.history.mutation", + "app.application.history.registration", "app.application.history.retry", "app.application.image", "app.application.invocation", @@ -11464,6 +11497,7 @@ "app.db.adapters.download", "app.db.adapters.history", "app.db.adapters.history.download", + "app.db.adapters.history.registration", "app.db.adapters.history.transfer", "app.db.adapters.invocation", "app.db.adapters.mediaserver", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index f802fbbee..61139c006 100644 --- a/tests/fixtures/architecture/startup-performance-baseline.json +++ b/tests/fixtures/architecture/startup-performance-baseline.json @@ -6,7 +6,7 @@ "repeat": 3, "targets": { "app.startup.lifecycle": { - "loaded_app_module_count": 599, + "loaded_app_module_count": 600, "max_ms": 1293.338, "median_ms": 1156.239, "min_ms": 1102.806, @@ -17,7 +17,7 @@ ] }, "app.factory": { - "loaded_app_module_count": 613, + "loaded_app_module_count": 614, "max_ms": 1127.911, "median_ms": 1122.382, "min_ms": 1119.221, @@ -28,7 +28,7 @@ ] }, "app.main": { - "loaded_app_module_count": 615, + "loaded_app_module_count": 616, "max_ms": 1188.652, "median_ms": 1183.509, "min_ms": 1174.522, diff --git a/tests/test_agent_api_bodies.py b/tests/test_agent_api_bodies.py index 6d9934c28..eb2181be0 100644 --- a/tests/test_agent_api_bodies.py +++ b/tests/test_agent_api_bodies.py @@ -71,6 +71,12 @@ def api_transport(monkeypatch: pytest.MonkeyPatch) -> tuple[MoviePilotApiTool, A "description": "{\"原始副标题\": \"不应被递归解码\"}", }, "media_source": "tmdb", "media_id": "568160"}), ("transfer.history.delete", {"id": 7}), + ("transfer.history.register", { + "src": "/downloads/a.mkv", "src_storage": "local", + "dest": "/library/a.mkv", "dest_storage": "local", + "mode": "copy", "type": "电影", "title": "A", + "media_source": "tmdb", "media_id": "42", + }), ("slash.run", {"command": "/help"}), ("media.classification.policy.preview", {"input": { "kind": "media", "media": {"title": "天气之子"}, diff --git a/tests/test_agent_api_gateway.py b/tests/test_agent_api_gateway.py index 2eb46702f..f7a0ddbcb 100644 --- a/tests/test_agent_api_gateway.py +++ b/tests/test_agent_api_gateway.py @@ -47,12 +47,12 @@ def test_api_operation_registry_matches_migration_batches() -> None: """ API 操作注册表必须覆盖各迁移批次且每项具有固定路由。 """ - assert len(API_FIRST_BATCH_OPERATION_SPECS) == 53 + assert len(API_FIRST_BATCH_OPERATION_SPECS) == 54 assert len(API_PARITY_OPERATION_SPECS) == 17 assert len(API_MUSIC_OPERATION_SPECS) == 10 assert len(API_SYSTEM_OPERATION_SPECS) == 7 assert len(API_EXTENDED_OPERATION_SPECS) == 145 - assert len(API_OPERATION_SPECS) == 232 + assert len(API_OPERATION_SPECS) == 233 assert {spec.operation_id for spec in API_OPERATION_SPECS} == set(API_OPERATION_ROUTES) assert { "download.list", diff --git a/tests/test_agent_api_identity.py b/tests/test_agent_api_identity.py index 0fd7ad7e0..70bf06612 100644 --- a/tests/test_agent_api_identity.py +++ b/tests/test_agent_api_identity.py @@ -345,7 +345,7 @@ def test_identity_scopes_cover_every_operation_and_match_rest_dependencies(): path = re.sub(r"\{[^}]+\}", "{}", "/api/v1" + router_spec.prefix + route.path) for method in route.methods: endpoints[method, path] = route - assert len(API_OPERATION_SPECS) == len(API_OPERATION_ROUTES) == 232 + assert len(API_OPERATION_SPECS) == len(API_OPERATION_ROUTES) == 233 for spec in API_OPERATION_SPECS: assert isinstance(spec.identity_scope, ApiIdentityScope) if spec.identity_scope is not ApiIdentityScope.PUBLIC: diff --git a/tests/test_agent_skills_middleware.py b/tests/test_agent_skills_middleware.py index 20a4efaa0..bbd5d4008 100644 --- a/tests/test_agent_skills_middleware.py +++ b/tests/test_agent_skills_middleware.py @@ -187,7 +187,7 @@ async def test_bundled_moviepilot_api_skill_loads_complete_contract() -> None: assert payload["content_limit_bytes"] == MAX_SKILL_CONTENT_BYTES assert payload["truncated"] is False assert payload["truncation_message"] is None - assert len(payload["skill"]["allowed_api_operations"]) == 232 + assert len(payload["skill"]["allowed_api_operations"]) == 233 assert "## API Category Index" in payload["content"] assert "### `workflow.update`" not in payload["content"] assert "api/workflow.md" in payload["supporting_files"] diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index 8bdbf8762..bf3594694 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -127,10 +127,10 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: expected_versions = { "browser-use": "4", "command-dispatch": "2", - "database-operation": "17", + "database-operation": "18", "feedback-issue": "12", "anysearch": "3", - "moviepilot-api": "71", + "moviepilot-api": "72", "moviepilot-update": "7", "organize-files": "6", "transfer-failed-retry": "6", diff --git a/tests/test_external_transfer_registration.py b/tests/test_external_transfer_registration.py new file mode 100644 index 000000000..8570848c2 --- /dev/null +++ b/tests/test_external_transfer_registration.py @@ -0,0 +1,879 @@ +""" +外部整理登记的请求、只读文件证明和原子历史回执回归。 +""" + +import asyncio +import importlib +import os +from concurrent.futures import ThreadPoolExecutor +from dataclasses import FrozenInstanceError, replace +from functools import partial +from pathlib import Path +from threading import Barrier +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from pydantic import ValidationError +from sqlalchemy import create_engine, select +from sqlalchemy.orm import sessionmaker + +from app.db.models.transferhistory import TransferHistory +from app.schemas import history as history_schema +from app.schemas.file import FileItem + + +def _request(**changes): + """ + 构造管理员明确提供的媒体和路径声明。 + """ + request_type = getattr(history_schema, "TransferHistoryRegisterRequest", None) + assert request_type is not None, "缺少外部整理登记请求" + payload = { + "src": "/downloads/a.mkv", "dest": "/library/a.mkv", + "src_storage": "local", "dest_storage": "local", "mode": "copy", + "type": "电影", "title": "A", "year": "2026", + "media_source": "tmdb", "media_id": "42", + } + return request_type.model_validate(payload | changes) + + +@pytest.mark.parametrize("mode", ["copy", "link", "move"]) +def test_registration_request_accepts_only_explicit_metadata(mode): + """ + 三种登记模式保留规范身份和路径，写入用例另行冻结声明。 + """ + request = _request(mode=mode, media_id=" 42 ", src="/downloads//a.mkv") + assert request.media_id == "42" + assert request.src == "/downloads/a.mkv" + + +@pytest.mark.parametrize("changes", [ + {"src": "relative.mkv"}, {"src": "/downloads/../a.mkv"}, + {"src": "/bad\x00.mkv"}, {"dest": "/downloads/a.mkv"}, + {"src": "/downloads/a.mkv\n"}, {"src_storage": "\tlocal"}, + {"dest": "/library/\u0085a.mkv"}, + {"src_storage": ""}, {"title": ""}, {"title": "x" * 513}, + {"src": "/" + "x" * 4096}, {"dest_storage": "x" * 65}, + {"media_id": "x" * 129}, {"download_hash": "x" * 129}, + {"mode": "softlink"}, {"type": "音乐"}, {"year": "tomorrow"}, + {"media_id": "0"}, {"media_source": None}, {"seasons": "season 1"}, + {"episodes": "E00"}, {"actor_id": 99}, {"origin": "external"}, + {"status": True}, {"src_fileitem": {}}, {"transfer_task_id": "task"}, +]) +def test_registration_request_rejects_ambiguous_or_injected_fields(changes): + """ + 无法证明的路径和服务端拥有的身份、状态字段不得由请求注入。 + """ + with pytest.raises(ValidationError): + _request(**changes) + + +@pytest.fixture +def registration_modules(): + """ + 加载待实现的显式登记用例与适配器。 + """ + names = [ + "app.application.history.registration", + "app.db.adapters.history.registration", + "app.adapters.system.registration", + ] + for name in names: + assert importlib.util.find_spec(name) is not None, f"缺少登记模块 {name}" + return tuple(importlib.import_module(name) for name in names) + + +@pytest.fixture +def registration_store(tmp_path, registration_modules): + """ + 为并发和回滚回归创建真实文件数据库和独立会话工厂。 + """ + _, database, _ = registration_modules + engine = create_engine( + f"sqlite:///{tmp_path / 'registration.db'}", + connect_args={"check_same_thread": False, "timeout": 10}, + ) + TransferHistory.__table__.create(engine) + sessions = sessionmaker(engine) + yield sessions, database.TransferRegistrationScopeFactory(sessions) + engine.dispose() + + +def _local_query(storage, path): + """ + 用真实文件系统模拟严格 provider 查询，不执行任何写操作。 + """ + try: + info = path.stat() + except FileNotFoundError: + return None + return FileItem( + path=str(path), storage=storage, type="file" if path.is_file() else "dir", + size=info.st_size, modify_time=info.st_mtime, + fileid=f"{info.st_dev}:{info.st_ino}", + url="https://private.invalid/?passkey=secret", + ) + + +def _command(modules, scope, query=_local_query): + """ + 只注入只读查询和短事务能力，缺少删除、整理、下载和事件能力。 + """ + application, _, local = modules + return application.TransferRegistrationCommand( + scope_factory=scope, query_file=query, + local_entry=partial(local.probe_local_entry, entry_factory=application.LocalRegistrationEntry), + ) + + +def _actor(modules, name="admin"): + """ + 模拟认证依赖返回的服务端管理员身份。 + """ + return modules[0].RegistrationActor(id=1, name=name) + + +def test_copy_registers_frozen_facts_and_returns_original_receipt( + tmp_path, registration_modules, registration_store, +): + """ + 登记只记录存在事实，重复请求保留最初审计人和时间且不重新探测文件。 + """ + sessions, scope = registration_store + src, dest = tmp_path / "a.mkv", tmp_path / "b.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + request = _request(src=str(src), dest=str(dest)) + command = _command(registration_modules, scope) + result = command.execute(request, _actor(registration_modules)) + assert result.created + assert result.history.origin == "external" + assert result.history.transfer_task_id is None + assert result.history.transfer_settlement_revision is None + assert result.history.registration["verification"] == "matching_sizes" + assert result.history.registration["actor_name"] == "admin" + assert result.history.src_fileitem["size"] == 4 + assert "url" not in result.history.src_fileitem + with pytest.raises(TypeError): + result.history.registration["actor_name"] = "changed" + src.unlink() + dest.unlink() + repeated = command.execute(request, _actor(registration_modules, "another")) + assert not repeated.created + assert repeated.history == result.history + with sessions() as session: + assert len(session.scalars(select(TransferHistory)).all()) == 1 + + +def test_move_records_only_destination_facts(tmp_path, registration_modules, registration_store): + """ + 移动后的源确实不存在时不伪造源大小、标识或内容一致证据。 + """ + _, scope = registration_store + dest = tmp_path / "b.mkv" + dest.write_bytes(b"film") + src = tmp_path / "missing.mkv" + result = _command(registration_modules, scope).execute( + _request(src=str(src), dest=str(dest), mode="move"), _actor(registration_modules), + ) + assert result.history.src_fileitem == {"path": str(src), "storage": "local"} + assert result.history.registration["verification"] == "destination_only" + assert result.history.registration["source_state"] == "absent" + + +@pytest.mark.parametrize("case", ["missing", "size", "directory", "symlink", "dangling_move", "present_move"]) +def test_invalid_file_facts_cannot_create_history( + case, tmp_path, registration_modules, registration_store, +): + """ + 缺失、不等大小、目录和符号链接不得变成成功历史。 + """ + sessions, scope = registration_store + src, dest = tmp_path / "a.mkv", tmp_path / "b.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + mode = "copy" + if case == "missing": + dest.unlink() + elif case == "size": + dest.write_bytes(b"different") + elif case == "directory": + dest.unlink() + dest.mkdir() + elif case == "symlink": + dest.unlink() + dest.symlink_to(src) + elif case == "dangling_move": + src.unlink() + src.symlink_to(tmp_path / "missing") + mode = "move" + elif case == "present_move": + mode = "move" + with pytest.raises(registration_modules[0].RegistrationValidationError): + _command(registration_modules, scope).execute( + _request(src=str(src), dest=str(dest), mode=mode), _actor(registration_modules), + ) + with sessions() as session: + assert session.scalars(select(TransferHistory)).all() == [] + + +@pytest.mark.parametrize("mode", ["copy", "link"]) +def test_copy_and_link_reject_confirmed_missing_source( + mode, tmp_path, registration_modules, registration_store, +): + """ + 目标普通文件有效时，复制或硬链接仍须因源确认缺失而拒绝，且不创建文件或历史。 + """ + sessions, scope = registration_store + src, dest = tmp_path / "missing.mkv", tmp_path / "dest.mkv" + dest.write_bytes(b"film") + before = dest.stat() + with pytest.raises( + registration_modules[0].RegistrationValidationError, + match="^源文件不存在，不能核验复制或硬链接$", + ): + _command(registration_modules, scope).execute( + _request(src=str(src), dest=str(dest), mode=mode), _actor(registration_modules), + ) + assert not src.exists() + assert dest.read_bytes() == b"film" + after = dest.stat() + assert (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns) == ( + before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns, + ) + with sessions() as session: + assert session.scalars(select(TransferHistory)).all() == [] + + +def test_local_link_rejects_equal_size_files_with_distinct_inodes( + tmp_path, registration_modules, registration_store, +): + """ + 两个等长普通文件具有不同 inode 时必须进入原生硬链接拒绝分支，保留原文件且不写历史。 + """ + sessions, scope = registration_store + src, dest = tmp_path / "source.mkv", tmp_path / "dest.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + before = [path.stat() for path in (src, dest)] + assert before[0].st_size == before[1].st_size + assert (before[0].st_dev, before[0].st_ino) != (before[1].st_dev, before[1].st_ino) + with pytest.raises( + registration_modules[0].RegistrationValidationError, + match="^源文件与目标文件不是同一原生硬链接$", + ): + _command(registration_modules, scope).execute( + _request(src=str(src), dest=str(dest), mode="link"), _actor(registration_modules), + ) + for path, original in zip((src, dest), before): + assert path.read_bytes() == b"film" + after = path.stat() + assert (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns) == ( + original.st_dev, original.st_ino, original.st_size, original.st_mtime_ns, original.st_ctime_ns, + ) + with sessions() as session: + assert session.scalars(select(TransferHistory)).all() == [] + + +@pytest.mark.parametrize("enabled", [False, True]) +def test_local_entry_proof_uses_lstat_in_both_proxy_modes( + enabled, tmp_path, monkeypatch, registration_modules, +): + """ + 开启和关闭代理都能识别最终符号链接及不存在状态。 + """ + from app.adapters.system.fsproxy import FileSystemProxy + + monkeypatch.setattr(FileSystemProxy, "_enabled", staticmethod(lambda: enabled)) + src, alias = tmp_path / "a.mkv", tmp_path / "alias.mkv" + src.write_bytes(b"film") + alias.symlink_to(src) + probe = partial( + registration_modules[2].probe_local_entry, entry_factory=registration_modules[0].LocalRegistrationEntry, + ) + assert probe(str(src)).is_file + assert probe(str(alias)).is_symlink + assert probe(str(tmp_path / "missing")) is None + + +def test_real_hardlink_is_accepted_but_aliases_are_rejected( + tmp_path, registration_modules, registration_store, +): + """ + 硬链接须为不同目录项，父目录别名和仅大小写不同的同项均拒绝。 + """ + _, scope = registration_store + folder = tmp_path / "real" + folder.mkdir() + src = folder / "Entry.mkv" + dest = folder / "other.mkv" + src.write_bytes(b"film") + os.link(src, dest) + command = _command(registration_modules, scope) + result = command.execute( + _request(src=str(src), dest=str(dest), mode="link"), _actor(registration_modules), + ) + assert result.history.registration["verification"] == "native_hardlink" + alias = tmp_path / "alias" + alias.symlink_to(folder, target_is_directory=True) + for path in (alias / src.name, folder / "entry.mkv"): + with pytest.raises(registration_modules[0].RegistrationValidationError): + # 换一个独立 source identity，避免已登记记录提前返回冲突。 + command._verify_files(_request(src=str(src), dest=str(path), mode="link")) + + +@pytest.mark.parametrize("mode", ["copy", "link"]) +def test_unicode_equivalent_names_cannot_register_the_same_entry( + mode, tmp_path, registration_modules, registration_store, +): + """ + NFC/NFD 地址在支持规范等价的文件系统上必须作为同一目录项拒绝。 + """ + application, _, local = registration_modules + sessions, scope = registration_store + src, alias = tmp_path / "caf\u00e9.mkv", tmp_path / "cafe\u0301.mkv" + src.write_bytes(b"film") + supports_alias = alias.exists() + if supports_alias: + assert src.samefile(alias) + probe = partial(local.probe_local_entry, entry_factory=application.LocalRegistrationEntry) + + def local_entry(path): + """ + 优先读取真实路径；区分规范形式的平台模拟同一目录项的第二种拼写。 + """ + if supports_alias or path == str(src): + return probe(path) + return replace(probe(str(src)), name=Path(path).name) + + def query(storage, path): + """ + 让不支持规范等价的平台也提供同文件、同父目录的可复现查询事实。 + """ + if supports_alias or path == src: + return _local_query(storage, path) + return _local_query(storage, src).model_copy(update={"path": str(path)}) + + command = application.TransferRegistrationCommand( + scope_factory=scope, query_file=query, local_entry=local_entry, + ) + with pytest.raises(application.RegistrationValidationError): + command.execute(_request(src=str(src), dest=str(alias), mode=mode), _actor(registration_modules)) + with sessions() as session: + assert session.scalars(select(TransferHistory)).all() == [] + + +def test_concurrent_registration_returns_one_created_receipt( + tmp_path, registration_modules, registration_store, +): + """ + 两线程独立 Session 必须由数据库唯一约束仲裁为同一回执。 + """ + sessions, scope = registration_store + src, dest = tmp_path / "a.mkv", tmp_path / "b.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + request = _request(src=str(src), dest=str(dest)) + barrier = Barrier(2) + + def query(storage, path): + """ + 两个独立读会话都确认无历史后才开始文件核验，强制覆盖原子插入竞争。 + """ + if path == src: + barrier.wait(timeout=10) + return _local_query(storage, path) + + command = _command(registration_modules, scope, query) + with ThreadPoolExecutor(max_workers=2) as executor: + results = list(executor.map( + lambda _: command.execute(request, _actor(registration_modules)), range(2), + )) + assert sorted(result.created for result in results) == [False, True] + assert results[0].history.id == results[1].history.id + with sessions() as session: + assert len(session.scalars(select(TransferHistory)).all()) == 1 + + +@pytest.mark.parametrize("change", [ + {"dest": "/other.mkv"}, {"title": "Other"}, {"mode": "move"}, +]) +def test_existing_source_is_never_replaced( + change, tmp_path, registration_modules, registration_store, +): + """ + 同源的不同目标或媒体声明必须冲突，不能覆盖原审计。 + """ + sessions, scope = registration_store + src, dest = tmp_path / "a.mkv", tmp_path / "b.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + request = _request(src=str(src), dest=str(dest)) + command = _command(registration_modules, scope) + first = command.execute(request, _actor(registration_modules)) + with pytest.raises(registration_modules[0].RegistrationConflictError): + command.execute(request.model_copy(update=change), _actor(registration_modules)) + with sessions() as session: + assert session.get(TransferHistory, first.history.id).title == "A" + + +@pytest.mark.parametrize("active,superuser,manage,expected", [ + (True, True, False, 200), (True, False, False, 403), + (True, False, True, 403), (False, True, True, 401), +]) +def test_registration_route_requires_active_superuser( + active, superuser, manage, expected, tmp_path, registration_modules, registration_store, +): + """ + 路由调用真实认证依赖；仅 manage 用户和未激活管理员均不能登记。 + """ + from app.api.dependencies import auth + from app.api.dependencies import history as dependencies + from app.api.endpoints import history + + dependency = getattr(dependencies, "get_transfer_registration_command", None) + assert dependency is not None, "缺少登记依赖" + _, scope = registration_store + src, dest = tmp_path / "a.mkv", tmp_path / "b.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + application = FastAPI() + application.include_router(history.router, prefix="/api/v1/history") + application.dependency_overrides[auth.get_current_user] = lambda: SimpleNamespace( + id=7, name="server-admin", is_active=active, is_superuser=superuser, permissions={"manage": manage}, + ) + application.dependency_overrides[dependency] = lambda: _command(registration_modules, scope) + with TestClient(application) as client: + response = client.post( + "/api/v1/history/transfer", + json=_request(src=str(src), dest=str(dest)).model_dump(mode="json"), + ) + assert response.status_code == expected + if expected == 200: + body = response.json()["data"] + assert body["created"] + assert body["history"]["registration"]["actor_id"] == 7 + assert body["history"]["registration"]["actor_name"] == "server-admin" + + +@pytest.mark.parametrize("name,display", [ + ("n" * 128, "n" * 128), ("名" * 129, "名" * 127 + "…"), + ("", "#7"), ("admin\n", "admin\n"), +], ids=["maximum", "long", "empty", "control"]) +def test_registration_command_preserves_identity_with_bounded_actor_display( + name, display, tmp_path, registration_modules, registration_store, +): + """ + 既有合法用户名用有界显示快照登记，真实身份和重试回执均保持最初值。 + """ + from app.schemas.user import UserCreate + + user = UserCreate(name=name, is_active=True, is_superuser=True) + _, scope = registration_store + src, dest = tmp_path / "source.mkv", tmp_path / "dest.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + request = _request(src=str(src), dest=str(dest)) + command = _command(registration_modules, scope) + first = command.execute(request, registration_modules[0].RegistrationActor(id=7, name=user.name)) + assert first.created + assert first.history.registration["actor_id"] == 7 + assert first.history.registration["actor_name"] == display + assert 1 <= len(first.history.registration["actor_name"]) <= 128 + src.unlink() + dest.unlink() + repeated = command.execute(request, registration_modules[0].RegistrationActor(id=8, name="another")) + assert not repeated.created + assert repeated.history == first.history + + +@pytest.mark.parametrize("name,display", [ + ("n" * 128, "n" * 128), ("名" * 129, "名" * 127 + "…"), + ("", "#7"), ("admin\n", "admin\n"), +], ids=["maximum", "long", "empty", "control"]) +def test_registration_http_accepts_existing_admin_name_contract( + name, display, tmp_path, registration_modules, registration_store, +): + """ + 真实认证依赖接受的长名和空名不得导致 500，HTTP 重试保留首次身份和显示快照。 + """ + from app.api.dependencies import auth + from app.api.dependencies import history as dependencies + from app.api.endpoints import history + from app.schemas.user import UserCreate + + user = UserCreate(name=name, is_active=True, is_superuser=True) + actor = SimpleNamespace(id=7, name=user.name, is_active=user.is_active, is_superuser=user.is_superuser) + sessions, scope = registration_store + src, dest = tmp_path / "source.mkv", tmp_path / "dest.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + payload = _request(src=str(src), dest=str(dest)).model_dump(mode="json") + application = FastAPI() + application.include_router(history.router, prefix="/api/v1/history") + application.dependency_overrides[auth.get_current_user] = lambda: actor + application.dependency_overrides[dependencies.get_transfer_registration_command] = lambda: _command( + registration_modules, scope, + ) + with TestClient(application, raise_server_exceptions=False) as client: + response = client.post("/api/v1/history/transfer", json=payload) + assert response.status_code == 200 + first = response.json()["data"] + assert first["created"] + assert first["history"]["registration"]["actor_id"] == 7 + assert first["history"]["registration"]["actor_name"] == display + assert 1 <= len(first["history"]["registration"]["actor_name"]) <= 128 + src.unlink() + dest.unlink() + actor.id, actor.name = 8, "another" + response = client.post("/api/v1/history/transfer", json=payload) + assert response.status_code == 200 + repeated = response.json()["data"] + assert not repeated["created"] + assert repeated["history"] == first["history"] + with sessions() as session: + rows = session.scalars(select(TransferHistory)).all() + assert len(rows) == 1 + assert rows[0].registration["actor_id"] == 7 + assert rows[0].registration["actor_name"] == display + + +@pytest.mark.parametrize("title", [None, "Movie", "*.mkv"]) +@pytest.mark.parametrize("status", [None, True, False]) +def test_external_origin_filter_count_and_page_agree(title, status, registration_store): + """ + 来源过滤贯穿普通标题、通配标题、状态和分页，旧查询仍包含未记录来源行。 + """ + from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + + from app.application.history import HistoryQueryService + from app.db.adapters.history.transfer import TransactionalTransferHistoryRepository + + sessions, _ = registration_store + with sessions() as session: + for number, (origin, successful) in enumerate([ + (None, True), ("external", True), ("external", True), ("external", False), + ]): + session.add(TransferHistory( + src=f"/src/{number}.mkv", src_storage="local", dest=f"/dest/{number}.mkv", + title="Movie", origin=origin, status=successful, date=f"2026-10-0{number + 1}", + )) + session.commit() + engine = create_async_engine(str(sessions.kw["bind"].url).replace("sqlite:", "sqlite+aiosqlite:")) + async_sessions = async_sessionmaker(engine) + repository = TransactionalTransferHistoryRepository(sync_session=sessions, async_session=async_sessions) + service = HistoryQueryService(download_repository=None, transfer_repository=repository) + + async def check(): + """ + 在同一事件循环验证 count、page 和未过滤查询。 + """ + try: + result = await service.list_transfer(title=title, status=status, origin="external", page=1, count=1) + assert result.total == (3 if status is None else 2 if status else 1) + assert len(result.list) == 1 + assert result.list[0].origin == "external" + ordinary = await service.list_transfer(title=title, status=status) + assert ordinary.total == (4 if status is None else 3 if status else 1) + finally: + await engine.dispose() + + asyncio.run(check()) + + +@pytest.mark.parametrize("origin", [None, "external"]) +def test_history_route_passes_only_explicit_origin_to_query(origin): + """ + 真实路由保留旧查询的参数省略形态，显式外部来源仍完整传给查询服务。 + """ + from app.api.dependencies import history as dependencies + from app.api.endpoints import history + + query = SimpleNamespace(list_transfer=AsyncMock(return_value={"list": [], "total": 0})) + application = FastAPI() + application.include_router(history.router, prefix="/api/v1/history") + application.dependency_overrides[history.verify_token] = lambda: None + application.dependency_overrides[dependencies.get_history_query_service] = lambda: query + parameters = {"title": "Movie*", "page": 2, "count": 5, "status": "true"} + expected = {"title": "Movie*", "page": 2, "count": 5, "status": True} + if origin is not None: + parameters["origin"] = origin + expected["origin"] = origin + with TestClient(application) as client: + response = client.get("/api/v1/history/transfer", params=parameters) + assert response.status_code == 200 + assert response.json()["data"] == {"list": [], "total": 0} + query.list_transfer.assert_awaited_once_with(**expected) + + +def test_registration_mcp_route_is_admin_idempotent_private_write(): + """ + MCP 只提供登记业务入口，声明私人结果和可重试的管理员数据库写操作。 + """ + from app.agent.policy.api import API_OPERATION_ROUTES, API_OPERATION_SPECS + from app.agent.policy.contracts import ActionEffect, PrincipalRole, RecoveryMode + + route = API_OPERATION_ROUTES.get("transfer.history.register") + assert route is not None + assert route.method == "POST" + assert route.path == "/api/v1/history/transfer" + spec = next(spec for spec in API_OPERATION_SPECS if spec.operation_id == "transfer.history.register") + assert spec.effect == ActionEffect.REVERSIBLE_WRITE + assert spec.recovery == RecoveryMode.IDEMPOTENT + assert spec.required_role == PrincipalRole.SYSTEM_ADMIN + + +def test_registration_atomic_insert_rejects_unknown_dialect(monkeypatch, registration_store): + """ + 未明确支持的数据库不能退化为先查后插或替换写入。 + """ + from app.db.oper.transferhistory import TransferHistoryOper + + sessions, _ = registration_store + with sessions() as session: + monkeypatch.setattr(session.get_bind().dialect, "name", "unsupported") + with pytest.raises(RuntimeError, match="不支持原子"): + TransferHistoryOper(session).stage_insert_if_source_absent({ + "src": "/a.mkv", "src_storage": "local", + }) + + +@pytest.mark.parametrize("provide_async_session", [False, True]) +def test_registration_stage_requires_explicit_sync_session(provide_async_session, monkeypatch): + """ + 原子登记拒绝缺失或异步调用方会话，不得自行创建会话、执行语句或提交事务。 + """ + from sqlalchemy.ext.asyncio import AsyncSession + from sqlalchemy.orm import Session + + from app.db.oper.transferhistory import TransferHistoryOper + + supplied_session = AsyncSession() if provide_async_session else None + database_access = Mock(side_effect=AssertionError("拒绝无同步会话的登记时不得访问数据库")) + try: + with monkeypatch.context() as guarded: + for session_type in (Session, AsyncSession): + for method in ("__init__", "execute", "flush", "commit"): + guarded.setattr(session_type, method, database_access) + oper = TransferHistoryOper(supplied_session) + with pytest.raises(RuntimeError, match="^登记整理历史需要调用方提供同步 Session$"): + oper.stage_insert_if_source_absent({"src": "/a.mkv", "src_storage": "local"}) + database_access.assert_not_called() + finally: + if supplied_session is not None: + asyncio.run(supplied_session.close()) + + +def test_registration_accepts_canonical_series_identity(): + """ + 电视剧的季集范围与非 TMDB 原生媒体身份完整保留。 + """ + request = _request(type="电视剧", media_source="imdb", media_id="tt1234567", seasons="S01", episodes="E01-E12") + assert request.media_id == "tt1234567" + assert request.seasons == "S01" + assert request.episodes == "E01-E12" + + +def test_move_accepts_missing_source_parent(tmp_path, registration_modules, registration_store): + """ + 移动后源父目录已清理也属于确认不存在，不要求收集该父目录的身份。 + """ + _, scope = registration_store + src, dest = tmp_path / "removed-parent" / "source.mkv", tmp_path / "dest.mkv" + dest.write_bytes(b"film") + result = _command(registration_modules, scope).execute( + _request(src=str(src), dest=str(dest), mode="move"), _actor(registration_modules), + ) + assert result.history.registration["source_state"] == "absent" + + +@pytest.mark.parametrize("phase", ["stage", "commit"]) +def test_registration_rolls_back_failed_write( + phase, tmp_path, monkeypatch, registration_modules, registration_store, +): + """ + 写入或提交异常后数据库没有半条成功历史，调用方也拿不到成功回执。 + """ + from app.db.uow import SqlAlchemyUnitOfWork + + sessions, scope = registration_store + src, dest = tmp_path / "a.mkv", tmp_path / "b.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + if phase == "stage": + repository = registration_modules[1].SessionTransferRegistrationRepository + original = repository.insert_if_absent + + def fail_after_insert(self, write): + """ + 在真实 INSERT 之后模拟后续投影失败。 + """ + original(self, write) + raise RuntimeError("stage failed") + + monkeypatch.setattr(repository, "insert_if_absent", fail_after_insert) + else: + def fail_commit(_self): + """ + 在真实事务提交前模拟失败。 + """ + raise RuntimeError("commit failed") + + monkeypatch.setattr(SqlAlchemyUnitOfWork, "commit", fail_commit) + with pytest.raises(RuntimeError, match=f"{phase} failed"): + _command(registration_modules, scope).execute( + _request(src=str(src), dest=str(dest)), _actor(registration_modules), + ) + with sessions() as session: + assert session.scalars(select(TransferHistory)).all() == [] + + +@pytest.mark.parametrize("existing", [ + {"origin": None, "status": True}, + {"origin": "external", "status": False}, + {"origin": "external", "status": True, "transfer_task_id": "durable-task"}, +]) +def test_existing_internal_failure_and_task_history_are_protected( + existing, registration_modules, registration_store, +): + """ + 无论目标事实如何，都不得替换内部、失败或持久任务绑定历史。 + """ + sessions, scope = registration_store + with sessions() as session: + record = TransferHistory(src="/downloads/a.mkv", src_storage="local", title="original", **existing) + session.add(record) + session.commit() + original_id = record.id + with pytest.raises(registration_modules[0].RegistrationConflictError): + _command(registration_modules, scope).execute(_request(), _actor(registration_modules)) + with sessions() as session: + rows = session.scalars(select(TransferHistory)).all() + assert len(rows) == 1 + assert rows[0].id == original_id + assert rows[0].title == "original" + + +@pytest.mark.parametrize("case", ["unsupported", "error", "wrong_path", "wrong_storage", "size", "nan", "remote_link"]) +def test_unknown_or_inconsistent_provider_facts_fail_safely( + case, registration_modules, registration_store, +): + """ + 远端 None、异常和错误文件项不能混同，云盘相同 fileid 也不能证明原生硬链接。 + """ + from app.schemas.exception import StorageQueryError + + sessions, scope = registration_store + + def query(storage, path): + """ + 模拟有明确失败或不一致回执的远端提供方。 + """ + if case in ("unsupported", "error"): + raise StorageQueryError("private URL https://secret.invalid/?apikey=secret") + changes = { + "wrong_path": {"path": "/other.mkv"}, "wrong_storage": {"storage": "other"}, + "size": {"size": -1}, "nan": {"modify_time": float("nan")}, + }.get(case, {}) + return FileItem.model_validate({ + "path": str(path), "storage": storage, "type": "file", "size": 4, "fileid": "same", + } | changes) + + with pytest.raises(registration_modules[0].RegistrationValidationError) as error: + _command(registration_modules, scope, query).execute( + _request(src_storage="u115", dest_storage="u115", mode="link" if case == "remote_link" else "copy"), + _actor(registration_modules), + ) + assert "secret" not in str(error.value) + with sessions() as session: + assert session.scalars(select(TransferHistory)).all() == [] + + +def test_external_declaration_is_frozen_before_file_io( + tmp_path, registration_modules, registration_store, +): + """ + 文件查询回调改变原请求也不能篡改已经冻结的媒体声明。 + """ + _, scope = registration_store + src, dest = tmp_path / "a.mkv", tmp_path / "b.mkv" + src.write_bytes(b"film") + dest.write_bytes(b"film") + request = _request(src=str(src), dest=str(dest)) + declaration = registration_modules[0].TransferRegistrationDeclaration.from_request(request) + with pytest.raises(FrozenInstanceError): + declaration.title = "changed" + + def query(storage, path): + """ + 模拟调用方在 I/O 期间仍持有并改动请求对象。 + """ + request.title = "changed" + return _local_query(storage, path) + + result = _command(registration_modules, scope, query).execute(request, _actor(registration_modules)) + assert result.history.title == "A" + + +def test_registration_migration_preserves_old_rows_and_unique_source(tmp_path): + """ + 迁移可重放且保留旧行 NULL 来源、源地址唯一约束，降级不丢历史行。 + """ + from alembic.migration import MigrationContext + from alembic.operations import Operations + from sqlalchemy import Column, Index, Integer, MetaData, String, Table, inspect + from sqlalchemy.exc import IntegrityError + + path = Path("database/versions/b849e613a72f_external_history_registration.py") + spec = importlib.util.spec_from_file_location("registration_migration", path) + migration = importlib.util.module_from_spec(spec) + spec.loader.exec_module(migration) + engine = create_engine(f"sqlite:///{tmp_path / 'migration.db'}") + metadata = MetaData() + table = Table( + "transferhistory", metadata, Column("id", Integer, primary_key=True), + Column("src", String), Column("src_storage", String), + Index("ux_transferhistory_src_storage", "src", "src_storage", unique=True), + ) + metadata.create_all(engine) + with engine.begin() as connection: + connection.execute(table.insert().values(src="/a.mkv", src_storage="local")) + migration.op = Operations(MigrationContext.configure(connection)) + migration.upgrade() + migration.upgrade() + reflected = Table("transferhistory", MetaData(), autoload_with=connection) + row = connection.execute(select(reflected)).one() + assert row.origin is None and row.registration is None + connection.execute(reflected.insert().values( + src="/b.mkv", src_storage="local", origin="external", registration={"actor_id": 1}, + )) + with pytest.raises(IntegrityError): + connection.execute(reflected.insert().values(src="/a.mkv", src_storage="local")) + migration.downgrade() + assert {column["name"] for column in inspect(connection).get_columns("transferhistory")} == { + "id", "src", "src_storage", + } + assert len(connection.execute(select(table)).all()) == 2 + engine.dispose() + + +@pytest.mark.parametrize("locale", ["en-US", "zh-TW"]) +def test_registration_errors_have_translations(locale): + """ + 登记的权限外错误也提供英文和繁体中文，不能向调用方泄漏存储原始异常。 + """ + from app.runtime.localization import LocaleHelper + + for message in ( + "该源地址已有内部整理或失败历史，不能覆盖", + "该源地址的历史不属于可重复的外部登记", + "该源地址已有不同目标或媒体声明的历史，不能覆盖", + "无法确认存储文件状态，请检查存储后重试", + "外部登记只接受普通文件，不接受符号链接或目录", + "移动登记要求源文件已经不存在", + "当前仅支持核验本地存储的原生硬链接", + ): + assert LocaleHelper.translate_text(message, locale=locale) != message diff --git a/tests/test_host_runtime_context.py b/tests/test_host_runtime_context.py index 94800ea0b..8deb01fb7 100644 --- a/tests/test_host_runtime_context.py +++ b/tests/test_host_runtime_context.py @@ -234,6 +234,8 @@ def _runtime() -> HostRuntime: transfer_mutation_repository=_Repository, media_server_repository=_Repository, transfer_execution_repository=_Repository(object()), + registration_scope=lambda: None, + registration_local_entry=lambda _path: None, ), site=SiteRuntime( repository=_Repository, diff --git a/tests/test_runtime_composition.py b/tests/test_runtime_composition.py index 846e8bf65..f7274b483 100644 --- a/tests/test_runtime_composition.py +++ b/tests/test_runtime_composition.py @@ -125,6 +125,27 @@ def test_runtime_composition_reuses_dependencies_and_projects_api_identity() -> assert composition.site_health._repository is inputs.dependencies.site +def test_runtime_composition_injects_application_owned_registration_fact(tmp_path) -> None: + """ + 系统适配器通过组合根注入的构造器返回应用冻结 DTO，不反向依赖应用包。 + """ + from app.application.history.registration import LocalRegistrationEntry + + source = tmp_path / "registration.mkv" + source.write_bytes(b"film") + runtime = runtime_composition.compose_runtime(_inputs()).runtime + entry = runtime.history.registration_local_entry(str(source)) + assert isinstance(entry, LocalRegistrationEntry) + assert entry.name == source.name + assert entry.is_file + assert not entry.is_symlink + assert entry.size == 4 + with runtime.history.registration_scope() as first: + with runtime.history.registration_scope() as second: + assert first.repository is not second.repository + assert first.unit_of_work is not second.unit_of_work + + def test_runtime_composition_publishes_and_resets_exact_projection(monkeypatch) -> None: """ 运行时 owner 应发布并按对称顺序撤销其拥有的全部投影。 diff --git a/tests/test_subscription_identity_review_migration.py b/tests/test_subscription_identity_review_migration.py index 7c00445ef..b27d32003 100644 --- a/tests/test_subscription_identity_review_migration.py +++ b/tests/test_subscription_identity_review_migration.py @@ -87,7 +87,8 @@ def test_identity_upgrade_chain_preserves_plugin_installation_snapshot(monkeypat config = Config() config.set_main_option("script_location", "database") script = ScriptDirectory.from_config(config) - assert script.get_heads() == ["97ad7239c1e0"] + assert script.get_heads() == ["b849e613a72f"] + assert script.get_revision("b849e613a72f").down_revision == "97ad7239c1e0" assert script.get_revision("97ad7239c1e0").down_revision == "4f62a73e81c9" assert script.get_revision("4f62a73e81c9").down_revision == "e4a8c7b29016" assert script.get_revision("e4a8c7b29016").down_revision == "da73b6f20418"