simonkoson/tps-dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
tps-dashboard/doco/backup_before_spk/fusion_align.py
T

1761 lines
59 KiB
Python
Raw Permalink Normal View History

doco: v2说话人分段模式 — ASR说话人分离+大block拆分+三维动画解说识别
2026-06-24 16:26:05 +08:00
# -*- coding: utf-8 -*-
"""
C4: 融合B稿(743行碎句) + A稿分段骨架 → 融合A稿.docx(公文格式)
=============================================================
职责: AI 只做分段对齐(决定每句归哪段),正文一字不改、纯规则拼接。
P3 新增: LLM 分段骨架抽取 (doco skeleton 命令)
- extract_a_paragraphs: 纯 docx 段落样式提取
- extract_skeleton_llm: LLM 判断分段结构,产出 JSON 骨架
- validate_skeleton_coverage: 全覆盖硬校验(无缺口/无重叠/无越界)
- parse_a_segments: 改读 skeleton JSON 替代正则(ep001 fallback 保留)
"""
import json
import re
import sys
import unicodedata
from pathlib import Path
from typing import List, Dict, Optional, Tuple, Any
from docx import Document
from docx.shared import Pt, Cm
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml.ns import qn
from docx.oxml import OxmlElement
from .llm import chat, LLMConfigError
from .fusion_review import parse_timed_lines
# --------------------------------------------------------------------------
# 常量
# --------------------------------------------------------------------------
SEG_HEADER_PATTERN = re.compile(r"^【.+?】$")
# 隔断识别: "隔断：【标题】"
SEG_BREAK_PATTERN = re.compile(r"^隔断：【(.+?)】")
# 句尾标点: 行尾已有这些字符则不补逗号
SENTENCE_END_PUNCT = frozenset("。！？；…!?;")
# 字体
TITLE_FONT_PRIMARY = "方正小标宋_GBK"
TITLE_FONT_FALLBACK = "宋体"
HEADER_FONT = "黑体"
BODY_FONT = "仿宋_GB2312"
BREAK_HEADER_FONT = "仿宋_GB2312" # 隔断段头用仿宋加粗
EMPTY_SEG_PLACEHOLDER = "（本段无对应字幕，待编导核）"
# 中文冒号/英文冒号（用于切分 inline 段头）
INLINE_HEADER_SEP = re.compile(r"[：:]")
# ====================================================================
# 1a. 纯 docx 段落样式提取
# ====================================================================
def extract_a_paragraphs(docx_path: str) -> List[Dict[str, Any]]:
"""
读 A 稿 docx，返回带样式信息的段落列表。
每条: {"idx": int, "text": str, "bold": bool, "align": str}
- bold: 该段任一 run 含加粗即为 True
- align: "left" | "center" | "justify" | "right" | None
- 跳过全空段落
"""
p = Path(docx_path)
if not p.exists():
raise FileNotFoundError(f"A稿 docx 不存在: {docx_path}")
doc = Document(str(p))
paras_out = []
ALIGN_MAP = {
WD_ALIGN_PARAGRAPH.LEFT: "left",
WD_ALIGN_PARAGRAPH.CENTER: "center",
WD_ALIGN_PARAGRAPH.JUSTIFY: "justify",
WD_ALIGN_PARAGRAPH.RIGHT: "right",
}
for idx, para in enumerate(doc.paragraphs):
text = para.text.strip()
if not text:
continue
# 检测加粗: 任一 run 的 bold=True
bold = any(
(run.bold is not None and run.bold) for run in para.runs
)
# 对齐
raw_align = para.alignment
align = ALIGN_MAP.get(raw_align, None)
paras_out.append(
{
"idx": len(paras_out), # 重新编号，连续无跳
"text": text,
"bold": bold,
"align": align,
}
)
return paras_out
# ====================================================================
# 1b. 段落列表序列化（喂 LLM）
# ====================================================================
def _serialize_paras_for_llm(paras: List[Dict[str, Any]]) -> str:
"""把段落列表格式化为 LLM 输入文本: [idx] bold=X align=Y | text"""
lines = []
for p in paras:
flags = []
if p.get("bold"):
flags.append("bold")
if p.get("align"):
flags.append(f"align={p['align']}")
flag_str = ",".join(flags) if flags else "-"
lines.append(f"[{p['idx']}] {flag_str} | {p['text']}")
return "\n".join(lines)
# ====================================================================
# 2. LLM 分段骨架提取
# ====================================================================
SKELETON_SYSTEM_PROMPT = """你是电视专题片文稿结构分析器。给你全文段落列表(每段带编号和样式标注),请判断每段的结构角色并输出 JSON 骨架。
【核心铁律】
1. 你只做结构判断,只输出 JSON 骨架。**严禁输出、复述、生成任何正文文字。**
2. role_label 只包含**角色类型**,严禁包含任何真人姓名/配音员/编导署名。
3. 所有段落必须归入某个 segment,骨架的 [para_start,para_end] 区间必须恰好覆盖
第一个段头之后的所有段落一次——无缺口、无重叠、无越界。
【角色类型参考集(开放可超集)】
导视 / 主持人 / 解说 / 专家 / 嘉宾 / 旁白 / 三维动画解说 / 演播室主持人 /
小剧场角色(小剧场N-角色名,如小剧场1-主持人、小剧场1-卡拉什尼科夫、
小剧场1-尤金斯通纳) / 子标题(break 型,原文字保留)
【段头识别规则】
- 冒号式: 文本含"："或":"且冒号左边是角色类型标识(如"解说：""主持人："
"专家：""三维动画解说：""演播室主持人："等)→ header_inline=true,
段头与正文同段。该段 para_start==para_end==header_para_idx。
- 独占一行式: 独立段落标记角色/子标题(如"【手枪】""【步枪】""小剧场1-主持人"
"解密鱼鳔的潜艇压载水舱系统")→ header_inline=false,段头占一个段落,
正文从下一段开始。
- 子标题(break): 加粗+总结性的短语+无冒号→ type=break,header_inline=false。
【role_label 命名规则(最高优先级红线)】
- role_label 格式: "【角色类型N】",如"【解说1】""【主持人1】""【三维动画解说1】"
"【演播室主持人1】""【小剧场1-主持人】""【小剧场1-卡拉什尼科夫】"
- 同类角色按出现顺序自动编号: 第一个解说→【解说1】,第二个→【解说2】...
- break 型: role_label 为子标题原文(不加工)
- ignore 型: role_label 为标记类型原文(如"【固】""【摇】""【轨】")
- **红线**: role_label 中**禁止包含**真人姓名/配音员姓名/编导署名。
反面案例: "【演播室主持人:刘通】""【解说:穆佩弦】""【旁白:孙逸昊】"
"【主持人:左鑫】"——全部非法。正确写法:"【演播室主持人1】""【解说1】"
"【旁白1】""【主持人1】"。
- **小剧场角色(如卡拉什尼科夫、尤金斯通纳、小剧场1-主持人)保留**:
这是被演绎的历史人物/剧情角色,非真人配音员,可以保留。
- 清理垃圾前缀: 如"Xr演播室"、"Xr"→ 提取真正的角色类型后命名。
【陷阱(必须归为 ignore,type="ignore")】
- 【固】【摇】【轨】【推】【拉】【跟】等运镜/镜头标记 → type=ignore
- (演播室)(此处为舞台表演)(掌声)等圆括号舞台提示 → type=ignore
- 纯标点/纯空白段 → type=ignore
- 注意: 整段加粗 ≠ 段头(如枪王期全文字幕加粗,需结合语义判断)
- 注意: "三维动画"、"三维动画解说"无冒号时可能是独立段头,结合上下文判断
【输出格式】
纯 JSON 数组,无 markdown 包装,无额外解释:
[
{
"type": "normal" | "break" | "ignore",
"para_start": int, // 此段覆盖的起始段落下标(inclusive)
"para_end": int, // 此段覆盖的结束段落下标(inclusive)
"role_label": "【演播室主持人1】",
"header_inline": bool, // 段头是否与正文同段(冒号式)
"header_para_idx": int // 段头所在段落下标(用于后续切除段头)
},
...
]
【注意】
- para_start/para_end 基于输入给你的 [idx] 编号
- 全文第一个段落是标题(含副标题等),请自动忽略(不纳入任何 segment)
- break 段: 单一段落,header_inline=false,body 为空
- ignore 段: 不进最终输出,但区间必须覆盖以通过全覆盖校验"""
def extract_skeleton_llm(
paras: List[Dict[str, Any]],
max_tokens: int = 16000,
) -> List[Dict[str, Any]]:
"""
把全文段落列表喂给 LLM,返回 JSON 骨架。
Args:
paras: extract_a_paragraphs 的输出
max_tokens: LLM max_tokens,长稿需调大(默认 16000)
Returns:
骨架列表 [{type, para_start, para_end, role_label, header_inline, header_para_idx}]
Raises:
ValueError: JSON 解析失败 / 返回格式不对
LLMConfigError: API key 未配置
"""
if not paras:
raise ValueError("段落列表为空,无法提取骨架")
# 序列化全文
full_text = _serialize_paras_for_llm(paras)
user_content = (
f"全文共 {len(paras)} 个段落(第0段是标题,请自动忽略):\n\n"
f"{full_text}\n\n"
f"请输出 JSON 骨架数组。"
)
messages = [
{"role": "system", "content": SKELETON_SYSTEM_PROMPT},
{"role": "user", "content": user_content},
]
raw_response = chat(
messages,
thinking=True,
max_tokens=max_tokens,
temperature=0.0,
)
# 解析 JSON
parsed = _parse_skeleton_json(raw_response)
# 校验每条记录的必要字段
required_fields = ["type", "para_start", "para_end", "role_label", "header_inline", "header_para_idx"]
for i, item in enumerate(parsed):
for field in required_fields:
if field not in item:
raise ValueError(
f"骨架第 {i} 条缺少字段 '{field}': {json.dumps(item, ensure_ascii=False)[:200]}"
)
# 校验 type 值
if item["type"] not in ("normal", "break", "ignore"):
raise ValueError(
f"骨架第 {i} 条 type='{item['type']}' 不合法,应为 normal/break/ignore"
)
return parsed
def _parse_skeleton_json(raw: str) -> List[dict]:
"""解析 LLM 返回的骨架 JSON 数组,去除 markdown code fences 等包装。"""
text = raw.strip()
# 去掉 markdown code fences
if text.startswith("```"):
lines = text.splitlines()
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]}\n"
f"原始响应后 200 字符: {raw[-200:]}"
)
if not isinstance(result, list):
raise ValueError(
f"LLM 返回不是 JSON 数组, 类型为 {type(result).__name__}"
)
if len(result) == 0:
raise ValueError("LLM 返回空骨架数组")
return result
# ====================================================================
# 2b. 全覆盖硬校验
# ====================================================================
def validate_skeleton_coverage(
skeleton: List[Dict[str, Any]],
total_paras: int,
) -> Tuple[bool, str]:
"""
校验骨架的 [para_start, para_end] 区间是否恰好覆盖 title 之后
的每个段落一次(无缺口/无重叠/无越界)。
注意: 第0段(title)不纳入覆盖范围。
Args:
skeleton: LLM 返回的骨架列表
total_paras: 总段落数(含 title)
Returns:
(passed: bool, message: str)
"""
if total_paras <= 1:
return True, ""
covered = set()
expected_start = 1 # title 后第一个段落下标
expected_end = total_paras - 1
errors = []
for i, seg in enumerate(skeleton):
ps = seg.get("para_start")
pe = seg.get("para_end")
if ps is None or pe is None:
errors.append(f"骨架第 {i} 条缺少 para_start/para_end")
continue
if not isinstance(ps, int) or not isinstance(pe, int):
errors.append(
f"骨架第 {i} 条 para_start/para_end 非整数: ps={ps}, pe={pe}"
)
continue
if ps > pe:
errors.append(
f"骨架第 {i} 条 para_start({ps}) > para_end({pe})"
)
continue
if ps < expected_start:
errors.append(
f"骨架第 {i} 条 para_start({ps}) < 预期最小值 {expected_start}(越界到 title 区)"
)
continue
if pe > expected_end:
errors.append(
f"骨架第 {i} 条 para_end({pe}) > 预期最大值 {expected_end}(越界)"
)
continue
# 检查重叠
seg_range = set(range(ps, pe + 1))
overlap = seg_range & covered
if overlap:
errors.append(
f"骨架第 {i} 条 [{ps},{pe}] 与已覆盖下标重叠: {sorted(overlap)}"
)
covered |= seg_range
# 检查缺口
all_expected = set(range(expected_start, expected_end + 1))
missing = all_expected - covered
if missing:
errors.append(
f"段落下标缺口(missing): {sorted(missing)[:20]}"
+ ("..." if len(missing) > 20 else "")
)
# 检查多余
extra = covered - all_expected
if extra:
errors.append(
f"段落下标多余(extra): {sorted(extra)[:20]}"
+ ("..." if len(extra) > 20 else "")
)
if errors:
return False, "; ".join(errors)
else:
return True, ""
# ====================================================================
# 3. 骨架落盘 + 预览
# ====================================================================
def save_skeleton(
skeleton: List[Dict[str, Any]],
paras: List[Dict[str, Any]],
episode_id: str,
output_dir: str,
) -> Path:
"""
将骨架写入 programs/<episode_id>/<episode_id>_a_skeleton.json
同时为每条 segment 附加正文前20字预览(仅用于人工核验,不入 JSON 机器读字段)。
Returns: skeleton 文件路径
"""
out_dir = Path(output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
# 构建保存用的骨架(不含正文预览,纯机器可读)
skeleton_path = out_dir / f"{episode_id}_a_skeleton.json"
skeleton_path.write_text(
json.dumps(skeleton, ensure_ascii=False, indent=2),
encoding="utf-8",
)
return skeleton_path
def print_skeleton_table(
skeleton: List[Dict[str, Any]],
paras: List[Dict[str, Any]],
title: str = "",
) -> None:
"""
打印人类可读预览表:
seq | type | role_label | 正文前20字
用于制片人肉眼核验: role_label 是否含真人姓名? ignore 是否漏/多?
"""
print(f"\n{'=' * 70}")
print(f" {title} 骨架预览表")
print(f"{'=' * 70}")
print(f"{'seq':>4} | {'type':<7} | {'role_label':<35} | 正文前20字")
print(f"{'-' * 4}-+-{'-' * 7}-+-{'-' * 35}-+-{'-' * 20}")
for seq_idx, seg in enumerate(skeleton, start=1):
seg_type = seg.get("type", "?")
role_label = seg.get("role_label", "")
ps = seg.get("para_start", -1)
pe = seg.get("para_end", -1)
header_inline = seg.get("header_inline", False)
# 取该段实际正文的前20字(非段头文字)
preview = ""
if 0 <= ps < len(paras):
if seg_type == "break":
# break 段无正文,显示子标题原文
preview = paras[ps]["text"][:20]
elif seg_type == "ignore":
preview = paras[ps]["text"][:20]
elif header_inline:
# 冒号式: para_start 段落含"角色：正文",切掉角色部分
full_text = paras[ps]["text"]
# 找第一个中文/英文冒号
colon_match = re.search(r"[：:]", full_text)
if colon_match:
preview = full_text[colon_match.end():].strip()[:20]
else:
preview = full_text[:20]
else:
# 独立段头: 段头占 ps,正文从 ps+1 开始
body_start = ps + 1
if body_start < len(paras) and body_start <= pe:
preview = paras[body_start]["text"][:20]
else:
# 无正文段(如段头后无内容)
preview = "(无正文)"
print(
f"{seq_idx:>4} | {seg_type:<7} | {role_label:<35} | {preview}"
)
print(f"{'=' * 70}")
print("[WARN] 请人工核验: role_label是否含真人姓名? ignore是否漏/多? type是否正确?")
print()
# ====================================================================
# 4. run_skeleton — Step 1-3 编排
# ====================================================================
def run_skeleton(
episode_id: str,
a_script_path: str,
output_dir: Optional[str] = None,
max_tokens: int = 16000,
) -> Dict[str, Any]:
"""
doco skeleton 命令主流程:
1. extract_a_paragraphs
2. extract_skeleton_llm
3. validate_skeleton_coverage (全覆盖硬校验)
4. save_skeleton + print_skeleton_table
Returns 统计 dict: {skeleton_path, total_paras, skeleton_count, coverage_ok}
"""
if output_dir is None:
out_dir = Path("programs") / episode_id
else:
out_dir = Path(output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
# Step 1: 段落提取
print(f"[skeleton] 读取 A 稿: {a_script_path}")
paras = extract_a_paragraphs(str(a_script_path))
print(f"[skeleton] 段落数: {len(paras)} (含标题)")
if not paras:
raise ValueError(f"A稿无有效文本: {a_script_path}")
title = paras[0]["text"]
print(f"[skeleton] 标题: {title}")
# Step 2: LLM 骨架
print(f"[skeleton] 调用 LLM 提取分段骨架 (thinking=True, max_tokens={max_tokens})...")
skeleton = extract_skeleton_llm(paras, max_tokens=max_tokens)
print(f"[skeleton] LLM 返回骨架段数: {len(skeleton)}")
# Step 3: 全覆盖校验
coverage_ok, coverage_msg = validate_skeleton_coverage(skeleton, len(paras))
if not coverage_ok:
print(f"[skeleton] [FAIL] 全覆盖校验失败: {coverage_msg}", file=sys.stderr)
# 仍打印预览表,方便人工排查
print_skeleton_table(skeleton, paras, title=f"{episode_id} (校验失败)")
raise ValueError(
f"骨架全覆盖校验失败: {coverage_msg}\n"
f"可能原因: LLM 数错下标/JSON 被截断/段落编号理解偏差。\n"
f"建议: 调大 max_tokens(当前={max_tokens})后重试,或检查 prompt。"
)
print(f"[skeleton] [PASS] 全覆盖校验通过")
# Step 4: 落盘(原始骨架) + 预览(合并去ignore 后的最终 skeleton)
skeleton_path = save_skeleton(skeleton, paras, episode_id, str(out_dir))
print(f"[skeleton] 骨架已保存: {skeleton_path}")
# 合并展示(预览表展示 compose 真正会用的那份)
skeleton_display = _merge_skeleton_segments(skeleton)
if len(skeleton_display) != len(skeleton):
print(
f"[skeleton] 预览(合并后): {len(skeleton)} → {len(skeleton_display)} "
f"段 ({len(skeleton) - len(skeleton_display)} 条同角色合并/丢弃 ignore)"
)
print_skeleton_table(skeleton_display, paras, title=episode_id)
return {
"skeleton_path": str(skeleton_path),
"total_paras": len(paras),
"skeleton_count": len(skeleton),
"coverage_ok": coverage_ok,
}
# ====================================================================
# 4b. 合并被 ignore 隔断的同角色段
# ====================================================================
def _merge_skeleton_segments(
skeleton: List[Dict[str, Any]],
) -> List[Dict[str, Any]]:
"""
后处理: 若两个相邻 normal 段 role_label 完全相同,
且它们之间只隔着 ignore 段(无其它 normal/break),合并为一段。
目的: 消除主持人话被运镜提示(（出枪柜）（换枪）)切断后产生的重复段头。
Args:
skeleton: LLM 产出的原始骨架列表
Returns:
合并后的骨架(ignore 已丢弃,相邻同角色 normal 已合并)
"""
if not skeleton:
return []
merged: List[Dict[str, Any]] = []
pending_normal: Optional[Dict[str, Any]] = None # 待合并的 normal 段
for seg in skeleton:
seg_type = seg.get("type", "normal")
if seg_type == "normal":
if pending_normal is not None and pending_normal["role_label"] == seg.get("role_label", ""):
# 同角色: 扩展 para 区间,保留第一个段头的 header 信息
pending_normal["para_end"] = max(pending_normal["para_end"], seg.get("para_end", pending_normal["para_end"]))
else:
# 不同角色或首个 normal: 先 flush 前一个,再缓存当前
if pending_normal is not None:
merged.append(pending_normal)
# 浅拷贝一份避免修改原始 skeleton
pending_normal = dict(seg)
elif seg_type == "ignore":
# 跳过,不打断 pending 链(就是这些 ignore 造成了切断)
continue
else:
# break 或其它类型: flush pending,然后直接加入
if pending_normal is not None:
merged.append(pending_normal)
pending_normal = None
merged.append(seg)
# 收尾
if pending_normal is not None:
merged.append(pending_normal)
return merged
# ====================================================================
# 5. 解析 A 稿分段(改读 skeleton 替代正则)
# ====================================================================
def parse_a_segments(docx_path: str) -> dict:
"""
读 A 稿 docx,返回 {
"title": str,
"segments": [
{"seg_id":int, "type":"normal"|"break",
"header":str, "body":str}
]
}
优先读 <stem>_a_skeleton.json (LLM 骨架),
若不存在且首段匹配旧正则格式(ep001),走旧正则逻辑 fallback。
否则报错。
"""
p = Path(docx_path)
if not p.exists():
raise FileNotFoundError(f"A稿 docx 不存在: {docx_path}")
# 推断 skeleton 路径
# docx 在 programs/<episode_id>/xxx.docx
# skeleton 在 programs/<episode_id>/<episode_id>_a_skeleton.json
episode_dir = p.parent
episode_id = episode_dir.name
skeleton_path = episode_dir / f"{episode_id}_a_skeleton.json"
if skeleton_path.exists():
return _parse_a_segments_from_skeleton(str(p), str(skeleton_path))
else:
# Fallback: 检查是否为旧格式(ep001)
doc = Document(str(p))
paras = [para.text.strip() for para in doc.paragraphs if para.text.strip()]
if paras and SEG_HEADER_PATTERN.match(paras[1] if len(paras) > 1 else ""):
print(
f"[fusion_align] 未找到 skeleton.json,检测到旧版【】格式,走正则解析"
)
return _parse_a_segments_regex(str(p), paras)
else:
raise FileNotFoundError(
f"未找到分段骨架文件: {skeleton_path}\n"
f"且 A 稿不匹配旧版【】段头格式。\n"
f"请先运行: doco skeleton --episode-id {episode_id} --a-script {docx_path}"
)
def _parse_a_segments_from_skeleton(
docx_path: str, skeleton_path: str
) -> dict:
"""
从 skeleton JSON 解析分段:
- 按 para_start/para_end 从 docx 原样抽取正文
- header_inline=true: 按第一个中文冒号切除段头,余下为 body
- header_inline=false: role_label 为 header,后续段落为 body
- type=ignore: 丢弃,不进 segments
- type=break: 单独产出 segment
"""
# 读 skeleton
skeleton_raw = json.loads(Path(skeleton_path).read_text(encoding="utf-8"))
# 合并被 ignore 隔断的同角色段（如主持人话被（出枪柜）（换枪）切断）
skeleton = _merge_skeleton_segments(skeleton_raw)
if len(skeleton) != len(skeleton_raw):
print(
f"[fusion_align] skeleton 合并: {len(skeleton_raw)} → {len(skeleton)} 段 "
f"(合并了 {len(skeleton_raw) - len(skeleton)} 条同角色拆分+丢弃 ignore)"
)
# 读 docx 段落(按原下标)
doc = Document(docx_path)
all_paras = [para.text.strip() for para in doc.paragraphs if para.text.strip()]
if not all_paras:
raise ValueError(f"A稿 docx 无有效文本: {docx_path}")
title = all_paras[0]
# 为 skeleton 建立原段落下标到实际文本的映射
# 注意: all_paras[0] = title, 所以 skeleton 的 para_start/para_end 对应 all_paras 的下标
segments = []
seg_id_counter = 0
for seg in skeleton:
seg_type = seg.get("type", "normal")
if seg_type == "ignore":
continue # 丢弃镜头标记等
ps = seg.get("para_start", -1)
pe = seg.get("para_end", -1)
role_label = seg.get("role_label", "")
header_inline = seg.get("header_inline", False)
if ps < 0 or pe < 0 or ps >= len(all_paras):
print(
f"[fusion_align] 警告: 骨架段 para_start/para_end 越界,跳过: {seg}",
file=sys.stderr,
)
continue
if seg_type == "break":
# 子标题: header=role_label(原文), body 为空
header_text = all_paras[ps] if ps < len(all_paras) else role_label
segments.append(
{
"seg_id": seg_id_counter,
"type": "break",
"header": role_label,
"body": "",
}
)
seg_id_counter += 1
continue
# normal 段
if header_inline:
# 段头与正文同段: 按第一个中文冒号切分
full_text = all_paras[ps] if ps < len(all_paras) else ""
header_text, body_text = _split_inline_header(full_text)
# header 用 role_label(规范化后), body 用切除后的
body_parts = [body_text] if body_text else []
# 如果 para_end > para_start,后续段落全是 body
for extra_idx in range(ps + 1, pe + 1):
if extra_idx < len(all_paras):
body_parts.append(all_paras[extra_idx])
else:
# 独立段头: header 段落自身是 role_label, body 从下一段开始
header_text = role_label
body_parts = []
for body_idx in range(ps + 1, pe + 1):
if body_idx < len(all_paras):
body_parts.append(all_paras[body_idx])
segments.append(
{
"seg_id": seg_id_counter,
"type": "normal",
"header": role_label,
"body": "\n".join(body_parts),
}
)
seg_id_counter += 1
return {"title": title, "segments": segments}
def _split_inline_header(text: str) -> Tuple[str, str]:
"""
按第一个中文冒号切分段头和正文。
如 "演播室主持人1：各位观众大家好" → ("演播室主持人1", "各位观众大家好")
如果没有冒号,全文本当 body。
"""
match = INLINE_HEADER_SEP.search(text)
if match:
header_part = text[: match.start()].strip()
body_part = text[match.end() :].strip()
return header_part, body_part
else:
return "", text
def _parse_a_segments_regex(docx_path: str, paras: List[str]) -> dict:
"""
旧正则逻辑(ep001 及兼容): ^【.+?】$ 段头 + ^隔断：【】识别。
保留不动。
"""
title = paras[0]
segments = []
current_header = None
current_body_parts = []
seg_id_counter = 0
for para_text in paras[1:]:
break_match = SEG_BREAK_PATTERN.match(para_text)
if break_match:
if current_header is not None:
segments.append(
{
"seg_id": seg_id_counter,
"type": "normal",
"header": current_header,
"body": "\n".join(current_body_parts),
}
)
seg_id_counter += 1
break_title = break_match.group(1)
segments.append(
{
"seg_id": seg_id_counter,
"type": "break",
"header": break_title,
"body": "",
}
)
seg_id_counter += 1
current_header = None
current_body_parts = []
continue
if SEG_HEADER_PATTERN.match(para_text):
if current_header is not None:
segments.append(
{
"seg_id": seg_id_counter,
"type": "normal",
"header": current_header,
"body": "\n".join(current_body_parts),
}
)
seg_id_counter += 1
current_header = para_text
current_body_parts = []
else:
if current_header is not None:
current_body_parts.append(para_text)
if current_header is not None:
segments.append(
{
"seg_id": seg_id_counter,
"type": "normal",
"header": current_header,
"body": "\n".join(current_body_parts),
}
)
return {"title": title, "segments": segments}
# ====================================================================
# 1b. 提取 normal 段（供对齐使用）
# ====================================================================
def _get_normal_segments(segments: List[dict]) -> List[dict]:
"""从全局 segments 中筛出 type=="normal" 的段,按 seg_id 排序。"""
normals = [seg for seg in segments if seg.get("type", "normal") == "normal"]
normals.sort(key=lambda s: s["seg_id"])
return normals
def _build_normal_seg_id_map(normal_segs: List[dict]) -> Dict[int, int]:
"""
建立映射: global_seg_id → normal_seg_id (0-based continuous)
反向映射: normal_seg_id → global_seg_id
"""
g2n = {}
n2g = {}
for nid, seg in enumerate(normal_segs):
g2n[seg["seg_id"]] = nid
n2g[nid] = seg["seg_id"]
return g2n, n2g
# ====================================================================
# 2. 构造分段对齐 Prompt
# ====================================================================
SYSTEM_PROMPT_ALIGN = """你是《军事科技》专题片分段对齐员。给你 A稿完整分段正文 和 融合B稿碎句(屏幕字幕,带行号,按播出时间排列)。
B句是播出字幕的逐句拆分,A稿是同一内容的书面稿,两者高度同源——请直接按"这句话的内容出现在A稿哪一段正文里"来归段。
规则:seg_id 随行号单调不减;边界以语义为准(如主持人开场白归主持人段,旁白归解说段)。
只返回JSON数组: [{"line_no":int,"seg_id":int,"confidence":0~1}]"""
def build_align_prompt(
batch_b: List[dict],
normal_segments: List[dict],
min_normal_seg_id: int,
) -> List[dict]:
"""
构造 messages:
(a) 完整 normal 段清单: 每段 "seg_id | header | A稿完整正文"
(b) 本批 B 句: 每句 "[全局行号] 文本"
(c) 下界约束: 传入上一批最后一行归到的 normal_seg_id,本批不得小于它
"""
seg_lines = []
for idx, seg in enumerate(normal_segments):
full_body = seg["body"].replace("\n", " ")
seg_lines.append(
f"seg_id={idx} | {seg['header']} | {full_body}"
)
seg_list_str = "\n".join(seg_lines)
b_lines_str = "\n".join(
f"[{bl['idx']}] {bl['text']}" for bl in batch_b
)
constraint_note = ""
if min_normal_seg_id > 0:
constraint_note = (
f"\n\n【重要约束】上一批最后一行归到了 seg_id={min_normal_seg_id}，"
f"本批所有行的 seg_id 必须 >= {min_normal_seg_id}，不得回退。"
)
user_content = (
f"A稿分段清单（共 {len(normal_segments)} 段）：\n\n"
f"{seg_list_str}\n\n"
f"--- 本批 B 稿碎句（共 {len(batch_b)} 行）---\n\n"
f"{b_lines_str}"
f"{constraint_note}"
)
messages = [
{"role": "system", "content": SYSTEM_PROMPT_ALIGN},
{"role": "user", "content": user_content},
]
return messages
# ====================================================================
# 3. 单批对齐
# ====================================================================
def align_batch(
batch_b: List[dict],
normal_segments: List[dict],
min_normal_seg_id: int,
no_ai: bool = False,
total_b_lines: int = 743,
) -> List[dict]:
"""
no_ai=True: 按全局行号比例均分到 normal 段(仅供验证管道)
no_ai=False: 调 LLM(thinking=True) 返回 JSON
返回 [{"line_no": int, "seg_id": int, "confidence": float}]
**注意**: 返回的 seg_id 是 normal_segments 体系(0-based 连续编号)
"""
if no_ai:
n_segs = len(normal_segments)
records = []
for bl in batch_b:
seg_idx = min(
int((bl["idx"] - 1) / total_b_lines * n_segs), n_segs - 1
)
records.append(
{"line_no": bl["idx"], "seg_id": seg_idx, "confidence": 0.5}
)
return records
messages = build_align_prompt(batch_b, normal_segments, min_normal_seg_id)
try:
raw_response = chat(
messages,
thinking=True,
max_tokens=4096,
temperature=0.0,
)
parsed = _parse_align_json(raw_response, len(batch_b))
except Exception as e:
print(
f"[fusion_align] LLM 调用失败,回退: {e}",
file=sys.stderr,
)
records = []
for bl in batch_b:
records.append(
{
"line_no": bl["idx"],
"seg_id": min_normal_seg_id,
"confidence": 0.3,
}
)
return records
records = []
for item in parsed:
records.append(
{
"line_no": int(item.get("line_no", 0)),
"seg_id": int(item.get("seg_id", min_normal_seg_id)),
"confidence": float(item.get("confidence", 0.5)),
}
)
return records
def _parse_align_json(raw: str, expected_len: int) -> List[dict]:
"""解析 LLM 返回的 JSON 数组,去除 markdown code fences 等包装。"""
text = raw.strip()
if text.startswith("```"):
lines = text.splitlines()
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} 条"
)
return result
# ====================================================================
# 4. 单调修正
# ====================================================================
def enforce_monotonic(
records: List[dict],
min_seg_id: int = 0,
) -> List[dict]:
"""
确保 seg_id 随 line_no 单调不减。
若出现回退(seg_id < 前值), 强制改回前值 + confidence 降到 0.3。
返回修正日志列表 [{line_no, original_seg_id, forced_seg_id}]。
"""
audit_log = []
prev_seg_id = min_seg_id
for rec in records:
if rec["seg_id"] < prev_seg_id:
audit_log.append(
{
"line_no": rec["line_no"],
"original_seg_id": rec["seg_id"],
"forced_seg_id": prev_seg_id,
}
)
rec["seg_id"] = prev_seg_id
rec["confidence"] = 0.3
else:
prev_seg_id = rec["seg_id"]
return audit_log
# ====================================================================
# 5. 分段对齐主函数
# ====================================================================
def align_lines_to_segments(
b_lines: List[dict],
segments: List[dict],
no_ai: bool = False,
batch_size: int = 40,
cache_dir: Optional[Path] = None,
) -> Tuple[List[dict], List[dict], List[dict]]:
"""
分批送 LLM 对齐(或 no_ai 均分),只对齐 normal 段。
返回:
(alignment_records, audit_logs, normal_segments)
"""
normal_segs = _get_normal_segments(segments)
if not normal_segs:
raise ValueError("normal segments 为空,无法对齐")
if cache_dir is None:
cache_dir = Path(".c4_cache")
cache_dir.mkdir(parents=True, exist_ok=True)
if no_ai:
n_segs = len(normal_segs)
total = len(b_lines)
records = []
for bl in b_lines:
seg_idx = min(
int((bl["idx"] - 1) / total * n_segs), n_segs - 1
)
records.append(
{"line_no": bl["idx"], "seg_id": seg_idx, "confidence": 0.5}
)
return records, [], normal_segs
all_records = []
all_audit = []
total_batches = (len(b_lines) + batch_size - 1) // batch_size
last_normal_seg_id = 0
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]
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_align] 复用缓存 batch_{batch_idx} ({len(cached)} 条)"
)
all_records.extend(cached)
if cached:
last_normal_seg_id = max(rec["seg_id"] for rec in cached)
continue
except Exception as e:
print(
f"[fusion_align] 缓存 batch_{batch_idx} 损坏,重新计算: {e}",
file=sys.stderr,
)
print(
f"[fusion_align] 对齐 batch {batch_idx + 1}/{total_batches} "
f"(行 {start + 1}-{end})..."
)
batch_records = align_batch(
batch_b, normal_segs, last_normal_seg_id, no_ai=False
)
audit = enforce_monotonic(batch_records, min_seg_id=last_normal_seg_id)
if audit:
all_audit.extend(audit)
print(
f"[fusion_align] batch {batch_idx + 1} 单调修正 {len(audit)} 行"
)
if batch_records:
last_normal_seg_id = max(rec["seg_id"] for rec in batch_records)
cache_path.write_text(
json.dumps(batch_records, ensure_ascii=False, indent=2),
encoding="utf-8",
)
all_records.extend(batch_records)
return all_records, all_audit, normal_segs
# ====================================================================
# 6. 硬校验
# ====================================================================
def hard_validate(
records: List[dict],
b_line_count: int,
seg_count: int,
) -> None:
"""硬校验,任一不过 raise ValueError,不写出半成品文件。"""
total = len(records)
if total != b_line_count:
raise ValueError(
f"对齐结果行数 {total} != B稿行数 {b_line_count}"
)
line_nos = [rec["line_no"] for rec in records]
expected = list(range(1, b_line_count + 1))
if line_nos != expected:
missing = set(expected) - set(line_nos)
extra = set(line_nos) - set(expected)
msg_parts = []
if missing:
msg_parts.append(f"缺失行号: {sorted(missing)[:15]}")
if extra:
msg_parts.append(f"多余行号: {sorted(extra)[:15]}")
raise ValueError(f"line_no 不连续: {'; '.join(msg_parts)}")
prev_seg = -1
for rec in records:
if rec["seg_id"] < prev_seg:
raise ValueError(
f"行 {rec['line_no']} seg_id={rec['seg_id']} ＜ 前值 "
f"{prev_seg},单调性被破坏"
)
prev_seg = rec["seg_id"]
max_seg = seg_count - 1
for rec in records:
sid = rec["seg_id"]
if sid < 0 or sid > max_seg:
raise ValueError(
f"行 {rec['line_no']} seg_id={sid} 越界 [0, {max_seg}]"
)
# ====================================================================
# 7. 正文拼接（纯规则，零改字）
# ====================================================================
def compose_segment_text(seg_lines: List[str]) -> str:
"""
逐句顺接,零改字规则:
- 行尾若有标点(。！？；…!?;)则保留,否则补"，"
- 整段最后一句末尾的"，"换成"。"
- 空列表返回 ""
"""
if not seg_lines:
return ""
parts = []
for text in seg_lines:
t = text.strip()
if not t:
continue
if t[-1] in SENTENCE_END_PUNCT:
parts.append(t)
else:
parts.append(t + "，")
if not parts:
return ""
last = parts[-1]
if last.endswith("，"):
last = last[:-1] + "。"
elif last[-1] not in "。！？…!?":
last = last + "。"
parts[-1] = last
return "".join(parts)
# ====================================================================
# 7b. 标点恢复（AI 加标点，硬校验守死）
# ====================================================================
PUNCTUATE_SYSTEM_PROMPT = """你是文稿标点校订员。给你一段【无标点的电视字幕文本】和一段【仅供参考的书面稿】。
你的唯一任务:在【字幕文本】相邻字符之间插入中文标点(。，、；：？！""''《》()…),让它符合阅读习惯。
铁律(违反即作废):
1. 只能插入标点。禁止增加任何汉字——包括"那么""此外""相比之下""防不胜防"这类连接词或成语,一个字都不能加。
2. 禁止删除任何汉字。禁止替换任何字("的/地/得"不能互换,"专门用于"不能改成"专司")。
3. 即使字幕读起来不通顺、有重复、缺字、有错别字,也原样保留每一个字,只在字之间加标点。
4. 参照稿只帮你判断在哪断句、哪里用顿号,绝不照抄它的字词。
5. 正常使用句号断句。一个完整陈述句结束时用句号"。",不要把所有断句都用逗号","连接。问句用问号"？",感叹用感叹号"！"。逗号只用于句内停顿、并列成分之间。
错误示范:
❌ 它的重量仅有3.6千克,非常便于携带,该枪采用了短行程活塞导气式自动方式,这种设计……
✅ 它的重量仅有3.6千克,非常便于携带。该枪采用了短行程活塞导气式自动方式,这种设计……
只返回加好标点的字幕文本,不要解释。"""
def strip_punct(text: str) -> str:
"""剔除 Unicode 标点 + 空白字符。用于硬校验。"""
result = []
for ch in text:
cat = unicodedata.category(ch)
if cat.startswith("P"):
continue
if ch.isspace():
continue
result.append(ch)
return "".join(result)
def punctuate_segment(
bare_text: str,
ref_body: str,
cache_dir: Optional[Path] = None,
seg_id: int = -1,
) -> Tuple[str, bool]:
"""
调 LLM 为 bare_text 添加标点符号。
返回 (文本, punct_ok)
"""
if not bare_text.strip():
return bare_text, True
if cache_dir is not None and seg_id >= 0:
cache_path = cache_dir / f"punct_seg_{seg_id}.json"
if cache_path.exists():
try:
cached = json.loads(cache_path.read_text(encoding="utf-8"))
if cached.get("bare_text") == bare_text:
print(f"[fusion_align] 复用标点缓存 seg_{seg_id}")
return cached.get("punct_text", bare_text), cached.get(
"punct_ok", False
)
except Exception as e:
print(
f"[fusion_align] 标点缓存 seg_{seg_id} 损坏: {e}",
file=sys.stderr,
)
user_content = f"【书面参照稿】\n{ref_body}\n\n【无标点字幕文本】\n{bare_text}"
messages = [
{"role": "system", "content": PUNCTUATE_SYSTEM_PROMPT},
{"role": "user", "content": user_content},
]
last_punct_text = bare_text
for attempt in range(3):
try:
raw_response = chat(
messages,
thinking=False,
max_tokens=3000,
temperature=0.0,
)
punct_text = raw_response.strip()
if strip_punct(punct_text) == strip_punct(bare_text):
if cache_dir is not None and seg_id >= 0:
cache_path = cache_dir / f"punct_seg_{seg_id}.json"
cache_path.write_text(
json.dumps(
{
"bare_text": bare_text,
"punct_text": punct_text,
"punct_ok": True,
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
if attempt > 0:
print(
f"[fusion_align] seg_{seg_id} 标点恢复第 {attempt + 1} 次重试通过",
file=sys.stderr,
)
return punct_text, True
else:
last_punct_text = punct_text
print(
f"[fusion_align] seg_{seg_id} 标点恢复第 {attempt + 1} 次尝试失败(模型改字),"
+ (f" 将重试..." if attempt < 2 else f" 已达上限,回退"),
file=sys.stderr,
)
except Exception as e:
print(
f"[fusion_align] seg_{seg_id} 标点 LLM 调用异常(第 {attempt + 1} 次): {e}",
file=sys.stderr,
)
if cache_dir is not None and seg_id >= 0:
cache_path = cache_dir / f"punct_seg_{seg_id}.json"
cache_path.write_text(
json.dumps(
{
"bare_text": bare_text,
"punct_text": bare_text,
"punct_ok": False,
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
return bare_text, False
# ====================================================================
# 8. 出 docx（GB/T 9704 公文格式）
# ====================================================================
def _set_run_font(run, font_name: str, size_pt: float, bold: bool = False):
run.font.name = font_name
run.font.size = Pt(size_pt)
run.bold = bold
rPr = run._element.get_or_add_rPr()
rFonts = rPr.find(qn("w:rFonts"))
if rFonts is None:
rFonts = OxmlElement("w:rFonts")
rPr.insert(0, rFonts)
rFonts.set(qn("w:ascii"), font_name)
rFonts.set(qn("w:hAnsi"), font_name)
rFonts.set(qn("w:eastAsia"), font_name)
def _set_line_spacing(paragraph, ratio: float = 1.25):
pPr = paragraph._element.get_or_add_pPr()
spacing = pPr.find(qn("w:spacing"))
if spacing is None:
spacing = OxmlElement("w:spacing")
pPr.append(spacing)
spacing.set(qn("w:line"), str(int(ratio * 240)))
spacing.set(qn("w:lineRule"), "auto")
def _set_first_line_indent(
paragraph, chars: float = 2.0, font_size_pt: float = 14.0
):
pPr = paragraph._element.get_or_add_pPr()
ind = pPr.find(qn("w:ind"))
if ind is None:
ind = OxmlElement("w:ind")
pPr.append(ind)
indent_twips = int(chars * font_size_pt * 20)
ind.set(qn("w:firstLine"), str(indent_twips))
def render_docx(
title: str,
segments: List[dict],
seg_texts: List[str],
out_path: str,
) -> None:
"""输出公文格式 docx (GB/T 9704)。"""
doc = Document()
section = doc.sections[0]
section.page_width = Cm(21.0)
section.page_height = Cm(29.7)
section.top_margin = Cm(3.7)
section.bottom_margin = Cm(3.5)
section.left_margin = Cm(2.8)
section.right_margin = Cm(2.6)
title_para = doc.add_paragraph()
title_para.alignment = WD_ALIGN_PARAGRAPH.CENTER
_set_run_font(title_para.add_run(title), TITLE_FONT_PRIMARY, 22, bold=False)
for seg, seg_text in zip(segments, seg_texts):
if seg.get("type") == "break":
h_para = doc.add_paragraph()
h_para.alignment = WD_ALIGN_PARAGRAPH.CENTER
_set_run_font(
h_para.add_run(seg["header"]),
BREAK_HEADER_FONT,
15,
bold=True,
)
_set_line_spacing(h_para, 1.25)
else:
h_para = doc.add_paragraph()
_set_run_font(h_para.add_run(seg["header"]), HEADER_FONT, 16)
_set_line_spacing(h_para, 1.25)
body_text = seg_text if seg_text else EMPTY_SEG_PLACEHOLDER
b_para = doc.add_paragraph()
_set_run_font(b_para.add_run(body_text), BODY_FONT, 14)
_set_line_spacing(b_para, 1.25)
_set_first_line_indent(b_para, 2.0, 14.0)
doc.save(out_path)
# ====================================================================
# 9. 写留痕 CSV (c4_alignment.csv)
# ====================================================================
def write_alignment_csv(
csv_path: str,
segments: List[dict],
alignment: List[dict],
audit_logs: List[dict],
punct_results: Dict[int, bool],
) -> None:
"""写 c4_alignment.csv。"""
normal_segs = _get_normal_segments(segments)
g2n, n2g = _build_normal_seg_id_map(normal_segs)
seg_data: Dict[int, dict] = {}
for rec in alignment:
normal_sid = rec["seg_id"]
global_sid = n2g.get(normal_sid, normal_sid)
if global_sid not in seg_data:
seg_data[global_sid] = {"lines": [], "confidences": []}
seg_data[global_sid]["lines"].append(rec["line_no"])
seg_data[global_sid]["confidences"].append(rec["confidence"])
forced_lines = {a["line_no"] for a in audit_logs}
rows = [
"seg_id,header,start_line,end_line,line_count,min_confidence,punct_ok,note"
]
for seg in segments:
sid = seg["seg_id"]
if seg.get("type") == "break":
note = "隔断"
rows.append(
f'{sid},"{seg["header"]}",,,0,0.0000,,"{note}"'
)
continue
sd = seg_data.get(sid, {"lines": [], "confidences": []})
lines = sorted(sd["lines"])
if lines:
start_line = lines[0]
end_line = lines[-1]
line_count = len(lines)
min_conf = min(sd["confidences"])
else:
start_line = ""
end_line = ""
line_count = 0
min_conf = 1.0
punct_ok = punct_results.get(sid, True)
notes = []
if line_count == 0:
notes.append("空段:无对应字幕")
if min_conf < 0.6 and line_count > 0:
notes.append(f"低把握(min_confidence={min_conf:.2f})")
if not punct_ok:
notes.append("标点恢复失败已回退,需人工")
forced_in_seg = [ln for ln in lines if ln in forced_lines]
if forced_in_seg:
preview = forced_in_seg[:10]
suffix = "..." if len(forced_in_seg) > 10 else ""
notes.append(
f"单调强制修正行: {preview}{suffix}"
)
note = "; ".join(notes)
header_escaped = seg["header"].replace('"', '""')
note_escaped = note.replace('"', '""')
rows.append(
f'{sid},"{header_escaped}",{start_line},{end_line},'
f'{line_count},{min_conf:.4f},{punct_ok},"{note_escaped}"'
)
Path(csv_path).write_text(
"\n".join(rows) + "\n", encoding="utf-8"
)
# ====================================================================
# 10. 主流程 (doco compose)
# ====================================================================
def run_compose(
episode_id: str,
output_dir: str,
no_ai: bool = False,
batch_size: int = 40,
) -> dict:
"""
C4 主流程:
1. 找 A 稿 docx + 检查 skeleton 先决条件
2. parse_a_segments 解析分段(优先读 skeleton,fallback 正则)
3. align_lines_to_segments 分批对齐 normal 段
4. hard_validate 硬校验
5. compose_segment_text 纯规则拼接正文
6. punctuate_segment AI标点恢复 + 硬校验
7. render_docx 出融合A稿.docx
8. write_alignment_csv 保留痕
返回统计 dict
"""
out_dir = Path(output_dir)
out_dir.mkdir(parents=True, exist_ok=True)
# ---- 找 A 稿 docx ----
# 只认原始 A 稿:排除本流程产出的"融合"稿与人工"批改"稿,
# 否则重跑 compose 时 glob 可能把自己的输出当成输入 A 稿。
a_candidates = [
f
for f in out_dir.glob("*.docx")
if not f.name.startswith("~$")
and "融合" not in f.name
and "_批改" not in f.name
]
if not a_candidates:
raise FileNotFoundError(
f"在 {out_dir} 中找不到 docx 文件\n"
f"目录下的 docx 文件: {[d.name for d in out_dir.glob('*.docx')]}"
)
a_path = a_candidates[0]
print(f"[fusion_align] A稿: {a_path}")
# ---- 前置检查: skeleton ----
skeleton_path = out_dir / f"{episode_id}_a_skeleton.json"
if skeleton_path.exists():
print(f"[fusion_align] 分段骨架: {skeleton_path}")
else:
# 检查是否 ep001 格式(旧正则能过)
doc = Document(str(a_path))
paras = [para.text.strip() for para in doc.paragraphs if para.text.strip()]
first_non_empty = paras[1] if len(paras) > 1 else ""
if SEG_HEADER_PATTERN.match(first_non_empty):
print(
"[fusion_align] 未找到 skeleton.json,检测到旧版【】格式,走正则 fallback"
)
else:
raise FileNotFoundError(
f"未找到分段骨架文件: {skeleton_path}\n"
f"请先运行: doco skeleton --episode-id {episode_id} --a-script {a_path}"
)
# ---- 读 B 稿 ----
b_path = out_dir / "融合B稿.txt"
if not b_path.exists():
raise FileNotFoundError(f"融合B稿.txt 不存在: {b_path}")
b_lines = parse_timed_lines(b_path)
print(f"[fusion_align] B稿: {len(b_lines)} 行")
# ---- 解析 A 稿 ----
a_data = parse_a_segments(str(a_path))
title = a_data["title"]
segments = a_data["segments"]
print(f"[fusion_align] 标题: {title}")
print(f"[fusion_align] 全局段数: {len(segments)}")
for seg in segments:
type_str = seg.get("type", "normal")
print(
f" seg_id={seg['seg_id']} [{type_str}] | {seg['header']} "
f"| body_len={len(seg['body'])}"
)
# ---- 对齐（只对 normal 段） ----
cache_dir = out_dir / ".c4_cache"
alignment, audit_logs, normal_segs = align_lines_to_segments(
b_lines,
segments,
no_ai=no_ai,
batch_size=batch_size,
cache_dir=cache_dir,
)
g2n, n2g = _build_normal_seg_id_map(normal_segs)
# ---- 硬校验 (normal 段数) ----
hard_validate(alignment, len(b_lines), len(normal_segs))
print("[fusion_align] [OK] 硬校验通过")
# ---- 拼接各段正文(裸文本) + 标点恢复 ----
seg_b_texts: Dict[int, list] = {}
for rec, bl in zip(alignment, b_lines):
normal_sid = rec["seg_id"]
global_sid = n2g.get(normal_sid, normal_sid)
seg_b_texts.setdefault(global_sid, []).append(bl["text"])
seg_texts = []
punct_results: Dict[int, bool] = {}
for seg in segments:
sid = seg["seg_id"]
if seg.get("type") == "break":
seg_texts.append("")
continue
lines = seg_b_texts.get(sid, [])
if not lines:
seg_texts.append("")
punct_results[sid] = True
continue
bare_text = compose_segment_text(lines)
ref_body = seg["body"]
punct_text, punct_ok = punctuate_segment(
bare_text,
ref_body,
cache_dir=cache_dir,
seg_id=sid,
)
seg_texts.append(punct_text)
punct_results[sid] = punct_ok
# ---- 出 docx ----
# 出稿命名 = 原始 A 稿名 + "_融合A稿"(不覆盖原始定稿,且带本期日期/节目/编导信息)
docx_path = out_dir / f"{a_path.stem}_融合A稿.docx"
render_docx(title, segments, seg_texts, str(docx_path))
print(f"[fusion_align] 融合A稿: {docx_path}")
# ---- 出 CSV ----
csv_path = out_dir / "c4_alignment.csv"
write_alignment_csv(
str(csv_path), segments, alignment, audit_logs, punct_results
)
print(f"[fusion_align] 留痕 CSV: {csv_path}")
# ---- 统计 ----
seg_line_counts: Dict[int, int] = {}
for rec, bl in zip(alignment, b_lines):
global_sid = n2g.get(rec["seg_id"], rec["seg_id"])
seg_line_counts[global_sid] = seg_line_counts.get(global_sid, 0) + 1
empty_segs = 0
low_conf_segs = 0
punct_failed_segs = 0
for seg in segments:
sid = seg["seg_id"]
if seg.get("type") == "break":
continue
count = seg_line_counts.get(sid, 0)
if count == 0:
empty_segs += 1
else:
min_c = min(
rec["confidence"]
for rec, bl in zip(alignment, b_lines)
if n2g.get(rec["seg_id"], rec["seg_id"]) == sid
)
if min_c < 0.6:
low_conf_segs += 1
if not punct_results.get(sid, True):
punct_failed_segs += 1
stats = {
"total_lines": len(b_lines),
"segment_count": len(segments),
"seg_line_counts": seg_line_counts,
"empty_segments": empty_segs,
"low_confidence_segments": low_conf_segs,
"audit_forced_lines": len(audit_logs),
"punct_failed_segs": punct_failed_segs,
"docx_path": str(docx_path),
"csv_path": str(csv_path),
}
print(f"\n[fusion_align] === 统计 ===")
print(f" 总行数: {stats['total_lines']}")
print(f" 段数: {stats['segment_count']}")
print(f" 空段数: {stats['empty_segments']}")
print(f" 低把握段数: {stats['low_confidence_segments']}")
print(f" 单调修正行数: {stats['audit_forced_lines']}")
print(f" 标点回退段数: {stats['punct_failed_segs']}")
print(f" 各段行数分布:")
for seg in segments:
sid = seg["seg_id"]
if seg.get("type") == "break":
print(f" [{sid:2d}] [隔断] {seg['header']}")
continue
count = seg_line_counts.get(sid, 0)
flag = ""
if count == 0:
flag = " [空段]"
elif count > 0:
min_c = min(
(rec["confidence"] for rec, bl in zip(alignment, b_lines)
if n2g.get(rec["seg_id"], rec["seg_id"]) == sid),
default=1.0,
)
if min_c < 0.6:
flag = f" [低把握 min_c={min_c:.2f}]"
punct_flag = " [标点回退]" if not punct_results.get(sid, True) else ""
print(f" [{sid:2d}] {seg['header']}: {count} 行{flag}{punct_flag}")
return stats
Reference in New Issue Copy Permalink