GithubBackup/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-09-07 16:36:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(monitor): 目录监控自愈、快照语义修正与覆盖保护闭环 (#6210)

Browse Source
This commit is contained in:
Aqr-K
2026-07-30 06:53:52 +08:00
committed by GitHub
parent e011b20210
commit cf80b551f9
20 changed files with 2353 additions and 1021 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/chain/transfer.py
+3 -1
View File
@@ -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
app/modules/filemanager/storages/__init__.py
+7
View File
@@ -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]:
"""
获取父目录
app/modules/filemanager/storages/alipan.py
+39 -15
View File
@@ -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]:
"""
获取指定路径的文件夹，如不存在则创建
app/modules/filemanager/storages/local.py
+18
View File
@@ -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]:
"""
获取文件详情
app/modules/filemanager/storages/u115.py
+46 -23
View File
@@ -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]:
"""
获取指定路径的文件夹，如不存在则创建
app/modules/filemanager/transhandler.py
+18 -2
View File
@@ -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
app/monitor.py
-954
View File
@@ -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()
app/monitor/__init__.py
+14
View File
@@ -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"]
app/monitor/dispatcher.py
+210
View File
@@ -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
app/monitor/monitor.py
+508
View File
@@ -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("本地目录监控服务已停止")
# 缓存与快照存储是共享后端的代理，生命周期由应用全局管理，这里不再关闭
app/monitor/poller.py
+228
View File
@@ -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
app/monitor/snapshot.py
+148
View File
@@ -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分钟
app/monitor/syslimits.py
+134
View File
@@ -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
app/monitor/watcher.py
+303
View File
@@ -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 "新增"
app/schemas/exception.py
+9
View File
@@ -36,3 +36,12 @@ class OperationInterrupted(KeyboardInterrupt):
用于表示操作被中断
"""
pass
class StorageQueryError(Exception):
"""
用于表示存储查询无法确认结果的异常类。
当文件信息查询因网络、限流或接口错误失败（区别于「确认不存在」）时抛出，
调用方不应把该状态当作文件不存在处理。
"""
pass
tests/test_monitor_resilience.py
+273
View File
@@ -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 == {}
tests/test_monitor_snapshot_semantics.py
+177
View File
@@ -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
tests/test_monitor_watchfiles.py
+29 -26
View File
@@ -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,
tests/test_transfer_job_manager.py
+3
View File
@@ -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")
tests/test_transfer_overwrite_guard.py
+186
View File
@@ -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()