You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Windows下Watchdog输出路径可手动打开但os.startfile无法访问

解决Watchdog触发文件创建事件后os.startfile找不到文件的问题

问题核心原因

Watchdog的created事件会在文件**刚被初始化(空文件状态)**时立刻触发,此时文件可能还在被其他程序写入数据,系统尚未完成文件的完整创建流程,导致os.startfile调用时无法识别到完整的文件实体。另外,虽然路径打印显示正常,但通过pathlib标准化路径格式能进一步避免潜在的路径兼容问题。

解决方案

以下是修改后的代码,主要添加了文件完整性等待逻辑,并优化了路径处理与事件响应:

import time
import os 
from pathlib import Path

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class Watcher:
    INCOMING_DIR = "C:/Users/username/documents/test/Incoming/"
    def __init__(self):
        self.observer = Observer()

    def run(self):
        event_handler = Handler()
        self.observer.schedule(event_handler, self.INCOMING_DIR, recursive=True)
        self.observer.start()
        try:
            while True:
                time.sleep(5)
        except KeyboardInterrupt:
            self.observer.stop()
        self.observer.join()

class Handler(FileSystemEventHandler):
    # 提取路径为类常量,避免重复定义
    PROCESSED_DIR = Path("C:/Users/username/documents/test/Processed/")
    INCOMING_DIR = Path("C:/Users/username/documents/test/Incoming/")

    def on_created(self, event):
        # 只处理文件事件,跳过目录
        if event.is_directory:
            return
        
        file_path = Path(event.src_path)
        print(f"Received created event - {file_path}.")
        print(file_path.name)

        # 等待文件完全创建完成,最多等待10秒
        max_wait = 10
        wait_count = 0
        while wait_count < max_wait:
            if file_path.exists() and file_path.is_file():
                try:
                    # 尝试打开文件,确保可访问
                    with open(file_path, 'r'):
                        break
                except:
                    pass
            time.sleep(1)
            wait_count += 1
        else:
            print(f"等待文件超时:{file_path}")
            return
        
        # 文件就绪后执行打印操作
        try:
            os.startfile(str(file_path), "print")
            print(f"已触发打印:{file_path}")
        except Exception as e:
            print(f"打印失败:{str(e)}")

if __name__ == '__main__':
    w = Watcher()
    w.run()

关键修改点说明

  • 改用on_created方法:替代on_any_event,精准响应文件创建事件,减少不必要的事件判断。
  • 文件完整性等待逻辑:循环检查文件是否存在且可读取,最多等待10秒(可根据需求调整),确保文件完全写入后再执行操作。
  • Path对象标准化路径:用pathlib.Path处理路径,自动适配Windows的路径格式,避免手动拼接路径的错误。
  • 异常捕获:添加try-except包裹os.startfile,捕获并打印可能的执行错误,便于排查问题。

验证要点

测试时可以尝试复制大文件到监控目录,此时created事件会提前触发,但等待逻辑会等到文件完全复制完成后才执行打印,避免出现找不到文件的错误。

内容的提问来源于stack exchange,提问作者Beliar

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.19 13:07:26