使用行计数和触发归档时遇io.UnsupportedOperation: not writable错误

核心问题梳理 #

1. 行计数逻辑错误：仅判断单个文件行数是否超75万，未累加已归档的总行数
2. 文件权限冲突：更新归档行数时，若文件以只读模式打开，会触发io.UnsupportedOperation: not writable错误
3. 新归档文件模式缺失：创建新归档时未指定写模式，导致后续写入操作失败
4. 提前触发归档创建：计数逻辑错误导致未达75万行就错误生成新归档

分步修复实现 #

1. 正确统计归档行数 #

用只读模式打开主归档文件统计行数，避免占用写权限引发冲突；若CSV含表头，需跳过表头行统计有效数据行。

2. 累加式行数判断 #

先获取当前归档总行数，遍历新CSV文件时累加待追加行数，当总和超过75万时，创建序号递增的新归档文件，并重置当前归档行数。

3. 规范文件打开模式 #

• 统计行数用r只读模式
• 追加数据到归档用a追加模式
• 创建新归档用w写入模式（或x模式避免覆盖已有文件）

4. 精准获取最新CSV文件 #

按文件修改时间倒序排序，筛选出目录下最新的6-9个CSV文件（排除主归档文件）

修正后的完整代码 #

import os
import glob

# 配置参数
ARCHIVE_DIR = "./"  # 归档文件所在目录
MAIN_ARCHIVE = "main_archive.csv"
MAX_ROWS_BEFORE_NEW_ARCHIVE = 750000
MAX_NEW_FILES = 9
MIN_NEW_FILES = 6

def get_file_mod_time(file_path):
    """获取文件修改时间戳"""
    return os.path.getmtime(file_path)

def count_csv_rows(file_path):
    """统计CSV文件行数（默认跳过表头，无表头可删除减1逻辑）"""
    row_count = 0
    with open(file_path, 'r', encoding='utf-8') as f:
        lines = f.readlines()
        if lines:
            row_count = len(lines) - 1  # 减去表头行
        else:
            row_count = 0
    return row_count

def get_latest_csv_files():
    """获取目录下最新的6-9个CSV文件（排除主归档）"""
    csv_files = glob.glob(os.path.join(ARCHIVE_DIR, "*.csv"))
    csv_files = [f for f in csv_files if os.path.basename(f) != MAIN_ARCHIVE]
    # 按修改时间倒序排序
    csv_files.sort(key=get_file_mod_time, reverse=True)
    # 取最新的6-9个
    return csv_files[:min(len(csv_files), MAX_NEW_FILES)] if len(csv_files) >= MIN_NEW_FILES else csv_files

def main():
    # 获取当前主归档的行数
    current_archive_rows = 0
    if os.path.exists(MAIN_ARCHIVE):
        current_archive_rows = count_csv_rows(MAIN_ARCHIVE)
    
    # 获取最新待追加的CSV文件
    latest_csvs = get_latest_csv_files()
    if not latest_csvs:
        print("无待追加的CSV文件")
        return
    
    output_file = MAIN_ARCHIVE
    for csv_file in latest_csvs:
        file_rows = count_csv_rows(csv_file)
        # 判断是否需要创建新归档
        if current_archive_rows + file_rows > MAX_ROWS_BEFORE_NEW_ARCHIVE:
            # 生成序号递增的新归档文件名
            archive_num = 1
            while True:
                new_archive = f"archive_{archive_num}.csv"
                if not os.path.exists(os.path.join(ARCHIVE_DIR, new_archive)):
                    break
                archive_num += 1
            output_file = new_archive
            current_archive_rows = 0
        
        # 追加文件内容到归档
        with open(csv_file, 'r', encoding='utf-8') as in_file:
            with open(output_file, 'a', encoding='utf-8') as out_file:
                # 新归档文件先写入表头
                if current_archive_rows == 0:
                    out_file.write(in_file.readline())
                # 写入剩余数据行
                for line in in_file:
                    out_file.write(line)
        # 更新当前归档行数
        current_archive_rows += file_rows
        print(f"已追加: {os.path.basename(csv_file)} -> {os.path.basename(output_file)}")

if __name__ == "__main__":
    main()

关键细节说明 #

• 若CSV无表头，需将count_csv_rows函数中的row_count = len(lines) - 1改为row_count = len(lines)
• 新归档命名逻辑会自动找到第一个未存在的序号文件，避免覆盖已有数据
• 新建归档时仅写入一次表头，避免重复追加表头导致数据格式错误

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