From 1965cc4de728979530164efd6bce3a7e12fd3cca Mon Sep 17 00:00:00 2001 From: ranzhigg <126594517+ranzhigg@users.noreply.github.com> Date: Thu, 1 Oct 2026 15:49:12 +0800 Subject: [PATCH 1/5] perf(monitor): index incremental snapshot reconciliation --- app/modules/filemanager/storages/__init__.py | 42 +++++++-- ...est_snapshot_reconciliation_performance.py | 90 +++++++++++++++++++ 2 files changed, 123 insertions(+), 9 deletions(-) create mode 100644 tests/test_snapshot_reconciliation_performance.py diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index 64f7bf2f7..78667a7f3 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -330,6 +330,37 @@ class StorageBase(metaclass=ABCMeta): if PurePosixPath(file_path).is_relative_to(root_path) } + # 每次目录列举只需要核对该目录的直接子项。旧实现会在每层目录 + # 上重新扫描整份 files_info，云盘监控目录较多时会退化为 + # O(目录数 * 文件数)。预先按“目录 -> 直接子项”建立索引，删除 + # 目录时沿索引移除其完整子树，但不再重复遍历全量快照。 + previous_children: Dict[PurePosixPath, set[PurePosixPath]] = {} + previous_files: Dict[PurePosixPath, str] = {} + if files_info: + for file_path in files_info: + file_posix_path = PurePosixPath(file_path) + previous_files[file_posix_path] = file_path + try: + relative_parts = file_posix_path.relative_to(root_path).parts + except ValueError: + # files_info 已经按 root_path 过滤；保留防御性判断，避免 + # 非标准路径对象让增量索引中断整轮监控。 + continue + for depth in range(len(relative_parts)): + directory_path = root_path.joinpath(*relative_parts[:depth]) + child_path = directory_path / relative_parts[depth] + previous_children.setdefault(directory_path, set()).add(child_path) + + def __remove_previous_subtree(path: PurePosixPath) -> None: + """沿旧快照索引移除一个已经消失的直接子树。""" + pending = [path] + while pending: + current_path = pending.pop() + old_file_path = previous_files.pop(current_path, None) + if old_file_path is not None: + files_info.pop(old_file_path, None) + pending.extend(previous_children.pop(current_path, ())) + def __remove_deleted_children(_fileitm: _SchemaFileItem, sub_files: List[_SchemaFileItem]) -> None: """ @@ -338,16 +369,9 @@ class StorageBase(metaclass=ABCMeta): """ directory_path = PurePosixPath(_fileitm.path) child_paths = {PurePosixPath(sub_file.path) for sub_file in sub_files} - for old_file_path in list(files_info): - try: - relative_path = PurePosixPath(old_file_path).relative_to(directory_path) - except ValueError: - continue - if not relative_path.parts: - continue - direct_child_path = directory_path / relative_path.parts[0] + for direct_child_path in previous_children.pop(directory_path, ()): if direct_child_path not in child_paths: - files_info.pop(old_file_path, None) + __remove_previous_subtree(direct_child_path) def __snapshot_file(_fileitm: _SchemaFileItem, current_depth: int = 0): """ diff --git a/tests/test_snapshot_reconciliation_performance.py b/tests/test_snapshot_reconciliation_performance.py new file mode 100644 index 000000000..9362c99ec --- /dev/null +++ b/tests/test_snapshot_reconciliation_performance.py @@ -0,0 +1,90 @@ +from pathlib import Path +from unittest.mock import MagicMock + +from app import schemas +from app.modules.filemanager.storages import StorageBase + + +def _directory(path: str, *, modify_time: int = 200): + return schemas.FileItem( + storage="alist", + type="dir", + path=f"{path.rstrip('/')}/", + name=Path(path).name, + modify_time=modify_time, + ) + + +def _file(path: str, *, size: int = 1): + return schemas.FileItem( + storage="alist", + type="file", + path=path, + name=Path(path).name, + size=size, + modify_time=200, + ) + + +def test_snapshot_removes_a_deleted_directory_without_scanning_other_subtrees(): + """ + 目录删除只应移除该目录的文件，而不影响其它目录的旧基线。 + """ + storage = MagicMock() + storage.snapshot_check_folder_modtime = True + storage.snapshot_strict_query = False + storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item) + root = _directory("/mon") + keep = _directory("/mon/keep", modify_time=50) + storage.get_item.return_value = root + storage.list.return_value = [keep] + + previous_snapshot = { + "/mon/gone/episode-1.mkv": {"size": 1, "modify_time": 100}, + "/mon/gone/episode-2.mkv": {"size": 2, "modify_time": 100}, + "/mon/keep/episode-1.mkv": {"size": 3, "modify_time": 50}, + } + + snapshot = StorageBase.snapshot( + storage, + Path("/mon"), + last_snapshot_time=100, + previous_snapshot=previous_snapshot, + ) + + assert snapshot == { + "/mon/keep/episode-1.mkv": {"size": 3, "modify_time": 50} + } + + +def test_snapshot_reconciles_nested_deleted_files_with_bounded_directory_work(): + """ + 每个被列举目录只需处理自己的直接子项；嵌套文件删除仍应正确收敛。 + """ + storage = MagicMock() + storage.snapshot_check_folder_modtime = False + storage.snapshot_strict_query = False + storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item) + root = _directory("/mon") + season = _directory("/mon/show/Season 1") + storage.get_item.return_value = root + storage.list.side_effect = [ + [_directory("/mon/show")], + [season], + [_file("/mon/show/Season 1/episode-2.mkv", size=20)], + ] + + previous_snapshot = { + "/mon/show/Season 1/episode-1.mkv": {"size": 10, "modify_time": 100}, + "/mon/show/Season 1/episode-2.mkv": {"size": 20, "modify_time": 100}, + } + + snapshot = StorageBase.snapshot( + storage, + Path("/mon"), + previous_snapshot=previous_snapshot, + ) + + assert snapshot == { + "/mon/show/Season 1/episode-2.mkv": {"size": 20, "modify_time": 200, "fileid": None, "type": "file"} + } From cac298fd022a1fbd8b14d9dc2fd6d51caa4c2c1d Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 3 Oct 2026 08:33:15 +0800 Subject: [PATCH 2/5] fix(monitor): keep snapshot reconciliation within complexity budget --- app/modules/filemanager/storages/__init__.py | 243 ++++++++++++------- 1 file changed, 156 insertions(+), 87 deletions(-) diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index 78667a7f3..9b936c57e 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -312,6 +312,150 @@ class StorageBase(metaclass=ABCMeta): raise StorageQueryError(f"读取快照路径失败: {fileitem.path} - {error}") from error logger.debug(f"Snapshot error for {fileitem.path}: {error}") + @staticmethod + def _snapshot_previous_state( + root_path: PurePosixPath, + previous_snapshot: Optional[Dict[str, Dict]], + ) -> Tuple[ + Dict[str, Dict], + Dict[PurePosixPath, set[PurePosixPath]], + Dict[PurePosixPath, str], + ]: + """ + 为增量快照建立旧文件和目录直接子项索引。 + + :param root_path: 本轮快照的根路径 + :param previous_snapshot: 上一轮完整快照 + :return: 当前文件信息、目录直接子项索引和路径反向索引 + """ + files_info = { + file_path: file_info + for file_path, file_info in (previous_snapshot or {}).items() + if PurePosixPath(file_path).is_relative_to(root_path) + } + previous_children: Dict[PurePosixPath, set[PurePosixPath]] = {} + previous_files: Dict[PurePosixPath, str] = {} + for file_path in files_info: + file_posix_path = PurePosixPath(file_path) + previous_files[file_posix_path] = file_path + try: + relative_parts = file_posix_path.relative_to(root_path).parts + except ValueError: + # files_info 已经按 root_path 过滤；保留防御性判断，避免 + # 非标准路径对象让增量索引中断整轮监控。 + continue + for depth in range(len(relative_parts)): + directory_path = root_path.joinpath(*relative_parts[:depth]) + child_path = directory_path / relative_parts[depth] + previous_children.setdefault(directory_path, set()).add(child_path) + return files_info, previous_children, previous_files + + @staticmethod + def _snapshot_remove_previous_subtree( + path: PurePosixPath, + files_info: Dict[str, Dict], + previous_children: Dict[PurePosixPath, set[PurePosixPath]], + previous_files: Dict[PurePosixPath, str], + ) -> None: + """沿旧快照索引移除一个已经消失的直接子树。""" + pending = [path] + while pending: + current_path = pending.pop() + old_file_path = previous_files.pop(current_path, None) + if old_file_path is not None: + files_info.pop(old_file_path, None) + pending.extend(previous_children.pop(current_path, ())) + + def _snapshot_remove_deleted_children( + self, + fileitem: _SchemaFileItem, + sub_files: List[_SchemaFileItem], + files_info: Dict[str, Dict], + previous_children: Dict[PurePosixPath, set[PurePosixPath]], + previous_files: Dict[PurePosixPath, str], + ) -> None: + """ + 清理已确认遍历目录中不再存在的直接子项。 + + :param fileitem: 当前已列举的目录 + :param sub_files: 当前目录返回的直接子项 + :param files_info: 当前正在构建的快照 + :param previous_children: 旧快照目录直接子项索引 + :param previous_files: 旧快照路径反向索引 + """ + directory_path = PurePosixPath(fileitem.path) + child_paths = {PurePosixPath(sub_file.path) for sub_file in sub_files} + for direct_child_path in previous_children.pop(directory_path, ()): + if direct_child_path not in child_paths: + StorageBase._snapshot_remove_previous_subtree( + direct_child_path, files_info, previous_children, previous_files + ) + + def _snapshot_file( + self, + fileitem: _SchemaFileItem, + files_info: Dict[str, Dict], + previous_children: Dict[PurePosixPath, set[PurePosixPath]], + previous_files: Dict[PurePosixPath, str], + last_snapshot_time: float, + max_depth: int, + current_depth: int = 0, + ) -> None: + """ + 递归读取目录并更新快照文件信息。 + + :param fileitem: 当前文件或目录 + :param files_info: 当前正在构建的快照 + :param previous_children: 旧快照目录直接子项索引 + :param previous_files: 旧快照路径反向索引 + :param last_snapshot_time: 上次快照时间 + :param max_depth: 最大递归深度 + :param current_depth: 当前递归深度 + """ + try: + if fileitem.type == "dir": + if current_depth >= max_depth: + return + + # 根目录每轮至少列举一次；子目录按修改时间跳过未变化分支。 + if (current_depth > 0 and + self.snapshot_check_folder_modtime and + last_snapshot_time and + fileitem.modify_time and + fileitem.modify_time <= last_snapshot_time): + return + + # 只有目录列表成功返回后才清理旧基线，查询异常时保留待下轮重试。 + sub_files = self._snapshot_list(fileitem) + if sub_files is None: + return + sub_files = list(sub_files) + StorageBase._snapshot_remove_deleted_children( + self, + fileitem, sub_files, files_info, previous_children, previous_files + ) + for sub_file in sub_files: + StorageBase._snapshot_file( + self, + sub_file, + files_info, + previous_children, + previous_files, + last_snapshot_time, + max_depth, + current_depth + 1, + ) + else: + # 始终记录文件完整信息，由 compare_snapshots 负责检测变化。 + files_info[fileitem.path] = { + 'size': fileitem.size or 0, + 'modify_time': getattr(fileitem, 'modify_time', 0), + 'fileid': getattr(fileitem, 'fileid', None), + 'type': fileitem.type + } + except Exception as error: + self._snapshot_read_error(fileitem, error) + def snapshot(self, path: Path, last_snapshot_time: float = None, max_depth: int = 5, previous_snapshot: Optional[Dict[str, Dict]] = None) -> Optional[Dict[str, Dict]]: """ @@ -324,98 +468,23 @@ class StorageBase(metaclass=ABCMeta): """ strict_query = self.snapshot_strict_query root_path = PurePosixPath(path.as_posix()) - files_info = { - file_path: file_info - for file_path, file_info in (previous_snapshot or {}).items() - if PurePosixPath(file_path).is_relative_to(root_path) - } - - # 每次目录列举只需要核对该目录的直接子项。旧实现会在每层目录 - # 上重新扫描整份 files_info，云盘监控目录较多时会退化为 - # O(目录数 * 文件数)。预先按“目录 -> 直接子项”建立索引，删除 - # 目录时沿索引移除其完整子树，但不再重复遍历全量快照。 - previous_children: Dict[PurePosixPath, set[PurePosixPath]] = {} - previous_files: Dict[PurePosixPath, str] = {} - if files_info: - for file_path in files_info: - file_posix_path = PurePosixPath(file_path) - previous_files[file_posix_path] = file_path - try: - relative_parts = file_posix_path.relative_to(root_path).parts - except ValueError: - # files_info 已经按 root_path 过滤；保留防御性判断，避免 - # 非标准路径对象让增量索引中断整轮监控。 - continue - for depth in range(len(relative_parts)): - directory_path = root_path.joinpath(*relative_parts[:depth]) - child_path = directory_path / relative_parts[depth] - previous_children.setdefault(directory_path, set()).add(child_path) - - def __remove_previous_subtree(path: PurePosixPath) -> None: - """沿旧快照索引移除一个已经消失的直接子树。""" - pending = [path] - while pending: - current_path = pending.pop() - old_file_path = previous_files.pop(current_path, None) - if old_file_path is not None: - files_info.pop(old_file_path, None) - pending.extend(previous_children.pop(current_path, ())) - - def __remove_deleted_children(_fileitm: _SchemaFileItem, - sub_files: List[_SchemaFileItem]) -> None: - """ - 清理已确认遍历目录中不再存在的直接子项。 - 未变化的子目录仍保留旧基线，避免增量遍历将其误删。 - """ - directory_path = PurePosixPath(_fileitm.path) - child_paths = {PurePosixPath(sub_file.path) for sub_file in sub_files} - for direct_child_path in previous_children.pop(directory_path, ()): - if direct_child_path not in child_paths: - __remove_previous_subtree(direct_child_path) - - def __snapshot_file(_fileitm: _SchemaFileItem, current_depth: int = 0): - """ - 递归获取文件信息 - """ - try: - if _fileitm.type == "dir": - # 检查递归深度限制 - if current_depth >= max_depth: - return - - # 根目录每轮至少列举一次，用于清理已移走的直接子项；子目录仍按修改时间增量遍历 - if (current_depth > 0 and - self.snapshot_check_folder_modtime and - last_snapshot_time and - _fileitm.modify_time and - _fileitm.modify_time <= last_snapshot_time): - return - - # 只有目录列表成功返回后才清理旧基线，查询异常时继续保留待下轮重试 - sub_files = self._snapshot_list(_fileitm) - if sub_files is None: - return - sub_files = list(sub_files) - __remove_deleted_children(_fileitm, sub_files) - for sub_file in sub_files: - __snapshot_file(sub_file, current_depth + 1) - else: - # 记录文件的完整信息用于比对（始终包含所有文件，由 compare_snapshots 负责检测变化） - files_info[_fileitm.path] = { - 'size': _fileitm.size or 0, - 'modify_time': getattr(_fileitm, 'modify_time', 0), - 'fileid': getattr(_fileitm, 'fileid', None), - 'type': _fileitm.type - } - - except Exception as e: - self._snapshot_read_error(_fileitm, e) + files_info, previous_children, previous_files = StorageBase._snapshot_previous_state( + root_path, previous_snapshot + ) try: fileitem = self.get_item_strict(path) if strict_query else self.get_item(path) if not fileitem: return {} - __snapshot_file(fileitem) + StorageBase._snapshot_file( + self, + fileitem, + files_info, + previous_children, + previous_files, + last_snapshot_time, + max_depth, + ) except StorageQueryError as err: logger.warning(f"存储快照失败，保留上次基线: {self.schema}:{path} - {err}") return None From 66e791603cf9d6ff2a0919825a51e1fc0457d584 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 3 Oct 2026 08:43:31 +0800 Subject: [PATCH 3/5] fix(monitor): type snapshot reconciliation state --- app/modules/filemanager/storages/__init__.py | 15 +++++++++------ tests/fixtures/architecture/mypy-baseline.json | 2 +- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index 9b936c57e..5ccb60a5e 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -1,6 +1,6 @@ from abc import ABCMeta, abstractmethod from pathlib import Path, PurePosixPath -from typing import Callable, Dict, List, Optional, Tuple, Union +from typing import Any, Callable, Dict, List, Optional, Tuple, Union from tqdm import tqdm @@ -14,6 +14,9 @@ from app.schemas.system import StorageConf as _SchemaStorageConf from app.schemas.workflow import FileItem as _SchemaFileItem +_SnapshotInfo = Dict[str, Dict[str, Any]] + + def transfer_process(path: str) -> Callable[[int | float], None]: """ 传输进度回调 @@ -315,9 +318,9 @@ class StorageBase(metaclass=ABCMeta): @staticmethod def _snapshot_previous_state( root_path: PurePosixPath, - previous_snapshot: Optional[Dict[str, Dict]], + previous_snapshot: Optional[_SnapshotInfo], ) -> Tuple[ - Dict[str, Dict], + _SnapshotInfo, Dict[PurePosixPath, set[PurePosixPath]], Dict[PurePosixPath, str], ]: @@ -353,7 +356,7 @@ class StorageBase(metaclass=ABCMeta): @staticmethod def _snapshot_remove_previous_subtree( path: PurePosixPath, - files_info: Dict[str, Dict], + files_info: _SnapshotInfo, previous_children: Dict[PurePosixPath, set[PurePosixPath]], previous_files: Dict[PurePosixPath, str], ) -> None: @@ -370,7 +373,7 @@ class StorageBase(metaclass=ABCMeta): self, fileitem: _SchemaFileItem, sub_files: List[_SchemaFileItem], - files_info: Dict[str, Dict], + files_info: _SnapshotInfo, previous_children: Dict[PurePosixPath, set[PurePosixPath]], previous_files: Dict[PurePosixPath, str], ) -> None: @@ -394,7 +397,7 @@ class StorageBase(metaclass=ABCMeta): def _snapshot_file( self, fileitem: _SchemaFileItem, - files_info: Dict[str, Dict], + files_info: _SnapshotInfo, previous_children: Dict[PurePosixPath, set[PurePosixPath]], previous_files: Dict[PurePosixPath, str], last_snapshot_time: float, diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 0186acfc2..d72d2f797 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -1517,7 +1517,7 @@ "attr-defined": 4, "index": 1, "no-untyped-call": 2, - "no-untyped-def": 8, + "no-untyped-def": 7, "return-value": 1, "type-arg": 6, "var-annotated": 1 From db4e89800570692eaecbc26df35afe8c497d99c1 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 3 Oct 2026 08:47:31 +0800 Subject: [PATCH 4/5] style(monitor): normalize snapshot imports --- app/modules/filemanager/storages/__init__.py | 1 - 1 file changed, 1 deletion(-) diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index 5ccb60a5e..dd6beb4c5 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -13,7 +13,6 @@ from app.schemas.file import StorageUsage as _SchemaStorageUsage from app.schemas.system import StorageConf as _SchemaStorageConf from app.schemas.workflow import FileItem as _SchemaFileItem - _SnapshotInfo = Dict[str, Dict[str, Any]] From afa6a42a55cd2ff688f9cd5cf27fe5eb79738623 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 3 Oct 2026 09:03:56 +0800 Subject: [PATCH 5/5] docs(architecture): sync mypy debt metric --- docs/refactor/optimization-checklist.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 61b7474fb..91db547be 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -106,7 +106,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Event Contract | 53 | 均已有 payload model，但当前全部是 diagnostic enforcement | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`；61 个文件超过 1,000 行，11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 | -| 全量 mypy 历史债务 | 9,177 / 508 文件 | 系统版本查询补齐类型后的现状基线；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | +| 全量 mypy 历史债务 | 9,176 / 508 文件 | 系统版本查询补齐类型后的现状基线；消息媒体交互类型债务已清零，canonical Facade 与 endpoint 类型边界已补齐，低水位只允许继续下降 | | Ruff 历史诊断 | 505 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%，Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |