非常规CSV文件解析需求：实现多文件结构化查询输出

针对你提到的这种以分号分隔、首行是#注释、数据行包含固定字段+不定数量k=v键值对的文件，直接用read_csv无法处理，下面提供两种可行的实现方式：

方式一：用SQL引擎（以DuckDB为例） #

通过分步处理，先读取原始行，再拆分、转结构：

1. 读取原始数据并过滤注释行
   先按行读取所有文件，跳过首行的注释，同时保留文件名作为Group列的备选值：

   WITH raw_rows AS (
     SELECT
       filename AS Group,
       row_text
     FROM read_csv(
       '*.txt',
       sep='\n',
       header=False,
       columns={'row_text': 'VARCHAR'},
       filename=True
     )
     WHERE row_text NOT LIKE '#%'
   )
   

2. 拆分固定字段与k=v键值对
   假设数据行的前两个字段是固定文本和时间戳，剩下的是k=v格式，拆分后提取键值对数组：

   , split_data AS (
     SELECT
       Group,
       split_part(row_text, ';', 1) AS fixed_text,
       split_part(row_text, ';', 2) AS timestamp,
       -- 过滤出k=v格式的部分
       list_filter(string_split(row_text, ';'), x -> x LIKE '%=%') AS kv_list
     FROM raw_rows
   )
   

3. 将键值对转为结构化列
   展开键值对数组，再通过PIVOT转成列，缺失的键会自动填充null：

   , kv_flattened AS (
     SELECT
       Group,
       fixed_text,
       timestamp,
       split_part(kv_item, '=', 1) AS key,
       split_part(kv_item, '=', 2) AS value
     FROM split_data, unnest(kv_list) AS kv_item
   )
   SELECT *
   FROM kv_flattened
   PIVOT (
     MAX(value) FOR key IN ('key1', 'key2', 'key3') -- 替换为实际存在的键，或用动态SQL生成
   )
   

4. 动态获取所有键（适配未知键集合）
   如果不确定所有键的名称，可以先查询出所有唯一键，再生成动态PIVOT语句：

   -- 先查询所有唯一键
   SELECT DISTINCT split_part(kv_item, '=', 1) AS key
   FROM raw_rows, unnest(string_split(row_text, ';')) AS kv_item
   WHERE kv_item LIKE '%=%';
   
   -- 将上述结果拼接成PIVOT的IN子句，执行动态SQL
   

方式二：用Python+Pandas手动解析 #

如果更习惯用脚本处理，可以写自定义解析逻辑：

import pandas as pd
import glob

def parse_single_file(file_path):
    records = []
    with open(file_path, 'r', encoding='utf-8') as f:
        # 读取首行注释作为Group，没有的话用文件名
        first_line = f.readline().strip()
        group_name = first_line.lstrip('#') if first_line.startswith('#') else file_path.split('/')[-1]
        
        for line in f:
            line = line.strip()
            if not line:
                continue
            parts = line.split(';')
            # 初始化行记录，填充Group和固定字段
            record = {
                'Group': group_name,
                'fixed_text': parts[0],
                'timestamp': parts[1]
            }
            # 处理剩余的k=v键值对
            for part in parts[2:]:
                if '=' in part:
                    key, val = part.split('=', 1)
                    record[key] = val
            records.append(record)
    return records

# 批量处理所有txt文件
all_records = []
for file in glob.glob('*.txt'):
    all_records.extend(parse_single_file(file))

# 转为DataFrame，缺失的键自动填充null
result_df = pd.DataFrame(all_records)
print(result_df)

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