如何用Python实时高效监控含子文件夹的目录并收集新增文件列表?
如何用Python实时高效监控含子文件夹的目录并收集新增文件列表?
嗨,我来给你捋捋这个问题的解决思路,刚好我之前也处理过类似的目录监控场景!结合你的需求——要实时收集新增的CSV文件路径列表,还要处理日期命名的子文件夹,我给你两个实用的方案,你可以根据自己的场景选:
方案一:用调度库定期扫描(简单易上手,适配你每分钟新增文件的节奏)
你提到过类似Cron的调度思路,那咱们就用schedule库来实现每分钟扫描一次,同时轻松搞定子文件夹的遍历。核心要注意去重,避免重复收集已经处理过的文件。
具体实现步骤:
- 先装依赖:
pip install schedule
- 完整代码示例:
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),能实时触发事件,效率拉满。
具体实现步骤:
- 先装依赖:
pip install watchdog
- 完整代码示例:
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-2秒再检查文件大小是否稳定,确认写入完成后再收集。
- 路径格式转换:如果你需要的是相对路径(比如
mydata/2026_01_01/xxx.csv),可以用os.path.relpath(file_path)把绝对路径转成相对路径。 - 去重逻辑:一定要用
processed_files集合来记录已收集的文件,不然会重复往列表里加同一个文件。
最后给你个选型建议:
如果你的文件是固定每分钟新增一个,那方案一的轮询扫描完全够用,实现简单还省资源;如果要追求极致实时(比如文件可能随时新增,不是固定节奏),那方案二的watchdog是更好的选择。你可以根据自己的实际需求挑~




