python

关注公众号 jb51net

关闭
首页 > 脚本专栏 > python > Python自动监控文件夹

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

作者:张大狸

这篇文章主要介绍了如何使用Python实现文件监控自动化系统,解决日常工作中的文件管理痛点,系统基于Watchdog库实时监控文件夹,支持文件自动分类(按类型/大小/日期)和微信通知提醒,文章详细讲解了核心代码实现,需要的朋友可以参考下

一、为什么需要文件监控自动化?

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

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

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

二、核心技术:Watchdog库详解

1. 基础监控实现

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

关键参数说明​:

2. 增强功能:文件智能分类

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

分类规则扩展建议​:

三、微信实时通知实现

1. 通过Server酱推送

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)

2. 企业微信机器人通知

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)

通知内容优化建议​:

四、完整实战案例

1. 监控+分类+通知完整代码

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

2. 部署与优化建议

部署方式​:

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自动监控文件夹的资料请关注脚本之家其它相关文章!

您可能感兴趣的文章:
阅读全文