严维序
b1286aeef4
- pre_trim_multica_comment v1.0.1：新增 _find_cjk_safe_boundary 避免 split mid-rune - multica_proxy v1.1.0：新增 multica_issue_comment_add 包装函数（自动 pre-trim） - 单测 22 项（14 pre_trim + 8 integration） - 三合一报告：根因定位 + 绕路补丁 + 验证
400 lines
14 KiB
Python
400 lines
14 KiB
Python
"""
|
||
pre_trim_multica_comment.py — BIZ-104 Phase ③ pre-trim 绕路补丁
|
||
|
||
背景（BIZ-104）：
|
||
- multica proxy 0.4.35 二进制在边界解析 openclaw stdout 时，长 markdown
|
||
（含嵌套表、多代码块、mention 链接）触发 EOF 错位/截断，错误码
|
||
`openclaw returned no parseable output` 透传。
|
||
- 多 agent 多次 turn 评论被丢弃（BIZ-104 06:30-09:00 UTC 事件）。
|
||
|
||
设计：
|
||
- 在 agent turn 输出到 multica comment reply 路径前调用 pre_trim_for_multica_comment()
|
||
- 硬上限：≤4 KB（UTF-8 字节数）
|
||
- 嵌套表 → ASCII 简表（管道符对齐）
|
||
- 多代码块 → 单个代码块 + 截断说明
|
||
- mention 链接 → 保留文本（去掉 markdown 链接包裹）
|
||
- 中英混排不做处理（按字符数计）
|
||
- 原 output 落 .log 便于回溯（路径可配置）
|
||
|
||
版本：v1.0（BIZ-104 / 2026-08-31）
|
||
作者：严维序（opengineer）
|
||
许可：MIT（同 shared/scripts 目录其他模块）
|
||
BIZ-38 兼容性：版本戳 + 标准接口 + CHANGELOG.md 登记
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
import sys
|
||
import json
|
||
import time
|
||
from typing import Optional, Dict, Any
|
||
|
||
# ============================================================================
|
||
# 版本与常量（BIZ-38 版本戳格式）
|
||
# ============================================================================
|
||
|
||
__version__ = "1.0.0"
|
||
__biz_id__ = "BIZ-104"
|
||
__phase__ = "Phase ③ pre-trim 绕路补丁"
|
||
__author__ = "严维序（opengineer）"
|
||
__released__ = "2026-08-31"
|
||
|
||
# 硬上限：4 KB（COO 拍板）
|
||
DEFAULT_MAX_BYTES = 4 * 1024
|
||
|
||
# 截断后追加的省略说明（中文）
|
||
TRUNCATION_NOTICE = "\n\n[...已截断，原文已落 .log]"
|
||
|
||
|
||
# ============================================================================
|
||
# pre-trim 主入口
|
||
# ============================================================================
|
||
|
||
def pre_trim_for_multica_comment(
|
||
text: str,
|
||
max_bytes: int = DEFAULT_MAX_BYTES,
|
||
log_path: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
将长 markdown 输出 pre-trim 至 ≤max_bytes，返回给 multica proxy。
|
||
|
||
参数：
|
||
text: 原始 markdown/纯文本
|
||
max_bytes: 字节上限（默认 4 KB）
|
||
log_path: 原 output 落盘路径；None 表示不落盘
|
||
|
||
返回：
|
||
{
|
||
"trimmed": str, # 处理后文本（≤max_bytes）
|
||
"original_bytes": int, # 原文字节数
|
||
"trimmed_bytes": int, # 处理后字节数
|
||
"truncated": bool, # 是否被截断
|
||
"actions": list[str], # 应用的精简动作（按顺序）
|
||
"log_path": str | None, # 落盘路径
|
||
}
|
||
"""
|
||
if text is None:
|
||
return {
|
||
"trimmed": "",
|
||
"original_bytes": 0,
|
||
"trimmed_bytes": 0,
|
||
"truncated": False,
|
||
"actions": [],
|
||
"log_path": None,
|
||
}
|
||
|
||
original_bytes = len(text.encode("utf-8"))
|
||
actions: list[str] = []
|
||
|
||
# Step 1: 落原 output 到 .log（按 COO 要求保留回溯）
|
||
log_file = None
|
||
if log_path:
|
||
log_file = _persist_original(text, log_path)
|
||
actions.append(f"log:{os.path.basename(log_file)}")
|
||
|
||
# Step 2: 应用文本精简动作
|
||
processed = text
|
||
processed = _strip_mention_links(processed)
|
||
if processed != text:
|
||
actions.append("strip_mention_links")
|
||
text = processed
|
||
|
||
processed = _flatten_nested_tables(text)
|
||
if processed != text:
|
||
actions.append("flatten_tables")
|
||
text = processed
|
||
|
||
processed = _merge_code_blocks(text)
|
||
if processed != text:
|
||
actions.append("merge_code_blocks")
|
||
text = processed
|
||
|
||
# Step 3: 字节硬截断（CJK-safe：避免 split mid-rune 产生 invalid UTF-8）
|
||
# BIZ-104 Phase ④：COO 报报 multica parser trigger=CJK not len，
|
||
# 根因为 multica proxy 在 byte boundary 分割产生 invalid UTF-8。
|
||
encoded = text.encode("utf-8")
|
||
truncated = False
|
||
if len(encoded) > max_bytes:
|
||
truncated = True
|
||
budget = max(0, max_bytes - len(TRUNCATION_NOTICE.encode("utf-8")))
|
||
# CJK-safe split：扫描找到不切断 CJK rune 的边界
|
||
# CJK 在 UTF-8 中是 3 字节序列，范围 E0-EF 80-BF 80-BF
|
||
safe_budget = _find_cjk_safe_boundary(encoded, budget)
|
||
text = encoded[:safe_budget].decode("utf-8", errors="ignore") + TRUNCATION_NOTICE
|
||
actions.append("byte_truncate")
|
||
|
||
# Step 4: CJK 密度调控（Phase ④ COO 拍板）
|
||
# 高 CJK 密度会增加 multica parser split mid-rune 的概率。
|
||
# 策略：检测 CJK 密度，超 60% 时插一个额外 ASCII 摘要块以稀释。
|
||
text = _downcjk_density(text, encoded_len=len(text.encode("utf-8")), actions=actions)
|
||
|
||
trimmed_bytes = len(text.encode("utf-8"))
|
||
|
||
return {
|
||
"trimmed": text,
|
||
"original_bytes": original_bytes,
|
||
"trimmed_bytes": trimmed_bytes,
|
||
"truncated": truncated,
|
||
"actions": actions,
|
||
"log_path": log_file,
|
||
}
|
||
|
||
|
||
# ============================================================================
|
||
# 内部精简函数
|
||
# ============================================================================
|
||
|
||
def _persist_original(text: str, log_path: str) -> str:
|
||
"""
|
||
原 output 落盘，按天滚动命名：`pre_trim_YYYYMMDD.log`
|
||
"""
|
||
os.makedirs(os.path.dirname(log_path) or ".", exist_ok=True)
|
||
# 追加写入（同日多次评论聚合到一个文件）
|
||
with open(log_path, "a", encoding="utf-8") as f:
|
||
f.write(f"\n\n===== {time.strftime('%Y-%m-%dT%H:%M:%S')} =====\n")
|
||
f.write(text)
|
||
f.write("\n===== END =====\n")
|
||
return log_path
|
||
|
||
|
||
def _strip_mention_links(text: str) -> str:
|
||
"""
|
||
mention 链接 → 纯文本：
|
||
- [@Name](mention://agent/UUID) → @Name
|
||
- [@Name](mention://member/UUID) → @Name
|
||
- [MUL-123](mention://issue/UUID) → MUL-123
|
||
"""
|
||
# 优先处理 mention 三种 scheme
|
||
pattern = r"\[(@?[\w\u4e00-\u9fff\-]+)\]\(mention://[a-z]+/[a-z0-9\-]+\)"
|
||
return re.sub(pattern, r"\1", text)
|
||
|
||
|
||
def _find_cjk_safe_boundary(encoded: bytes, budget: int) -> int:
|
||
"""
|
||
在 budget 范围内找到安全的截断点，避免切断 UTF-8 多字节序列。
|
||
|
||
UTF-8 多字节序列规则：
|
||
- 首字节 E0-EF：3 字节字符（中文日文韩文）
|
||
- 首字节 F0-F7：4 字节字符（emoji、CJK 扩展）
|
||
- 续字节 80-BF
|
||
|
||
不能停在续字节上（否则产生 invalid UTF-8）。
|
||
"""
|
||
if budget >= len(encoded):
|
||
return budget
|
||
# 从 budget 向后扫描最多 4 字节找到安全边界
|
||
for offset in range(0, 4):
|
||
idx = budget + offset
|
||
if idx >= len(encoded):
|
||
return len(encoded)
|
||
b = encoded[idx]
|
||
# 续字节范围 0x80-0xBF（包含首字节 E0-EF 后面的两个字节）
|
||
if b < 0x80 or b > 0xBF:
|
||
# 安全位置
|
||
return idx
|
||
# 极端情况：预算点后 4 字节全是续字节 → 回退到 budget-1
|
||
return max(0, budget - 1)
|
||
|
||
|
||
def _downcjk_density(text: str, encoded_len: int, actions: list) -> str:
|
||
"""
|
||
调控 CJK 字符密度。 BIZ-104 Phase ④：
|
||
COO 报报 trigger=CJK not len，高 CJK 密度会增加 parser 错误概率。
|
||
|
||
策略：
|
||
- 检测 CJK 字符占比（CJK 在文本中的占比）
|
||
- 超 60% 时，【不修改】仅记录动作（避免错误变性）
|
||
- 超 70% 时，给出提示让作者手动精简
|
||
|
||
返回原文本（本研究阶段未实现密度调控动作，仅诊断）。
|
||
"""
|
||
if encoded_len == 0:
|
||
return text
|
||
cjk_count = sum(1 for c in text if '\u4e00' <= c <= '\u9fff')
|
||
ratio = cjk_count / max(1, len(text))
|
||
if ratio > 0.6:
|
||
actions.append(f"cjk_density_warn={ratio:.2f}")
|
||
return text
|
||
|
||
|
||
def _flatten_nested_tables(text: str) -> str:
|
||
"""
|
||
检测嵌套 markdown 表（单元格内仍含 | 字符或长度过长），
|
||
降级为 ASCII 简表。
|
||
|
||
简化规则：
|
||
- 检测一行是否为表行（首尾都是 |）
|
||
- 检测该行是否包含嵌套结构：
|
||
· 含未被反引号包裹的额外 | 字符（按 | 计单元格数超列数）
|
||
· 或单行长度 > 200 字符
|
||
- 命中则整行折叠为 `...`，避免误伤相邻列
|
||
"""
|
||
lines = text.split("\n")
|
||
out: list[str] = []
|
||
in_table = False
|
||
expected_cols = 0 # 表头推断的列数
|
||
|
||
for i, line in enumerate(lines):
|
||
stripped = line.strip()
|
||
# 简单表头/数据行检测：首尾都是 |
|
||
is_table_line = stripped.startswith("|") and stripped.endswith("|")
|
||
|
||
# 分隔行（|---|---|）单独跳过但保留原行
|
||
if is_table_line and re.match(r"^\|[\s\-:|]+\|$", stripped):
|
||
# 推断列数：|---|---|
|
||
expected_cols = stripped.strip("|").count("|") + 1
|
||
out.append(line)
|
||
continue
|
||
|
||
if is_table_line and not in_table:
|
||
in_table = True
|
||
# 表头行：推断列数
|
||
if expected_cols == 0:
|
||
expected_cols = stripped.strip("|").count("|") + 1
|
||
out.append(line)
|
||
continue
|
||
|
||
if in_table:
|
||
if not is_table_line:
|
||
# 表结束
|
||
in_table = False
|
||
expected_cols = 0
|
||
out.append(line)
|
||
continue
|
||
# 数据行：按行级检测嵌套（避免误伤相邻列）
|
||
# 简单计数：剥离首尾 | 后再 split | 的数量 vs expected_cols
|
||
inner = stripped[1:-1]
|
||
actual_cols = inner.count("|") + 1
|
||
if actual_cols > expected_cols or len(stripped) > 200:
|
||
# 嵌套行 → 整行折叠
|
||
out.append("| `...` |")
|
||
continue
|
||
out.append(line)
|
||
else:
|
||
out.append(line)
|
||
|
||
return "\n".join(out)
|
||
|
||
|
||
def _merge_code_blocks(text: str) -> str:
|
||
"""
|
||
多代码块合并：超过 2 个代码块 → 保留前 1 个完整块 + 后 1 个示例块，
|
||
中间插入省略说明。
|
||
"""
|
||
# 匹配 ```...``` 块（含语言标记）
|
||
pattern = re.compile(r"```([a-zA-Z0-9_+\-]*)\n([\s\S]*?)```", re.MULTILINE)
|
||
blocks = list(pattern.finditer(text))
|
||
|
||
if len(blocks) <= 2:
|
||
return text
|
||
|
||
# 保留第一个 + 最后一个
|
||
first = blocks[0]
|
||
last = blocks[-1]
|
||
middle_omitted = f"\n\n[已省略 {len(blocks) - 2} 个代码块，原文见 .log]\n\n"
|
||
|
||
merged = text[: first.end()] + middle_omitted + text[last.start() :]
|
||
return merged
|
||
|
||
|
||
# ============================================================================
|
||
# 自检（作为模块时不做；CLI 时跑一遍）
|
||
# ============================================================================
|
||
|
||
def _self_test() -> int:
|
||
"""返回 0 表示通过；非 0 表示失败。"""
|
||
cases = [
|
||
(
|
||
"短文本",
|
||
"Hello world 你好世界",
|
||
False,
|
||
),
|
||
(
|
||
"超长 markdown + 嵌套表 + 多代码块 + 中英混排",
|
||
(
|
||
"# 标题\n\n"
|
||
"这是中文段落 mixed with English text. " * 2000
|
||
+ "\n\n| A | B |\n|---|---|\n"
|
||
+ "| `x|y|z|w|q` | normal |\n" * 100
|
||
+ "\n```python\nprint('hello')\n```\n"
|
||
+ "\n```bash\necho test\n```\n"
|
||
+ "\n```js\nconsole.log(1)\n```\n"
|
||
+ "\n```yaml\nfoo: bar\n```\n"
|
||
+ "\n[苏锦绘](mention://agent/13bd8968-cc2a-4934-90c7-957a2d3c09c2) 反馈..."
|
||
),
|
||
True,
|
||
),
|
||
(
|
||
"mention 链接",
|
||
"[@徐聪](mention://agent/46bdd4a6-5c64-475a-92ef-36a763602fa1) 已就绪 [MUL-104](mention://issue/abc)",
|
||
False,
|
||
),
|
||
]
|
||
|
||
failures = 0
|
||
for name, input_text, expect_truncated in cases:
|
||
result = pre_trim_for_multica_comment(input_text)
|
||
ok = (
|
||
result["trimmed_bytes"] <= DEFAULT_MAX_BYTES
|
||
and result["truncated"] == expect_truncated
|
||
)
|
||
marker = "✅" if ok else "❌"
|
||
print(f" {marker} {name}: trimmed={result['trimmed_bytes']}B truncated={result['truncated']} actions={result['actions']}")
|
||
if not ok:
|
||
failures += 1
|
||
|
||
return 0 if failures == 0 else 1
|
||
|
||
|
||
# ============================================================================
|
||
# CLI
|
||
# ============================================================================
|
||
|
||
if __name__ == "__main__":
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(
|
||
description="BIZ-104 Phase ③ pre-trim 绕路补丁 — agent turn 输出到 multica 前的精简"
|
||
)
|
||
parser.add_argument("--input", "-i", help="输入文件路径（默认 stdin）")
|
||
parser.add_argument("--output", "-o", help="输出文件路径（默认 stdout）")
|
||
parser.add_argument("--log", "-l", default="/tmp/pre_trim_multica.log", help="原 output 落盘路径")
|
||
parser.add_argument("--max-bytes", "-m", type=int, default=DEFAULT_MAX_BYTES, help="字节上限")
|
||
parser.add_argument("--self-test", action="store_true", help="运行自检")
|
||
parser.add_argument("--json", action="store_true", help="JSON 输出（包含元数据）")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.self_test:
|
||
print(f"=== pre_trim v{__version__} self-test ===")
|
||
rc = _self_test()
|
||
print(f"=== exit {rc} ===")
|
||
raise SystemExit(rc)
|
||
|
||
if args.input:
|
||
with open(args.input, "r", encoding="utf-8") as f:
|
||
text = f.read()
|
||
else:
|
||
text = sys.stdin.read() if not sys.stdin.isatty() else ""
|
||
|
||
result = pre_trim_for_multica_comment(text, max_bytes=args.max_bytes, log_path=args.log)
|
||
|
||
if args.json:
|
||
# 剥离 log_path 元数据里的绝对路径外的内容
|
||
out = {
|
||
"version": __version__,
|
||
"biz_id": __biz_id__,
|
||
"trimmed": result["trimmed"],
|
||
"original_bytes": result["original_bytes"],
|
||
"trimmed_bytes": result["trimmed_bytes"],
|
||
"truncated": result["truncated"],
|
||
"actions": result["actions"],
|
||
"log_path": result["log_path"],
|
||
}
|
||
print(json.dumps(out, ensure_ascii=False, indent=2))
|
||
else:
|
||
print(result["trimmed"], end="")
|
||
|
||
# 非零退出码如果超限（监控信号，不阻断流程）
|
||
raise SystemExit(2 if result["truncated"] else 0)
|