From 44200d575a23b85f29bec73c1eff0944143ac958 Mon Sep 17 00:00:00 2001 From: InfinityPacer Date: Mon, 31 Aug 2026 06:03:40 +0800 Subject: [PATCH] refactor(history): isolate mutation commands --- app/application/history.py | 242 +-------------- app/application/history_mutation.py | 277 ++++++++++++++++++ app/schemas/exports.py | 4 +- docs/architecture-overview.md | 2 +- docs/architecture/optimization-checklist.md | 2 +- .../architecture/dependency-baseline.json | 11 +- .../startup-performance-baseline.json | 6 +- 7 files changed, 302 insertions(+), 242 deletions(-) create mode 100644 app/application/history_mutation.py diff --git a/app/application/history.py b/app/application/history.py index 626bca904..31ee879fa 100644 --- a/app/application/history.py +++ b/app/application/history.py @@ -1,10 +1,18 @@ from __future__ import annotations from dataclasses import asdict, dataclass -from pathlib import Path from typing import Any, Callable, Dict, NoReturn, Optional, Protocol, Union from app.application.configuration import TransferRetryConfig, get_transfer_retry_config +from app.application.history_mutation import ( + DownloadFileMutationRepository as DownloadFileMutationRepository, + DownloadHistoryMutationCommand as DownloadHistoryMutationCommand, + DownloadHistoryMutationRepository as DownloadHistoryMutationRepository, + HistoryMutationResult as HistoryMutationResult, + HistoryUnitOfWork as HistoryUnitOfWork, + TransferHistoryMutationCommand as TransferHistoryMutationCommand, + TransferHistoryMutationRepository as TransferHistoryMutationRepository, +) from app.domain.context import MediaInfo, MusicInfo from app.domain.meta.metabase import MetaBase from app.domain.meta.metamusic import MetaMusic @@ -16,8 +24,6 @@ from app.schemas.file import FileItem from app.schemas.history import ( DownloadHistory, TransferHistory, - TransferHistoryDeleteResult, - TransferHistoryDeleteStep, TransferHistoryPage, ) from app.schemas.media import resolve_media_identity @@ -37,14 +43,6 @@ FAILED_RETRY_TTL = 24 * 3600 _failed_retry_counts = TTLCache(region="transfer_failed_retry", maxsize=5000, ttl=FAILED_RETRY_TTL) -@dataclass(frozen=True, slots=True) -class HistoryMutationResult: - """描述历史记录维护操作是否成功及兼容提示。""" - - success: bool - message: str = "" - - class _FrozenJsonDict(dict[str, JsonData]): """保留 JSON 字典读取与序列化行为，并拒绝常规原地修改。""" @@ -759,228 +757,6 @@ class HistoryQueryService: return result.replace("*", "%").replace("?", "_") -class DownloadHistoryMutationRepository(Protocol): - """下载历史删除用例需要的最小持久化端口。""" - - def stage_delete_history(self, history_id: int) -> None: - """暂存下载历史删除。""" - ... - - -class TransferHistoryMutationRepository(Protocol): - """整理历史删除与清理用例需要的最小持久化端口。""" - - def get(self, history_id: int) -> Optional[TransferHistorySnapshot]: - """读取整理历史。""" - ... - - def stage_delete(self, history_id: int) -> None: - """暂存整理历史删除。""" - ... - - def stage_truncate(self) -> None: - """暂存全部整理历史删除。""" - ... - - -class DownloadFileMutationRepository(Protocol): - """整理历史删除时关联下载文件状态更新端口。""" - - def stage_delete_file_by_fullpath(self, fullpath: str) -> None: - """暂存下载文件删除状态。""" - ... - - -class HistoryUnitOfWork(Protocol): - """同步历史维护用例使用的事务端口。""" - - def commit(self) -> None: - """提交当前事务。""" - ... - - def rollback(self) -> None: - """回滚当前事务。""" - ... - - -class DownloadHistoryMutationCommand: - """统一提交下载历史删除，避免 API 直接持有数据库事务。""" - - def __init__( - self, - *, - repository: DownloadHistoryMutationRepository, - unit_of_work: HistoryUnitOfWork, - ) -> None: - """保存下载历史持久化和事务端口。""" - self._repository = repository - self._unit_of_work = unit_of_work - - def delete(self, history_id: int) -> HistoryMutationResult: - """暂存并提交单条下载历史删除。""" - self._repository.stage_delete_history(history_id) - self._commit() - return HistoryMutationResult(True) - - def _commit(self) -> None: - """提交事务，失败时回滚。""" - try: - self._unit_of_work.commit() - except Exception: - self._unit_of_work.rollback() - raise - - -class TransferHistoryMutationCommand: - """协调整理历史、关联文件状态和外部存储删除。""" - - def __init__( - self, - *, - repository: TransferHistoryMutationRepository, - download_repository: DownloadFileMutationRepository, - unit_of_work: HistoryUnitOfWork, - file_item_factory: Callable[[dict[str, JsonData]], Any], - delete_media_file: Callable[[Any], bool], - publish_download_file_deleted: Callable[[dict[str, Any]], None], - clear_failures: Callable[[Optional[str], Optional[str]], None], - file_exists: Optional[Callable[[Any], Optional[bool]]] = None, - ) -> None: - """保存历史事务、存储删除、事件和失败状态清理端口。""" - self._repository = repository - self._download_repository = download_repository - self._unit_of_work = unit_of_work - self._file_item_factory = file_item_factory - self._file_exists = file_exists or (lambda _fileitem: True) - self._delete_media_file = delete_media_file - self._publish_download_file_deleted = publish_download_file_deleted - self._clear_failures = clear_failures - - def delete( - self, - history_id: int, - *, - delete_source: bool = False, - delete_destination: bool = False, - ) -> TransferHistoryDeleteResult: - """删除整理记录并返回源、目标文件及历史记录的分项结果。""" - history = self._repository.get(history_id) - if not history: - return TransferHistoryDeleteResult( - source=TransferHistoryDeleteStep(status="not_requested"), - destination=TransferHistoryDeleteStep(status="not_requested"), - history="not_found", - message="记录不存在", - ) - if history.transfer_task_id: - return TransferHistoryDeleteResult( - source=TransferHistoryDeleteStep(status="not_requested"), - destination=TransferHistoryDeleteStep(status="not_requested"), - history="retained", - message="持久整理失败记录不可删除，请使用重试或人工复核入口", - ) - - destination_result = self._delete_file( - history.dest_fileitem if delete_destination else None, - requested=delete_destination, - label="目标文件", - ) - source_result = self._delete_file( - history.src_fileitem if delete_source else None, - requested=delete_source, - label="源文件", - ) - source_completed = source_result.status in {"deleted", "already_missing"} - all_requested_completed = all( - step.status in {"not_requested", "deleted", "already_missing"} - for step in (source_result, destination_result) - ) - - if source_completed: - source_payload = self._file_item_payload(history.src_fileitem) - if source_payload is not None: - source = self._file_item_factory(source_payload) - self._download_repository.stage_delete_file_by_fullpath( - Path(source.path).as_posix() - ) - - if all_requested_completed: - self._repository.stage_delete(history_id) - - if source_completed or all_requested_completed: - self._commit() - if source_completed: - self._publish_download_file_deleted({ - "src": history.src, - "hash": history.download_hash, - }) - if all_requested_completed: - self._clear_failures(history.src, history.src_storage) - - return TransferHistoryDeleteResult( - source=source_result, - destination=destination_result, - history="deleted" if all_requested_completed else "retained", - message=( - "已删除整理记录" - if all_requested_completed - else next( - ( - step.message - for step in (source_result, destination_result) - if step.status == "failed" and step.message - ), - "整理记录已保留，部分文件处理失败", - ) - ), - ) - - def _delete_file( - self, - value: Optional[JsonData], - *, - requested: bool, - label: str, - ) -> TransferHistoryDeleteStep: - """删除一个文件目标；原目标已不存在时按完成处理，失败不抛出以便继续汇总其他步骤。""" - if not requested: - return TransferHistoryDeleteStep(status="not_requested") - payload = self._file_item_payload(value) - if payload is None: - return TransferHistoryDeleteStep(status="failed", message=f"{label}历史数据无效") - try: - fileitem = self._file_item_factory(payload) - if self._file_exists(fileitem) is False: - return TransferHistoryDeleteStep(status="already_missing") - if not self._delete_media_file(fileitem): - return TransferHistoryDeleteStep( - status="failed", - message=f"{fileitem.path} 删除失败", - ) - except Exception: - return TransferHistoryDeleteStep(status="failed", message=f"{label}删除失败") - return TransferHistoryDeleteStep(status="deleted") - - @staticmethod - def _file_item_payload(value: JsonData) -> Optional[dict[str, JsonData]]: - """仅接受可安全构造文件项的历史 JSON 对象。""" - return value if isinstance(value, dict) else None - - def truncate(self) -> HistoryMutationResult: - """在单一事务中清空旧历史，并保留当前失败任务记录。""" - self._repository.stage_truncate() - self._commit() - return HistoryMutationResult(True, "已清空旧整理记录，失败任务记录已保留") - - def _commit(self) -> None: - """提交历史事务，失败时回滚且不发布事件或清缓存。""" - try: - self._unit_of_work.commit() - except Exception: - self._unit_of_work.rollback() - raise - - class HistoryGateAction: """ 整理历史查重闸的判定结果。 diff --git a/app/application/history_mutation.py b/app/application/history_mutation.py new file mode 100644 index 000000000..7c80adecf --- /dev/null +++ b/app/application/history_mutation.py @@ -0,0 +1,277 @@ +"""历史删除命令及其事务端口。""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Optional, Protocol + +from app.schemas.common import JsonData +from app.schemas.history import TransferHistoryDeleteResult, TransferHistoryDeleteStep + + +@dataclass(frozen=True, slots=True) +class HistoryMutationResult: + """描述历史记录维护操作是否成功及兼容提示。""" + + success: bool + message: str = "" + + +class DownloadHistoryMutationRepository(Protocol): + """下载历史删除用例需要的最小持久化端口。""" + + def stage_delete_history(self, history_id: int) -> None: + """暂存下载历史删除。""" + ... + + +class TransferHistoryMutationRecord(Protocol): + """整理历史删除用例读取的稳定字段投影。""" + + @property + def transfer_task_id(self) -> Optional[str]: + """返回持久整理任务标识。""" + ... + + @property + def src(self) -> Optional[str]: + """返回源路径。""" + ... + + @property + def src_storage(self) -> Optional[str]: + """返回源存储标识。""" + ... + + @property + def src_fileitem(self) -> Optional[JsonData]: + """返回源文件项快照。""" + ... + + @property + def dest_fileitem(self) -> Optional[JsonData]: + """返回目标文件项快照。""" + ... + + @property + def download_hash(self) -> Optional[str]: + """返回下载任务 Hash。""" + ... + + +class TransferHistoryMutationRepository(Protocol): + """整理历史删除与清理用例需要的最小持久化端口。""" + + def get(self, history_id: int) -> Optional[TransferHistoryMutationRecord]: + """读取整理历史。""" + ... + + def stage_delete(self, history_id: int) -> None: + """暂存整理历史删除。""" + ... + + def stage_truncate(self) -> None: + """暂存全部整理历史删除。""" + ... + + +class DownloadFileMutationRepository(Protocol): + """整理历史删除时关联下载文件状态更新端口。""" + + def stage_delete_file_by_fullpath(self, fullpath: str) -> None: + """暂存下载文件删除状态。""" + ... + + +class HistoryUnitOfWork(Protocol): + """同步历史维护用例使用的事务端口。""" + + def commit(self) -> None: + """提交当前事务。""" + ... + + def rollback(self) -> None: + """回滚当前事务。""" + ... + + +class DownloadHistoryMutationCommand: + """统一提交下载历史删除，避免 API 直接持有数据库事务。""" + + def __init__( + self, + *, + repository: DownloadHistoryMutationRepository, + unit_of_work: HistoryUnitOfWork, + ) -> None: + """保存下载历史持久化和事务端口。""" + self._repository = repository + self._unit_of_work = unit_of_work + + def delete(self, history_id: int) -> HistoryMutationResult: + """暂存并提交单条下载历史删除。""" + self._repository.stage_delete_history(history_id) + self._commit() + return HistoryMutationResult(True) + + def _commit(self) -> None: + """提交事务，失败时回滚。""" + try: + self._unit_of_work.commit() + except Exception: + self._unit_of_work.rollback() + raise + + +class TransferHistoryMutationCommand: + """协调整理历史、关联文件状态和外部存储删除。""" + + def __init__( + self, + *, + repository: TransferHistoryMutationRepository, + download_repository: DownloadFileMutationRepository, + unit_of_work: HistoryUnitOfWork, + file_item_factory: Callable[[dict[str, JsonData]], Any], + delete_media_file: Callable[[Any], bool], + publish_download_file_deleted: Callable[[dict[str, Any]], None], + clear_failures: Callable[[Optional[str], Optional[str]], None], + file_exists: Optional[Callable[[Any], Optional[bool]]] = None, + ) -> None: + """保存历史事务、存储删除、事件和失败状态清理端口。""" + self._repository = repository + self._download_repository = download_repository + self._unit_of_work = unit_of_work + self._file_item_factory = file_item_factory + self._file_exists = file_exists or (lambda _fileitem: True) + self._delete_media_file = delete_media_file + self._publish_download_file_deleted = publish_download_file_deleted + self._clear_failures = clear_failures + + def delete( + self, + history_id: int, + *, + delete_source: bool = False, + delete_destination: bool = False, + ) -> TransferHistoryDeleteResult: + """删除整理记录并返回源、目标文件及历史记录的分项结果。""" + history = self._repository.get(history_id) + if not history: + return TransferHistoryDeleteResult( + source=TransferHistoryDeleteStep(status="not_requested"), + destination=TransferHistoryDeleteStep(status="not_requested"), + history="not_found", + message="记录不存在", + ) + if history.transfer_task_id: + return TransferHistoryDeleteResult( + source=TransferHistoryDeleteStep(status="not_requested"), + destination=TransferHistoryDeleteStep(status="not_requested"), + history="retained", + message="持久整理失败记录不可删除，请使用重试或人工复核入口", + ) + + destination_result = self._delete_file( + history.dest_fileitem if delete_destination else None, + requested=delete_destination, + label="目标文件", + ) + source_result = self._delete_file( + history.src_fileitem if delete_source else None, + requested=delete_source, + label="源文件", + ) + source_completed = source_result.status in {"deleted", "already_missing"} + all_requested_completed = all( + step.status in {"not_requested", "deleted", "already_missing"} + for step in (source_result, destination_result) + ) + + if source_completed: + source_payload = self._file_item_payload(history.src_fileitem) + if source_payload is not None: + source = self._file_item_factory(source_payload) + self._download_repository.stage_delete_file_by_fullpath( + Path(source.path).as_posix() + ) + + if all_requested_completed: + self._repository.stage_delete(history_id) + + if source_completed or all_requested_completed: + self._commit() + if source_completed: + self._publish_download_file_deleted({ + "src": history.src, + "hash": history.download_hash, + }) + if all_requested_completed: + self._clear_failures(history.src, history.src_storage) + + return TransferHistoryDeleteResult( + source=source_result, + destination=destination_result, + history="deleted" if all_requested_completed else "retained", + message=( + "已删除整理记录" + if all_requested_completed + else next( + ( + step.message + for step in (source_result, destination_result) + if step.status == "failed" and step.message + ), + "整理记录已保留，部分文件处理失败", + ) + ), + ) + + def _delete_file( + self, + value: Optional[JsonData], + *, + requested: bool, + label: str, + ) -> TransferHistoryDeleteStep: + """删除一个文件目标；原目标已不存在时按完成处理，失败不抛出以便继续汇总其他步骤。""" + if not requested: + return TransferHistoryDeleteStep(status="not_requested") + payload = self._file_item_payload(value) + if payload is None: + return TransferHistoryDeleteStep( + status="failed", + message=f"{label}历史数据无效", + ) + try: + fileitem = self._file_item_factory(payload) + if self._file_exists(fileitem) is False: + return TransferHistoryDeleteStep(status="already_missing") + if not self._delete_media_file(fileitem): + return TransferHistoryDeleteStep( + status="failed", + message=f"{fileitem.path} 删除失败", + ) + except Exception: + return TransferHistoryDeleteStep(status="failed", message=f"{label}删除失败") + return TransferHistoryDeleteStep(status="deleted") + + @staticmethod + def _file_item_payload(value: JsonData) -> Optional[dict[str, JsonData]]: + """仅接受可安全构造文件项的历史 JSON 对象。""" + return value if isinstance(value, dict) else None + + def truncate(self) -> HistoryMutationResult: + """在单一事务中清空旧历史，并保留当前失败任务记录。""" + self._repository.stage_truncate() + self._commit() + return HistoryMutationResult(True, "已清空旧整理记录，失败任务记录已保留") + + def _commit(self) -> None: + """提交历史事务，失败时回滚且不发布事件或清缓存。""" + try: + self._unit_of_work.commit() + except Exception: + self._unit_of_work.rollback() + raise diff --git a/app/schemas/exports.py b/app/schemas/exports.py index 4d8ea0699..2110a2a05 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -413,6 +413,8 @@ SCHEMA_EXPORTS = { 'TorrentReidentifyData': ('app.schemas.cache', 'TorrentReidentifyData'), 'TransferDirectoryConf': ('app.schemas.system', 'TransferDirectoryConf'), 'TransferHistory': ('app.schemas.history', 'TransferHistory'), + 'TransferHistoryDeleteResult': ('app.schemas.history', 'TransferHistoryDeleteResult'), + 'TransferHistoryDeleteStep': ('app.schemas.history', 'TransferHistoryDeleteStep'), 'TransferHistoryPage': ('app.schemas.history', 'TransferHistoryPage'), 'TransferInfo': ('app.schemas.transfer', 'TransferInfo'), 'TransferInterceptEventData': ('app.schemas.event', 'TransferInterceptEventData'), @@ -483,7 +485,7 @@ SCHEMA_CONFLICTS = { 'FilterRuleGroup': ['app.schemas.rule', 'app.schemas.system'], 'JsonData': ['app.schemas.agent', 'app.schemas.common', 'app.schemas.context', 'app.schemas.dashboard', 'app.schemas.event', 'app.schemas.history', 'app.schemas.mediaserver', 'app.schemas.message', 'app.schemas.mfa', 'app.schemas.music', 'app.schemas.plugin', 'app.schemas.search', 'app.schemas.openai', 'app.schemas.servcookie', 'app.schemas.site', 'app.schemas.transfer', 'app.schemas.user', 'app.schemas.workflow', 'app.schemas.mcp'], 'List': ['app.schemas.agent', 'app.schemas.context', 'app.schemas.event', 'app.schemas.history', 'app.schemas.mediaserver', 'app.schemas.message', 'app.schemas.monitoring', 'app.schemas.plugin', 'app.schemas.openai', 'app.schemas.subscribe', 'app.schemas.transfer', 'app.schemas.workflow'], - 'Literal': ['app.schemas.agent', 'app.schemas.music', 'app.schemas.plugin', 'app.schemas.search', 'app.schemas.servcookie', 'app.schemas.system', 'app.schemas.transfer', 'app.schemas.mcp'], + 'Literal': ['app.schemas.agent', 'app.schemas.history', 'app.schemas.music', 'app.schemas.plugin', 'app.schemas.search', 'app.schemas.servcookie', 'app.schemas.system', 'app.schemas.transfer', 'app.schemas.mcp'], 'LocaleHelper': ['app.schemas.dashboard', 'app.schemas.response'], 'MediaInfo': ['app.schemas.context', 'app.schemas.system', 'app.schemas.transfer', 'app.schemas.workflow'], 'MediaSource': ['app.schemas.cache', 'app.schemas.context', 'app.schemas.event', 'app.schemas.history', 'app.schemas.mediaserver', 'app.schemas.music', 'app.schemas.subscribe', 'app.schemas.transfer'], diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 74527dfb2..c620e70cf 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -755,7 +755,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 973 | -| 内部导入边 | 8,422 | +| 内部导入边 | 8,426 | | 非平凡 SCC | 1（精确 containment 的 TMDB 移植包环） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53（债务已清零，53 条精确 containment） | diff --git a/docs/architecture/optimization-checklist.md b/docs/architecture/optimization-checklist.md index 844345d16..d7b5a8280 100644 --- a/docs/architecture/optimization-checklist.md +++ b/docs/architecture/optimization-checklist.md @@ -93,7 +93,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 981 / 8,422 | `dependency-baseline.json` 当前快照 | +| 宿主 Python 模块 / 内部依赖边 | 982 / 8,426 | `dependency-baseline.json` 当前快照 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 86cfc9b08..d7bddcce6 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1089,8 +1089,8 @@ "runtime_only": true } }, - "edge_count": 8422, - "edge_sha256": "3971beb13490710d6054b8cedbc66da9c0920bb30107332440f8c788d27feebd", + "edge_count": 8426, + "edge_sha256": "d361e646f1b0afdd79c52a75291c27af74ff9976291a048c18f6679cf18aaaa7", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3796,6 +3796,7 @@ "app.application.formatting -> app.schemas.workflow", "app.application.history -> app.application", "app.application.history -> app.application.configuration", + "app.application.history -> app.application.history_mutation", "app.application.history -> app.domain", "app.application.history -> app.domain.context", "app.application.history -> app.domain.meta", @@ -3813,6 +3814,9 @@ "app.application.history -> app.schemas.media", "app.application.history -> app.schemas.transfer", "app.application.history -> app.schemas.types", + "app.application.history_mutation -> app.schemas", + "app.application.history_mutation -> app.schemas.common", + "app.application.history_mutation -> app.schemas.history", "app.application.image -> app.application", "app.application.image -> app.application.configuration", "app.application.image -> app.application.security", @@ -9515,7 +9519,7 @@ "app.workflow.actions.transfer_file -> app.workflow", "app.workflow.actions.transfer_file -> app.workflow.actions" ], - "module_count": 981, + "module_count": 982, "modules": [ "app", "app.adapters", @@ -9794,6 +9798,7 @@ "app.application.downloader", "app.application.formatting", "app.application.history", + "app.application.history_mutation", "app.application.image", "app.application.maintenance", "app.application.mediaserver", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index 9842026cc..d9cb92f8b 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": 498, + "loaded_app_module_count": 499, "max_ms": 1138.629, "median_ms": 1108.684, "min_ms": 1097.855, @@ -17,7 +17,7 @@ ] }, "app.factory": { - "loaded_app_module_count": 510, + "loaded_app_module_count": 511, "max_ms": 1118.665, "median_ms": 1117.279, "min_ms": 1111.113, @@ -28,7 +28,7 @@ ] }, "app.main": { - "loaded_app_module_count": 512, + "loaded_app_module_count": 513, "max_ms": 1172.187, "median_ms": 1161.652, "min_ms": 1155.282,