Posted inPython

Tự động hóa giám sát File với Python Watchdog: Ngừng ‘Polling’ và bắt đầu ‘Listening’

Python tutorial - IT technology blog
Python tutorial - IT technology blog

Vấn đề của việc kiểm tra thủ công và những giải pháp “chắp vá”

Hồi mới làm IT, mình từng nhận một task khá oái oăm. Cứ khi nào khách upload file thiết kế lên server qua FTP, mình phải copy ngay vào folder dự phòng và nhắn tin báo cho team. Nghe thì đơn giản, nhưng khách có thể upload bất cứ lúc nào, kể cả 2 giờ sáng hay đúng lúc mình đang ăn cơm.

Ban đầu, mình cứ 15 phút lại remote vào server để kiểm tra một lần. Cách này vừa tốn sức vừa dễ sót việc. Sau đó, mình viết một script Python dùng vòng lặp while True kết hợp time.sleep(60) để quét danh sách file. Tuy nhiên, nếu để thời gian chờ lâu thì bị trễ, còn nếu để 1 giây quét một lần thì CPU server luôn nhảy vọt lên 20-30% chỉ để đọc ổ cứng liên tục.

Sau nhiều lần “vấp cỏ”, mình nhận ra việc ngồi đợi hệ thống file tự thông báo khi có thay đổi sẽ hiệu quả hơn nhiều so với việc mình chủ động đi hỏi nó.

Tại sao Polling (quét định kỳ) lại là giải pháp tồi?

Trong lập trình, kỹ thuật quét định kỳ này gọi là Polling. Bạn liên tục hỏi hệ điều hành: “Có file mới chưa?”. Cách này có hai nhược điểm chí mạng:

  • Ngốn tài nguyên I/O: Việc đọc danh sách hàng ngàn file liên tục là tác vụ cực kỳ nặng nề. Nó khiến ổ cứng phải hoạt động liên tục, làm giảm tuổi thọ thiết bị và chậm các ứng dụng khác.
  • Độ trễ (Latency): Nếu bạn quét mỗi 5 phút, file có thể đã nằm chờ 4 phút 59 giây trước khi được xử lý. Với các hệ thống cần phản hồi ngay lập tức, con số này là không thể chấp nhận.

Các hệ điều hành hiện đại như Linux (inotify), macOS (FSEvents) hay Windows đều có cơ chế Event-driven. Thay vì đi hỏi, chương trình của chúng ta sẽ đăng ký với OS để nhận tín hiệu ngay khi có sự kiện tạo, xóa hoặc sửa file.

Watchdog – Công cụ giám sát file chuyên nghiệp

Thư viện Watchdog của Python là giải pháp tối ưu để tận dụng cơ chế Event-driven này. Nó hoạt động đa nền tảng và cực kỳ tiết kiệm tài nguyên CPU.

Watchdog vận hành dựa trên hai thành phần cốt lõi:

  1. Observer: Một thread chạy ngầm làm nhiệm vụ “hóng” các tín hiệu từ hệ điều hành.
  2. Event Handler: Tập hợp các hàm xử lý logic khi có sự kiện cụ thể xảy ra (như on_modified hay on_created).

Triển khai công cụ giám sát cơ bản

Đầu tiên, hãy cài đặt thư viện qua pip bằng lệnh:

pip install watchdog

Dưới đây là khung code (boilerplate) mình thường dùng. Nó sẽ theo dõi một thư mục và in thông báo ngay khi có biến động.

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} vừa được cập nhật.")

    def on_created(self, event):
        if not event.is_directory:
            print(f"Phát hiện file mới: {event.src_path}")

if __name__ == "__main__":
    path = "./my_folder" 
    event_handler = MyHandler()
    observer = Observer()
    observer.schedule(event_handler, path, recursive=True)
    
    print(f"Hệ thống đang giám sát thư mục: {path}...")
    observer.start()
    
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

Lưu ý nhỏ: Tham số recursive=True giúp bạn quản lý luôn cả các thư mục con. Nếu chỉ cần theo dõi thư mục gốc để giảm tải cho memory, hãy chuyển nó thành False.

Ứng dụng: Tự động phân loại file Downloads

Hãy thử nâng cấp script để giải quyết một vấn đề thực tế: Tự động dọn dẹp thư mục Downloads. Cứ mỗi khi có file .jpg hoặc .pdf tải về, hệ thống sẽ tự động chuyển chúng vào các folder tương ứng.

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

class CleanupHandler(FileSystemEventHandler):
    def on_created(self, event):
        if event.is_directory:
            return
        
        filename = os.path.basename(event.src_path)
        ext = os.path.splitext(filename)[1].lower()
        
        dest_map = {
            ".jpg": "./Images", ".png": "./Images",
            ".pdf": "./Documents", ".docx": "./Documents"
        }
        
        destination = dest_map.get(ext)
        if destination:
            os.makedirs(destination, exist_ok=True)
            # Chờ 1s để đảm bảo file đã được ghi xong xuống đĩa
            time.sleep(1)
            shutil.move(event.src_path, os.path.join(destination, filename))
            print(f"Đã di chuyển {filename} sang {destination}")

Mẹo nhỏ ở đây là hàm time.sleep(1). Khi file đang tải về, OS có thể phát ra sự kiện created trước khi dữ liệu được ghi xong. Việc chờ một chút giúp tránh lỗi “File in use” khi bạn thực hiện di chuyển file.

Lưu ý quan trọng khi triển khai thực tế

Để script chạy ổn định trong môi trường production, bạn cần chú ý 3 điểm sau:

  1. Tránh vòng lặp sự kiện: Nếu script ghi log vào chính thư mục đang giám sát, Watchdog sẽ bắt được sự kiện ghi log đó và kích hoạt tiếp. Điều này tạo ra một vòng lặp vô tận khiến script bị treo.
  2. Xử lý ngoại lệ: Luôn sử dụng try-except khi thao tác với file. Nếu một file đang bị khóa bởi phần mềm khác (như Excel), lệnh shutil.move sẽ gây crash toàn bộ chương trình.
  3. Duy trì hoạt động: Trên Linux, hãy cài đặt script như một systemd service. Cách này đảm bảo script tự khởi động lại nếu server reboot hoặc gặp lỗi bất ngờ.

Việc áp dụng Watchdog giúp mình tiết kiệm ít nhất 30-40 phút mỗi ngày cho các tác vụ kiểm tra vụn vặt. Nếu bạn đang có những luồng công việc lặp lại liên quan đến file, hãy thử viết một script giám sát ngay hôm nay.

Share:
Tags: