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

目录

• 一、为什么需要文件监控自动化？
• 二、核心技术：Watchdog库详解
• 1. 基础监控实现
• 2. 增强功能：文件智能分类
• 三、微信实时通知实现
• 1. 通过Server酱推送
• 2. 企业微信机器人通知
• 四、完整实战案例
• 1. 监控+分类+通知完整代码
• 2. 部署与优化建议
• 五、企业级增强方案
• 安全增强实现部分源码

在日常工作中，你是否经常遇到这样的困扰：

• 同事发来的文件混杂在下载文件夹里，需要手动整理
• 重要文件被修改却没能及时察觉
• 紧急合同上传后，需要反复刷新查看是否到位

​传统方式 vs 自动化方案对比​：

方式	响应速度	准确性	持续工作能力
人工检查	分钟级	依赖注意力	8小时/天
Python监控	​秒级​	​100%​​	24小时/天

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()

​关键参数说明​：

• recursive=True：监控子文件夹
• event.src_path：获取文件完整路径
• on_created：响应文件创建事件（还有on_modified, on_deleted等）

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"📁 新文件提醒nn" 
              f"文件名：{file.name}n" 
              f"类型：{file.suffix}n" 
              f"大小：{file.stat().st_size/1024:.2f}KBn" 
              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()

​部署方式​：

Windows系统：打包为exe后台运行

pyinstaller --onefile --windowed file_monitor.py

Linux系统：配置为systemd服务

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)  # 隔离可疑文件

以上就是一文详解如何使用Python实现文件监控自动化系统的详细内容，更多关于Python自动监控文件夹的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

• Python办公自动化之网络监控和压缩文件处理
• 利用Python实现自动化监控文件夹完成服务部署
• python基于watchdog库全自动化监控目录文件
• Python实现实时文件监控和变更通知
• 基于Python Watchdog库实现文件系统监控