You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何在Python中轻松记录所有磁盘IO操作?

记录所有磁盘操作的实现方法

要捕获包括自动刷盘在内的所有磁盘操作,仅重写顶层函数别名不够,得从模块劫持和文件对象包装入手,以下是可行方案:

1. 劫持os模块与包装文件对象

通过替换sys.modules['os']中的目标函数,同时包装open()返回的文件对象,拦截flush、close等涉及自动刷盘的操作:

import os
import sys
from functools import wraps

# 记录操作的日志函数
def log_operation(func_name, *args, **kwargs):
    print(f"磁盘操作: {func_name} | 参数: {args} | 关键字参数: {kwargs}")

# 包装os模块的函数
def wrap_os_func(func):
    @wraps(func)
    def wrapped(*args, **kwargs):
        log_operation(func.__name__, *args, **kwargs)
        return func(*args, **kwargs)
    return wrapped

# 替换os模块中的关键磁盘操作函数
for func_name in ['remove', 'rename', 'mkdir', 'rmdir', 'unlink']:
    if hasattr(os, func_name):
        setattr(os, func_name, wrap_os_func(getattr(os, func_name)))

# 包装文件对象的方法,捕获刷盘相关操作
class LoggedFile:
    def __init__(self, file_obj):
        self.file_obj = file_obj
    
    def __getattr__(self, name):
        attr = getattr(self.file_obj, name)
        # 拦截触发刷盘或写入的方法
        if name in ['flush', 'close', 'write', 'writelines']:
            @wraps(attr)
            def wrapped(*args, **kwargs):
                log_operation(f"file.{name}", *args, **kwargs)
                return attr(*args, **kwargs)
            return wrapped
        return attr

# 替换内置open函数,记录打开操作并返回包装后的文件对象
original_open = open
def logged_open(*args, **kwargs):
    log_operation("open", *args, **kwargs)
    return LoggedFile(original_open(*args, **kwargs))

open = logged_open

这样不仅能捕获os.remove()这类显式操作,还能拦截文件对象的flush()(主动刷盘)、close()(隐式触发刷盘)以及写入类操作。

2. 使用sys.settrace全局跟踪函数调用

通过设置全局跟踪函数,自动捕获所有模块中与磁盘操作相关的函数调用,无需逐个包装:

import sys

def trace_calls(frame, event, arg):
    if event != 'call':
        return trace_calls
    func_name = frame.f_code.co_name
    module_name = frame.f_globals.get('__name__', '')
    # 过滤需要跟踪的磁盘操作相关函数
    target_funcs = {'remove', 'rename', 'mkdir', 'rmdir', 'flush', 'close', 'open'}
    if (module_name == 'os' and func_name in target_funcs) or func_name == 'open':
        args = frame.f_locals
        print(f"跟踪到磁盘操作: {module_name}.{func_name} | 参数: {args}")
    return trace_calls

sys.settrace(trace_calls)

这种方式覆盖范围广,但会产生大量非目标操作的日志,需要根据实际需求调整过滤规则。

补充说明

以上Python层面的方法仅能捕获代码触发的磁盘操作,若涉及C扩展模块、操作系统内核隐式触发的磁盘写入(如系统缓存刷盘),则需要借助操作系统原生工具(如Linux的inotify、Windows的FileSystemWatcher)来实现全量跟踪。

内容的提问来源于stack exchange,提问作者Omar Abdul'Azeez

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.23 19:35:22