nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-02 13:28:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(history): isolate mutation commands

Browse Source
This commit is contained in:
InfinityPacer
2026-08-31 06:03:40 +08:00
parent 6c53086afe
commit 44200d575a
7 changed files with 302 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files

242
app/application/history.py
View File

@@ -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:
"""
整理历史查重闸的判定结果。

277
app/application/history_mutation.py Normal file
View File

@@ -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

4
app/schemas/exports.py
View File

@@ -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'],

2
docs/architecture-overview.md
View File

@@ -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） |

2
docs/architecture/optimization-checklist.md
View File

@@ -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 等基线均为零 |

11
tests/fixtures/architecture/dependency-baseline.json vendored
View File

@@ -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",

6
tests/fixtures/architecture/startup-performance-baseline.json vendored
View File

@@ -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,