Python 中文消息字数未超限却被拒绝：UTF-8 字节预算的定位与修复

适用场景与故障现象

某通知适配器约定：content 字段编码为 UTF-8 后最多 20 字节。英文联调一直正常，切换中文告警后却收到长度校验失败。发送端已经执行 content[:20]，于是问题被误判为接收方限流。

这里的 20 是便于复现的协议约定，不代表任何第三方平台的真实限额。接入实际服务时，先查明限制对象究竟是字段字节数、字符数，还是完整 HTTP 请求体。

先确认测量单位

在 Python 3 中运行：

text = "数据库连接失败"
print(len(text))                  # 7 个码点
print(len(text.encode("utf-8")))  # 21 字节
print(text[:20] == text)          # True，没有截短

len(str) 计算 Unicode 码点数量，len(bytes) 计算字节数量。常见汉字在 UTF-8 中占 3 字节，部分表情占 4 字节，不能统一按“三倍字数”估算。组合字符和连接表情还可能由多个码点组成。

排查时按下面顺序取证：

对照协议确认编码、限制单位，以及是否包含标题、前缀和截断标记。
在发送边界记录 codepoint_count、content_bytes、limit_bytes 和 request_id，固定日志说明使用中文，不记录原始告警正文。
用相同语义的英文、中文、表情构造本地样本，观察字节数是否超过阈值。
如果字段没有超限，再核对实际序列化后的请求体长度和服务端错误码，避免把不同限制混在一起。

不要先修改网络超时或增加重试次数。确定性的长度拒绝不会因为重试而恢复。

修复：为标记预留空间，按码点累加字节

本文约定：原文能放下就原样返回；超限时保留最长的码点前缀，再添加标记；标记自身放不下则报错。预算必须是非负整数，不接受布尔值。对可能隐藏在截断部分的孤立代理码点也拒绝，避免截断掩盖非法输入。

将下面代码保存为 utf8_budget.py：

def fit_utf8(text: str, limit_bytes: int, marker: str = "…") -> str:
    """按 UTF-8 字节预算截断展示文本，保留完整码点并预留标记。"""
    if not isinstance(text, str) or not isinstance(marker, str):
        raise TypeError("正文和截断标记必须是字符串")
    if isinstance(limit_bytes, bool) or not isinstance(limit_bytes, int):
        raise TypeError("字节预算必须是整数")
    if limit_bytes < 0:
        raise ValueError("字节预算不得为负数")
    for value in (text, marker):
        if any(0xD800 <= ord(char) <= 0xDFFF for char in value):
            raise ValueError("输入包含不能编码为标准 UTF-8 的代理码点")

    if len(text.encode("utf-8")) <= limit_bytes:
        return text
    marker_bytes = len(marker.encode("utf-8"))
    if marker_bytes > limit_bytes:
        raise ValueError("字节预算不足以容纳截断标记")

    remaining_bytes = limit_bytes - marker_bytes
    prefix: list[str] = []
    for char in text:
        char_bytes = len(char.encode("utf-8"))
        if char_bytes > remaining_bytes:
            break
        prefix.append(char)
        remaining_bytes -= char_bytes
    return "".join(prefix) + marker

不直接切编码后的字节，是为了避免落在多字节序列中间；也不使用 errors="ignore"，以免把损坏数据和正常截断混为一谈。代码在第一个放不下的码点停止，不跳过它去拼接后文，因此输出始终是原文前缀加标记。

例如 fit_utf8("数据库连接失败", 20) 得到 数据库连接…，总计 18 字节。剩下的 2 字节不足以容纳下一个汉字，留空是正确结果。预算不是必须填满的配额。

该实现需要扫描完整输入，编码时也会创建中间字节串，时间及额外空间上界为 O(n)。服务端仍须在上游设置合理的输入长度上限；这不是用于无限日志流的流式截断器。

用回归测试覆盖中文、表情和失败路径

将下面内容保存为同目录下的 test_utf8_budget.py：

import unittest

from utf8_budget import fit_utf8


class Utf8BudgetTests(unittest.TestCase):
    """校验协议字节上限与截断边界。"""

    def test_exact_boundary(self):
        self.assertEqual(fit_utf8("中文", 6), "中文")

    def test_chinese_budget(self):
        result = fit_utf8("数据库连接失败", 20)
        self.assertEqual(result, "数据库连接…")
        self.assertLessEqual(len(result.encode("utf-8")), 20)

    def test_emoji_boundary(self):
        self.assertEqual(fit_utf8("A😀中文", 8), "A😀…")

    def test_ascii_marker(self):
        self.assertEqual(fit_utf8("abcdef", 5, "..."), "ab...")

    def test_empty_and_zero(self):
        self.assertEqual(fit_utf8("", 0), "")
        self.assertEqual(fit_utf8("中文", 0, ""), "")

    def test_marker_does_not_fit(self):
        with self.assertRaises(ValueError):
            fit_utf8("中文", 2)

    def test_invalid_budget(self):
        for budget in (True, 1.5, "20"):
            with self.subTest(budget=budget):
                with self.assertRaises(TypeError):
                    fit_utf8("内容", budget)
        with self.assertRaises(ValueError):
            fit_utf8("内容", -1)

    def test_invalid_text(self):
        with self.assertRaises(TypeError):
            fit_utf8(None, 20)
        with self.assertRaises(TypeError):
            fit_utf8("内容", 20, None)
        with self.assertRaises(ValueError):
            fit_utf8("前缀\ud800", 3)

    def test_prefix_and_budget_matrix(self):
        for text in ("abc", "中文混合A😀", "e\u0301", "👨‍👩‍👧"):
            for budget in range(24):
                with self.subTest(text=text, budget=budget):
                    result = fit_utf8(text, budget, "")
                    self.assertTrue(text.startswith(result))
                    self.assertLessEqual(len(result.encode("utf-8")), budget)
                    if len(result) < len(text):
                        next_prefix = text[:len(result) + 1]
                        self.assertGreater(len(next_prefix.encode("utf-8")), budget)


if __name__ == "__main__":
    unittest.main()

运行 python -m unittest -v test_utf8_budget。验证重点不是“输出看起来正常”，而是编码后始终不超限、未截断时内容不变、截断时满足最长前缀语义。最后一项矩阵测试还覆盖组合字符及连接表情，但只断言码点边界。

为了确认测试能发现原问题，可临时将实现替换为 return text[:limit_bytes] 再运行；中文、表情和字节预算断言应失败。完成后恢复正确实现并重新运行，禁止把这个缺陷版本用于发送。

发送边界的两个陷阱

字段字节上限不等于 JSON 请求体上限。 json.dumps() 的 ensure_ascii 设置会影响中文是否被转义，加上 JSON 键名、引号及转义字符，请求体长度可能远大于字段的 UTF-8 长度。若协议限制整个请求体，应对实际发送的序列化字节做最终检查，发送与测量使用同一份字节，不能拿本函数的结果代替请求体校验。

完整码点不等于完整可见字符。 组合重音、肤色修饰及连接表情可能在码点之间被截开。本文仅保证合法 UTF-8 和字节预算；产品若要求完整可见字符，应按 Unicode 字素簇分段，再对每个簇累加字节，另加对应测试。

仅对告警摘要、展示描述等允许损失尾部的字段使用截断。订单号、签名原文、SQL、URL、审计证据及机器解析载荷超限时应明确拒绝或改用分片、附件等协议方案，不能静默改变语义。上线后观察长度拒绝计数和截断比例；回滚适配器时也要保留发送前字节校验，避免重新发送已知非法载荷。

验证范围与结论

本文在 Python 3.14.6 上执行上述 9 项 unittest，全部通过；恢复按字符切片的缺陷后出现 41 个断言失败、0 个执行错误，恢复修复后再次全部通过。UTF-8 无 BOM、代码语法和 Markdown 围栏检查通过。环境没有安装 Ruff，未执行其格式化及静态检查，也未新增依赖。

本地测试没有调用真实通知平台，也没有验证平台的序列化、网关或服务端限额。上线前应使用测试环境核对精确边界、错误码和最终请求字节。

这类故障的关键是统一计量单位：协议限制 UTF-8 字节，就在发送边界按 UTF-8 字节验证；需要截断时，将标记也计入预算，并明确码点边界与字素簇边界的区别。

参考：Python Unicode HOWTO、字符串编码方法。