You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何用Python实时高效监控含子文件夹的目录并收集新增文件列表?

如何用Python实时高效监控含子文件夹的目录并收集新增文件列表?

嗨,我来给你捋捋这个问题的解决思路,刚好我之前也处理过类似的目录监控场景!结合你的需求——要实时收集新增的CSV文件路径列表,还要处理日期命名的子文件夹,我给你两个实用的方案,你可以根据自己的场景选:


方案一:用调度库定期扫描(简单易上手,适配你每分钟新增文件的节奏)

你提到过类似Cron的调度思路,那咱们就用schedule库来实现每分钟扫描一次,同时轻松搞定子文件夹的遍历。核心要注意去重,避免重复收集已经处理过的文件。

具体实现步骤:

  1. 先装依赖:
pip install schedule
  1. 完整代码示例:
import os
import schedule
import time

# 全局变量:存储已收集的文件(避免重复)、实时更新的目标文件列表
processed_files = set()
target_file_list = []
# 你的根目录路径
ROOT_DIR = "mydata"

def scan_new_files():
    global target_file_list, processed_files
    # 遍历根目录下所有子文件夹,os.walk会自动递归所有层级
    for root, _, files in os.walk(ROOT_DIR):
        for file in files:
            # 只筛选CSV文件
            if file.endswith(".csv"):
                file_path = os.path.join(root, file)
                # 只把未收集过的文件加入列表
                if file_path not in processed_files:
                    processed_files.add(file_path)
                    target_file_list.append(file_path)
                    print(f"新增文件已加入列表:{file_path}")

# 配置调度:每分钟执行一次扫描
schedule.every(1).minutes.do(scan_new_files)

# 初始化扫描:先把目录里已有的CSV文件全部加载进来
scan_new_files()

# 保持程序持续运行
print("开始监控目录,每分钟自动扫描新增文件...")
while True:
    schedule.run_pending()
    time.sleep(1)

优化小技巧:

因为你的子文件夹是按日期命名的,每天新增的文件只会出现在当天的子文件夹里,咱们可以优化扫描范围——每次只扫当天的日期文件夹,不用遍历所有历史文件夹,能大幅提升扫描速度:

from datetime import datetime

def scan_new_files():
    global target_file_list, processed_files
    # 获取今天的日期字符串,匹配你的子文件夹命名格式
    today = datetime.now().strftime("%Y_%m_%d")
    today_dir = os.path.join(ROOT_DIR, today)
    # 先判断当天文件夹是否存在
    if os.path.exists(today_dir):
        for root, _, files in os.walk(today_dir):
            for file in files:
                if file.endswith(".csv"):
                    file_path = os.path.join(root, file)
                    if file_path not in processed_files:
                        processed_files.add(file_path)
                        target_file_list.append(file_path)
                        print(f"新增文件已加入列表:{file_path}")

方案二:用watchdog实时监控(极致无延迟,文件一创建就捕捉)

如果你不想等那1分钟的轮询延迟,想要文件刚被其他程序创建完成就立刻加入列表,那watchdog库更适合你。它基于系统原生的文件监控API(比如Linux的inotify、Windows的ReadDirectoryChangesW),能实时触发事件,效率拉满。

具体实现步骤:

  1. 先装依赖:
pip install watchdog
  1. 完整代码示例:
import os
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 全局变量:存储已收集的文件、实时更新的目标文件列表
processed_files = set()
target_file_list = []
ROOT_DIR = "mydata"

class NewFileHandler(FileSystemEventHandler):
    def on_created(self, event):
        # 只处理文件创建事件,忽略文件夹创建
        if not event.is_directory:
            file_path = event.src_path
            # 只筛选CSV文件
            if file_path.endswith(".csv"):
                # 关键:其他程序可能还在写入文件,直接处理会报错,加个小判断
                try:
                    # 尝试打开文件,确认写入已完成
                    with open(file_path, "r") as f:
                        pass
                    # 去重后加入列表
                    if file_path not in processed_files:
                        processed_files.add(file_path)
                        target_file_list.append(file_path)
                        print(f"新增文件已加入列表:{file_path}")
                except:
                    # 打开失败说明文件还在写入,跳过本次,后续不会重复触发(因为已加入processed_files)
                    pass

if __name__ == "__main__":
    # 初始化事件处理器和监控器
    event_handler = NewFileHandler()
    observer = Observer()
    # recursive=True 表示监控所有子目录
    observer.schedule(event_handler, path=ROOT_DIR, recursive=True)
    observer.start()
    print("开始实时监控目录,文件一创建就自动收集...")

    # 先加载已有的CSV文件到列表
    for root, _, files in os.walk(ROOT_DIR):
        for file in files:
            if file.endswith(".csv"):
                file_path = os.path.join(root, file)
                processed_files.add(file_path)
                target_file_list.append(file_path)

    # 保持程序运行,按Ctrl+C停止
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

几个必注意的细节:

  1. 文件写入完成判断:不管用哪种方案,都要处理「其他程序还在写文件就被监控到」的情况——比如上面代码里的尝试打开文件,或者你也可以加个延迟,等1-2秒再检查文件大小是否稳定,确认写入完成后再收集。
  2. 路径格式转换:如果你需要的是相对路径(比如mydata/2026_01_01/xxx.csv),可以用os.path.relpath(file_path)把绝对路径转成相对路径。
  3. 去重逻辑:一定要用processed_files集合来记录已收集的文件,不然会重复往列表里加同一个文件。

最后给你个选型建议:

如果你的文件是固定每分钟新增一个,那方案一的轮询扫描完全够用,实现简单还省资源;如果要追求极致实时(比如文件可能随时新增,不是固定节奏),那方案二的watchdog是更好的选择。你可以根据自己的实际需求挑~

火山引擎 最新活动