Python watchdog灵活监控文件和目录的变化

pip install watchdog

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 创建一个自定义事件处理器
class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"File {event.src_path} has been modified")
# 创建一个观察者并启动
observer = Observer()
event_handler = MyHandler()
observer.schedule(event_handler, path="path/to/your/file", recursive=False)
observer.start()
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 创建一个自定义事件处理器
class MyHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"File {event.src_path} has been modified")
# 创建一个观察者并启动
observer = Observer()
event_handler = MyHandler()
observer.schedule(event_handler, path="path/to/your/directory", recursive=True)
observer.start()
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
# 自定义事件处理器
class BuildHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory:
            print(f"File {event.src_path} has been modified. Triggering build...")
        # 在此处添加构建和测试的操作
# 创建观察者并启动
observer = Observer()
event_handler = BuildHandler()
observer.schedule(event_handler, path="path/to/your/project", recursive=True)
observer.start()
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()

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

source_dir = "path/to/source"
target_dir = "path/to/target"

# 自定义事件处理器
class SyncHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            src_file = event.src_path
            dest_file = src_file.replace(source_dir, target_dir)
            shutil.copy2(src_file, dest_file)
            print(f"File {src_file} has been created. Syncing to {dest_file}")

    def on_modified(self, event):
        if not event.is_directory:
            src_file = event.src_path
            dest_file = src_file.replace(source_dir, target_dir)
            shutil.copy2(src_file, dest_file)
            print(f"File {src_file} has been modified. Syncing to {dest_file}")

    def on_deleted(self, event):
        if not event.is_directory:
            src_file = event.src_path
            dest_file = src_file.replace(source_dir, target_dir)
            try:
                os.remove(dest_file)
                print(f"File {src_file} has been deleted. Removing from {dest_file}")
            except FileNotFoundError:
                pass

# 创建观察者并启动
observer = Observer()
event_handler = SyncHandler()
observer.schedule(event_handler, path=source_dir, recursive=True)
observer.start()

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

observer.join()

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
log_file = "path/to/your/logfile.log"
# 自定义事件处理器
class LogHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory and event.src_path == log_file:
            with open(log_file, "r") as file:
                new_entries = file.readlines()
                for entry in new_entries:
                    # 在这里添加日志分析和处理的操作
                    print(f"New log entry: {entry}")
# 创建观察者并启动
observer = Observer()
event_handler = LogHandler()
observer.schedule(event_handler, path="path/to/your", recursive=False)
observer.start()
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    observer.stop()
observer.join()