Обнаружение создания файла с помощью сторожевого таймера

Я пытаюсь определить, когда файл с заданным именем создается в каталоге. Я делаю это благодаря сторожевому псу. Создание правильно обнаружено, но я не знаю, как правильно завершить приложение после обнаружения.

Мой фрагмент кода следующий:

#!/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() ожидает завершения этого потока. Таким образом, я предполагаю, что я не вызываю правильный метод для наблюдателя, чтобы остановить ожидание/зацикливание, но документация по сторожевому таймеру кажется действительно неясной, чтобы указать, какие методы можно использовать.

У кого-нибудь есть идея, как я могу достичь своей цели?


python multithreading exit-code watchdog termination
person Laurent    schedule 09.08.2012    source источник

Ответы (2)


arrow_upward
17
arrow_downward

Наконец, взглянув на реализацию сторожевого таймера, нет необходимости вызывать unschedule_all перед stop, это делается автоматически. Удаление строки, содержащей вызов этого метода, устраняет проблему, и приложение работает нормально.

person Laurent    schedule 09.08.2012

arrow_upward
1
arrow_downward

Кроме того, приведенный ниже сценарий используется для наблюдения за filename в определенном path с помощью 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
person Milovan Tomašević    schedule 09.02.2021