在Python中创建看门狗以查找文件系统更改
很多时候,需要在创建或修改文件时对其进行处理。这可以通过跟踪特定目录中的更改来完成。 Python中有很多方法可以跟踪目录中所做的更改。一种这样的方法是使用watchdog
模块。顾名思义,这个模块观察给定的目录,并且可以通知文件是否被创建或更改。
需要的模块
- 看门狗
要安装看门狗,请在终端中运行此命令。pip install watchdog
- 日志记录
它与Python一起内置,因此无需外部安装。
现在让我们深入研究将记录所有已注册更改的代码。此代码将记录仅在当前目录中注册的更改。
# import the modules
import sys
import time
import logging
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler
if __name__ == "__main__":
# Set the format for logging info
logging.basicConfig(level=logging.INFO,
format='%(asctime)s - %(message)s',
datefmt='%Y-%m-%d %H:%M:%S')
# Set format for displaying path
path = sys.argv[1] if len(sys.argv) > 1 else '.'
# Initialize logging event handler
event_handler = LoggingEventHandler()
# Initialize Observer
observer = Observer()
observer.schedule(event_handler, path, recursive=True)
# Start the observer
observer.start()
try:
while True:
# Set the thread sleep time
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
输出:
上面的代码将记录在目录中注册的所有更改。如果要在创建或修改文件时进行更改,可以使用以下代码进行。
# import time module, Observer, FileSystemEventHandler
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class OnMyWatch:
# Set the directory on watch
watchDirectory = "/give / the / address / of / directory"
def __init__(self):
self.observer = Observer()
def run(self):
event_handler = Handler()
self.observer.schedule(event_handler, self.watchDirectory, recursive = True)
self.observer.start()
try:
while True:
time.sleep(5)
except:
self.observer.stop()
print("Observer Stopped")
self.observer.join()
class Handler(FileSystemEventHandler):
@staticmethod
def on_any_event(event):
if event.is_directory:
return None
elif event.event_type == 'created':
# Event is created, you can process it now
print("Watchdog received created event - % s." % event.src_path)
elif event.event_type == 'modified':
# Event is modified, you can process it now
print("Watchdog received modified event - % s." % event.src_path)
if __name__ == '__main__':
watch = OnMyWatch()
watch.run()
输出:
破译密码
观察者
Observer
是监视任何文件系统更改然后将事件分派给事件处理程序的类。它监视文件系统并查找任何更改。
事件处理程序
事件处理程序是一个对象,当文件系统发生某些事情时将得到通知。通常,编写脚本来监视创建或修改的任何类型的新文件,例如 jpg、xml 等。例如,在下面的代码中, PatternMatchingEventHandler
继承自FileSystemEventHandler
类并用于执行此操作。这个类的一些有用的方法是:
- on_any_event:将针对任何事件执行。
- on_created:在创建文件或目录时执行。
- on_modified:当文件被修改或目录被重命名时执行。
- on_deleted:删除文件或目录时执行。
- on_moved:移动文件或目录时执行。
这些方法中的每一个都接收事件对象作为第一个参数,并且事件对象具有 3 个属性:
- event_type:修改/创建/移动/删除
- is_directory:真/假
- src_path:路径/到/观察/文件
以下脚本用于仅使用PatternMAtchingEventHandler
观察 .csv 文件。如果您想观察不止一种类型的文件,您可以进一步扩展模式列表。
import watchdog.events
import watchdog.observers
import time
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_created(self, event):
print("Watchdog received created event - % s." % event.src_path)
# Event is created, you can process it now
def on_modified(self, event):
print("Watchdog received modified event - % s." % event.src_path)
# Event is modified, you can process it now
if __name__ == "__main__":
src_path = r"C:\Users\GeeksforGeeks\PycharmProjects\requests hotel"
event_handler = Handler()
observer = watchdog.observers.Observer()
observer.schedule(event_handler, path=src_path, recursive=True)
observer.start()
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
observer.stop()
observer.join()
输出:
在PatternMAtchingEventHandler
的帮助下,我们可以利用仅处理与扩展名为 .csv 的文件相关的事件。通过扩展 Watchdog 提供的事件处理程序类,我们可以通过实现上述类方法来处理修改、创建、删除和移动的事件。