场景与现象
一个日志导入程序按固定字节数读取 UTF-8 文件，每块调用 decode("utf-8")。英文测试一直正常，上线处理中文时却偶发 UnicodeDecodeError；改成 errors="ignore" 后程序不再报错，客户名称里却少了字。
本文处理的是同一条连续 UTF-8 字节流的分块读取，适用于文件导入或自己维护的字节流适配器。传输块边界不保证与字符边界重合。若上游实际使用 GB18030，必须先按协议确定编码，增量解码不能修复编码选择错误。
先复现，不急着放宽错误策略
raw = "甲乙".encode("utf-8")
chunks = [raw[:4], raw[4:]]
print("每块字节数：", [len(chunk) for chunk in chunks])
try:
print("".join(chunk.decode("utf-8") for chunk in chunks))
except UnicodeDecodeError as exc:
print("严格解码失败：", exc.reason)
print("忽略错误后的结果：", "".join(
chunk.decode("utf-8", errors="ignore") for chunk in chunks
))
输出中块长度为 [4, 2]，严格解码失败，忽略错误只剩“甲”。“乙”的三个字节被拆成一字节和两字节，两次独立解码都无法还原它。读取块大小改成 4096 只能改变故障概率；UTF-8 字符长度可变，不能靠固定块长保证边界。
排查时检查完整原始字节在严格模式下是否可解码，再比较不同分块方式。只在测试夹具中输出字节内容；生产日志记录 encoding、chunk_index、bytes_read 和错误原因即可，不记录原始敏感文本。异常位置通常相对于本次解码输入，增量模式还可能包含缓存字节，不要未经换算就当成文件绝对偏移。
修复：每条流保留一个解码器
把下面代码保存为 stream_decode.py。仅依赖标准库，适用于 Python 3.10 及以上。
import codecs
from collections.abc import Iterable, Iterator
def iter_utf8_text(chunks: Iterable[bytes]) -> Iterator[str]:
"""在同一条字节流内保留跨块字符，并在流结束时检查残留字节。"""
decoder = codecs.getincrementaldecoder("utf-8")(errors="strict")
for chunk in chunks:
if not isinstance(chunk, bytes):
raise TypeError("字节流的每个分块必须是 bytes")
text = decoder.decode(chunk, final=False)
if text:
yield text
tail = decoder.decode(b"", final=True)
if tail:
yield tail
解码器在循环外创建，不完整字符会留在其内部，后续字节到达再继续解码。空块合法，不代表这条流结束；只有迭代器正常耗尽，才执行 final=True。结束时若还有不完整字符，严格模式抛出异常，防止把截断误认作成功。
这些行为依据 Python 官方 codecs 文档。标准库也提供 codecs.iterdecode(chunks, "utf-8", errors="strict")；这里显式写出生命周期，便于说明结束检查和输入合同。
普通文件无需自行维护分块解码，通常直接使用 open(path, encoding="utf-8", errors="strict") 的文本接口。只有调用方已经拿到字节块时，才需要上述适配器。解码器不能跨文件共享，也不能让多个线程或任务交错喂入不同流的数据。
回归测试：不能只测完整中文块
将以下内容保存为同目录的 test_stream_decode.py，执行 python -m unittest -v test_stream_decode。
import unittest
from stream_decode import iter_utf8_text
class StreamDecodeTests(unittest.TestCase):
"""检查块边界、输入合同和流结束的错误传播。"""
def test_every_split(self) -> None:
text = "A甲乙\U0001f680\r\n尾"
raw = text.encode("utf-8")
for split in range(len(raw) + 1):
with self.subTest(split=split):
self.assertEqual(
"".join(iter_utf8_text([raw[:split], raw[split:]])), text
)
def test_one_byte_chunks(self) -> None:
text = "中文\U0001f680"
chunks = (bytes([value]) for value in text.encode("utf-8"))
self.assertEqual("".join(iter_utf8_text(chunks)), text)
def test_empty_input(self) -> None:
self.assertEqual(list(iter_utf8_text([])), [])
self.assertEqual(list(iter_utf8_text([b"", b""])), [])
def test_truncated_end(self) -> None:
with self.assertRaises(UnicodeDecodeError):
list(iter_utf8_text(["甲".encode("utf-8")[:2]]))
def test_invalid_bytes(self) -> None:
with self.assertRaises(UnicodeDecodeError):
list(iter_utf8_text([b"ok", b"\xff"]))
def test_invalid_chunk_type(self) -> None:
with self.assertRaisesRegex(TypeError, "bytes"):
list(iter_utf8_text(["中文"]))
def test_streams_are_independent(self) -> None:
with self.assertRaises(UnicodeDecodeError):
list(iter_utf8_text([b"\xe7"]))
self.assertEqual("".join(iter_utf8_text([b"ok"])), "ok")
if __name__ == "__main__":
unittest.main()
第一项遍历每个双块切分位置，覆盖二、三、四字节字符以及换行。单字节块用来防止实现偶然依赖较大块长。最后一项验证失败流的状态不会污染下一次调用。
解码正确不等于导入原子提交
这个函数是惰性迭代器，前面合法文本可能已经产出，后面才遇到非法字节或末尾截断。如果消费方提前退出，也不会运行结束检查。必须完整消费且未报错，才能把整条流标记为解码成功。
因此，导入任务不要每收到一个文本片段就宣布成功。需要整批原子性时，先写入受限的暂存区，完整消费并校验业务记录后再提交；失败则清理暂存。不要为验证完整性把超大文件全部拼进内存。上游读取异常应向外传播，异常中断不能冒充正常 EOF。
解码器输出的片段也不保证是一行、一个 JSON 对象或一条业务记录。分行与协议解析应放在解码之后，并对记录长度设置独立限制。本文没有实现记录解析、事务暂存、压缩解包或断点续传；有 BOM 时也应由文件合同明确是否接受，不能无条件删除文本首字符。
发布前验证与落地检查
本文示例在 Python 3.14.6 中执行，七项 unittest 全部通过。将增量解码故意退回逐块独立解码后，跨块用例出现 10 个预期的解码执行错误；删除结束刷新后，两项测试因未抛出预期异常而失败。恢复正确实现后七项重新通过。编码、代码语法和围栏检查通过；本地未安装 Ruff，未执行其格式和静态检查，也未安装额外依赖。本地检查不能代替真实文件源、网络断流、解压链路和生产吞吐量验证。
上线前保留包含中文、四字节字符、尾部截断和非法字节的匿名夹具，对照原始字节合同验收。观察解码失败数和导入失败状态，避免用 ignore 或 replace 掩盖数据损坏。回滚可切回旧版本并暂停异常批次，保留受控原始文件供重放，不能把已经丢失字符的结果继续作为正确数据传播。
Discussion
评论