From c822b4ff5387957287a652caf9f469125e058365 Mon Sep 17 00:00:00 2001 From: wshy Date: Thu, 9 Jul 2026 13:28:20 +0800 Subject: [PATCH] =?UTF-8?q?fix(wecom):=20=E8=B6=85=E9=95=BF=E5=A4=8D?= =?UTF-8?q?=E7=9B=98=E6=B6=88=E6=81=AF=E6=88=AA=E6=96=AD=E5=90=8E=E8=BF=BD?= =?UTF-8?q?=E5=8A=A0=E6=8F=90=E7=A4=BA=E8=A1=8C=E5=B9=B6=E6=8F=90=E9=87=8F?= =?UTF-8?q?=20(#85)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 企业微信群机器人 markdown 上限 4096 字节(经官方文档核实), 复盘报告超长时 会被静默截断, 用户无从知晓还有更多内容。 - 截断阈值由 3800 提到 4000 字节, 在 4096 硬限内尽量多显示正文 - 超长截断时追加提示行「…内容较长已截断, 更多详情请回到 TickFlow 应用内查看」 - _truncate_to_bytes 支持自定义 suffix, 按字节为提示行预留空间 - 顺手修正企业微信字段名 msg_type→msgtype, 推送失败日志 debug→warning (此前企业微信推送失败静默吞掉, 日志里看不到) --- backend/app/services/webhook_adapter.py | 44 +++++++++++++++++++++---- 1 file changed, 37 insertions(+), 7 deletions(-) diff --git a/backend/app/services/webhook_adapter.py b/backend/app/services/webhook_adapter.py index 605c473..735cf40 100644 --- a/backend/app/services/webhook_adapter.py +++ b/backend/app/services/webhook_adapter.py @@ -28,6 +28,13 @@ _MAX_LEN = 500 # 卡片消息正文最长字符 (飞书 interactive 卡片上限 30KB, 保守留余量给标题/结构) _CARD_MAX_LEN = 28000 +# 企业微信群机器人 markdown 消息上限 4096 字节 (非字符; 中文每字 3 字节), +# 留余量给标题、格式符及截断提示行。 +_WECOM_MD_MAX_BYTES = 4000 + +# 截断提示行: 正文超长被截断时追加, 引导用户回应用内查看完整内容。 +_WECOM_TRUNCATED_HINT = "\n\n…内容较长已截断，更多详情请回到 TickFlow 应用内查看。" + # 飞书自定义机器人 Webhook 前缀 (用于 URL 合法性校验) FEISHU_HOOK_PREFIX = "https://open.feishu.cn/open-apis/bot/v2/hook/" @@ -63,6 +70,26 @@ def _truncate_card(text: str) -> str: return text[:_CARD_MAX_LEN] + ("…" if len(text) > _CARD_MAX_LEN else "") +def _truncate_to_bytes(text: str, max_bytes: int, suffix: str = "…") -> str: + """按 UTF-8 字节数安全截断 (不截断在多字节字符中间, 末尾补 suffix)。 + + 企业微信 markdown 按**字节**计长 (上限 4096), 中文每字 3 字节, + 不能用字符数截断 (len(str) 算的是字符数, 会超字节上限)。 + + Args: + max_bytes: 含 suffix 在内的总字节上限。 + suffix: 被截断时追加的结尾 (如省略号或提示行)。 + """ + text = (text or "").strip() + encoded = text.encode("utf-8") + if len(encoded) <= max_bytes: + return text + suffix_bytes = suffix.encode("utf-8") + # 为 suffix 留出字节, 再用 errors='ignore' 丢弃被截断的尾字节, 避免半个字符 + cut = encoded[:max_bytes - len(suffix_bytes)] + return cut.decode("utf-8", errors="ignore") + suffix + + _FEISHU_MAX_ATTEMPTS = 3 @@ -195,7 +222,7 @@ def send_feishu_card(webhook_url: str, title: str, subtitle: str, body_md: str, # 关键差异: # 1. Webhook 形态: https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx # 2. 无需签名校验 (key 本身即凭证; 企业微信群机器人可选"签名校验"但极少用) -# 3. Markdown 原生支持 (msg_type=markdown), 不必像飞书那样包进 interactive 卡片 +# 3. Markdown 原生支持 (msgtype=markdown), 不必像飞书那样包进 interactive 卡片 # 4. 成功响应: {"errcode":0,"errmsg":"ok"} # # 限制: 每个机器人每分钟最多 20 条消息 (超出会被限流 460min 内不可用), @@ -250,14 +277,14 @@ def _post_wecom(webhook_url: str, payload: dict) -> bool: # errcode=0 表示成功; 45009=频率限制, 其它非零=业务失败 if data.get("errcode") == 0: return True - logger.debug("企业微信推送业务失败: %s", data) + logger.warning("企业微信推送业务失败: %s", data) return False except ValueError: return True - logger.debug("企业微信推送 HTTP %s: %s", resp.status_code, resp.text[:200]) + logger.warning("企业微信推送 HTTP %s: %s", resp.status_code, resp.text[:200]) return False except Exception as e: # noqa: BLE001 - logger.debug("企业微信 Webhook 推送失败: %s", e) + logger.warning("企业微信 Webhook 推送失败: %s", e) return False @@ -281,7 +308,7 @@ def send_wecom(webhook_url: str, title: str, body: str) -> bool: if not text: return False - payload: dict = {"msg_type": "text", "text": {"content": text}} + payload: dict = {"msgtype": "text", "text": {"content": text}} return _post_wecom(webhook_url, payload) @@ -303,10 +330,13 @@ def send_wecom_markdown(webhook_url: str, title: str, body_md: str) -> bool: if not is_valid_wecom_url(webhook_url): return False - content = f"## {title}\n\n{_truncate_card(body_md)}" + # 企业微信 markdown 上限 4096 字节 (含标题), 用按字节截断避免超限被拒。 + # 超长时追加提示行, 引导用户回应用内查看完整报告。 + raw = f"## {title}\n\n{body_md}".strip() + content = _truncate_to_bytes(raw, _WECOM_MD_MAX_BYTES, suffix=_WECOM_TRUNCATED_HINT) if not content.strip(): return False - payload: dict = {"msg_type": "markdown", "markdown": {"content": content}} + payload: dict = {"msgtype": "markdown", "markdown": {"content": content}} return _post_wecom(webhook_url, payload)