Python迭代处理CSV数据直到满足指定长度条件的实现咨询

import os
import pandas as pd
from typing import Final
# 导入process.py中你自己实现的数据处理函数，替换成实际的函数名
from process import data_process_func

def iteration():
    # 获取项目根目录（read.py所在的目录就是项目根目录）
    PROJECT_ROOT: Final = os.path.dirname(os.path.abspath(__file__))
    new_data_path = os.path.join(PROJECT_ROOT, "my_new_data", "new_data.csv")
    original_data_path = os.path.join(PROJECT_ROOT, "my_original_data", "original_data.csv")

    # 首次合并去重，获取基准长度
    df_new = pd.read_csv(new_data_path)
    df_original = pd.read_csv(original_data_path)
    all_data = pd.concat([df_new, df_original], ignore_index=True)
    all_data.drop_duplicates(subset="text", keep=False, inplace=True)
    len_first_comb: Final = len(all_data)

    while True:
        # 把当前合并去重后的数据传给process处理
        processed_data = data_process_func(all_data)
        # 处理结果替换new_data.csv
        processed_data.to_csv(new_data_path, index=False, encoding="utf-8")

        # 重新读取文件合并去重，计算当前长度
        df_new = pd.read_csv(new_data_path)
        all_data = pd.concat([df_new, df_original], ignore_index=True)
        all_data.drop_duplicates(subset="text", keep=False, inplace=True)
        len_current = len(all_data)

        # 长度不再大于首次长度时停止迭代
        if len_current <= len_first_comb:
            break

    print('iteration is finished.')
    return all_data

关键修改说明 #

• 补充了路径计算逻辑：通过__file__动态获取项目根目录，避免相对路径读取文件失败的问题
• 修正了迭代逻辑：每次循环先处理当前数据、更新new_data.csv，再重新合并计算长度，直到新长度不大于首次基准长度就终止
• 补充了process函数调用、结果写回csv的完整流程，符合跨模块调用和文件替换需求
• 移除了原代码中无法进入的while循环逻辑，原代码初始长度和基准长度相等，永远不会触发循环
• 可选优化：可以在while循环中加入最大迭代次数限制，避免特殊情况出现死循环，比如设置max_iter = 100，循环次数超过阈值就主动终止

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