From 75de932ea03821004ad3b397afd007bd93ac822b Mon Sep 17 00:00:00 2001 From: bottlefish <781230111@qq.com> Date: Sun, 26 Jul 2026 23:43:04 +0800 Subject: [PATCH] =?UTF-8?q?feat(tools):=20=E6=B7=BB=E5=8A=A0=20Yarn=20?= =?UTF-8?q?=E6=96=87=E6=9C=AC=E5=AE=A1=E6=A0=A1=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Cursor --- Tools/yarn_text_audit.py | 812 +++++++++++++++++++++++++++++++++++++++ 1 file changed, 812 insertions(+) create mode 100644 Tools/yarn_text_audit.py diff --git a/Tools/yarn_text_audit.py b/Tools/yarn_text_audit.py new file mode 100644 index 000000000..a4ab987d9 --- /dev/null +++ b/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"(?>(?:\s*//.*)?\s*$") +INLINE_IF_RE = re.compile(r"<>") +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)", 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"]+>", text[insert_at:], flags=re.IGNORECASE) + if wait: + insert_at += wait.end() + insertion = "
" + else: + insertion = "[[comma-pause]]
" + 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"(? 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)", 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 " 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)", 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]]
就逃回那个着火的房子。"], + ["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)", new_text)) > len(re.split(r"(?i)", 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)", old_body)), + int((old_width + MANUAL_BREAK_WIDTH - 0.001) // MANUAL_BREAK_WIDTH), + ) + new_lines = len(re.split(r"(?i)", 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]]文字"), "文字") + self.assertEqual(visible_width("中文ab"), 3.0) + + def test_break_rule(self): + rec = { + "text": "很多时候，比起幸福更难以让人离
开的，是熟悉。", + "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]]
", changed) + with_wait = "别因为外面很冷，就逃回那个着火的房子。" + changed_wait = add_comfortable_breaks(with_wait) + self.assertIn("，
", 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())