解析固定宽度TXT文件时处理非ASCII字符的问题求助

我有多个GB级的固定宽度表格型TXT文件，列宽由表头下方的短横线数量确定。编写了逐行读取并转换为XML的脚本，但遇到以下问题：

• 多数内容是UTF-8编码，但存在少量非UTF-8内容，解码会报错，目前只能处理字节字符串；
• 列宽基于解码后的字符长度计算，但UTF-8中非ASCII字符占多字节，导致字节长度超过列宽，破坏后续解析（比如字符串´Zürich, Albisgütli´解码后字符长度18，UTF-8字节长度20，超出列宽19）；
• 尝试过先解码再解析（但编码错误无法解决）、枚举非ASCII字符调整解析（无法穷举且效率极低），均未解决问题；
• 当前使用基于struct.unpack_from的解析代码，近期发现部分内容是ANSI/Windows-1252编码（如b'Z\xfcrich'），ftfy库无法处理这种情况，不想用大量try-except处理编码问题，求可行方案。

当前代码 #

def convert(infile, outfile):
    secondline = infile.readline() # 短横线所在行
    maxlen = len(secondline)
    fieldwidths = get_widths(secondline) # 统计短横线数量得到列宽

    # 代码来自Stack Overflow
    fmtstring = ' '.join('{}{}'.format(abs(fw), 'x' if fw < 0 else 's')
                        for fw in fieldwidths)
    fieldstruct = struct.Struct(fmtstring)
    parse = fieldstruct.unpack_from
    
    c = 0
    
    outfile.write(b"<?xml version='1.0' encoding='UTF-8'?>\n")
    
    namespace = f'xmlns="http://www.bar.admin.ch/xmlns/siard/2/table.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.bar.admin.ch/xmlns/siard/2/table.xsd {table_w_num}.xsd" version="2.1"'.encode()
    
    outfile.write(b'<table ' + namespace + b'>\n')
    for line in infile:
        diff = maxlen - len(line)
        padded_line = bytearray()
        padded_line += line
        for _ in range(diff):
            padded_line += b' '

        data = [elem.strip() for elem in parse(padded_line)]
        data = parse(padded_line)
        
        if b"Albis" in line:
            print(line)
            print(data)
        row = b''
        for elem, n in zip(data, range(1, len(data)+1)):
            # 时间戳修正
            elem = re.sub(b"(\d{4}\-\d{2}\-\d{2}) (\d{2}:\d{2}:\d{2}(\.\d+)?\S*?)", b"\g<1>T\g<2>Z", elem)
            if elem == b'' or elem == b'NULL':
                pass
            else:
                row = b'%s<c%s>%s</c%s>' % (row, str(n).encode(), xml_escape(elem), str(n).encode())
        row = b"<row>%s</row>" % (row)
        outfile.write(b''.join([row, b'\n']))
        
        c += 1
        if c % infostep == 0:
            timestamp = int(time.time() - start_time)
            print(f"已处理{c}行，耗时：{str(timestamp)}秒")
    
    outfile.write(b'</table>')

可行方案建议 #

1. 先统一编码：解决混合编码问题 #

先批量检测文件编码，将所有内容转为UTF-8，避免字节级处理的混乱：

• 用cchardet（比chardet更快）检测每行或整个文件的编码；
• 对Windows-1252编码的内容直接转码为UTF-8，解码错误用errors='backslashreplace'保留原始字节信息，避免丢失数据；
• 转码完成后，全程用字符串处理列宽切割，从根源解决字节长度和字符长度不匹配的问题。

示例代码片段：

import cchardet

def detect_and_fix_encoding(line_bytes):
    detect_result = cchardet.detect(line_bytes)
    encoding = detect_result['encoding'] or 'utf-8'
    # 优先处理Windows-1252
    if encoding.lower() in ['windows-1252', 'ansi']:
        return line_bytes.decode('windows-1252').encode('utf-8')
    # 处理其他编码，容错转码
    try:
        return line_bytes.decode(encoding).encode('utf-8')
    except UnicodeDecodeError:
        return line_bytes.decode(encoding, errors='backslashreplace').encode('utf-8')

2. 替换struct.unpack_from：基于字符长度切割列 #

因为列宽是解码后的字符数，所以必须在字符串层面切割：

• 预计算每列的字符起始和结束索引：根据fieldwidths（字符数），生成索引区间，比如[0,19], [19,38], ...；
• 每行先解码为字符串，按索引切片，再处理空白；
• 切割完成后再转回字节生成XML，或者直接用字符串拼接XML（注意转义）。

示例代码片段：

# 预计算字符索引区间
def get_char_ranges(fieldwidths):
    ranges = []
    start = 0
    for width in fieldwidths:
        if width > 0:
            ranges.append((start, start + width))
            start += width
        else:
            start += abs(width)
    return ranges

# 在convert函数中替换解析逻辑
char_ranges = get_char_ranges(fieldwidths)
# 计算表头行的字符长度作为补全基准
maxlen_char = len(secondline.decode('utf-8').strip('\n'))

for line in infile:
    # 先修复编码
    fixed_line_bytes = detect_and_fix_encoding(line.strip(b'\n'))
    fixed_line = fixed_line_bytes.decode('utf-8')
    # 补全字符到指定长度
    padded_line = fixed_line.ljust(maxlen_char)
    # 切割列
    data = []
    for start, end in char_ranges:
        field = padded_line[start:end].strip()
        data.append(field.encode('utf-8') if field else b'')

3. 大文件处理优化 #

GB级文件逐行处理没问题，但可以优化内存和速度：

• 用生成器逐行读取，避免一次性加载文件；
• 预计算索引区间后，直接切片比循环更高效；
• XML生成可以用字符串拼接代替字节拼接，减少编码转换开销（最后统一编码为UTF-8写入）。

4. 避免大量try-except：编码处理前置 #

把编码检测和修复逻辑封装成单独函数，在解析前完成所有编码转换，后续流程只处理UTF-8字符串或字节，不用在解析过程中处理编码错误，彻底避免try-except嵌套。

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