如何在Python 3中处理fixed-width files并转存为CSV？

Hey there! 完全不需要先把每个链接的内容保存成TXT文件再解析，直接流式读取并处理是更高效的方案——既能节省磁盘空间，又能减少IO操作的耗时。下面给你两种靠谱的实现思路，适配你的需求：

方案一：用Pandas快速处理固定宽度文件 #

Pandas的read_fwf函数天生就是为固定宽度格式设计的，而且支持直接从URL读取，非常适合你的场景。因为文件里包含两种行类型（01开头的企业信息行、02开头的合伙人信息行），我们可以先拆分再处理，最后合并关联数据：

步骤与代码示例 #

import pandas as pd
import requests

# 你的地区-链接字典（示例）
dict_val = {
    "Acre": "http://idg.receita.fazenda.gov.br/orientacao/tributaria/cadastros/cadastro-nacional-de-pessoas-juridicas-cnpj/consultas/download/F.K03200UF.D71214AC",
    "Espírito Santo": "http://idg.receita.fazenda.gov.br/orientacao/tributaria/cadastros/cadastro-nacional-de-pessoas-juridicas-cnpj/consultas/download/F.K03200UF.D71214ES"
}

# 定义两种行的字段宽度和列名
# 01行：行标识(2) + 企业代码(14) + 企业名称(150)
fwf_spec_01 = [2, 14, 150]
cols_01 = ["行类型", "企业代码", "企业名称"]
# 02行：行标识(2) + 企业代码(14) + 合伙人识别码1(1) + 预留字段(14) + 合伙人识别码2(2) + 合伙人姓名(150)
fwf_spec_02 = [2, 14, 1, 14, 2, 150]
cols_02 = ["行类型", "企业代码", "合伙人识别码1", "预留字段", "合伙人识别码2", "合伙人姓名"]

for region, url in dict_val.items():
    # 流式获取文件内容，避免加载大文件到内存
    response = requests.get(url, stream=True)
    response.encoding = "latin-1"  # 巴西官方文件常用编码，务必设置
    
    # 拆分两种行类型
    lines_01 = []
    lines_02 = []
    for line in response.iter_lines(decode_unicode=True):
        if line.startswith("01"):
            lines_01.append(line)
        elif line.startswith("02"):
            lines_02.append(line)
    
    # 解析为DataFrame
    df_empresas = pd.read_fwf(pd.io.common.StringIO("\n".join(lines_01)), widths=fwf_spec_01, names=cols_01)
    df_socios = pd.read_fwf(pd.io.common.StringIO("\n".join(lines_02)), widths=fwf_spec_02, names=cols_02)
    
    # 合并企业与合伙人数据（按企业代码关联）
    df_merged = pd.merge(df_empresas, df_socios, on="企业代码")
    
    # 保存为CSV，按地区命名
    csv_path = f"empresas_socios_{region.replace(' ', '_')}.csv"
    df_merged.to_csv(csv_path, index=False, encoding="utf-8")
    print(f"✅ 已完成{region}地区的文件处理，保存为：{csv_path}")

方案二：手动解析（轻量无依赖） #

如果不想依赖Pandas，手动按字符位置截取内容也是可行的，代码更轻量，适合对内存占用要求高的场景：

步骤与代码示例 #

import requests
import csv

# 你的地区-链接字典（示例）
dict_val = {
    "Acre": "http://idg.receita.fazenda.gov.br/orientacao/tributaria/cadastros/cadastro-nacional-de-pessoas-juridicas-cnpj/consultas/download/F.K03200UF.D71214AC",
    "Espírito Santo": "http://idg.receita.fazenda.gov.br/orientacao/tributaria/cadastros/cadastro-nacional-de-pessoas-juridicas-cnpj/consultas/download/F.K03200UF.D71214ES"
}

for region, url in dict_val.items():
    response = requests.get(url, stream=True)
    response.encoding = "latin-1"
    
    # 准备CSV文件写入
    csv_filename = f"empresas_socios_{region.replace(' ', '_')}.csv"
    with open(csv_filename, "w", encoding="utf-8", newline="") as csvfile:
        fieldnames = ["企业代码", "企业名称", "合伙人识别码1", "合伙人识别码2", "合伙人姓名"]
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()
        
        current_enterprise = None  # 缓存当前企业信息，用于关联合伙人
        for line in response.iter_lines(decode_unicode=True):
            if line.startswith("01"):
                # 解析企业信息行
                enterprise_code = line[2:16].strip()
                enterprise_name = line[16:166].strip()
                current_enterprise = (enterprise_code, enterprise_name)
            elif line.startswith("02") and current_enterprise:
                # 解析合伙人信息行，并关联当前企业
                partner_code1 = line[16].strip()
                partner_code2 = line[31:33].strip()
                partner_name = line[33:183].strip()
                # 写入CSV行
                writer.writerow({
                    "企业代码": current_enterprise[0],
                    "企业名称": current_enterprise[1],
                    "合伙人识别码1": partner_code1,
                    "合伙人识别码2": partner_code2,
                    "合伙人姓名": partner_name
                })
    print(f"✅ {region}地区的CSV文件已生成：{csv_filename}")

关键注意事项 #

• 编码问题：这类巴西官方文件几乎都用latin-1编码，一定要设置正确，否则会出现乱码。
• 流式处理：使用stream=True和iter_lines()可以避免将大文件一次性加载到内存，处理大体积文件时更稳定。
• 数据关联：02行依赖01行的企业代码，所以处理时要缓存当前企业的信息，确保合伙人数据能正确关联到对应的企业。

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