From 5c7e09d40727f4cf8994d9214da0ec1c044373c0 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sun, 9 Aug 2026 16:37:25 +0800 Subject: [PATCH] =?UTF-8?q?fix(monitor):=20=E4=BF=AE=E6=AD=A3=E8=BF=9C?= =?UTF-8?q?=E7=A8=8B=E5=BF=AB=E7=85=A7=E6=96=87=E4=BB=B6=E6=95=B0=E9=87=8F?= =?UTF-8?q?=E7=B4=AF=E8=AE=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/chain/storage.py | 7 +- app/modules/filemanager/__init__.py | 14 ++- app/modules/filemanager/storages/__init__.py | 43 +++++++-- app/monitor/monitor.py | 10 ++- app/monitor/poller.py | 50 ++++++++--- app/monitor/snapshot.py | 11 ++- tests/test_monitor_snapshot_semantics.py | 93 +++++++++++++++++++- 7 files changed, 199 insertions(+), 29 deletions(-) diff --git a/app/chain/storage.py b/app/chain/storage.py index 73b125198..471bb76bd 100644 --- a/app/chain/storage.py +++ b/app/chain/storage.py @@ -122,16 +122,19 @@ class StorageChain(ChainBase): return self.run_module("get_parent_item", fileitem=fileitem) def snapshot_storage(self, storage: str, path: Path, - last_snapshot_time: float = None, max_depth: int = 5) -> Optional[Dict[str, Dict]]: + last_snapshot_time: float = None, max_depth: int = 5, + previous_snapshot: Optional[Dict[str, Dict]] = None) -> Optional[Dict[str, Dict]]: """ 快照存储 :param storage: 存储类型 :param path: 路径 :param last_snapshot_time: 上次快照时间，用于增量快照 :param max_depth: 最大递归深度，避免过深遍历 + :param previous_snapshot: 上次完整快照，用于增量对账 """ return self.run_module("snapshot_storage", storage=storage, path=path, - last_snapshot_time=last_snapshot_time, max_depth=max_depth) + last_snapshot_time=last_snapshot_time, max_depth=max_depth, + previous_snapshot=previous_snapshot) def storage_usage(self, storage: str) -> Optional[schemas.StorageUsage]: """ diff --git a/app/modules/filemanager/__init__.py b/app/modules/filemanager/__init__.py index 6fd2f412d..11579991d 100644 --- a/app/modules/filemanager/__init__.py +++ b/app/modules/filemanager/__init__.py @@ -32,6 +32,7 @@ class FileManagerModule(_ModuleBase): self.messagehelper = MessageHelper() def init_module(self) -> None: + """初始化文件整理模块支持的存储实现""" # 加载模块 self._storage_schemas = ModuleHelper.load('app.modules.filemanager.storages', filter_func=lambda _, obj: hasattr(obj, 'schema') and obj.schema) @@ -40,6 +41,7 @@ class FileManagerModule(_ModuleBase): @staticmethod def get_name() -> str: + """获取模块名称""" return "文件整理" @staticmethod @@ -64,6 +66,7 @@ class FileManagerModule(_ModuleBase): return 4 def stop(self): + """停止文件整理模块""" pass def test(self) -> Tuple[bool, str]: @@ -384,13 +387,15 @@ class FileManagerModule(_ModuleBase): return storage_oper.get_parent(fileitem) def snapshot_storage(self, storage: str, path: Path, - last_snapshot_time: float = None, max_depth: int = 5) -> Optional[Dict[str, Dict]]: + last_snapshot_time: float = None, max_depth: int = 5, + previous_snapshot: Optional[Dict[str, Dict]] = None) -> Optional[Dict[str, Dict]]: """ 快照存储 :param storage: 存储类型 :param path: 路径 :param last_snapshot_time: 上次快照时间，用于增量快照 :param max_depth: 最大递归深度，避免过深遍历 + :param previous_snapshot: 上次完整快照，用于增量对账 """ if storage not in self._support_storages: return None @@ -398,7 +403,12 @@ class FileManagerModule(_ModuleBase): if not storage_oper: logger.error(f"不支持 {storage} 的快照处理") return None - return storage_oper.snapshot(path, last_snapshot_time=last_snapshot_time, max_depth=max_depth) + return storage_oper.snapshot( + path, + last_snapshot_time=last_snapshot_time, + max_depth=max_depth, + previous_snapshot=previous_snapshot + ) def storage_usage(self, storage: str) -> Optional[StorageUsage]: """ diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index 83edc8440..8987c827e 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -55,6 +55,7 @@ class StorageBase(metaclass=ABCMeta): pass def generate_qrcode(self, *args, **kwargs) -> Optional[Tuple[dict, str]]: + """生成存储登录二维码""" pass def generate_auth_url(self, *args, **kwargs) -> Optional[Tuple[dict, str]]: @@ -64,6 +65,7 @@ class StorageBase(metaclass=ABCMeta): return {}, "此存储不支持 OAuth2 授权" def check_login(self, *args, **kwargs) -> Optional[Dict[str, str]]: + """检查存储登录状态""" pass def get_config(self) -> Optional[schemas.StorageConf]: @@ -269,14 +271,40 @@ class StorageBase(metaclass=ABCMeta): """ pass - def snapshot(self, path: Path, last_snapshot_time: float = None, max_depth: int = 5) -> Dict[str, Dict]: + def snapshot(self, path: Path, last_snapshot_time: float = None, max_depth: int = 5, + previous_snapshot: Optional[Dict[str, Dict]] = None) -> Dict[str, Dict]: """ 快照文件系统，输出所有层级文件信息（不含目录） :param path: 路径 :param last_snapshot_time: 上次快照时间，用于增量快照 :param max_depth: 最大递归深度，避免过深遍历 + :param previous_snapshot: 上次完整快照，用于保留未变化目录并清理已删除文件 """ - files_info = {} + 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) + } + + def __remove_deleted_children(_fileitm: schemas.FileItem, + sub_files: List[schemas.FileItem]) -> None: + """ + 清理已确认遍历目录中不再存在的直接子项。 + 未变化的子目录仍保留旧基线，避免增量遍历将其误删。 + """ + 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] + if direct_child_path not in child_paths: + files_info.pop(old_file_path, None) def __snapshot_file(_fileitm: schemas.FileItem, current_depth: int = 0): """ @@ -288,15 +316,20 @@ class StorageBase(metaclass=ABCMeta): if current_depth >= max_depth: return - # 增量检查：如果目录修改时间早于上次快照，跳过 - if (self.snapshot_check_folder_modtime and + # 根目录每轮至少列举一次，用于清理已移走的直接子项；子目录仍按修改时间增量遍历 + 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.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: diff --git a/app/monitor/monitor.py b/app/monitor/monitor.py index 2c5335ab9..a4e67d81a 100644 --- a/app/monitor/monitor.py +++ b/app/monitor/monitor.py @@ -167,12 +167,18 @@ class Monitor(ConfigReloadMixin, metaclass=SingletonClass): # 远程目录监控 - 使用智能间隔 # 先尝试加载已有快照获取文件数量 snapshot_data = self._store.load(storage) - file_count = snapshot_data.get('file_count', 0) if snapshot_data else 0 + snapshot_is_current = ( + snapshot_data + and snapshot_data.get('version') == SnapshotStore.VERSION + ) + file_count = snapshot_data.get('file_count', 0) if snapshot_is_current else 0 interval = SnapshotStore.adjust_interval(file_count) for path in paths: logger.info(f"正在启动远程目录监控: {path} [{storage}]") logger.info("*** 重要提示：远程目录监控只处理新增和修改的文件，不会处理监控启动前已存在的文件 ***") - logger.info(f"预估文件数量: {file_count}, 监控间隔: {interval}分钟") + if snapshot_data and not snapshot_is_current: + logger.info(f"检测到旧版远程快照，将在首次轮询后重新校准: {storage}") + logger.info(f"上次快照文件数量: {file_count}, 监控间隔: {interval}分钟") self._scheduler.add_job( self.polling_observer, diff --git a/app/monitor/poller.py b/app/monitor/poller.py index b840a3893..3b1e0e8fd 100644 --- a/app/monitor/poller.py +++ b/app/monitor/poller.py @@ -1,5 +1,5 @@ import traceback -from pathlib import Path +from pathlib import Path, PurePosixPath from threading import Lock from typing import Callable, Dict, List, Optional @@ -64,6 +64,21 @@ class RemotePoller: self._alert_cb(storage, f"远程目录监控已恢复: {storage}") self._failure_counts[storage] = 0 + @staticmethod + def _snapshot_for_path(snapshot: Dict[str, Dict], mon_path: Path) -> Dict[str, Dict]: + """ + 提取指定监控目录范围内的快照。 + :param snapshot: 完整存储快照 + :param mon_path: 监控目录 + :return: 目录范围内的快照 + """ + root_path = PurePosixPath(mon_path.as_posix()) + return { + file_path: file_info + for file_path, file_info in snapshot.items() + if PurePosixPath(file_path).is_relative_to(root_path) + } + def poll(self, storage: str, mon_paths: List[Path]) -> Optional[int]: """ 执行一轮轮询监控。 @@ -83,7 +98,7 @@ class RemotePoller: last_snapshot_time = old_snapshot_data.get('timestamp', 0) if old_snapshot_data else 0 is_first_snapshot = old_snapshot_data is None - new_snapshot = {} + path_snapshots = [] failed_paths = [] for mon_path in mon_paths: logger.debug(f"开始对 {storage}:{mon_path} 进行快照...") @@ -92,40 +107,47 @@ class RemotePoller: snapshot = StorageChain().snapshot_storage( storage=storage, path=mon_path, - last_snapshot_time=last_snapshot_time + last_snapshot_time=last_snapshot_time, + previous_snapshot=old_snapshot ) if snapshot is None: - failed_paths.append(str(mon_path)) + failed_paths.append(mon_path) logger.warn(f"获取 {storage}:{mon_path} 快照失败") continue - new_snapshot.update(snapshot) + path_snapshots.append(snapshot) logger.info(f"{storage}:{mon_path} 快照完成，发现 {len(snapshot)} 个文件") if failed_paths and (is_first_snapshot or len(failed_paths) == len(mon_paths)): # 首次基线必须完整建立；全部路径失败时本轮没有有效数据，均不落盘 - self._note_failure(storage, f"快照失败: {','.join(failed_paths)}") + self._note_failure(storage, f"快照失败:{','.join(str(path) for path in failed_paths)}") return None - # 增量快照只包含变化子树，必须与基线合并才是完整视图； - # 直接把增量当基线会导致下一轮把未扫到的旧文件全部误判为新增 - merged_snapshot = {**old_snapshot, **new_snapshot} - file_count = len(merged_snapshot) + # 成功路径已在存储层完成增量对账；失败路径继续保留旧基线，避免临时故障丢失状态 + current_snapshot = {} + for failed_path in failed_paths: + current_snapshot.update(self._snapshot_for_path(old_snapshot, failed_path)) + for path_snapshot in path_snapshots: + current_snapshot.update(path_snapshot) + file_count = len(current_snapshot) if not is_first_snapshot: - self._handle_changes(storage, old_snapshot, new_snapshot) + self._handle_changes(storage, old_snapshot, current_snapshot) else: logger.info(f"{storage} 首次快照完成，共 {file_count} 个文件") logger.info("*** 首次快照仅建立基准，不会处理现有文件。后续监控将处理新增和修改的文件 ***") - # 保存合并后的基线 - if not self._store.save(storage, merged_snapshot, file_count, last_snapshot_time): + # 保存当前完整基线 + if not self._store.save(storage, current_snapshot, file_count, last_snapshot_time): self._note_failure(storage, "保存快照基线失败") return None if failed_paths: # 部分路径失败：成功路径已合并，失败路径保留旧基线，下轮重试 - self._note_failure(storage, f"部分路径快照失败: {','.join(failed_paths)}") + self._note_failure( + storage, + f"部分路径快照失败: {','.join(str(path) for path in failed_paths)}" + ) else: self._note_success(storage) return file_count diff --git a/app/monitor/snapshot.py b/app/monitor/snapshot.py index 0e21dda1b..3d66affc5 100644 --- a/app/monitor/snapshot.py +++ b/app/monitor/snapshot.py @@ -11,6 +11,7 @@ class SnapshotStore: """ 远程目录监控快照的存取与比对。 """ + VERSION = 2 def __init__(self, cache: Optional[FileCache] = None): """ @@ -30,10 +31,14 @@ class SnapshotStore: :return: 是否保存成功 """ try: - snapshot_time = max((item.get('modify_time', 0) for item in snapshot.values()), default=None) - if snapshot_time is None: - snapshot_time = last_snapshot_time or time.time() + snapshot_time = max( + last_snapshot_time or 0, + max((item.get('modify_time', 0) for item in snapshot.values()), default=0) + ) + if not snapshot_time: + snapshot_time = time.time() snapshot_data = { + 'version': self.VERSION, 'timestamp': snapshot_time, 'file_count': file_count, 'snapshot': snapshot diff --git a/tests/test_monitor_snapshot_semantics.py b/tests/test_monitor_snapshot_semantics.py index 516616b3a..fdfd9b2de 100644 --- a/tests/test_monitor_snapshot_semantics.py +++ b/tests/test_monitor_snapshot_semantics.py @@ -1,7 +1,11 @@ +import json from pathlib import Path from unittest.mock import MagicMock +from app import schemas +from app.modules.filemanager.storages import StorageBase from app.monitor.poller import RemotePoller +from app.monitor.snapshot import SnapshotStore from app.monitor.watcher import LocalDirectoryWatcher @@ -46,13 +50,17 @@ def test_poll_merges_incremental_into_baseline(monkeypatch): """ poller, store, dispatcher = _build_poller() store.load_checked.return_value = (dict(BASELINE), True) - _mock_storage_chain(monkeypatch, [{'/mon/b.mkv': {'size': 2, 'modify_time': 200}}]) + chain = _mock_storage_chain(monkeypatch, [{ + '/mon/a.mkv': {'size': 1, 'modify_time': 100}, + '/mon/b.mkv': {'size': 2, 'modify_time': 200} + }]) file_count = poller.poll("u115", [Path("/mon")]) assert file_count == 2 saved_snapshot = store.save.call_args.args[1] assert set(saved_snapshot.keys()) == {'/mon/a.mkv', '/mon/b.mkv'} + assert chain.snapshot_storage.call_args.kwargs["previous_snapshot"] == BASELINE["snapshot"] dispatcher.handle_file.assert_called_once() assert dispatcher.handle_file.call_args.kwargs["event_path"] == Path('/mon/b.mkv') @@ -73,6 +81,89 @@ def test_poll_detects_modified_files(monkeypatch): dispatcher.handle_file.assert_called_once() +def test_poll_removes_deleted_files_from_count(monkeypatch): + """ + 已移出监控目录的文件应从完整基线和动态间隔计数中移除。 + """ + poller, store, dispatcher = _build_poller() + store.load_checked.return_value = (dict(BASELINE), True) + _mock_storage_chain(monkeypatch, [{}]) + + assert poller.poll("alist", [Path("/mon")]) == 0 + assert store.save.call_args.args[1] == {} + assert store.save.call_args.args[2] == 0 + dispatcher.handle_file.assert_not_called() + + +def test_storage_snapshot_reconciles_deleted_children_and_keeps_skipped_subtree(): + """ + 增量遍历应删除已消失的直接子项，同时保留未变化子目录的旧基线。 + """ + storage = MagicMock() + storage.snapshot_check_folder_modtime = True + root = schemas.FileItem(storage="alist", type="dir", path="/mon/", name="mon", modify_time=200) + kept_dir = schemas.FileItem( + storage="alist", type="dir", path="/mon/keep/", name="keep", modify_time=50 + ) + storage.get_item.return_value = root + storage.list.return_value = [kept_dir] + previous_snapshot = { + '/mon/gone.mkv': {'size': 1, 'modify_time': 100}, + '/mon/keep/a.mkv': {'size': 2, 'modify_time': 50} + } + + snapshot = StorageBase.snapshot( + storage, + Path("/mon"), + last_snapshot_time=100, + previous_snapshot=previous_snapshot + ) + + assert snapshot == {'/mon/keep/a.mkv': {'size': 2, 'modify_time': 50}} + + +def test_storage_snapshot_always_lists_monitor_root_for_deletions(): + """ + 即使根目录修改时间未推进，也应列举其直接子项以清理已移走文件。 + """ + storage = MagicMock() + storage.snapshot_check_folder_modtime = True + storage.get_item.return_value = schemas.FileItem( + storage="alist", type="dir", path="/mon/", name="mon", modify_time=50 + ) + storage.list.return_value = [] + + snapshot = StorageBase.snapshot( + storage, + Path("/mon"), + last_snapshot_time=100, + previous_snapshot={'/mon/gone.mkv': {'size': 1, 'modify_time': 100}} + ) + + assert snapshot == {} + storage.list.assert_called_once() + + +def test_snapshot_store_marks_reconciled_format_and_keeps_cursor(): + """ + 新快照应标记对账格式，删除最新文件后游标也不能倒退。 + """ + cache = MagicMock() + store = SnapshotStore(cache=cache) + + assert store.save( + "alist", + {'/mon/a.mkv': {'size': 1, 'modify_time': 50}}, + file_count=1, + last_snapshot_time=100 + ) is True + + payload = json.loads(cache.set.call_args.args[1].decode("utf-8")) + assert payload["version"] == SnapshotStore.VERSION + assert payload["timestamp"] == 100 + assert payload["file_count"] == 1 + + def test_poll_partial_failure_merges_success_and_keeps_baseline(monkeypatch): """ 部分路径快照失败时，成功路径合并落盘，失败路径保留旧基线。