使用看门狗将新创建的文件名放入变量中

Question

我想使用看门狗来查找在一个文件夹中创建的新文件。然后文件名将用于不同的功能。我在这里使用此代码：

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=' %(message)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()

这给出了例如控制台中的输出：

Created file: ./test_file.h5
Modified directory: .
Modified file: ./test_file.h5
Modified directory: .

我只想在创建新文件时返回它们的名称，我不需要在控制台中返回它，而是在变量中返回，这样我就可以将它用作不同的功能。有没有办法做到这一点？

Answer 1

您需要创建自定义处理程序，可以通过继承FileSystemEventHandler 并覆盖您要使用的事件来完成。

class CustomHandler(FileSystemEventHandler):

    def __init__(self, callback: Callable):
        self.callback = callback

    def on_created(self, event: Union[DirCreatedEvent, FileCreatedEvent]):
        print(f"Event type: {event.event_type}\nAt: {event.src_path}\n")

        if isinstance(event, FileCreatedEvent):
            file = pathlib.Path(event.src_path)

            print(f"Processing file {file.name}\n")

            self.callback(file)

可用的事件有：

• on_modified（自我，事件）
• on_deleted(self, event)
• on_closed（自我，事件）
• on_moved（自我，事件）
• on_any_event(self, event)

对于 on_created，每个事件都可能有所不同 - 可以安全地假设它只有 DirCreatedEvent 和 FileCreatedEvent。

--

示例代码

import time
import pathlib
from typing import Union

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, DirCreatedEvent, FileCreatedEvent


class CustomHandler(FileSystemEventHandler):
    """Custom handler for Watchdog"""

    def __init__(self):
        # List to store path
        self.path_strings = []

    # callback for File/Directory created event, called by Observer.
    def on_created(self, event: Union[DirCreatedEvent, FileCreatedEvent]):
        print(f"Event type: {event.event_type}\nAt: {event.src_path}")

        # check if it's File creation, not Directory creation
        if isinstance(event, FileCreatedEvent):
            
            # if so, do something with event.src_path - it's path of the created file.
            self.path_strings.append(event.src_path)

            print(f"Path content: \n{self.path_strings}")


def main():
    # get current path as absolute, linux-style path.
    working_path = pathlib.Path(".").absolute().as_posix()

    # create instance of observer and CustomHandler
    observer = Observer()
    handler = CustomHandler()

    # start observer, checks files recursively
    observer.schedule(handler, path=working_path, recursive=True)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()


if __name__ == '__main__':
    main()

示例输出

Event type: created
At: E:/github/ProjectIncubator/single_run_scripts\test.a
Path content: 
['E:/github/ProjectIncubator/single_run_scripts\\test.a']
Event type: created
At: E:/github/ProjectIncubator/single_run_scripts\nyan.txt
Path content: 
['E:/github/ProjectIncubator/single_run_scripts\\test.a', 'E:/github/ProjectIncubator/single_run_scripts\\nyan.txt']

---

旧

完整示例代码

import time
import pathlib
import argparse
from typing import Union, Callable

from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler, DirCreatedEvent, FileCreatedEvent


class CustomHandler(FileSystemEventHandler):

    def __init__(self, callback: Callable):
        self.callback = callback

        # Store callback to be called on every on_created event

    def on_created(self, event: Union[DirCreatedEvent, FileCreatedEvent]):
        print(f"Event type: {event.event_type}\nAt: {event.src_path}\n")

        # check if it's File creation, not Directory creation
        if isinstance(event, FileCreatedEvent):
            file = pathlib.Path(event.src_path)

            print(f"Processing file {file.name}\n")

            # call callback
            self.callback(file)


def main():
    path: pathlib.Path = args.dir

    # list for new files
    created_files = []

    # create callback
    def callback(path_: pathlib.Path):
        print(f"Adding {path_.name} to list!")
        created_files.append(path_)

    # create instance of observer and CustomHandler
    observer = Observer()
    handler = CustomHandler(callback)

    observer.schedule(handler, path=path.absolute().as_posix(), recursive=True)
    observer.start()

    print("Observer started")

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

    print(f"{len(created_files)} new files was created!", "\n".join(p.name for p in created_files), sep="\n")
    input("Press enter to exit!")


if __name__ == '__main__':

    # get script's root
    ROOT = pathlib.Path(__file__).parent

    # parse argument - if provided given directory is used, otherwise 
    parser = argparse.ArgumentParser(description="Listen for file change in directory.")

    parser.add_argument("dir", metavar="DIR", type=pathlib.Path, default=ROOT, nargs="?", help="Directory to listen for. If omitted, script path is used.")

    args = parser.parse_args()

    main()

---

示例输出（换行很乱，抱歉！）

❯ py .\watchdog_test.py X:\test
Observer started
Event type: created
At: X:/test\새 폴더

Event type: created
At: X:/test\meow.txt

Processing file meow.txt

Adding meow.txt to list!
Event type: created
At: X:/test\meow.txt

Processing file meow.txt

Adding meow.txt to list!
Event type: created
At: X:/test\meow - 복사본.txt

Processing file meow - 복사본.txt

Adding meow - 복사본.txt to list!
Event type: created
At: X:/test\meow - 복사본 (2).txt

Processing file meow - 복사본 (2).txt

Adding meow - 복사본 (2).txt to list!
Event type: created
At: X:/test\meow - 복사본 (3).txt

Processing file meow - 복사본 (3).txt

Adding meow - 복사본 (3).txt to list!
5 new files was created!
meow.txt
meow.txt
meow - 복사본.txt
meow - 복사본 (2).txt
meow - 복사본 (3).txt
Press enter to exit!