Python Watchdog报WinError32：文件被占用问题求助

问题描述 #

用Python的Watchdog监控test_folder文件夹，当特定Parquet文件创建时，调用处理函数读取文件并生成JPEG图片保存到同一目录。首次运行正常，但上传不同内容或名称的文件时，触发PermissionError: [WinError 32]，提示文件被另一进程占用。

问题根源 #

1. 文件未完全写入就触发处理：Windows系统中，文件创建事件会在文件开始上传时立即触发，此时文件还被上传进程占用，直接读取会报错。
2. 同名输出文件冲突：原代码中JPEG文件名固定为GAF_photo.jpeg，如果之前的文件还未完全释放，新的写入操作会触发占用错误。
3. 文件资源释放不彻底：处理函数中读取Parquet文件的方式可能未完全释放文件句柄，增加了占用风险。

解决步骤 #

• 等待文件写入完成：在触发创建事件后，循环尝试读取文件，直到能正常访问再开始处理。
• 生成唯一输出文件名：基于原Parquet文件名生成JPEG文件名，避免同名覆盖冲突。
• 优化文件资源管理：使用上下文管理器确保Parquet文件读取后及时释放资源。

修改后的代码 #

Watchdog监控代码 #

import sys
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import Final
from pathlib import Path

class OnMyWatch:
    watchDirectory = r'C:/Users/WEL4HO/Desktop/test_folder'
    def __init__(self):
        self.observer = Observer()
    def run(self):
        event_handler = Handler()
        self.observer.schedule(event_handler, self.watchDirectory, recursive=True)
        self.observer.start()
        try:
            while True:
                time.sleep(1)
        except:
            self.observer.stop() 
        self.observer.join()
        
class Handler(FileSystemEventHandler):
    @staticmethod
    def wait_for_file_ready(file_path, timeout=30):
        """循环等待文件可被读取，超时返回False"""
        start_time = time.time()
        while time.time() - start_time < timeout:
            try:
                with open(file_path, 'rb'):
                    return True
            except (PermissionError, OSError):
                time.sleep(0.5)
        return False

    @staticmethod
    def on_any_event(event):
        if event.is_directory:
            return None
  
        elif event.event_type == 'created':
            print(f'事件：{event.src_path} 已创建！')
            if event.src_path.endswith('Sampling2000KHz_AEKi-0.parquet'):
                # 等待文件写入完成
                if Handler.wait_for_file_ready(event.src_path):
                    # 生成唯一输出文件名
                    file_stem = Path(event.src_path).stem
                    output_name = f'{file_stem}_GAF_photo.jpeg'
                    Final.GAF(event.src_path, output_name)
                else:
                    print(f'超时：无法访问文件 {event.src_path}')
            
if __name__ == '__main__':
    watch = OnMyWatch()
    watch.run()

处理函数代码 #

import matplotlib.pyplot as plt
from mpl_toolkits.axes_grid1 import ImageGrid
from pyts.image import GramianAngularField
import numpy as np
import pyarrow.parquet as pq
from scipy.signal import savgol_filter

def GAF(path, outputname):
    Sampling_rate = 2000000  # AE采样率
    # 使用上下文管理器确保Parquet文件资源释放
    with pq.ParquetFile(path) as pf:
        df = pf.read().to_pandas()
    
    acoustic_emission_1 = df.values
    time_length = len(acoustic_emission_1) / Sampling_rate

    # 降采样
    downsampling_rate = 90
    clip_length = len(acoustic_emission_1) - (len(acoustic_emission_1) % downsampling_rate)
    acoustic_emission_1_clip = acoustic_emission_1[:clip_length]
    acoustic_emission_1_downsampled = acoustic_emission_1_clip.reshape(-1, downsampling_rate).mean(axis=1)

    # 提取正数值（简化写法）
    arr = acoustic_emission_1_downsampled[acoustic_emission_1_downsampled >= 0]

    # 滤波窗口长度处理
    window_length = int(round(len(arr) * 0.285, 0))
    window_length = window_length + 1 if window_length % 2 == 0 else window_length
    print('window_length:', window_length)

    y_filtered = savgol_filter(arr, window_length, 3)
    print('滤波后长度:', len(y_filtered))

    # 生成x轴数据（简化写法）
    x = np.arange(len(y_filtered))
    print('x轴长度:', len(x))

    X = np.array([y_filtered, x], dtype=object)

    # 转换为Gramian Angular Field
    gasf = GramianAngularField(image_size=1000, method='summation')
    X_gasf = gasf.transform(X)

    # 绘制并保存图像
    fig = plt.figure(figsize=(20, 10))
    grid = ImageGrid(fig, 111,
                     nrows_ncols=(1, 1),
                     axes_pad=0.15,
                     share_all=True,
                     cbar_location="right",
                     cbar_mode="single",
                     cbar_size="7%",
                     cbar_pad=0.3,
                     )
    for image, title, ax in zip([X_gasf[0]], ['Summation'], grid):
        im = ax.imshow(image, cmap='rainbow', origin='lower')
        ax.set_title(title, fontdict={'fontsize': 12})
    ax.cax.colorbar(im)
    ax.cax.toggle_label(True)
    plt.suptitle('Gramian Angular Field', y=0.98, fontsize=16)
    
    fig.savefig(outputname, dpi=300)
    plt.close(fig)
    # 显式清理变量，加速资源释放
    del fig, df, acoustic_emission_1, y_filtered, X, X_gasf

内容的提问来源于stack exchange，提问作者jasmin