Phát hiện thay đổi trong tệp python

pip install watchdog
# If you're using Poetry instead of Virtualenv
poetry add watchdog.

Đây là một ví dụ để bắt đầu với. Tập lệnh Python sau sẽ theo dõi các thay đổi của tệp trong thư mục hiện tại. Nó sẽ ghi lại tất cả các thay đổi khi chúng xảy ra.  

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


class FileCreateHandler[FileSystemEventHandler]:
    def on_created[self, event]:
        print["Created: " + event.src_path]

if __name__ == "__main__":

    event_handler = FileCreateHandler[]

    # Create an observer.
    observer = Observer[]

    # Attach the observer to the event handler.
    observer.schedule[event_handler, ".", recursive=True]

    # Start the observer.
    observer.start[]

    try:
        while observer.is_alive[]:
            observer.join[1]
    finally:
        observer.stop[]
        observer.join[]

Phần quan trọng của đoạn mã trên là lớp FileProcessor. Nhưng trước khi vào đó, chúng ta nên tạo một đối tượng quan sát viên để đính kèm một trình xử lý sự kiện.  

Chúng ta có thể đính kèm một trình xử lý sự kiện cho người quan sát bằng cách sử dụng phương thức lịch biểu. Trong ví dụ trên, chúng tôi đã đính kèm nó để xem các sự kiện trong thư mục hiện tại và tất cả các thư mục xuôi dòng của nó.  

Nếu bạn chạy mã này và tạo một tệp mới trong thư mục hiện tại, chúng ta có thể thấy tập lệnh Python in sự kiện trên thiết bị đầu cuối.  

Chúng tôi đã sử dụng thư mục hiện tại trong phương thức schedule của đối tượng người quan sát. Bạn cũng có thể sử dụng bất kỳ con đường nào bạn chọn. Bạn cũng có thể chọn lấy nó từ đối số dòng lệnh.  

Có liên quan. Làm cách nào để thực thi các lệnh Shell với Python?

Đây là một sửa đổi đối với cùng mã chuyển đổi tập lệnh của chúng tôi thành CLI. Bây giờ bạn có thể chuyển đường dẫn tới màn hình bằng đối số dòng lệnh.  

...
import argparse

...


if __name__ == "__main__":
    parser = argparse.ArgumentParser[
        description="Watchdog script to watch for new CSV files and load them into the database"
    ]
    parser.add_argument["path", help="Path to watch for new CSV files"]

    args = parser.parse_args[]
    path = args.path

    # Create an observer.
    observer = Observer[]

    observer.schedule[event_handler, path, recursive=True]
    
    ...

Bây giờ bạn có thể chạy tập lệnh của mình như sau trong thiết bị đầu cuối của mình

python .py /somewhare/in/your/computer

Trong bài viết trước của tôi, bạn có thể tìm hiểu thêm về cách tạo giao diện dòng lệnh bằng Python

Xử lý các thay đổi đối với tệp trong lớp trình xử lý

Trong ví dụ của chúng tôi, chúng tôi đã tạo một trình xử lý sự kiện bằng cách phân lớp con lớp 'FileSystemEventHandler'. Tất cả các trình xử lý sự kiện phải như thế này.  

Lớp cha có trình giữ chỗ cho một số phương thức cho các sự kiện hệ thống tệp. Chúng tôi đã sử dụng phương thức 'on_create' để xử lý tất cả các tệp mới được tạo. Tương tự như vậy, bạn cũng có thể sử dụng các phương thức on_deleted, on_modified, on_moved và on_any_event để xử lý các loại sự kiện khác.  

Hãy cập nhật on_create để xử lý tệp và chèn giá trị vào cơ sở dữ liệu. Vui lòng bỏ qua phần này nếu nó không liên quan đến trường hợp sử dụng của bạn.  

import os

import pandas as pd
from sqlalchemy import create_engine
from watchdog.events import FileSystemEventHandler
...


class FileCreateHandler[FileSystemEventHandler]:
    def on_created[self, event]:
        # Create SQLAlchemy engine for postgres database using environment variables
        engine = create_engine[os.environ["DATABASE_URL"]]
        
        with engine.connect[] as con:
            # Read the CSV data
            df = pd.read_csv[event.src_path]

            # Do the required transformations
            df["date"] = pd.to_datetime[df["date"]]

            # Write the data to the database
            df.to_sql["weather", con, if_exists="append", index=False]
            
...

Mã này có thể trông rất quen thuộc nếu bạn đã làm việc với Sqlalchemy và Pandas. Điều đáng chú ý là cách chúng ta lấy đường dẫn của tệp mới tạo.  

Mỗi trình kích hoạt sự kiện trong lớp 'FileSystemEventHandler' tham chiếu đường dẫn trong đối số sự kiện của nó. Chúng tôi có thể truy cập nó bằng thẻ src-path như trong mã.  

Nếu bạn chạy mã và để tập lệnh Python lắng nghe các thay đổi, nó cũng sẽ ngay lập tức đẩy các thay đổi đó vào cơ sở dữ liệu.  

Phục vụ ứng dụng của bạn trong nền

Đến bây giờ, bạn đã nhận thấy rằng ứng dụng của chúng tôi chạy trên thiết bị đầu cuối trực tiếp. Nhưng nó không phải là khôn ngoan để làm điều này trong sản xuất. Bất cứ điều gì đối với phiên cuối có thể ảnh hưởng đến ứng dụng.  

Cách tốt nhất để chạy các dịch vụ đó trong nền là thông qua một dịch vụ hệ thống. Người dùng Windows có thể sử dụng công cụ NSSM [Non-sucking Service Manager. ] Sẽ khá đơn giản nếu bạn lướt qua tài liệu của họ.  

Nhưng trong bài đăng này, tôi sẽ đề cập đến việc sử dụng systemctl của hệ thống Linux.  

Bạn có thể tạo một dịch vụ hệ thống mới bằng cách tạo một tệp có nội dung sau trong thư mục '/etc/systemd/system'. Bạn có thể đặt tên cho nó bất cứ thứ gì với phần mở rộng là '. Dịch vụ'

[Unit]
Description="Process Data"

[Service]
Restart=always
WorkingDirectory=
ExecStart= .py

[Install]
WantedBy=multi-user target

Sau khi hoàn tất, bạn có thể chạy các lệnh sau trên thiết bị đầu cuối để kích hoạt dịch vụ.  

# To make your new service available to systemctl utility.
$ systemctl daemon-reload

# To start the service 
$ systemctl start .service

Điều này sẽ bắt đầu quá trình. Bây giờ, khi các tệp mới đang được tạo trên đích FTP của chúng tôi, dịch vụ này sẽ xử lý và tải chúng lên cơ sở dữ liệu. Và cơ sở dữ liệu thời gian thực của chúng tôi sẽ nhận được dữ liệu mới mà không có bất kỳ sự chậm trễ nào.  

Bạn có thể kiểm tra xem dịch vụ của mình có chạy đúng không bằng lệnh sau.  

systemctl status .service

suy nghĩ cuối cùng

Ngày nay, việc xử lý thay đổi hệ thống tệp rất hiếm khi thế giới hướng tới sự tích hợp mạnh mẽ hơn giữa các hệ thống. Nhưng điều đó không có nghĩa là trình kích hoạt hệ thống tập tin không có cách sử dụng.  

Có nhiều trường hợp chúng ta cần theo dõi việc tạo hoặc sửa đổi tệp mới. Lấy ví dụ, xử lý luồng nhật ký. Bạn có thể sử dụng kỹ thuật được mô tả ở đây để xử lý một dòng nhật ký mới và đẩy nó vào kho dữ liệu.  

Tôi đã sử dụng nó trong một thời gian dài. Và tôi không thấy cần phải giảm bớt.  

Bạn có thích những gì bạn đọc?

Làm thế nào để Python biết các thay đổi đối với một thư mục?

Trình theo dõi tệp là gì?