fix(monitor): 修正远程快照文件数量累计

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]:
         """

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]:
         """

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:

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,

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

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

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):
     """
     部分路径快照失败时，成功路径合并落盘，失败路径保留旧基线。