如何让Python脚本等待文本文件内容发生变更后再执行后续操作？

以下提供两种常用实现方式，可根据你的使用场景选择：

方案1：无第三方依赖的跨平台轮询实现 #

通过定时检查文件的修改时间和大小变化判断是否有新内容追加，无需安装额外依赖，适配所有操作系统。

import os
import time
import subprocess

# 自定义配置
TARGET_FILE = "/path/to/your/target.txt"  # 替换为你的目标文件路径
EXEC_COMMAND = ["your", "command", "args"]  # 替换为你要执行的命令

def watch_and_exec():
    # 初始定位到文件末尾，忽略历史内容
    with open(TARGET_FILE, 'r', encoding='utf-8') as f:
        f.seek(0, os.SEEK_END)
        last_read_pos = f.tell()
    last_modify_time = os.path.getmtime(TARGET_FILE)

    while True:
        # 调整休眠时间平衡延迟和CPU占用，单位为秒
        time.sleep(1)
        current_mtime = os.path.getmtime(TARGET_FILE)
        current_size = os.path.getsize(TARGET_FILE)

        # 仅当文件修改时间更新且大小变大，判定为有新内容追加
        if current_mtime > last_modify_time and current_size > last_read_pos:
            with open(TARGET_FILE, 'r', encoding='utf-8') as f:
                f.seek(last_read_pos)
                new_content = f.read()
                last_read_pos = f.tell()
                last_modify_time = current_mtime
            
            # 执行指定命令，可按需将new_content传入命令
            subprocess.run(EXEC_COMMAND, check=True)

if __name__ == "__main__":
    watch_and_exec()

方案2：低延迟系统级事件监听（需第三方库watchdog） #

基于操作系统原生的文件事件通知机制实现，闲置状态无CPU占用，内容追加触发延迟极低。

1. 先安装依赖库：
   pip install watchdog

2. 示例代码：

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

# 自定义配置
TARGET_FILE = "/path/to/your/target.txt"
EXEC_COMMAND = ["your", "command", "args"]
last_read_pos = 0

class AppendMonitor(FileSystemEventHandler):
    def on_modified(self, event):
        global last_read_pos
        # 过滤非目标文件的修改事件
        if event.is_directory or os.path.abspath(event.src_path) != os.path.abspath(TARGET_FILE):
            return
        current_size = os.path.getsize(TARGET_FILE)
        if current_size > last_read_pos:
            with open(TARGET_FILE, 'r', encoding='utf-8') as f:
                f.seek(last_read_pos)
                new_content = f.read()
                last_read_pos = f.tell()
            subprocess.run(EXEC_COMMAND, check=True)

if __name__ == "__main__":
    # 初始定位到文件末尾
    with open(TARGET_FILE, 'r', encoding='utf-8') as f:
        f.seek(0, os.SEEK_END)
        last_read_pos = f.tell()
    
    observer = Observer()
    observer.schedule(AppendMonitor(), path=os.path.dirname(TARGET_FILE), recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

注意事项 #

• 若业务场景存在文件被截断重写而非仅追加的情况，可自行添加内容哈希校验逻辑避免误判
• 如需将新追加的内容传入执行命令，可直接将读取到的new_content作为参数传入subprocess.run调用
• Windows平台如遇编码报错，调整open函数的encoding参数适配即可

内容的提问来源于stack exchange，提问作者George Wilhelm Hegel