From cac298fd022a1fbd8b14d9dc2fd6d51caa4c2c1d Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 3 Oct 2026 08:33:15 +0800 Subject: [PATCH] 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