مانیتور تغییرات فایل با watchdog در پایتون؛ راهنمای جامع و کاربردی

۱. مانیتورینگ تغییرات فایل در پایتون چگونه کار می‌کند؟

مانیتورینگ فایل در پایتون از طریق اتصال به APIهای سطح پایین سیستم‌عامل انجام می‌شود تا هرگونه تغییر در ساختار دایرکتوری‌ها را بدون تاخیر گزارش کند. کتابخانه قدرتمند Watchdog با مپ کردن رویدادهای سیستم‌عامل به توابع پایتون، به شما امکان می‌دهد تا برنامه‌های کاملاً واکنشی (Reactive) بسازید.

در سیستم‌عامل‌های مختلف، مکانیزم‌های متفاوتی برای این کار وجود دارد؛ به عنوان مثال در لینوکس از inotify، در ویندوز از ReadDirectoryChangesW و در مک از FSEvents استفاده می‌شود. Watchdog به عنوان یک لایه انتزاعی (Abstraction) روی تمام این ابزارها قرار می‌گیرد و کدی کاملاً کراس‌پلتفرم (مستقل از سیستم‌عامل) به شما تحویل می‌دهد.

۲. چرا کتابخانه Watchdog؟ مقایسه روش‌ها

برای مانیتور کردن فایل‌ها دو راهکار کلی وجود دارد: بررسی دوره‌ای (Polling) و برنامه‌نویسی مبتنی بر رویداد (Event-driven). روش اول به شدت منابع سیستم را مصرف می‌کند، در حالی که روش دوم کاملاً بهینه‌سازی شده است.

۳. راه‌اندازی سریع و نوشتن اولین اسکریپت رصد فایل

نصب کتابخانه Watchdog بسیار ساده است و پیش‌نیاز خاصی ندارد. کافی است ترمینال خود را باز کرده و دستور زیر را اجرا کنید:

pip install watchdog

پس از نصب، برای استفاده از این ابزار در برنامه‌های خود، باید دو بخش اصلی آن را بشناسیم: Observer (ناظر که وظیفه رصد فیزیکی دایرکتوری را دارد) و EventHandler (مدیریت‌کننده رویداد که مشخص می‌کند پس از بروز تغییر چه کاری باید انجام شود). اگر علاقمند به استفاده از پایتون در پروژه‌های مختلف هستید، می‌توانید بخش تکه‌کدهای پایتون را برای پروژه‌های کاربردی دیگر بررسی کنید.

قطعه کد پایه برای مانیتور یک پوشه

اسکریپت زیر ساده‌ترین حالت پیاده‌سازی مانیتورینگ است که تمام رویدادها را در کنسول چاپ می‌کند:

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

class SimpleHandler(FileSystemEventHandler):
    def on_any_event(self, event):
        # این متد هر نوع تغییری را ثبت می‌کند
        event_type = event.event_type
        is_directory = "پوشه" if event.is_directory else "فایل"
        print(f"رویداد: {event_type} | نوع: {is_directory} | مسیر: {event.src_path}")

if __name__ == "__main__":
    path_to_monitor = "./my_folder"  # پوشه مورد نظر برای رصد
    
    event_handler = SimpleHandler()
    observer = Observer()
    observer.schedule(event_handler, path=path_to_monitor, recursive=True)
    
    print(f"شروع نظارت بر روی پوشه: {path_to_monitor}")
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("توقف نظارت...")
        observer.stop()
    observer.join()

۴. مدیریت انواع رویدادها (ایجاد، ویرایش، حذف و جابجایی)

به جای استفاده از متد کلی on_any_event، ایده آل‌تر این است که برای هر تغییر فیزیکی، رفتار منحصربه‌فردی تعریف کنید. کلاس FileSystemEventHandler متدهای تخصصی برای این کار در اختیارتان قرار می‌دهد.

در زیر این متدها و نحوه شخصی‌سازی آن‌ها را به تفکیک مشاهده می‌کنید:

• on_created: زمانی اجرا می‌شود که یک فایل یا پوشه جدید ساخته شود.
• on_deleted: در صورت حذف هر فایلی فعال می‌شود.
• on_modified: پرکاربردترین متد است که زمان ویرایش محتوا یا تغییر ویژگی‌های فایل اجرا می‌شود.
• on_moved: زمانی که فایل تغییر نام داده یا به مسیر جدیدی منتقل شود فراخوانی می‌گردد. این متد علاوه بر src_path، ویژگی dest_path را نیز ارائه می‌دهد.

class CustomFileHandler(FileSystemEventHandler):
    def on_created(self, event):
        print(f"فایل جدید ساخته شد: {event.src_path}")

    def on_deleted(self, event):
        print(f"هشدار! فایل حذف شد: {event.src_path}")

    def on_modified(self, event):
        print(f"تغییر در فایل رخ داد: {event.src_path}")

    def on_moved(self, event):
        print(f"انتقال فایل از {event.src_path} به {event.dest_path}")

۵. چالش‌های پیشرفته و راه‌حل‌های دنیای واقعی (Debouncing و Queue)

در پروژه‌های واقعی، مانیتور کردن فایل‌ها به سادگیِ چاپ کردن پیام در کنسول نیست. زمانی که شروع به کار با سیستم‌های پر ترافیک می‌کنید، با مشکلات جدی مانند رویدادهای تکراری و بلاک شدن نخ اصلی (Thread) مواجه می‌شوید.

چالش اول: مشکل رویدادهای تکراری (Duplicate Events)

بسیاری از نرم‌افزارها (مانند ویرایشگرهای متن یا سیستم‌های فشرده‌سازی) هنگام ذخیره یک فایل، آن را در چند مرحله کوچک می‌نویسند یا یک فایل موقت ایجاد کرده و سپس جایگزین می‌کنند. این کار باعث می‌شود رویداد on_modified برای یک عملیات ذخیره‌سازی ساده، ۴ الی ۵ بار پشت سر هم اجرا شود.

راه‌حل: استفاده از تکنیک Debouncing با پیاده‌سازی کش زمانی کوتاه.

import time
from watchdog.events import FileSystemEventHandler

class DebouncedHandler(FileSystemEventHandler):
    def __init__(self, delay=1.0):
        self.delay = delay
        self.last_modified = {}

    def on_modified(self, event):
        if event.is_directory:
            return
            
        current_time = time.time()
        last_time = self.last_modified.get(event.src_path, 0)
        
        # اگر فاصله آخرین تغییر کمتر از مقدار مشخص شده بود، نادیده بگیر
        if current_time - last_time < self.delay:
            return
            
        self.last_modified[event.src_path] = current_time
        self.process_file(event.src_path)

    def process_file(self, filepath):
        print(f"پردازش ایمن فایل بدون تکرار بیهوده: {filepath}")

چالش دوم: مسدود شدن نخ ناظر (Blocking the Observer Thread)

نخ ناظرِ Watchdog نباید درگیر پردازش‌های سنگین (مانند آپلود فایل، بهینه‌سازی تصاویر یا خواندن دیتابیس‌های حجیم) شود. اگر یک تابع زمان‌بر را مستقیم در کلاس EventHandler بنویسید، رصد کردن بقیه تغییرات متوقف خواهد شد.

راه‌حل: استفاده از یک صف (Queue) مجزا برای پردازش رویدادها توسط یک Worker در بک‌گراند.

import queue
import threading
from watchdog.events import FileSystemEventHandler

# ایجاد یک صف سراسری یا محلی
event_queue = queue.Queue()

class QueueHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            # فقط مسیر فایل را در صف قرار می‌دهیم و خارج می‌شویم
            event_queue.put(event.src_path)

def worker():
    while True:
        # کارگر در نخ جداگانه صف را بررسی می‌کند و کار سنگین انجام می‌دهد
        file_path = event_queue.get()
        if file_path is None:
            break
        print(f"نخ فرعی در حال پردازش سنگین روی فایل: {file_path}")
        time.sleep(3) # شبیه‌سازی کار سنگین
        event_queue.task_done()

# راه‌اندازی نخ کارگر
worker_thread = threading.Thread(target=worker, daemon=True)
worker_thread.start()

چالش سوم: محدودیت‌های سیستم‌عامل (به خصوص در لینوکس)

در لینوکس، سرویس inotify محدودیت مشخصی برای رصد همزمان فایل‌ها دارد (fs.inotify.max_user_watches). اگر پوشه پروژه شما بسیار بزرگ باشد (مثلاً شامل فولدرهای سنگینی مانند node_modules)، با ارور مواجه خواهید شد.

راه‌حل: فیلتر کردن پوشه‌ها در کدهای پایتون یا افزایش لیمیت سیستم‌عامل با وارد کردن فرمان زیر در لینوکس:

echo fs.inotify.max_user_watches=524288 | sudo tee -a /etc/sysctl.conf && sudo sysctl -p

۶. پروژه عملی: پشتیبان‌گیری خودکار و هوشمند فایل‌ها

بیایید یک سناریوی واقعی را پیاده کنیم: یک دایرکتوری به نام /source داریم. می‌خواهیم به محض قرار گرفتن هر فایل تصویری (با فرمت‌های jpg, png)، این فایل به صورت خودکار فشرده شده یا به دایرکتوری /backup کپی شود.

import os
import shutil
import time
from watchdog.observers import Observer
from watchdog.events import PatternMatchingEventHandler

class ImageBackupHandler(PatternMatchingEventHandler):
    # مشخص کردن فیلترها روی فرمت‌های دلخواه
    def __init__(self):
        super().__init__(
            patterns=["*.jpg", "*.jpeg", "*.png"], 
            ignore_directories=True, 
            case_sensitive=False
        )
        self.backup_dir = "./backup_folder"
        if not os.path.exists(self.backup_dir):
            os.makedirs(self.backup_dir)

    def on_created(self, event):
        filename = os.path.basename(event.src_path)
        dest_path = os.path.join(self.backup_dir, filename)
        
        # تاخیر جزیی برای اطمینان از پایان کپی فایل به پوشه مبدا
        time.sleep(0.5)
        
        try:
            shutil.copy2(event.src_path, dest_path)
            print(f"با موفقیت پشتیبان‌گیری شد: {filename} -> {dest_path}")
        except Exception as e:
            print(f"خطا در کپی فایل {filename}: {e}")

if __name__ == "__main__":
    src_dir = "./source_folder"
    if not os.path.exists(src_dir):
        os.makedirs(src_dir)
        
    event_handler = ImageBackupHandler()
    observer = Observer()
    observer.schedule(event_handler, path=src_dir, recursive=False)
    
    print(f"سیستم پشتیبان‌گیری خودکار فعال شد. پوشه تحت رصد: {src_dir}")
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

۷. عیب‌یابی سریع و رفع خطاهای رایج

هنگام توسعه این اسکریپت‌ها ممکن است با خطاهای زیر مواجه شوید. در این بخش راه‌حل‌های سریع هر یک را به زبان ساده آورده‌ایم:

from watchdog.observers.polling import PollingObserver as Observer

۸. پرسش‌های متداول (FAQ)