使用看门狗检测文件创建

Question

我正在尝试检测何时在目录中创建了具有给定名称的文件。感谢watchdog，我正在这样做。正确检测到创建，但是一旦检测完成，我不知道如何正确终止应用程序。

我的一段代码如下：

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import logging
import sys
import time

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

logging.basicConfig(level=logging.ERROR)

class MyEventHandler(FileSystemEventHandler):
    def __init__(self, observer, filename):
        self.observer = observer
        self.filename = filename

    def on_created(self, event):
        print "e=", event
        if not event.is_directory and event.src_path.endswith(self.filename):
            print "file created"
            self.observer.unschedule_all()
            self.observer.stop()

def main(argv=None):
    path = argv[1]
    filename = argv[2]
    observer = Observer()
    event_handler = MyEventHandler(observer, filename)
    observer.schedule(event_handler, path, recursive=False)
    observer.start()
    observer.join()
    return 0

if __name__ == "__main__":
    sys.exit(main(sys.argv))

我是 python 新手，我不知道出了什么问题。检测似乎安排在专用线程中，并且 join() 方法正在等待该线程终止。因此，我想我没有在观察者上调用正确的方法来停止等待/循环，但是看门狗文档似乎真的不清楚指出可以使用哪些方法。

有人知道如何实现我的目标吗？

Answer 1

最后，在看了看门狗实现之后，不必在stop之前调用unschedule_all，这是自动完成的。删除包含此方法调用的行可以解决问题，并且应用程序运行正常。

Answer 2

此外，以下脚本用于使用PatternMatchingEventHandler 在特定path 观察filename。

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from watchdog.events import PatternMatchingEventHandler
import sys


class Watcher:
    def __init__(self, path, filename):
        self.observer = Observer()
        self.path = path
        self.filename = filename

    def run(self):
        event_handler = Handler(self.filename)
        self.observer.schedule(event_handler, self.path, recursive=True)
        self.observer.start()
        try:
            while True:
                time.sleep(1)
        except:
            self.observer.stop()
            print("Error")

        self.observer.join()


class Handler(PatternMatchingEventHandler):
    def __init__(self, filename):
        super(Handler, self).__init__(
            patterns=[filename],
            ignore_patterns=["*.tmp"],
            ignore_directories=True,
            case_sensitive=False,
        )

    def on_any_event(self, event):
        print(
            "[{}] noticed: [{}] on: [{}] ".format(
                time.asctime(), event.event_type, event.src_path
            )
        )


if __name__ == "__main__":
    path = "."
    filename = "test.csv"

    w = Watcher(path, filename)
    w.run()

输出：

[Tue Feb  9 01:55:38 2021] noticed: [created] on: [/Users/mt/Documents/stackoverflow/test.csv] 
[Tue Feb  9 01:55:44 2021] noticed: [modified] on: [/Users/mt/Documents/stackoverflow/test.csv] 
[Tue Feb  9 01:56:01 2021] noticed: [deleted] on: [/Users/mt/Documents/stackoverflow/test.csv] 

也可以在不安装其他库的情况下确定新文件的创建。

import os
import time


def watch_file(filename, time_limit=3600, check_interval=60):
    """Return true if filename exists, if not keep checking once every check_interval seconds for time_limit seconds.
    time_limit defaults to 1 hour
    check_interval defaults to 1 minute
    """
    now = time.time()
    last_time = now + time_limit
    
    while time.time() <= last_time:
        if os.path.exists(filename):
            return True
        else:
            # Wait for check interval seconds, then check again.
            time.sleep(check_interval)
    return False


if __name__ == "__main__":
    filename = "test.csv"
    time_limit = 60
    check_interval = 1

    if watch_file(filename, time_limit, check_interval):
        print(f"File created: {os.path.abspath(filename)}")
    else:
        print(
            f"File {filename} not found after waiting: {time_limit} seconds!"
        )

输出：

File created: /Users/mt/Documents/stackoverflow/test.csv