当 Python 看门狗的目录发生任何变化时，如何运行一个函数？

Question

我正在尝试使用 watchdog 到运行同步脚本，只要目录中有任何更改（一个特定文件除外）。我只是从自述文件（粘贴在下面）中复制了代码，它按照它说的做；记录哪个文件已更改。

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    event_handler = LoggingEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

我现在想要运行一个功能（将整个文件夹同步到远程机器）每当有任何变化时。所以我只是用我自己的函数替换了 event_handler 。但这给了我以下错误：

Traceback (most recent call last):
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 810, in __bootstrap_inner
    self.run()
  File "/Library/Python/2.7/site-packages/watchdog/observers/api.py", line 199, in run
    self.dispatch_events(self.event_queue, self.timeout)
  File "/Library/Python/2.7/site-packages/watchdog/observers/api.py", line 368, in dispatch_events
    handler.dispatch(event)
AttributeError: 'function' object has no attribute 'dispatch'

有人知道我做错了什么吗？欢迎所有 tips！

ps。我还想排除文件夹中的一个文件被监视。我应该怎么做？

Answer 1

你需要 subclass 并在 dispatch 中做任何你想做的事情:

import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

class Event(LoggingEventHandler):
    def dispatch(self, event):
        print("Foobar")

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO,
                        format='%(asctime)s - %(message)s',
                        datefmt='%Y-%m-%d %H:%M:%S')
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    event_handler = Event()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

如果您运行代码，您将在检测到更改时看到 Foobar 输出，要忽略您可能需要使用 [events.PatternMatchingEventHandler][1] 的文件。每个都有各种方法

要在某事被修改的情况下做某事，我们可以覆盖 on_modified:

class Event(LoggingEventHandler):
    def on_modified(self, event):
        print("Doh")

和运行将上面的class代码与event_handler = Event()结合使用并更改文件将输出如下内容：

Doh
Doh
Doh
Doh
Doh
Doh
Doh
2015-10-03 15:33:55 - Created file: ./test.txt___jb_bak___
2015-10-03 15:33:55 - Moved file: from ./test.txt to ./test.txt___jb_old___
2015-10-03 15:33:55 - Moved file: from ./test.txt___jb_bak___ to ./test.txt
2015-10-03 15:33:55 - Deleted file: ./test.txt___jb_old___
Doh

[1]: http://pythonhosted.org/watchdog/api.html#watchdog.events.PatternMatchingEventHandler EventHandler class 你可以覆盖，这完全取决于你想做什么。 LoggingEventHandler class 自身是 watchdog.events.FileSystemEventHandler:

的子class

class watchdog.events.FileSystemEventHandler 基地：对象

Base file system event handler that you can override methods from.

调度（事件） 将事件分派给适当的方法。

Parameters: event (FileSystemEvent) – The event object representing the file system event.

on_any_event（事件） 捕获所有事件处理程序。

Parameters: event (FileSystemEvent) – The event object representing the file system event.

on_created（事件） 创建文件或目录时调用。

Parameters: event (DirCreatedEvent or FileCreatedEvent) – Event representing file/directory creation.

on_deleted（事件） 删除文件或目录时调用。

Parameters: event (DirDeletedEvent or FileDeletedEvent) – Event representing file/directory deletion.

on_modified（事件） 修改文件或目录时调用。

Parameters: event (DirModifiedEvent or FileModifiedEvent) – Event representing file/directory modification.

on_moved（事件） 移动或重命名文件或目录时调用。

Parameters: event (DirMovedEvent or FileMovedEvent) – Event representing file/directory movement.

Answer 2

python 中有很多方法可以跟踪目录中所做的更改。一种方法是使用 watchdog 模块。

需要模块
- 要在终端中安装看门狗运行此命令。

pip install watchdog

如果要在创建或修改 file/directory 时进行更改（当前位于根目录 - "."），可以使用以下代码：

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


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

    def run(self):
        event_handler = Handler()
        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(FileSystemEventHandler):
    @staticmethod
    def on_any_event(event):
        # if event.is_directory:
        #     return None
        print(
            "[{}] noticed: [{}] on: [{}] ".format(
                time.asctime(), event.event_type, event.src_path
            )
        )


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

event.src_path 将是完整的文件路径
event.event_type 将创建、移动等

If you want to ignore directory changes just remove the comment.

输出：

[Tue Feb  9 00:16:02 2021] noticed: [created] on: [/Users/mt/Documents/Whosebug/test.txt] 
[Tue Feb  9 00:16:02 2021] noticed: [modified] on: [/Users/mt/Documents/Whosebug] 
[Tue Feb  9 00:16:19 2021] noticed: [created] on: [/Users/mt/Documents/Whosebug/download.jpg] 
[Tue Feb  9 00:16:19 2021] noticed: [modified] on: [/Users/mt/Documents/Whosebug] 
[Tue Feb  9 00:16:30 2021] noticed: [created] on: [/Users/mt/Documents/Whosebug/new_folder] 
[Tue Feb  9 00:16:30 2021] noticed: [modified] on: [/Users/mt/Documents/Whosebug] 
[Tue Feb  9 00:16:46 2021] noticed: [deleted] on: [/Users/mt/Documents/Whosebug/new_folder] 
[Tue Feb  9 00:16:46 2021] noticed: [modified] on: [/Users/mt/Documents/Whosebug] 
[Tue Feb  9 00:16:52 2021] noticed: [deleted] on: [/Users/mt/Documents/Whosebug/download.jpg] 
[Tue Feb  9 00:16:52 2021] noticed: [modified] on: [/Users/mt/Documents/Whosebug] 
[Tue Feb  9 00:17:00 2021] noticed: [deleted] on: [/Users/mt/Documents/Whosebug/test.txt] 
[Tue Feb  9 00:17:00 2021] noticed: [modified] on: [/Users/mt/Documents/Whosebug]

Observer 是 class，它监视任何文件系统更改，然后将事件分派给事件处理程序。它监视文件系统并查找任何更改。

EventHandler 是一个对象，当文件系统发生某些事情时将被通知。通常，会编写一个脚本来监视创建或修改的任何类型的新文件，如 csv、txt、xml、jpg 等

例如，在下面的代码中，PatternMatchingEventHandler 继承自 FileSystemEventHandler class 并用于执行此操作。 class 的一些有用方法是：

on_any_event: 任何事件都会被执行。
on_created: 创建文件或目录时执行。
on_modified: 当文件被修改或目录重命名时执行。
on_deleted: 删除文件或目录时执行。
on_moved: 移动文件或目录时执行。

以下脚本仅用于使用 PatternMAtchingEventHandler 观察 .csv 文件。如果你想观察不止一种类型的文件，你可以进一步扩展模式列表。

import watchdog.events
import watchdog.observers
import time
import sys


class Handler(watchdog.events.PatternMatchingEventHandler):
    def __init__(self):
        # Set the patterns for PatternMatchingEventHandler
        watchdog.events.PatternMatchingEventHandler.__init__(
            self,
            patterns=["*.csv"],
            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 = sys.argv[1] if len(sys.argv) > 1 else "."
    event_handler = Handler()
    observer = watchdog.observers.Observer()
    observer.schedule(event_handler, path=path, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

在 PatternMAtchingEventHandler 的帮助下，我们可以利用仅处理与扩展名为 .csv 的文件相关的事件。

输出：

[Tue Feb  9 00:18:51 2021] noticed: [created] on: [/Users/mt/Documents/Whosebug/test.csv] 
[Tue Feb  9 00:18:59 2021] noticed: [modified] on: [/Users/mt/Documents/Whosebug/test.csv] 
[Tue Feb  9 00:19:12 2021] noticed: [deleted] on: [/Users/mt/Documents/Whosebug/test.csv]

当 Python 看门狗的目录发生任何变化时，如何运行一个函数？

How to run an function when anything changes in a dir with Python Watchdog?

python

filesystems

filesystemwatcher

watchdog

python-watchdog

当 Python 看门狗的目录发生任何变化时，如何 运行 一个函数？

How to run an function when anything changes in a dir with Python Watchdog?

python

filesystems

filesystemwatcher

watchdog

python-watchdog

当 Python 看门狗的目录发生任何变化时，如何运行一个函数？