如何批量处理指定格式txt文件，为每个文件生成两个DataFrame？

核心实现思路 #

• 新增块标记变量，区分当前读取的是[Level1]还是[Level2]段落
• Level1段落逐行解析键值对，统一清理首尾空格、引号后存入字典，最终转为单行DataFrame
• Level2段落逐行拆分数值对，存入列表后转为多行列DataFrame
• 生成的两个DataFrame可直接通过pandas内置方法推送至数据库

完整可运行代码 #

import os
import re
import pandas as pd

energy_root= r'c:\data\Desktop\Studio\Energyfiles'

# 生成所有txt文件路径列表
def read_txt_file(path):
    list_file_path = []
    for root, dirs, files in os.walk(path):
        for file in files:
            if file.endswith('.txt'):
                file_path = os.path.join(root, file)
                list_file_path.append(file_path)
    return list_file_path     

def create_df():
    all_file = read_txt_file(energy_root)
    for file_path in all_file:
        file_name = os.path.basename(file_path)
        # 提取文件名中的时间字段，根据实际需求可自行加入生成的DataFrame中
        datetime = re.findall(r'_(\d{8}_\d{6})\.', file_name)[0]

        # 读取文件内容，提前清理空行和首尾空白
        with open(file_path, 'r', encoding='utf-8') as f:
            lines = [line.strip() for line in f.readlines() if line.strip()]
        
        level1_dict = {}
        level2_list = []
        current_block = None

        for line in lines:
            # 切换块标记
            if line == '[Level1]':
                current_block = 'level1'
                continue
            if line == '[Level2]':
                current_block = 'level2'
                continue
            
            # 处理Level1段落键值对
            if current_block == 'level1':
                k, v = line.split('=', 1)
                k = k.strip()
                v = v.strip().strip('"')
                # 替换字段名匹配需求表头
                if k == 'Energy level':
                    k = 'Energy'
                level1_dict[k] = v
            
            # 处理Level2段落数值对
            elif current_block == 'level2':
                speed, energy_level = line.split()
                level2_list.append({
                    'Speed': float(speed),
                    'Energylevel': float(energy_level)
                })
        
        # 生成符合要求的两个DataFrame
        df_level1 = pd.DataFrame([level1_dict])
        df_level2 = pd.DataFrame(level2_list)

        # 此处可加入数据推送数据库的逻辑
        print('=========当前文件处理完成=========')
        print('df_level1：\n', df_level1)
        print('df_level2：\n', df_level2)

if __name__ == '__main__':
    create_df()

数据推送数据库示例 #

以MySQL为例，使用sqlalchemy实现一键推送，其他类型数据库只需调整连接字符串即可：

from sqlalchemy import create_engine
# 替换为实际的数据库连接信息
engine = create_engine('mysql+pymysql://用户名:密码@数据库地址:端口/库名?charset=utf8')
# if_exists参数可选：replace覆盖/append追加/fail存在就报错，按实际需求选择
df_level1.to_sql(name='level1对应的表名', con=engine, if_exists='append', index=False)
df_level2.to_sql(name='level2对应的表名', con=engine, if_exists='append', index=False)

内容的提问来源于stack exchange，提问作者Al-Andalus