使用看门狗将新创建的文件名放入变量中
Using watchdog to put newly created file names into variables
我想使用 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=' %(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: .
我只希望在创建新文件时返回它们的名称，我不需要在控制台中返回它，而只是在变量中返回，这样我就可以将它用作输入对于不同的功能。有办法吗？
您需要创建自定义处理程序，可以通过继承 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（自我，事件）
- on_closed（自我，事件）
- on_moved（自我，事件）
- on_any_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!
我想使用 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=' %(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: .
我只希望在创建新文件时返回它们的名称，我不需要在控制台中返回它，而只是在变量中返回，这样我就可以将它用作输入对于不同的功能。有办法吗？
您需要创建自定义处理程序，可以通过继承 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（自我，事件）
- on_closed（自我，事件）
- on_moved（自我，事件）
- on_any_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!