try:
    observer.start()
except OSError as e:
    logger.error(f"文件系統(tǒng)監(jiān)控啟動失敗: {str(e)}")
    if e.errno == errno.ENOSPC:
        logger.error("系統(tǒng)inotify限制 reached，請增加fs.inotify.max_user_watches")
    raise FileMonitorError(f"監(jiān)控啟動失敗: {str(e)}")
except Exception as e:
    logger.error(f"文件監(jiān)控異常: {str(e)}")
    raise FileMonitorError(f"監(jiān)控異常: {str(e)}")

try:
    send_notification(event)
except smtplib.SMTPException as e:
    logger.error(f"郵件發(fā)送失敗: {str(e)}")
    # 重試機制
    retry_send_notification(event, max_retries=3)
except requests.RequestException as e:
    logger.error(f"Webhook請求失敗: {str(e)}")
    handle_webhook_failure(event)
except Exception as e:
    logger.error(f"通知發(fā)送異常: {str(e)}")

try:
    config = load_config(config_file)
except json.JSONDecodeError as e:
    logger.error(f"配置文件JSON格式錯誤: {str(e)}")
    raise ConfigError(f"配置文件格式無效: {str(e)}")
except yaml.YAMLError as e:
    logger.error(f"配置文件YAML格式錯誤: {str(e)}")
    raise ConfigError(f"配置文件格式無效: {str(e)}")
except FileNotFoundError:
    logger.warning(f"配置文件不存在，使用默認配置: {config_file}")
    config = create_default_config()

try:
    os.access(path, os.R_OK)
except PermissionError as e:
    logger.error(f"無權限訪問目錄: {path}")
    send_alert(f"監(jiān)控目錄訪問被拒絕: {path}")
except Exception as e:
    logger.error(f"目錄訪問異常: {str(e)}")

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
文件監(jiān)控和變更通知器
功能：監(jiān)控文件系統(tǒng)變更并發(fā)送通知
作者：Cline
版本：1.0
"""

import argparse
import sys
import json
import yaml
import logging
import os
import time
import threading
import re
import fnmatch
from datetime import datetime
from typing import Dict, List, Set, Optional, Callable
from pathlib import Path
import hashlib
import smtplib
import requests
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

# 根據(jù)平臺導入文件監(jiān)控庫
try:
    from watchdog.observers import Observer
    from watchdog.events import FileSystemEventHandler
    WATCHDOG_AVAILABLE = True
except ImportError:
    WATCHDOG_AVAILABLE = False
    logger.warning("watchdog庫未安裝，將使用輪詢模式")

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('file_monitor.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

class FileMonitorError(Exception):
    """文件監(jiān)控器異常類"""
    pass

class ConfigError(Exception):
    """配置異常類"""
    pass

class FileEvent:
    """文件事件類"""
    def __init__(self, event_type: str, src_path: str, dest_path: str = None):
        self.event_type = event_type  # created, modified, deleted, moved
        self.src_path = src_path
        self.dest_path = dest_path
        self.timestamp = datetime.now()
        self.file_size = None
        self.file_hash = None
        
        # 獲取文件信息
        if os.path.exists(src_path):
            try:
                self.file_size = os.path.getsize(src_path)
                # 計算文件哈希值（僅對小文件）
                if self.file_size < 10 * 1024 * 1024:  # 10MB以下
                    self.file_hash = self._calculate_hash(src_path)
            except Exception:
                pass
                
    def _calculate_hash(self, file_path: str) -> str:
        """計算文件MD5哈希值"""
        hash_md5 = hashlib.md5()
        try:
            with open(file_path, "rb") as f:
                for chunk in iter(lambda: f.read(4096), b""):
                    hash_md5.update(chunk)
            return hash_md5.hexdigest()
        except Exception:
            return None
            
    def to_dict(self) -> Dict:
        """轉換為字典"""
        return {
            'event_type': self.event_type,
            'src_path': self.src_path,
            'dest_path': self.dest_path,
            'timestamp': self.timestamp.isoformat(),
            'file_size': self.file_size,
            'file_hash': self.file_hash
        }

class NotificationManager:
    """通知管理器"""
    def __init__(self, config: Dict):
        self.config = config.get('notifications', {})
        self.email_config = self.config.get('email', {})
        self.webhook_config = self.config.get('webhook', {})
        self.desktop_config = self.config.get('desktop', {})
        
    def send_email(self, subject: str, body: str, attachments: List[str] = None):
        """發(fā)送郵件通知"""
        if not self.email_config.get('enabled', False):
            return
            
        try:
            msg = MIMEMultipart()
            msg['From'] = self.email_config.get('sender')
            msg['To'] = ', '.join(self.email_config.get('recipients', []))
            msg['Subject'] = subject
            
            msg.attach(MIMEText(body, 'plain'))
            
            server = smtplib.SMTP(
                self.email_config.get('smtp_server'), 
                self.email_config.get('smtp_port', 587)
            )
            server.starttls()
            server.login(
                self.email_config.get('sender'), 
                self.email_config.get('password')
            )
            server.send_message(msg)
            server.quit()
            
            logger.info(f"郵件通知已發(fā)送: {subject}")
            
        except Exception as e:
            logger.error(f"郵件發(fā)送失敗: {str(e)}")
            
    def send_webhook(self, event: FileEvent):
        """發(fā)送webhook通知"""
        if not self.webhook_config.get('enabled', False):
            return
            
        try:
            payload = {
                'event': event.to_dict(),
                'monitor_name': self.config.get('name', 'Unknown')
            }
            
            response = requests.post(
                self.webhook_config.get('url'),
                json=payload,
                headers=self.webhook_config.get('headers', {}),
                timeout=self.webhook_config.get('timeout', 30)
            )
            response.raise_for_status()
            
            logger.info(f"Webhook通知已發(fā)送: {event.event_type}")
            
        except Exception as e:
            logger.error(f"Webhook發(fā)送失敗: {str(e)}")
            
    def send_desktop_notification(self, title: str, message: str):
        """發(fā)送桌面通知"""
        if not self.desktop_config.get('enabled', False):
            return
            
        try:
            # 根據(jù)平臺選擇通知庫
            if sys.platform == 'win32':
                # Windows通知
                from plyer import notification
                notification.notify(
                    title=title,
                    message=message,
                    app_name='File Monitor'
                )
            elif sys.platform == 'darwin':
                # macOS通知
                import pync
                pync.notify(message, title=title)
            else:
                # Linux通知
                import notify2
                notify2.init('File Monitor')
                notice = notify2.Notification(title, message)
                notice.show()
                
            logger.info(f"桌面通知已發(fā)送: {title}")
            
        except Exception as e:
            logger.error(f"桌面通知發(fā)送失敗: {str(e)}")

class FileEventHandler(FileSystemEventHandler):
    """文件事件處理器"""
    def __init__(self, monitor_config: Dict, notification_manager: NotificationManager):
        self.config = monitor_config
        self.notification_manager = notification_manager
        self.ignore_patterns = self.config.get('ignore_patterns', [])
        self.include_patterns = self.config.get('include_patterns', ['*'])
        self.min_file_size = self.config.get('min_file_size', 0)
        self.max_file_size = self.config.get('max_file_size', float('inf'))
        
    def on_any_event(self, event):
        """處理所有文件事件"""
        if event.is_directory:
            return
            
        # 檢查是否應該忽略此文件
        if self._should_ignore(event.src_path):
            return
            
        # 創(chuàng)建文件事件對象
        file_event = FileEvent(
            event_type=event.event_type,
            src_path=event.src_path,
            dest_path=getattr(event, 'dest_path', None)
        )
        
        # 檢查文件大小過濾
        if file_event.file_size is not None:
            if file_event.file_size < self.min_file_size or file_event.file_size > self.max_file_size:
                return
                
        # 發(fā)送通知
        self._send_notifications(file_event)
        
    def _should_ignore(self, file_path: str) -> bool:
        """檢查是否應該忽略文件"""
        filename = os.path.basename(file_path)
        
        # 檢查排除模式
        for pattern in self.ignore_patterns:
            if fnmatch.fnmatch(filename, pattern):
                return True
                
        # 檢查包含模式
        included = False
        for pattern in self.include_patterns:
            if fnmatch.fnmatch(filename, pattern):
                included = True
                break
                
        return not included
        
    def _send_notifications(self, event: FileEvent):
        """發(fā)送通知"""
        # 構造通知內(nèi)容
        subject = f"文件變更通知 - {event.event_type}"
        message = f"""
文件監(jiān)控器檢測到文件變更

事件類型: {event.event_type}
文件路徑: {event.src_path}
目標路徑: {event.dest_path or 'N/A'}
文件大小: {event.file_size or 'N/A'} bytes
時間戳: {event.timestamp.strftime('%Y-%m-%d %H:%M:%S')}
文件哈希: {event.file_hash or 'N/A'}
        """
        
        # 發(fā)送各種通知
        self.notification_manager.send_email(subject, message)
        self.notification_manager.send_webhook(event)
        self.notification_manager.send_desktop_notification(subject, message)

class PollingFileMonitor:
    """輪詢文件監(jiān)控器（當watchdog不可用時使用）"""
    def __init__(self, path: str, interval: int = 1):
        self.path = path
        self.interval = interval
        self.files = {}
        self.running = False
        self.callbacks = []
        
        # 初始化文件狀態(tài)
        self._scan_directory()
        
    def _scan_directory(self):
        """掃描目錄獲取文件狀態(tài)"""
        try:
            for root, dirs, files in os.walk(self.path):
                for file in files:
                    file_path = os.path.join(root, file)
                    try:
                        stat = os.stat(file_path)
                        self.files[file_path] = {
                            'mtime': stat.st_mtime,
                            'size': stat.st_size
                        }
                    except OSError:
                        continue
        except Exception as e:
            logger.error(f"掃描目錄失敗: {str(e)}")
            
    def add_callback(self, callback: Callable):
        """添加回調(diào)函數(shù)"""
        self.callbacks.append(callback)
        
    def start(self):
        """啟動監(jiān)控"""
        self.running = True
        thread = threading.Thread(target=self._monitor_loop)
        thread.daemon = True
        thread.start()
        
    def stop(self):
        """停止監(jiān)控"""
        self.running = False
        
    def _monitor_loop(self):
        """監(jiān)控循環(huán)"""
        while self.running:
            try:
                current_files = {}
                for root, dirs, files in os.walk(self.path):
                    for file in files:
                        file_path = os.path.join(root, file)
                        try:
                            stat = os.stat(file_path)
                            current_files[file_path] = {
                                'mtime': stat.st_mtime,
                                'size': stat.st_size
                            }
                        except OSError:
                            continue
                            
                # 檢查文件變更
                self._check_changes(current_files)
                self.files = current_files
                
                time.sleep(self.interval)
                
            except Exception as e:
                logger.error(f"輪詢監(jiān)控異常: {str(e)}")
                time.sleep(self.interval)
                
    def _check_changes(self, current_files: Dict):
        """檢查文件變更"""
        # 檢查新文件
        for file_path, info in current_files.items():
            if file_path not in self.files:
                event = FileEvent('created', file_path)
                for callback in self.callbacks:
                    try:
                        callback(event)
                    except Exception as e:
                        logger.error(f"回調(diào)執(zhí)行失敗: {str(e)}")
                        
        # 檢查刪除的文件
        for file_path in self.files:
            if file_path not in current_files:
                event = FileEvent('deleted', file_path)
                for callback in self.callbacks:
                    try:
                        callback(event)
                    except Exception as e:
                        logger.error(f"回調(diào)執(zhí)行失敗: {str(e)}")
                        
        # 檢查修改的文件
        for file_path, info in current_files.items():
            if file_path in self.files:
                old_info = self.files[file_path]
                if (info['mtime'] != old_info['mtime'] or 
                    info['size'] != old_info['size']):
                    event = FileEvent('modified', file_path)
                    for callback in self.callbacks:
                        try:
                            callback(event)
                        except Exception as e:
                            logger.error(f"回調(diào)執(zhí)行失敗: {str(e)}")

class FileMonitor:
    """文件監(jiān)控器主類"""
    def __init__(self, config_file: str = None):
        self.config_file = config_file
        self.config = {}
        self.monitors = {}
        self.notification_managers = {}
        self.running = False
        
        # 加載配置
        self.load_config()
        
    def load_config(self):
        """加載配置文件"""
        if not self.config_file or not os.path.exists(self.config_file):
            logger.info("未指定配置文件或文件不存在，使用默認配置")
            self.config = self._create_default_config()
            return
            
        try:
            with open(self.config_file, 'r', encoding='utf-8') as f:
                if self.config_file.endswith('.yaml') or self.config_file.endswith('.yml'):
                    self.config = yaml.safe_load(f)
                else:
                    self.config = json.load(f)
                    
            logger.info(f"成功加載配置文件: {self.config_file}")
            
        except Exception as e:
            logger.error(f"加載配置文件失敗: {str(e)}")
            raise ConfigError(f"配置加載失敗: {str(e)}")
            
    def _create_default_config(self) -> Dict:
        """創(chuàng)建默認配置"""
        return {
            "monitors": [
                {
                    "name": "downloads_monitor",
                    "path": "~/Downloads",
                    "recursive": True,
                    "ignore_patterns": ["*.tmp", "*.part"],
                    "include_patterns": ["*"],
                    "min_file_size": 0,
                    "max_file_size": 1073741824,  # 1GB
                    "enabled": True
                }
            ],
            "notifications": {
                "email": {
                    "enabled": False,
                    "smtp_server": "smtp.gmail.com",
                    "smtp_port": 587,
                    "sender": "your_email@gmail.com",
                    "password": "your_app_password",
                    "recipients": ["admin@example.com"]
                },
                "webhook": {
                    "enabled": False,
                    "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
                    "headers": {},
                    "timeout": 30
                },
                "desktop": {
                    "enabled": True
                }
            }
        }
        
    def start(self):
        """啟動所有監(jiān)控器"""
        if self.running:
            logger.warning("文件監(jiān)控器已在運行")
            return
            
        logger.info("啟動文件監(jiān)控器...")
        self.running = True
        
        # 為每個監(jiān)控任務創(chuàng)建監(jiān)控器
        for monitor_config in self.config.get('monitors', []):
            if not monitor_config.get('enabled', True):
                continue
                
            name = monitor_config.get('name', 'unnamed')
            path = os.path.expanduser(monitor_config.get('path', '.'))
            
            if not os.path.exists(path):
                logger.warning(f"監(jiān)控路徑不存在: {path}")
                continue
                
            try:
                # 創(chuàng)建通知管理器
                notification_manager = NotificationManager(self.config)
                self.notification_managers[name] = notification_manager
                
                if WATCHDOG_AVAILABLE:
                    # 使用watchdog監(jiān)控
                    event_handler = FileEventHandler(monitor_config, notification_manager)
                    observer = Observer()
                    observer.schedule(event_handler, path, recursive=monitor_config.get('recursive', True))
                    observer.start()
                    self.monitors[name] = observer
                    logger.info(f"已啟動監(jiān)控器: {name} -> {path}")
                else:
                    # 使用輪詢監(jiān)控
                    interval = monitor_config.get('poll_interval', 1)
                    poll_monitor = PollingFileMonitor(path, interval)
                    
                    # 創(chuàng)建事件處理器
                    def handle_event(event):
                        event_handler = FileEventHandler(monitor_config, notification_manager)
                        event_handler._send_notifications(event)
                        
                    poll_monitor.add_callback(handle_event)
                    poll_monitor.start()
                    self.monitors[name] = poll_monitor
                    logger.info(f"已啟動輪詢監(jiān)控器: {name} -> {path}")
                    
            except Exception as e:
                logger.error(f"啟動監(jiān)控器 {name} 失敗: {str(e)}")
                
        logger.info("文件監(jiān)控器啟動完成")
        
    def stop(self):
        """停止所有監(jiān)控器"""
        logger.info("停止文件監(jiān)控器...")
        self.running = False
        
        for name, monitor in self.monitors.items():
            try:
                if WATCHDOG_AVAILABLE and hasattr(monitor, 'stop'):
                    monitor.stop()
                    monitor.join()
                elif hasattr(monitor, 'stop'):
                    monitor.stop()
                logger.info(f"已停止監(jiān)控器: {name}")
            except Exception as e:
                logger.error(f"停止監(jiān)控器 {name} 失敗: {str(e)}")
                
        self.monitors.clear()
        logger.info("文件監(jiān)控器已停止")
        
    def add_monitor(self, monitor_config: Dict):
        """動態(tài)添加監(jiān)控任務"""
        name = monitor_config.get('name', 'unnamed')
        path = os.path.expanduser(monitor_config.get('path', '.'))
        
        if name in self.monitors:
            logger.warning(f"監(jiān)控器已存在: {name}")
            return
            
        if not os.path.exists(path):
            logger.warning(f"監(jiān)控路徑不存在: {path}")
            return
            
        try:
            notification_manager = NotificationManager(self.config)
            self.notification_managers[name] = notification_manager
            
            if WATCHDOG_AVAILABLE:
                event_handler = FileEventHandler(monitor_config, notification_manager)
                observer = Observer()
                observer.schedule(event_handler, path, recursive=monitor_config.get('recursive', True))
                observer.start()
                self.monitors[name] = observer
            else:
                interval = monitor_config.get('poll_interval', 1)
                poll_monitor = PollingFileMonitor(path, interval)
                
                def handle_event(event):
                    event_handler = FileEventHandler(monitor_config, notification_manager)
                    event_handler._send_notifications(event)
                    
                poll_monitor.add_callback(handle_event)
                poll_monitor.start()
                self.monitors[name] = poll_monitor
                
            logger.info(f"已添加監(jiān)控器: {name} -> {path}")
            
        except Exception as e:
            logger.error(f"添加監(jiān)控器 {name} 失敗: {str(e)}")
            
    def remove_monitor(self, name: str):
        """移除監(jiān)控任務"""
        if name not in self.monitors:
            logger.warning(f"監(jiān)控器不存在: {name}")
            return
            
        try:
            monitor = self.monitors[name]
            if WATCHDOG_AVAILABLE and hasattr(monitor, 'stop'):
                monitor.stop()
                monitor.join()
            elif hasattr(monitor, 'stop'):
                monitor.stop()
                
            del self.monitors[name]
            if name in self.notification_managers:
                del self.notification_managers[name]
                
            logger.info(f"已移除監(jiān)控器: {name}")
            
        except Exception as e:
            logger.error(f"移除監(jiān)控器 {name} 失敗: {str(e)}")

def create_sample_config():
    """創(chuàng)建示例配置文件"""
    sample_config = {
        "monitors": [
            {
                "name": "downloads_monitor",
                "path": "~/Downloads",
                "recursive": True,
                "ignore_patterns": ["*.tmp", "*.part", "~*"],
                "include_patterns": ["*.pdf", "*.doc", "*.docx", "*.jpg", "*.png"],
                "min_file_size": 1024,  # 1KB
                "max_file_size": 1073741824,  # 1GB
                "enabled": True
            },
            {
                "name": "config_monitor",
                "path": "/etc",
                "recursive": False,
                "ignore_patterns": [],
                "include_patterns": ["*.conf", "*.cfg"],
                "min_file_size": 0,
                "max_file_size": 10485760,  # 10MB
                "enabled": True
            }
        ],
        "notifications": {
            "email": {
                "enabled": True,
                "smtp_server": "smtp.gmail.com",
                "smtp_port": 587,
                "sender": "your_email@gmail.com",
                "password": "your_app_password",
                "recipients": ["admin@example.com", "security@example.com"]
            },
            "webhook": {
                "enabled": True,
                "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
                "headers": {
                    "Content-Type": "application/json"
                },
                "timeout": 30
            },
            "desktop": {
                "enabled": True
            }
        }
    }
    
    with open('file_monitor_sample_config.json', 'w', encoding='utf-8') as f:
        json.dump(sample_config, f, indent=2, ensure_ascii=False)
    logger.info("示例配置文件已創(chuàng)建: file_monitor_sample_config.json")

def main():
    parser = argparse.ArgumentParser(description='文件監(jiān)控和變更通知器')
    parser.add_argument('-c', '--config', help='配置文件路徑')
    parser.add_argument('--start', action='store_true', help='啟動監(jiān)控器')
    parser.add_argument('--sample-config', action='store_true', help='創(chuàng)建示例配置文件')
    
    args = parser.parse_args()
    
    if args.sample_config:
        create_sample_config()
        return
        
    monitor = FileMonitor(args.config)
    
    if args.start:
        try:
            monitor.start()
            # 保持程序運行
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            logger.info("收到中斷信號，正在停止監(jiān)控器...")
        finally:
            monitor.stop()
    else:
        parser.print_help()

if __name__ == '__main__':
    main()

# 基礎依賴
pip install watchdog requests pyyaml plyer

# macOS通知支持
pip install pync

# Linux通知支持
pip install notify2

python file_monitor.py --sample-config

python file_monitor.py --config file_monitor_config.json --start

{
  "monitors": [
    {
      "name": "downloads_monitor",
      "path": "~/Downloads",
      "recursive": true,
      "ignore_patterns": ["*.tmp", "*.part", "~*"],
      "include_patterns": ["*.pdf", "*.doc", "*.docx", "*.jpg", "*.png"],
      "min_file_size": 1024,
      "max_file_size": 1073741824,
      "enabled": true
    },
    {
      "name": "config_monitor",
      "path": "/etc",
      "recursive": false,
      "ignore_patterns": [],
      "include_patterns": ["*.conf", "*.cfg"],
      "min_file_size": 0,
      "max_file_size": 10485760,
      "enabled": true
    }
  ],
  "notifications": {
    "email": {
      "enabled": true,
      "smtp_server": "smtp.gmail.com",
      "smtp_port": 587,
      "sender": "your_email@gmail.com",
      "password": "your_app_password",
      "recipients": ["admin@example.com"]
    },
    "webhook": {
      "enabled": true,
      "url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK",
      "headers": {
        "Content-Type": "application/json"
      },
      "timeout": 30
    },
    "desktop": {
      "enabled": true
    }
  }
}

monitors:
  - name: downloads_monitor
    path: ~/Downloads
    recursive: true
    ignore_patterns:
      - "*.tmp"
      - "*.part"
      - "~*"
    include_patterns:
      - "*.pdf"
      - "*.doc"
      - "*.docx"
      - "*.jpg"
      - "*.png"
    min_file_size: 1024
    max_file_size: 1073741824
    enabled: true

  - name: config_monitor
    path: /etc
    recursive: false
    ignore_patterns: []
    include_patterns:
      - "*.conf"
      - "*.cfg"
    min_file_size: 0
    max_file_size: 10485760
    enabled: true

notifications:
  email:
    enabled: true
    smtp_server: smtp.gmail.com
    smtp_port: 587
    sender: your_email@gmail.com
    password: your_app_password
    recipients:
      - admin@example.com

  webhook:
    enabled: true
    url: https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK
    headers:
      Content-Type: application/json
    timeout: 30

  desktop:
    enabled: true