From f3c1449c8b43fcd902ca3ec8c38bf4e2e62fdb30 Mon Sep 17 00:00:00 2001 From: Aqr-K <95741669+Aqr-K@users.noreply.github.com> Date: Fri, 23 Oct 2026 06:51:54 +0800 Subject: [PATCH] fix(storage): preserve transfer contents and cleanup identities (#7142) --- app/adapters/system/fsproxy.py | 2 +- app/adapters/system/fsworker.py | 6 + app/adapters/system/host.py | 5 +- app/api/endpoints/storage.py | 4 +- app/api/endpoints/transferhistory.py | 4 + app/application/audio.py | 3 - app/application/transfer/history.py | 1 + app/chain/scraping.py | 2 + app/modules/filemanager/module.py | 9 +- app/modules/filemanager/storages/alipan.py | 16 +- app/modules/filemanager/storages/alist.py | 26 +- app/modules/filemanager/storages/local.py | 11 +- app/modules/filemanager/transhandler.py | 161 +++++++---- app/schemas/file.py | 13 + docs/mcp-api.md | 12 + skills/organize-files/SKILL.md | 10 +- .../architecture/complexity-baseline.json | 2 +- tests/test_builtin_skill_boundaries.py | 2 +- .../test_local_transfer_review_regressions.py | 214 +++++++++++++++ tests/test_storage_review_regressions.py | 165 +++++++++++ tests/test_system_reflink.py | 41 ++- tests/test_transfer_execution_runner.py | 8 +- tests/test_transfer_review_regressions.py | 256 ++++++++++++++++++ tests/test_transfer_safe_overwrite.py | 12 +- 24 files changed, 874 insertions(+), 111 deletions(-) create mode 100644 tests/test_local_transfer_review_regressions.py create mode 100644 tests/test_storage_review_regressions.py create mode 100644 tests/test_transfer_review_regressions.py diff --git a/app/adapters/system/fsproxy.py b/app/adapters/system/fsproxy.py index 1480b1b76..868622557 100644 --- a/app/adapters/system/fsproxy.py +++ b/app/adapters/system/fsproxy.py @@ -82,7 +82,7 @@ class FileSystemProxy: 读取路径属性。 :param path: 目标路径 - :return: {"size", "mtime", "dev", "ino", "is_dir", "is_file"}；dev/ino 标识当前文件身份 + :return: 大小、秒及纳秒修改时间、纳秒状态时间、设备和 inode、文件类型；dev/ino 标识当前文件身份 """ return cast(dict[str, Any], self._call("stat", path=str(path))) diff --git a/app/adapters/system/fsworker.py b/app/adapters/system/fsworker.py index da85d1288..fec4b4108 100644 --- a/app/adapters/system/fsworker.py +++ b/app/adapters/system/fsworker.py @@ -56,6 +56,8 @@ class _StatResult(TypedDict): :ivar size: 文件大小，单位为字节 :ivar mtime: 文件修改时间的 Unix 秒级时间戳 + :ivar mtime_ns: 文件修改时间的 Unix 纳秒级时间戳 + :ivar ctime_ns: 文件状态变化时间的 Unix 纳秒级时间戳 :ivar dev: 文件所属设备号，与 ino 一起用于核对身份 :ivar ino: 文件系统索引节点号 :ivar is_dir: 查询时目标是否为目录 @@ -64,6 +66,8 @@ class _StatResult(TypedDict): size: int mtime: float + mtime_ns: int + ctime_ns: int dev: int ino: int is_dir: bool @@ -127,6 +131,8 @@ def _stat(payload: _Request, _emit: _Emit) -> _StatResult: 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": os.path.isdir(path), diff --git a/app/adapters/system/host.py b/app/adapters/system/host.py index 7decb3b08..8747591ac 100644 --- a/app/adapters/system/host.py +++ b/app/adapters/system/host.py @@ -572,11 +572,8 @@ class SystemUtils: :param dest: 目标文件路径 :return: (状态码, 错误信息)，0 表示成功 """ - tmp_path = dest.with_suffix(dest.suffix + ".mp") + tmp_path = dest.with_name(f".{dest.name}.{uuid.uuid4().hex}.mp") try: - # 检查临时目标路径是否已存在，如果存在则先清理 - if tmp_path.exists(): - tmp_path.unlink() # 调用私有辅助方法执行底层克隆 code, msg = SystemUtils._reflink_copy(src, tmp_path) if code != 0: diff --git a/app/api/endpoints/storage.py b/app/api/endpoints/storage.py index 8138ca55b..873915070 100644 --- a/app/api/endpoints/storage.py +++ b/app/api/endpoints/storage.py @@ -368,7 +368,7 @@ def rename( runtime_config.media_extensions + runtime_config.subtitle_extensions + runtime_config.audio_extensions ) # 递归修改目录内文件（智能识别命名） - sub_files: List[_SchemaFileItem] = StorageChain().list_files(fileitem) or [] + sub_files: List[_SchemaFileItem] = StorageChain().list_files(fileitem, recursion=True) or [] if sub_files: # 开始进度 progress = ProgressHelper(ProgressKey.BatchRename) @@ -384,7 +384,7 @@ def rename( continue if f".{sub_file.extension.lower()}" not in media_exts: continue - sub_path = Path(f"{fileitem.path}{sub_file.name}") + sub_path = Path(sub_file.path or "") context = MediaChain().recognize_by_path( sub_path, obtain_images=False, diff --git a/app/api/endpoints/transferhistory.py b/app/api/endpoints/transferhistory.py index 5a23fa56b..90453ca97 100644 --- a/app/api/endpoints/transferhistory.py +++ b/app/api/endpoints/transferhistory.py @@ -66,6 +66,10 @@ def restore_manual_transfer_history_batch( _common_history_value(histories, "music_type") or transer_item.music_type ) + transer_item.episode_group = ( + _common_history_value(histories, "episode_group") + or transer_item.episode_group + ) downloader = _common_history_value(histories, "downloader") download_hash = _common_history_value(histories, "download_hash") logger.info("手动整理历史批次还原 %s 个源文件", len(src_fileitems)) diff --git a/app/application/audio.py b/app/application/audio.py index c4f26466f..cdbe348a3 100644 --- a/app/application/audio.py +++ b/app/application/audio.py @@ -153,10 +153,7 @@ def _audio_write_target(path: Path) -> Iterator[Path]: shutil.copy2(path, temporary) # NAS ACL 挂载可将可读源文件映射为 000；只给独立副本补齐读写权限，源权限保持不变。 temporary.chmod(stat.st_mode | S_IRUSR | S_IWUSR) - initial_copy = _audio_signature(temporary) yield temporary - if _audio_signature(temporary) == initial_copy: - return if path.lstat().st_ino != link_inode or _audio_signature(path) != signature: raise OSError("标签写入期间目标文件已变化，保留当前目标并取消替换") temporary.replace(path) diff --git a/app/application/transfer/history.py b/app/application/transfer/history.py index 634a6b6b1..ac834a3d1 100644 --- a/app/application/transfer/history.py +++ b/app/application/transfer/history.py @@ -107,6 +107,7 @@ def success_fields( image=mediainfo.get_poster_image(), downloader=downloader, download_hash=download_hash, + episode_group=mediainfo.episode_group, status=True, retry_count=0, auto_paused=False, diff --git a/app/chain/scraping.py b/app/chain/scraping.py index 79d70023d..9a268621d 100644 --- a/app/chain/scraping.py +++ b/app/chain/scraping.py @@ -2454,6 +2454,8 @@ class ScrapingChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): logger.warn(f"{filepath.name} 无法识别文件媒体信息！") return + file_mediainfo.scrape_source = mediainfo.scrape_source + self._scrape_nfo_generic( current_fileitem=fileitem, meta=file_meta, diff --git a/app/modules/filemanager/module.py b/app/modules/filemanager/module.py index fde416891..1ec146845 100644 --- a/app/modules/filemanager/module.py +++ b/app/modules/filemanager/module.py @@ -1044,13 +1044,14 @@ class FileManagerModule(_ModuleBase): for fileitem in fileitems: file_meta = MetaInfo(fileitem.basename or "") season_index = file_meta.begin_season if file_meta.begin_season is not None else 1 - episode_index = file_meta.begin_episode - if not episode_index: + episode_indexes = file_meta.episode_list + if not episode_indexes: continue if season_index not in seasons: seasons[season_index] = [] - if episode_index not in seasons[season_index]: - seasons[season_index].append(episode_index) + for episode_index in episode_indexes: + if episode_index not in seasons[season_index]: + seasons[season_index].append(episode_index) logger.info(f"{mediainfo.title_year} 在本地文件系统中找到了这些季集：{seasons}") return ExistMediaInfo(type=MediaType.TV, seasons=seasons) return None diff --git a/app/modules/filemanager/storages/alipan.py b/app/modules/filemanager/storages/alipan.py index 57b1f8a35..b2410dc50 100644 --- a/app/modules/filemanager/storages/alipan.py +++ b/app/modules/filemanager/storages/alipan.py @@ -135,8 +135,8 @@ class AliPan(StorageBase, metaclass=WeakSingleton): refreshed_tokens = self.__refresh_access_token(refresh_token) if not refreshed_tokens: return None - tokens = refreshed_tokens - self.set_config({"refresh_time": int(time.time()), **tokens}) + tokens = {**tokens, **refreshed_tokens} + self.set_config({**tokens, "refresh_time": int(time.time())}) access_token = tokens.get("access_token") if access_token: self._request_utils.update_headers( @@ -454,7 +454,7 @@ class AliPan(StorageBase, metaclass=WeakSingleton): """ 分页列出目录直接子项，文件输入则刷新自身详情。 - 每页最多读取 100 项，少于 100 项时停止；无响应抛出 FileNotFoundError。 + 每页最多读取 100 项，没有下一页标记时停止；无响应抛出 FileNotFoundError。 业务错误未统一抛出，可能返回已获取的部分结果。 :param fileitem: 待查询项；非根目录必须带有文件标识和空间标识 @@ -497,7 +497,7 @@ class AliPan(StorageBase, metaclass=WeakSingleton): current_items = resp.get("items") or [] for item in current_items: items.append(self.__get_fileitem(item, parent=str(fileitem.path))) - if len(current_items) < 100: + if not next_marker: break return items @@ -977,18 +977,16 @@ class AliPan(StorageBase, metaclass=WeakSingleton): """ 请求将文件或目录移入云盘回收站。 - 当前实现不检查接口返回的业务错误或空响应，True 不能证明目标已删除。 - :param fileitem: 携带空间和文件标识的待删除项 - :return: 未抛出 HTTP 请求异常时返回 True，否则返回 False + :return: 收到无业务错误的 JSON 响应时返回 True；拒绝、无响应或请求异常返回 False """ try: - self._request_api( + response = self._request_api( "POST", "/adrive/v1.0/openFile/recyclebin/trash", json={"drive_id": fileitem.drive_id, "file_id": fileitem.fileid}, ) - return True + return isinstance(response, dict) and not response.get("code") except HttpRequestError: return False diff --git a/app/modules/filemanager/storages/alist.py b/app/modules/filemanager/storages/alist.py index 54625e4d7..163bf4cfd 100644 --- a/app/modules/filemanager/storages/alist.py +++ b/app/modules/filemanager/storages/alist.py @@ -564,13 +564,13 @@ class Alist(StorageBase, metaclass=WeakSingleton): """ 取得下载地址后流式写入本地文件。 - 当前捕获传输异常后仍返回目标路径，不能据此断定文件完整甚至已经创建； - 取消和失败都可能留下部分文件，调用方需另行核验。 + 文件下载不发送 OpenList API 令牌，使用服务端返回的直链或签名地址。 + 只有流正常结束且已知大小匹配才返回成功；中断、取消或截断均删除未完成文件。 - :param fileitem: 源远端文件；name 直接用于拼接保存路径 + :param fileitem: 源远端文件，提供文件名称和可选的预期大小 :param path: 本地保存目录；为空时使用 TEMP_PATH :param password: 详情查询使用的可选访问密码 - :return: 本地目标路径；详情查询失败或取消返回 None + :return: 完整本地目标路径；详情查询、传输、大小校验失败或取消返回 None """ if not fileitem.path: return None @@ -613,22 +613,32 @@ class Alist(StorageBase, metaclass=WeakSingleton): if local_path is None or not download_url: return None - request_utils = RequestUtils(headers=self.__get_header_with_token()) + # 下载使用 raw_url 或带 sign 的 /d 地址，API 令牌不属于文件服务及其重定向目标。 + request_utils = RequestUtils() try: with request_utils.get_stream(download_url, raise_exception=True) as r: if r is None: return None r.raise_for_status() + downloaded_size = 0 with open(local_path, "wb") as f: for chunk in r.iter_content(chunk_size=8192): if runtime_stop_state.consume_transfer_stop(fileitem.path): logger.info(f"【OpenList】{fileitem.path} 下载已取消！") - return None + raise IOError("OpenList 下载已取消") f.write(chunk) + downloaded_size += len(chunk) + expected_size = fileitem.size + if expected_size is None: + expected_size = result["data"].get("size") + if expected_size is None and r.headers.get("Content-Length"): + expected_size = int(r.headers["Content-Length"]) + if expected_size is not None and downloaded_size != expected_size: + raise IOError(f"下载大小不一致：预期 {expected_size}，实际 {downloaded_size}") except Exception as e: logger.error(f"【OpenList】下载文件 {fileitem.path} 失败：{e}") - if local_path.exists(): - return local_path + local_path.unlink(missing_ok=True) + return None return local_path diff --git a/app/modules/filemanager/storages/local.py b/app/modules/filemanager/storages/local.py index 250514c53..ab9dad9b8 100644 --- a/app/modules/filemanager/storages/local.py +++ b/app/modules/filemanager/storages/local.py @@ -5,6 +5,7 @@ import os import shutil import time +import uuid from pathlib import Path from typing import Any, List, Optional @@ -70,6 +71,8 @@ class LocalStorage(StorageBase): extension=path.suffix[1:], size=info["size"], modify_time=info["mtime"], + modify_time_ns=info.get("mtime_ns"), + change_time_ns=info.get("ctime_ns"), fileid=f"{info['dev']}:{info['ino']}" if info.get("ino") else None, ) @@ -293,14 +296,14 @@ class LocalStorage(StorageBase): @classmethod def _partial_path(cls, dest: Path) -> Path: """ - 生成同目录、带进程标识的中间写入路径。 + 生成同目录、每次调用独立的中间写入路径。 - 同目录便于完成后原子替换；进程标识隔离不同进程，但不保证同进程并发写同一目标互不干扰。 + 同目录便于完成后原子替换；随机标识隔离同进程和跨进程的同目标任务。 :param dest: 最终文件路径 :return: 隐藏的 .mp-partial 文件路径 """ - return dest.parent / f".{dest.name}.{os.getpid()}{cls.PARTIAL_SUFFIX}" + return dest.parent / f".{dest.name}.{uuid.uuid4().hex}{cls.PARTIAL_SUFFIX}" @classmethod def _cleanup_stale_partials(cls, directory: Path) -> None: @@ -348,7 +351,7 @@ class LocalStorage(StorageBase): return False src_stat = src.stat() os.utime(dest, ns=(src_stat.st_atime_ns, src_stat.st_mtime_ns)) - sentinel = dest.parent / f".mode-probe-{os.getpid()}" + sentinel = dest.parent / f".mode-probe-{uuid.uuid4().hex}" try: sentinel.write_bytes(b"") dest_mode = sentinel.stat().st_mode & 0o777 diff --git a/app/modules/filemanager/transhandler.py b/app/modules/filemanager/transhandler.py index 95fe1abce..21ed0246f 100644 --- a/app/modules/filemanager/transhandler.py +++ b/app/modules/filemanager/transhandler.py @@ -813,7 +813,9 @@ class TransHandler: 清理前比对冻结文件的大小、类型及可用身份；证据不足时保留文件。 本地链接的查询沿用 stat 语义，硬链接和软链接均比较实际内容的身份。 - 云盘修改时间可能在上传后变化或缺失，不参与校验；旧快照没有 fileid 时使用大小凭据。 + 本地必须核对修改时间，优先纳秒证据；同路径还核对状态时间，防止等长改写后还原 mtime。 + rename 合法改变状态时间，因此跨路径只比较修改时间。旧本地快照回退到秒级修改时间， + 缺少时间凭据时转人工复核。云盘修改时间可能在上传后变化或缺失，不参与校验。 :param expected: 冻结的文件身份，必须包含大小 :param current: 当前严格查询结果；不可见时视为冲突 @@ -833,6 +835,22 @@ class TransHandler: raise StorageQueryError(f"文件 {expected.path} 缺少 fileid 身份凭据，需人工复核") if expected.fileid != current.fileid: raise TransferExecutionConflictError(f"文件 {expected.path} 已被替换，禁止清理") + if expected.storage == "local": + expected_time: Optional[int | float] + current_time: Optional[int | float] + if expected.modify_time_ns is not None: + expected_time, current_time = expected.modify_time_ns, current.modify_time_ns + else: + expected_time, current_time = expected.modify_time, current.modify_time + if expected_time is None or current_time is None: + raise StorageQueryError(f"文件 {expected.path} 缺少修改时间身份凭据，需人工复核") + if expected_time != current_time: + raise TransferExecutionConflictError(f"文件 {expected.path} 的修改时间已变化，禁止清理") + if expected.path == current.path and expected.change_time_ns is not None: + if current.change_time_ns is None: + raise StorageQueryError(f"文件 {expected.path} 缺少状态时间身份凭据，需人工复核") + if expected.change_time_ns != current.change_time_ns: + raise TransferExecutionConflictError(f"文件 {expected.path} 的状态时间已变化，禁止清理") @staticmethod def __restore_move_target_identity(source: FileItem, target: FileItem) -> FileItem: @@ -1550,34 +1568,47 @@ class TransHandler: ), ) - @classmethod def __resolve_leaf_overwrite_with_step( - cls, + self, *, step_runner: Optional[TransferStepRunner], planned_item: TransferPlanItem, target_oper: StorageBase, transfer_type: str, - ) -> bool: + meta: MetaBase, + mediainfo: MediaInfo | MusicInfo, + overwrite_mode: Optional[str], + need_notify: bool, + ) -> tuple[bool, Optional[TransferInfo]]: """ - 冻结目录叶子目标的存在性，恢复时沿用同一操作序列。 - - 此处只判断叶子目标是否存在，不再次执行单文件覆盖策略。 + 按冻结覆盖策略裁决目录叶目标，保持既有叶子步骤的稳定身份。 :param step_runner: 步骤执行器；None 时直接查询 :param planned_item: 含冻结源快照和目标路径的叶子操作 :param target_oper: 负责严格查询的目标适配器 :param transfer_type: 整理方式，参与步骤身份 - :return: 目标已存在且需要替换时为 True + :param meta: 目录媒体元数据 + :param mediainfo: 已识别媒体信息 + :param overwrite_mode: 检查点冻结的覆盖策略 + :param need_notify: 拒绝覆盖时是否请求通知 + :return: 是否覆盖及可选的拒绝结果 """ def execute() -> TransferStepResult: """ - 严格查询当前叶目标，失败时禁止对旧文件执行任何操作。 + 目录叶与单文件共用覆盖合同，并冻结拒绝原因。 """ - exists = target_oper.get_item_strict(Path(planned_item.target_path)) is not None - return TransferStepResult(payload={"over_flag": exists}) + over_flag, _, failure = self.__resolve_overwrite( + fileitem=FileItem.model_validate(planned_item.source_fileitem), + meta=meta, mediainfo=mediainfo, target_oper=target_oper, + target_storage=planned_item.target_storage, target_file=Path(planned_item.target_path), + transfer_type=transfer_type, overwrite_mode=overwrite_mode, need_notify=need_notify, + ) + return TransferStepResult(payload={ + "over_flag": over_flag, "policy_checked": True, + "failure": failure.model_dump(mode="json") if failure else None, + }) - decision = cls.__run_persisted_step( + decision = self.__run_persisted_step( step_runner, phase="decision", kind="resolve_leaf_overwrite", payload={"source": planned_item.source_fileitem, "target_storage": planned_item.target_storage, @@ -1588,7 +1619,19 @@ class TransHandler: evidence=TransferStepResult(payload={"reason": "read-only leaf decision may be repeated"}), ), ) - return decision.payload.get("over_flag") is True + failure = decision.payload.get("failure") + if (decision.payload.get("over_flag") is True and overwrite_mode == "never" + and decision.payload.get("policy_checked") is not True): + # 旧版本只记录存在性，不能用旧回执授权本次禁止的破坏性覆盖。 + return False, TransferInfo( + success=False, overwrite_skipped=True, + message="旧目录覆盖裁决未校验不覆盖策略，保留源文件和目标，需人工复核", + fileitem=FileItem.model_validate(planned_item.source_fileitem), + transfer_type=transfer_type, need_notify=need_notify, + ) + return decision.payload.get("over_flag") is True, ( + TransferInfo.model_validate(failure) if isinstance(failure, dict) else None + ) def __resolve_overwrite_with_step( self, @@ -1715,6 +1758,53 @@ class TransHandler: ) return bool(result.payload.get("allowed")), str(result.payload.get("reason") or "") + def __execute_directory_leaf( + self, *, planned_item: TransferPlanItem, checkpoint: TransferPlanCheckpoint, + meta: MetaBase, mediainfo: MediaInfo | MusicInfo, + source_oper: StorageBase, target_oper: StorageBase, + step_runner: Optional[TransferStepRunner], + ) -> tuple[Optional[FileItem], Optional[TransferInfo]]: + """ + 校验目录叶操作并消费覆盖裁决，拒绝时不执行文件副作用。 + + :param planned_item: 冻结的目录叶整理操作 + :param checkpoint: 目录计划及其覆盖、通知策略 + :param meta: 目录媒体元数据 + :param mediainfo: 目录媒体识别结果 + :param source_oper: 源存储适配器 + :param target_oper: 目标存储适配器 + :param step_runner: 持久步骤执行器；为空时同步执行 + :return: 新目标及可选失败结果；拒绝覆盖保留 overwrite_skipped 语义 + """ + transfer_type = checkpoint.resolved_transfer_type + if planned_item.action != "transfer" or planned_item.target_storage != checkpoint.target_storage: + return None, TransferInfo( + success=False, message="整理计划包含不支持的操作或目标存储", + fileitem=FileItem.model_validate(checkpoint.planning_input.source_fileitem), + transfer_type=transfer_type, need_notify=checkpoint.need_notify, + ) + over_flag, failure = self.__resolve_leaf_overwrite_with_step( + step_runner=step_runner, planned_item=planned_item, + target_oper=target_oper, transfer_type=transfer_type, + meta=meta, mediainfo=mediainfo, overwrite_mode=checkpoint.overwrite_mode, + need_notify=checkpoint.need_notify, + ) + if failure: + return None, failure + execute_transfer = self.__replace_target_with_steps if over_flag else self.__execute_transfer_with_steps + new_item, error = execute_transfer( + step_runner=step_runner, source_fileitem=planned_item.source_fileitem, + target_storage=planned_item.target_storage, source_oper=source_oper, target_oper=target_oper, + target_file=Path(planned_item.target_path), transfer_type=transfer_type, + ) + if not new_item: + return None, TransferInfo( + success=False, message=error, + fileitem=FileItem.model_validate(checkpoint.planning_input.source_fileitem), + transfer_type=transfer_type, need_notify=checkpoint.need_notify, + ) + return new_item, None + def execute_transfer_plan( self, checkpoint: TransferPlanCheckpoint, @@ -1842,47 +1932,16 @@ class TransHandler: need_notify=checkpoint.need_notify, ) for planned_item in checkpoint.items: - if ( - planned_item.action != "transfer" - or planned_item.target_storage != target_storage - ): - return TransferInfo( - success=False, - message="整理计划包含不支持的操作或目标存储", - fileitem=fileitem, - transfer_type=transfer_type, - need_notify=checkpoint.need_notify, - ) - source_item = FileItem.model_validate(planned_item.source_fileitem) - over_flag = self.__resolve_leaf_overwrite_with_step( - step_runner=step_runner, planned_item=planned_item, - target_oper=target_oper, transfer_type=transfer_type, + new_item, failure = self.__execute_directory_leaf( + planned_item=planned_item, checkpoint=checkpoint, meta=meta, mediainfo=mediainfo, + source_oper=source_oper, target_oper=target_oper, step_runner=step_runner, ) - execute_transfer = ( - self.__replace_target_with_steps if over_flag - else self.__execute_transfer_with_steps - ) - new_item, error = execute_transfer( - step_runner=step_runner, - source_fileitem=planned_item.source_fileitem, - target_storage=planned_item.target_storage, - source_oper=source_oper, - target_oper=target_oper, - target_file=Path(planned_item.target_path), - transfer_type=transfer_type, - ) - if not new_item: - return TransferInfo( - success=False, - message=error, - fileitem=fileitem, - transfer_type=transfer_type, - need_notify=checkpoint.need_notify, - ) + if failure: + return failure self.__update_result( result=result, - file_list=[source_item.path], - file_list_new=[new_item.path], + file_list=[planned_item.source_fileitem.get("path")], + file_list_new=[cast(FileItem, new_item).path], ) self.__cleanup_with_step( step_runner=step_runner, diff --git a/app/schemas/file.py b/app/schemas/file.py index 703edd101..a6c44eb50 100644 --- a/app/schemas/file.py +++ b/app/schemas/file.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import Optional from pydantic import BaseModel, Field +from pydantic import field_serializer as _field_serializer from app.schemas.types import StorageSchema @@ -62,6 +63,8 @@ class FileItem(FileURI): extension: Optional[str] = Field(default=None, description="文件后缀") size: Optional[int] = Field(default=None, description="文件大小，单位为字节；目录或来源未提供时可为 None") modify_time: Optional[float] = Field(default=None, description="文件修改时间的 Unix 时间戳，单位为秒，未提供时为 None") + modify_time_ns: Optional[int] = Field(default=None, description="本地修改时间的 Unix 纳秒值，JSON 使用十进制字符串避免精度丢失；旧快照或远端未提供时为 None") + change_time_ns: Optional[int] = Field(default=None, description="本地状态变化时间的 Unix 纳秒值，JSON 使用十进制字符串；用于同路径清理核验，未提供时为 None") children: Optional[list["FileItem"]] = Field(default_factory=list, description="子节点") fileid: Optional[str] = Field(default=None, description="存储服务分配的文件标识") parent_fileid: Optional[str] = Field(default=None, description="父ID") @@ -70,6 +73,16 @@ class FileItem(FileURI): drive_id: Optional[str] = Field(default=None, description="云盘驱动器标识") url: Optional[str] = Field(default=None, description="访问地址") + @_field_serializer("modify_time_ns", "change_time_ns", when_used="json") + def serialize_local_timestamp(self, value: Optional[int]) -> Optional[str]: + """ + 使用十进制字符串传输纳秒时间，避免浏览器数字精度改变恢复凭据。 + + :param value: Python 中的纳秒整数；未提供时为 None + :return: 可无损回传的十进制字符串，未提供时保持 None + """ + return str(value) if value is not None else None + class StorageUsage(BaseModel): """ diff --git a/docs/mcp-api.md b/docs/mcp-api.md index c47e10aec..3acba4635 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -26,6 +26,18 @@ 升级前历史没有来源标记，界面显示“未记录”，不会按当前订阅设置追溯推断。 普通订阅的首次提交与洗版订阅补缺在活动中都标记为首次订阅。 +### 文件整理与恢复 + +整理成功历史保存 `episode_group`；批量复用历史时仅恢复所有记录共同的剧集组， +不同剧集组应分别提交。目录整理的叶子文件同样遵守冻结的覆盖策略， +`overwrite_skipped` 表示策略拒绝，不能当作文件已导入。 + +`FileItem` 的可选 `modify_time_ns` 和 `change_time_ns` 保存本地纳秒版本证据， +JSON 使用十进制字符串避免浏览器数字精度丢失，客户端应原样保留查询返回值，不自行生成。 +恢复清理发现本地源或目标版本变化时会保留副本并转人工复核； +云盘时间戳变化仍按大小与稳定文件标识核验。OpenList 中断或大小不符的下载不会作为移动成功结果删除源文件。 +`storage.rename` 的 `recursive=true` 会识别并重命名所有子目录内媒体，随后重命名所选目录。 + ## 1. 连接与认证 | 配置项 | 值与含义 | diff --git a/skills/organize-files/SKILL.md b/skills/organize-files/SKILL.md index f9bb5fbd9..7cbf5428c 100644 --- a/skills/organize-files/SKILL.md +++ b/skills/organize-files/SKILL.md @@ -1,6 +1,6 @@ --- name: organize-files -version: 5 +version: 6 description: >- Use this skill when the user asks MoviePilot to identify and organize a local or downloaded video/music file, season folder, recording, album directory, or @@ -64,3 +64,11 @@ recognition, transfer, and history tools are not available. Stop and report instead of transferring when the source is missing, directory configuration is absent, identity remains ambiguous, an album appears mixed or incomplete, or the requested target would overwrite unrelated media. + +## 整理恢复与历史上下文 + +保留 `storage.list` 返回的完整文件项，包括可选的 `modify_time_ns` 和 +`change_time_ns`；不要自行生成本地版本凭据。恢复发现源或目标版本变化时， +应报告人工复核状态并保留现场，不通过重新提交整个批次绕过核验。 +复用历史整理电视剧时保留 `episode_group`；不同剧集组分批处理。 +目录整理的 `overwrite_skipped` 同样表示未导入，不能报告为成功。 diff --git a/tests/fixtures/architecture/complexity-baseline.json b/tests/fixtures/architecture/complexity-baseline.json index 753896e14..3429993c1 100644 --- a/tests/fixtures/architecture/complexity-baseline.json +++ b/tests/fixtures/architecture/complexity-baseline.json @@ -93,7 +93,7 @@ "app/modules/filemanager/storages/u115.py:U115Pan.upload": 33, "app/modules/filemanager/transhandler.py:TransHandler.__resolve_overwrite": 16, "app/modules/filemanager/transhandler.py:TransHandler.__transfer_command": 19, - "app/modules/filemanager/transhandler.py:TransHandler.execute_transfer_plan": 20, + "app/modules/filemanager/transhandler.py:TransHandler.execute_transfer_plan": 19, "app/modules/filter/__init__.py:FilterModule.__match_rule": 24, "app/modules/imdb/__init__.py:ImdbModule._to_media_info": 17, "app/modules/indexer/spider/__init__.py:SiteSpider.__get_search_url": 24, diff --git a/tests/test_builtin_skill_boundaries.py b/tests/test_builtin_skill_boundaries.py index a11455830..040a190fb 100644 --- a/tests/test_builtin_skill_boundaries.py +++ b/tests/test_builtin_skill_boundaries.py @@ -132,7 +132,7 @@ def test_modified_builtin_skills_have_incremented_versions() -> None: "anysearch": "3", "moviepilot-api": "59", "moviepilot-update": "7", - "organize-files": "5", + "organize-files": "6", "transfer-failed-retry": "6", "generate-identifiers": "4", "create-moviepilot-plugin": "7", diff --git a/tests/test_local_transfer_review_regressions.py b/tests/test_local_transfer_review_regressions.py new file mode 100644 index 000000000..dbe51e4ad --- /dev/null +++ b/tests/test_local_transfer_review_regressions.py @@ -0,0 +1,214 @@ +""" +验证本地并发暂存、音频链接写入和递归重命名的审查回归。 +""" + +import errno +import os +import shutil +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from threading import Barrier, Event +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from app.adapters.system.fsproxy import fsproxy +from app.adapters.system.host import SystemUtils +from app.api.endpoints import storage as storage_endpoint +from app.application import audio +from app.modules.filemanager.storages.local import LocalStorage +from app.schemas.file import FileItem + + +@pytest.mark.parametrize("operation", ["copy", "move"]) +def test_concurrent_transfers_to_same_target_use_separate_partials(tmp_path, monkeypatch, operation): + """ + 并发同目标复制不能发布另一任务的暂存内容，跨盘移动也不能丢失源内容。 + """ + first, second, target = tmp_path / "a.bin", tmp_path / "b.bin", tmp_path / "target.bin" + first.write_bytes(b"A" * 64) + second.write_bytes(b"B" * 64) + copied_first, copied_second, published_first = Event(), Event(), Event() + partials = [] + original_copy = fsproxy.copy + storage = object.__new__(LocalStorage) + monkeypatch.setattr(storage, "_try_reflink_copy", lambda *_args: False) + monkeypatch.setattr(storage, "_LocalStorage__should_show_progress", lambda *_args: False) + + def copy(source, destination, **kwargs): + """ + 固定两份内容完成复制和提升的先后顺序。 + """ + partials.append(destination) + if source == first: + result = original_copy(source, destination, **kwargs) + copied_first.set() + assert copied_second.wait(10) + return result + assert copied_first.wait(10) + result = original_copy(source, destination, **kwargs) + copied_second.set() + assert published_first.wait(10) + return result + + monkeypatch.setattr(fsproxy, "copy", copy) + if operation == "move": + replace = os.replace + + def cross_device_replace(source, destination): + """ + 只让原始移动跨设备，暂存提升仍使用真实原子替换。 + """ + if source in (first, second): + raise OSError(errno.EXDEV, "cross-device") + return replace(source, destination) + + monkeypatch.setattr(os, "replace", cross_device_replace) + + def transfer(source): + """ + 返回每个任务确认成功时实际发布的内容。 + """ + try: + result = getattr(storage, operation)(FileItem(path=str(source), type="file"), tmp_path, target.name) + return result, target.read_bytes() + finally: + if source == first: + published_first.set() + + with ThreadPoolExecutor(max_workers=2) as workers: + first_result = workers.submit(transfer, first) + second_result = workers.submit(transfer, second) + assert first_result.result(20) == (True, b"A" * 64) + assert second_result.result(20) == (True, b"B" * 64) + assert len(set(partials)) == 2 + assert not list(tmp_path.glob("*.mp-partial")) + + +def test_reflink_uses_distinct_private_staging_names(tmp_path, monkeypatch): + """ + 直接调用宿主 reflink 的同目标任务也必须独立分配暂存名称。 + """ + source, target = tmp_path / "source", tmp_path / "target" + source.write_bytes(b"data") + seen = [] + copied = Barrier(2) + + def clone(src, dest): + """ + 回放并发克隆，强制两份暂存同时存活。 + """ + seen.append(dest) + shutil.copyfile(src, dest) + copied.wait(10) + return 0, "" + + monkeypatch.setattr(SystemUtils, "_reflink_copy", clone) + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(lambda _index: SystemUtils.reflink(source, target), range(2))) + assert results == [(0, ""), (0, "")] + assert len(set(seen)) == 2 + + +def test_audio_link_write_commits_despite_unchanged_timestamp_signature(tmp_path, monkeypatch): + """ + 相同长度标签更新即使文件系统时间戳没有变化，也不能丢弃独立副本。 + """ + source, target = tmp_path / "seed.flac", tmp_path / "library.flac" + source.write_bytes(b"old-title") + os.link(source, target) + real_signature = audio._audio_signature + signatures = {} + + def signature(path): + """ + 模拟时间粒度不足的文件系统，保持每条路径的首次版本签名。 + """ + signatures.setdefault(path, real_signature(path)) + return signatures[path] + + monkeypatch.setattr(audio, "_audio_signature", signature) + with audio._audio_write_target(target) as writable: + writable.write_bytes(b"new-title") + assert target.read_bytes() == b"new-title" + assert source.read_bytes() == b"old-title" + assert not target.samefile(source) + + +def test_recursive_rename_recognizes_nested_original_paths(tmp_path, monkeypatch): + """ + 递归重命名必须处理季目录内媒体，并用真实子路径进行识别。 + """ + root = tmp_path / "Original" + nested = root / "Season1" / "Show.S01E01.mkv" + nested.parent.mkdir(parents=True) + nested.write_bytes(b"fictional") + storage = object.__new__(LocalStorage) + recognized = [] + + def list_files(item, recursion=False): + """ + 在临时目录提供与存储链一致的递归列表合同。 + """ + directory = Path(item.path) + paths = directory.rglob("*") if recursion else directory.iterdir() + return [storage.get_item_strict(path) for path in paths] + + def recognize(path, **_kwargs): + """ + 记录实际识别位置，不读取外部媒体或来源。 + """ + recognized.append(path) + return SimpleNamespace(meta_info=path, media_info=object()) + + monkeypatch.setattr(storage_endpoint, "StorageChain", lambda: SimpleNamespace( + list_files=list_files, rename_file=storage.rename, + )) + monkeypatch.setattr(storage_endpoint, "MediaChain", lambda: SimpleNamespace(recognize_by_path=recognize)) + monkeypatch.setattr(storage_endpoint, "TransferChain", lambda: SimpleNamespace( + recommend_name=lambda **_kwargs: "Recognized.mkv")) + monkeypatch.setattr(storage_endpoint, "ProgressHelper", Mock()) + monkeypatch.setattr(storage_endpoint, "get_api_runtime_config_snapshot", lambda: SimpleNamespace( + media_extensions=[".mkv"], subtitle_extensions=[], audio_extensions=[])) + result = storage_endpoint.rename(storage.get_item_strict(root), "Renamed", recursive=True) + assert result.success + assert recognized == [nested] + assert (tmp_path / "Renamed" / "Season1" / "Recognized.mkv").read_bytes() == b"fictional" + + +def test_parallel_reflink_permission_probes_are_private(tmp_path, monkeypatch): + """ + 并发权限探针各自持有临时名称，清理另一任务探针不能使克隆退回复制。 + """ + source = tmp_path / "source" + source.write_bytes(b"data") + probes = [] + ready = Barrier(2) + write_bytes = Path.write_bytes + + def write_probe(path, data): + """ + 强制两个权限探针同时存在，捕获各自名称。 + """ + result = write_bytes(path, data) + if path.name.startswith(".mode-probe-"): + probes.append(path) + ready.wait(10) + return result + + def clone(src, dest): + """ + 使用真实复制回放支持 reflink 的文件系统。 + """ + shutil.copyfile(src, dest) + return 0, "" + + monkeypatch.setattr(Path, "write_bytes", write_probe) + monkeypatch.setattr(SystemUtils, "is_same_disk", lambda *_args: True) + monkeypatch.setattr(SystemUtils, "reflink", clone) + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(lambda index: LocalStorage._try_reflink_copy(source, tmp_path / str(index)), range(2))) + assert results == [True, True] + assert len(set(probes)) == 2 + assert not list(tmp_path.glob(".mode-probe-*")) diff --git a/tests/test_storage_review_regressions.py b/tests/test_storage_review_regressions.py new file mode 100644 index 000000000..fa6b8eefb --- /dev/null +++ b/tests/test_storage_review_regressions.py @@ -0,0 +1,165 @@ +""" +验证存储审查发现的分页、凭据和中断下载回归。 +""" + +from contextlib import contextmanager +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from app.modules.filemanager.storages import alist as alist_module +from app.modules.filemanager.storages.alipan import AliPan +from app.modules.filemanager.storages.alist import Alist +from app.schemas.file import FileItem + + +def test_alipan_token_refresh_preserves_drive_configuration(monkeypatch): + """ + 自动刷新只替换令牌字段，保留已授权空间和用户配置。 + """ + storage = object.__new__(AliPan) + configuration = {"refresh_token": "old", "access_token": "expired", "expires_in": 1, + "refresh_time": 1, "resource_drive_id": "drive", "custom": "preserved"} + storage._request_utils = Mock() + monkeypatch.setattr(storage, "get_conf", lambda: dict(configuration)) + monkeypatch.setattr(storage, "set_config", lambda value: (configuration.clear(), configuration.update(value))) + monkeypatch.setattr(storage, "_AliPan__refresh_access_token", lambda _token: { + "access_token": "fresh", "refresh_token": "next", "expires_in": 7200, + }) + assert storage.access_token == "fresh" + assert configuration["resource_drive_id"] == "drive" + assert configuration["custom"] == "preserved" + assert configuration["refresh_token"] == "next" + + +@pytest.mark.parametrize("response, expected", [(None, False), ({"code": "AccessDenied"}, False), ({}, True)]) +def test_alipan_delete_requires_success_response(monkeypatch, response, expected): + """ + 删除拒绝和无响应不能落账为成功，空 JSON 是开放接口正常成功响应。 + """ + storage = object.__new__(AliPan) + monkeypatch.setattr(storage, "_request_api", lambda *_args, **_kwargs: response) + assert storage.delete(FileItem(storage="alipan", fileid="file", drive_id="drive")) is expected + + +def test_alipan_full_last_page_stops_at_empty_marker(monkeypatch): + """ + 满页但没有下一页标记时只发送一次目录请求。 + """ + storage = object.__new__(AliPan) + monkeypatch.setattr(storage, "get_conf", lambda: {"resource_drive_id": "drive"}) + request = Mock(side_effect=[{"items": [{"file_id": str(i)} for i in range(100)], "next_marker": ""}, + AssertionError("末页不应重取第一页")]) + monkeypatch.setattr(storage, "_request_api", request) + monkeypatch.setattr(storage, "_AliPan__get_fileitem", lambda info, **_kwargs: FileItem(fileid=info["file_id"])) + assert len(storage.list(FileItem(storage="alipan", path="/", type="dir"))) == 100 + assert request.call_count == 1 + + +@pytest.mark.parametrize("failure", ["stream", "short", "cancel"]) +def test_openlist_incomplete_download_is_not_a_transfer_source(tmp_path, monkeypatch, failure): + """ + 流中断、正常提前结束和取消都必须返回失败并删除未完成文件。 + """ + storage = object.__new__(Alist) + monkeypatch.setattr(storage, "get_conf", lambda: {"url": "https://openlist.invalid", "token": "secret"}) + response = Mock(status_code=200) + response.json.return_value = {"code": 200, "data": {"raw_url": "https://cdn.invalid/file", "sign": ""}} + stream = Mock(headers={"Content-Length": "12"}) + + def chunks(**_kwargs): + """ + 只提供部分内容，并可模拟连接中断。 + """ + yield b"part" + if failure == "stream": + raise ConnectionError("stream interrupted") + + stream.iter_content = chunks + + @contextmanager + def get_stream(*_args, **_kwargs): + """ + 在网络边界回放下载流。 + """ + yield stream + + request = Mock() + request.post_res.return_value = response + request.get_stream = get_stream + monkeypatch.setattr(alist_module, "RequestUtils", lambda **_kwargs: request) + monkeypatch.setattr(alist_module.runtime_stop_state, "consume_transfer_stop", lambda _path: failure == "cancel") + item = FileItem(storage="alist", path="/movie.mkv", name="movie.mkv", type="file", size=12) + assert storage.download(item, tmp_path) is None + assert not list(tmp_path.iterdir()) + + +@pytest.mark.parametrize("raw_url", ["https://cdn.invalid/file", "https://openlist.invalid/d/file"]) +def test_openlist_download_never_forwards_access_token(tmp_path, monkeypatch, raw_url): + """ + 文件直链和签名下载地址不携带 OpenList API 令牌，跨源重定向也没有令牌可泄漏。 + """ + storage = object.__new__(Alist) + monkeypatch.setattr(storage, "get_conf", lambda: {"url": "https://openlist.invalid", "token": "secret"}) + response = Mock(status_code=200) + response.json.return_value = {"code": 200, "data": {"raw_url": raw_url, "sign": ""}} + configurations = [] + stream = Mock(headers={"Content-Length": "4"}) + stream.iter_content.return_value = [b"data"] + + @contextmanager + def get_stream(*_args, **_kwargs): + """ + 返回完整的离线内容响应。 + """ + yield stream + + def request_factory(**kwargs): + """ + 记录 API 和下载各自的请求头。 + """ + configurations.append(kwargs) + return SimpleNamespace(post_res=lambda *_args, **_kwargs: response, get_stream=get_stream) + + monkeypatch.setattr(alist_module, "RequestUtils", request_factory) + item = FileItem(storage="alist", path="/file.mkv", name="file.mkv", type="file", size=4) + result = storage.download(item, tmp_path) + assert result.read_bytes() == b"data" + assert "Authorization" not in configurations[-1].get("headers", {}) + + +def test_openlist_detail_size_prevents_legacy_move_deleting_truncated_source(tmp_path, monkeypatch): + """ + 旧文件项及无长度 HTTP 响应仍须使用详情大小阻止截断移动删除源。 + """ + from app.modules.filemanager.transhandler import TransHandler + storage = object.__new__(Alist) + monkeypatch.setattr(storage, "get_conf", lambda: {"url": "https://openlist.invalid", "token": "secret"}) + response = Mock(status_code=200) + response.json.return_value = {"code": 200, "data": { + "raw_url": "https://cdn.invalid/file", "sign": "", "size": 100, + }} + stream = Mock(headers={}) + stream.iter_content.return_value = [b"PART"] + + @contextmanager + def get_stream(*_args, **_kwargs): + """ + 模拟正常结束但短于详情大小的无长度下载。 + """ + yield stream + + request = Mock() + request.post_res.return_value = response + request.get_stream = get_stream + monkeypatch.setattr(alist_module, "RequestUtils", lambda **_kwargs: request) + deleted = Mock(return_value=True) + monkeypatch.setattr(storage, "delete", deleted) + source = FileItem(storage="alist", path="/Movie.mkv", name="Movie.mkv", type="file") + target = tmp_path / "Movie.mkv" + result, _ = TransHandler._TransHandler__download_to_local(source, storage, target, "move") + assert result is False + deleted.assert_not_called() + assert not target.exists() + assert not list(tmp_path.glob(".mp-download-*")) diff --git a/tests/test_system_reflink.py b/tests/test_system_reflink.py index 854d381d9..6cfb2d21d 100644 --- a/tests/test_system_reflink.py +++ b/tests/test_system_reflink.py @@ -68,13 +68,10 @@ def test_reflink_success_on_current_platform(tmp_path: Path): assert dest.read_bytes() == b"Modified cloned content" -def test_reflink_cleans_preexisting_temp_file(tmp_path: Path): +def test_reflink_preserves_unowned_legacy_temp_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): """ - 测试当目标临时 .mp 文件预先存在时，reflink 能先清理临时文件并成功完成克隆。 + 私有暂存不清理固定旧名称，避免删除其它任务仍在写入的文件。 """ - if not _reflink_available(tmp_path): - pytest.skip("当前文件系统不支持引用链接") - src = tmp_path / "source.txt" dest = tmp_path / "target.txt" temp_file = dest.with_suffix(dest.suffix + ".mp") @@ -82,11 +79,21 @@ def test_reflink_cleans_preexisting_temp_file(tmp_path: Path): src.write_text("source content", encoding="utf-8") temp_file.write_text("old stale temp content", encoding="utf-8") + def clone(source: Path, destination: Path): + """ + 在真实文件上回放底层克隆，避免依赖当前文件系统的 CoW 支持。 + """ + shutil.copyfile(source, destination) + return 0, "" + + monkeypatch.setattr(SystemUtils, "_reflink_copy", clone) + code, msg = SystemUtils.reflink(src, dest) assert code == 0 assert msg == "" assert dest.read_text(encoding="utf-8") == "source content" - assert not temp_file.exists() + assert temp_file.read_text(encoding="utf-8") == "old stale temp content" + assert not list(tmp_path.glob(".target.txt.*.mp")) def test_reflink_copy_failure_cleans_temp_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): @@ -95,10 +102,14 @@ def test_reflink_copy_failure_cleans_temp_file(tmp_path: Path, monkeypatch: pyte """ src = tmp_path / "source.txt" dest = tmp_path / "target.txt" - temp_file = dest.with_suffix(dest.suffix + ".mp") + temporary_files = [] src.write_text("source data", encoding="utf-8") def _mock_reflink_copy(_src: Path, _dest: Path): + """ + 记录本次实际暂存路径后回放部分内容写入失败。 + """ + temporary_files.append(_dest) # 模拟底层复制过程中生成了临时文件但最终报告失败 _dest.write_text("partial data", encoding="utf-8") return -1, "底层克隆失败模拟错误" @@ -109,7 +120,7 @@ def test_reflink_copy_failure_cleans_temp_file(tmp_path: Path, monkeypatch: pyte assert code == -1 assert "底层克隆失败模拟错误" in msg # 验证临时文件已被清理且目标未生成 - assert not temp_file.exists() + assert temporary_files and all(not path.exists() for path in temporary_files) assert not dest.exists() @@ -119,12 +130,11 @@ def test_reflink_source_file_not_found(tmp_path: Path): """ src = tmp_path / "nonexistent_source.txt" dest = tmp_path / "target.txt" - temp_file = dest.with_suffix(dest.suffix + ".mp") code, msg = SystemUtils.reflink(src, dest) assert code != 0 assert msg != "" - assert not temp_file.exists() + assert not list(tmp_path.glob(".target.txt.*.mp")) assert not dest.exists() @@ -134,14 +144,21 @@ def test_reflink_unexpected_exception_handling(tmp_path: Path, monkeypatch: pyte """ src = tmp_path / "source.txt" dest = tmp_path / "target.txt" - temp_file = dest.with_suffix(dest.suffix + ".mp") + temporary_files = [] src.write_text("source data", encoding="utf-8") def _mock_reflink_copy(_src: Path, _dest: Path): + """ + 记录实际克隆暂存路径供异常清理断言使用。 + """ + temporary_files.append(_dest) _dest.write_text("cloned data", encoding="utf-8") return 0, "" def _mock_move(_src, _dest): + """ + 模拟提升克隆文件失败。 + """ raise RuntimeError("模拟 shutil.move 移动文件异常") monkeypatch.setattr(SystemUtils, "_reflink_copy", _mock_reflink_copy) @@ -150,7 +167,7 @@ def test_reflink_unexpected_exception_handling(tmp_path: Path, monkeypatch: pyte code, msg = SystemUtils.reflink(src, dest) assert code == -1 assert "模拟 shutil.move 移动文件异常" in msg - assert not temp_file.exists() + assert temporary_files and all(not path.exists() for path in temporary_files) assert not dest.exists() diff --git a/tests/test_transfer_execution_runner.py b/tests/test_transfer_execution_runner.py index d4ada5903..d315c343a 100644 --- a/tests/test_transfer_execution_runner.py +++ b/tests/test_transfer_execution_runner.py @@ -403,7 +403,7 @@ class _ImmediateStepRunner: @pytest.mark.parametrize("frozen_time,live_time", [(1.0, 1.0), (1.0, 5.0), (None, 5.0), (1.0, None)]) def test_cross_storage_move_materializes_before_independent_source_delete(tmp_path, frozen_time, live_time): """ - 跨存储 move 先复制再独立删除源，源和目标时间戳变化或缺失均不阻止清理。 + 跨存储 move 先复制再独立删除源，云端时间戳变化或缺失不阻止清理，本地源版本保持稳定。 """ source_path = tmp_path / "source.mkv" source_path.write_bytes(b"movie") @@ -413,7 +413,7 @@ def test_cross_storage_move_materializes_before_independent_source_delete(tmp_pa name=source_path.name, type="file", size=source_path.stat().st_size, - modify_time=frozen_time, + modify_time=1.0, extension="mkv", ) target_item = FileItem( @@ -422,12 +422,12 @@ def test_cross_storage_move_materializes_before_independent_source_delete(tmp_pa name="source.mkv", type="file", size=source_item.size, - modify_time=source_item.modify_time, + modify_time=frozen_time, extension="mkv", ) source_oper = Mock() source_oper.delete.return_value = True - live_source = source_item.model_copy(update={"modify_time": live_time}) + live_source = source_item.model_copy() source_oper.get_item_strict.return_value = live_source target_oper = Mock() target_oper.get_item_strict.return_value = target_item.model_copy(update={"modify_time": live_time}) diff --git a/tests/test_transfer_review_regressions.py b/tests/test_transfer_review_regressions.py new file mode 100644 index 000000000..aef7a0b0d --- /dev/null +++ b/tests/test_transfer_review_regressions.py @@ -0,0 +1,256 @@ +""" +验证整理恢复、目录覆盖和上下文传递的审查回归。 +""" + +import os +import time +from contextlib import contextmanager +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from app.adapters.network.http import RequestUtils +from app.api.endpoints.transferhistory import restore_manual_transfer_history_batch +from app.application.transfer.execution import TransferExecutionConflictError, TransferExecutionState +from app.application.transfer.history import success_fields +from app.application.transfer.models import TransferPlanCheckpoint, TransferPlanItem, TransferPlanningInput +from app.chain.scraping import ScrapingChain +from app.chain.transfer.execution import _TransferManualReviewRequired +from app.domain.context import MediaInfo +from app.domain.meta.metavideo import MetaVideo +from app.modules.filemanager.module import FileManagerModule +from app.modules.filemanager.storages.alist import Alist +from app.modules.filemanager.storages.local import LocalStorage +from app.modules.filemanager.transhandler import TransHandler +from app.schemas.file import FileItem +from app.schemas.transfer import ManualTransferItem, TransferInfo +from app.schemas.types import MediaType +from tests.test_transfer_safe_overwrite import _checkpoint, _durable_runner, _execute +from tests.test_transfer_safe_overwrite import repository_factory as _repository_factory + + +@pytest.fixture +def repository_factory(): + """ + 复用真实 SQLite 整理账本载具，每个用例结束时关闭独立引擎。 + """ + yield from _repository_factory.__wrapped__() + + +def test_success_history_preserves_episode_group(): + """ + 整理成功历史应与失败历史一样保存自定义剧集组。 + """ + media = MediaInfo(type=MediaType.TV, title="Show", episode_group="group") + result = success_fields(FileItem(path="/source"), "copy", MetaVideo("Show.S01E01"), media, TransferInfo()) + assert result["episode_group"] == "group" + + +@pytest.mark.parametrize("groups, expected", [(["group", "group"], "group"), (["a", "b"], None)]) +def test_batch_retransfer_restores_only_common_episode_group(groups, expected): + """ + 批量重整只复用共同剧集组，混合批次不误用单条记录。 + """ + histories = [SimpleNamespace(status=True, mode="copy", src_fileitem={"path": f"/{index}"}, + type="电视剧", media_source="themoviedb", media_id="123", + music_type=None, downloader=None, download_hash=None, episode_group=group) + for index, group in enumerate(groups)] + request = ManualTransferItem(logids=list(range(len(groups))), from_history=True) + result = restore_manual_transfer_history_batch(request, SimpleNamespace(get=lambda identifier: histories[identifier])) + assert result[-1] is None + assert request.episode_group == expected + + +def test_episode_scraping_preserves_explicit_source(monkeypatch): + """ + 单集重新识别后继续使用本次请求指定的刮削来源。 + """ + media = MediaInfo(type=MediaType.TV, media_source="themoviedb", media_id="123", scrape_source="imdb") + recognized = MediaInfo(type=MediaType.TV, media_source="themoviedb", media_id="123") + monkeypatch.setattr("app.chain.scraping.MediaChain", lambda: SimpleNamespace(recognize_media=lambda **_kwargs: recognized)) + chain = object.__new__(ScrapingChain) + nfo, images = Mock(), Mock() + monkeypatch.setattr(chain, "_scrape_nfo_generic", nfo) + monkeypatch.setattr(chain, "_scrape_images_generic", images) + chain._handle_tv_episode_file(FileItem(path="/Show.S01E01.mkv"), Path("/Show.S01E01.mkv"), media, None, False) + assert nfo.call_args.kwargs["mediainfo"].scrape_source == "imdb" + assert images.call_args.kwargs["mediainfo"].scrape_source == "imdb" + + +def test_local_exists_includes_all_episodes(monkeypatch): + """ + 合并集媒体应将文件中声明的全部剧集加入本地查重结果。 + """ + module = object.__new__(FileManagerModule) + monkeypatch.setattr("app.modules.filemanager.module.get_runtime_setting", lambda *_args: True) + monkeypatch.setattr(module, "media_files", lambda _media: [FileItem(basename="Show.S01E01-E02")]) + assert module.media_exists(MediaInfo(type=MediaType.TV, title="Show")).seasons == {1: [1, 2]} + + +def test_directory_never_overwrite_preserves_library_and_source(tmp_path, monkeypatch): + """ + 原盘目录的 never 策略必须跳过覆盖，且不能误触发移动源清理。 + """ + source, target = tmp_path / "disc", tmp_path / "library" + source.mkdir() + target.mkdir() + leaf, existing = source / "00001.m2ts", target / "00001.m2ts" + leaf.write_bytes(b"new") + existing.write_bytes(b"old-library") + storage = object.__new__(LocalStorage) + item = storage.get_item_strict(source).model_dump(mode="json") + checkpoint = replace(_checkpoint(leaf, existing, transfer_type="move", mode="never"), + planning_input=TransferPlanningInput(source_fileitem=item), final_target_path=str(target), + items=(TransferPlanItem(sequence=0, source_fileitem=storage.get_item_strict(leaf).model_dump(mode="json"), + target_storage="local", target_path=str(existing)),)) + result = _execute(checkpoint, monkeypatch, None) + assert existing.read_bytes() == b"old-library" + assert leaf.read_bytes() == b"new" + assert not result.success and result.overwrite_skipped + + +@pytest.mark.parametrize("mutation", ["source", "target"]) +def test_recovery_retains_all_copies_after_same_size_mutation(tmp_path, monkeypatch, repository_factory, mutation): + """ + 提升落账后发生等长原地修改，恢复必须隔离任务并保留源和备份。 + """ + source, target = tmp_path / "download.mkv", tmp_path / "Movie.2026.mkv" + source.write_bytes(b"new-version") + target.write_bytes(b"old-version") + checkpoint = _checkpoint(source, target, transfer_type="move") + repository = repository_factory(checkpoint) + runner = _durable_runner(repository, checkpoint) + real_run = runner.run + + class ProcessStopped(BaseException): + """ + 模拟删除源之前进程退出，不让执行器转译成普通失败。 + """ + + def stop_before_cleanup(**kwargs): + """ + 在提升已有持久回执后中断。 + """ + if kwargs["kind"] == "delete_overwrite_source": + raise ProcessStopped() + return real_run(**kwargs) + + monkeypatch.setattr(runner, "run", stop_before_cleanup) + with pytest.raises(ProcessStopped): + _execute(checkpoint, monkeypatch, runner) + backup = next(tmp_path.glob(".mp-backup-*")) + changed = source if mutation == "source" else target + before = changed.stat() + changed.write_bytes(b"NEW-CONTENT" if mutation == "source" else b"CORRUPTION!") + os.utime(changed, ns=(before.st_atime_ns, before.st_mtime_ns + 1)) + with pytest.raises(_TransferManualReviewRequired): + _execute(checkpoint, monkeypatch, _durable_runner(repository, checkpoint)) + assert repository.get_snapshot(task_id="safe-overwrite").state is TransferExecutionState.MANUAL_REVIEW + assert source.exists() and backup.exists() + assert backup.read_bytes() == b"old-version" + + +@pytest.mark.parametrize("storage", ["u115", "alipan", "alist"]) +def test_cloud_cleanup_identity_allows_timestamp_changes(storage): + """ + 云端大小与稳定 ID 相同时保留上传后时间戳变化的兼容合同。 + """ + expected = FileItem(storage=storage, path="/file", type="file", size=10, fileid="id", modify_time=1.0) + TransHandler._TransHandler__assert_file_identity(expected, expected.model_copy(update={"modify_time": 5.0})) + + +def test_local_cleanup_rejects_mutation_with_restored_mtime(tmp_path): + """ + 原地改写后还原 mtime 仍会改变 ctime，本地清理必须拒绝这类版本变化。 + """ + path = tmp_path / "file" + path.write_bytes(b"old") + local = object.__new__(LocalStorage) + expected = local.get_item_strict(path) + before = path.stat() + time.sleep(0.02) + path.write_bytes(b"new") + os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns)) + current = local.get_item_strict(path) + assert expected.modify_time_ns == current.modify_time_ns + assert expected.change_time_ns != current.change_time_ns + with pytest.raises(TransferExecutionConflictError, match="状态时间已变化"): + TransHandler._TransHandler__assert_file_identity(expected, current) + + +@pytest.mark.parametrize("failure", ["exception", "detail_size_only"]) +def test_openlist_truncated_move_never_deletes_remote_source(tmp_path, monkeypatch, repository_factory, failure): + """ + 真实下载与持久步骤执行器遇到截断流时保留远端原件，不发布本地损坏文件。 + """ + source = object.__new__(Alist) + target = object.__new__(LocalStorage) + remote = FileItem(storage="alist", path="/download/Movie.mkv", name="Movie.mkv", + basename="Movie", extension="mkv", type="file", + size=None if failure == "detail_size_only" else 100, modify_time=1.0) + payload = remote.model_dump(mode="json") + destination = tmp_path / "Movie.mkv" + checkpoint = TransferPlanCheckpoint( + planning_input=TransferPlanningInput(source_fileitem=payload), target_storage="local", + root_target_path=str(tmp_path), final_target_path=str(destination), resolved_transfer_type="move", + overwrite_mode="never", items=(TransferPlanItem(sequence=0, source_fileitem=payload, + target_storage="local", target_path=str(destination)),), + ) + repository = repository_factory(checkpoint) + response = Mock(status_code=200) + response.json.return_value = {"code": 200, "data": { + "raw_url": "https://cdn.invalid/file", "sign": "", "size": 100, + }} + stream = Mock(headers={} if failure == "detail_size_only" else {"Content-Length": "100"}) + + def chunks(**_kwargs): + """ + 在远端下载传输边界回放部分写入后的连接中断。 + """ + yield b"PART" + if failure == "exception": + raise ConnectionError("interrupted") + + @contextmanager + def download_stream(*_args, **_kwargs): + """ + 保留真实下载代码，只替换 HTTP 响应。 + """ + yield stream + + stream.iter_content = chunks + monkeypatch.setattr(source, "get_conf", lambda: {"url": "https://openlist.invalid", "token": "secret"}) + monkeypatch.setattr(RequestUtils, "post_res", lambda *_args, **_kwargs: response) + monkeypatch.setattr(RequestUtils, "get_stream", download_stream) + deleted = Mock(return_value=True) + monkeypatch.setattr(source, "delete", deleted) + handler = TransHandler() + monkeypatch.setattr(handler, "_TransHandler__intercept_transfer", lambda **_kwargs: (True, "")) + with pytest.raises(RuntimeError, match="下载"): + handler.execute_transfer_plan( + checkpoint, meta=MetaVideo("Movie.2026.mkv"), mediainfo=MediaInfo(type=MediaType.MOVIE, title="Movie"), + source_oper=source, target_oper=target, step_runner=_durable_runner(repository, checkpoint), + ) + deleted.assert_not_called() + assert not destination.exists() + assert not list(tmp_path.glob(".mp-download-*")) + assert all(step.kind != "delete_move_source" for step in repository.get_snapshot(task_id="safe-overwrite").steps) + + +def test_local_timestamp_json_roundtrip_keeps_nanosecond_precision(): + """ + 纳秒版本证据超过浏览器整数精度时以字符串无损传输，回传后恢复为 Python 整数。 + """ + stamp = 1_792_691_575_120_037_153 + item = FileItem(modify_time_ns=stamp, change_time_ns=stamp + 1) + payload = item.model_dump(mode="json") + assert payload["modify_time_ns"] == str(stamp) + assert payload["change_time_ns"] == str(stamp + 1) + restored = FileItem.model_validate(payload) + assert restored.modify_time_ns == stamp + assert restored.change_time_ns == stamp + 1 + assert restored.model_dump()["modify_time_ns"] == stamp + assert FileItem().model_dump(mode="json")["modify_time_ns"] is None diff --git a/tests/test_transfer_safe_overwrite.py b/tests/test_transfer_safe_overwrite.py index 08837b83f..74615333a 100644 --- a/tests/test_transfer_safe_overwrite.py +++ b/tests/test_transfer_safe_overwrite.py @@ -1022,11 +1022,11 @@ def test_completed_promotion_rechecks_live_files_before_destructive_cleanup( assert backup.read_bytes() == expected_backup -def test_completed_promotion_allows_cleanup_after_timestamp_changes( +def test_completed_local_promotion_retains_copies_after_timestamp_changes( tmp_path, monkeypatch, repository_factory, ): """ - 提升结果已落账后，即使源、目标和备份的时间戳均变化，恢复仍按大小和身份完成清理。 + 本地提升落账后时间戳改变意味着内容版本不确定，恢复须保留所有副本。 """ source, target = tmp_path / "download.mkv", tmp_path / "Movie.2026.mkv" source.write_bytes(b"new-version") @@ -1057,11 +1057,11 @@ def test_completed_promotion_allows_cleanup_after_timestamp_changes( before = path.stat() os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns + 4_000_000_000)) - result = _execute(checkpoint, monkeypatch, _durable_runner(repository, checkpoint)) - assert result.success + with pytest.raises(execution_module._TransferManualReviewRequired): + _execute(checkpoint, monkeypatch, _durable_runner(repository, checkpoint)) assert target.read_bytes() == b"new-version" - assert not source.exists() - assert not backup.exists() + assert source.read_bytes() == b"new-version" + assert backup.read_bytes() == b"old-version" @pytest.mark.parametrize("missing_evidence", ["source", "target"])