python watchdog监视文件的变化

Cma*_*mag 50 python monitoring file watchdog

伙计们,我需要查看日志文件以进行更改.在查看了stackoverflow问题之后,我看到人们推荐"看门狗".所以我正在尝试测试,并且不确定在文件更改时添加代码的位置:

#!/usr/bin/python
import time
from watchdog.observers import Observer
from watchdog.events import LoggingEventHandler

if __name__ == "__main__":
    event_handler = LoggingEventHandler()
    observer = Observer()
    observer.schedule(event_handler, path='.', recursive=False)
    observer.start()
    try:
        while True:
            time.sleep(1)
        else:
            print "got it"
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
Run Code Online (Sandbox Code Playgroud)

如果文件已被添加/更改,我在哪里添加"得到它",在while循环中?

ale*_*cxe 99

而不是LoggingEventHandler定义您的处理程序:

#!/usr/bin/python
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler


class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        print(f'event type: {event.event_type}  path : {event.src_path}')


if __name__ == "__main__":
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path='/data/', recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()
Run Code Online (Sandbox Code Playgroud)

on_modified 在修改文件或目录时调用.

  • @DanTenenbaum你有没有找到如何修复重复的'得到'消息?我也连续收到两条消息. (3认同)
  • 非常好,谢谢!现在我怎么只看一个特定的文件,现在整个目录? (2认同)
  • 别客气.您可以查看包含该文件的目录,然后在处理程序方法中检查`event.src_path`变量. (2认同)
  • 此代码锁定文件,因此其他程序无法使用它 (2认同)
  • ```observer.join()``` 在代码中做了什么?观察者是否有自己的线程,还是在主线程中运行(试图找出为什么“time.sleep(1)”调用会起作用,因为观察者似乎使用线程)?watdog 文档似乎有点稀疏,所以试图拼凑出如何使用这个东西...... (2认同)

run*_*ace 6

这是一个防止其运行两次的代码段,因为其他人在@alecxe答案中对此进行了评论:

from datetime import datetime, timedelta

class MyHandler(FileSystemEventHandler):
    def __init__(self):
        self.last_modified = datetime.now()

    def on_modified(self, event):
        if datetime.now() - self.last_modified < timedelta(seconds=1):
            return
        else:
            self.last_modified = datetime.now()
        print(f'Event type: {event.event_type}  path : {event.src_path}')
        print(event.is_directory) # This attribute is also available
Run Code Online (Sandbox Code Playgroud)