Merge pull request #6843 from cwd0204/fix/smb-snapshot-recovery

app/modules/filemanager/storages/__init__.py

@@ -51,6 +51,11 @@ class StorageBase(metaclass=ABCMeta):
         """返回存储快照目录时间检查的默认开关。"""
         return True
 
+    @property
+    def snapshot_strict_query(self) -> bool:
+        """声明快照是否使用严格查询；未适配的存储沿用原查询合同。"""
+        return False
+
     def __init__(self):
         self.storagehelper = StorageHelper()
 
@@ -164,6 +169,10 @@ class StorageBase(metaclass=ABCMeta):
         """
         pass
 
+    def list_strict(self, fileitem: _SchemaFileItem) -> List[_SchemaFileItem]:
+        """完整列举目录；空列表表示确认无子项，无法确认时抛出 StorageQueryError。"""
+        raise StorageQueryError(f"存储 {self.schema} 未实现严格目录查询: {fileitem.path}")
+
     @abstractmethod
     def create_folder(self, fileitem: _SchemaFileItem, name: str) -> Optional[_SchemaFileItem]:
         """
@@ -286,15 +295,34 @@ class StorageBase(metaclass=ABCMeta):
         """
         pass
 
+    def _snapshot_list(self, fileitem: _SchemaFileItem) -> Optional[List[_SchemaFileItem]]:
+        """按存储能力读取目录，严格快照不能把无结果当作可以跳过的子树。"""
+        if not self.snapshot_strict_query:
+            return self.list(fileitem)
+        items = self.list_strict(fileitem)
+        if items is None:
+            raise StorageQueryError(f"目录查询没有返回完整结果: {fileitem.path}")
+        return items
+
+    def _snapshot_read_error(self, fileitem: _SchemaFileItem, error: Exception) -> None:
+        """严格快照传播读取故障；未适配的存储保留原来的子树容错行为。"""
+        if self.snapshot_strict_query:
+            if isinstance(error, StorageQueryError):
+                raise error
+            raise StorageQueryError(f"读取快照路径失败: {fileitem.path} - {error}") from error
+        logger.debug(f"Snapshot error for {fileitem.path}: {error}")
+
     def snapshot(self, path: Path, last_snapshot_time: float = None, max_depth: int = 5,
-                 previous_snapshot: Optional[Dict[str, Dict]] = None) -> Dict[str, Dict]:
+                 previous_snapshot: Optional[Dict[str, Dict]] = None) -> Optional[Dict[str, Dict]]:
         """
         快照文件系统，输出所有层级文件信息（不含目录）
         :param path: 路径
         :param last_snapshot_time: 上次快照时间，用于增量快照
         :param max_depth: 最大递归深度，避免过深遍历
         :param previous_snapshot: 上次完整快照，用于保留未变化目录并清理已删除文件
+        :return: 有效快照（允许为空）；严格查询失败返回 None，调用方必须保留旧基线
         """
+        strict_query = self.snapshot_strict_query
         root_path = PurePosixPath(path.as_posix())
         files_info = {
             file_path: file_info
@@ -340,7 +368,7 @@ class StorageBase(metaclass=ABCMeta):
                         return
 
                     # 只有目录列表成功返回后才清理旧基线，查询异常时继续保留待下轮重试
-                    sub_files = self.list(_fileitm)
+                    sub_files = self._snapshot_list(_fileitm)
                     if sub_files is None:
                         return
                     sub_files = list(sub_files)
@@ -357,12 +385,15 @@ class StorageBase(metaclass=ABCMeta):
                     }
 
             except Exception as e:
-                logger.debug(f"Snapshot error for {_fileitm.path}: {e}")
+                self._snapshot_read_error(_fileitm, e)
 
-        fileitem = self.get_item(path)
-        if not fileitem:
-            return {}
-
-        __snapshot_file(fileitem)
+        try:
+            fileitem = self.get_item_strict(path) if strict_query else self.get_item(path)
+            if not fileitem:
+                return {}
+            __snapshot_file(fileitem)
+        except StorageQueryError as err:
+            logger.warning(f"存储快照失败，保留上次基线: {self.schema}:{path} - {err}")
+            return None
 
         return files_info

app/modules/filemanager/storages/smb.py

@@ -47,6 +47,16 @@ class SMB(StorageBase, metaclass=WeakSingleton):
     # 文件块大小，默认10MB
     chunk_size = 10 * 1024 * 1024
 
+    @property
+    def snapshot_check_folder_modtime(self) -> bool:
+        """SMB 目录时间不保证传播后代变化，快照必须遍历到配置的深度上限。"""
+        return False
+
+    @property
+    def snapshot_strict_query(self) -> bool:
+        """SMB 快照必须区分真实空目录与读取失败，防止错误更新监控基线。"""
+        return True
+
     def __init__(self):
         """加载当前 SMB 配置并尝试建立连接。"""
         super().__init__()
@@ -202,10 +212,10 @@ class SMB(StorageBase, metaclass=WeakSingleton):
         return "/" + file_path[len(prefix):].replace("\\", "/")
 
     def _create_fileitem(
-        self, stat_result, file_path: str, name: str
+        self, stat_result, file_path: str, name: str, *, strict: bool = False
     ) -> _SchemaFileItem:
         """
-        创建文件项
+        创建文件项；严格查询不能把元数据读取失败回退成普通文件。
         """
         import smbclient
         try:
@@ -247,6 +257,8 @@ class SMB(StorageBase, metaclass=WeakSingleton):
                     modify_time=modify_time,
                 )
         except Exception as e:
+            if strict:
+                raise StorageQueryError(f"【SMB】创建文件项失败: {file_path} - {e}") from e
             logger.error(f"【SMB】创建文件项失败：{e}")
             # 返回基本的文件项信息
             return _SchemaFileItem(
@@ -282,9 +294,18 @@ class SMB(StorageBase, metaclass=WeakSingleton):
             self._connected = False
             return False
 
-    def list(self, fileitem: _SchemaFileItem) -> List[_SchemaFileItem]:
+    def list_strict(self, fileitem: _SchemaFileItem) -> List[_SchemaFileItem]:
+        """用于快照的完整目录查询，任一子项读取失败都拒绝返回部分结果。"""
+        try:
+            return self.list(fileitem, strict=True)
+        except StorageQueryError:
+            raise
+        except Exception as err:
+            raise StorageQueryError(f"【SMB】查询目录失败: {fileitem.path} - {err}") from err
+
+    def list(self, fileitem: _SchemaFileItem, *, strict: bool = False) -> List[_SchemaFileItem]:
         """
-        浏览文件
+        浏览文件；strict 供快照查询传播异常，普通浏览沿用容错回退。
         """
         import smbclient
         from smbprotocol.exceptions import SMBException, SMBResponseException
@@ -292,7 +313,9 @@ class SMB(StorageBase, metaclass=WeakSingleton):
             self._check_connection()
 
             if fileitem.type == "file":
-                item = self.detail(fileitem)
+                if not fileitem.path:
+                    raise StorageQueryError("【SMB】文件查询缺少路径")
+                item = self.get_item_strict(Path(fileitem.path)) if strict else self.detail(fileitem)
                 if item:
                     return [item]
                 return []
@@ -310,11 +333,10 @@ class SMB(StorageBase, metaclass=WeakSingleton):
             # 列出目录内容
             try:
                 entries = smbclient.listdir(smb_path)
-            except SMBResponseException as e:
-                logger.error(f"【SMB】列出目录失败: {smb_path} - {e}")
-                return []
-            except SMBException as e:
+            except (SMBResponseException, SMBException) as e:
                 logger.error(f"【SMB】列出目录失败: {smb_path} - {e}")
+                if strict:
+                    raise
                 return []
 
             items = []
@@ -325,14 +347,18 @@ class SMB(StorageBase, metaclass=WeakSingleton):
                 entry_path = f"{smb_path}\\{entry}"
                 try:
                     stat_result = smbclient.stat(entry_path)
-                    item = self._create_fileitem(stat_result, entry_path, entry)
+                    item = self._create_fileitem(stat_result, entry_path, entry, strict=strict)
                     items.append(item)
                 except Exception as e:
+                    if strict:
+                        raise
                     logger.debug(f"【SMB】获取文件信息失败: {entry_path} - {e}")
                     continue
 
             return items
         except Exception as e:
+            if strict:
+                raise
             logger.error(f"【SMB】列出文件失败: {e}")
             return []
 
@@ -453,7 +479,7 @@ class SMB(StorageBase, metaclass=WeakSingleton):
                 if err.errno in (errno.ENOENT, errno.ENOTDIR):
                     return None
                 raise StorageQueryError(f"【SMB】查询文件项失败: {path} - {err}") from err
-            return self._create_fileitem(stat_result, smb_path, Path(path).name)
+            return self._create_fileitem(stat_result, smb_path, Path(path).name, strict=True)
         except StorageQueryError:
             raise
         except Exception as e:

docs/rules/05-architecture.md

@@ -717,6 +717,12 @@ SMB 同存储整理由 `app.modules.filemanager.storages.smb` 持有协议调用
 下载器返回的远程路径保留存储 URI，整理 Chain 通过 `StorageChain` 查询源文件，
 不直接依赖 SMB SDK；断线后目标状态不明沿用持久步骤的保守恢复合同。
 
+SMB 目录快照不依赖祖先目录 mtime 推断后代变化，而在既有递归深度上限内逐层列举。
+存储通过 `snapshot_strict_query` 声明已实现 `get_item_strict` / `list_strict`；
+严格快照的任一查询失败必须返回 `None`，由远程轮询器保留旧基线并重试。
+只有确认目录清空或根路径不存在时才返回空字典。SMB 普通文件浏览继续允许容错回退，
+严格快照不能复用其部分结果；其他存储需按各自查询合同逐步接入。
+
 `app.modules.filemanager` is a lazy compatibility entrypoint. The concrete
 `FileManagerModule` implementation lives in `app.modules.filemanager.module`,
 while the historical capability path and class module identity remain

docs/testing.md

@@ -82,6 +82,23 @@ with patch.object(SomeModule, "fetch", new=AsyncMock(return_value=FAKE)):
 
 > 注意：识别这类端到端流程往往不止一个外部出口。例如 TMDB 识别除了目录请求，链路层还会向 MP 服务器上报/查询「共享识别 API」——这类旁路出口必须一并打桩。用下文的 socket 探针确认确实零出站。
 
+### 存储与监控的边界回归
+
+存储快照用例优先在 SDK 的目录列表、stat、类型查询和连接边界打桩，保留真实的路径映射、
+文件项构造、快照遍历及轮询逻辑。仅替换 `snapshot_storage` 的返回值可以验证轮询器，
+但不能证明存储适配器正确区分了空目录与查询失败。
+
+`tests/test_smb_storage.py` 提供以下离线场景：
+
+- 单共享和多共享下，祖先目录时间不变时仍发现后代新增、修改、保留旧 mtime 的文件移入和删除。
+- 根目录、子目录、单个文件或类型查询失败时，不提交空快照或部分快照。
+- 连接失效保留旧基线；恢复后发现新文件，再次轮询不重复派发。
+- 真正清空的目录和确认不存在的根路径可以提交空快照。
+- 完整遍历仍遵守递归深度上限，不因本次修复增加扫描频率。
+
+这些场景只承诺已验证的 SMB 查询语义；为其他存储接入严格快照时，应在其 SDK/API
+边界复现同样的状态转换，并评估目录时间语义、分页和请求成本。
+
 ## 自隔离：用了什么，就还原什么
 
 用例若修改了**进程级状态**——`sys.modules` 桩、单例（`Singleton._instances`）、`lru_cache`、环境变量、`settings` 字段——必须在用例或模块结束时还原。pytest 一次性导入全部测试模块，未还原的污染会扩散到后续用例，产生“单独跑过、一起跑挂”的测不准现象。

tests/test_monitor_snapshot_semantics.py

@@ -136,6 +136,8 @@ def test_storage_snapshot_reconciles_deleted_children_and_keeps_skipped_subtree(
     """
     storage = MagicMock()
     storage.snapshot_check_folder_modtime = True
+    storage.snapshot_strict_query = False
+    storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item)
     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
@@ -163,6 +165,8 @@ def test_storage_snapshot_always_lists_monitor_root_for_deletions():
     """
     storage = MagicMock()
     storage.snapshot_check_folder_modtime = True
+    storage.snapshot_strict_query = False
+    storage._snapshot_list = lambda item: StorageBase._snapshot_list(storage, item)
     storage.get_item.return_value = schemas.FileItem(
         storage="alist", type="dir", path="/mon/", name="mon", modify_time=50
     )

tests/test_smb_storage.py

@@ -1,4 +1,4 @@
-"""SMB 保存重连和同一共享内服务端整理的回归测试。"""
+"""SMB 保存重连、服务端整理与远程快照恢复的回归测试。"""
 
 import errno
 from pathlib import Path
@@ -12,6 +12,8 @@ from app.application.storage import StorageHelper
 from app.modules.filemanager.module import FileManagerModule
 from app.modules.filemanager.storages import smb as smb_module
 from app.modules.filemanager.transhandler import TransHandler
+from app.monitor.poller import RemotePoller
+from app.monitor.snapshot import SnapshotStore
 from app.schemas.file import FileItem
 
 
@@ -332,3 +334,175 @@ def test_multi_share_usage_query_failure_remains_unavailable(multi_share, monkey
     monkeypatch.setattr(smbclient, "stat_volume", stat_volume)
     assert multi_share.storage.usage() is None
     stat_volume.assert_called_once_with(r"\\10.10.10.11\video")
+
+
+@pytest.fixture
+def snapshot_server(smb_storage, monkeypatch):
+    """仅替换 SMB SDK 的文件系统边界，保留真实目录映射、文件项构造和快照遍历。"""
+    storage = smb_storage.storage
+    smb_storage.state["conf"] = {"host": "smb.example", "share": "media"}
+    storage.init_storage()
+    tree = {
+        "/downloads": {"directory": True, "mtime": 50},
+        "/downloads/Series": {"directory": True, "mtime": 50},
+        "/downloads/Series/Season01": {"directory": True, "mtime": 50},
+        "/downloads/Series/Season01/old.mkv": {"directory": False, "mtime": 100, "size": 10},
+    }
+    fault = {"operation": None, "path": None}
+
+    def relative(path):
+        return storage._relative_path(path).rstrip("/")
+
+    def read_node(operation, path):
+        key = relative(path)
+        if fault == {"operation": operation, "path": key}:
+            raise OSError(errno.EIO, "temporary SMB read failure")
+        if key not in tree:
+            raise FileNotFoundError(errno.ENOENT, "missing")
+        return key, tree[key]
+
+    def list_directory(path):
+        key, _node = read_node("list", path)
+        return [Path(item).name for item in tree if Path(item).parent.as_posix() == key]
+
+    def stat_item(path):
+        _key, node = read_node("stat", path)
+        return SimpleNamespace(st_mtime=node["mtime"], st_size=node.get("size", 0))
+
+    def is_directory(path):
+        _key, node = read_node("type", path)
+        return node["directory"]
+
+    smb_storage.client.listdir.side_effect = list_directory
+    monkeypatch.setattr(smbclient, "stat", Mock(side_effect=stat_item))
+    monkeypatch.setattr(smbclient.path, "isdir", Mock(side_effect=is_directory))
+    monkeypatch.setattr(smbclient.path, "exists", Mock(side_effect=lambda path: relative(path) in tree))
+    return SimpleNamespace(
+        storage=storage, module=smb_storage.module, client=smb_storage.client,
+        tree=tree, fault=fault, root=Path("/downloads"),
+    )
+
+
+@pytest.mark.parametrize("multiple_shares", [False, True])
+@pytest.mark.parametrize("change", ["added", "modified", "moved_old", "deleted"])
+def test_snapshot_detects_descendant_changes_without_parent_mtime(snapshot_server, multiple_shares, change):
+    """SMB 祖先目录时间不变时，单共享与多共享仍能发现后代新增、修改、旧文件移入和删除。"""
+    state = snapshot_server
+    if multiple_shares:
+        state.storage._configure_shares({"shares": ["downloads", "library"]})
+    old_path = "/downloads/Series/Season01/old.mkv"
+    new_path = "/downloads/Series/Season01/new.mkv"
+    baseline = state.storage.snapshot(state.root)
+    assert set(baseline) == {old_path}
+
+    if change in {"added", "moved_old"}:
+        state.tree[new_path] = {
+            "directory": False, "mtime": 200 if change == "added" else 1, "size": 20,
+        }
+    elif change == "modified":
+        state.tree[old_path].update(size=20, mtime=200)
+    else:
+        del state.tree[old_path]
+
+    snapshot = state.module.snapshot_storage(
+        "smb", state.root, last_snapshot_time=100, previous_snapshot=baseline,
+    )
+    assert snapshot is not None
+    changes = SnapshotStore.compare(baseline, snapshot)
+    expected = {"added": [], "modified": []}
+    if change == "deleted":
+        assert old_path not in snapshot
+    else:
+        expected["added" if change == "moved_old" else change] = [
+            new_path if change in {"added", "moved_old"} else old_path
+        ]
+    assert changes == expected
+    assert baseline[old_path]["size"] == 10
+
+
+@pytest.mark.parametrize(
+    ("operation", "path"),
+    [
+        ("stat", "/downloads"),
+        ("list", "/downloads"),
+        ("list", "/downloads/Series/Season01"),
+        ("stat", "/downloads/Series/Season01/old.mkv"),
+        ("type", "/downloads/Series"),
+    ],
+)
+def test_snapshot_query_failure_is_not_an_empty_or_partial_tree(snapshot_server, operation, path):
+    """根目录、子目录或文件元数据读取失败时，整个监控路径都不能提交为有效快照。"""
+    state = snapshot_server
+    baseline = state.storage.snapshot(state.root)
+    state.fault.update(operation=operation, path=path)
+
+    assert state.module.snapshot_storage(
+        "smb", state.root, last_snapshot_time=0, previous_snapshot=baseline,
+    ) is None
+    assert len(baseline) == 1
+
+
+def test_disconnected_snapshot_preserves_failure_semantics(snapshot_server):
+    """连接失效不能冒充目录已删除；普通文件浏览仍沿用空列表回退。"""
+    state = snapshot_server
+    state.storage._connected = False
+
+    assert state.module.snapshot_storage("smb", state.root) is None
+    assert state.storage.list(FileItem(storage="smb", type="dir", path="/downloads")) == []
+
+
+@pytest.mark.parametrize("root_missing", [False, True])
+def test_confirmed_empty_snapshot_is_success(snapshot_server, root_missing):
+    """确认目录已清空或根目录已删除时应返回空快照，而不是失败。"""
+    state = snapshot_server
+    baseline = state.storage.snapshot(state.root)
+    if root_missing:
+        state.tree.clear()
+    else:
+        del state.tree["/downloads/Series/Season01/old.mkv"]
+
+    assert state.module.snapshot_storage(
+        "smb", state.root, last_snapshot_time=100, previous_snapshot=baseline,
+    ) == {}
+
+
+def test_snapshot_keeps_depth_limit(snapshot_server):
+    """遍历不因 mtime 剪枝，但仍遵守调用方的递归深度预算。"""
+    state = snapshot_server
+    state.client.listdir.reset_mock()
+
+    assert state.storage.snapshot(state.root, max_depth=2) == {}
+    assert state.client.listdir.call_count == 2
+
+
+def test_poll_retries_failed_smb_scan_without_losing_baseline(snapshot_server, monkeypatch):
+    """SMB 断线时保留旧基线，恢复后发现保留旧 mtime 的移入文件且只派发一次。"""
+    state = snapshot_server
+    records = {}
+    cache = Mock()
+    cache.get.side_effect = lambda key, **_kwargs: records.get(key)
+    cache.set.side_effect = lambda key, value, **_kwargs: records.__setitem__(key, value)
+    store = SnapshotStore(cache=cache)
+    baseline = state.storage.snapshot(state.root)
+    assert store.save("smb", baseline, file_count=1, last_snapshot_time=100)
+    saved_baseline = store.load("smb")
+    dispatcher = Mock()
+    dispatcher.is_transfer_candidate_path.return_value = True
+    dispatcher.handle_file.return_value = True
+    chain = SimpleNamespace(snapshot_storage=state.module.snapshot_storage)
+    monkeypatch.setattr("app.monitor.poller.StorageChain", lambda: chain)
+    poller = RemotePoller(store=store, dispatcher=dispatcher)
+    new_path = "/downloads/Series/Season01/new.mkv"
+    state.tree[new_path] = {"directory": False, "mtime": 1, "size": 20}
+    state.fault.update(operation="list", path="/downloads/Series/Season01")
+
+    assert poller.poll("smb", [state.root]) is None
+    assert store.load("smb") == saved_baseline
+    dispatcher.handle_file.assert_not_called()
+
+    state.fault.update(operation=None, path=None)
+    assert poller.poll("smb", [state.root]) == 2
+    dispatcher.handle_file.assert_called_once()
+    assert dispatcher.handle_file.call_args.kwargs["event_path"] == Path(new_path)
+    assert poller.poll("smb", [state.root]) == 2
+    dispatcher.handle_file.assert_called_once()

tests/test_storage_strict_query_coverage.py

@@ -19,6 +19,14 @@ def test_base_get_item_strict_fails_conservatively():
         StorageBase.get_item_strict(stub, Path("/media/示例.mkv"))
 
 
+def test_base_list_strict_fails_conservatively():
+    """未适配严格目录查询的存储不能把宽松查询的空结果当作完整快照。"""
+    stub = SimpleNamespace(schema="dummy")
+
+    with pytest.raises(StorageQueryError):
+        StorageBase.list_strict(stub, SimpleNamespace(path="/media"))
+
+
 @pytest.mark.parametrize("module_name", BUILTIN_STORAGE_MODULES)
 def test_builtin_storage_overrides_strict_query(module_name):
     """每个在树存储都必须覆写严格查询，否则整理会被基类保守拒绝。"""