diff --git a/app/chain/transfer.py b/app/chain/transfer.py index ad0568dd2..b6cb9de20 100755 --- a/app/chain/transfer.py +++ b/app/chain/transfer.py @@ -1565,7 +1565,9 @@ class TransferChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): logger.info(__end_msg) self._progress.update(value=100, text=__end_msg) self._progress.end() - # 重置计数 + # 重置计数，_total_num 一并归零，否则会作为历史最大值一直 + # 累积，令后续批次的「当前共 N 个文件」与进度百分比失真 + self._total_num = 0 self._processed_num = 0 self._fail_num = 0 diff --git a/app/modules/filemanager/storages/__init__.py b/app/modules/filemanager/storages/__init__.py index 5c495484c..83edc8440 100644 --- a/app/modules/filemanager/storages/__init__.py +++ b/app/modules/filemanager/storages/__init__.py @@ -174,6 +174,13 @@ class StorageBase(metaclass=ABCMeta): """ pass + def get_item_strict(self, path: Path) -> Optional[schemas.FileItem]: + """ + 获取文件或目录，确认不存在返回None；无法确认状态时抛出 StorageQueryError。 + 默认实现不区分「不存在」与「查询失败」，由具体存储按需覆写。 + """ + return self.get_item(path) + def get_parent(self, fileitem: schemas.FileItem) -> Optional[schemas.FileItem]: """ 获取父目录 diff --git a/app/modules/filemanager/storages/alipan.py b/app/modules/filemanager/storages/alipan.py index 7fae228ae..d468af4e6 100644 --- a/app/modules/filemanager/storages/alipan.py +++ b/app/modules/filemanager/storages/alipan.py @@ -13,6 +13,7 @@ from app.core.config import settings, global_vars from app.log import logger from app.modules.filemanager import StorageBase from app.modules.filemanager.storages import transfer_process +from app.schemas.exception import StorageQueryError from app.schemas.types import StorageSchema from app.utils.http import RequestUtils from app.utils.singleton import WeakSingleton @@ -834,30 +835,53 @@ class AliPan(StorageBase, metaclass=WeakSingleton): return False return True + def __get_by_path_item(self, path: Path, drive_id: str = None) -> Optional[schemas.FileItem]: + """ + 按路径查询文件/目录项，无法确认状态时抛出 StorageQueryError。 + NotFound 系列错误码表示确认不存在，其余错误（网络失败、限流、 + 权限或未知业务错误）均无法确认目标状态。 + """ + resp = self._request_api( + "POST", + "/adrive/v1.0/openFile/get_by_path", + json={ + "drive_id": drive_id or self._default_drive_id, + "file_path": path.as_posix(), + }, + no_error_log=True, + ) + if resp is None: + raise StorageQueryError(f"【阿里云盘】无法确认文件状态（请求失败）: {path}") + code = resp.get("code") + if code: + if "NotFound" in str(code): + # 明确的不存在错误码，确认目标不存在 + return None + raise StorageQueryError( + f"【阿里云盘】查询文件信息出错: {path} - {code} {resp.get('message')}") + return self.__get_fileitem(resp, parent=str(path.parent)) + def get_item(self, path: Path, drive_id: str = None) -> Optional[schemas.FileItem]: """ 获取指定路径的文件/目录项 """ try: - resp = self._request_api( - "POST", - "/adrive/v1.0/openFile/get_by_path", - json={ - "drive_id": drive_id or self._default_drive_id, - "file_path": path.as_posix(), - }, - no_error_log=True, - ) - if not resp: - return None - if resp.get("code"): - logger.debug(f"【阿里云盘】获取文件信息失败: {resp.get('message')}") - return None - return self.__get_fileitem(resp, parent=str(path.parent)) + return self.__get_by_path_item(path, drive_id=drive_id) except Exception as e: logger.debug(f"【阿里云盘】获取文件信息失败: {str(e)}") return None + def get_item_strict(self, path: Path) -> Optional[schemas.FileItem]: + """ + 获取指定路径的文件/目录项，无法确认状态时抛出 StorageQueryError。 + """ + try: + return self.__get_by_path_item(path) + except StorageQueryError: + raise + except Exception as e: + raise StorageQueryError(f"【阿里云盘】查询文件信息失败: {path} - {e}") from e + def get_folder(self, path: Path) -> Optional[schemas.FileItem]: """ 获取指定路径的文件夹，如不存在则创建 diff --git a/app/modules/filemanager/storages/local.py b/app/modules/filemanager/storages/local.py index 872744730..cf3c0ec80 100644 --- a/app/modules/filemanager/storages/local.py +++ b/app/modules/filemanager/storages/local.py @@ -8,6 +8,7 @@ from app.core.config import global_vars, settings from app.helper.directory import DirectoryHelper from app.log import logger from app.modules.filemanager.storages import StorageBase, transfer_process +from app.schemas.exception import StorageQueryError from app.schemas.types import StorageSchema from app.utils.system import SystemUtils @@ -148,6 +149,23 @@ class LocalStorage(StorageBase): return self.__get_fileitem(path) return self.__get_diritem(path) + def get_item_strict(self, path: Path) -> Optional[schemas.FileItem]: + """ + 获取文件或目录，无法确认状态时抛出 StorageQueryError。 + Path.exists() 会把部分 errno（如 EBADF/ELOOP）归入「不存在」， + 网络/FUSE 挂载抖动时会误判，这里用 stat 显式区分。 + """ + try: + path.stat() + except (FileNotFoundError, NotADirectoryError): + return None + except OSError as e: + raise StorageQueryError(f"【本地】读取文件状态失败: {path} - {e}") from e + try: + return self.get_item(path) + except OSError as e: + raise StorageQueryError(f"【本地】读取文件信息失败: {path} - {e}") from e + def detail(self, fileitem: schemas.FileItem) -> Optional[schemas.FileItem]: """ 获取文件详情 diff --git a/app/modules/filemanager/storages/u115.py b/app/modules/filemanager/storages/u115.py index 8ecd2be27..fb3c925ab 100644 --- a/app/modules/filemanager/storages/u115.py +++ b/app/modules/filemanager/storages/u115.py @@ -17,6 +17,7 @@ from app.core.config import settings, global_vars from app.log import logger from app.modules.filemanager import StorageBase from app.modules.filemanager.storages import transfer_process +from app.schemas.exception import StorageQueryError from app.schemas.types import StorageSchema from app.utils.singleton import WeakSingleton from app.utils.string import StringUtils @@ -906,38 +907,60 @@ class U115Pan(StorageBase, metaclass=WeakSingleton): return True return False + def __get_info_item(self, path: Path) -> Optional[schemas.FileItem]: + """ + 查询指定路径的文件/目录项，无法确认状态时抛出 StorageQueryError。 + 接口业务码 20004（记录不存在）与 0 一样视为确认结果，其余错误 + （网络失败、限流重试用尽、未知业务错误）均无法确认目标状态。 + """ + resp = self._request_api( + "POST", + "/open/folder/get_info", + data={"path": path.as_posix()}, + no_error_log=True, + ) + if resp is None: + raise StorageQueryError(f"【115】无法确认文件状态（请求失败或接口错误）: {path}") + data = resp.get("data") if isinstance(resp, dict) else None + if not data or not data.get("file_id"): + # code 20004（记录不存在）等场景，确认目标不存在 + return None + return schemas.FileItem( + storage=self.schema.value, + fileid=str(data["file_id"]), + path=path.as_posix() + ("/" if data["file_category"] == "0" else ""), + type="file" if data["file_category"] == "1" else "dir", + name=data["file_name"], + basename=Path(data["file_name"]).stem, + extension=Path(data["file_name"]).suffix[1:] + if data["file_category"] == "1" + else None, + pickcode=data["pick_code"], + size=data["size_byte"] if data["file_category"] == "1" else None, + modify_time=data["utime"], + ) + def get_item(self, path: Path) -> Optional[schemas.FileItem]: """ 获取指定路径的文件/目录项 """ try: - resp = self._request_api( - "POST", - "/open/folder/get_info", - "data", - data={"path": path.as_posix()}, - no_error_log=True, - ) - if not resp: - return None - return schemas.FileItem( - storage=self.schema.value, - fileid=str(resp["file_id"]), - path=path.as_posix() + ("/" if resp["file_category"] == "0" else ""), - type="file" if resp["file_category"] == "1" else "dir", - name=resp["file_name"], - basename=Path(resp["file_name"]).stem, - extension=Path(resp["file_name"]).suffix[1:] - if resp["file_category"] == "1" - else None, - pickcode=resp["pick_code"], - size=resp["size_byte"] if resp["file_category"] == "1" else None, - modify_time=resp["utime"], - ) + return self.__get_info_item(path) except Exception as e: logger.debug(f"【115】获取文件信息失败: {str(e)}") return None + def get_item_strict(self, path: Path) -> Optional[schemas.FileItem]: + """ + 获取指定路径的文件/目录项，无法确认状态时抛出 StorageQueryError。 + """ + try: + return self.__get_info_item(path) + except StorageQueryError: + raise + except Exception as e: + raise StorageQueryError(f"【115】查询文件信息失败: {path} - {e}") from e + def get_folder(self, path: Path) -> Optional[schemas.FileItem]: """ 获取指定路径的文件夹，如不存在则创建 diff --git a/app/modules/filemanager/transhandler.py b/app/modules/filemanager/transhandler.py index 95803d400..c78d5eca7 100644 --- a/app/modules/filemanager/transhandler.py +++ b/app/modules/filemanager/transhandler.py @@ -23,6 +23,7 @@ from app.schemas import ( TransferRenameBuildEventData, TransferRenameEventData, ) +from app.schemas.exception import StorageQueryError from app.schemas.types import MediaType, ChainEventType from app.utils.system import SystemUtils @@ -405,8 +406,23 @@ class TransHandler: # 判断是否要覆盖，附加文件强制覆盖 overflag = False if not __is_extra_file(fileitem): - # 目标文件 - target_item = target_oper.get_item(new_file) + # 目标文件（严格查询：无法确认状态时拒绝覆盖，避免已有文件被误覆盖） + try: + target_item = target_oper.get_item_strict(new_file) + except StorageQueryError as query_err: + errmsg = f"无法确认目标文件状态，已跳过整理以避免误覆盖：{new_file} - {query_err}" + logger.warn(errmsg) + self.__update_result( + result=result, + success=False, + message=errmsg, + fileitem=fileitem, + target_diritem=target_diritem, + fail_list=[fileitem.path], + transfer_type=transfer_type, + need_notify=need_notify, + ) + return result if target_item: # 目标文件已存在 target_file = new_file diff --git a/app/monitor.py b/app/monitor.py deleted file mode 100644 index b9282067b..000000000 --- a/app/monitor.py +++ /dev/null @@ -1,954 +0,0 @@ -import json -import platform -import re -import threading -import time -import traceback -from dataclasses import dataclass -from pathlib import Path -from threading import Lock -from typing import Any, Optional, Dict, List - -from apscheduler.schedulers.background import BackgroundScheduler -from watchfiles import Change, DefaultFilter, watch - -from app.chain import ChainBase -from app.chain.storage import StorageChain -from app.chain.transfer import TransferChain -from app.core.cache import TTLCache, FileCache -from app.core.config import settings -from app.db.transferhistory_oper import TransferHistoryOper -from app.helper.directory import DirectoryHelper -from app.helper.message import MessageHelper -from app.log import logger -from app.schemas import FileItem -from app.schemas.types import SystemConfigKey -from app.utils.mixins import ConfigReloadMixin -from app.utils.singleton import SingletonClass -from app.utils.system import SystemUtils - -lock = Lock() -snapshot_lock = Lock() - - -class MonitorChain(ChainBase): - pass - - -@dataclass(frozen=True) -class DirectoryChangeEvent: - """ - 目录文件变化事件，隔离底层 watchfiles 事件结构。 - """ - change_type: Change - src_path: str - is_directory: bool - - -class LocalDirectoryWatcher: - """ - 基于 watchfiles 的本地目录监控线程。 - """ - _HANDLE_CHANGES = {Change.added, Change.modified} - - def __init__(self, mon_path: Path, callback: Any, force_polling: Optional[bool] = None): - """ - 初始化本地目录监控。 - :param mon_path: 监控目录 - :param callback: 目录变化回调对象 - :param force_polling: 是否强制使用轮询模式，None 表示由 watchfiles 自动选择 - """ - self._watch_path = mon_path - self._callback = callback - self._force_polling = force_polling - self._stop_event = threading.Event() - self._thread: Optional[threading.Thread] = None - self._watch_filter = DefaultFilter() - - @property - def watch_path(self) -> Path: - """ - 获取监控目录。 - :return: 监控目录 - """ - return self._watch_path - - def start(self): - """ - 启动本地目录监控线程。 - """ - if not self._watch_path.exists(): - raise FileNotFoundError(f"监控目录不存在: {self._watch_path}") - if not self._watch_path.is_dir(): - raise NotADirectoryError(f"监控路径不是目录: {self._watch_path}") - if self.is_alive(): - logger.info(f"本地目录监控已在运行中: {self._watch_path}") - return - self._stop_event.clear() - self._thread = threading.Thread( - target=self._run, - name=f"MoviePilot-DirectoryWatcher-{self._watch_path.name}", - daemon=True - ) - self._thread.start() - - def stop(self): - """ - 请求停止本地目录监控线程。 - """ - self._stop_event.set() - - def join(self, timeout: Optional[float] = None): - """ - 等待本地目录监控线程退出。 - :param timeout: 最长等待秒数 - """ - if self._thread: - self._thread.join(timeout=timeout) - - def is_alive(self) -> bool: - """ - 判断监控线程是否仍在运行。 - :return: 线程存活状态 - """ - return bool(self._thread and self._thread.is_alive()) - - def _run(self): - """ - 运行 watchfiles 主循环，并在快速模式不可用时回退到轮询。 - """ - try: - self._run_watch(force_polling=self._force_polling) - except Exception as err: - if self._stop_event.is_set(): - return - if self._force_polling is True: - logger.error(f"本地目录监控发生错误: {self._watch_path} - {err}") - logger.debug(traceback.format_exc()) - return - logger.warn(f"快速模式监控 {self._watch_path} 失败，将自动切换到兼容模式: {err}") - try: - self._run_watch(force_polling=True) - except Exception as fallback_err: - if not self._stop_event.is_set(): - logger.error(f"兼容模式监控 {self._watch_path} 仍然失败: {fallback_err}") - logger.debug(traceback.format_exc()) - - def _run_watch(self, force_polling: Optional[bool]): - """ - 执行一次 watchfiles 监控循环。 - :param force_polling: 是否强制轮询 - """ - for changes in watch( - str(self._watch_path), - watch_filter=self._watch_filter, - stop_event=self._stop_event, - rust_timeout=1000, - yield_on_timeout=True, - force_polling=force_polling, - recursive=True, - ignore_permission_denied=True): - if self._stop_event.is_set(): - break - if not changes: - continue - self._handle_changes(changes) - - def _handle_changes(self, changes: set[tuple[Change, str]]): - """ - 将 watchfiles 原始变更转换为目录监控事件。 - :param changes: watchfiles 返回的变更集合 - """ - changes = self._expand_added_directories(changes) - for change_type, path_str in sorted(changes, key=lambda item: item[1]): - if change_type not in self._HANDLE_CHANGES: - continue - event_path = Path(path_str) - event = self._build_event(change_type=change_type, event_path=event_path) - if not event or event.is_directory: - continue - file_size = self._get_file_size(event_path) - if file_size is None: - continue - text = self._change_text(change_type) - try: - self._callback.event_handler( - event=event, - text=text, - event_path=path_str, - file_size=file_size - ) - except Exception as err: - logger.error(f"处理本地目录监控事件失败: {path_str} - {err}") - - def _expand_added_directories(self, changes: set[tuple[Change, str]]) -> set[tuple[Change, str]]: - """ - 将整体移入监控范围的新增目录展开为内部文件事件。 - :param changes: watchfiles 返回的变更集合 - :return: 包含目录内新增文件的变更集合 - """ - expanded_changes = set(changes) - for change_type, path_str in changes: - if change_type != Change.added: - continue - event_path = Path(path_str) - try: - if not event_path.is_dir(): - continue - for nested_path in event_path.rglob("*"): - if not nested_path.is_file(): - continue - nested_path_str = nested_path.as_posix() - if self._watch_filter(Change.added, nested_path_str): - expanded_changes.add((Change.added, nested_path_str)) - except OSError as err: - logger.debug(f"扫描新增目录失败: {event_path} - {err}") - return expanded_changes - - @staticmethod - def _build_event(change_type: Change, event_path: Path) -> Optional[DirectoryChangeEvent]: - """ - 构建目录变化事件，路径已不存在时忽略。 - :param change_type: watchfiles 变化类型 - :param event_path: 变化路径 - :return: 目录变化事件 - """ - try: - is_directory = event_path.is_dir() - except OSError as err: - logger.debug(f"读取目录监控事件路径失败: {event_path} - {err}") - return None - if not event_path.exists(): - return None - return DirectoryChangeEvent( - change_type=change_type, - src_path=event_path.as_posix(), - is_directory=is_directory - ) - - @staticmethod - def _get_file_size(event_path: Path) -> Optional[int]: - """ - 读取事件文件大小，文件已消失时返回 None。 - :param event_path: 事件文件路径 - :return: 文件大小 - """ - try: - return event_path.stat().st_size - except OSError as err: - logger.debug(f"读取目录监控文件大小失败: {event_path} - {err}") - return None - - @staticmethod - def _change_text(change_type: Change) -> str: - """ - 转换 watchfiles 事件类型为日志文案。 - :param change_type: watchfiles 变化类型 - :return: 事件描述 - """ - if change_type == Change.modified: - return "修改" - return "新增" - - -class Monitor(ConfigReloadMixin, metaclass=SingletonClass): - """ - 目录监控处理链，单例模式 - """ - CONFIG_WATCH = {SystemConfigKey.Directories.value} - - def __init__(self): - super().__init__() - # 本地目录监控服务 - self._watchers = [] - # 定时服务 - self._scheduler = None - # 存储过照间隔（分钟） - self._snapshot_interval = 5 - # TTL缓存，10秒钟有效 - self._cache = TTLCache(region="monitor", maxsize=1024, ttl=10) - # 快照文件缓存 - self._snapshot_cache = FileCache(base=settings.CACHE_PATH / "snapshots") - # 监控的文件扩展名 - self.all_exts = settings.RMT_MEDIAEXT + settings.RMT_SUBEXT + settings.RMT_AUDIOEXT - # 启动目录监控和文件整理 - self.init() - - def on_config_changed(self): - self.init() - - def get_reload_name(self): - return "目录监控" - - def save_snapshot(self, storage: str, snapshot: Dict, file_count: int = 0, - last_snapshot_time: Optional[float] = None): - """ - 保存快照到文件缓存 - :param storage: 存储名称 - :param snapshot: 快照数据 - :param last_snapshot_time: 上次快照时间戳 - :param file_count: 文件数量，用于调整监控间隔 - """ - 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_data = { - 'timestamp': snapshot_time, - 'file_count': file_count, - 'snapshot': snapshot - } - # 使用FileCache保存快照数据 - cache_key = f"{storage}_snapshot" - snapshot_json = json.dumps(snapshot_data, ensure_ascii=False, indent=2) - self._snapshot_cache.set(cache_key, snapshot_json.encode('utf-8'), region="snapshots") - logger.debug(f"快照已保存到缓存: {storage}") - except Exception as e: - logger.error(f"保存快照失败: {e}") - - def reset_snapshot(self, storage: str) -> bool: - """ - 重置快照，强制下次扫描时重新建立基准 - :param storage: 存储名称 - :return: 是否成功 - """ - try: - cache_key = f"{storage}_snapshot" - if self._snapshot_cache.exists(cache_key, region="snapshots"): - self._snapshot_cache.delete(cache_key, region="snapshots") - logger.info(f"快照已重置: {storage}") - return True - logger.debug(f"快照文件不存在，无需重置: {storage}") - return True - except Exception as e: - logger.error(f"重置快照失败: {storage} - {e}") - return False - - def force_full_scan(self, storage: str, mon_path: Path) -> bool: - """ - 强制全量扫描并处理所有文件（包括已存在的文件） - :param storage: 存储名称 - :param mon_path: 监控路径 - :return: 是否成功 - """ - try: - logger.info(f"开始强制全量扫描: {storage}:{mon_path}") - - # 生成快照 - new_snapshot = StorageChain().snapshot_storage( - storage=storage, - path=mon_path, - last_snapshot_time=0 # 全量扫描，不使用增量 - ) - - if new_snapshot is None: - logger.warn(f"获取 {storage}:{mon_path} 快照失败") - return False - - file_count = len(new_snapshot) - logger.info(f"{storage}:{mon_path} 全量扫描完成，发现 {file_count} 个文件") - - # 处理所有文件 - processed_count = 0 - for file_path, file_info in new_snapshot.items(): - try: - if not self.__is_transfer_candidate_path(Path(file_path)): - continue - file_size = file_info.get('size', 0) if isinstance(file_info, dict) else file_info - if self.__handle_file(storage=storage, event_path=Path(file_path), file_size=file_size): - processed_count += 1 - except Exception as e: - logger.error(f"处理文件 {file_path} 失败: {e}") - continue - - logger.info(f"{storage}:{mon_path} 全量扫描完成，共处理 {processed_count}/{file_count} 个文件") - - # 保存快照 - self.save_snapshot(storage, new_snapshot, file_count) - - return True - - except Exception as e: - logger.error(f"强制全量扫描失败: {storage}:{mon_path} - {e}") - return False - - def load_snapshot(self, storage: str) -> Optional[Dict]: - """ - 从文件缓存加载快照 - :param storage: 存储名称 - :return: 快照数据或None - """ - try: - cache_key = f"{storage}_snapshot" - snapshot_data = self._snapshot_cache.get(cache_key, region="snapshots") - if snapshot_data: - data = json.loads(snapshot_data.decode('utf-8')) - logger.debug(f"成功加载快照: {storage}, 包含 {len(data.get('snapshot', {}))} 个文件") - return data - logger.debug(f"快照文件不存在: {storage}") - return None - except Exception as e: - logger.error(f"加载快照失败: {e}") - return None - - @staticmethod - def adjust_monitor_interval(file_count: int) -> int: - """ - 根据文件数量动态调整监控间隔 - :param file_count: 文件数量 - :return: 监控间隔（分钟） - """ - if file_count < 100: - return 5 # 5分钟 - elif file_count < 500: - return 10 # 10分钟 - elif file_count < 1000: - return 15 # 15分钟 - else: - return 30 # 30分钟 - - @staticmethod - def compare_snapshots(old_snapshot: Dict, new_snapshot: Dict) -> Dict[str, List]: - """ - 比对快照，找出变化的文件（只处理新增和修改，不处理删除） - :param old_snapshot: 旧快照 - :param new_snapshot: 新快照 - :return: 变化信息 - """ - changes = { - 'added': [], - 'modified': [] - } - - old_files = set(old_snapshot.keys()) - new_files = set(new_snapshot.keys()) - - # 新增文件 - changes['added'] = list(new_files - old_files) - - # 修改文件（大小或时间变化） - for file_path in old_files & new_files: - old_info = old_snapshot[file_path] - new_info = new_snapshot[file_path] - - # 检查文件大小变化 - old_size = old_info.get('size', 0) if isinstance(old_info, dict) else old_info - new_size = new_info.get('size', 0) if isinstance(new_info, dict) else new_info - - # 检查修改时间变化（如果有的话） - old_time = old_info.get('modify_time', 0) if isinstance(old_info, dict) else 0 - new_time = new_info.get('modify_time', 0) if isinstance(new_info, dict) else 0 - - if old_size != new_size or (old_time and new_time and old_time != new_time): - changes['modified'].append(file_path) - - return changes - - @staticmethod - def __is_bluray_sub(_path: Path) -> bool: - """ - 判断是否蓝光原盘目录内的媒体流文件。 - """ - return True if re.search(r"BDMV[/\\]STREAM", _path.as_posix(), re.IGNORECASE) else False - - @staticmethod - def __get_bluray_dir(_path: Path) -> Optional[Path]: - """ - 获取蓝光原盘BDMV目录的上级目录。 - """ - for p in _path.parents: - if p.name == "BDMV": - return p.parent - return None - - @staticmethod - def __has_suffix_in(file_path: Path, extensions: List[str]) -> bool: - """ - 判断路径后缀是否命中给定扩展名列表。 - """ - if not file_path.suffix: - return False - return file_path.suffix.casefold() in {ext.casefold() for ext in extensions} - - def __is_transfer_candidate_path(self, file_path: Path) -> bool: - """ - 判断监控事件路径是否需要进入整理链。 - """ - if self.__has_suffix_in(file_path, settings.DOWNLOAD_TMPEXT): - return False - return self.__has_suffix_in(file_path, self.all_exts) - - @staticmethod - def __build_transfer_src_path(event_path: Path, is_bluray_folder: bool) -> str: - """ - 生成整理记录使用的源路径。 - """ - if is_bluray_folder: - return f"{event_path.as_posix()}/" - return event_path.as_posix() - - @staticmethod - def __has_transfer_history(storage: str, src_path: str) -> Optional[bool]: - """ - 判断源文件是否已经存在整理记录。 - """ - try: - return bool(TransferHistoryOper().get_by_src(src_path, storage=storage)) - except Exception as err: - logger.error(f"查询整理历史失败: {src_path} - {err}") - return None - - @staticmethod - def count_directory_files(directory: Path, max_check: int = 10000) -> int: - """ - 统计目录下的文件数量（用于检测是否超过系统限制） - :param directory: 目录路径 - :param max_check: 最大检查数量，避免长时间阻塞 - :return: 文件数量 - """ - try: - count = 0 - import os - for root, dirs, files in os.walk(str(directory)): - count += len(files) - if count > max_check: - return count - return count - except Exception as err: - logger.debug(f"统计目录文件数量失败: {err}") - return 0 - - @staticmethod - def check_system_limits() -> Dict[str, Any]: - """ - 检查系统限制 - :return: 系统限制信息 - """ - limits = { - 'max_user_watches': 0, - 'max_user_instances': 0, - 'current_watches': 0, - 'warnings': [] - } - - try: - system = platform.system() - if system == 'Linux': - # 检查 inotify 限制 - try: - with open('/proc/sys/fs/inotify/max_user_watches', 'r', encoding='utf-8', errors='replace') as f: - limits['max_user_watches'] = int(f.read().strip()) - except Exception as e: - logger.debug(f"读取 inotify 限制失败: {e}") - limits['max_user_watches'] = 8192 # 默认值 - - try: - with open('/proc/sys/fs/inotify/max_user_instances', 'r', encoding='utf-8', errors='replace') as f: - limits['max_user_instances'] = int(f.read().strip()) - except Exception as e: - logger.debug(f"读取 inotify 实例限制失败: {e}") - - # 检查当前使用的watches - try: - import subprocess - result = subprocess.run(['find', '/proc/*/fd', '-lname', 'anon_inode:inotify', '-printf', '%h\n'], - capture_output=True, text=True, timeout=5) - if result.returncode == 0: - limits['current_watches'] = len(result.stdout.strip().split('\n')) - except Exception as e: - logger.debug(f"检查当前 inotify 使用失败: {e}") - - except Exception as e: - limits['warnings'].append(f"检查系统限制时出错: {e}") - - return limits - - @staticmethod - def get_system_optimization_tips() -> List[str]: - """ - 获取系统优化建议 - :return: 优化建议列表 - """ - tips = [] - system = platform.system() - - if system == 'Linux': - tips.extend([ - "增加 inotify 监控数量限制:", - "echo fs.inotify.max_user_watches=524288 | sudo tee -a /etc/sysctl.conf", - "echo fs.inotify.max_user_instances=524288 | sudo tee -a /etc/sysctl.conf", - "sudo sysctl -p", - "", - "如果在Docker中运行，请在宿主机上执行以上命令" - ]) - elif system == 'Darwin': - tips.extend([ - "macOS 系统优化建议:", - "sudo sysctl kern.maxfiles=65536", - "sudo sysctl kern.maxfilesperproc=32768", - "ulimit -n 32768" - ]) - elif system == 'Windows': - tips.extend([ - "Windows 系统优化建议:", - "1. 关闭不必要的实时保护软件对监控目录的扫描", - "2. 将监控目录添加到Windows Defender排除列表", - "3. 确保有足够的可用内存" - ]) - - return tips - - @staticmethod - def should_use_polling(directory: Path, monitor_mode: str, - file_count: int, limits: dict) -> tuple[bool, str]: - """ - 判断是否应该使用轮询模式 - :param directory: 监控目录 - :param monitor_mode: 配置的监控模式 - :param file_count: 目录文件数量 - :param limits: 系统限制信息 - :return: (是否使用轮询, 原因) - """ - if monitor_mode == "compatibility": - return True, "用户配置为兼容模式" - - # 检查网络文件系统 - if SystemUtils.is_network_filesystem(directory): - return True, "检测到网络文件系统，建议使用兼容模式" - - max_watches = limits.get('max_user_watches') - if max_watches and file_count > max_watches * 0.8: - return True, f"目录文件数量({file_count})接近系统限制({max_watches})" - return False, "使用快速模式" - - def init(self): - """ - 启动监控 - """ - # 停止现有任务 - self.stop() - - # 读取目录配置 - monitor_dirs = DirectoryHelper().get_download_dirs() - if not monitor_dirs: - logger.info("未找到任何目录监控配置") - return - - # 按下载目录去重 - monitor_dirs = list({f"{d.storage}_{d.download_path}": d for d in monitor_dirs}.values()) - logger.info(f"找到 {len(monitor_dirs)} 个目录监控配置") - - # 启动定时服务进程 - self._scheduler = BackgroundScheduler(timezone=settings.TZ) - - messagehelper = MessageHelper() - mon_storages = {} - for mon_dir in monitor_dirs: - if not mon_dir.library_path: - logger.warn(f"跳过监控配置 {mon_dir.download_path}：未设置媒体库目录") - continue - if mon_dir.monitor_type != "monitor": - logger.debug(f"跳过监控配置 {mon_dir.download_path}：监控类型为 {mon_dir.monitor_type}") - continue - - # 检查媒体库目录是不是下载目录的子目录 - mon_path = Path(mon_dir.download_path) - target_path = Path(mon_dir.library_path) - if target_path.is_relative_to(mon_path): - logger.warn(f"{target_path} 是监控目录 {mon_path} 的子目录，无法监控！") - messagehelper.put(f"{target_path} 是监控目录 {mon_path} 的子目录，无法监控", title="目录监控") - continue - - # 启动监控 - if mon_dir.storage == "local": - # 本地目录监控 - logger.info(f"正在启动本地目录监控: {mon_path}") - logger.info("*** 重要提示：目录监控只处理新增和修改的文件，不会处理监控启动前已存在的文件 ***") - - try: - # 统计文件数量并给出提示 - file_count = self.count_directory_files(mon_path) - logger.info(f"监控目录 {mon_path} 包含约 {file_count} 个文件") - - # 检查系统限制 - limits = self.check_system_limits() - - # 检查是否需要使用轮询模式 - use_polling, reason = self.should_use_polling(mon_path, - monitor_mode=mon_dir.monitor_mode, - file_count=file_count, - limits=limits) - logger.info(f"监控模式决策: {reason}") - - mode_name = "兼容模式(轮询)" if use_polling else "快速模式" - logger.info(f"使用{mode_name}监控 {mon_path}") - if not use_polling: - if limits['warnings']: - for warning in limits['warnings']: - logger.warn(f"系统限制警告: {warning}") - if limits['max_user_watches'] > 0: - usage_percent = (file_count / limits['max_user_watches']) * 100 - logger.info( - f"系统监控资源使用率: {usage_percent:.1f}% ({file_count}/{limits['max_user_watches']})") - - watcher = LocalDirectoryWatcher( - mon_path=mon_path, - callback=self, - force_polling=True if use_polling else None - ) - self._watchers.append(watcher) - watcher.start() - - logger.info(f"✓ 本地目录监控已启动: {mon_path} [{mode_name}]") - - except Exception as e: - err_msg = str(e) - logger.error(f"启动本地目录监控失败: {mon_path}") - logger.error(f"错误详情: {err_msg}") - - if "inotify" in err_msg.lower(): - logger.error("inotify 相关错误，这通常是由于系统监控数量限制导致的") - logger.error("解决方案:") - tips = self.get_system_optimization_tips() - for tip in tips: - logger.error(f" {tip}") - logger.error("执行上述命令后重启 MoviePilot") - elif "permission" in err_msg.lower(): - logger.error("权限错误，请检查 MoviePilot 是否有足够的权限访问监控目录") - else: - logger.error("建议尝试使用兼容模式进行监控") - - messagehelper.put(f"启动本地目录监控失败: {mon_path}\n错误: {err_msg}", title="目录监控") - else: - if not mon_storages.get(mon_dir.storage): - mon_storages[mon_dir.storage] = [] - mon_storages[mon_dir.storage].append(mon_path) - - for storage, paths in mon_storages.items(): - # 远程目录监控 - 使用智能间隔 - # 先尝试加载已有快照获取文件数量 - snapshot_data = self.load_snapshot(storage) - file_count = snapshot_data.get('file_count', 0) if snapshot_data else 0 - interval = self.adjust_monitor_interval(file_count) - for path in paths: - logger.info(f"正在启动远程目录监控: {path} [{storage}]") - logger.info("*** 重要提示：远程目录监控只处理新增和修改的文件，不会处理监控启动前已存在的文件 ***") - logger.info(f"预估文件数量: {file_count}, 监控间隔: {interval}分钟") - - self._scheduler.add_job( - self.polling_observer, - 'interval', - minutes=interval, - kwargs={ - 'storage': storage, - 'mon_paths': paths - }, - id=f"monitor_{storage}", - replace_existing=True - ) - logger.info(f"✓ 远程目录监控已启动: [间隔: {interval}分钟]") - - # 启动定时服务 - if self._scheduler.get_jobs(): - self._scheduler.print_jobs() - self._scheduler.start() - logger.info("定时监控服务已启动") - - # 输出监控总结 - local_count = len([d for d in monitor_dirs if d.storage == "local" and d.monitor_type == "monitor"]) - remote_count = len([d for d in monitor_dirs if d.storage != "local" and d.monitor_type == "monitor"]) - logger.info(f"目录监控启动完成: 本地监控 {local_count} 个，远程监控 {remote_count} 个") - - def polling_observer(self, storage: str, mon_paths: List[Path]): - """ - 轮询监控（改进版） - """ - monitor_scope = ",".join(str(mon_path) for mon_path in mon_paths) or "未配置路径" - with snapshot_lock: - try: - # 加载上次快照数据 - old_snapshot_data = self.load_snapshot(storage) - old_snapshot = old_snapshot_data.get('snapshot', {}) if old_snapshot_data else {} - 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 = {} - for mon_path in mon_paths: - logger.debug(f"开始对 {storage}:{mon_path} 进行快照...") - - # 生成新快照（增量模式） - snapshot = StorageChain().snapshot_storage( - storage=storage, - path=mon_path, - last_snapshot_time=last_snapshot_time - ) - - if snapshot is None: - logger.warn(f"获取 {storage}:{mon_path} 快照失败") - continue - new_snapshot.update(snapshot) - file_count = len(snapshot) - logger.info(f"{storage}:{mon_path} 快照完成，发现 {file_count} 个文件") - file_count = len(new_snapshot) - if not is_first_snapshot: - # 比较快照找出变化 - changes = self.compare_snapshots(old_snapshot, new_snapshot) - added_files = [ - file_path - for file_path in changes['added'] - if self.__is_transfer_candidate_path(Path(file_path)) - ] - modified_files = [ - file_path - for file_path in changes['modified'] - if self.__is_transfer_candidate_path(Path(file_path)) - ] - - # 处理新增文件 - handled_added_count = 0 - for new_file in added_files: - file_info = new_snapshot.get(new_file, {}) - file_size = file_info.get('size', 0) if isinstance(file_info, dict) else file_info - if self.__handle_file(storage=storage, event_path=Path(new_file), file_size=file_size): - handled_added_count += 1 - - # 处理修改文件 - handled_modified_count = 0 - for modified_file in modified_files: - file_info = new_snapshot.get(modified_file, {}) - file_size = file_info.get('size', 0) if isinstance(file_info, dict) else file_info - if self.__handle_file(storage=storage, event_path=Path(modified_file), file_size=file_size): - handled_modified_count += 1 - - if handled_added_count or handled_modified_count: - logger.info( - f"{storage} 发现 {handled_added_count} 个新增文件，{handled_modified_count} 个修改文件") - else: - logger.debug(f"{storage} 无文件变化") - else: - logger.info(f"{storage} 首次快照完成，共 {file_count} 个文件") - logger.info("*** 首次快照仅建立基准，不会处理现有文件。后续监控将处理新增和修改的文件 ***") - - # 保存新快照 - self.save_snapshot(storage, new_snapshot, file_count, last_snapshot_time) - - # 动态调整监控间隔 - new_interval = self.adjust_monitor_interval(file_count) - current_job = self._scheduler.get_job(f"monitor_{storage}") - if current_job and current_job.trigger.interval.total_seconds() / 60 != new_interval: - # 重新安排任务 - self._scheduler.modify_job( - f"monitor_{storage}", - trigger='interval', - minutes=new_interval - ) - logger.info(f"{storage}:{monitor_scope} 监控间隔已调整为 {new_interval} 分钟") - - except Exception as e: - logger.error(f"轮询监控 {storage}:{monitor_scope} 出现错误：{e}") - logger.debug(traceback.format_exc()) - - def event_handler(self, event, text: str, event_path: str, file_size: float = None): - """ - 处理文件变化 - :param event: 事件 - :param text: 事件描述 - :param event_path: 事件文件路径 - :param file_size: 文件大小 - """ - if not event.is_directory: - if not self.__is_transfer_candidate_path(Path(event_path)): - return - # 整理文件 - self.__handle_file(storage="local", event_path=Path(event_path), file_size=file_size) - - def __handle_file(self, storage: str, event_path: Path, file_size: float = None) -> bool: - """ - 整理一个文件 - :param storage: 存储 - :param event_path: 事件文件路径 - :param file_size: 文件大小 - :return: 是否进入整理链 - """ - # 全程加锁 - with lock: - is_bluray_folder = False - # 蓝光原盘文件处理 - if self.__is_bluray_sub(event_path): - event_path = self.__get_bluray_dir(event_path) - if not event_path: - return False - is_bluray_folder = True - elif not self.__is_transfer_candidate_path(event_path): - return False - - # TTL缓存控重 - if self._cache.get(str(event_path)): - return False - self._cache[str(event_path)] = True - - src_path = self.__build_transfer_src_path( - event_path=event_path, - is_bluray_folder=is_bluray_folder, - ) - has_transfer_history = self.__has_transfer_history( - storage=storage, - src_path=src_path, - ) - if has_transfer_history is not False: - return False - - try: - if is_bluray_folder: - logger.info(f"开始整理蓝光原盘: {event_path}") - else: - logger.info(f"开始整理文件: {event_path}") - # 开始整理 - TransferChain().do_transfer( - fileitem=FileItem( - storage=storage, - path=src_path, - type="file" if not is_bluray_folder else "dir", - name=event_path.name, - basename=event_path.stem, - extension=event_path.suffix[1:], - size=file_size - ) - ) - return True - except Exception as e: - logger.error("目录监控整理文件发生错误：%s - %s" % (str(e), traceback.format_exc())) - return False - - def stop(self): - """ - 退出监控 - """ - if self._watchers: - logger.info("正在停止本地目录监控服务...") - for watcher in self._watchers: - try: - watcher.stop() - watcher.join(timeout=5) - if watcher.is_alive(): - logger.warning(f"本地目录监控线程在5秒内未能停止: {watcher.watch_path}") - else: - logger.debug(f"已停止本地目录监控服务: {watcher.watch_path}") - except Exception as e: - logger.error(f"停止目录监控服务出现了错误：{e}") - self._watchers = [] - logger.info("本地目录监控服务已停止") - if self._scheduler: - self._scheduler.remove_all_jobs() - if self._scheduler.running: - try: - self._scheduler.shutdown() - logger.info("定时监控服务已停止") - except Exception as e: - logger.error(f"停止定时服务出现了错误：{e}") - self._scheduler = None - if self._cache: - self._cache.close() - if self._snapshot_cache: - self._snapshot_cache.close() diff --git a/app/monitor/__init__.py b/app/monitor/__init__.py new file mode 100644 index 000000000..7fb947733 --- /dev/null +++ b/app/monitor/__init__.py @@ -0,0 +1,14 @@ +""" +目录监控包。 + +- watcher.py 本地目录监控线程（watchfiles） +- syslimits.py 系统限制探测与监控模式决策 +- snapshot.py 远程快照存取与比对 +- dispatcher.py 监控事件到整理链的分发 +- poller.py 远程目录轮询监控 +- monitor.py Monitor 门面：装配、生命周期与健康检查 +""" +from app.monitor.watcher import DirectoryChangeEvent, LocalDirectoryWatcher +from app.monitor.monitor import Monitor + +__all__ = ["DirectoryChangeEvent", "LocalDirectoryWatcher", "Monitor"] diff --git a/app/monitor/dispatcher.py b/app/monitor/dispatcher.py new file mode 100644 index 000000000..6a99f6606 --- /dev/null +++ b/app/monitor/dispatcher.py @@ -0,0 +1,210 @@ +import re +import traceback +from pathlib import Path +from threading import Lock +from typing import Any, Dict, List, Optional + +from app.chain.transfer import TransferChain +from app.core.cache import TTLCache +from app.core.config import settings +from app.db.transferhistory_oper import TransferHistoryOper +from app.log import logger +from app.schemas import FileItem + + +class TransferDispatcher: + """ + 将监控事件分发到整理链：候选判定、TTL 去重、整理历史查重与整理触发。 + """ + # 历史查询失败待重试队列上限，防止长时间故障期间无限增长 + MAX_PENDING_RETRIES = 1000 + # 单个文件的最大重试次数（按健康检查周期计，60 次约 1 小时） + MAX_RETRY_ATTEMPTS = 60 + + def __init__(self, all_exts: Optional[List[str]] = None, cache: Optional[Any] = None): + """ + 初始化整理分发器。 + :param all_exts: 监控的文件扩展名，默认取系统配置 + :param cache: 去重缓存，默认使用 10 秒 TTL 缓存 + """ + self.all_exts = all_exts if all_exts is not None else ( + settings.RMT_MEDIAEXT + settings.RMT_SUBEXT + settings.RMT_AUDIOEXT) + self._cache = cache if cache is not None else TTLCache(region="monitor", maxsize=1024, ttl=10) + self._lock = Lock() + # 历史查询失败待重试的文件 + self._pending_retries: Dict[str, Dict[str, Any]] = {} + self._pending_guard = Lock() + + @staticmethod + def _is_bluray_sub(_path: Path) -> bool: + """ + 判断是否蓝光原盘目录内的媒体流文件。 + """ + return True if re.search(r"BDMV[/\\]STREAM", _path.as_posix(), re.IGNORECASE) else False + + @staticmethod + def _get_bluray_dir(_path: Path) -> Optional[Path]: + """ + 获取蓝光原盘BDMV目录的上级目录。 + """ + for p in _path.parents: + if p.name == "BDMV": + return p.parent + return None + + @staticmethod + def _has_suffix_in(file_path: Path, extensions: List[str]) -> bool: + """ + 判断路径后缀是否命中给定扩展名列表。 + """ + if not file_path.suffix: + return False + return file_path.suffix.casefold() in {ext.casefold() for ext in extensions} + + def is_transfer_candidate_path(self, file_path: Path) -> bool: + """ + 判断监控事件路径是否需要进入整理链。 + """ + if self._has_suffix_in(file_path, settings.DOWNLOAD_TMPEXT): + return False + return self._has_suffix_in(file_path, self.all_exts) + + @staticmethod + def _build_transfer_src_path(event_path: Path, is_bluray_folder: bool) -> str: + """ + 生成整理记录使用的源路径。 + """ + if is_bluray_folder: + return f"{event_path.as_posix()}/" + return event_path.as_posix() + + @staticmethod + def _has_transfer_history(storage: str, src_path: str) -> Optional[bool]: + """ + 判断源文件是否已经存在整理记录。 + :return: True/False 查询成功，None 查询失败 + """ + try: + return bool(TransferHistoryOper().get_by_src(src_path, storage=storage)) + except Exception as err: + logger.error(f"查询整理历史失败: {src_path} - {err}") + return None + + @staticmethod + def _pending_key(storage: str, event_path: Path) -> str: + """ + 生成待重试文件的唯一键。 + """ + return f"{storage}:{Path(event_path).as_posix()}" + + def _register_pending(self, storage: str, event_path: Path, file_size: float = None): + """ + 登记历史查询失败的文件待重试，重复失败累计次数，超限后放弃。 + :param storage: 存储 + :param event_path: 原始事件路径 + :param file_size: 文件大小 + """ + key = self._pending_key(storage, event_path) + with self._pending_guard: + entry = self._pending_retries.get(key) + if entry: + entry["attempts"] += 1 + if entry["attempts"] >= self.MAX_RETRY_ATTEMPTS: + self._pending_retries.pop(key, None) + logger.error(f"整理历史查询持续失败，已放弃重试: {key}") + return + if len(self._pending_retries) >= self.MAX_PENDING_RETRIES: + logger.error(f"整理重试队列已满，丢弃: {key}") + return + self._pending_retries[key] = { + "storage": storage, + "event_path": event_path, + "file_size": file_size, + "attempts": 1 + } + logger.warn(f"整理历史查询失败，已登记待重试: {key}") + + def _discard_pending(self, storage: str, event_path: Path): + """ + 历史查询已得到确定结果，移除待重试登记。 + :param storage: 存储 + :param event_path: 原始事件路径 + """ + with self._pending_guard: + self._pending_retries.pop(self._pending_key(storage, event_path), None) + + def retry_pending(self): + """ + 重试历史查询失败的文件，由健康检查周期驱动。 + 成功或得到确定结果的条目在 handle_file 内部自动移除。 + """ + with self._pending_guard: + items = list(self._pending_retries.values()) + for item in items: + logger.info(f"重试整理: {item['storage']}:{item['event_path']}") + self.handle_file(storage=item["storage"], event_path=item["event_path"], + file_size=item["file_size"]) + + def handle_file(self, storage: str, event_path: Path, file_size: float = None) -> bool: + """ + 整理一个文件。 + :param storage: 存储 + :param event_path: 事件文件路径 + :param file_size: 文件大小 + :return: 是否进入整理链 + """ + with self._lock: + # 登记重试用原始事件路径，蓝光目录解析在重试时重新执行 + origin_path = event_path + is_bluray_folder = False + # 蓝光原盘文件处理 + if self._is_bluray_sub(event_path): + event_path = self._get_bluray_dir(event_path) + if not event_path: + return False + is_bluray_folder = True + elif not self.is_transfer_candidate_path(event_path): + return False + + # TTL缓存控重 + if self._cache.get(str(event_path)): + return False + self._cache[str(event_path)] = True + + src_path = self._build_transfer_src_path( + event_path=event_path, + is_bluray_folder=is_bluray_folder, + ) + has_transfer_history = self._has_transfer_history( + storage=storage, + src_path=src_path, + ) + if has_transfer_history is None: + # 查询失败是暂时故障，登记待重试（由健康检查周期驱动），不能永久跳过 + self._register_pending(storage=storage, event_path=origin_path, file_size=file_size) + return False + self._discard_pending(storage=storage, event_path=origin_path) + if has_transfer_history: + return False + + try: + if is_bluray_folder: + logger.info(f"开始整理蓝光原盘: {event_path}") + else: + logger.info(f"开始整理文件: {event_path}") + # 开始整理 + TransferChain().do_transfer( + fileitem=FileItem( + storage=storage, + path=src_path, + type="file" if not is_bluray_folder else "dir", + name=event_path.name, + basename=event_path.stem, + extension=event_path.suffix[1:], + size=file_size + ) + ) + return True + except Exception as e: + logger.error("目录监控整理文件发生错误：%s - %s" % (str(e), traceback.format_exc())) + return False diff --git a/app/monitor/monitor.py b/app/monitor/monitor.py new file mode 100644 index 000000000..2c5335ab9 --- /dev/null +++ b/app/monitor/monitor.py @@ -0,0 +1,508 @@ +import traceback +from pathlib import Path +from threading import Lock +from typing import Any, Dict, List, Optional + +from apscheduler.schedulers.background import BackgroundScheduler + +from app.core.config import settings +from app.helper.directory import DirectoryHelper +from app.helper.message import MessageHelper +from app.log import logger +from app.monitor.dispatcher import TransferDispatcher +from app.monitor.poller import RemotePoller +from app.monitor.snapshot import SnapshotStore +from app.monitor.syslimits import decide_monitor_mode, get_system_optimization_tips +from app.monitor.watcher import LocalDirectoryWatcher +from app.schemas.types import SystemConfigKey +from app.utils.mixins import ConfigReloadMixin +from app.utils.singleton import SingletonClass +from app.utils.system import SystemUtils + + +class Monitor(ConfigReloadMixin, metaclass=SingletonClass): + """ + 目录监控门面，单例模式：装配本地/远程监控、维护生命周期与健康检查。 + """ + CONFIG_WATCH = {SystemConfigKey.Directories.value} + # 目录监控健康检查间隔（秒） + WATCHDOG_INTERVAL = 60 + # 连续多少个健康检查周期无新增重启后才宣告恢复，避免反复崩溃时告警刷屏 + RECOVERY_STABLE_CYCLES = 5 + + def __init__(self): + super().__init__() + # 本地目录监控服务 + self._watchers = [] + # 本地目录监控列表读写锁 + self._watcher_lock = Lock() + # 启动失败待重试的本地监控配置 + self._pending_locals: List[Dict[str, Any]] = [] + # 已告警的监控目录，避免重复推送 + self._alerted_paths: set = set() + # 各监控目录已告警过的自动重启次数 + self._restart_marks: Dict[str, int] = {} + # 各监控目录连续稳定的健康检查周期数 + self._stable_cycles: Dict[str, int] = {} + # 定时服务 + self._scheduler = None + # 整理分发器 + self._dispatcher = TransferDispatcher() + # 快照存储 + self._store = SnapshotStore() + # 远程轮询监控 + self._poller = RemotePoller(store=self._store, dispatcher=self._dispatcher, + alert_cb=self.__poller_alert) + # 启动目录监控和文件整理 + self.init() + + def on_config_changed(self): + self.init() + + def get_reload_name(self): + return "目录监控" + + def save_snapshot(self, storage: str, snapshot: Dict, file_count: int = 0, + last_snapshot_time: Optional[float] = None): + """ + 保存快照到文件缓存。 + """ + self._store.save(storage, snapshot, file_count=file_count, last_snapshot_time=last_snapshot_time) + + def load_snapshot(self, storage: str) -> Optional[Dict]: + """ + 从文件缓存加载快照。 + """ + return self._store.load(storage) + + def reset_snapshot(self, storage: str) -> bool: + """ + 重置快照，强制下次扫描时重新建立基准。 + """ + return self._store.reset(storage) + + def force_full_scan(self, storage: str, mon_path: Path) -> bool: + """ + 强制全量扫描并处理所有文件（包括已存在的文件）。 + """ + return self._poller.force_full_scan(storage=storage, mon_path=mon_path) + + @staticmethod + def adjust_monitor_interval(file_count: int) -> int: + """ + 根据文件数量动态调整监控间隔。 + """ + return SnapshotStore.adjust_interval(file_count) + + @staticmethod + def compare_snapshots(old_snapshot: Dict, new_snapshot: Dict) -> Dict[str, List]: + """ + 比对快照，找出变化的文件。 + """ + return SnapshotStore.compare(old_snapshot, new_snapshot) + + def init(self): + """ + 启动监控 + """ + # 停止现有任务 + self.stop() + + # 读取目录配置 + monitor_dirs = DirectoryHelper().get_download_dirs() + if not monitor_dirs: + logger.info("未找到任何目录监控配置") + return + + messagehelper = MessageHelper() + + # 先筛出有效的监控配置，再按下载目录去重，避免非监控配置顶掉监控配置 + valid_dirs = [] + for mon_dir in monitor_dirs: + if not mon_dir.library_path: + logger.warn(f"跳过监控配置 {mon_dir.download_path}：未设置媒体库目录") + continue + if mon_dir.monitor_type != "monitor": + logger.debug(f"跳过监控配置 {mon_dir.download_path}：监控类型为 {mon_dir.monitor_type}") + continue + valid_dirs.append(mon_dir) + + deduped: Dict[str, Any] = {} + for mon_dir in valid_dirs: + key = f"{mon_dir.storage}_{mon_dir.download_path}" + if key in deduped: + logger.warn(f"监控配置重复，忽略后一条: {mon_dir.download_path}" + f"（媒体库 {mon_dir.library_path}）") + continue + deduped[key] = mon_dir + monitor_dirs = list(deduped.values()) + logger.info(f"找到 {len(monitor_dirs)} 个目录监控配置") + + # 启动定时服务进程 + self._scheduler = BackgroundScheduler(timezone=settings.TZ) + + mon_storages: Dict[str, List[Path]] = {} + # 本地监控启动结果计数，用于输出真实的启动总结 + local_started = 0 + local_failed = 0 + for mon_dir in monitor_dirs: + # 检查媒体库目录是不是下载目录的子目录 + mon_path = Path(mon_dir.download_path) + target_path = Path(mon_dir.library_path) + if target_path.is_relative_to(mon_path): + logger.warn(f"{target_path} 是监控目录 {mon_path} 的子目录，无法监控！") + messagehelper.put(f"{target_path} 是监控目录 {mon_path} 的子目录，无法监控", title="目录监控") + continue + + # 启动监控 + if mon_dir.storage == "local": + if self.__start_local_monitor(mon_path=mon_path, monitor_mode=mon_dir.monitor_mode): + local_started += 1 + else: + local_failed += 1 + else: + mon_storages.setdefault(mon_dir.storage, []).append(mon_path) + + for storage, paths in mon_storages.items(): + # 远程目录监控 - 使用智能间隔 + # 先尝试加载已有快照获取文件数量 + snapshot_data = self._store.load(storage) + file_count = snapshot_data.get('file_count', 0) if snapshot_data 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}分钟") + + self._scheduler.add_job( + self.polling_observer, + 'interval', + minutes=interval, + kwargs={ + 'storage': storage, + 'mon_paths': paths + }, + id=f"monitor_{storage}", + replace_existing=True + ) + logger.info(f"✓ 远程目录监控已启动: [间隔: {interval}分钟]") + + # 监控健康检查：重建异常监控线程、重试启动失败目录、重试历史查询失败的文件 + if local_started or local_failed or mon_storages: + self._scheduler.add_job( + self.watchdog, + 'interval', + seconds=self.WATCHDOG_INTERVAL, + id="monitor_watchdog", + replace_existing=True + ) + logger.info(f"✓ 目录监控健康检查已启动: [间隔: {self.WATCHDOG_INTERVAL}秒]") + + # 启动定时服务 + if self._scheduler.get_jobs(): + self._scheduler.print_jobs() + self._scheduler.start() + logger.info("定时监控服务已启动") + + # 输出监控总结，报告实际启动成功数而不是配置数 + remote_count = sum(len(paths) for paths in mon_storages.values()) + summary = f"目录监控启动完成: 本地监控 {local_started} 个成功" + if local_failed: + summary += f"、{local_failed} 个失败（将自动退避重试）" + summary += f"，远程监控 {remote_count} 个" + if local_failed: + logger.warn(summary) + else: + logger.info(summary) + + def __start_local_monitor(self, mon_path: Path, monitor_mode: str) -> bool: + """ + 启动单个本地目录监控，失败时登记待重试。 + :param mon_path: 监控目录 + :param monitor_mode: 配置的监控模式 + :return: 是否启动成功 + """ + logger.info(f"正在启动本地目录监控: {mon_path}") + logger.info("*** 重要提示：目录监控只处理新增和修改的文件，不会处理监控启动前已存在的文件 ***") + + try: + # 检查是否需要使用轮询模式（兼容模式/网络存储不做启动期目录遍历） + use_polling, reason, limits, file_count = decide_monitor_mode(mon_path, monitor_mode) + logger.info(f"监控模式决策: {reason}") + + mode_name = "兼容模式(轮询)" if use_polling else "快速模式" + logger.info(f"使用{mode_name}监控 {mon_path}") + if file_count is not None: + logger.info(f"监控目录 {mon_path} 包含约 {file_count} 个文件") + if not use_polling and limits: + if limits['warnings']: + for warning in limits['warnings']: + logger.warn(f"系统限制警告: {warning}") + if limits['max_user_watches'] > 0 and file_count is not None: + usage_percent = (file_count / limits['max_user_watches']) * 100 + logger.info( + f"系统监控资源使用率: {usage_percent:.1f}% ({file_count}/{limits['max_user_watches']})") + + # 网络/FUSE 挂载轮询降频，减少监控自身对挂载后端的持续 stat 压力 + poll_delay_ms = None + if use_polling and SystemUtils.is_network_filesystem(mon_path): + poll_delay_ms = LocalDirectoryWatcher.POLL_DELAY_NETWORK_MS + logger.info(f"检测到网络文件系统，轮询扫描间隔调整为 {poll_delay_ms}ms: {mon_path}") + + watcher = LocalDirectoryWatcher( + mon_path=mon_path, + callback=self, + force_polling=True if use_polling else None, + poll_delay_ms=poll_delay_ms + ) + # 启动成功后再登记，避免失败的监控残留在列表中 + watcher.start() + with self._watcher_lock: + self._watchers.append(watcher) + self._pending_locals = [ + pending for pending in self._pending_locals + if pending["mon_path"] != mon_path + ] + self.__clear_alert(mon_path, f"本地目录监控已恢复: {mon_path} [{mode_name}]") + + logger.info(f"✓ 本地目录监控已启动: {mon_path} [{mode_name}]") + return True + + except Exception as e: + self.__handle_start_failure(mon_path=mon_path, monitor_mode=monitor_mode, err=e) + return False + + def __handle_start_failure(self, mon_path: Path, monitor_mode: str, err: Exception): + """ + 处理本地目录监控启动失败，登记待重试并按需告警。 + :param mon_path: 监控目录 + :param monitor_mode: 配置的监控模式 + :param err: 启动异常 + """ + err_msg = str(err) + logger.error(f"启动本地目录监控失败: {mon_path}") + logger.error(f"错误详情: {err_msg}") + + if "inotify" in err_msg.lower(): + logger.error("inotify 相关错误，这通常是由于系统监控数量限制导致的") + logger.error("解决方案:") + for tip in get_system_optimization_tips(): + logger.error(f" {tip}") + logger.error("执行上述命令后重启 MoviePilot") + elif "permission" in err_msg.lower(): + logger.error("权限错误，请检查 MoviePilot 是否有足够的权限访问监控目录") + elif isinstance(err, (FileNotFoundError, NotADirectoryError)): + logger.error("监控目录当前不可用，网络存储/FUSE 挂载可能尚未就绪，将自动重试") + elif monitor_mode != "compatibility": + logger.error("建议尝试使用兼容模式进行监控") + + with self._watcher_lock: + if all(pending["mon_path"] != mon_path for pending in self._pending_locals): + self._pending_locals.append({ + "mon_path": mon_path, + "monitor_mode": monitor_mode + }) + self.__send_alert(mon_path, + f"启动本地目录监控失败: {mon_path}\n错误: {err_msg}\n" + f"将自动退避重试") + + def watchdog(self): + """ + 目录监控健康检查：重建崩溃或静默失效的监控线程，并重试启动失败的监控目录。 + """ + try: + self.__check_watchers() + self.__retry_pending_locals() + self._dispatcher.retry_pending() + except Exception as e: + logger.error(f"目录监控健康检查出现错误：{e}\n{traceback.format_exc()}") + + def __check_watchers(self): + """ + 检查本地目录监控线程状态，异常时重建。 + """ + with self._watcher_lock: + watchers = list(self._watchers) + for watcher in watchers: + key = str(watcher.watch_path) + if watcher.is_stalled(): + reason = f"监控循环超过 {LocalDirectoryWatcher.STALL_TIMEOUT} 秒无任何活动，判定为静默失效" + elif not watcher.is_alive(): + reason = "监控线程已退出" + else: + # 线程已自愈，但崩溃过就要告警，避免自动重启把故障变成新的静默 + if watcher.restart_count > self._restart_marks.get(key, 0): + self._restart_marks[key] = watcher.restart_count + self._stable_cycles[key] = 0 + self.__send_alert(watcher.watch_path, + f"目录监控发生错误并已自动重启" + f"（累计 {watcher.restart_count} 次）: {watcher.watch_path}") + else: + # 稳定满恢复窗口才宣告恢复，避免反复崩溃时告警/恢复消息来回刷屏 + self._stable_cycles[key] = self._stable_cycles.get(key, 0) + 1 + if self._stable_cycles[key] >= self.RECOVERY_STABLE_CYCLES: + self.__clear_alert(watcher.watch_path, f"目录监控已恢复正常: {watcher.watch_path}") + continue + logger.error(f"目录监控异常: {watcher.watch_path} - {reason}，正在重建监控线程 ...") + self.__send_alert(watcher.watch_path, + f"目录监控异常: {watcher.watch_path}\n原因: {reason}\n正在自动重建监控") + self.__rebuild_watcher(watcher) + + def __rebuild_watcher(self, watcher: LocalDirectoryWatcher): + """ + 重建一个本地目录监控线程。 + :param watcher: 需要重建的监控 + """ + # 卡死的线程阻塞在底层调用中无法强制回收，只能请求停止后由守护线程自然退出 + watcher.stop() + new_watcher = LocalDirectoryWatcher( + mon_path=watcher.watch_path, + callback=self, + force_polling=watcher.force_polling, + poll_delay_ms=watcher.poll_delay_ms + ) + try: + new_watcher.start() + except Exception as e: + logger.error(f"重建目录监控失败: {watcher.watch_path} - {e}") + with self._watcher_lock: + self._watchers = [item for item in self._watchers if item is not watcher] + if all(pending["mon_path"] != watcher.watch_path for pending in self._pending_locals): + self._pending_locals.append({ + "mon_path": watcher.watch_path, + # 重建沿用原监控模式，force_polling 为 True 即兼容模式 + "monitor_mode": "compatibility" if watcher.force_polling else "fast" + }) + return + with self._watcher_lock: + self._watchers = [new_watcher if item is watcher else item for item in self._watchers] + # 新监控的重启计数从零开始，同步重置告警基准 + self._restart_marks.pop(str(watcher.watch_path), None) + self._stable_cycles.pop(str(watcher.watch_path), None) + logger.info(f"✓ 目录监控已重建: {watcher.watch_path}") + self.__clear_alert(watcher.watch_path, f"目录监控已自动恢复: {watcher.watch_path}") + + def __retry_pending_locals(self): + """ + 重试启动失败的本地目录监控，给网络存储/FUSE 挂载留出就绪时间。 + """ + with self._watcher_lock: + pending = list(self._pending_locals) + for item in pending: + # 失败次数越多重试间隔越长（按健康检查周期数退避），长时间故障时不刷屏 + if item.get("skip_cycles", 0) > 0: + item["skip_cycles"] -= 1 + continue + logger.info(f"重试启动本地目录监控: {item['mon_path']}") + if not self.__start_local_monitor(mon_path=item["mon_path"], monitor_mode=item["monitor_mode"]): + item["attempts"] = item.get("attempts", 0) + 1 + item["skip_cycles"] = min(item["attempts"], 10) + + def __send_alert(self, mon_path: Path, message: str): + """ + 推送目录监控异常告警，同一目录仅在状态变化时推送一次。 + :param mon_path: 监控目录 + :param message: 告警内容 + """ + key = str(mon_path) + with self._watcher_lock: + if key in self._alerted_paths: + return + self._alerted_paths.add(key) + MessageHelper().put(message, title="目录监控") + + @staticmethod + def __poller_alert(storage: str, message: str): + """ + 远程轮询监控告警回调，复用消息渠道推送。 + :param storage: 存储名称 + :param message: 告警内容 + """ + logger.warn(f"[{storage}] {message}") + MessageHelper().put(message, title="目录监控") + + def __clear_alert(self, mon_path: Path, message: str): + """ + 清除目录监控异常告警状态，并在此前告警过时推送恢复消息。 + :param mon_path: 监控目录 + :param message: 恢复内容 + """ + key = str(mon_path) + with self._watcher_lock: + if key not in self._alerted_paths: + return + self._alerted_paths.discard(key) + logger.info(message) + MessageHelper().put(message, title="目录监控") + + def polling_observer(self, storage: str, mon_paths: List[Path]): + """ + 轮询监控：执行一轮快照并按结果动态调整监控间隔。 + """ + file_count = self._poller.poll(storage=storage, mon_paths=mon_paths) + if file_count is None or not self._scheduler: + return + # 动态调整监控间隔 + new_interval = SnapshotStore.adjust_interval(file_count) + try: + current_job = self._scheduler.get_job(f"monitor_{storage}") + if current_job and current_job.trigger.interval.total_seconds() / 60 != new_interval: + self._scheduler.modify_job( + f"monitor_{storage}", + trigger='interval', + minutes=new_interval + ) + logger.info(f"{storage} 监控间隔已调整为 {new_interval} 分钟") + except Exception as e: + logger.error(f"调整监控间隔失败: {storage} - {e}") + + def event_handler(self, event, text: str, event_path: str, file_size: float = None): + """ + 处理文件变化。 + :param event: 事件 + :param text: 事件描述 + :param event_path: 事件文件路径 + :param file_size: 文件大小 + """ + if event.is_directory: + return + if not self._dispatcher.is_transfer_candidate_path(Path(event_path)): + return + # 整理文件 + self._dispatcher.handle_file(storage="local", event_path=Path(event_path), file_size=file_size) + + def stop(self): + """ + 退出监控 + """ + # 先停定时服务，避免健康检查在停止过程中重建监控线程 + if self._scheduler: + self._scheduler.remove_all_jobs() + if self._scheduler.running: + try: + self._scheduler.shutdown() + logger.info("定时监控服务已停止") + except Exception as e: + logger.error(f"停止定时服务出现了错误：{e}") + self._scheduler = None + with self._watcher_lock: + watchers = self._watchers + self._watchers = [] + self._pending_locals = [] + self._alerted_paths = set() + self._restart_marks = {} + self._stable_cycles = {} + if watchers: + logger.info("正在停止本地目录监控服务...") + for watcher in watchers: + try: + watcher.stop() + watcher.join(timeout=5) + if watcher.is_alive(): + logger.warning(f"本地目录监控线程在5秒内未能停止: {watcher.watch_path}") + else: + logger.debug(f"已停止本地目录监控服务: {watcher.watch_path}") + except Exception as e: + logger.error(f"停止目录监控服务出现了错误：{e}") + logger.info("本地目录监控服务已停止") + # 缓存与快照存储是共享后端的代理，生命周期由应用全局管理，这里不再关闭 diff --git a/app/monitor/poller.py b/app/monitor/poller.py new file mode 100644 index 000000000..b840a3893 --- /dev/null +++ b/app/monitor/poller.py @@ -0,0 +1,228 @@ +import traceback +from pathlib import Path +from threading import Lock +from typing import Callable, Dict, List, Optional + +from app.chain.storage import StorageChain +from app.log import logger +from app.monitor.dispatcher import TransferDispatcher +from app.monitor.snapshot import SnapshotStore + + +class RemotePoller: + """ + 远程目录轮询监控：快照、比对并分发变化文件。 + """ + # 同一存储连续异常达到该次数后推送告警 + FAILURE_ALERT_THRESHOLD = 3 + + def __init__(self, store: SnapshotStore, dispatcher: TransferDispatcher, + alert_cb: Optional[Callable[[str, str], None]] = None): + """ + 初始化远程轮询监控。 + :param store: 快照存储 + :param dispatcher: 整理分发器 + :param alert_cb: 告警回调 (storage, message) + """ + self._store = store + self._dispatcher = dispatcher + self._alert_cb = alert_cb + # 快照锁按存储隔离，避免一个慢存储阻塞其他存储的轮询 + self._locks: Dict[str, Lock] = {} + self._locks_guard = Lock() + # 各存储连续异常次数 + self._failure_counts: Dict[str, int] = {} + + def _get_lock(self, storage: str) -> Lock: + """ + 获取指定存储的快照锁。 + :param storage: 存储名称 + :return: 快照锁 + """ + with self._locks_guard: + return self._locks.setdefault(storage, Lock()) + + def _note_failure(self, storage: str, reason: str): + """ + 记录一次轮询异常，连续异常达到阈值时推送告警。 + :param storage: 存储名称 + :param reason: 异常原因 + """ + count = self._failure_counts.get(storage, 0) + 1 + self._failure_counts[storage] = count + logger.warn(f"远程目录监控异常（连续第 {count} 次）: {storage} - {reason}") + if count == self.FAILURE_ALERT_THRESHOLD and self._alert_cb: + self._alert_cb(storage, + f"远程目录监控连续 {count} 次异常: {storage}\n原因: {reason}\n将继续按周期重试") + + def _note_success(self, storage: str): + """ + 记录一次轮询成功，此前告警过时推送恢复消息。 + :param storage: 存储名称 + """ + if self._failure_counts.get(storage, 0) >= self.FAILURE_ALERT_THRESHOLD and self._alert_cb: + self._alert_cb(storage, f"远程目录监控已恢复: {storage}") + self._failure_counts[storage] = 0 + + def poll(self, storage: str, mon_paths: List[Path]) -> Optional[int]: + """ + 执行一轮轮询监控。 + :param storage: 存储名称 + :param mon_paths: 监控路径列表 + :return: 基线文件数量，本轮无有效结果时返回 None + """ + monitor_scope = ",".join(str(mon_path) for mon_path in mon_paths) or "未配置路径" + with self._get_lock(storage): + try: + # 加载上次快照数据，读取失败不能当作首次快照，否则会丢弃已有基线 + old_snapshot_data, load_ok = self._store.load_checked(storage) + if not load_ok: + self._note_failure(storage, "读取快照基线失败，跳过本轮") + return None + old_snapshot = old_snapshot_data.get('snapshot', {}) if old_snapshot_data else {} + 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 = {} + failed_paths = [] + for mon_path in mon_paths: + logger.debug(f"开始对 {storage}:{mon_path} 进行快照...") + + # 生成新快照（增量模式） + snapshot = StorageChain().snapshot_storage( + storage=storage, + path=mon_path, + last_snapshot_time=last_snapshot_time + ) + + if snapshot is None: + failed_paths.append(str(mon_path)) + logger.warn(f"获取 {storage}:{mon_path} 快照失败") + continue + new_snapshot.update(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)}") + return None + + # 增量快照只包含变化子树，必须与基线合并才是完整视图； + # 直接把增量当基线会导致下一轮把未扫到的旧文件全部误判为新增 + merged_snapshot = {**old_snapshot, **new_snapshot} + file_count = len(merged_snapshot) + + if not is_first_snapshot: + self._handle_changes(storage, old_snapshot, new_snapshot) + else: + logger.info(f"{storage} 首次快照完成，共 {file_count} 个文件") + logger.info("*** 首次快照仅建立基准，不会处理现有文件。后续监控将处理新增和修改的文件 ***") + + # 保存合并后的基线 + if not self._store.save(storage, merged_snapshot, file_count, last_snapshot_time): + self._note_failure(storage, "保存快照基线失败") + return None + + if failed_paths: + # 部分路径失败：成功路径已合并，失败路径保留旧基线，下轮重试 + self._note_failure(storage, f"部分路径快照失败: {','.join(failed_paths)}") + else: + self._note_success(storage) + return file_count + + except Exception as e: + logger.error(f"轮询监控 {storage}:{monitor_scope} 出现错误：{e}\n{traceback.format_exc()}") + self._note_failure(storage, str(e)) + return None + + def _handle_changes(self, storage: str, old_snapshot: dict, new_snapshot: dict): + """ + 比对快照并把变化文件送入整理链。 + :param storage: 存储名称 + :param old_snapshot: 旧基线 + :param new_snapshot: 本轮增量快照 + """ + changes = SnapshotStore.compare(old_snapshot, new_snapshot) + added_files = [ + file_path + for file_path in changes['added'] + if self._dispatcher.is_transfer_candidate_path(Path(file_path)) + ] + modified_files = [ + file_path + for file_path in changes['modified'] + if self._dispatcher.is_transfer_candidate_path(Path(file_path)) + ] + + # 处理新增文件 + handled_added_count = 0 + for new_file in added_files: + file_info = new_snapshot.get(new_file, {}) + file_size = file_info.get('size', 0) if isinstance(file_info, dict) else file_info + if self._dispatcher.handle_file(storage=storage, event_path=Path(new_file), file_size=file_size): + handled_added_count += 1 + + # 处理修改文件 + handled_modified_count = 0 + for modified_file in modified_files: + file_info = new_snapshot.get(modified_file, {}) + file_size = file_info.get('size', 0) if isinstance(file_info, dict) else file_info + if self._dispatcher.handle_file(storage=storage, event_path=Path(modified_file), file_size=file_size): + handled_modified_count += 1 + + if handled_added_count or handled_modified_count: + logger.info(f"{storage} 发现 {handled_added_count} 个新增文件，{handled_modified_count} 个修改文件") + else: + logger.debug(f"{storage} 无文件变化") + + def force_full_scan(self, storage: str, mon_path: Path) -> bool: + """ + 强制全量扫描并处理所有文件（包括已存在的文件）。 + :param storage: 存储名称 + :param mon_path: 监控路径 + :return: 是否成功 + """ + try: + logger.info(f"开始强制全量扫描: {storage}:{mon_path}") + + # 生成快照 + new_snapshot = StorageChain().snapshot_storage( + storage=storage, + path=mon_path, + last_snapshot_time=0 # 全量扫描，不使用增量 + ) + + if new_snapshot is None: + logger.warn(f"获取 {storage}:{mon_path} 快照失败") + return False + + file_count = len(new_snapshot) + logger.info(f"{storage}:{mon_path} 全量扫描完成，发现 {file_count} 个文件") + + # 处理所有文件 + processed_count = 0 + for file_path, file_info in new_snapshot.items(): + try: + if not self._dispatcher.is_transfer_candidate_path(Path(file_path)): + continue + file_size = file_info.get('size', 0) if isinstance(file_info, dict) else file_info + if self._dispatcher.handle_file(storage=storage, event_path=Path(file_path), + file_size=file_size): + processed_count += 1 + except Exception as e: + logger.error(f"处理文件 {file_path} 失败: {e}") + continue + + logger.info(f"{storage}:{mon_path} 全量扫描完成，共处理 {processed_count}/{file_count} 个文件") + + # 全量扫描覆盖单个路径，与已有基线合并后落盘，避免覆盖其他监控路径的基线 + old_snapshot_data, load_ok = self._store.load_checked(storage) + old_snapshot = old_snapshot_data.get('snapshot', {}) if (load_ok and old_snapshot_data) else {} + merged_snapshot = {**old_snapshot, **new_snapshot} + self._store.save(storage, merged_snapshot, len(merged_snapshot)) + + return True + + except Exception as e: + logger.error(f"强制全量扫描失败: {storage}:{mon_path} - {e}") + return False diff --git a/app/monitor/snapshot.py b/app/monitor/snapshot.py new file mode 100644 index 000000000..0e21dda1b --- /dev/null +++ b/app/monitor/snapshot.py @@ -0,0 +1,148 @@ +import json +import time +from typing import Dict, List, Optional, Tuple + +from app.core.cache import FileCache +from app.core.config import settings +from app.log import logger + + +class SnapshotStore: + """ + 远程目录监控快照的存取与比对。 + """ + + def __init__(self, cache: Optional[FileCache] = None): + """ + 初始化快照存储。 + :param cache: 快照文件缓存，默认使用 CACHE_PATH/snapshots + """ + self._cache = cache if cache is not None else FileCache(base=settings.CACHE_PATH / "snapshots") + + def save(self, storage: str, snapshot: Dict, file_count: int = 0, + last_snapshot_time: Optional[float] = None) -> bool: + """ + 保存快照到文件缓存。 + :param storage: 存储名称 + :param snapshot: 快照数据 + :param file_count: 文件数量，用于调整监控间隔 + :param last_snapshot_time: 上次快照时间戳 + :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_data = { + 'timestamp': snapshot_time, + 'file_count': file_count, + 'snapshot': snapshot + } + cache_key = f"{storage}_snapshot" + snapshot_json = json.dumps(snapshot_data, ensure_ascii=False, indent=2) + self._cache.set(cache_key, snapshot_json.encode('utf-8'), region="snapshots") + logger.debug(f"快照已保存到缓存: {storage}") + return True + except Exception as e: + logger.error(f"保存快照失败: {e}") + return False + + def load_checked(self, storage: str) -> Tuple[Optional[Dict], bool]: + """ + 从文件缓存加载快照，并区分「快照不存在」与「读取失败」。 + 读取失败时不能当作首次快照处理，否则会静默丢弃已有基线。 + :param storage: 存储名称 + :return: (快照数据或None, 是否读取成功) + """ + try: + cache_key = f"{storage}_snapshot" + snapshot_data = self._cache.get(cache_key, region="snapshots") + if snapshot_data: + data = json.loads(snapshot_data.decode('utf-8')) + logger.debug(f"成功加载快照: {storage}, 包含 {len(data.get('snapshot', {}))} 个文件") + return data, True + logger.debug(f"快照文件不存在: {storage}") + return None, True + except Exception as e: + logger.error(f"加载快照失败: {e}") + return None, False + + def load(self, storage: str) -> Optional[Dict]: + """ + 从文件缓存加载快照。 + :param storage: 存储名称 + :return: 快照数据或None + """ + data, _ = self.load_checked(storage) + return data + + def reset(self, storage: str) -> bool: + """ + 重置快照，强制下次扫描时重新建立基准。 + :param storage: 存储名称 + :return: 是否成功 + """ + try: + cache_key = f"{storage}_snapshot" + if self._cache.exists(cache_key, region="snapshots"): + self._cache.delete(cache_key, region="snapshots") + logger.info(f"快照已重置: {storage}") + return True + logger.debug(f"快照文件不存在，无需重置: {storage}") + return True + except Exception as e: + logger.error(f"重置快照失败: {storage} - {e}") + return False + + @staticmethod + def compare(old_snapshot: Dict, new_snapshot: Dict) -> Dict[str, List]: + """ + 比对快照，找出变化的文件（只处理新增和修改，不处理删除）。 + :param old_snapshot: 旧快照 + :param new_snapshot: 新快照 + :return: 变化信息 + """ + changes = { + 'added': [], + 'modified': [] + } + + old_files = set(old_snapshot.keys()) + new_files = set(new_snapshot.keys()) + + # 新增文件 + changes['added'] = list(new_files - old_files) + + # 修改文件（大小或时间变化） + for file_path in old_files & new_files: + old_info = old_snapshot[file_path] + new_info = new_snapshot[file_path] + + # 检查文件大小变化 + old_size = old_info.get('size', 0) if isinstance(old_info, dict) else old_info + new_size = new_info.get('size', 0) if isinstance(new_info, dict) else new_info + + # 检查修改时间变化（如果有的话） + old_time = old_info.get('modify_time', 0) if isinstance(old_info, dict) else 0 + new_time = new_info.get('modify_time', 0) if isinstance(new_info, dict) else 0 + + if old_size != new_size or (old_time and new_time and old_time != new_time): + changes['modified'].append(file_path) + + return changes + + @staticmethod + def adjust_interval(file_count: int) -> int: + """ + 根据文件数量动态调整监控间隔。 + :param file_count: 文件数量 + :return: 监控间隔（分钟） + """ + if file_count < 100: + return 5 # 5分钟 + elif file_count < 500: + return 10 # 10分钟 + elif file_count < 1000: + return 15 # 15分钟 + else: + return 30 # 30分钟 diff --git a/app/monitor/syslimits.py b/app/monitor/syslimits.py new file mode 100644 index 000000000..4e54393a5 --- /dev/null +++ b/app/monitor/syslimits.py @@ -0,0 +1,134 @@ +import os +import platform +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +from app.log import logger +from app.utils.system import SystemUtils + + +def count_directory_entries(directory: Path, max_check: int = 10000) -> Tuple[int, int]: + """ + 统计目录下的文件与子目录数量（用于检测是否超过系统限制）。 + :param directory: 目录路径 + :param max_check: 最大检查文件数量，避免长时间阻塞 + :return: (文件数量, 目录数量) + """ + file_count = 0 + dir_count = 0 + try: + for _, dirs, files in os.walk(str(directory)): + file_count += len(files) + dir_count += len(dirs) + if file_count > max_check: + break + except Exception as err: + logger.debug(f"统计目录规模失败: {err}") + return file_count, dir_count + + +def count_directory_files(directory: Path, max_check: int = 10000) -> int: + """ + 统计目录下的文件数量。 + :param directory: 目录路径 + :param max_check: 最大检查数量，避免长时间阻塞 + :return: 文件数量 + """ + file_count, _ = count_directory_entries(directory, max_check=max_check) + return file_count + + +def check_system_limits() -> Dict[str, Any]: + """ + 检查系统监控相关限制。 + :return: 系统限制信息 + """ + limits = { + 'max_user_watches': 0, + 'max_user_instances': 0, + 'warnings': [] + } + + try: + if platform.system() == 'Linux': + # 检查 inotify 限制 + try: + with open('/proc/sys/fs/inotify/max_user_watches', 'r', encoding='utf-8', errors='replace') as f: + limits['max_user_watches'] = int(f.read().strip()) + except Exception as e: + logger.debug(f"读取 inotify 限制失败: {e}") + limits['max_user_watches'] = 8192 # 默认值 + + try: + with open('/proc/sys/fs/inotify/max_user_instances', 'r', encoding='utf-8', errors='replace') as f: + limits['max_user_instances'] = int(f.read().strip()) + except Exception as e: + logger.debug(f"读取 inotify 实例限制失败: {e}") + except Exception as e: + limits['warnings'].append(f"检查系统限制时出错: {e}") + + return limits + + +def get_system_optimization_tips() -> List[str]: + """ + 获取系统优化建议。 + :return: 优化建议列表 + """ + tips = [] + system = platform.system() + + if system == 'Linux': + tips.extend([ + "增加 inotify 监控数量限制:", + "echo fs.inotify.max_user_watches=524288 | sudo tee -a /etc/sysctl.conf", + "echo fs.inotify.max_user_instances=524288 | sudo tee -a /etc/sysctl.conf", + "sudo sysctl -p", + "", + "如果在Docker中运行，请在宿主机上执行以上命令" + ]) + elif system == 'Darwin': + tips.extend([ + "macOS 系统优化建议:", + "sudo sysctl kern.maxfiles=65536", + "sudo sysctl kern.maxfilesperproc=32768", + "ulimit -n 32768" + ]) + elif system == 'Windows': + tips.extend([ + "Windows 系统优化建议:", + "1. 关闭不必要的实时保护软件对监控目录的扫描", + "2. 将监控目录添加到Windows Defender排除列表", + "3. 确保有足够的可用内存" + ]) + + return tips + + +def decide_monitor_mode(directory: Path, + monitor_mode: str) -> Tuple[bool, str, Optional[Dict[str, Any]], Optional[int]]: + """ + 决策监控模式。兼容模式与网络文件系统直接短路，只有快速模式候选才统计 + 目录规模与系统限制，避免启动期对网络挂载做无谓的全量遍历。 + + inotify 的 max_user_watches 按监视点（目录）计数，因此用目录数量而不是 + 文件数量与上限比较。 + + :param directory: 监控目录 + :param monitor_mode: 配置的监控模式 + :return: (是否使用轮询, 原因, 系统限制信息或None, 文件数量或None) + """ + if monitor_mode == "compatibility": + return True, "用户配置为兼容模式", None, None + + # 检查网络文件系统 + if SystemUtils.is_network_filesystem(directory): + return True, "检测到网络文件系统，建议使用兼容模式", None, None + + limits = check_system_limits() + file_count, dir_count = count_directory_entries(directory) + max_watches = limits.get('max_user_watches') + if max_watches and dir_count > max_watches * 0.8: + return (True, f"目录数量({dir_count})接近 inotify 监控上限({max_watches})", + limits, file_count) + return False, "使用快速模式", limits, file_count diff --git a/app/monitor/watcher.py b/app/monitor/watcher.py new file mode 100644 index 000000000..b9638b64e --- /dev/null +++ b/app/monitor/watcher.py @@ -0,0 +1,303 @@ +import threading +import time +import traceback +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +from watchfiles import Change, DefaultFilter, watch + +from app.log import logger + + +@dataclass(frozen=True) +class DirectoryChangeEvent: + """ + 目录文件变化事件，隔离底层 watchfiles 事件结构。 + """ + change_type: Change + src_path: str + is_directory: bool + + +class LocalDirectoryWatcher: + """ + 基于 watchfiles 的本地目录监控线程。 + """ + _HANDLE_CHANGES = {Change.added, Change.modified} + # 监控循环异常退出后的重启退避秒数，网络存储/FUSE 挂载抖动通常是暂时的 + RESTART_BACKOFF = (5, 15, 30, 60, 120, 300) + # 单次监控循环存活超过该秒数视为已恢复，重置退避 + HEALTHY_UPTIME = 60 + # 超过该秒数监控循环没有任何活动，判定为静默失效 + STALL_TIMEOUT = 600 + # 轮询模式目录扫描间隔（毫秒）：本地磁盘用 watchfiles 默认值 + POLL_DELAY_LOCAL_MS = 300 + # 网络/FUSE 挂载轮询降频，减少监控自身对挂载后端的持续 stat 压力 + POLL_DELAY_NETWORK_MS = 5000 + + def __init__(self, mon_path: Path, callback: Any, force_polling: Optional[bool] = None, + poll_delay_ms: Optional[int] = None): + """ + 初始化本地目录监控。 + :param mon_path: 监控目录 + :param callback: 目录变化回调对象 + :param force_polling: 是否强制使用轮询模式，None 表示由 watchfiles 自动选择 + :param poll_delay_ms: 轮询模式目录扫描间隔（毫秒），仅轮询时生效 + """ + self._watch_path = mon_path + self._callback = callback + self._force_polling = force_polling + self._poll_delay_ms = poll_delay_ms or self.POLL_DELAY_LOCAL_MS + self._stop_event = threading.Event() + self._thread: Optional[threading.Thread] = None + self._watch_filter = DefaultFilter() + # 最近一次监控循环活动时间（monotonic），用于检测静默失效 + self._last_activity: float = 0.0 + # 累计自动重启次数 + self._restart_count: int = 0 + + @property + def watch_path(self) -> Path: + """ + 获取监控目录。 + :return: 监控目录 + """ + return self._watch_path + + @property + def force_polling(self) -> Optional[bool]: + """ + 获取监控模式配置，重建监控线程时沿用。 + :return: 是否强制轮询 + """ + return self._force_polling + + @property + def restart_count(self) -> int: + """ + 获取累计自动重启次数。 + :return: 自动重启次数 + """ + return self._restart_count + + @property + def poll_delay_ms(self) -> int: + """ + 获取轮询模式目录扫描间隔（毫秒），重建监控线程时沿用。 + :return: 扫描间隔 + """ + return self._poll_delay_ms + + def start(self): + """ + 启动本地目录监控线程。 + """ + if not self._watch_path.exists(): + raise FileNotFoundError(f"监控目录不存在: {self._watch_path}") + if not self._watch_path.is_dir(): + raise NotADirectoryError(f"监控路径不是目录: {self._watch_path}") + if self.is_alive(): + logger.info(f"本地目录监控已在运行中: {self._watch_path}") + return + self._stop_event.clear() + self._mark_activity() + self._thread = threading.Thread( + target=self._run, + name=f"MoviePilot-DirectoryWatcher-{self._watch_path.name}", + daemon=True + ) + self._thread.start() + + def stop(self): + """ + 请求停止本地目录监控线程。 + """ + self._stop_event.set() + + def join(self, timeout: Optional[float] = None): + """ + 等待本地目录监控线程退出。 + :param timeout: 最长等待秒数 + """ + if self._thread: + self._thread.join(timeout=timeout) + + def is_alive(self) -> bool: + """ + 判断监控线程是否仍在运行。 + :return: 线程存活状态 + """ + return bool(self._thread and self._thread.is_alive()) + + def is_stalled(self) -> bool: + """ + 判断监控线程是否已静默失效（线程存活但监控循环长时间无任何活动）。 + :return: 是否静默失效 + """ + if self._stop_event.is_set() or not self.is_alive(): + return False + if not self._last_activity: + return False + return (time.monotonic() - self._last_activity) > self.STALL_TIMEOUT + + def _mark_activity(self): + """ + 记录一次监控循环活动时间，作为静默失效检测的心跳。 + """ + self._last_activity = time.monotonic() + + def _run(self): + """ + 运行 watchfiles 主循环，异常时退避重启，避免一次故障导致监控永久停摆。 + """ + # 快速模式失败后降级为轮询，降级后的失败一律走退避重启 + force_polling = self._force_polling + attempt = 0 + while not self._stop_event.is_set(): + started_at = time.monotonic() + try: + self._mark_activity() + self._run_watch(force_polling=force_polling) + # 正常返回表示收到停止信号 + return + except Exception as err: + if self._stop_event.is_set(): + return + # 崩溃堆栈按 ERROR 级输出，生产环境 LOG_LEVEL=ERROR 时也能落盘 + logger.error(f"本地目录监控异常堆栈: {self._watch_path}\n{traceback.format_exc()}") + if force_polling is not True: + logger.warn(f"快速模式监控 {self._watch_path} 失败，将自动切换到兼容模式: {err}") + force_polling = True + continue + if time.monotonic() - started_at >= self.HEALTHY_UPTIME: + # 上一轮监控已稳定运行过，重新从最短间隔开始退避 + attempt = 0 + delay = self.RESTART_BACKOFF[min(attempt, len(self.RESTART_BACKOFF) - 1)] + attempt += 1 + self._restart_count += 1 + logger.error(f"本地目录监控发生错误，{delay} 秒后自动重启" + f"（累计第 {self._restart_count} 次）: {self._watch_path} - {err}") + if self._stop_event.wait(timeout=delay): + return + + def _run_watch(self, force_polling: Optional[bool]): + """ + 执行一次 watchfiles 监控循环。 + :param force_polling: 是否强制轮询 + """ + for changes in watch( + str(self._watch_path), + watch_filter=self._watch_filter, + stop_event=self._stop_event, + rust_timeout=1000, + yield_on_timeout=True, + force_polling=force_polling, + poll_delay_ms=self._poll_delay_ms, + recursive=True, + ignore_permission_denied=True): + self._mark_activity() + if self._stop_event.is_set(): + break + if not changes: + continue + self._handle_changes(changes) + self._mark_activity() + + def _handle_changes(self, changes: set[tuple[Change, str]]): + """ + 将 watchfiles 原始变更转换为目录监控事件。 + :param changes: watchfiles 返回的变更集合 + """ + changes = self._expand_added_directories(changes) + for change_type, path_str in sorted(changes, key=lambda item: item[1]): + # 批量整理可能持续较久，逐个文件刷新心跳，避免被误判为静默失效 + self._mark_activity() + if change_type not in self._HANDLE_CHANGES: + continue + event_path = Path(path_str) + event = self._build_event(change_type=change_type, event_path=event_path) + if not event or event.is_directory: + continue + file_size = self._get_file_size(event_path) + if file_size is None: + continue + text = self._change_text(change_type) + try: + self._callback.event_handler( + event=event, + text=text, + event_path=path_str, + file_size=file_size + ) + except Exception as err: + logger.error(f"处理本地目录监控事件失败: {path_str} - {err}") + + def _expand_added_directories(self, changes: set[tuple[Change, str]]) -> set[tuple[Change, str]]: + """ + 将整体移入监控范围的新增目录展开为内部文件事件。 + :param changes: watchfiles 返回的变更集合 + :return: 包含目录内新增文件的变更集合 + """ + expanded_changes = set(changes) + for change_type, path_str in changes: + if change_type != Change.added: + continue + event_path = Path(path_str) + try: + if not event_path.is_dir(): + continue + for nested_path in event_path.rglob("*"): + if not nested_path.is_file(): + continue + nested_path_str = nested_path.as_posix() + if self._watch_filter(Change.added, nested_path_str): + expanded_changes.add((Change.added, nested_path_str)) + except OSError as err: + logger.debug(f"扫描新增目录失败: {event_path} - {err}") + return expanded_changes + + @staticmethod + def _build_event(change_type: Change, event_path: Path) -> Optional[DirectoryChangeEvent]: + """ + 构建目录变化事件，路径已不存在时忽略。 + :param change_type: watchfiles 变化类型 + :param event_path: 变化路径 + :return: 目录变化事件 + """ + try: + is_directory = event_path.is_dir() + except OSError as err: + logger.debug(f"读取目录监控事件路径失败: {event_path} - {err}") + return None + if not event_path.exists(): + return None + return DirectoryChangeEvent( + change_type=change_type, + src_path=event_path.as_posix(), + is_directory=is_directory + ) + + @staticmethod + def _get_file_size(event_path: Path) -> Optional[int]: + """ + 读取事件文件大小，文件已消失时返回 None。 + :param event_path: 事件文件路径 + :return: 文件大小 + """ + try: + return event_path.stat().st_size + except OSError as err: + logger.debug(f"读取目录监控文件大小失败: {event_path} - {err}") + return None + + @staticmethod + def _change_text(change_type: Change) -> str: + """ + 转换 watchfiles 事件类型为日志文案。 + :param change_type: watchfiles 变化类型 + :return: 事件描述 + """ + if change_type == Change.modified: + return "修改" + return "新增" diff --git a/app/schemas/exception.py b/app/schemas/exception.py index edc2e4556..9165339b1 100644 --- a/app/schemas/exception.py +++ b/app/schemas/exception.py @@ -36,3 +36,12 @@ class OperationInterrupted(KeyboardInterrupt): 用于表示操作被中断 """ pass + + +class StorageQueryError(Exception): + """ + 用于表示存储查询无法确认结果的异常类。 + 当文件信息查询因网络、限流或接口错误失败（区别于「确认不存在」）时抛出， + 调用方不应把该状态当作文件不存在处理。 + """ + pass diff --git a/tests/test_monitor_resilience.py b/tests/test_monitor_resilience.py new file mode 100644 index 000000000..779797a35 --- /dev/null +++ b/tests/test_monitor_resilience.py @@ -0,0 +1,273 @@ +from pathlib import Path +from unittest.mock import MagicMock + +from app.monitor import LocalDirectoryWatcher, Monitor + + +def _build_watcher(tmp_path, force_polling): + """ + 构造测试用目录监控。 + :param tmp_path: 监控目录 + :param force_polling: 是否强制轮询 + :return: 目录监控 + """ + return LocalDirectoryWatcher(tmp_path, callback=MagicMock(), force_polling=force_polling) + + +def test_run_retries_with_backoff_in_compatibility_mode(tmp_path, monkeypatch): + """ + 兼容模式下监控循环抛异常后应退避重启，而不是直接结束线程。 + """ + monkeypatch.setattr(LocalDirectoryWatcher, "RESTART_BACKOFF", (0,)) + watcher = _build_watcher(tmp_path, force_polling=True) + calls = [] + + def fake_run_watch(force_polling): + """ + 模拟底层监控循环持续抛出 FUSE 错误。 + """ + calls.append(force_polling) + if len(calls) >= 3: + watcher.stop() + raise OSError(131, "State not recoverable") + + monkeypatch.setattr(watcher, "_run_watch", fake_run_watch) + + watcher._run() + + assert calls == [True, True, True] + assert watcher.restart_count == 2 + + +def test_run_falls_back_to_polling_before_backoff(tmp_path, monkeypatch): + """ + 快速模式失败应先降级为兼容模式重试，且降级不计入退避重启次数。 + """ + monkeypatch.setattr(LocalDirectoryWatcher, "RESTART_BACKOFF", (0,)) + watcher = _build_watcher(tmp_path, force_polling=None) + calls = [] + + def fake_run_watch(force_polling): + """ + 模拟快速模式与兼容模式先后失败。 + """ + calls.append(force_polling) + if len(calls) >= 2: + watcher.stop() + raise OSError("inotify watch limit reached") + + monkeypatch.setattr(watcher, "_run_watch", fake_run_watch) + + watcher._run() + + assert calls == [None, True] + assert watcher.restart_count == 0 + + +def test_run_returns_when_stop_requested(tmp_path, monkeypatch): + """ + 收到停止信号后监控循环正常返回，不应触发重启。 + """ + watcher = _build_watcher(tmp_path, force_polling=True) + calls = [] + + def fake_run_watch(force_polling): + """ + 模拟收到停止信号后正常退出的监控循环。 + """ + calls.append(force_polling) + + monkeypatch.setattr(watcher, "_run_watch", fake_run_watch) + + watcher._run() + + assert calls == [True] + assert watcher.restart_count == 0 + + +def test_is_stalled_detects_silent_failure(tmp_path): + """ + 监控线程存活但长时间无活动时应判定为静默失效。 + """ + watcher = _build_watcher(tmp_path, force_polling=True) + + # 线程未启动时不做判定 + assert watcher.is_stalled() is False + + thread = MagicMock() + thread.is_alive.return_value = True + watcher._thread = thread + watcher._mark_activity() + assert watcher.is_stalled() is False + + watcher._last_activity -= LocalDirectoryWatcher.STALL_TIMEOUT + 1 + assert watcher.is_stalled() is True + + +def test_is_stalled_ignores_stopped_watcher(tmp_path): + """ + 已请求停止的监控不应再被判定为静默失效。 + """ + watcher = _build_watcher(tmp_path, force_polling=True) + thread = MagicMock() + thread.is_alive.return_value = True + watcher._thread = thread + watcher._mark_activity() + watcher._last_activity -= LocalDirectoryWatcher.STALL_TIMEOUT + 1 + + watcher.stop() + + assert watcher.is_stalled() is False + + +def _build_monitor(monkeypatch, put_recorder): + """ + 构造测试用 Monitor 骨架，绕过单例初始化。 + :param monkeypatch: pytest monkeypatch + :param put_recorder: 消息推送记录器 + :return: Monitor 骨架 + """ + from threading import Lock + monkeypatch.setattr("app.monitor.monitor.MessageHelper", MagicMock(return_value=put_recorder)) + monitor = object.__new__(Monitor) + monitor._watchers = [] + monitor._watcher_lock = Lock() + monitor._pending_locals = [] + monitor._alerted_paths = set() + monitor._restart_marks = {} + monitor._stable_cycles = {} + return monitor + + +def _fake_watcher(mon_path, alive=True, stalled=False, restart_count=0): + """ + 构造测试用监控线程替身。 + :param mon_path: 监控目录 + :param alive: 线程是否存活 + :param stalled: 是否静默失效 + :param restart_count: 自动重启次数 + :return: 监控线程替身 + """ + watcher = MagicMock() + watcher.watch_path = mon_path + watcher.is_alive.return_value = alive + watcher.is_stalled.return_value = stalled + watcher.restart_count = restart_count + return watcher + + +def test_watchdog_rebuilds_dead_watcher(tmp_path, monkeypatch): + """ + 监控线程退出后健康检查应重建线程并告警。 + """ + put_recorder = MagicMock() + monitor = _build_monitor(monkeypatch, put_recorder) + watcher = _fake_watcher(tmp_path, alive=False) + monitor._watchers = [watcher] + rebuild = MagicMock() + setattr(monitor, "_Monitor__rebuild_watcher", rebuild) + + monitor._Monitor__check_watchers() + + rebuild.assert_called_once_with(watcher) + put_recorder.put.assert_called_once() + + +def test_watchdog_rebuilds_stalled_watcher(tmp_path, monkeypatch): + """ + 静默失效的监控线程也应被健康检查重建。 + """ + put_recorder = MagicMock() + monitor = _build_monitor(monkeypatch, put_recorder) + watcher = _fake_watcher(tmp_path, alive=True, stalled=True) + monitor._watchers = [watcher] + rebuild = MagicMock() + setattr(monitor, "_Monitor__rebuild_watcher", rebuild) + + monitor._Monitor__check_watchers() + + rebuild.assert_called_once_with(watcher) + + +def test_watchdog_alerts_on_restart_and_recovers_after_stable_window(tmp_path, monkeypatch): + """ + 自动重启应触发一次告警，恢复消息需等满稳定窗口，避免来回刷屏。 + """ + put_recorder = MagicMock() + monitor = _build_monitor(monkeypatch, put_recorder) + watcher = _fake_watcher(tmp_path, alive=True, stalled=False, restart_count=1) + monitor._watchers = [watcher] + + monitor._Monitor__check_watchers() + assert str(tmp_path) in monitor._alerted_paths + assert put_recorder.put.call_count == 1 + + for _ in range(Monitor.RECOVERY_STABLE_CYCLES - 1): + monitor._Monitor__check_watchers() + assert str(tmp_path) in monitor._alerted_paths + + monitor._Monitor__check_watchers() + assert str(tmp_path) not in monitor._alerted_paths + assert put_recorder.put.call_count == 2 + + +def test_retry_pending_locals_backs_off(tmp_path, monkeypatch): + """ + 启动失败的监控重试应按失败次数退避，避免持续故障时刷屏。 + """ + put_recorder = MagicMock() + monitor = _build_monitor(monkeypatch, put_recorder) + monitor._pending_locals = [{"mon_path": tmp_path, "monitor_mode": "compatibility"}] + start = MagicMock(return_value=False) + setattr(monitor, "_Monitor__start_local_monitor", start) + + for _ in range(6): + monitor._Monitor__retry_pending_locals() + + assert start.call_count == 3 + + +def test_dispatcher_retries_after_history_query_failure(monkeypatch): + """ + 整理历史查询失败应登记待重试，重试成功后进入整理链并清除登记。 + """ + from app.monitor.dispatcher import TransferDispatcher + dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={}) + event_path = Path("/downloads/movie.mkv") + history = MagicMock(side_effect=[None, False]) + monkeypatch.setattr(dispatcher, "_has_transfer_history", history) + transfer_chain_instance = MagicMock() + monkeypatch.setattr("app.monitor.dispatcher.TransferChain", + MagicMock(return_value=transfer_chain_instance)) + + # 首次查询失败：不整理，登记待重试 + assert dispatcher.handle_file(storage="local", event_path=event_path, file_size=1) is False + assert len(dispatcher._pending_retries) == 1 + transfer_chain_instance.do_transfer.assert_not_called() + + # 模拟 TTL 缓存过期后由健康检查驱动重试 + dispatcher._cache.clear() + dispatcher.retry_pending() + + transfer_chain_instance.do_transfer.assert_called_once() + assert dispatcher._pending_retries == {} + + +def test_dispatcher_drops_pending_after_max_attempts(monkeypatch): + """ + 历史查询持续失败达到上限后应放弃重试，避免队列无限累积。 + """ + from app.monitor.dispatcher import TransferDispatcher + dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={}) + event_path = Path("/downloads/movie.mkv") + monkeypatch.setattr(dispatcher, "_has_transfer_history", MagicMock(return_value=None)) + + dispatcher.handle_file(storage="local", event_path=event_path, file_size=1) + key = f"local:{event_path.as_posix()}" + assert key in dispatcher._pending_retries + dispatcher._pending_retries[key]["attempts"] = TransferDispatcher.MAX_RETRY_ATTEMPTS - 1 + + dispatcher._cache.clear() + dispatcher.retry_pending() + + assert dispatcher._pending_retries == {} diff --git a/tests/test_monitor_snapshot_semantics.py b/tests/test_monitor_snapshot_semantics.py new file mode 100644 index 000000000..516616b3a --- /dev/null +++ b/tests/test_monitor_snapshot_semantics.py @@ -0,0 +1,177 @@ +from pathlib import Path +from unittest.mock import MagicMock + +from app.monitor.poller import RemotePoller +from app.monitor.watcher import LocalDirectoryWatcher + + +def _build_poller(alert_cb=None): + """ + 构造测试用远程轮询监控。 + :param alert_cb: 告警回调替身 + :return: (poller, store, dispatcher) + """ + store = MagicMock() + store.save.return_value = True + dispatcher = MagicMock() + dispatcher.is_transfer_candidate_path.return_value = True + dispatcher.handle_file.return_value = True + poller = RemotePoller(store=store, dispatcher=dispatcher, alert_cb=alert_cb) + return poller, store, dispatcher + + +def _mock_storage_chain(monkeypatch, side_effect): + """ + 替换 StorageChain 的快照返回。 + :param monkeypatch: pytest monkeypatch + :param side_effect: snapshot_storage 的返回序列 + :return: StorageChain 实例替身 + """ + chain_instance = MagicMock() + chain_instance.snapshot_storage.side_effect = side_effect + monkeypatch.setattr("app.monitor.poller.StorageChain", MagicMock(return_value=chain_instance)) + return chain_instance + + +BASELINE = { + 'timestamp': 100, + 'file_count': 1, + 'snapshot': {'/mon/a.mkv': {'size': 1, 'modify_time': 100}} +} + + +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}}]) + + 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'} + dispatcher.handle_file.assert_called_once() + assert dispatcher.handle_file.call_args.kwargs["event_path"] == Path('/mon/b.mkv') + + +def test_poll_detects_modified_files(monkeypatch): + """ + 增量中已有文件的大小变化应作为修改事件分发，并更新基线。 + """ + poller, store, dispatcher = _build_poller() + store.load_checked.return_value = (dict(BASELINE), True) + _mock_storage_chain(monkeypatch, [{'/mon/a.mkv': {'size': 5, 'modify_time': 300}}]) + + file_count = poller.poll("u115", [Path("/mon")]) + + assert file_count == 1 + saved_snapshot = store.save.call_args.args[1] + assert saved_snapshot['/mon/a.mkv']['size'] == 5 + dispatcher.handle_file.assert_called_once() + + +def test_poll_partial_failure_merges_success_and_keeps_baseline(monkeypatch): + """ + 部分路径快照失败时，成功路径合并落盘，失败路径保留旧基线。 + """ + alert_cb = MagicMock() + poller, store, dispatcher = _build_poller(alert_cb) + store.load_checked.return_value = (dict(BASELINE), True) + _mock_storage_chain(monkeypatch, [None, {'/mon2/b.mkv': {'size': 2, 'modify_time': 200}}]) + + file_count = poller.poll("u115", [Path("/mon"), Path("/mon2")]) + + assert file_count == 2 + saved_snapshot = store.save.call_args.args[1] + assert set(saved_snapshot.keys()) == {'/mon/a.mkv', '/mon2/b.mkv'} + # 单次失败未达告警阈值 + alert_cb.assert_not_called() + + +def test_poll_all_paths_failed_skips_save(monkeypatch): + """ + 全部路径快照失败时本轮不落盘，基线保持不变。 + """ + poller, store, dispatcher = _build_poller() + store.load_checked.return_value = (dict(BASELINE), True) + _mock_storage_chain(monkeypatch, [None]) + + assert poller.poll("u115", [Path("/mon")]) is None + store.save.assert_not_called() + dispatcher.handle_file.assert_not_called() + + +def test_poll_first_snapshot_failure_builds_no_empty_baseline(monkeypatch): + """ + 首次快照失败时不得落盘空基线，否则下一轮会把全部存量当作新增。 + """ + poller, store, dispatcher = _build_poller() + store.load_checked.return_value = (None, True) + _mock_storage_chain(monkeypatch, [None]) + + assert poller.poll("u115", [Path("/mon")]) is None + store.save.assert_not_called() + + +def test_poll_first_snapshot_success_saves_baseline_without_dispatch(monkeypatch): + """ + 首次快照成功仅建立基准，不应处理存量文件。 + """ + poller, store, dispatcher = _build_poller() + store.load_checked.return_value = (None, True) + _mock_storage_chain(monkeypatch, [{'/mon/a.mkv': {'size': 1, 'modify_time': 100}}]) + + assert poller.poll("u115", [Path("/mon")]) == 1 + store.save.assert_called_once() + dispatcher.handle_file.assert_not_called() + + +def test_poll_load_error_skips_round(monkeypatch): + """ + 基线读取失败不能当作首次快照，应跳过本轮避免丢弃已有基线。 + """ + poller, store, dispatcher = _build_poller() + store.load_checked.return_value = (None, False) + chain = _mock_storage_chain(monkeypatch, [{'/mon/a.mkv': {'size': 1, 'modify_time': 100}}]) + + assert poller.poll("u115", [Path("/mon")]) is None + chain.snapshot_storage.assert_not_called() + store.save.assert_not_called() + + +def test_poll_failure_alert_threshold_and_recovery(monkeypatch): + """ + 连续异常达到阈值只告警一次，恢复后推送恢复消息。 + """ + alert_cb = MagicMock() + poller, store, dispatcher = _build_poller(alert_cb) + store.load_checked.return_value = (dict(BASELINE), True) + _mock_storage_chain( + monkeypatch, + [None] * RemotePoller.FAILURE_ALERT_THRESHOLD + [{'/mon/b.mkv': {'size': 2, 'modify_time': 200}}] + ) + + for _ in range(RemotePoller.FAILURE_ALERT_THRESHOLD): + poller.poll("u115", [Path("/mon")]) + assert alert_cb.call_count == 1 + + poller.poll("u115", [Path("/mon")]) + assert alert_cb.call_count == 2 + assert "已恢复" in alert_cb.call_args.args[1] + + +def test_watcher_poll_delay_defaults_and_override(tmp_path): + """ + 轮询扫描间隔默认取本地值，显式传入网络值时生效。 + """ + default_watcher = LocalDirectoryWatcher(tmp_path, callback=MagicMock(), force_polling=True) + assert default_watcher.poll_delay_ms == LocalDirectoryWatcher.POLL_DELAY_LOCAL_MS + + network_watcher = LocalDirectoryWatcher( + tmp_path, callback=MagicMock(), force_polling=True, + poll_delay_ms=LocalDirectoryWatcher.POLL_DELAY_NETWORK_MS + ) + assert network_watcher.poll_delay_ms == LocalDirectoryWatcher.POLL_DELAY_NETWORK_MS diff --git a/tests/test_monitor_watchfiles.py b/tests/test_monitor_watchfiles.py index 2f008c031..a7a9075ac 100644 --- a/tests/test_monitor_watchfiles.py +++ b/tests/test_monitor_watchfiles.py @@ -4,6 +4,7 @@ from unittest.mock import MagicMock from watchfiles import Change from app.monitor import DirectoryChangeEvent, LocalDirectoryWatcher, Monitor +from app.monitor.dispatcher import TransferDispatcher class CallbackRecorder: @@ -28,6 +29,20 @@ class CallbackRecorder: self.events.append((event, text, event_path, file_size)) +def _build_monitor_with_dispatcher(handle_file: MagicMock = None): + """ + 构造带分发器的测试用 Monitor 骨架。 + :param handle_file: 替换分发器 handle_file 的替身 + :return: (Monitor 骨架, 分发器) + """ + monitor = object.__new__(Monitor) + dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={}) + if handle_file is not None: + dispatcher.handle_file = handle_file + monitor._dispatcher = dispatcher + return monitor, dispatcher + + def test_handle_changes_dispatches_added_and_modified_files(tmp_path): """ 新增和修改文件应转换成目录监控整理回调。 @@ -120,10 +135,8 @@ def test_event_handler_routes_file_events_to_transfer_handler(): """ 文件事件应继续按 local 存储交给整理流程。 """ - monitor = object.__new__(Monitor) - monitor.all_exts = [".mkv"] handle_file = MagicMock() - setattr(monitor, "_Monitor__handle_file", handle_file) + monitor, _ = _build_monitor_with_dispatcher(handle_file) event_path = Path("/downloads/movie.mkv") event = DirectoryChangeEvent( change_type=Change.added, @@ -149,10 +162,8 @@ def test_event_handler_ignores_directory_events(): """ 目录事件不应进入文件整理流程。 """ - monitor = object.__new__(Monitor) - monitor.all_exts = [".mkv"] handle_file = MagicMock() - setattr(monitor, "_Monitor__handle_file", handle_file) + monitor, _ = _build_monitor_with_dispatcher(handle_file) event_path = Path("/downloads/folder") event = DirectoryChangeEvent( change_type=Change.added, @@ -173,10 +184,8 @@ def test_event_handler_ignores_download_temp_files(): """ 下载器临时文件不应进入整理流程。 """ - monitor = object.__new__(Monitor) - monitor.all_exts = [".mkv"] handle_file = MagicMock() - setattr(monitor, "_Monitor__handle_file", handle_file) + monitor, _ = _build_monitor_with_dispatcher(handle_file) event_path = Path("/downloads/movie.mkv.!qB") event = DirectoryChangeEvent( change_type=Change.modified, @@ -198,10 +207,8 @@ def test_event_handler_ignores_non_transferable_files(): """ 非可整理后缀文件不应进入整理流程。 """ - monitor = object.__new__(Monitor) - monitor.all_exts = [".mkv"] handle_file = MagicMock() - setattr(monitor, "_Monitor__handle_file", handle_file) + monitor, _ = _build_monitor_with_dispatcher(handle_file) event_path = Path("/downloads/movie.nfo") event = DirectoryChangeEvent( change_type=Change.added, @@ -223,9 +230,7 @@ def test_handle_file_skips_transfer_when_history_exists(monkeypatch): """ 已有整理记录的源文件不应再次进入整理链。 """ - monitor = object.__new__(Monitor) - monitor.all_exts = [".mkv"] - monitor._cache = {} + dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={}) event_path = Path("/downloads/movie.mkv") lookups = [] @@ -244,12 +249,12 @@ def test_handle_file_skips_transfer_when_history_exists(monkeypatch): transfer_chain = MagicMock() logger_info = MagicMock() logger_debug = MagicMock() - monkeypatch.setattr("app.monitor.TransferHistoryOper", FakeTransferHistoryOper) - monkeypatch.setattr("app.monitor.TransferChain", transfer_chain) - monkeypatch.setattr("app.monitor.logger.info", logger_info) - monkeypatch.setattr("app.monitor.logger.debug", logger_debug) + monkeypatch.setattr("app.monitor.dispatcher.TransferHistoryOper", FakeTransferHistoryOper) + monkeypatch.setattr("app.monitor.dispatcher.TransferChain", transfer_chain) + monkeypatch.setattr("app.monitor.dispatcher.logger.info", logger_info) + monkeypatch.setattr("app.monitor.dispatcher.logger.debug", logger_debug) - handled = monitor._Monitor__handle_file( + handled = dispatcher.handle_file( storage="local", event_path=event_path, file_size=1024, @@ -266,9 +271,7 @@ def test_handle_file_invokes_transfer_when_history_missing(monkeypatch): """ 没有整理记录的源文件应继续进入整理链。 """ - monitor = object.__new__(Monitor) - monitor.all_exts = [".mkv"] - monitor._cache = {} + dispatcher = TransferDispatcher(all_exts=[".mkv"], cache={}) event_path = Path("/downloads/movie.mkv") class FakeTransferHistoryOper: @@ -284,10 +287,10 @@ def test_handle_file_invokes_transfer_when_history_missing(monkeypatch): transfer_chain_instance = MagicMock() transfer_chain = MagicMock(return_value=transfer_chain_instance) - monkeypatch.setattr("app.monitor.TransferHistoryOper", FakeTransferHistoryOper) - monkeypatch.setattr("app.monitor.TransferChain", transfer_chain) + monkeypatch.setattr("app.monitor.dispatcher.TransferHistoryOper", FakeTransferHistoryOper) + monkeypatch.setattr("app.monitor.dispatcher.TransferChain", transfer_chain) - handled = monitor._Monitor__handle_file( + handled = dispatcher.handle_file( storage="local", event_path=event_path, file_size=1024, diff --git a/tests/test_transfer_job_manager.py b/tests/test_transfer_job_manager.py index 044ac6cae..25aad1ef8 100644 --- a/tests/test_transfer_job_manager.py +++ b/tests/test_transfer_job_manager.py @@ -183,6 +183,7 @@ class TransferJobManagerTest(unittest.TestCase): target_oper = SimpleNamespace( get_folder=lambda path: target_folder, get_item=lambda path: None, + get_item_strict=lambda path: None, ) new_item, errmsg = TransHandler._TransHandler__transfer_command( @@ -243,6 +244,7 @@ class TransferJobManagerTest(unittest.TestCase): target_oper = SimpleNamespace( get_folder=lambda path: target_folder, get_item=lambda path: None, + get_item_strict=lambda path: None, ) with patch.object( @@ -313,6 +315,7 @@ class TransferJobManagerTest(unittest.TestCase): target_oper = SimpleNamespace( get_folder=lambda path: target_folder, get_item=lambda path: None, + get_item_strict=lambda path: None, ) in_meta = MetaVideo("Test.Show.S02E03") diff --git a/tests/test_transfer_overwrite_guard.py b/tests/test_transfer_overwrite_guard.py new file mode 100644 index 000000000..3d4a4da4f --- /dev/null +++ b/tests/test_transfer_overwrite_guard.py @@ -0,0 +1,186 @@ +from pathlib import Path +from unittest.mock import MagicMock + +import pytest + +from app.modules.filemanager.storages.alipan import AliPan +from app.modules.filemanager.storages.local import LocalStorage +from app.modules.filemanager.storages.rclone import Rclone +from app.modules.filemanager.storages.u115 import U115Pan +from app.schemas.exception import StorageQueryError + + +def _local() -> LocalStorage: + """ + 构造本地存储实例（跳过初始化）。 + """ + return object.__new__(LocalStorage) + + +def _u115() -> U115Pan: + """ + 构造 115 存储实例（跳过初始化）。 + """ + return object.__new__(U115Pan) + + +def _alipan(monkeypatch) -> AliPan: + """ + 构造阿里云盘存储实例（跳过初始化，_default_drive_id 为只读属性需在类级替换）。 + """ + monkeypatch.setattr(AliPan, "_default_drive_id", "drive-1", raising=False) + return object.__new__(AliPan) + + +def test_local_strict_missing_file_returns_none(tmp_path): + """ + 目标文件确实不存在时应确认为不存在，允许正常整理。 + """ + assert _local().get_item_strict(tmp_path / "missing.mkv") is None + + +def test_local_strict_existing_file_returns_item(tmp_path): + """ + 目标文件存在时应返回文件项。 + """ + target = tmp_path / "movie.mkv" + target.write_bytes(b"movie") + + item = _local().get_item_strict(target) + + assert item is not None + assert item.path == target.as_posix() + + +def test_local_strict_broken_symlink_returns_none(tmp_path): + """ + 失效软链接视为目标不存在，不应阻断整理。 + """ + target = tmp_path / "movie.mkv" + target.symlink_to(tmp_path / "gone.mkv") + + assert _local().get_item_strict(target) is None + + +def test_local_strict_raises_on_stat_error(tmp_path, monkeypatch): + """ + FUSE 挂载抖动导致 stat 失败时应抛出 StorageQueryError，拒绝覆盖。 + """ + target = tmp_path / "movie.mkv" + + def raise_stat_error(self, *args, **kwargs): + """ + 模拟 CloudDrive FUSE 挂载返回 ENOTRECOVERABLE。 + """ + raise OSError(131, "State not recoverable") + + monkeypatch.setattr(Path, "stat", raise_stat_error) + + with pytest.raises(StorageQueryError): + _local().get_item_strict(target) + + +def test_u115_strict_transport_failure_raises(): + """ + 115 请求失败（网络/限流重试用尽）时应抛出 StorageQueryError。 + """ + storage = _u115() + storage._request_api = MagicMock(return_value=None) + + with pytest.raises(StorageQueryError): + storage.get_item_strict(Path("/movie.mkv")) + + +def test_u115_get_item_keeps_swallowing_transport_failure(): + """ + 宽松版 get_item 行为保持兼容：请求失败仍返回 None。 + """ + storage = _u115() + storage._request_api = MagicMock(return_value=None) + + assert storage.get_item(Path("/movie.mkv")) is None + + +def test_u115_strict_confirmed_absent_returns_none(): + """ + 115 业务码返回记录不存在（data 为空）时应确认为不存在。 + """ + storage = _u115() + storage._request_api = MagicMock(return_value={"state": True, "code": 20004, "data": {}}) + + assert storage.get_item_strict(Path("/movie.mkv")) is None + + +def test_u115_strict_returns_item(): + """ + 115 返回有效文件数据时应构造文件项。 + """ + storage = _u115() + storage._request_api = MagicMock(return_value={"state": True, "code": 0, "data": { + "file_id": 123, + "file_category": "1", + "file_name": "movie.mkv", + "pick_code": "abc", + "size_byte": 1024, + "utime": 100, + }}) + + item = storage.get_item_strict(Path("/movie.mkv")) + + assert item is not None + assert item.fileid == "123" + assert item.size == 1024 + + +def test_alipan_strict_notfound_returns_none(monkeypatch): + """ + 阿里云盘 NotFound 系列错误码应确认为不存在。 + """ + storage = _alipan(monkeypatch) + storage._request_api = MagicMock(return_value={"code": "NotFound.File", "message": "not found"}) + + assert storage.get_item_strict(Path("/movie.mkv")) is None + + +def test_alipan_strict_other_error_raises(monkeypatch): + """ + 阿里云盘非 NotFound 的业务错误（如限流）应抛出 StorageQueryError。 + """ + storage = _alipan(monkeypatch) + storage._request_api = MagicMock(return_value={"code": "TooManyRequests", "message": "limit"}) + + with pytest.raises(StorageQueryError): + storage.get_item_strict(Path("/movie.mkv")) + + +def test_alipan_strict_transport_failure_raises(monkeypatch): + """ + 阿里云盘请求失败时应抛出 StorageQueryError。 + """ + storage = _alipan(monkeypatch) + storage._request_api = MagicMock(return_value=None) + + with pytest.raises(StorageQueryError): + storage.get_item_strict(Path("/movie.mkv")) + + +def test_alipan_strict_returns_item(monkeypatch): + """ + 阿里云盘返回有效数据时应构造文件项。 + """ + storage = _alipan(monkeypatch) + storage._request_api = MagicMock(return_value={"file_id": "f1", "name": "movie.mkv"}) + setattr(storage, "_AliPan__get_fileitem", MagicMock(return_value="ITEM")) + + assert storage.get_item_strict(Path("/movie.mkv")) == "ITEM" + + +def test_storage_base_strict_defaults_to_get_item(): + """ + 未覆写的存储沿用 get_item 判定，行为不变。 + """ + storage = object.__new__(Rclone) + storage.get_item = MagicMock(return_value=None) + + assert storage.get_item_strict(Path("/movie.mkv")) is None + storage.get_item.assert_called_once()