Python如何读取持续更新文本文件新增行及指定重叠行

方案核心优势 #

对比全量读入DataFrame、从文件末尾倒序扫行匹配锚点这两种方案，基于文件偏移量记录的增量读取方案有几个不可替代的优点：

• 无全量IO开销：每次仅读取上次读取位置之后的新增字节，文件体积再大也不会影响读取效率
• 无匹配漏读风险：不需要拿上一轮最后一行做锚点比对，天然规避连续相似行导致的锚点匹配错误、漏读问题
• 自动适配写入波动：能自动识别写入过程中的半行残缺内容，不会读到未写完的无效数据，适配网络波动导致的写入时长不固定场景
• 上下文留存逻辑简单：用固定长度队列自动留存需要的N行历史内容，不需要额外计算倒读行数

具体实现逻辑 #

• 初始化阶段：首次运行时打开目标文件，直接将读取指针移动到文件当前末尾，通过tell()记录当前偏移量；同时初始化一个最大长度为N（需要留存的历史上下文行数）的双端队列作为历史行缓存，队列满了之后会自动淘汰最早存入的行，不需要手动维护。
• 轮询阶段：每次轮询先检查文件当前大小，如果当前大小小于上次记录的偏移量，说明文件被截断/重建（比如日志切分、文件被覆盖重写），直接将偏移量重置为0，清空历史缓存，避免漏读。
• 读取阶段：将文件指针移动到上次记录的偏移量位置，读取从该位置到当前文件末尾的所有内容；判断读取到的内容末尾是否为换行符，如果不是说明最后一行是正在写入的半行，直接丢弃这部分内容，将指针回退到上一个完整换行符的位置，等下一轮写入完成后再读取。
• 处理阶段：将读取到的完整新增行，和历史缓存中留存的N行上下文一起传入分析函数——注意区分两类内容：历史上下文仅做关联参考，不需要重复走分析流程，仅对新增行执行核心分析逻辑即可。
• 状态更新：分析完成后，将本次读到的所有新增行存入历史缓存，队列会自动裁剪到N行长度；同时更新文件偏移量为本次读完的完整行末尾位置，进入下一轮轮询。

可直接复用的代码实现 #

import os
import time
from collections import deque

def incremental_file_reader(
    file_path: str,
    keep_context_lines: int,
    poll_interval: int = 10,
    analysis_callback = None
):
    """
    增量读取持续追加的文本文件
    :param file_path: 目标文件路径
    :param keep_context_lines: 需要留存的上一轮最后N行历史上下文数量
    :param poll_interval: 轮询检查间隔，单位秒
    :param analysis_callback: 后续分析函数，接收两个入参：context(历史上下文行列表), new_lines(本次新增行列表)
    """
    # 初始化状态变量
    last_read_offset = 0
    context_cache = deque(maxlen=keep_context_lines)

    # 首次运行直接定位到文件末尾，不处理历史全量；如果需要首次启动就读取全量内容，可注释掉下方初始化块
    if os.path.exists(file_path):
        with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
            f.seek(0, os.SEEK_END)
            last_read_offset = f.tell()

    while True:
        try:
            current_file_size = os.path.getsize(file_path)
            # 处理文件被截断/重建的场景
            if current_file_size < last_read_offset:
                last_read_offset = 0
                context_cache.clear()

            read_content = ""
            current_offset = last_read_offset
            with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
                f.seek(last_read_offset)
                read_content = f.read()
                current_offset = f.tell()

            # 没有新内容直接休眠等待
            if not read_content:
                time.sleep(poll_interval)
                continue

            # 拆分内容，过滤未写完的半行
            lines = read_content.splitlines()
            # 内容末尾不是换行符，说明最后一行不完整，剔除并回退偏移量
            if not read_content.endswith(('\n', '\r\n')):
                incomplete_line_bytes = len(lines[-1].encode('utf-8'))
                lines = lines[:-1]
                current_offset -= (incomplete_line_bytes + len(os.linesep.encode('utf-8')))

            # 有完整新增行就执行分析
            if lines and analysis_callback:
                analysis_callback(
                    context=list(context_cache),
                    new_lines=lines
                )
                # 更新上下文缓存
                context_cache.extend(lines)

            # 更新读取偏移量
            last_read_offset = current_offset

        except PermissionError:
            # 文件被写入锁定时短暂重试
            time.sleep(1)
            continue
        except Exception as e:
            # 可根据自身需求扩展异常处理逻辑
            print(f"读取文件异常: {str(e)}")

        time.sleep(poll_interval)


# 调用示例
def your_analysis_logic(context, new_lines):
    # 在这里写你的分析代码，context是留存的N行历史，new_lines是本次新增内容
    print(f"本次读取到{len(new_lines)}行新内容，关联历史上下文{len(context)}行")

if __name__ == "__main__":
    incremental_file_reader(
        file_path="your_target_file.log",
        keep_context_lines=30,  # 按自己需要的上下文行数设置
        poll_interval=10,
        analysis_callback=your_analysis_logic
    )

适配优化建议 #

• 轮询间隔不用严格设置为60秒，设为10-15秒即可，因为每次仅读新增内容，IO开销极低，还能适配网络波动导致的写入时长不稳定问题，不用等整分钟才触发读取。
• 如果分析函数耗时较长，不要在读取线程里同步执行分析逻辑，可以把新增行和上下文打包扔进任务队列，交给独立的工作线程/进程处理，避免阻塞读取流程导致内容积压。
• 如果文件存在多编码混存的情况，open时保持errors='replace'参数，避免个别乱码行导致整个读取流程崩溃。

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