refactor：优化目录监控

2026-02-03 02:25:32 +08:00 · 2025-06-30 13:16:05 +08:00
parent 8593a6cdd0
commit 4c03759d3f
1 changed files with 285 additions and 33 deletions
--- a/app/monitor.py
+++ b/app/monitor.py
@@ -1,6 +1,7 @@
 import json
 import platform
 import re
+import subprocess
 import threading
 import time
 import traceback
@@ -114,7 +115,7 @@ class Monitor(metaclass=Singleton):
                'snapshot': snapshot
            }
            with open(cache_file, 'w', encoding='utf-8') as f:
-                json.dump(snapshot_data, f, ensure_ascii=False, indent=2)
+                json.dump(snapshot_data, f, ensure_ascii=False, indent=2)  # noqa
            logger.debug(f"快照已保存到 {cache_file}")
        except Exception as e:
            logger.error(f"保存快照失败: {e}")
@@ -189,6 +190,159 @@ class Monitor(metaclass=Singleton):

        return changes

+    @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') 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') 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
+
+    def should_use_polling(self, 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 == "compatible":
+            return True, "用户配置为兼容模式"
+
+        # 检查网络文件系统
+        if self.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, "使用快速模式"
+
+    @staticmethod
+    def is_network_filesystem(directory: Path) -> bool:
+        """
+        检测是否为网络文件系统
+        :param directory: 目录路径
+        :return: 是否为网络文件系统
+        """
+        try:
+            system = platform.system()
+            if system == 'Linux':
+                # 检查挂载信息
+                result = subprocess.run(['df', '-T', str(directory)],
+                                        capture_output=True, text=True, timeout=5)
+                if result.returncode == 0:
+                    output = result.stdout.lower()
+                    network_fs = ['nfs', 'cifs', 'smbfs', 'fuse', 'sshfs', 'ftpfs']
+                    return any(fs in output for fs in network_fs)
+            elif system == 'Darwin':
+                # macOS 检查
+                result = subprocess.run(['df', '-T', str(directory)],
+                                        capture_output=True, text=True, timeout=5)
+                if result.returncode == 0:
+                    output = result.stdout.lower()
+                    return 'nfs' in output or 'smbfs' in output
+            elif system == 'Windows':
+                # Windows 检查网络驱动器
+                return str(directory).startswith('\\\\')
+        except Exception as e:
+            logger.debug(f"检测网络文件系统时出错: {e}")
+        return False
+
    def init(self):
        """
        启动监控
@@ -199,10 +353,12 @@ class Monitor(metaclass=Singleton):
        # 读取目录配置
        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)
@@ -210,9 +366,12 @@ class Monitor(metaclass=Singleton):
        messagehelper = MessageHelper()
        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)
@@ -224,31 +383,70 @@ class Monitor(metaclass=Singleton):
            # 启动监控
            if mon_dir.storage == "local":
                # 本地目录监控
+                logger.info(f"正在启动本地目录监控: {mon_path}")
+                logger.info("*** 重要提示：目录监控只处理新增和修改的文件，不会处理监控启动前已存在的文件 ***")
+
                try:
-                    if mon_dir.monitor_mode == "fast":
-                        observer = self.__choose_observer()
-                    else:
+                    # 统计文件数量并给出提示
+                    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}")
+
+                    if use_polling:
                        observer = PollingObserver()
+                        logger.info(f"使用兼容模式(轮询)监控 {mon_path}")
+                    else:
+                        observer = self.__choose_observer()
+                        if observer is None:
+                            logger.warn(f"快速模式不可用，自动切换到兼容模式监控 {mon_path}")
+                            observer = PollingObserver()
+                        else:
+                            logger.info(f"使用快速模式监控 {mon_path}")
+                            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']})")
+
                    self._observers.append(observer)
                    observer.schedule(FileMonitorHandler(mon_path=mon_path, callback=self),
                                      path=str(mon_path),
                                      recursive=True)
                    observer.daemon = True
                    observer.start()
-                    logger.info(f"已启动 {mon_path} 的目录监控服务, 监控模式：{mon_dir.monitor_mode}")
+
+                    mode_name = "兼容模式(轮询)" if use_polling else "快速模式"
+                    logger.info(f"✓ 本地目录监控已启动: {mon_path} [{mode_name}]")
+
                except Exception as e:
                    err_msg = str(e)
-                    if "inotify" in err_msg and "reached" in err_msg:
-                        logger.warn(
-                            f"目录监控服务启动出现异常：{err_msg}，请在宿主机上（不是docker容器内）执行以下命令并重启："
-                            + """
-                             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
-                             """)
+                    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(f"{mon_path} 启动目录监控失败：{err_msg}")
-                    messagehelper.put(f"{mon_path} 启动目录监控失败：{err_msg}", title="目录监控")
+                        logger.error("建议尝试使用兼容模式进行监控")
+
+                    messagehelper.put(f"启动本地目录监控失败: {mon_path}\n错误: {err_msg}", title="目录监控")
            else:
                # 远程目录监控 - 使用智能间隔
                # 先尝试加载已有快照获取文件数量
@@ -256,6 +454,10 @@ class Monitor(metaclass=Singleton):
                file_count = snapshot_data.get('file_count', 0) if snapshot_data else 0
                interval = self.adjust_monitor_interval(file_count)

+                logger.info(f"正在启动远程目录监控: {mon_path} [{mon_dir.storage}]")
+                logger.info("*** 重要提示：远程目录监控只处理新增和修改的文件，不会处理监控启动前已存在的文件 ***")
+                logger.info(f"预估文件数量: {file_count}, 监控间隔: {interval}分钟")
+
                self._scheduler.add_job(
                    self.polling_observer,
                    'interval',
@@ -267,33 +469,76 @@ class Monitor(metaclass=Singleton):
                    id=f"monitor_{mon_dir.storage}_{mon_dir.download_path}",
                    replace_existing=True
                )
-                logger.info(f"已启动 {mon_path} 的远程目录监控，存储：{mon_dir.storage}，间隔：{interval}分钟")
+                logger.info(f"✓ 远程目录监控已启动: {mon_path} [间隔: {interval}分钟]")

        # 启动定时服务
        if self._scheduler.get_jobs():
            self._scheduler.print_jobs()
            self._scheduler.start()
+            logger.info("定时监控服务已启动")

-    @staticmethod
-    def __choose_observer() -> Any:
+        # 输出监控总结
+        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 __choose_observer(self) -> Optional[Any]:
        """
-        选择最优的监控模式
+        选择最优的监控模式（带错误处理和自动回退）
        """
        system = platform.system()

+        observers_to_try = []
+
        try:
            if system == 'Linux':
-                from watchdog.observers.inotify import InotifyObserver
-                return InotifyObserver()
+                observers_to_try = [
+                    ('InotifyObserver',
+                     lambda: self.__try_import_observer('watchdog.observers.inotify', 'InotifyObserver')),
+                ]
            elif system == 'Darwin':
-                from watchdog.observers.fsevents import FSEventsObserver
-                return FSEventsObserver()
+                observers_to_try = [
+                    ('FSEventsObserver',
+                     lambda: self.__try_import_observer('watchdog.observers.fsevents', 'FSEventsObserver')),
+                ]
            elif system == 'Windows':
-                from watchdog.observers.read_directory_changes import WindowsApiObserver
-                return WindowsApiObserver()
-        except Exception as error:
-            logger.warn(f"导入模块错误：{error}，将使用 PollingObserver 监控目录")
-        return PollingObserver()
+                observers_to_try = [
+                    ('WindowsApiObserver',
+                     lambda: self.__try_import_observer('watchdog.observers.read_directory_changes',
+                                                        'WindowsApiObserver')),
+                ]
+
+            # 尝试每个观察者
+            for observer_name, observer_func in observers_to_try:
+                try:
+                    observer_class = observer_func()
+                    if observer_class:
+                        # 尝试创建实例以验证是否可用
+                        test_observer = observer_class()
+                        test_observer.stop()  # 立即停止测试实例
+                        logger.debug(f"成功初始化 {observer_name}")
+                        return observer_class()
+                except Exception as e:
+                    logger.debug(f"初始化 {observer_name} 失败: {e}")
+                    continue
+
+        except Exception as e:
+            logger.debug(f"选择观察者时出错: {e}")
+
+        logger.debug("所有快速监控模式都不可用，将使用兼容模式")
+        return None
+
+    @staticmethod
+    def __try_import_observer(module_name: str, class_name: str):
+        """
+        尝试导入观察者类
+        """
+        try:
+            module = __import__(module_name, fromlist=[class_name])
+            return getattr(module, class_name)
+        except (ImportError, AttributeError) as e:
+            logger.debug(f"导入 {module_name}.{class_name} 失败: {e}")
+            return None

    def polling_observer(self, storage: str, mon_path: Path):
        """
@@ -343,8 +588,11 @@ class Monitor(metaclass=Singleton):
                    if changes['added'] or changes['modified']:
                        logger.info(
                            f"{storage}:{mon_path} 发现 {len(changes['added'])} 个新增文件，{len(changes['modified'])} 个修改文件")
+                    else:
+                        logger.debug(f"{storage}:{mon_path} 无文件变化")
                else:
-                    logger.info(f"{storage}:{mon_path} 首次快照，暂不处理文件")
+                    logger.info(f"{storage}:{mon_path} 首次快照完成，共 {file_count} 个文件")
+                    logger.info("*** 首次快照仅建立基准，不会处理现有文件。后续监控将处理新增和修改的文件 ***")

                # 保存新快照
                self.save_snapshot(storage, new_snapshot, file_count)
@@ -375,7 +623,7 @@ class Monitor(metaclass=Singleton):
        """
        if not event.is_directory:
            # 文件发生变化
-            logger.debug(f"文件 {event_path} 发生了 {text}")
+            logger.debug(f"检测到文件变化: {event_path} [{text}]")
            # 整理文件
            self.__handle_file(storage="local", event_path=Path(event_path), file_size=file_size)

@@ -412,10 +660,12 @@ class Monitor(metaclass=Singleton):

            # TTL缓存控重
            if self._cache.get(str(event_path)):
+                logger.debug(f"文件 {event_path} 在缓存中，跳过处理")
                return
            self._cache[str(event_path)] = True

            try:
+                logger.info(f"开始整理文件: {event_path}")
                # 开始整理
                TransferChain().do_transfer(
                    fileitem=FileItem(
@@ -429,7 +679,7 @@ class Monitor(metaclass=Singleton):
                    )
                )
            except Exception as e:
-                logger.error("目录监控发生错误：%s - %s" % (str(e), traceback.format_exc()))
+                logger.error("目录监控整理文件发生错误：%s - %s" % (str(e), traceback.format_exc()))

    def stop(self):
        """
@@ -437,20 +687,22 @@ class Monitor(metaclass=Singleton):
        """
        self._event.set()
        if self._observers:
+            logger.info("正在停止本地目录监控服务...")
            for observer in self._observers:
                try:
-                    logger.info(f"正在停止目录监控服务：{observer}...")
                    observer.stop()
                    observer.join()
-                    logger.info(f"{observer} 目录监控已停止")
+                    logger.debug(f"已停止监控服务: {observer}")
                except Exception as e:
                    logger.error(f"停止目录监控服务出现了错误：{e}")
            self._observers = []
+            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