From 60f22e66600372ca2eaf6ce7abca9db8c970e539 Mon Sep 17 00:00:00 2001 From: simonkoson <28867558@qq.com> Date: Wed, 17 Jun 2026 13:27:41 +0800 Subject: [PATCH] =?UTF-8?q?P3-C3:=20=E4=BA=A4=E5=8F=89=E5=A4=8D=E5=AE=A1?= =?UTF-8?q?=20fusion=5Freview.py=20+=20doco=20fuse=20=E5=91=BD=E4=BB=A4?= =?UTF-8?q?=EF=BC=8C=E4=BA=A7=E8=9E=8D=E5=90=88B=E7=A8=BF743=E8=A1=8C+fusi?= =?UTF-8?q?on=5Freview=E7=95=99=E7=97=95=EF=BC=8C=E6=9B=B4=E6=96=B0CLAUDE.?= =?UTF-8?q?md=E4=BA=A4=E6=8E=A5C4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- doco/CLAUDE.md | 31 +- doco/src/doco/cli.py | 69 +++++ doco/src/doco/fusion_review.py | 534 +++++++++++++++++++++++++++++++++ 3 files changed, 623 insertions(+), 11 deletions(-) create mode 100644 doco/src/doco/fusion_review.py diff --git a/doco/CLAUDE.md b/doco/CLAUDE.md index 231342e..9b420b0 100644 --- a/doco/CLAUDE.md +++ b/doco/CLAUDE.md @@ -17,7 +17,7 @@ ## 🔖 状态栏 (STATUS — 每次结束 session 前必须更新这三行) - **最后更新**：账号A ｜ 2026-06-17 -- **当前状态一句话**：P3-C2 讯飞 ASR 跑通并收口（真转写310句带时间戳，热词命中良好），本轮已 git 提交推 main（9340edc）；**B稿_v2.txt 已就位入库，C3 卡点解除，下一步可直接开 C3**。 +- **当前状态一句话**：P3-C3 交叉复审完工（融合B稿743行+fusion_review 5条留痕落盘），产品文件已就位，下一步 C4 语义融合→融合A稿 docx。 - **下一个动手的人从这里开始**：见下方「⏩ 交接备注」 --- @@ -62,26 +62,29 @@ - `src/doco/term_extract.py` — C1：规则层 + AI 层提取 + 词典累积 + 热词清洗 - `src/doco/asr_adapter.py` — C2 讯飞 ASR 适配层（已完工）：`get_hot_words(episode_id)` 读热词表、`transcribe()` 返回 `(句子, raw)`、`write_asr_result()` 落 `asr_v2_timed.txt` + `asr_result_raw.json` - `src/doco/video_split.py` — P1：含 `extract_audio()`（16kHz/单声道/16bit WAV），C2 直接复用，不另写 ffmpeg - - `src/doco/cli.py` — 命令入口（split / terms / asr） + - `src/doco/fusion_review.py` — C3：B稿v2 ⊕ ASR 交叉复审 → 融合B稿(743行) + fusion_review.csv + - `src/doco/cli.py` — 命令入口（split / terms / asr / fuse） - `doco/data/term_dict.json` — 累积词典（当前 110 条） - `doco/programs//` — 每期产物（热词表、各阶段输出） - **常用命令**： - C1（已实现）：`doco terms --episode-id --a-script [--no-ai]` - C2（已实现）：`doco asr --episode-id --input-video <绝对路径mp4> --output-dir <绝对路径 doco/programs/>`（`--skip-asr` 只分离不转写）。**务必传绝对 `--output-dir`，否则落到当前工作目录的 `programs/` 会与 doco 产物分家。** - - C3/C4：待实现 + - C3（已实现）：`doco fuse --episode-id [--no-ai] [--batch-size 35]` + - C4：待实现 - **环境变量 / 密钥**：只放 `doco/.env`（已在 `.gitignore`）。需要的 key：`DEEPSEEK_API_KEY` / `DEEPSEEK_BASE_URL` / `DEEPSEEK_MODEL`、`XFYUN_APP_ID` / `XFYUN_SECRET_KEY`。**代码与对话里只引用变量名，绝不出现真值。** --- ## 3. 当前进度（核心交接区，以最新快照为准） -- **正在做**：C2 已收口并推 main，准备进 C3（B稿v2 ⊕ ASR 交叉复审）。 -- **卡点 / 待解**：无。C3 三路输入均已就位于 `doco/programs/ep001_20260612_fangkong_fandao/`：`B稿_v2.txt`（743行，P2）、`asr_v2_timed.txt`（310句，C2）、A稿 docx。 +- **正在做**：C3 已收口，C4 待开工。 +- **卡点 / 待解**：无。 --- ## 4. 已完成（只追加，最新在最上） +- [2026-06-17 ｜账号A] **P3-C3 收口**：`fusion_review.py` 逐行复审 B稿v2 ⊕ ASR（22 批×35行=743行），产 `融合B稿.txt`（行数/时间戳零偏移）+ `fusion_review.csv`（5条真OCR纠错留痕）。关键机制：断点缓存 `.c3_cache/batch_N.json` 可复用；硬校验行数/时间戳/change_type枚举不过不写文件；后处理修正 `final_text==B原文` 被 LLM 误标为 `minor_edit` 的行（19→5）。`--no-ai` 全 unchanged 可验证管道。 - [2026-06-17 ｜账号A] **P3-C2 收口**：密钥外置三验收点全过（硬编码已删/读 .env/已 gitignore）+ 旧 key 已轮换；`asr_adapter.py` 并入讯飞上传·轮询·解析，复用 P1 `extract_audio` 分离 16kHz/单声道/16bit WAV；新增 `doco asr` 命令读 C1 64条热词。**真转写《现代防空反导大对决》27分钟音频 → 310句带时间戳**，萨德×6/塔米尔/爱国者/箭二箭三/见证者等术语命中良好，raw json 730KB 落盘。（顺带修 `extract_audio` 在 Windows 的 `text=True` 解码崩溃。） - [2026-06-16 ｜账号A] **P3-C1 收口**：A稿→规则+DeepSeek AI 提取→累积词典110条→热词清洗→本期热词表64条；AI 44 候选已逐个回查 A 稿原文无幻觉，算术对账一致。 - [P2 ｜本轮之前] 本地 OCR（DeepSeek-OCR/Ollama）+ 扒词流水线，产出 **B稿_v2.txt 743 行**。 @@ -91,8 +94,8 @@ ## 5. 待办（按优先级，做完打勾） -- [ ] **C3（下一步，输入已齐）**：B稿v2 ⊕ 新ASR 交叉复审 → 融合B稿（保持743行碎句粒度 + 密集时间戳原样，ASR只复审纠错不改行结构）+ `fusion_review.csv`。 -- [ ] **C4**：融合B稿 + A稿语义对齐 → 融合A稿（公文 docx，用 docx skill，GB/T 9704 格式）。 +- [x] **C3**：B稿v2 ⊕ ASR 交叉复审 → 融合B稿(743行时间戳零偏移) + fusion_review.csv(5条留痕)。 +- [ ] **C4**（下一步）**：融合B稿 + A稿语义对齐 → 融合A稿（公文 docx，GB/T 9704 格式）。按 A稿分段把融合B稿碎句归拢、套 docx 格式，C4 语义融合可开 thinking。 - [x] **C2** 讯飞 ASR 适配层（密钥外置 + asr_adapter 并入 + `doco asr` 命令 + 真转写310句） - [x] P3-C1 术语提取 + 词典 + 热词表 - [x] P2 本地 OCR + B稿v2 @@ -113,6 +116,9 @@ - [2026-06-15] **B稿v2 权威升级**：屏幕术语/型号/番号 **B稿v2 ≈ A稿 并列权威**（B稿是屏幕实打的字，更贴近播出）。口语/语序/语气信 ASR；书面结构/分段信 A稿；同音事实错（美以→美伊）用 A稿/B稿v2 覆盖 ASR。 - [设计期] **分层不变**：OCR 错字纠正、术语统一、时间戳对齐走**规则层**（确定性）；段落归属、整段删除判定、编导笔误识别走 **AI 层**（需语义判断）。 - [设计期] **顺序约束**：术语提取（C1，产热词）必须在 ASR（C2）之前——已成立。 +- [2026-06-17] **`fusion_review.csv` 定位 = 留痕档案（给爱德华/复核），不是编导读物。** 编导读的是 C4 docx。CSV 可读性不优化，只看 final_text 准不准。 +- [2026-06-17] **`final_text==B原文` 强制归 `unchanged` 不进 review。** LLM 有时把"考虑后决定保留 B 稿原文"标成 `minor_edit`，但实际没改字。`fusion_review.py` 的 `_normalize_unchanged_when_no_edit()` 在写文件前修正（review 19→5）。 +- [2026-06-17] **`change_type` 的 `minor_edit`/`editor_typo` 边界 LLM 把握不稳**（OCR 错字两类混标）。仅 metadata 标签不一致，`final_text` 全对，不影响融合B稿正文。本轮决定忽略；若 C4/爱德华按 `change_type` 做逻辑分流，届时在 prompt 里收紧两个枚举的边界定义。 - [工程纪律] 各阶段解耦、中间产物落缓存、可断点续跑、可单独重跑；长批处理放独立终端。 --- @@ -121,10 +127,13 @@ > 下次开工读完这段应能 0 摩擦续上。接手后可清空重写。 -- **第一句话该干的**：C2 已收口推 main，C3 三路输入已齐（`B稿_v2.txt` 743行 / `asr_v2_timed.txt` 310句 / A稿 docx，都在 `doco/programs/ep001_20260612_fangkong_fandao/`）。直接给 Cline 出 C3 的 ACT 指令开工。 -- **C3 目标**：B稿v2 ⊕ asr_v2_timed 交叉复审 → 融合B稿（**保持 743 行碎句 + 密集时间戳原样，ASR 只复审纠错不改行结构**）+ `fusion_review.csv`。权威优先级见关键决策「B稿v2 权威升级」那条。 -- **重跑真转写命令**（复用已分离 WAV，27分钟音频讯飞轮询等几分钟别中断）： - `doco asr --episode-id ep001_20260612_fangkong_fandao --input-video "E:\tps-dashboard\doco\programs\ep001_20260612_fangkong_fandao\现代防空反导大对决VA0.mp4" --output-dir "E:\tps-dashboard\doco\programs\ep001_20260612_fangkong_fandao"` +- **第一句话该干的**：C3 已完成，产物就位（`融合B稿.txt` 743行 + `fusion_review.csv` 5条留痕，都在 `doco/programs/ep001_20260612_fangkong_fandao/`）。直接开 C4。 +- **C4 目标**：融合B稿（743行碎句）+ A稿 docx 语义对齐 → **融合A稿 docx**（公文格式，GB/T 9704）。按 A稿分段（【导视】【主持人N】【解说N】【专家N】【隔断】）把融合B稿碎句归拢到对应段落，套 docx 模板格式输出。C4 语义融合可开 thinking（非抽取类任务，质量优先）。一致性约束：融合A稿由融合B稿生成，不是拿 A 稿改。 +- **C4 输入**（均就位）： + - `programs/ep001_20260612_fangkong_fandao/融合B稿.txt`（743行，`[XmYs] 文本`） + - `programs/ep001_20260612_fangkong_fandao/现代防空反导大对决A稿.docx`（A稿，python-docx 可读） + - `programs/ep001_20260612_fangkong_fandao/fusion_review.csv`（5条改动留痕，可选参考） +- **C3 重跑命令**：`doco fuse --episode-id ep001_20260612_fangkong_fandao`（`--no-ai` 全 unchanged 验证管道，缓存 `.c3_cache/` 复用无需重跑 AI） - 通哥连轴转了几轮，回复要紧凑。 - **已知坑（务必记住）**： 1. **A稿真实约 6500–7300 汉字**（栏目常态 6000–8000），**不是 2.2 万**——那是用 `wc -m` 把 UTF-8 字节当字数的乌龙（一个汉字 3 字节）。docx 无文本框无表格，`python-docx` 提取的 ~7.5k 字符就是全文。 diff --git a/doco/src/doco/cli.py b/doco/src/doco/cli.py index 4bf4280..d82e1f0 100644 --- a/doco/src/doco/cli.py +++ b/doco/src/doco/cli.py @@ -19,6 +19,9 @@ from .term_extract import run_terms # P3 C2 讯飞 ASR from .asr_adapter import get_hot_words, transcribe, write_asr_result +# P3 C3 B稿⊕ASR 交叉复审融合 +from .fusion_review import run_fusion + @click.group() @click.version_option(version="0.1.0") @@ -216,6 +219,72 @@ def terms( sys.exit(1) +@main.command("fuse") +@click.option("--episode-id", required=True, help="节目 ID,如 ep001_20260612_fangkong_fandao") +@click.option( + "--output-dir", + default=None, + type=click.Path(), + help="输出目录(默认 programs//)", +) +@click.option( + "--no-ai", + is_flag=True, + default=False, + help="跳过 LLM 只跑规则层(=全 unchanged)", +) +@click.option( + "--batch-size", + default=35, + type=int, + help="每批送审行数(默认 35)", +) +def fuse( + episode_id: str, + output_dir: str, + no_ai: bool, + batch_size: int, +): + """ + P3 C3: B稿 ⊕ ASR 交叉复审融合 + + 逐行复审 B稿(屏幕字幕OCR),以 ASR(口语转写)为上下文参考, + 只做纠错,绝不合并行、不拆行、不增删行、不改时间戳。 + + --no-ai: 跳过 LLM,全 unchanged(验证管道) + --batch-size: 每批送审行数,默认 35 + + 产物: + - 融合B稿.txt(与 B稿_v2 逐行时间戳一致) + - fusion_review.csv(仅含 change_type≠unchanged 或 confidence<0.8 的行) + """ + if output_dir is None: + out_dir = Path("programs") / episode_id + else: + out_dir = Path(output_dir) + + click.echo(f"[doco fuse] episode_id={episode_id}") + click.echo(f"[doco fuse] output_dir={out_dir}") + click.echo(f"[doco fuse] no_ai={no_ai}") + click.echo(f"[doco fuse] batch_size={batch_size}") + + try: + stats = run_fusion( + episode_id=episode_id, + output_dir=str(out_dir), + no_ai=no_ai, + batch_size=batch_size, + ) + click.echo(f"[ok] 总行数: {stats['total_lines']}") + click.echo(f"[ok] 各 change_type 计数: {stats['change_counts']}") + click.echo(f"[ok] 进 review 行数: {stats['review_lines']}") + click.echo(f"[ok] 融合B稿: {stats['fused_path']}") + click.echo(f"[ok] review CSV: {stats['csv_path']}") + except Exception as e: + click.echo(f"[error] {e}", err=True) + sys.exit(1) + + @main.command("asr") @click.option("--episode-id", required=True, help="节目 ID,如 ep001_20260612_fangkong_fandao") @click.option( diff --git a/doco/src/doco/fusion_review.py b/doco/src/doco/fusion_review.py new file mode 100644 index 0000000..3ad1103 --- /dev/null +++ b/doco/src/doco/fusion_review.py @@ -0,0 +1,534 @@ +# -*- coding: utf-8 -*- +""" +C3: B稿v2 ⊕ ASR 交叉复审 → 融合B稿(743行) + fusion_review.csv +============================================================= +职责:逐行复审 B稿(屏幕字幕OCR),以 ASR(口语转写)为上下文参考, +只做纠错,严禁改行数/时间戳。 +""" + +import json +import re +import sys +from pathlib import Path +from typing import List, Dict, Optional + +from .llm import chat + +# -------------------------------------------------------------------------- +# 常量 +# -------------------------------------------------------------------------- + +CHANGE_TYPE_ENUM = frozenset( + { + "unchanged", + "minor_edit", + "term_normalize", + "rewrite_large", + "segment_delete", + "segment_add", + "editor_typo", + } +) + +SYSTEM_PROMPT = """你是《军事科技》专题片文稿校审员。给你 B稿(屏幕字幕OCR,逐行碎句,带时间戳) 和对应的 ASR(口语转写)。 +你的任务:逐行复审 B稿,只做纠错,绝不合并行、不拆行、不增删行、不改时间戳。 +权威优先级: +- 屏幕术语/型号/番号(箭-3/萨德/见证者-136等): B稿为准(屏幕实打的字) +- B稿明显是OCR错字而ASR是对的: 用ASR覆盖 +- 同音事实错(如"美以"vs"美伊"): 以书面规范为准,存疑进review +- 一两个字的等价差异(的/地、啊等语气): 算 unchanged,不要改 +每行输出: line_no, final_text(纠错后,默认等于B原文), change_type(7选1), confidence(0~1), reason(简短,unchanged时留空) +只返回JSON数组,不要任何解释文字。 +change_type枚举: unchanged/minor_edit/term_normalize/rewrite_large/segment_delete/segment_add/editor_typo""" + + +# -------------------------------------------------------------------------- +# 1. 解析带时间戳的行 +# -------------------------------------------------------------------------- + +def parse_timed_lines(path) -> List[dict]: + r""" + 解析 "[XmYs] 文本" → [{"idx":int, "ts_raw":"0m8s", "ts_sec":8, "text":"导弹呼啸而过"}] + 正则: ^\[(\d+)m(\d+)s\]\s*(.*)$ ; ts_sec = m*60+s + 解析失败的行要抛异常并打印行号,不许静默跳过 + """ + p = Path(path) + if not p.exists(): + raise FileNotFoundError(f"文件不存在: {path}") + + pattern = re.compile(r"^\[(\d+)m(\d+)s\]\s*(.*)$") + lines_raw = p.read_text(encoding="utf-8").splitlines() + result = [] + + for idx, line in enumerate(lines_raw, start=1): + line = line.strip() + if not line: + continue # 跳过空行 + m = pattern.match(line) + if not m: + raise ValueError( + f"行 {idx} 解析失败,无法匹配时间戳格式: {repr(line[:120])}\n" + f"文件: {path}" + ) + minutes = int(m.group(1)) + seconds = int(m.group(2)) + ts_raw = f"{minutes}m{seconds}s" + ts_sec = minutes * 60 + seconds + text = m.group(3).strip() + result.append( + { + "idx": idx, + "ts_raw": ts_raw, + "ts_sec": ts_sec, + "text": text, + } + ) + + return result + + +# -------------------------------------------------------------------------- +# 2. 对齐 ASR 上下文 +# -------------------------------------------------------------------------- + +def align_asr_context(b_lines: List[dict], asr_lines: List[dict]) -> List[str]: + """ + 为每个 B 行找时间窗内的 ASR 上下文(用于喂 LLM) + 规则: 取 ts_sec 落在 [b_ts-3, b_next_ts+3] 区间的 ASR 句拼接; + 边界用前后各扩 1 句兜底。返回与 b_lines 等长的 context 列表 + """ + n = len(b_lines) + contexts = [] + + # 预计算 B 行的时间窗: [b[i].ts_sec - 3, b[i+1].ts_sec + 3] + # 最后一行用 b[i].ts_sec + 10 作为上界 + windows = [] + for i, bl in enumerate(b_lines): + lo = bl["ts_sec"] - 3 + if i + 1 < n: + hi = b_lines[i + 1]["ts_sec"] + 3 + else: + hi = bl["ts_sec"] + 10 + windows.append((lo, hi)) + + asr_count = len(asr_lines) + + for i, (lo, hi) in enumerate(windows): + # 找到落在窗口内的 ASR 句索引 + hit_indices = [] + for j, al in enumerate(asr_lines): + if lo <= al["ts_sec"] <= hi: + hit_indices.append(j) + + if not hit_indices: + # 无命中:取距离最近的 1 句 + best_j = None + best_dist = float("inf") + mid_ts = (lo + hi) / 2 + for j, al in enumerate(asr_lines): + dist = abs(al["ts_sec"] - mid_ts) + if dist < best_dist: + best_dist = dist + best_j = j + if best_j is not None: + start_j = max(0, best_j - 1) + end_j = min(asr_count - 1, best_j + 1) + else: + start_j = 0 + end_j = 0 + else: + # 命中句的范围 + 前后各扩 1 + start_j = max(0, hit_indices[0] - 1) + end_j = min(asr_count - 1, hit_indices[-1] + 1) + + # 拼接 [start_j, end_j] 的 ASR 文本 + selected = asr_lines[start_j : end_j + 1] + context = " ".join(s["text"] for s in selected) + contexts.append(context) + + assert len(contexts) == n, f"context 列表长度 {len(contexts)} != B 稿行数 {n}" + return contexts + + +# -------------------------------------------------------------------------- +# 3. 构造 Prompt +# -------------------------------------------------------------------------- + +def build_prompt(batch_b: List[dict], batch_ctx: List[str]) -> List[dict]: + """ + 构造 messages,见下方"四、Prompt 模板" + """ + assert len(batch_b) == len(batch_ctx), ( + f"batch_b({len(batch_b)}) 与 batch_ctx({len(batch_ctx)}) 长度不一致" + ) + + user_lines = [] + for bl, ctx in zip(batch_b, batch_ctx): + line_no = bl["idx"] + b_text = bl["text"] + asr_text = ctx if ctx else "(无ASR上下文)" + user_lines.append( + f"[行{line_no}] B稿: \"{b_text}\" ASR上下文: \"{asr_text}\"" + ) + + user_content = "\n".join(user_lines) + + messages = [ + {"role": "system", "content": SYSTEM_PROMPT}, + {"role": "user", "content": user_content}, + ] + return messages + + +# -------------------------------------------------------------------------- +# 4. 单批复审 +# -------------------------------------------------------------------------- + +def review_batch( + batch_b: List[dict], batch_ctx: List[str], no_ai: bool = False +) -> List[dict]: + """ + no_ai=True: 直接回填 unchanged(final_text=b原文, change_type="unchanged", confidence=1.0) + no_ai=False: 调 llm.chat(messages, thinking=False, max_tokens=4000, temperature=0.0) + 解析返回 JSON 数组; 每元素 {line_no, final_text, change_type, confidence, reason} + 返回标准化记录列表 + """ + if no_ai: + records = [] + for bl in batch_b: + records.append( + { + "line_no": bl["idx"], + "final_text": bl["text"], + "change_type": "unchanged", + "confidence": 1.0, + "reason": "", + } + ) + return records + + # ---- AI 路径 ---- + messages = build_prompt(batch_b, batch_ctx) + + try: + raw_response = chat( + messages, + thinking=False, + max_tokens=4000, + temperature=0.0, + ) + except Exception as e: + print( + f"[fusion_review] LLM 调用失败,回退为 unchanged 批次: {e}", + file=sys.stderr, + ) + # 回退:全部 unchanged + records = [] + for bl in batch_b: + records.append( + { + "line_no": bl["idx"], + "final_text": bl["text"], + "change_type": "unchanged", + "confidence": 1.0, + "reason": f"LLM调用失败回退: {str(e)[:80]}", + } + ) + return records + + # 解析 JSON + parsed = _parse_llm_json_response(raw_response, len(batch_b)) + + # 标准化并校验 + records = [] + for item in parsed: + line_no = item.get("line_no") + final_text = item.get("final_text", "") + change_type = item.get("change_type", "unchanged") + confidence = item.get("confidence", 1.0) + reason = item.get("reason", "") + + # 校验 change_type + if change_type not in CHANGE_TYPE_ENUM: + original_ct = change_type + print( + f"[fusion_review] 行 {line_no} 非法 change_type='{original_ct}', 强制改为 unchanged", + file=sys.stderr, + ) + change_type = "unchanged" + final_text = "" # 下面会回填 + reason = f"LLM返回非法change_type({original_ct}),回退unchanged" + + # 如果 change_type 被改为 unchanged 但 final_text 为空,回填 B 原文 + if change_type == "unchanged" and not final_text: + # 从 batch_b 找回原文 + for bl in batch_b: + if bl["idx"] == line_no: + final_text = bl["text"] + break + + records.append( + { + "line_no": line_no, + "final_text": final_text, + "change_type": change_type, + "confidence": float(confidence), + "reason": reason or "", + } + ) + + return records + + +def _parse_llm_json_response(raw: str, expected_len: int) -> List[dict]: + """解析 LLM 返回的 JSON,处理 markdown code fences 等常见包装。""" + text = raw.strip() + + # 去掉可能的 markdown code fences + if text.startswith("```"): + lines = text.splitlines() + # 去掉第一行 ```json 或 ``` + if lines and lines[0].startswith("```"): + lines = lines[1:] + # 去掉最后一行 ``` + if lines and lines[-1].strip() == "```": + lines = lines[:-1] + text = "\n".join(lines).strip() + + try: + result = json.loads(text) + except json.JSONDecodeError as e: + raise ValueError( + f"LLM 返回 JSON 解析失败: {e}\n" + f"原始响应前 500 字符: {raw[:500]}" + ) + + if not isinstance(result, list): + raise ValueError( + f"LLM 返回不是 JSON 数组,类型为 {type(result).__name__}" + ) + + if len(result) != expected_len: + raise ValueError( + f"LLM 返回 {len(result)} 条记录,期望 {expected_len} 条。" + f"该批次将回退为 unchanged 并重新请求。" + ) + + return result + + +# -------------------------------------------------------------------------- +# 5. 主流程 +# -------------------------------------------------------------------------- + +def run_fusion( + episode_id: str, + output_dir: str, + no_ai: bool = False, + batch_size: int = 35, +) -> dict: + """ + 主流程: + 1. 读 output_dir/B稿_v2.txt → b_lines（断言行数>0） + 2. 读 output_dir/asr_v2_timed.txt → asr_lines + 3. align_asr_context 生成等长 context + 4. 按 batch_size 分块；每块结果落缓存,已存在则复用（断点续跑） + 5. 逐块 review_batch,汇总所有记录 + 6. 硬校验（任一不过就 raise,不写出文件） + 7. 写 output_dir/融合B稿.txt + 8. 写 output_dir/fusion_review.csv + 9. 返回统计 dict + """ + out_dir = Path(output_dir) + out_dir.mkdir(parents=True, exist_ok=True) + + b_path = out_dir / "B稿_v2.txt" + asr_path = out_dir / "asr_v2_timed.txt" + + if not b_path.exists(): + raise FileNotFoundError(f"B稿_v2.txt 不存在: {b_path}") + if not asr_path.exists(): + raise FileNotFoundError(f"asr_v2_timed.txt 不存在: {asr_path}") + + # Step 1: 解析 B 稿 + b_lines = parse_timed_lines(b_path) + assert len(b_lines) > 0, f"B稿_v2.txt 解析后为空: {b_path}" + + # Step 2: 解析 ASR + asr_lines = parse_timed_lines(asr_path) + + # Step 3: 对齐 ASR 上下文 + contexts = align_asr_context(b_lines, asr_lines) + assert len(contexts) == len(b_lines), ( + f"context 长度 {len(contexts)} != B 稿行数 {len(b_lines)}" + ) + + # Step 4: 分块 + 缓存 + cache_dir = out_dir / ".c3_cache" + cache_dir.mkdir(parents=True, exist_ok=True) + + all_records = [] + total_batches = (len(b_lines) + batch_size - 1) // batch_size + + for batch_idx in range(total_batches): + start = batch_idx * batch_size + end = min(start + batch_size, len(b_lines)) + batch_b = b_lines[start:end] + batch_ctx = contexts[start:end] + + cache_path = cache_dir / f"batch_{batch_idx}.json" + + if cache_path.exists(): + # 断点续跑:复用缓存 + try: + cached = json.loads(cache_path.read_text(encoding="utf-8")) + print(f"[fusion_review] 复用缓存 batch_{batch_idx} ({len(cached)} 条)") + all_records.extend(cached) + continue + except Exception as e: + print( + f"[fusion_review] 缓存 batch_{batch_idx} 损坏,重新计算: {e}", + file=sys.stderr, + ) + + print( + f"[fusion_review] 复审 batch {batch_idx + 1}/{total_batches} " + f"(行 {start + 1}-{end})..." + ) + try: + batch_records = review_batch(batch_b, batch_ctx, no_ai=no_ai) + except Exception as e: + print( + f"[fusion_review] batch {batch_idx + 1} 失败,跳过缓存写入: {e}", + file=sys.stderr, + ) + # 不写缓存,下次重跑时重新请求该批 + continue + + # 写入缓存 + cache_path.write_text( + json.dumps(batch_records, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + all_records.extend(batch_records) + + # Step 6: 硬校验 + _hard_validate(all_records, b_lines) + + # Step 6.5: 修正语义——final_text 等于 B 原文的行强制归为 unchanged + _normalize_unchanged_when_no_edit(all_records, b_lines) + + # Step 7: 写 融合B稿.txt + fused_path = out_dir / "融合B稿.txt" + fused_lines = [] + for rec, bl in zip(all_records, b_lines): + fused_lines.append(f"[{bl['ts_raw']}] {rec['final_text']}") + fused_path.write_text("\n".join(fused_lines) + "\n", encoding="utf-8") + + # Step 8: 写 fusion_review.csv + csv_path = out_dir / "fusion_review.csv" + csv_rows = [ + "line_no,timestamp,b_original,asr_context,final_text,change_type,confidence,reason" + ] + for rec, bl, ctx in zip(all_records, b_lines, contexts): + if rec["change_type"] == "unchanged" and rec["confidence"] >= 0.8: + continue # 只写需要 review 的行 + # CSV 转义: 字段含逗号或引号时用双引号包裹 + row_fields = [ + str(rec["line_no"]), + bl["ts_raw"], + bl["text"], + ctx, + rec["final_text"], + rec["change_type"], + str(rec["confidence"]), + rec["reason"], + ] + csv_rows.append(_csv_row(row_fields)) + csv_path.write_text("\n".join(csv_rows) + "\n", encoding="utf-8") + + # Step 9: 统计 + stats = { + "total_lines": len(b_lines), + "change_counts": {}, + "review_lines": 0, + } + for rec in all_records: + ct = rec["change_type"] + stats["change_counts"][ct] = stats["change_counts"].get(ct, 0) + 1 + if ct != "unchanged" or rec["confidence"] < 0.8: + stats["review_lines"] += 1 + + print(f"[fusion_review] 融合完成: 总行数={stats['total_lines']}") + print(f"[fusion_review] 各 change_type 计数: {stats['change_counts']}") + print(f"[fusion_review] 进 review 行数: {stats['review_lines']}") + print(f"[fusion_review] 融合B稿: {fused_path}") + print(f"[fusion_review] review CSV: {csv_path}") + + stats["fused_path"] = str(fused_path) + stats["csv_path"] = str(csv_path) + + return stats + + +def _hard_validate(records: List[dict], b_lines: List[dict]) -> None: + """硬校验,任一不过就 raise ValueError,不写出文件。""" + # 校验 1: 行数必须相等 + if len(records) != len(b_lines): + raise ValueError( + f"行数不一致: records={len(records)}, B稿={len(b_lines)}" + ) + + # 校验 2: 逐行时间戳不能被动 + for i, (rec, bl) in enumerate(zip(records, b_lines)): + rec_line_no = rec.get("line_no") + expected_line_no = bl["idx"] + if rec_line_no != expected_line_no: + raise ValueError( + f"第 {i} 条记录 line_no={rec_line_no}, 期望 {expected_line_no}" + ) + + # 校验 3: change_type 枚举 + for rec in records: + ct = rec.get("change_type", "") + if ct not in CHANGE_TYPE_ENUM: + raise ValueError( + f"行 {rec.get('line_no')}: 非法 change_type='{ct}'" + ) + + +def _normalize_unchanged_when_no_edit( + records: List[dict], b_lines: List[dict] +) -> None: + """修正语义:final_text 等于 B 原文的行,强制归为 unchanged。 + + LLM 有时会把"考虑后决定保留 B 稿"标成 minor_edit, + 但实际 final_text == b_original, 这不在留痕范围内。 + """ + b_text_by_idx = {bl["idx"]: bl["text"] for bl in b_lines} + fixed = 0 + for rec in records: + line_no = rec.get("line_no") + b_orig = b_text_by_idx.get(line_no) + if b_orig is not None and rec.get("final_text") == b_orig: + if rec.get("change_type") != "unchanged": + rec["change_type"] = "unchanged" + rec["confidence"] = 1.0 + rec["reason"] = "" + fixed += 1 + if fixed: + print( + f"[fusion_review] 修正 {fixed} 行: final_text==B原文但change_type≠unchanged, 已强制归为 unchanged" + ) + + +def _csv_row(fields: List[str]) -> str: + """将字段列表格式化为 CSV 行,处理逗号和引号转义。""" + escaped = [] + for f in fields: + s = str(f) + if "," in s or '"' in s or "\n" in s: + s = s.replace('"', '""') + s = f'"{s}"' + escaped.append(s) + return ",".join(escaped)