feat(tools): 添加 Yarn 文本审校脚本

Co-authored-by: Cursor <cursoragent@cursor.com>

Tools/yarn_text_audit.py

@@ -0,0 +1,812 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import csv
+import hashlib
+import json
+import os
+import re
+import sys
+import tempfile
+import unicodedata
+from collections import Counter, defaultdict
+from pathlib import Path
+
+LINE_TAG_RE = re.compile(r"(?<!\\)#line:([0-9A-Za-z_-]+)")
+META_TAG_RE = re.compile(r"(?<!\\)\s*#(?:line:[0-9A-Za-z_-]+|auto_next|autonext|option_prompt)\b")
+COMMAND_RE = re.compile(r"^<<.*?>>(?:\s*//.*)?\s*$")
+INLINE_IF_RE = re.compile(r"<<if\b.*?>>")
+ANGLE_TAG_RE = re.compile(r"<[^>]+>")
+YARN_MARKUP_RE = re.compile(r"\[\[/?[^\]]+\]\]")
+CHAR_ATTR_RE = re.compile(r"^\[character\b[^\]]*/\]")
+SPEAKER_RE = re.compile(r"^([^:：\s][^:：]{0,29})[:：]\s*(.*)$")
+REPEAT_RE = re.compile(r"([\u3400-\u9fff]{2,6})\1")
+
+WEAK_LINE_START = tuple("的地得了着过而但就也却吗呢啊吧呀")
+TERM_PATTERNS = (
+    "UF模块", "ULM自训练进程", "UF", "ULM", "UnstableFusion", "非稳态聚合模块",
+    "幸福罐头公司", "色觉晶圆", "原型泄露", "表达模块", "情绪模块", "逻辑模块",
+    "销售语言拓展模块", "记忆打孔带", "视觉模块",
+)
+REVIEW_ACCEPTED = "接受修改"
+REVIEW_KEEP = "保留原文"
+REVIEW_FOLLOW_NOTE = "按照备注修改"
+MANUAL_BREAK_WIDTH = 18.0
+
+def sha256(path: Path) -> str:
+    h = hashlib.sha256()
+    with path.open("rb") as f:
+        for chunk in iter(lambda: f.read(1024 * 1024), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+def visible_width(text: str) -> float:
+    total = 0.0
+    for ch in text:
+        if unicodedata.combining(ch):
+            continue
+        total += 1.0 if unicodedata.east_asian_width(ch) in "WFA" else 0.5
+    return total
+
+def strip_markup(text: str) -> str:
+    text = ANGLE_TAG_RE.sub("", text)
+    text = YARN_MARKUP_RE.sub("", text)
+    return text.replace("\\#", "#").strip()
+
+def review_choice(value) -> str:
+    if isinstance(value, list):
+        return value[0] if value else ""
+    return value or ""
+
+def replace_ascii_quotes(text: str) -> str:
+    result = []
+    opening = True
+    for char in text:
+        if char != '"':
+            result.append(char)
+            continue
+        result.append("“" if opening else "”")
+        opening = not opening
+    return "".join(result)
+
+def dialogue_parts(raw: str) -> tuple[str, str, str]:
+    """Return the immutable dialogue prefix, editable body, and metadata suffix."""
+    metadata = META_TAG_RE.search(raw)
+    source = raw[:metadata.start()].rstrip() if metadata else raw.rstrip()
+    suffix = raw[len(source):]
+
+    cursor = 0
+    leading = re.match(r"^\s*(?:->\s*)?", source)
+    if leading:
+        cursor = leading.end()
+    attr = CHAR_ATTR_RE.match(source[cursor:])
+    if attr:
+        cursor += attr.end()
+        whitespace = re.match(r"\s*", source[cursor:])
+        cursor += whitespace.end() if whitespace else 0
+    speaker = SPEAKER_RE.match(source[cursor:])
+    if speaker and not any(c in speaker.group(1) for c in "<>[]{}，。！？…"):
+        cursor += speaker.start(2)
+    return source[:cursor], source[cursor:], suffix
+
+def wrap_phrases(text: str, phrases: list[str], marker: str) -> str:
+    phrases = sorted({x for x in phrases if x}, key=len, reverse=True)
+    if not phrases:
+        return text
+    pattern = re.compile("|".join(re.escape(x) for x in phrases))
+    return pattern.sub(lambda match: f"[[{marker}]]{match.group(0)}[[/{marker}]]", text)
+
+def detected_terms(text: str) -> list[str]:
+    plain = strip_markup(text)
+    return [term for term in TERM_PATTERNS if term in plain]
+
+def comma_break_positions(text: str) -> tuple[list[int], list[float]]:
+    visible_positions = []
+    width = 0.0
+    index = 0
+    while index < len(text):
+        if text[index] == "<":
+            close = text.find(">", index + 1)
+            if close >= 0:
+                index = close + 1
+                continue
+        if text.startswith("[[", index):
+            close = text.find("]]", index + 2)
+            if close >= 0:
+                index = close + 2
+                continue
+        char = text[index]
+        width += visible_width(char)
+        if char == "，":
+            visible_positions.append((index, width))
+        index += 1
+    return [x[0] for x in visible_positions], [x[1] for x in visible_positions] + [width]
+
+def add_comfortable_breaks(text: str) -> str:
+    if re.search(r"(?i)<br\s*/?>", text):
+        return text
+    positions, widths = comma_break_positions(text)
+    if not positions:
+        return text
+    total_width = widths[-1]
+    target_lines = max(1, int((total_width + MANUAL_BREAK_WIDTH - 0.001) // MANUAL_BREAK_WIDTH))
+    break_count = min(target_lines - 1, len(positions))
+    if break_count <= 0:
+        return text
+
+    comma_widths = widths[:-1]
+    selected = []
+    previous = -1.0
+    for part in range(1, break_count + 1):
+        target = total_width * part / (break_count + 1)
+        candidates = [
+            (abs(width - target), idx, width)
+            for idx, width in zip(positions, comma_widths)
+            if width > previous + 4 and idx not in selected
+        ]
+        if not candidates:
+            continue
+        _, chosen, chosen_width = min(candidates)
+        selected.append(chosen)
+        previous = chosen_width
+
+    for index in sorted(selected, reverse=True):
+        insert_at = index + 1
+        wait = re.match(r"<waitfor\s*=\s*[^>]+>", text[insert_at:], flags=re.IGNORECASE)
+        if wait:
+            insert_at += wait.end()
+            insertion = "<br>"
+        else:
+            insertion = "[[comma-pause]]<br>"
+        text = text[:insert_at] + insertion + text[insert_at:]
+    return text
+
+def apply_note(body: str, note: str, category: str) -> str:
+    quoted = re.findall(r"“([^”]+)”", note)
+    if "直接去掉#autonext" in note:
+        return body
+    if "改为“色觉晶圆”" in note:
+        body = re.sub(r"(?<!色觉)晶圆", "色觉晶圆", body)
+        return wrap_phrases(body, ["色觉晶圆"], "term")
+    if "改为“码”" in note:
+        body = body.replace("妈", "码")
+    if "去掉“*”" in note:
+        body = body.replace("*", "")
+    if "加wiggle" in note:
+        for phrase in quoted:
+            emphasized = f"*{phrase}*"
+            if emphasized in body:
+                body = body.replace(
+                    emphasized, f"*[[joke]]{phrase}[[/joke]]*")
+            else:
+                body = wrap_phrases(body, [phrase], "joke")
+        return body
+    if "只" in note and ("断行" in note or "换行" in note):
+        return add_comfortable_breaks(body)
+
+    if "并用蓝字" in note:
+        body = wrap_phrases(body, detected_terms(body), "term")
+    elif note.startswith("用黄字"):
+        body = wrap_phrases(body, detected_terms(body), "key")
+    elif "专有名词" in category:
+        body = wrap_phrases(body, detected_terms(body), "term")
+
+    if quoted and ("黄字" in note):
+        body = wrap_phrases(body, quoted, "key")
+    if "断行" in note:
+        body = add_comfortable_breaks(body)
+    return body
+
+def transform_reviewed_line(row: dict) -> str:
+    original = row["原文"]
+    decision = review_choice(row.get("审阅结论"))
+    if decision == REVIEW_KEEP:
+        return original
+
+    note = row.get("审阅备注") or ""
+    category = row.get("问题分类") or ""
+    suggested = row.get("建议文本") or ""
+    if decision == REVIEW_ACCEPTED and suggested:
+        return suggested
+
+    prefix, body, suffix = dialogue_parts(original)
+    if decision == REVIEW_FOLLOW_NOTE:
+        if "直接去掉#autonext" in note:
+            return original.replace("#autonext", "")
+        body = apply_note(body, note, category)
+    elif decision == REVIEW_ACCEPTED:
+        if "标点规范" in category:
+            body = replace_ascii_quotes(body)
+        if "空格" in category:
+            body = re.sub(r"\s+([，。！？；：])", r"\1", body)
+        if "专有名词" in category and "[[term]]" in (row.get("建议标记") or ""):
+            body = wrap_phrases(body, detected_terms(body), "term")
+        if "ULM效果" in category and "[[ulm]]" in (row.get("建议标记") or ""):
+            prefix += "[[ulm]]"
+        if "笑话效果" in category and "[[joke]]" in (row.get("建议标记") or ""):
+            body = f"[[joke]]{body}[[/joke]]"
+        if "换行候选" in category:
+            body = add_comfortable_breaks(body)
+
+    return prefix + body + suffix
+
+def source_without_metadata(raw: str) -> str:
+    match = META_TAG_RE.search(raw)
+    return raw[:match.start()].rstrip() if match else raw.rstrip()
+
+def split_speaker(text: str) -> tuple[str, str]:
+    attr = CHAR_ATTR_RE.match(text)
+    if attr:
+        text = text[attr.end():].lstrip()
+    match = SPEAKER_RE.match(text)
+    if not match:
+        return "", text
+    speaker = match.group(1).strip()
+    if any(c in speaker for c in "<>[]{}，。！？…"):
+        return "", text
+    return speaker, match.group(2)
+
+def issue(issue_type: str, severity: str, confidence: str, reason: str,
+          proposal: str = "", suggested_text: str = "") -> dict:
+    return {
+        "category": issue_type,
+        "severity": severity,
+        "confidence": confidence,
+        "reason": reason,
+        "proposal": proposal,
+        "suggested_text": suggested_text,
+    }
+
+def detect(record: dict) -> list[dict]:
+    text = record["text"]
+    visible = record["visible_text"]
+    speaker = record["speaker"]
+    found: list[dict] = []
+
+    if REPEAT_RE.search(visible):
+        found.append(issue("重复候选", "需确认", "低", "发现相邻重复片段，可能是强调或口语节奏，也可能是复制错误。"))
+    if re.search(r"[，。！？；：]{2,}", visible):
+        found.append(issue("标点", "建议优化", "中", "发现连续中文标点，需确认是否为误输入。"))
+    if "..." in visible or re.search(r'["\']', visible):
+        found.append(issue("标点规范", "建议优化", "中", "包含 ASCII 省略号或引号，建议统一为中文标点。"))
+    if re.search(r"\s+[，。！？；：]", visible):
+        found.append(issue("空格", "明确错误", "高", "中文标点前存在多余空格。"))
+    if re.search(r"(慢慢|悄悄|轻轻|渐渐|狠狠|迅速|飞快|大声|小声)的(?=[\u3400-\u9fff])", visible):
+        found.append(issue("的地得", "建议优化", "中", "状语后疑似误用“的”，建议结合谓语确认是否改为“地”。"))
+
+    segments = [strip_markup(x) for x in re.split(r"(?i)<br\s*/?>", text)]
+    widths = [visible_width(x) for x in segments]
+    if len(segments) > 1:
+        bad_start = any(seg.startswith(WEAK_LINE_START) for seg in segments[1:] if seg)
+        orphan = any(0 < width <= 2 for width in widths)
+        imbalance = max(widths, default=0) - min(widths, default=0) >= 8
+        if bad_start or orphan or imbalance:
+            found.append(issue("换行", "建议优化", "中",
+                "现有断行存在弱助词开头、孤行或明显失衡；仅在不增加显示行数时调整。"))
+    elif visible_width(visible) > 24 and "，" in visible:
+        found.append(issue("换行候选", "需确认", "低",
+            "单行较长且含逗号，可在保持总显示行数的前提下评估断行。"))
+
+    if speaker in {"me", "mecd", "med"} and (
+        re.match(r"^[（(].*[）)]$", visible) or any(k in visible for k in ("小声", "低声", "喃喃", "默念"))
+    ) and "[[aside]]" not in text and "#5F5F5F" not in text:
+        found.append(issue("自言自语/念文本", "需确认", "低",
+            "疑似自言自语、阅读或低声表达，请确认灰色语义标记。", "[[aside]]…[[/aside]]"))
+
+    if speaker == "ULM自训练进程" and "[[ulm]]" not in text and "vertexp" not in text:
+        found.append(issue("ULM效果", "样式标记", "高",
+            "活动 ULM 自训练进程文本应使用统一出现效果。", "[[ulm]]"))
+
+    terms = [term for term in TERM_PATTERNS if term in visible]
+    if terms and "[[term]]" not in text and "#52B6FF" not in text and "#FFB200" not in text:
+        found.append(issue("专有名词", "需确认", "低",
+            "包含游戏专有名词：" + "、".join(terms) + "。颜色需按本次语境逐处确认，不因术语出现而自动标色。",
+            "[[term]]…[[/term]]"))
+
+    if record["module"] == "FP_Huoshan1" and speaker == "hs" and (
+        any(k in visible.replace("*", "") for k in (
+            "看医生其实", "衣 帽", "衣帽间", "路师", "凶柿", "西红凶", "帽问题", "嫡中嫡"
+        ))
+    ) and "[[joke]]" not in text and "<wiggle" not in text:
+        found.append(issue("笑话效果", "样式标记", "中",
+            "疑似实际讲笑话的台词；确认后使用统一 wiggle 语义标记。", "[[joke]]…[[/joke]]"))
+
+    known_replacements = {
+        "信号接受正常": "信号接收正常",
+        "2000-1000hz": "2000-10000hz",
+        "打折回旋": "打着回旋",
+        "VERYFYING": "VERIFYING",
+        "检查到特征频域异常": "检测到特征频域异常",
+        "#autonext": "#auto_next",
+    }
+    raw_source = record.get("raw", record.get("source_text", ""))
+    for wrong, right in known_replacements.items():
+        if wrong in raw_source:
+            found.append(issue("错别字/漏字", "明确错误", "高",
+                f"检测到明确文本错误：{wrong} → {right}。",
+                suggested_text=raw_source.replace(wrong, right)))
+
+    # Do not duplicate categories for one source line.
+    merged = {}
+    for item in found:
+        merged.setdefault(item["category"], item)
+    return list(merged.values())
+
+def parse_file(path: Path, repo: Path) -> tuple[list[dict], dict]:
+    lines = path.read_text(encoding="utf-8-sig").splitlines()
+    records: list[dict] = []
+    node = ""
+    pending_title = ""
+    in_body = False
+    counts = Counter()
+    file_hash = sha256(path)
+    rel = path.relative_to(repo).as_posix()
+    module = next((part for part in path.parts if part.startswith("FP_")), path.parent.name)
+
+    for number, raw in enumerate(lines, 1):
+        stripped = raw.strip()
+        if not in_body:
+            if stripped.startswith("title:"):
+                pending_title = stripped.split(":", 1)[1].strip()
+            if stripped == "---":
+                node = pending_title
+                in_body = True
+                counts["delimiter"] += 1
+            elif stripped:
+                counts["metadata"] += 1
+            else:
+                counts["blank"] += 1
+            continue
+
+        if stripped == "===":
+            in_body = False
+            counts["delimiter"] += 1
+            continue
+        if not stripped:
+            counts["blank"] += 1
+            continue
+        if stripped.startswith("//"):
+            counts["comment"] += 1
+            continue
+        if COMMAND_RE.match(stripped):
+            counts["command"] += 1
+            continue
+
+        kind = "option" if stripped.startswith("->") else "text"
+        active = stripped[2:].lstrip() if kind == "option" else stripped
+        active = INLINE_IF_RE.sub("", active).rstrip()
+        source_text = source_without_metadata(active)
+        line_tag_match = LINE_TAG_RE.search(raw)
+        line_tag = line_tag_match.group(1) if line_tag_match else ""
+        speaker, text = split_speaker(source_text)
+        clean = strip_markup(text)
+        record_id = hashlib.sha1(f"{rel}:{number}:{raw}".encode("utf-8")).hexdigest()[:12]
+        record = {
+            "id": record_id,
+            "module": module,
+            "file": rel,
+            "file_sha256": file_hash,
+            "node": node,
+            "line": number,
+            "line_tag": line_tag,
+            "kind": kind,
+            "speaker": speaker,
+            "raw": raw,
+            "source_text": source_text,
+            "text": text,
+            "visible_text": clean,
+            "display_lines": len(re.split(r"(?i)<br\s*/?>", text)),
+        }
+        findings = detect(record)
+        record["findings"] = findings
+        record["review_status"] = "建议修改" if any(
+            x["confidence"] == "高" or x["severity"] == "明确错误" for x in findings
+        ) else ("需确认" if findings else "无问题")
+        record["review_method"] = "确定性全量筛查；语义候选待表内确认"
+        records.append(record)
+        counts[kind] += 1
+
+    unknown = 0 if not in_body else 1
+    summary = {
+        "module": module,
+        "file": rel,
+        "sha256": file_hash,
+        "physical_lines": len(lines),
+        "active_records": len(records),
+        "reviewed_records": len(records),
+        "unreviewed_records": 0,
+        "issue_records": sum(bool(r["findings"]) for r in records),
+        "unknown_active_lines": unknown,
+        "counts": dict(counts),
+        "status": "通过" if unknown == 0 else "失败",
+    }
+    return records, summary
+
+def write_csv(path: Path, fields: list[str], rows: list[dict]) -> None:
+    with path.open("w", encoding="utf-8-sig", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore")
+        writer.writeheader()
+        writer.writerows(rows)
+
+def scan(args: argparse.Namespace) -> int:
+    repo = Path(args.repo).resolve()
+    output = Path(args.output)
+    if not output.is_absolute():
+        output = (repo / output).resolve()
+    output.mkdir(parents=True, exist_ok=True)
+
+    files = []
+    for root_arg in args.root:
+        root = (repo / root_arg).resolve() if not Path(root_arg).is_absolute() else Path(root_arg)
+        files.extend(sorted(root.rglob("*.yarn")))
+    files = sorted(set(files))
+    all_records, coverage = [], []
+    for file in files:
+        records, summary = parse_file(file, repo)
+        all_records.extend(records)
+        coverage.append(summary)
+
+    reviewed_ids = set()
+    if args.review_memory:
+        memory = json.loads(Path(args.review_memory).read_text(encoding="utf-8"))
+        for row in memory:
+            expected = transform_reviewed_line(row)
+            fingerprint = f'{row["文件"]}:{int(row["行号"])}:{expected}'
+            reviewed_ids.add(hashlib.sha1(fingerprint.encode("utf-8")).hexdigest()[:12])
+        for record in all_records:
+            if record["id"] in reviewed_ids:
+                record["findings"] = []
+                record["review_status"] = "无问题"
+                record["review_method"] = "人工审阅结论已应用"
+        issues_by_file = Counter(
+            record["file"] for record in all_records if record["findings"])
+        for summary in coverage:
+            summary["issue_records"] = issues_by_file[summary["file"]]
+
+    issues = []
+    for record in all_records:
+        if not record["findings"]:
+            continue
+        categories = [x["category"] for x in record["findings"]]
+        severity_order = {"明确错误": 3, "样式标记": 2, "建议优化": 1, "需确认": 0}
+        confidence_order = {"高": 2, "中": 1, "低": 0}
+        severity = max((x["severity"] for x in record["findings"]), key=lambda x: severity_order.get(x, 0))
+        confidence = max((x["confidence"] for x in record["findings"]), key=lambda x: confidence_order.get(x, 0))
+        issues.append({
+            "稳定ID": record["id"], "模块": record["module"], "文件": record["file"],
+            "节点": record["node"], "行号": record["line"], "LineTag": record["line_tag"],
+            "说话者": record["speaker"], "问题分类": categories, "严重度": severity,
+            "置信度": confidence, "原文": record["raw"],
+            "建议文本": next((x["suggested_text"] for x in record["findings"] if x["suggested_text"]), ""),
+            "理由": "\n".join(x["reason"] for x in record["findings"]),
+            "建议标记": "；".join(x["proposal"] for x in record["findings"] if x["proposal"]),
+            "检测来源": "脚本规则", "审核状态": "待确认",
+            "file_sha256": record["file_sha256"], "display_lines": record["display_lines"],
+        })
+
+    manifest = {
+        "version": 1,
+        "repo": str(repo),
+        "roots": args.root,
+        "files": coverage,
+    }
+    total = {
+        "files": len(files),
+        "physical_lines": sum(x["physical_lines"] for x in coverage),
+        "active_records": len(all_records),
+        "reviewed_records": len(all_records),
+        "unreviewed_records": 0,
+        "issue_records": len(issues),
+        "unknown_active_lines": sum(x["unknown_active_lines"] for x in coverage),
+        "review_method": "确定性全量筛查；语义候选待表内确认",
+    }
+    (output / "manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8")
+    (output / "summary.json").write_text(json.dumps(total, ensure_ascii=False, indent=2), encoding="utf-8")
+    with (output / "ledger.jsonl").open("w", encoding="utf-8") as f:
+        for row in all_records:
+            f.write(json.dumps(row, ensure_ascii=False) + "\n")
+    (output / "issues.json").write_text(json.dumps(issues, ensure_ascii=False, indent=2), encoding="utf-8")
+    issue_fields = ["稳定ID","模块","文件","节点","行号","LineTag","说话者","问题分类","严重度",
+                    "置信度","原文","建议文本","理由","建议标记","检测来源","审核状态"]
+    csv_issues = [{**x, "问题分类": "、".join(x["问题分类"])} for x in issues]
+    write_csv(output / "issues.csv", issue_fields, csv_issues)
+    coverage_fields = ["module","file","sha256","physical_lines","active_records","reviewed_records",
+                       "unreviewed_records","issue_records","unknown_active_lines","status"]
+    write_csv(output / "coverage.csv", coverage_fields, coverage)
+    print(json.dumps(total, ensure_ascii=False, indent=2))
+    return 0 if total["unknown_active_lines"] == total["unreviewed_records"] == 0 else 2
+
+def verify(args: argparse.Namespace) -> int:
+    output = Path(args.output)
+    summary = json.loads((output / "summary.json").read_text(encoding="utf-8"))
+    ledger_count = sum(1 for line in (output / "ledger.jsonl").read_text(encoding="utf-8").splitlines() if line)
+    issues = json.loads((output / "issues.json").read_text(encoding="utf-8"))
+    ok = (
+        summary["unknown_active_lines"] == 0
+        and summary["unreviewed_records"] == 0
+        and summary["active_records"] == ledger_count
+        and summary["issue_records"] == len(issues)
+    )
+    print(json.dumps({"ok": ok, "ledger_count": ledger_count, **summary}, ensure_ascii=False, indent=2))
+    return 0 if ok else 2
+
+def build_lark_payloads(args: argparse.Namespace) -> int:
+    output = Path(args.output)
+    target = output / "lark"
+    target.mkdir(parents=True, exist_ok=True)
+    issues = json.loads((output / "issues.json").read_text(encoding="utf-8"))
+    manifest = json.loads((output / "manifest.json").read_text(encoding="utf-8"))
+
+    issue_fields = ["稳定ID","模块","文件","节点","行号","LineTag","说话者","问题分类",
+                    "严重度","置信度","原文","建议文本","理由","建议标记","检测来源","审核状态"]
+    issue_rows = []
+    for row in issues:
+        issue_rows.append([
+            row["稳定ID"], row["模块"], row["文件"], row["节点"], row["行号"],
+            row["LineTag"], row["说话者"], "、".join(row["问题分类"]), row["严重度"],
+            row["置信度"], row["原文"], row["建议文本"], row["理由"], row["建议标记"],
+            row["检测来源"], row["审核状态"],
+        ])
+    for index in range(0, len(issue_rows), 200):
+        payload = {"fields": issue_fields, "rows": issue_rows[index:index + 200]}
+        (target / f"issues-{index // 200 + 1:03d}.json").write_text(
+            json.dumps(payload, ensure_ascii=False), encoding="utf-8")
+
+    coverage_fields = ["文件","模块","SHA256","物理行数","活动文本数","已审数",
+                       "未审数","问题数","未知活动行","校验状态"]
+    coverage_rows = [[
+        row["file"], row["module"], row["sha256"], row["physical_lines"],
+        row["active_records"], row["reviewed_records"], row["unreviewed_records"],
+        row["issue_records"], row["unknown_active_lines"], row["status"],
+    ] for row in manifest["files"]]
+    (target / "coverage.json").write_text(json.dumps(
+        {"fields": coverage_fields, "rows": coverage_rows}, ensure_ascii=False), encoding="utf-8")
+
+    style_fields = ["名称","类别","语义标记","默认参数","适用规则","示例"]
+    style_rows = [
+        ["自言自语/念文本/小声","语义样式","[[aside]]…[[/aside]]","#5F5F5F","只用于真实自言自语、阅读或低声表达；斜体不自动等同","[[aside]]我是不是忘了什么……[[/aside]]"],
+        ["游戏专有名词","语义样式","[[term]]…[[/term]]","#52B6FF","游戏世界独有的模块、机构、技术和概念","[[term]]UF模块[[/term]]"],
+        ["玩家重点信息","语义样式","[[key]]…[[/key]]","#FFB200","任务目标、必须寻找或操作的对象；优先于专名","寻找[[key]]色觉晶圆[[/key]]"],
+        ["逗号停顿","节奏效果","[[comma-pause]]","waitfor=0.2","只用于调整断行或明确需要节奏的逗号","别因为外面很冷，[[comma-pause]]<br>就逃回那个着火的房子。"],
+        ["ULM自训练进程","出现效果","[[ulm]]","vertexp d=0.05 bot","FP_Peipei2 每条活动 ULM自训练进程文本","[[ulm]]ULM自训练进程: 正在训练……"],
+        ["火山笑话","动态效果","[[joke]]…[[/joke]]","wiggle a=0.3 s=0.1","只包裹实际讲出的笑话，不包裹说明性对白","[[joke]]看医生其实是被医生看。[[/joke]]"],
+    ]
+    (target / "styles.json").write_text(json.dumps(
+        {"fields": style_fields, "rows": style_rows}, ensure_ascii=False), encoding="utf-8")
+    print(json.dumps({
+        "issue_batches": (len(issue_rows) + 199) // 200,
+        "issue_rows": len(issue_rows),
+        "coverage_rows": len(coverage_rows),
+        "style_rows": len(style_rows),
+    }, ensure_ascii=False, indent=2))
+    return 0
+
+def apply_approved(args: argparse.Namespace) -> int:
+    issues_path = Path(args.issues)
+    manifest = json.loads(Path(args.manifest).read_text(encoding="utf-8"))
+    repo = Path(manifest["repo"])
+    issues = json.loads(issues_path.read_text(encoding="utf-8"))
+    approved = [x for x in issues if x.get("审核状态") == "接受" and x.get("建议文本")]
+    hashes = {x["file"]: x["sha256"] for x in manifest["files"]}
+    grouped = defaultdict(list)
+    for row in approved:
+        grouped[row["文件"]].append(row)
+
+    prepared = {}
+    for rel, rows in grouped.items():
+        path = repo / rel
+        if sha256(path) != hashes.get(rel):
+            raise SystemExit(f"stale file hash: {rel}")
+        lines = path.read_text(encoding="utf-8-sig").splitlines(keepends=True)
+        for row in sorted(rows, key=lambda x: int(x["行号"])):
+            idx = int(row["行号"]) - 1
+            old = lines[idx].rstrip("\r\n")
+            if old != row["原文"]:
+                raise SystemExit(f"source mismatch: {rel}:{idx+1}")
+            new = row["建议文本"]
+            if LINE_TAG_RE.findall(old) != LINE_TAG_RE.findall(new):
+                raise SystemExit(f"line tag changed: {rel}:{idx+1}")
+            old_text = source_without_metadata(old)
+            new_text = source_without_metadata(new)
+            if len(re.split(r"(?i)<br\s*/?>", new_text)) > len(re.split(r"(?i)<br\s*/?>", old_text)):
+                raise SystemExit(f"display line added: {rel}:{idx+1}")
+            ending = lines[idx][len(lines[idx].rstrip("\r\n")):]
+            lines[idx] = new + ending
+        prepared[path] = "".join(lines)
+
+    for path, content in prepared.items():
+        fd, tmp = tempfile.mkstemp(prefix=path.name, dir=path.parent)
+        try:
+            with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
+                f.write(content)
+            os.replace(tmp, path)
+        finally:
+            if os.path.exists(tmp):
+                os.unlink(tmp)
+    print(json.dumps({"applied": len(approved), "files": len(prepared)}, ensure_ascii=False))
+    return 0
+
+def apply_reviewed(args: argparse.Namespace) -> int:
+    manifest = json.loads(Path(args.manifest).read_text(encoding="utf-8"))
+    reviews = json.loads(Path(args.reviews).read_text(encoding="utf-8"))
+    repo = Path(manifest["repo"])
+    hashes = {x["file"]: x["sha256"] for x in manifest["files"]}
+    allowed = {REVIEW_ACCEPTED, REVIEW_KEEP, REVIEW_FOLLOW_NOTE}
+    decisions = Counter(review_choice(row.get("审阅结论")) for row in reviews)
+    invalid = sorted(choice for choice in decisions if choice not in allowed)
+    if invalid:
+        raise SystemExit(f"unresolved review decisions: {invalid}")
+
+    grouped = defaultdict(list)
+    changes = []
+    unchanged_actionable = []
+    for row in reviews:
+        decision = review_choice(row.get("审阅结论"))
+        new = transform_reviewed_line(row)
+        if decision != REVIEW_KEEP and new == row["原文"]:
+            unchanged_actionable.append(f'{row["文件"]}:{row["行号"]} ({decision})')
+        if new != row["原文"]:
+            grouped[row["文件"]].append((row, new))
+            changes.append({
+                "稳定ID": row["稳定ID"],
+                "文件": row["文件"],
+                "行号": row["行号"],
+                "问题分类": row["问题分类"],
+                "审阅结论": decision,
+                "审阅备注": row.get("审阅备注") or "",
+                "原文": row["原文"],
+                "应用文本": new,
+            })
+    if unchanged_actionable:
+        raise SystemExit("actionable reviews produced no change:\n" + "\n".join(unchanged_actionable))
+
+    prepared = {}
+    for rel, rows in grouped.items():
+        path = repo / rel
+        if sha256(path) != hashes.get(rel):
+            raise SystemExit(f"stale file hash: {rel}")
+        lines = path.read_text(encoding="utf-8-sig").splitlines(keepends=True)
+        before_tags = LINE_TAG_RE.findall("".join(lines))
+        for row, new in sorted(rows, key=lambda item: int(item[0]["行号"])):
+            idx = int(row["行号"]) - 1
+            old = lines[idx].rstrip("\r\n")
+            if old != row["原文"]:
+                raise SystemExit(f"source mismatch: {rel}:{idx + 1}")
+            if LINE_TAG_RE.findall(old) != LINE_TAG_RE.findall(new):
+                raise SystemExit(f"line tag changed: {rel}:{idx + 1}")
+            _, old_body, _ = dialogue_parts(old)
+            _, new_body, _ = dialogue_parts(new)
+            old_width = visible_width(strip_markup(old_body))
+            allowed_lines = max(
+                len(re.split(r"(?i)<br\s*/?>", old_body)),
+                int((old_width + MANUAL_BREAK_WIDTH - 0.001) // MANUAL_BREAK_WIDTH),
+            )
+            new_lines = len(re.split(r"(?i)<br\s*/?>", new_body))
+            if new_lines > allowed_lines:
+                raise SystemExit(
+                    f"display line estimate increased: {rel}:{idx + 1} "
+                    f"({new_lines} > {allowed_lines})"
+                )
+            for marker in ("aside", "term", "key", "joke"):
+                if new.count(f"[[{marker}]]") != new.count(f"[[/{marker}]]"):
+                    raise SystemExit(f"unbalanced semantic marker {marker}: {rel}:{idx + 1}")
+            ending = lines[idx][len(lines[idx].rstrip("\r\n")):]
+            lines[idx] = new + ending
+        content = "".join(lines)
+        if LINE_TAG_RE.findall(content) != before_tags:
+            raise SystemExit(f"file line tag set/order changed: {rel}")
+        prepared[path] = content
+
+    report = {
+        "review_records": len(reviews),
+        "decisions": dict(decisions),
+        "changed_records": len(changes),
+        "changed_files": len(prepared),
+        "dry_run": not args.apply,
+        "changes": changes,
+    }
+    if args.report:
+        Path(args.report).write_text(
+            json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
+    if args.apply:
+        for path, content in prepared.items():
+            fd, tmp = tempfile.mkstemp(prefix=path.name, dir=path.parent)
+            try:
+                with os.fdopen(fd, "w", encoding="utf-8", newline="") as f:
+                    f.write(content)
+                os.replace(tmp, path)
+            finally:
+                if os.path.exists(tmp):
+                    os.unlink(tmp)
+    print(json.dumps({
+        key: value for key, value in report.items() if key != "changes"
+    }, ensure_ascii=False, indent=2))
+    return 0
+
+def self_test() -> int:
+    import unittest
+
+    class AuditTests(unittest.TestCase):
+        def test_markup_and_width(self):
+            self.assertEqual(strip_markup("[[vertexp d=0.05 bot]]<color=\\#fff>文字</color>"), "文字")
+            self.assertEqual(visible_width("中文ab"), 3.0)
+
+        def test_break_rule(self):
+            rec = {
+                "text": "很多时候，比起幸福更难以让人离<br>开的，是熟悉。",
+                "visible_text": "很多时候，比起幸福更难以让人离开的，是熟悉。",
+                "speaker": "me", "source_text": "", "module": "FP_Test",
+            }
+            self.assertIn("换行", [x["category"] for x in detect(rec)])
+
+        def test_annotated_rules(self):
+            ulm = {"text":"正在训练","visible_text":"正在训练","speaker":"ULM自训练进程",
+                   "source_text":"","module":"FP_Peipei2"}
+            self.assertIn("ULM效果", [x["category"] for x in detect(ulm)])
+            task = {"text":"寻找模块","visible_text":"寻找模块","speaker":"Task",
+                    "source_text":"","module":"FP_Test"}
+            self.assertNotIn("重要信息", [x["category"] for x in detect(task)])
+
+        def test_reviewed_break_preserves_existing_wait(self):
+            text = "很多时候，比起幸福更难以让人离开的，是熟悉。"
+            changed = add_comfortable_breaks(text)
+            self.assertIn("，[[comma-pause]]<br>", changed)
+            with_wait = "别因为外面很冷，<waitfor=0.1>就逃回那个着火的房子。"
+            changed_wait = add_comfortable_breaks(with_wait)
+            self.assertIn("，<waitfor=0.1><br>", changed_wait)
+            self.assertNotIn("[[comma-pause]]", changed_wait)
+
+        def test_review_note_overrides_default_style(self):
+            row = {
+                "原文": "me: 去检查*情绪模块*。 #line:abc",
+                "审阅结论": [REVIEW_FOLLOW_NOTE],
+                "审阅备注": "用黄字，并且去掉“*”",
+                "问题分类": "专有名词",
+            }
+            changed = transform_reviewed_line(row)
+            self.assertIn("[[key]]情绪模块[[/key]]", changed)
+            self.assertNotIn("*", changed)
+            self.assertEqual(LINE_TAG_RE.findall(changed), ["abc"])
+
+    suite = unittest.defaultTestLoader.loadTestsFromTestCase(AuditTests)
+    result = unittest.TextTestRunner(verbosity=2).run(suite)
+    return 0 if result.wasSuccessful() else 2
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Deterministic Yarn text coverage and review ledger")
+    sub = parser.add_subparsers(dest="command", required=True)
+    p_scan = sub.add_parser("scan")
+    p_scan.add_argument("--repo", default=".")
+    p_scan.add_argument("--root", action="append", required=True)
+    p_scan.add_argument("--output", required=True)
+    p_scan.add_argument(
+        "--review-memory",
+        help="可选的飞书审阅导出；已应用或明确保留的记录不再重复报告")
+    p_scan.set_defaults(func=scan)
+    p_verify = sub.add_parser("verify")
+    p_verify.add_argument("--output", required=True)
+    p_verify.set_defaults(func=verify)
+    p_apply = sub.add_parser("apply")
+    p_apply.add_argument("--issues", required=True)
+    p_apply.add_argument("--manifest", required=True)
+    p_apply.set_defaults(func=apply_approved)
+    p_apply_review = sub.add_parser("apply-review")
+    p_apply_review.add_argument("--reviews", required=True)
+    p_apply_review.add_argument("--manifest", required=True)
+    p_apply_review.add_argument("--report")
+    p_apply_review.add_argument("--apply", action="store_true")
+    p_apply_review.set_defaults(func=apply_reviewed)
+    p_lark = sub.add_parser("lark-payloads")
+    p_lark.add_argument("--output", required=True)
+    p_lark.set_defaults(func=build_lark_payloads)
+    p_test = sub.add_parser("self-test")
+    p_test.set_defaults(func=lambda _: self_test())
+    return parser
+
+def main() -> int:
+    args = build_parser().parse_args()
+    return args.func(args)
+
+if __name__ == "__main__":
+    sys.exit(main())