gorakhargosh / watchdog

Python library and shell utilities to monitor filesystem events.
http://packages.python.org/watchdog/
Apache License 2.0
6.49k stars 695 forks source link

How to make the watchdog not monitor Python scripts' operations such as opening and modifying files #1067

Open chanly921 opened 1 week ago

chanly921 commented 1 week ago

How to make the watchdog not monitor Python scripts' operations such as opening and modifying files, modification, and other operations. Here is a simple example, when I open a file, I will execute the decrypt_file function, and the opening and modification of the file by the decrypt_file function will be detected by the watchdog.

def read_file(file_path: str) -> bytes:
    with open(file_path, 'rb') as f:
        data = f.read()
    return data

def encrypt_file(file_path: str):
    data = read_file(file_path)
    with open(file_path, 'wb') as f:
        print('File successfully encrypted...')
        f.write(data)

def decrypt_file(file_path: str):
    data = read_file(file_path)
    with open(file_path, 'wb') as f:
        print('File successfully decrypted...')
        f.write(data)

class FileEncryptHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory and re.search(FILE_PATTERN, os.path.basename(event.src_path)):
            print(f'File {os.path.basename(event.src_path)} modified...')
            encrypt_file(event.src_path)

    def on_opened(self, event):
        if not event.is_directory and re.search(FILE_PATTERN, os.path.basename(event.src_path)):
            print(f'File {os.path.basename(event.src_path)} opened up...')
            decrypt_file(event.src_path)

2024-09-06_17-31

BoboTiG commented 1 week ago

I am not sure to understand what you want :thinking:

chanly921 commented 1 week ago

I want to modify the target file in the on_opened and on_madified methods of watchdog. The action of modifying target file is not monitored by watchdog.

BoboTiG commented 1 week ago

Can you share the code?

chanly921 commented 1 week ago

Of course. After executing a Python program, opening Word will result in the current situation. After opening Word, executing the Python program and making modifications to Word will also result in the current situation.

import hashlib
import os
import re
import time

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

def read_file(file_path: str) -> bytes:
    with open(file_path, 'rb') as f:
        data = f.read()
    return data

def encrypt_file(file_path: str):
    try:
        data = read_file(file_path)
        if data[:len(key)] == key:
            return
        with open(file_path, 'wb') as f:
            print('File successfully encrypted...')
            f.write(key + data)
    except Exception:
        pass

def decrypt_file(file_path: str):
    try:
        data = read_file(file_path)
        if data[:len(key)] != key:
            return
        with open(file_path, 'wb') as f:
            print('File successfully decrypted...')
            f.write(data[len(key):])
    except Exception:
        pass

class FileEncryptHandler(FileSystemEventHandler):
    def on_modified(self, event):
        if not event.is_directory and re.search(r'^(?!.*\.\~).*\.doc$|^(?!.*\.\~).*\.docx$', event.src_path):
            print(f'File {os.path.basename(event.src_path)} modified...')
            encrypt_file(event.src_path)

    def on_opened(self, event):
        if not event.is_directory and re.search(r'^(?!.*\.\~).*\.doc$|^(?!.*\.\~).*\.docx$', event.src_path):
            print(f'File {os.path.basename(event.src_path)} opened up...')
            decrypt_file(event.src_path)

def start_monitoring():
    handler = FileEncryptHandler()
    observer = Observer()
    observer.schedule(handler, folder, recursive=True)
    observer.start()
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

if __name__ == "__main__":
    print('Start...')
    folder = r'/home/c/test'  # Change to test directory
    key = hashlib.sha256('12gs5349dadh2894lr34'.encode()).digest()
    start_monitoring()
    print('End...')