Python逐行逐字段解析固定格式TXT工资单并导出CSV的实现问题

之前的代码问题在于仅读取了文件首行，没有做分块识别、逐行遍历、多字段定位，下面是可直接运行的实现：

核心逻辑 #

• 一次性读取文件所有行，过滤无内容的空行，避免排版空行干扰位置判断
• 以「包含公司名+带/的结算周期」作为单张工资单的起始标记，循环识别所有工资单块直到文件末尾
• 按固定行偏移+字符切片提取对应字段，自动清洗多余空白、冗余符号
• 所有记录提取完成后按标准CSV格式输出，采用适配Excel直接打开的编码规则

完整代码 #

import csv
import re

# 字段字符位置切片，根据提供的样例校准，排版偏移时直接修改此处数值即可
SLICE_CONF = {
    "company": (0, 32),
    "competence": (32, 48),
    "id_employee": (6, 12),
    "employee_name": (14, 45),
    "job_code": (9, 19),
    "job_name": (21, 38),
    "total_earn": (45, 58),
    "total_deduct": (64, 76),
    "net_salary": (60, 74)
}

def clean_text(raw_str: str) -> str:
    """清洗多余空白、首尾冗余符号"""
    return re.sub(r"\s+", " ", raw_str.strip().strip("-")).strip()

def parse_payslip(input_path: str, output_csv_path: str):
    # 读取所有非空行
    with open(input_path, "r", encoding="utf-8") as f:
        valid_lines = []
        for line in f.readlines():
            line_content = line.rstrip("\n")
            if line_content.strip():
                valid_lines.append(line_content)
    
    payslip_records = []
    cursor = 0
    total_line_num = len(valid_lines)

    while cursor < total_line_num:
        current_line = valid_lines[cursor]
        # 匹配工资单头部起始行
        if "COMPANY" in current_line and "/" in current_line:
            single_record = {}
            # 提取第一行头部字段
            single_record["company"] = clean_text(current_line[SLICE_CONF["company"][0]:SLICE_CONF["company"][1]])
            comp_raw = current_line[SLICE_CONF["competence"][0]:SLICE_CONF["competence"][1]]
            single_record["competence"] = clean_text(comp_raw.split(" ")[0]) # 剔除工种后缀

            # 提取第二行：员工编号、姓名
            cursor += 1
            emp_line = valid_lines[cursor]
            single_record["id_employee"] = clean_text(emp_line[SLICE_CONF["id_employee"][0]:SLICE_CONF["id_employee"][1]])
            single_record["employee_name"] = clean_text(emp_line[SLICE_CONF["employee_name"][0]:SLICE_CONF["employee_name"][1]])

            # 提取第三行：岗位信息
            cursor += 1
            job_line = valid_lines[cursor]
            single_record["job_code"] = clean_text(job_line[SLICE_CONF["job_code"][0]:SLICE_CONF["job_code"][1]])
            single_record["job_name"] = clean_text(job_line[SLICE_CONF["job_name"][0]:SLICE_CONF["job_name"][1]])

            # 跳过薪资明细行，定位到汇总行
            cursor += 1
            while cursor < total_line_num:
                check_line = valid_lines[cursor]
                # 匹配同时包含总加项、总减项的汇总行
                if "+" in check_line and "-" in check_line and "." in check_line:
                    single_record["total_earnings"] = clean_text(check_line[SLICE_CONF["total_earn"][0]:SLICE_CONF["total_earn"][1]].replace("+",""))
                    single_record["total_deductions"] = clean_text(check_line[SLICE_CONF["total_deduct"][0]:SLICE_CONF["total_deduct"][1]].replace("-",""))
                    # 向下偏移2行取实发工资
                    cursor += 2
                    net_line = valid_lines[cursor]
                    single_record["net_salary"] = clean_text(net_line[SLICE_CONF["net_salary"][0]:SLICE_CONF["net_salary"][1]])
                    payslip_records.append(single_record)
                    break
                cursor += 1
        cursor += 1
    
    # 写入CSV文件
    csv_headers = ["competence","company","id_employee","employee_name","job_code","job_name","total_earnings","total_deductions","net_salary"]
    with open(output_csv_path, "w", encoding="utf-8-sig", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=csv_headers, delimiter=";")
        writer.writeheader()
        writer.writerows(payslip_records)
    print(f"解析完成，共处理{len(payslip_records)}张工资单，结果已写入{output_csv_path}")

if __name__ == "__main__":
    parse_payslip(r"I:\input\test.txt", r"I:\input\payslip_output.csv")

使用说明 #

• 输出CSV采用utf-8-sig编码，分号作为分隔符，直接双击用Excel打开不会乱码，适配巴西地区CSV默认格式
• 如果实际TXT存在排版偏移，只需要微调SLICE_CONF里的切片起止数字即可，不需要修改核心遍历逻辑
• 如果需要提取单条薪资明细（如正常工时工资、加班费、社保扣除等分项），只需在明细行遍历段增加对应正则匹配规则即可，现有框架已经预留遍历位置

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