一文详解如何使用Python实现文件监控自动化系统

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class MyHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(f'新文件出现: {event.src_path}')
        # 这里添加文件处理逻辑

if __name__ == "__main__":
    path = "C:/监控文件夹"  # 替换为你的监控路径，比如微信文件存储路径，各类下载文件夹等
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

def auto_classify(file_path):
    """根据扩展名自动分类文件"""
    from pathlib import Path
    import shutil
    
    ext_map = {
        '.pdf': '文档',
        '.jpg': '图片',
        '.xlsx': '表格'
    }
    
    file = Path(file_path)
    if file.is_file():
        ext = file.suffix.lower()
        target_dir = Path(file.parent) / ext_map.get(ext, '其他')
        target_dir.mkdir(exist_ok=True)
        shutil.move(str(file), target_dir)
        return target_dir / file.name

import requests

def wechat_notify(title, content):
    """发送微信通知"""
    SCKEY = "你的Server酱SCKEY"  # 在https://sct.ftqq.com申请
    url = f"https://sc.ftqq.com/{SCKEY}.send"
    data = {
        "text": title,
        "desp": content
    }
    requests.post(url, data=data)

def qywx_notify(content):
    """企业微信机器人通知"""
    webhook = "你的企业微信群机器人Webhook地址"
    headers = {'Content-Type': 'application/json'}
    data = {
        "msgtype": "text",
        "text": {
            "content": content,
            "mentioned_mobile_list": ["13800138000"]  # 需要@的人
        }
    }
    requests.post(webhook, headers=headers, json=data)

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from pathlib import Path
import shutil
import requests

class SmartFileHandler(FileSystemEventHandler):
    def __init__(self):
        self.ext_map = {
            '.pdf': '文档',
            '.docx': '文档',
            '.xlsx': '表格',
            '.jpg': '图片'
        }
    
    def on_created(self, event):
        file_path = event.src_path
        if Path(file_path).is_file():
            target_path = self.classify_file(file_path)
            self.send_notification(file_path, target_path)
    
    def classify_file(self, file_path):
        """文件自动分类"""
        file = Path(file_path)
        ext = file.suffix.lower()
        target_dir = Path(file.parent) / self.ext_map.get(ext, '其他')
        target_dir.mkdir(exist_ok=True)
        shutil.move(str(file), target_dir)
        return target_dir / file.name
    
    def send_notification(self, origin_path, target_path):
        """发送微信通知"""
        file = Path(origin_path)
        msg = f"📁 新文件提醒\n\n" \
              f"文件名：{file.name}\n" \
              f"类型：{file.suffix}\n" \
              f"大小：{file.stat().st_size/1024:.2f}KB\n" \
              f"存放位置：{target_path}"
        wechat_notify("文件监控提醒", msg)

def main():
    path = "C:/监控文件夹"  # 设置监控路径
    event_handler = SmartFileHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    
    print(f"开始监控文件夹: {path}")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    main()

pyinstaller --onefile --windowed file_monitor.py

sudo nano /etc/systemd/system/filemon.service

if not file_path.endswith(('.pdf', '.docx')):
    return

last_trigger = 0
def on_created(self, event):
    if time.time() - last_trigger < 5:  # 5秒内不重复处理
        return
    last_trigger = time.time()

def check_virus(file_path):
    """简单的文件安全检查"""
    blacklist = ['.exe', '.bat', '.vbs']
    if Path(file_path).suffix.lower() in blacklist:
        wechat_notify("安全警报", f"发现可疑文件: {file_path}")
        quarantine_file(file_path)  # 隔离可疑文件