使用监视器检测文件创建

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()方法正在等待该线程终止。因此，我认为我没有在观察者上调用正确的方法来停止等待/循环，但看门狗文档似乎并没有清楚地指出可以使用哪些方法。有人知道我如何实现我的目标吗？

- Laurent

Answer 1

最后，在研究看门狗实现之后，在调用 stop 之前不需要调用 unschedule_all，这是自动完成的。删除包含此方法调用的行可以解决问题，应用程序运行完美。

- Laurent

Answer 2

此外，以下脚本用于使用PatternMatchingEventHandler在特定的路径观察文件名。

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

- Milovan Tomašević