diff --git a/cca/CLAUDE.md b/cca/CLAUDE.md index fa0925d..c3e47d8 100644 --- a/cca/CLAUDE.md +++ b/cca/CLAUDE.md @@ -4,9 +4,9 @@ ## 🔖 状态栏 (STATUS — 每次结束 session 前必须更新这三行) -- **最后更新**：Claude Code（动手开发）｜ 2026-07-04 -- **当前状态一句话**：脚本版流水线 v3 完成（绝对时间戳+严格校对不润色），等制片人周一大洋系统验证 SRT 导入效果。 -- **下一个动手的人从这里开始**：读完本文件，运行 `python cca_pipeline.py --help` 了解用法。凭证需填入 `cca/.env`（XFYUN + DEEPSEEK）。可选下一步：用 `--audio` 跑真实 ASR（带热词注入）看是否比缓存版效果更好。 +- **最后更新**：Claude Code（动手开发）｜ 2026-07-05 +- **当前状态一句话**：CCA 唱词助手已部署至腾讯云（http://101.42.29.217/cca.html）。完整流程可用：上传音频+A稿 → ASR+AI校对 → 审稿台（含查找替换）→ 生成SRT下载。WAV自动转MP3。进入内测阶段。 +- **下一个动手的人从这里开始**：读完本文件。线上地址 http://101.42.29.217/cca.html（从配音首页"唱词助手"按钮进入）。服务器凭证见 `/workspace/military_tech_voice/backend/.env`。本地调试：`python -X utf8 cca_pipeline.py --asr-cache output/asr_raw.json --script "data/重走战争老路的日本军备（A稿）.docx" --output-dir output_v6`。 --- @@ -51,7 +51,7 @@ ## 2. 技术栈与运行方式 - **语言**：Python（与主项目 backend 对齐） -- **前端**：Web 界面（编导审稿台需要双屏对比 UI），技术选型待定（可复用主项目 React 或轻量方案） +- **前端**：纯 HTML/CSS/JS 单页（cca.html），暗色主题匹配 lanhao 配音系统，无框架依赖 - **ASR**：讯飞开放平台 录音文件转写标准版（已有 API Key，与 doco 共用同一套讯飞凭证） - **AI**：LLM 用于两处——① 从 A 稿提取专有名词词典；② ASR 稿校对（的地得、引号、错别字） - **输出格式**：SRT（大洋系统兼容格式，有样本参照） @@ -62,14 +62,17 @@ ## 3. 当前进度 -- **已完成至**：脚本版流水线 v3 跑通——绝对时间戳 + 严格校对（只改错别字/术语/填充词，不润色不调序）+ AI折行 + 5段SRT。 -- **正在做**：无，等制片人周一在大洋系统验证 SRT 导入效果。 -- **卡点/待解**：无硬卡点。可选优化：用 `--audio` 跑真实 ASR（热词注入生效）看是否在转写层就避免"建制→舰只"类错误。 +- **已完成至**：腾讯云部署完成 + 审稿台（含查找替换）上线。流水线 v6 + Web 审稿台 + WAV 自动转码。进入内测。 +- **正在做**：无（等待内测反馈）。 +- **卡点/待解**：无硬卡点。已知残留：ASR 切句边界跨越固定搭配（如"第二次/世界大战"）暂无法修复——需要跨句拆词检测（可做但需更大短语词典）。 --- ## 4. 已完成（只追加，最新在上） +- [2026-07-05] **腾讯云部署 + 审稿台上线**：① deploy/cca_route.py（Flask 蓝图，6 个 API 端点：upload/status/review/save/generate/download）；② deploy/cca.html（4 步单页流程：上传→处理→审稿→下载，暗色主题匹配配音系统）；③ 审稿台功能：左栏 ASR 原文 vs 右栏 AI 校对稿对比、逐句编辑确认、仅看修改过滤、全部确认、**查找替换**（Ctrl+H，支持逐个/全部替换+高亮定位）；④ WAV 大文件自动 ffmpeg 转 MP3（解决讯飞上传超时）；⑤ 服务器架构：Nginx→静态 HTML + Flask:5000，CCA 源码在 `/workspace/military_tech_voice/backend/cca_src/`，蓝图注册在 `app/routes/cca.py`。首页已加"唱词助手"入口按钮。 +- [2026-07-05] **脚本版流水线 v6（14项审片修复+专家段识别）**：① term_normalizer 新增：波浪号→"到"、顿号→空格、小数点保留（0.9马赫）、同音字映射表（建制→舰只/沉默→沉没/继承→击沉）、引号上下文感知（日向号加引号但日向级不加）、书名号补全（《军事科技》）；② ai_proofreader 新增：speaker 角色自动识别（解说/主持/专家）、专家采访增强 Prompt（严格删除嗯/呃/啊/这个/那么/就是说等口头语）、的地得规则大幅加强（+大量示例）、数字照抄 ASR 规则；③ ai_line_breaker 新增：引号不跨屏后处理（≤6字引号内容不拆两行）、极短行合并（≤3字+时长<1秒→并入相邻行）、极短句合并间隔放宽（≤4字句间隔阈值1200ms）；④ line_breaker 修改：clean_punctuation 保留小数点、顿号→空格；⑤ pipeline 新增 Step 2.5 校对后二次正则修复（兜住 AI 校对引入的新问题）。 +- [2026-07-04] **脚本版流水线 v5（四层纠错体系+折行优化）**：基于制片人逐帧审片反馈，解决 10+ 问题。新增四层纠错：① term_normalizer.py 正则层（型号短横线 F-15J/武器昵称引号"鱼鹰"/中文数字修复，零token）；② 校对 Prompt 升级（+代词他→它、+的地得纠错、+A稿权重规则）；③ 折行 Prompt 升级（+禁忌字规则"的了着过"不开头、+主谓宾拆分规则、+不可拆词示例）；④ 折行后处理三层（超长切分→禁忌字修复→拆词检测 _fix_split_words）。新增短句合并预处理（解决专家气口碎片句问题）。 - [2026-07-04] **脚本版流水线 v3（绝对时间戳+严格校对）**：① 恢复绝对时间戳（方便在大洋时间线对位）；② 重写校对 Prompt——铁律：只改错别字/同音字、术语格式、口语填充词，绝不润色/调序/替换实词；③ 校对效果：60 处修正（vs v2 的 100 处，去掉了过度修改）。输出目录 `output_v3/`。 - [2026-07-04] **脚本版流水线 v2 真实测试通过**：① 热词提取（规则+AI，127个术语）→ ② 讯飞ASR（94秒完成25MB音频，357句）→ ③ AI校对（DeepSeek，修正同音字/术语格式/口语填充，"建制"→"舰只"等）→ ④ AI折行（语义断句，98%行≤14字）→ ⑤ 5段SRT输出（段内相对时间戳，从00:00:00开始）。新增：hotword_extractor.py（热词提取）、ai_proofreader.py（AI校对）、ai_line_breaker.py（AI折行）。 - [2026-07-04] 脚本版流水线骨架完成：① asr_client.py（讯飞ASR适配，从doco复用）；② line_breaker.py（折行引擎，≤14字/语义断句/空白行检测）；③ srt_writer.py（大洋格式SRT输出）；④ segment_splitter.py（节目结构切分：导视/正片×3/预告）；⑤ cca_pipeline.py（主入口串联全流程）。本地测试全部通过。 @@ -80,12 +83,15 @@ ## 5. 待办（按优先级） - [x] ~~PRD / 业务规则确认~~ → 已在对话中完成（2026-07-04） -- [x] ~~脚本版流水线~~ → v3 完成（绝对时间戳+严格校对），等大洋验证 -- [x] ~~AI 校对层~~ → 已实现（两层防线：热词注入+DeepSeek 严格校对） -- [ ] **大洋系统验证**：周一导入 SRT 测试兼容性 +- [x] ~~脚本版流水线~~ → v5 完成（四层纠错+折行优化+短句合并） +- [x] ~~AI 校对层~~ → 已实现（四层防线：热词→正则→AI校对→折行后处理） +- [x] ~~制片人审片第一轮~~ → 10+ 问题全部解决 +- [x] ~~编导审稿台~~ → 已完成（查找替换+逐句对比+编辑确认，2026-07-05） +- [x] ~~部署至腾讯云~~ → 已完成（http://101.42.29.217/cca.html，2026-07-05） +- [ ] **内测反馈收集**：同事试用中，等待反馈 +- [ ] **大洋系统验证**：导入 SRT 测试兼容性 - [ ] **热词注入真实 ASR 测试**：用 `--audio` 跑完整流水线（非缓存），验证热词在转写层的效果 -- [ ] **编导审稿台**：双屏对比 UI、差异高亮、编导确认/手改交互（第二步） -- [ ] **部署至 lanhao 配音 2.0**：先跑起来测试 +- [ ] **首页入口按钮可能被遮挡**：index.html 已添加代码但可能需要样式调整（Ctrl+F5 刷新后可见） --- @@ -118,6 +124,11 @@ - [2026-07-04] **AI 校对严格纪律**：只允许改三类——① 错别字/同音字 ② 术语格式（F-15J）③ 口语填充词删除。绝不润色、绝不调序、绝不替换实词。ASR 是已录音频的转写，改不了内容。 - [2026-07-04] **两层 ASR 纠错防线**：第一层=热词注入（预防，让讯飞在转写时就认对专有名词）；第二层=AI 校对（修正，用 A 稿上下文判断同音字）。两层互补。 - [2026-07-04] **LLM 选型已定**：校对+折行+热词提取统一用 DeepSeek（deepseek-chat），性价比最优。 +- [2026-07-04] **四层纠错体系**（v5 确立）：① 热词注入（讯飞ASR层，预防中文同音字）→ ② term_normalizer 正则后处理（型号短横线/引号/中文数字，零token确定性替换）→ ③ AI 校对（DeepSeek，同音字/代词/的地得/术语/填充词）→ ④ 折行后处理（超长切分+禁忌字修复+拆词检测）。 +- [2026-07-04] **折行三条铁律**：① 词语不可拆分到两屏 ② "的了着过地得和与及或"不能作为新行第一个字 ③ 主谓宾优先折为"主语(折行)谓语+宾语"。 +- [2026-07-04] **短句合并策略**：ASR 按音频静音切句，专家气口会产生碎片短句（2-5字）。折行前先合并：≤8字+间隔<800ms→合并为一个语义单元再送AI折行。>2s静音仍插空白行。 +- [2026-07-04] **A稿与ASR权重规则**：内容冲突时ASR优先（配音员可能改过措辞），但专有名词格式/写法按A稿（如F-35A、"鱼鹰"引号）。 +- [2026-07-04] **国家代词不改**：指代国家时口语用"他"是可接受的，不纠正；只纠正指代武器/舰艇/飞机/导弹时的"他→它"。 --- @@ -127,13 +138,13 @@ - **"拍词"术语解释**：折行稿（去标点、按规则断行的文稿）+ 时间戳对位 = 拍词。传统靠人工实时听拍，CCA 用 ASR 时间戳代替。 - **与 doco 的区别**：doco 是"播出后"整理终版文稿（三方融合）；CCA 是"剪辑后、播出前"生成唱词字幕（ASR→校对→SRT）。两者共用讯飞 ASR 能力，但流程目的完全不同。 - **数据样本**：`data/` 下有 A 稿 docx + mp3 音频 + 3 个人工拍词 SRT（对应正片三段）。 -- **代码文件**：`src/` 下是核心流水线代码（asr_client / line_breaker / ai_line_breaker / ai_proofreader / srt_writer / segment_splitter / hotword_extractor），入口 `cca_pipeline.py`。 -- **凭证**：需在 `cca/.env` 中填写 `XFYUN_APP_ID`、`XFYUN_SECRET_KEY`、`DEEPSEEK_API_KEY`。 -- **输出目录**：`output/`（ASR 缓存 + v1 输出）、`output_v2/`（相对时间戳版）、`output_v3/`（绝对时间戳+严格校对，当前最优）。 +- **代码文件**：`src/` 下是核心流水线代码（asr_client / line_breaker / ai_line_breaker / ai_proofreader / srt_writer / segment_splitter / hotword_extractor / term_normalizer），入口 `cca_pipeline.py`。`deploy/` 下是部署文件（cca_route.py Flask 蓝图 + cca.html 前端页面 + deploy_to_server.py 部署脚本）。 +- **凭证**：本地需在 `cca/.env` 中填写；服务器凭证在 `/workspace/military_tech_voice/backend/.env`（讯飞大号 + DeepSeek）。 +- **服务器架构**：腾讯云 101.42.29.217，Nginx:80 → 静态文件(/var/www/voice/) + Flask:5000 代理(/api/)。CCA 源码部署在 `/workspace/military_tech_voice/backend/cca_src/`，任务数据在 `cca_data/`。Flask 无 systemd 服务，重启方式：`fuser -k 5000/tcp && cd backend && source venv/bin/activate && nohup python3 -m app.main > /tmp/flask_cca.log 2>&1 &`。 +- **输出目录**：`output/`（ASR 缓存 + v1 输出）、`output_v2/`~`output_v6/`（各版本输出）。 - **运行命令示例**： - - 从缓存跑（调试校对/折行）：`python -X utf8 cca_pipeline.py --asr-cache output/asr_raw.json --script "data/重走战争老路的日本军备（A稿）.docx" --output-dir output_v3` - - 完整流水线（含真实 ASR）：`python -X utf8 cca_pipeline.py --audio "data/重走战争老路的日本军备A0.mp3" --script "data/重走战争老路的日本军备（A稿）.docx" --output-dir output_v4` -- **时间压力**：这两天要出可用版本，功能不复杂但要快。 + - 从缓存跑（调试校对/折行）：`python -X utf8 cca_pipeline.py --asr-cache output/asr_raw.json --script "data/重走战争老路的日本军备（A稿）.docx" --output-dir output_v6` + - 完整流水线（含真实 ASR）：`python -X utf8 cca_pipeline.py --audio "data/重走战争老路的日本军备A0.mp3" --script "data/重走战争老路的日本军备（A稿）.docx" --output-dir output_v7` --- @@ -143,5 +154,7 @@ - [x] ~~大洋 SRT 样本文件~~ → data/ 下已有 3 个真实样本 - [x] ~~音频格式~~ → 纯人声 MP3，无需预处理 - [x] ~~LLM 选型~~ → DeepSeek（deepseek-chat），已验证效果好、价格低 -- [ ] 前端审稿台技术选型（第二步再定） -- [ ] 大洋系统 SRT 导入兼容性（周一验证） +- [x] ~~的地得纠错~~ → 已加入校对 Prompt（v5） +- [x] ~~前端审稿台技术选型~~ → 纯 HTML/JS 单页，无框架（2026-07-05） +- [ ] 大洋系统 SRT 导入兼容性（待验证） +- [ ] 跨句固定搭配拆词（"第二次/世界大战"类问题，需大短语词典，优先级低） diff --git a/cca/cca_pipeline.py b/cca/cca_pipeline.py index 63f3baf..69ada63 100644 --- a/cca/cca_pipeline.py +++ b/cca/cca_pipeline.py @@ -34,6 +34,7 @@ from srt_writer import write_srt, ms_to_srt_time from segment_splitter import split_into_segments from hotword_extractor import extract_hotwords from ai_proofreader import proofread_batch +from term_normalizer import normalize_terms def main(): @@ -104,6 +105,11 @@ def main(): print(f"[流水线] ASR 共 {len(sentences)} 句") + # ====== Step 1.5: 术语格式化（正则后处理，不耗 token）====== + if script_text: + print("[流水线] 术语格式化（型号短横线/武器昵称引号/中文数字）...") + sentences = normalize_terms(sentences, script_text) + # ====== Step 2: AI 校对 ====== if use_ai and not args.no_proofread and script_text: print("[流水线] AI 校对中 (DeepSeek)...") @@ -111,6 +117,12 @@ def main(): elif not script_text and not args.no_proofread and use_ai: print("[流水线] 未提供A稿(--script)，跳过AI校对") + # ====== Step 2.5: 校对后二次正则修复（兜住AI校对引入的新问题）====== + if script_text: + from term_normalizer import normalize_terms as post_normalize + print("[流水线] 校对后二次正则修复...") + sentences = post_normalize(sentences, script_text) + # ====== Step 3: 节目结构切分 ====== print("[流水线] 切分节目结构...") segments = split_into_segments(sentences) diff --git a/cca/deploy/cca.html b/cca/deploy/cca.html new file mode 100644 index 0000000..4343e54 --- /dev/null +++ b/cca/deploy/cca.html @@ -0,0 +1,910 @@ + + + + + + CCA 唱词助手 - 军事科技 + + + + +
+
+ +
+ 唱词字幕 +
+
+ + +
+
+
+
+
+
+ + +
+
+

上传素材

+

+ 上传编导 A 稿和粗编人声音频，AI 将自动完成 ASR 转写、术语校对、的地得纠错和口头语清除 +

+
+
+
🎤
+
人声音频
+
MP3 / WAV，纯人声（必传）
+
+ +
+
+
📄
+
A 稿文件
+
DOCX / TXT（强烈建议上传，用于术语校对）
+
+ +
+
+
+ +
+
+
+ + +
+
+
+ +
准备中...
+
请勿关闭页面，ASR 转写通常需要 1-2 分钟
+
+
+
+ + +
+
+
+ 审稿台 + + 总计 0 句 | + AI 修改 0 处 | + 已确认 0 + +
+
+ + + + + +
+
+ +
+
+ 查找 + +
+ + + +
+
+
+ 替换 + +
+ + + +
+
+
+
+ AI 已修改 + 已确认 + 点击文字可直接编辑，右侧勾选确认 +
+
+
+ + +
+
+
+
✅
+
SRT 字幕文件生成完成
+
+
+ + +
+
+
+
+ + + +
+ + + + diff --git a/cca/deploy/cca_route.py b/cca/deploy/cca_route.py new file mode 100644 index 0000000..87ff02b --- /dev/null +++ b/cca/deploy/cca_route.py @@ -0,0 +1,330 @@ +# -*- coding: utf-8 -*- +""" +CCA 唱词助手 — Flask 路由 +POST /api/cca/upload 上传 A稿+音频，启动流水线 +GET /api/cca/status/ 轮询任务状态 +GET /api/cca/review/ 获取审稿数据 +POST /api/cca/save/ 保存编导修改 +POST /api/cca/generate/ 生成最终 SRT +GET /api/cca/download/ 下载 SRT zip +""" + +import json +import os +import sys +import uuid +import threading +import traceback +import zipfile +from io import BytesIO +from pathlib import Path +from datetime import datetime + +from flask import Blueprint, request, jsonify, send_file + +bp = Blueprint('cca', __name__, url_prefix='/api/cca') + +# 运行时数据目录 +CCA_DATA_DIR = Path('/workspace/military_tech_voice/backend/cca_data') +CCA_DATA_DIR.mkdir(parents=True, exist_ok=True) + +# CCA 源码目录 +CCA_SRC_DIR = Path('/workspace/military_tech_voice/backend/cca_src') +if str(CCA_SRC_DIR) not in sys.path: + sys.path.insert(0, str(CCA_SRC_DIR)) + +# 任务状态存储（内存，重启丢失无所谓——编导重新上传即可） +tasks = {} + + +def _get_task(task_id): + t = tasks.get(task_id) + if not t: + return None + return t + + +@bp.route('/upload', methods=['POST']) +def upload(): + """接收 A稿 + 音频，创建任务""" + if 'audio' not in request.files: + return jsonify({'error': '请上传音频文件'}), 400 + + audio_file = request.files['audio'] + script_file = request.files.get('script') + + if not audio_file.filename: + return jsonify({'error': '音频文件为空'}), 400 + + task_id = datetime.now().strftime('%Y%m%d_%H%M%S') + '_' + uuid.uuid4().hex[:6] + task_dir = CCA_DATA_DIR / task_id + task_dir.mkdir(parents=True, exist_ok=True) + + # 保存音频 + audio_ext = os.path.splitext(audio_file.filename)[1] or '.mp3' + audio_path = task_dir / f'audio{audio_ext}' + audio_file.save(str(audio_path)) + + # 保存 A稿 + script_path = None + if script_file and script_file.filename: + script_ext = os.path.splitext(script_file.filename)[1] or '.docx' + script_path = task_dir / f'script{script_ext}' + script_file.save(str(script_path)) + + tasks[task_id] = { + 'id': task_id, + 'status': 'uploaded', + 'progress': '文件已上传，准备处理...', + 'audio_path': str(audio_path), + 'script_path': str(script_path) if script_path else None, + 'created_at': datetime.now().isoformat(), + 'error': None, + 'asr_sentences': None, + 'proofread_sentences': None, + 'review_data': None, + 'final_srt_dir': None, + } + + # 启动后台处理 + thread = threading.Thread(target=_run_pipeline, args=(task_id,), daemon=True) + thread.start() + + return jsonify({'task_id': task_id, 'status': 'processing'}) + + +def _run_pipeline(task_id): + """后台运行 CCA 流水线""" + task = tasks[task_id] + try: + task['status'] = 'processing' + task['progress'] = '正在提取热词...' + + audio_path = task['audio_path'] + script_path = task['script_path'] + + # WAV/大文件 → MP3 压缩（讯飞上传大文件容易超时） + if audio_path.lower().endswith('.wav'): + import subprocess + mp3_path = audio_path.rsplit('.', 1)[0] + '.mp3' + task['progress'] = '正在压缩音频（WAV→MP3）...' + subprocess.run( + ['ffmpeg', '-i', audio_path, '-b:a', '128k', '-y', mp3_path], + capture_output=True, timeout=300, + ) + if os.path.exists(mp3_path) and os.path.getsize(mp3_path) > 0: + audio_path = mp3_path + task['audio_path'] = mp3_path + + from hotword_extractor import extract_hotwords, read_docx_text, read_text_file + from asr_client import transcribe, parse_result + from term_normalizer import normalize_terms + from ai_proofreader import proofread_batch + from segment_splitter import split_into_segments + + # Step 0: 热词提取 + hot_words = [] + script_text = "" + if script_path: + hot_words = extract_hotwords(script_path, use_ai=True) + ext = os.path.splitext(script_path)[1].lower() + if ext == '.docx': + script_text = read_docx_text(script_path) + else: + script_text = read_text_file(script_path) + + task['progress'] = f'热词提取完成({len(hot_words)}个)，正在 ASR 转写...' + + # Step 1: ASR + sentences, raw_json = transcribe(audio_path, hot_words=hot_words if hot_words else None) + + # 缓存 ASR 原始结果 + task_dir = Path(audio_path).parent + cache_path = task_dir / 'asr_raw.json' + with open(cache_path, 'w', encoding='utf-8') as f: + f.write(raw_json) + + task['progress'] = f'ASR 完成({len(sentences)}句)，正在术语格式化...' + + # 保存 ASR 原始句子（校对前，供 diff 对比） + asr_original = [(bg, ed, text, spk) for bg, ed, text, spk in sentences] + + # Step 1.5: 术语格式化 + if script_text: + sentences = normalize_terms(sentences, script_text) + + task['progress'] = '术语格式化完成，正在 AI 校对...' + + # Step 2: AI 校对 + if script_text: + sentences = proofread_batch(sentences, script_text) + + # Step 2.5: 校对后二次正则 + if script_text: + sentences = normalize_terms(sentences, script_text) + + task['progress'] = 'AI 校对完成，正在准备审稿数据...' + + # 保存校对后的句子 + task['asr_sentences'] = asr_original + task['proofread_sentences'] = sentences + + # 构建审稿数据：逐句对比 + review_items = [] + for i, ((bg, ed, orig_text, spk), (_, _, proof_text, _)) in enumerate( + zip(asr_original, sentences) + ): + has_change = orig_text != proof_text + review_items.append({ + 'index': i, + 'start_ms': bg, + 'end_ms': ed, + 'speaker_id': spk, + 'original': orig_text, + 'corrected': proof_text, + 'edited': proof_text, + 'has_change': has_change, + 'confirmed': not has_change, + }) + + task['review_data'] = review_items + task['status'] = 'review' + task['progress'] = '审稿数据就绪，请编导审阅确认' + + # 同时保存到磁盘（防丢） + review_path = task_dir / 'review_data.json' + with open(review_path, 'w', encoding='utf-8') as f: + json.dump(review_items, f, ensure_ascii=False, indent=2) + + except Exception as e: + task['status'] = 'error' + err_msg = str(e) + if '余额不足' in err_msg or 'insufficient' in err_msg.lower() or '10317' in err_msg: + task['error'] = '讯飞录音文件转写余额不足，请联系管理员充值' + else: + task['error'] = f'处理出错: {err_msg}' + task['progress'] = task['error'] + traceback.print_exc() + + +@bp.route('/status/', methods=['GET']) +def status(task_id): + task = _get_task(task_id) + if not task: + return jsonify({'error': '任务不存在'}), 404 + return jsonify({ + 'task_id': task_id, + 'status': task['status'], + 'progress': task['progress'], + 'error': task['error'], + }) + + +@bp.route('/review/', methods=['GET']) +def review(task_id): + task = _get_task(task_id) + if not task: + return jsonify({'error': '任务不存在'}), 404 + if task['status'] not in ('review', 'completed'): + return jsonify({'error': '任务尚未就绪', 'status': task['status']}), 400 + return jsonify({ + 'task_id': task_id, + 'items': task['review_data'], + }) + + +@bp.route('/save/', methods=['POST']) +def save(task_id): + """保存编导的修改（自动保存用）""" + task = _get_task(task_id) + if not task: + return jsonify({'error': '任务不存在'}), 404 + + data = request.get_json() + edits = data.get('items', []) + + for edit in edits: + idx = edit.get('index') + if idx is not None and 0 <= idx < len(task['review_data']): + task['review_data'][idx]['edited'] = edit.get('edited', task['review_data'][idx]['edited']) + task['review_data'][idx]['confirmed'] = edit.get('confirmed', True) + + return jsonify({'ok': True}) + + +@bp.route('/generate/', methods=['POST']) +def generate(task_id): + """用编导确认后的文本生成最终 SRT""" + task = _get_task(task_id) + if not task: + return jsonify({'error': '任务不存在'}), 404 + if task['status'] not in ('review', 'completed'): + return jsonify({'error': '任务状态不对'}), 400 + + try: + from ai_line_breaker import process_sentences_with_ai + from srt_writer import write_srt, ms_to_srt_time + from segment_splitter import split_into_segments + + # 用编导确认后的文本重建句子列表 + confirmed_sentences = [] + for item in task['review_data']: + text = item['edited'] + confirmed_sentences.append(( + item['start_ms'], item['end_ms'], text, item['speaker_id'] + )) + + # 切分节目结构 + segments = split_into_segments(confirmed_sentences) + + # 折行 + 生成 SRT + task_dir = Path(task['audio_path']).parent + srt_dir = task_dir / 'srt_output' + srt_dir.mkdir(exist_ok=True) + + srt_files = [] + for seg_name, seg_sentences in segments: + if not seg_sentences: + continue + subtitle_lines = process_sentences_with_ai(seg_sentences) + srt_path = srt_dir / f'{seg_name}.srt' + write_srt(subtitle_lines, str(srt_path)) + srt_files.append(str(srt_path)) + + task['final_srt_dir'] = str(srt_dir) + task['status'] = 'completed' + task['progress'] = f'生成完成，共 {len(srt_files)} 个 SRT 文件' + + return jsonify({ + 'ok': True, + 'srt_count': len(srt_files), + 'message': task['progress'], + }) + + except Exception as e: + return jsonify({'error': f'生成 SRT 出错: {str(e)}'}), 500 + + +@bp.route('/download/', methods=['GET']) +def download(task_id): + """下载 SRT zip 包""" + task = _get_task(task_id) + if not task: + return jsonify({'error': '任务不存在'}), 404 + if not task.get('final_srt_dir'): + return jsonify({'error': 'SRT 尚未生成'}), 400 + + srt_dir = Path(task['final_srt_dir']) + srt_files = sorted(srt_dir.glob('*.srt')) + if not srt_files: + return jsonify({'error': '无 SRT 文件'}), 404 + + # 打包 zip + buf = BytesIO() + with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf: + for srt_file in srt_files: + zf.write(srt_file, srt_file.name) + buf.seek(0) + + filename = f'唱词字幕_{task_id}.zip' + return send_file(buf, mimetype='application/zip', as_attachment=True, download_name=filename) diff --git a/cca/deploy/deploy_to_server.py b/cca/deploy/deploy_to_server.py new file mode 100644 index 0000000..b2044ed --- /dev/null +++ b/cca/deploy/deploy_to_server.py @@ -0,0 +1,246 @@ +# -*- coding: utf-8 -*- +""" +CCA 部署脚本 — 通过 paramiko 上传文件到腾讯云服务器 +""" +import sys +sys.stdout.reconfigure(encoding='utf-8') + +import os +import paramiko +from pathlib import Path + +HOST = '101.42.29.217' +PORT = 22 +USER = 'root' +PASS = 'liutong65' + +CCA_ROOT = Path(__file__).resolve().parent.parent +SRC_DIR = CCA_ROOT / 'src' +DEPLOY_DIR = CCA_ROOT / 'deploy' + +# 服务器目标路径 +SERVER_CCA_SRC = '/workspace/military_tech_voice/backend/cca_src' +SERVER_FRONTEND = '/workspace/military_tech_voice/frontend' +SERVER_WWW = '/var/www/voice' +SERVER_BACKEND = '/workspace/military_tech_voice/backend' +SERVER_ROUTES = f'{SERVER_BACKEND}/app/routes' + +# 需要上传的 src 模块 +SRC_MODULES = [ + 'asr_client.py', + 'line_breaker.py', + 'ai_line_breaker.py', + 'ai_proofreader.py', + 'term_normalizer.py', + 'hotword_extractor.py', + 'srt_writer.py', + 'segment_splitter.py', +] + + +def connect(): + ssh = paramiko.SSHClient() + ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + ssh.connect(HOST, PORT, USER, PASS) + sftp = ssh.open_sftp() + return ssh, sftp + + +def run(ssh, cmd): + print(f' $ {cmd}') + _, stdout, stderr = ssh.exec_command(cmd) + out = stdout.read().decode('utf-8', errors='replace').strip() + err = stderr.read().decode('utf-8', errors='replace').strip() + if out: + print(f' {out[:500]}') + if err: + print(f' [stderr] {err[:500]}') + return out + + +def upload(sftp, local_path, remote_path): + print(f' ↑ {Path(local_path).name} → {remote_path}') + sftp.put(str(local_path), remote_path) + + +def main(): + print('=== CCA 部署开始 ===\n') + ssh, sftp = connect() + print('[1/7] 连接成功\n') + + # Step 2: 创建 cca_src 目录 + 上传源码 + print('[2/7] 上传 CCA 源码模块...') + run(ssh, f'mkdir -p {SERVER_CCA_SRC}') + for module in SRC_MODULES: + local = SRC_DIR / module + if local.exists(): + upload(sftp, local, f'{SERVER_CCA_SRC}/{module}') + else: + print(f' ⚠ 跳过（不存在）: {module}') + print() + + # Step 3: 上传 cca_route.py 到 app/routes/ + print('[3/7] 上传 cca_route.py...') + run(ssh, f'mkdir -p {SERVER_ROUTES}') + upload(sftp, DEPLOY_DIR / 'cca_route.py', f'{SERVER_ROUTES}/cca.py') + # 确保 __init__.py 存在 + run(ssh, f'touch {SERVER_ROUTES}/__init__.py') + print() + + # Step 4: 上传 cca.html 到前端目录 + print('[4/7] 上传 cca.html...') + upload(sftp, DEPLOY_DIR / 'cca.html', f'{SERVER_FRONTEND}/cca.html') + upload(sftp, DEPLOY_DIR / 'cca.html', f'{SERVER_WWW}/cca.html') + print() + + # Step 5: 配置 .env（追加 CCA 相关变量） + print('[5/7] 配置 .env...') + env_path = f'{SERVER_BACKEND}/.env' + existing_env = '' + try: + with sftp.open(env_path, 'r') as f: + existing_env = f.read().decode('utf-8', errors='replace') + except FileNotFoundError: + pass + + env_additions = [] + if 'XFYUN_APP_ID' not in existing_env: + env_additions.append('# === CCA 唱词助手凭证 ===') + env_additions.append('# 讯飞录音文件转写（账号1-默认）') + env_additions.append('XFYUN_APP_ID=4c423e35') + env_additions.append('XFYUN_SECRET_KEY=b9e0b97d5dda072c9b4b8fb59e7e3d22') + env_additions.append('# 讯飞备用账号2') + env_additions.append('# XFYUN_APP_ID=52ae3024') + env_additions.append('# XFYUN_SECRET_KEY=d65de0eb282a4339e2b1e14fd119e42e') + if 'DEEPSEEK_API_KEY' not in existing_env: + env_additions.append('# DeepSeek（校对+折行+热词）') + env_additions.append('DEEPSEEK_API_KEY=sk-01a7868a88a04ab494e4f05c1f3f06e2') + + if env_additions: + with sftp.open(env_path, 'a') as f: + f.write('\n' + '\n'.join(env_additions) + '\n') + print(' .env 已追加 CCA 凭证') + else: + print(' .env 已有 CCA 凭证，跳过') + print() + + # Step 6: 注册 CCA 蓝图到 Flask main.py + print('[6/7] 注册 CCA 蓝图...') + main_py_path = f'{SERVER_BACKEND}/app/main.py' + with sftp.open(main_py_path, 'r') as f: + main_content = f.read().decode('utf-8', errors='replace') + + if 'cca' not in main_content.lower(): + # 找到最后一个 register_blueprint 的位置，在其后追加 + lines = main_content.split('\n') + insert_idx = -1 + for i, line in enumerate(lines): + if 'register_blueprint' in line: + insert_idx = i + + if insert_idx >= 0: + indent = ' ' # 匹配现有缩进 + # 检查现有蓝图注册的缩进 + existing_line = lines[insert_idx] + indent = existing_line[:len(existing_line) - len(existing_line.lstrip())] + + cca_lines = [ + '', + f'{indent}# CCA 唱词助手', + f'{indent}from app.routes.cca import bp as cca_bp', + f'{indent}app.register_blueprint(cca_bp)', + ] + for j, cca_line in enumerate(cca_lines): + lines.insert(insert_idx + 1 + j, cca_line) + + new_content = '\n'.join(lines) + with sftp.open(main_py_path, 'w') as f: + f.write(new_content) + print(' 已注册 CCA 蓝图') + else: + print(' ⚠ 未找到 register_blueprint，请手动注册') + else: + print(' CCA 蓝图已注册，跳过') + print() + + # Step 7: 安装依赖 + 修改 index.html + 重启 + print('[7/7] 安装依赖、修改首页、重启服务...') + + # 安装 Python 依赖 + run(ssh, f'{SERVER_BACKEND}/venv/bin/pip install openai python-docx 2>&1 | tail -5') + + # 修改 index.html 添加 CCA 入口按钮 + index_path = f'{SERVER_FRONTEND}/index.html' + with sftp.open(index_path, 'r') as f: + index_content = f.read().decode('utf-8', errors='replace') + + if '唱词助手' not in index_content and 'cca.html' not in index_content: + # 在导航栏中找到合适位置插入按钮 + # 典型位置：header 或 nav 区域的最后一个链接之后 + if ' 或 前插入 + for marker in ['', '']: + if marker in index_content: + btn_html = ' 唱词助手\n' + index_content = index_content.replace(marker, btn_html + ' ' + marker) + break + else: + # 备用：在 body 开头加一个浮动按钮 + btn_html = '\n' + index_content = index_content.replace('', '\n' + btn_html) + + with sftp.open(index_path, 'w') as f: + f.write(index_content) + + # 同步到 /var/www/voice/ + run(ssh, f'cp {SERVER_FRONTEND}/index.html {SERVER_WWW}/index.html') + print(' index.html 已添加唱词助手入口') + else: + print(' index.html 已有唱词助手入口，跳过') + + # 增大 Nginx 上传限制（音频文件可能较大） + nginx_conf = '/etc/nginx/nginx.conf' + with sftp.open(nginx_conf, 'r') as f: + nginx_content = f.read().decode('utf-8', errors='replace') + + if 'client_max_body_size' not in nginx_content: + nginx_content = nginx_content.replace( + 'http {', + 'http {\n client_max_body_size 200m;' + ) + with sftp.open(nginx_conf, 'w') as f: + f.write(nginx_content) + run(ssh, 'nginx -t && nginx -s reload') + print(' Nginx 已增大上传限制至 200MB') + else: + print(' Nginx 上传限制已配置') + + # 重启 Flask 服务 + print('\n 重启 Flask...') + # 查找并重启 Flask 进程 + run(ssh, 'pkill -f "flask run" || pkill -f "gunicorn" || pkill -f "python.*app" || true') + # 给进程时间退出 + import time + time.sleep(2) + + # 检查服务启动方式 + service_out = run(ssh, 'systemctl list-units --type=service | grep -i voice || systemctl list-units --type=service | grep -i flask || true') + if 'voice' in service_out.lower() or 'flask' in service_out.lower(): + service_name = service_out.split()[0] if service_out else '' + if service_name: + run(ssh, f'systemctl restart {service_name}') + else: + # 直接后台启动 + run(ssh, f'cd {SERVER_BACKEND} && source venv/bin/activate && nohup python -m flask run --host=0.0.0.0 --port=5000 > /tmp/flask_cca.log 2>&1 &') + + print() + print('=== CCA 部署完成 ===') + print(f'访问地址: http://{HOST}/cca.html') + print(f'API 地址: http://{HOST}/api/cca/') + + sftp.close() + ssh.close() + + +if __name__ == '__main__': + main() diff --git a/cca/src/ai_line_breaker.py b/cca/src/ai_line_breaker.py index ae5ad05..211687a 100644 --- a/cca/src/ai_line_breaker.py +++ b/cca/src/ai_line_breaker.py @@ -31,13 +31,37 @@ SILENCE_THRESHOLD_MS = 2000 SYSTEM_PROMPT = """你是电视节目唱词字幕的折行助手。你的任务是将一段文字按照以下规则折成多行： -规则： +**基本规则：** 1. 每行最多14个字（中文字符、英文字母、数字各算1个字） -2. 去掉逗号、句号、感叹号、问号、顿号、分号、冒号、省略号等标点，只保留引号（""''「」）和书名号（《》） +2. 去掉逗号、句号、感叹号、问号、分号、冒号、省略号等标点，只保留引号（""''）和书名号（《》） 3. 折行要符合语义和阅读习惯，不能把词语切断 4. 每行不一定要凑满14字，可以是5字、8字、10字等，关键是语义完整 5. 保持原文内容不变，不增不减不改字 +**折行禁忌（硬规则，违反即错误）：** +- 禁止把一个词语拆到两行（"过程"不能变成"过"在行末、"程"在下行首；"实际上"不能拆开） +- 禁止"的""了""着""过""地""得"作为新行的第一个字 +- 禁止"和""与""及""或"作为新行的第一个字 +- 禁止拆断固定搭配（如"F-35A和F-35B"保持同行，"RQ-4全球鹰"保持同行） +- 禁止把动宾结构拆成：主语+动词(折行)宾语。应折成：主语(折行)动词+宾语 +- **引号不跨屏**：当引号""内的内容≤6个字时，上引号和下引号必须在同一行，不允许拆到两行。例如"鱼鹰"、"日向"号 必须保持在同一行内 + +**折行优先级（按此顺序选择折点）：** +1. 最优：在句子成分边界折（主语|谓语+宾语，状语|主句） +2. 次优：在并列分句之间折 +3. 可接受：在长定语与中心词之间折（但"的"必须跟前面，不能落到下一行开头） +4. 最末：硬切词间（仅当以上都超14字时） + +**示例：** +- 正确："重塑自身军事力量版图的" / "野心与企图" +- 错误："重塑自身军事力量版图" / "的野心与企图"（"的"开头） +- 正确："日本国会参议院" / "今天上午表决通过了" / "防卫省设置法修正案等法案" +- 错误："日本国会参议院今天上午" / "表决通过" / "了防卫省设置法修正案等法案"（"了"开头，且拆断动宾） +- 正确："2015年美国国务院" / "批准向日本军售RQ-4全球鹰" +- 错误："2015年美国国务院批准" / "向日本军售RQ-4全球鹰"（主+谓(折行)宾） +- 正确："所以发展"日向"级" / "直升机护卫舰"（引号内容不拆开） +- 错误："所以发展"日向" / "级直升机护卫舰"（引号被拆到两屏） + 输出格式：每行一句，不加序号，不加标点（引号和书名号除外）。""" USER_PROMPT_TEMPLATE = """请将以下文字折行（每行≤14字，去标点保引号，按语义断句）： @@ -125,6 +149,253 @@ def ai_break_batch(texts: List[str], client: OpenAI) -> List[List[str]]: return all_results +# 常见不可拆分的双字词（高频，不求全，兜底关键场景） +# 这些词如果被折行拆到两屏，观众体验极差 +_COMMON_WORDS = set([ + "过程", "中间", "实际", "日本", "美国", "中国", "问题", "发展", + "军事", "武器", "装备", "能力", "力量", "防御", "进攻", "导弹", + "战斗", "战机", "战争", "国家", "历史", "世界", "方面", "系统", + "技术", "任务", "目标", "计划", "项目", "部署", "改装", "航母", + "自卫", "海军", "空军", "陆军", "预算", "宪法", "和平", "安全", + "基础", "措施", "结构", "性能", "速度", "射程", "重量", "面积", + "时候", "之后", "以后", "之前", "目前", "现在", "所以", "因此", + "但是", "虽然", "而且", "或者", "如果", "这个", "那个", "已经", + "可以", "应该", "需要", "能够", "开始", "成为", "通过", "进行", + "实现", "提升", "完成", "建设", "研发", "生产", "采购", "引进", +]) + + +def _fix_split_words(lines: List[str]) -> List[str]: + """ + 检测并修复被拆到两行的词语。 + 如果行末1字+下行首1字构成常见双字词，把末字移到下行。 + """ + if len(lines) <= 1: + return lines + + fixed = list(lines) + changed = True + max_iterations = 3 # 防止无限循环 + + while changed and max_iterations > 0: + changed = False + max_iterations -= 1 + new_fixed = [fixed[0]] + + for j in range(1, len(fixed)): + prev_line = new_fixed[-1] + curr_line = fixed[j] + + if not prev_line or not curr_line: + new_fixed.append(curr_line) + continue + + # 检查行末字+下行首字是否构成词 + pair = prev_line[-1] + curr_line[0] + if pair in _COMMON_WORDS: + # 把上一行末字移到当前行首 + new_fixed[-1] = prev_line[:-1] + curr_line = prev_line[-1] + curr_line + changed = True + + # 如果上一行变空了，删掉 + if not new_fixed[-1].strip(): + new_fixed.pop() + + new_fixed.append(curr_line) + + fixed = new_fixed + + # 清理：过滤空行，检查是否有超长行需要重切 + from line_breaker import break_sentence + result = [] + for line in fixed: + line = line.strip() + if not line: + continue + if len(line) > MAX_CHARS_SOFT: + result.extend(break_sentence(line)) + else: + result.append(line) + + return result if result else lines + + +def _fix_quote_split(lines: List[str]) -> List[str]: + """ + 修复引号被拆到两屏的问题。 + 当上引号"在一行、下引号"在下一行，且引号内内容≤6字时，合并到同一行。 + """ + if len(lines) <= 1: + return lines + + from line_breaker import break_sentence + + fixed = [lines[0]] + i = 1 + while i < len(lines): + prev = fixed[-1] + curr = lines[i] + + # 检查：上一行有"但没有配对的"，当前行有" + if "“" in prev and "”" not in prev and "”" in curr: + # 找到上引号位置，计算引号内内容长度 + quote_start = prev.rfind("“") + # 引号内容 = 上一行从"开始的部分 + 当前行到"为止的部分 + quote_end_in_curr = curr.index("”") + quoted_content = prev[quote_start+1:] + curr[:quote_end_in_curr] + if len(quoted_content) <= 6: + # 合并这两行 + merged = prev + curr + if len(merged) <= MAX_CHARS_SOFT: + fixed[-1] = merged + else: + # 合并后超长，重新折行 + fixed[-1:] = break_sentence(merged) + i += 1 + continue + + fixed.append(curr) + i += 1 + + return fixed + + +def _merge_tiny_subtitle(result: List[Tuple[int, int, str]]) -> List[Tuple[int, int, str]]: + """ + 合并极短字幕行（≤3字且时长<1秒）到相邻行。 + 避免"东海"这种两个字单独闪一屏。 + """ + if len(result) <= 1: + return result + + merged = [] + skip_next = False + + for i, (bg, ed, text) in enumerate(result): + if skip_next: + skip_next = False + continue + + duration_ms = ed - bg + is_tiny = len(text) <= 3 and duration_ms < 1000 and text.strip() + + if is_tiny: + # 尝试与下一行合并 + if i + 1 < len(result) and result[i+1][2].strip(): + next_bg, next_ed, next_text = result[i+1] + combined = text + next_text + if len(combined) <= MAX_CHARS_SOFT: + merged.append((bg, next_ed, combined)) + skip_next = True + continue + + # 尝试与上一行合并 + if merged and merged[-1][2].strip(): + prev_bg, prev_ed, prev_text = merged[-1] + combined = prev_text + text + if len(combined) <= MAX_CHARS_SOFT: + merged[-1] = (prev_bg, ed, combined) + continue + + merged.append((bg, ed, text)) + + return merged + + +MERGE_THRESHOLD_CHARS = 8 # ≤8字的句子考虑合并 +MERGE_GAP_MS = 800 # 句间间隔<800ms才合并（>800ms视为有意停顿） +MERGE_GAP_TINY_MS = 1200 # 极短句(≤4字)放宽间隔阈值 + + +def _merge_short_sentences( + sentences: List[Tuple[int, int, str, int]], +) -> List[Tuple[int, int, str, int]]: + """ + 合并碎片短句：专家气口造成的短碎 ASR 句，合并成完整语义单元再折行。 + + 规则: + - 连续的短句（≤8字）且间隔<800ms → 合并为一句 + - 遇到 >2s 静音 → 不合并（是真正的话题停顿） + - 如果某句已经≥14字 → 作为独立单元不参与合并 + - 合并后的句子时间戳取第一句起点到最后一句终点 + """ + if not sentences: + return [] + + from line_breaker import clean_punctuation + + merged = [] + buffer = [] # [(bg, ed, text, spk), ...] + + def flush_buffer(): + if not buffer: + return + if len(buffer) == 1: + merged.append(buffer[0]) + else: + # 合并 buffer 中所有句子 + bg = buffer[0][0] + ed = buffer[-1][1] + text = "".join(item[2] for item in buffer) + spk = buffer[0][3] + merged.append((bg, ed, text, spk)) + + for i, (bg, ed, text, spk) in enumerate(sentences): + cleaned = clean_punctuation(text) + + # 检查与 buffer 最后一句的间隔 + if buffer: + gap = bg - buffer[-1][1] + # 极短句(≤4字)用更宽松的间隔阈值，让气口碎片更容易合并 + threshold = MERGE_GAP_TINY_MS if len(cleaned) <= 4 else MERGE_GAP_MS + if gap > threshold: + flush_buffer() + buffer = [] + + # 如果当前句子很长（>28字），独立处理 + if len(cleaned) > MAX_CHARS * 2: + flush_buffer() + buffer = [] + merged.append((bg, ed, text, spk)) + continue + + # 如果当前句子中等偏长（15-28字） + if len(cleaned) > MAX_CHARS: + # 如果前面buffer里有极短句（≤5字），合并进来（专家气口碎片） + if buffer and all(len(clean_punctuation(item[2])) <= 5 for item in buffer): + buffer.append((bg, ed, text, spk)) + else: + flush_buffer() + buffer = [] + merged.append((bg, ed, text, spk)) + continue + + # 短句，看是否要合并 + if len(cleaned) <= MERGE_THRESHOLD_CHARS: + # 检查合并后是否太长 + buffer_text = "".join(clean_punctuation(item[2]) for item in buffer) + cleaned + if len(buffer_text) > MAX_CHARS * 3: # 合并后超过3行的量就太多了 + flush_buffer() + buffer = [(bg, ed, text, spk)] + else: + buffer.append((bg, ed, text, spk)) + else: + # 中等长度（9-14字），如果buffer有内容就合并进去，否则独立 + if buffer: + buffer_text = "".join(clean_punctuation(item[2]) for item in buffer) + cleaned + if len(buffer_text) <= MAX_CHARS * 3: + buffer.append((bg, ed, text, spk)) + else: + flush_buffer() + buffer = [(bg, ed, text, spk)] + else: + merged.append((bg, ed, text, spk)) + + flush_buffer() + return merged + + def process_sentences_with_ai( sentences: List[Tuple[int, int, str, int]], batch_size: int = 15, @@ -136,6 +407,7 @@ def process_sentences_with_ai( 输出: [(start_ms, end_ms, text), ...] 策略: + - 先合并碎片短句（专家气口造成的短碎ASR句） - ≤14 字：直接输出（去标点） - >14 字：批量调 AI 折行 - 句间 >2秒：插入空白行 @@ -145,6 +417,12 @@ def process_sentences_with_ai( if not sentences: return [] + # 短句合并预处理 + original_count = len(sentences) + sentences = _merge_short_sentences(sentences) + if len(sentences) != original_count: + print(f"[AI折行] 短句合并: {original_count} 句 → {len(sentences)} 句") + client = _create_client() result = [] @@ -198,7 +476,7 @@ def process_sentences_with_ai( else: lines = [cleaned] - # 后处理：AI 偶尔返回超长行，强制二次切分 + # 后处理1：AI 偶尔返回超长行，强制二次切分 from line_breaker import break_sentence final_lines = [] for line in lines: @@ -208,6 +486,35 @@ def process_sentences_with_ai( final_lines.append(line) lines = final_lines + # 后处理2：禁忌字开头修复（把禁忌字并入上一行） + FORBIDDEN_START = set("的了着过地得和与及或") + if len(lines) > 1: + fixed_lines = [lines[0]] + for ln in lines[1:]: + if ln and ln[0] in FORBIDDEN_START and fixed_lines: + # 把这个字并回上一行 + fixed_lines[-1] = fixed_lines[-1] + ln[0] + remainder = ln[1:] + if remainder: + # 检查上一行是否超长了 + if len(fixed_lines[-1]) > MAX_CHARS_SOFT: + # 需要重新切分合并后的文本 + merged = fixed_lines[-1] + remainder + fixed_lines[-1:] = break_sentence(merged) + else: + fixed_lines.append(remainder) + else: + fixed_lines.append(ln) + lines = fixed_lines + + # 后处理3：拆词检测（行末+下行首构成常见双字词 → 调整折点） + if len(lines) > 1: + lines = _fix_split_words(lines) + + # 后处理4：引号不跨屏（≤6字的引号内容不拆到两行） + if len(lines) > 1: + lines = _fix_quote_split(lines) + # 为子行分配时间戳 total_chars = sum(len(l) for l in lines) duration = ed - bg @@ -221,4 +528,7 @@ def process_sentences_with_ai( result.append((current_ms, line_end, line)) current_ms = line_end + # 全局后处理：合并极短字幕行（≤3字+时长<1秒→并入相邻行） + result = _merge_tiny_subtitle(result) + return result diff --git a/cca/src/ai_proofreader.py b/cca/src/ai_proofreader.py index de8b6e0..767f29c 100644 --- a/cca/src/ai_proofreader.py +++ b/cca/src/ai_proofreader.py @@ -7,11 +7,13 @@ AI 校对器 — ASR 稿与 A 稿比对 + 上下文纠错 - 军事术语规范化（"f15j"→"F-15J"） - 的/地/得纠错 - 去除口语填充词（"嗯""那个""就是说"） +- 专家采访段落强化去口头语 策略: - 将 ASR 全文 + A 稿全文一起发给 DeepSeek - AI 结合节目主题和上下文做纠错 - 返回修正后的句子列表 + 修改说明 +- 专家采访段落用增强版 Prompt，更严格地删除口头语 """ import json @@ -37,21 +39,68 @@ PROOFREAD_SYSTEM_PROMPT = """你是电视军事节目《军事科技》的字幕 **铁律（违反任何一条都算失败）：** - ASR稿是已经录好的音频的转写，内容不能改——**绝不润色语句、绝不调整语序、绝不增删实词** -- 只修三类问题：① 错别字/同音字 ② 术语格式 ③ 口语填充词 -- 除这三类外的一切文字，原封不动照抄，一个字都不能动 -- A稿只用来判断"这个词在本期节目的语境下应该是哪个字"，不能把ASR稿往A稿的措辞上靠 +- 只修下列允许的几类问题，除此之外一个字都不能动 +- **A稿与ASR内容冲突时ASR优先**（配音员可能改过措辞），但专有名词的正确写法/格式按A稿 +- **数字表达照抄ASR原文**：不要参考A稿调整数字的位置、格式或表述方式。ASR说"马赫数0.9"就保持"马赫数0.9"，不要改成A稿的"0.9马赫" -**允许修的三类：** -1. **同音字/错别字**（ASR听错的字）：如"建制"→"舰只"、"舰手"→"舰艏"、"继承"→"击沉"、"空花弹"→"滑翔弹" -2. **术语格式**：英文型号大小写+连字符（"f15j"→"F-15J"、"v22"→"V-22"、"rq四"→"RQ-4"） -3. **口语填充词删除**：只删"嗯""呃""唉""啊""呢""那个""就是说""这个"这类纯填充词。如果"这个"后面紧跟名词作指示代词（"这个导弹"），保留不删 +**允许修的类别：** +1. **同音字/错别字**（ASR听错的字）：如"建制"→"舰只"、"舰手"→"舰艏"、"继承"→"击沉"、"空花弹"→"滑翔弹"、"沉默"→"沉没"（指船只） +2. **代词纠错**：武器装备/导弹/飞机/舰艇等的代词应为"它"而非"他"。注意：指代国家时不改（国家口语中用"他"是可接受的）。只纠正明确指代物件（武器、军舰、飞机、导弹）的情况 +3. **的/地/得纠错**（重要！ASR无法区分三个"de"，你必须逐句检查并修正）： + - **"的"用在名词前**（形容词/名词 + 的 + 名词）：强大的性能、日本的军备、重要的舰只 + - **"地"用在动词前**（副词 + 地 + 动词）：不断地进行、持续地推动、快速地发展、正式地把、大规模地改装、积极地推进、不断地扩大、明确地表示 + - **"得"用在补语前**（动词 + 得 + 补语）：发展得很快、做得很好、打得很准 + - 判断方法：看"de"后面跟的是名词还是动词——跟动词就用"地"，跟名词就用"的"，是评价/程度补语就用"得" + - 常见错误模式："不断的进行"→"不断地进行"、"持续的推动"→"持续地推动"、"正式的把"→"正式地把"、"大力的发展"→"大力地发展" +4. **术语格式**：英文型号大小写+连字符（"f15j"→"F-15J"、"v22"→"V-22"、"rq四"→"RQ-4"） +5. **中文数字保留**：ASR可能把"数十"转成"数10"、"几百"转成"几100"——必须改回中文写法 +6. **武器昵称引号**：如A稿中武器有引号昵称（"鱼鹰""战斧""全球鹰"），ASR中同一武器无引号时补上中文双引号 +7. **口语填充词删除**：只删"嗯""呃""唉""那个""就是说"这类纯填充词。"这个"后面紧跟名词作指示代词（"这个导弹"）时保留 **绝对不许做的（哪怕你觉得改了更好也不许）：** - 不许调整语序（"它在性质上就是"不许改成"它本质上就是"） - 不许替换实词（"不是那么特别的顺利"不许改成"不太顺利"） +- 不许参考A稿的数字表达方式来改ASR的数字写法 - 不许增删标点来改变句子结构 - 不许把口语化表达改成书面语 -- 不许根据A稿的措辞替换ASR中意思相同但用词不同的表达 +- 不许根据A稿的措辞替换ASR中意思相同但用词不同的表达（如A稿"陆续订购"，ASR说"先后采购"→保持"先后采购"） + +**输出格式：** +JSON数组，每个元素：{"id": 编号, "original": "原文", "corrected": "修正后", "changes": "修改说明（无修改写空字符串）"} +只输出JSON，不要其他内容。""" + + +PROOFREAD_EXPERT_SYSTEM_PROMPT = """你是电视军事节目《军事科技》的字幕校对专家。你将收到两份材料： +1. **ASR稿**：语音识别的转写结果，带有时间编号，是字幕的基础。**本批全部来自专家采访段落** +2. **A稿**：编导写的节目文稿（仅包含解说词，不包含专家采访内容——专家说的话A稿里没有） + +你的任务是校对 ASR 稿中的**语音识别错误**，同时**严格清除专家的口头语**。 + +**铁律（违反任何一条都算失败）：** +- ASR稿是已经录好的音频的转写，内容不能改——**绝不润色语句、绝不调整语序、绝不增删实词** +- 只修下列允许的几类问题，除此之外一个字都不能动 +- 由于是专家采访，A稿中没有对应内容，所以**不要用A稿措辞替换专家的话**，A稿只用于确认专有名词写法 + +**允许修的类别：** +1. **同音字/错别字**（ASR听错的字）：如"建制"→"舰只"、"舰手"→"舰艏"、"继承"→"击沉"、"沉默"→"沉没"（指船只） +2. **代词纠错**：武器装备/导弹/飞机/舰艇等的代词应为"它"而非"他"。指代国家时不改 +3. **的/地/得纠错**（重要！ASR无法区分三个"de"，你必须逐句检查并修正）： + - **"的"用在名词前**（形容词/名词 + 的 + 名词） + - **"地"用在动词前**（副词 + 地 + 动词）：不断地进行、持续地推动、正式地把、大规模地改装 + - **"得"用在补语前**（动词 + 得 + 补语）：发展得很快 + - 常见错误："不断的进行"→"不断地进行"、"持续的推动"→"持续地推动" +4. **术语格式**：英文型号大小写+连字符 +5. **口语填充词删除（专家采访重点！必须严格执行）**： + - **必删**：嗯、呃、唉、啊（句首或句中作语气词时）、那个、这个（非指示代词时）、那么（非表示程度时）、就是说、应该说、可以说、怎么说呢、相对来讲、相对来说 + - **判断"这个/那个"**：紧跟具体名词="指示代词"保留（"这个导弹"）；单独出现或后面是虚词/停顿=口头语删除（"这个呢它是"→删"这个"、"发展这个日向级"→删"这个"） + - **判断"啊"**：句首"啊射程""啊这个"=口头语删除；"啊"在感叹句末尾=保留（极少出现在专家采访中） + - **判断"那么"**："那么大""那么快"=程度副词保留；"那么它就是"=口头语删除 +6. **数字表达照抄ASR原文**，不参考A稿 + +**绝对不许做的：** +- 不许调整语序、替换实词、把口语化改书面语 +- 不许用A稿的措辞替换专家的话（专家说的内容A稿没有，不存在"参考"关系） +- 不许删除有意义的词（只删纯口头语填充词） **输出格式：** JSON数组，每个元素：{"id": 编号, "original": "原文", "corrected": "修正后", "changes": "修改说明（无修改写空字符串）"} @@ -76,6 +125,62 @@ def _create_client(): ) +def identify_speakers( + sentences: List[Tuple[int, int, str, int]], +) -> Dict[int, str]: + """ + 识别每个 speaker_id 的角色。 + + 规则（基于《军事科技》节目结构）： + - 找到说"各位观众你们好"或"欢迎收看军事科技"的 speaker → 主持人（也是解说配音员） + - 导视段（最早出现的）speaker 如果和主持人不同 → 也是解说（录音环境不同导致分裂） + - 剩余的 speaker → 专家/其他（统一按"专家采访"对待，加强去口头语） + + 返回: {speaker_id: "narration"|"host"|"expert"} + """ + if not sentences: + return {} + + speaker_texts: Dict[int, str] = {} + speaker_first_appear: Dict[int, int] = {} + for i, (bg, ed, text, spk) in enumerate(sentences): + if spk not in speaker_texts: + speaker_texts[spk] = "" + speaker_first_appear[spk] = i + speaker_texts[spk] += text + + roles: Dict[int, str] = {} + + # 找主持人：说过"各位观众你们好"或"欢迎收看军事科技" + host_spk = None + for spk, text in speaker_texts.items(): + if "各位观众" in text or "欢迎收看" in text or "主持人" in text: + host_spk = spk + roles[spk] = "host" + break + + # 最早出现的 speaker 是解说（导视段配音员） + earliest_spk = min(speaker_first_appear, key=speaker_first_appear.get) + if earliest_spk not in roles: + roles[earliest_spk] = "narration" + + # 如果主持人和解说是不同 speaker，两个都标记 + # 如果相同，那就是同一个人（标为 narration 即可） + if host_spk is not None and host_spk == earliest_spk: + roles[host_spk] = "narration" + + # 剩余的全部标为专家/其他 + for spk in speaker_texts: + if spk not in roles: + roles[spk] = "expert" + + role_summary = {spk: f"{role}({len([s for s in sentences if s[3]==spk])}句)" + for spk, role in roles.items()} + print(f"[校对] Speaker 角色识别: {role_summary}") + + return roles + + def proofread_batch( asr_sentences: List[Tuple[int, int, str, int]], script_text: str, @@ -83,81 +188,87 @@ def proofread_batch( ) -> List[Tuple[int, int, str, int]]: """ 对 ASR 句子列表做 AI 校对。 - - 输入: - asr_sentences: [(start_ms, end_ms, text, speaker_id), ...] - script_text: A稿全文 - batch_size: 每批处理的句子数 - - 返回: - 校对后的句子列表，格式同输入 + 专家采访段落使用增强版 Prompt（更严格的口头语清除）。 """ if not asr_sentences: return [] client = _create_client() - - # A稿截取（太长的话截前8000字，够提供上下文了） script_truncated = script_text[:8000] if len(script_text) > 8000 else script_text - corrected_sentences = list(asr_sentences) # 浅拷贝 + # 识别说话人角色 + speaker_roles = identify_speakers(asr_sentences) + + corrected_sentences = list(asr_sentences) total_changes = 0 - for batch_start in range(0, len(asr_sentences), batch_size): - batch = asr_sentences[batch_start:batch_start + batch_size] - batch_end = batch_start + len(batch) + # 按角色分组处理：专家用增强 Prompt，其余用标准 Prompt + expert_indices = [] + normal_indices = [] + for i, (bg, ed, text, spk) in enumerate(asr_sentences): + if speaker_roles.get(spk) == "expert": + expert_indices.append(i) + else: + normal_indices.append(i) - # 构建 ASR 文本（带编号） - asr_lines = [] - for i, (bg, ed, text, spk) in enumerate(batch): - asr_lines.append(f"[{i+1}] {text}") - asr_text = "\n".join(asr_lines) + print(f"[校对] 解说/主持 {len(normal_indices)} 句, 专家采访 {len(expert_indices)} 句") - print(f"[校对] 处理第 {batch_start+1}-{batch_end} 句...") + def _process_batch(indices, system_prompt, label): + nonlocal total_changes + for batch_start in range(0, len(indices), batch_size): + batch_idx = indices[batch_start:batch_start + batch_size] - try: - resp = client.chat.completions.create( - model=os.environ.get("DEEPSEEK_MODEL", "deepseek-chat"), - messages=[ - {"role": "system", "content": PROOFREAD_SYSTEM_PROMPT}, - {"role": "user", "content": PROOFREAD_USER_TEMPLATE.format( - script_text=script_truncated, - asr_text=asr_text, - )}, - ], - temperature=0.1, - max_tokens=4000, - ) + asr_lines = [] + for seq, idx in enumerate(batch_idx): + asr_lines.append(f"[{seq+1}] {asr_sentences[idx][2]}") + asr_text = "\n".join(asr_lines) - result_text = resp.choices[0].message.content.strip() + print(f"[校对-{label}] 处理第 {batch_start+1}-{batch_start+len(batch_idx)} 句...") - # 尝试解析 JSON - # 去掉可能的 markdown 代码块标记 - if result_text.startswith("```"): - result_text = result_text.split("\n", 1)[1] - if result_text.endswith("```"): - result_text = result_text[:-3] - result_text = result_text.strip() + try: + resp = client.chat.completions.create( + model=os.environ.get("DEEPSEEK_MODEL", "deepseek-chat"), + messages=[ + {"role": "system", "content": system_prompt}, + {"role": "user", "content": PROOFREAD_USER_TEMPLATE.format( + script_text=script_truncated, + asr_text=asr_text, + )}, + ], + temperature=0.1, + max_tokens=4000, + ) - corrections = json.loads(result_text) + result_text = resp.choices[0].message.content.strip() + if result_text.startswith("```"): + result_text = result_text.split("\n", 1)[1] + if result_text.endswith("```"): + result_text = result_text[:-3] + result_text = result_text.strip() - # 应用修正 - for item in corrections: - idx = item.get("id", 0) - 1 # 编号从1开始 - corrected = item.get("corrected", "") - changes = item.get("changes", "") + corrections = json.loads(result_text) - if 0 <= idx < len(batch) and corrected and changes: - original_idx = batch_start + idx - bg, ed, _, spk = corrected_sentences[original_idx] - corrected_sentences[original_idx] = (bg, ed, corrected, spk) - total_changes += 1 - print(f" 修正: '{item.get('original','')}' → '{corrected}' ({changes})") + for item in corrections: + seq = item.get("id", 0) - 1 + corrected = item.get("corrected", "") + changes = item.get("changes", "") - except json.JSONDecodeError as e: - print(f"[校对] JSON解析失败，跳过本批: {e}", file=sys.stderr) - except Exception as e: - print(f"[校对] 出错: {e}", file=sys.stderr) + if 0 <= seq < len(batch_idx) and corrected and changes: + original_idx = batch_idx[seq] + bg, ed, _, spk = corrected_sentences[original_idx] + corrected_sentences[original_idx] = (bg, ed, corrected, spk) + total_changes += 1 + print(f" 修正: '{item.get('original','')}' → '{corrected}' ({changes})") + + except json.JSONDecodeError as e: + print(f"[校对-{label}] JSON解析失败，跳过本批: {e}", file=sys.stderr) + except Exception as e: + print(f"[校对-{label}] 出错: {e}", file=sys.stderr) + + if normal_indices: + _process_batch(normal_indices, PROOFREAD_SYSTEM_PROMPT, "解说") + if expert_indices: + _process_batch(expert_indices, PROOFREAD_EXPERT_SYSTEM_PROMPT, "专家") print(f"[校对] 完成，共修正 {total_changes} 处") return corrected_sentences diff --git a/cca/src/line_breaker.py b/cca/src/line_breaker.py index 1515d2b..43d4a68 100644 --- a/cca/src/line_breaker.py +++ b/cca/src/line_breaker.py @@ -32,11 +32,18 @@ BREAK_PATTERNS = [ def clean_punctuation(text: str) -> str: - """去掉标点，保留引号类""" + """去掉标点，保留引号类。顿号替换为空格（唱词中并列词用空格分隔）。保留小数点。""" result = [] - for ch in text: + for i, ch in enumerate(text): if ch in KEEP_PUNCTUATION: result.append(ch) + elif ch == '、': + result.append(' ') + elif ch == '.' or ch == '．': + # 保留小数点（前后都是数字） + if i > 0 and i < len(text) - 1 and text[i-1].isdigit() and text[i+1].isdigit(): + result.append(ch) + # 其他句号/英文句点删掉 elif REMOVE_PUNCTUATION.match(ch): continue else: diff --git a/cca/src/term_normalizer.py b/cca/src/term_normalizer.py new file mode 100644 index 0000000..615baac --- /dev/null +++ b/cca/src/term_normalizer.py @@ -0,0 +1,282 @@ +# -*- coding: utf-8 -*- +""" +术语格式化器 — 正则后处理层（零 token 消耗） + +在 ASR 结果出来后、AI 校对之前执行。 +从 A 稿中提取正确的术语写法，构建映射表，对 ASR 文本做确定性替换。 + +解决的问题: +- 讯飞 ASR 丢失英文型号中的短横线（F-15J→F15J, V-22→V22） +- 武器昵称引号丢失（A稿有引号但ASR没带出来） +- 中文数字被转成阿拉伯数字（数十→数10） +- 数字范围符号（~→到） +- 顿号分隔词加空格 +- 小数点丢失修复（09马赫→0.9马赫） +- 军事领域高频同音字修正（建制→舰只等） +""" + +import re +from typing import List, Tuple, Dict, Set + + +# ======================================================================== +# 型号短横线修复 +# ======================================================================== + +MODEL_PATTERN = re.compile(r'[A-Z]{1,4}-\d{1,4}[A-Z]?(?:/[A-Z])?') + + +def _build_model_mapping(script_text: str) -> Dict[str, str]: + mapping = {} + models = set(MODEL_PATTERN.findall(script_text)) + for model in models: + no_hyphen = model.replace("-", "") + if no_hyphen != model: + mapping[no_hyphen] = model + return mapping + + +def _fix_model_hyphens(text: str, mapping: Dict[str, str]) -> str: + if not mapping: + return text + for no_hyphen in sorted(mapping.keys(), key=len, reverse=True): + correct = mapping[no_hyphen] + pattern = re.compile(re.escape(no_hyphen) + r'(?![A-Za-z0-9])') + text = pattern.sub(correct, text) + return text + + +# ======================================================================== +# 武器昵称引号修复（上下文感知版） +# ======================================================================== + +# 匹配 A 稿中 "xxx"号 / "xxx"级 / "xxx"型 / 单独 "xxx" 的模式 +QUOTED_WITH_SUFFIX = re.compile(r'“([^“”„‟""]{1,8})”([号级型式舰]?)') + + +def _build_quote_mapping(script_text: str) -> Dict[str, Set[str]]: + """ + 从 A 稿提取引号词及其后缀上下文。 + 返回 {词: {出现过的后缀集合}}，后缀为空字符串表示单独使用。 + 例: {"日向": {"号"}, "鱼鹰": {""}} 表示 A 稿有"日向"号但没有"日向"级，有单独的"鱼鹰" + """ + mapping: Dict[str, Set[str]] = {} + for match in QUOTED_WITH_SUFFIX.finditer(script_text): + word = match.group(1).strip() + suffix = match.group(2) + if 2 <= len(word) <= 6: + if word not in mapping: + mapping[word] = set() + mapping[word].add(suffix) + return mapping + + +def _check_bare_occurrences(script_text: str, word: str, suffixes: Set[str]) -> Set[str]: + """ + 检查 A 稿中该词的无引号出现，看哪些后缀组合是不加引号的。 + 例如 A 稿有 "日向级"（无引号），说明"日向级"不该加引号。 + """ + bare_suffixes = set() + for suffix in ["号", "级", "型", "式", "舰", ""]: + bare_pattern = word + suffix if suffix else word + quoted_pattern = f"“{word}”{suffix}" + # 在 A 稿中出现了无引号版本 且 没有对应的有引号版本 + if bare_pattern in script_text and quoted_pattern not in script_text: + bare_suffixes.add(suffix) + return bare_suffixes + + +def _fix_weapon_quotes(text: str, quote_mapping: Dict[str, Set[str]], script_text: str) -> str: + """对文本中无引号的武器昵称补上引号（上下文感知）""" + if not quote_mapping: + return text + for word in sorted(quote_mapping.keys(), key=len, reverse=True): + quoted_suffixes = quote_mapping[word] + bare_suffixes = _check_bare_occurrences(script_text, word, quoted_suffixes) + + # 对每个在 A 稿中确实带引号的后缀组合，在 ASR 文本中补引号 + for suffix in quoted_suffixes: + if suffix and suffix not in bare_suffixes: + # 匹配 "word+suffix"（无引号），替换为 "word"+suffix + target = word + suffix + replacement = f"“{word}”{suffix}" + pattern = re.compile( + r'(? str: + for pattern, replacement in CHINESE_NUM_FIXES: + text = pattern.sub(replacement, text) + return text + + +# ======================================================================== +# 数字范围符号修复：~ ～ → 到 +# ======================================================================== + +# 匹配 数字~数字 或 数字～数字 的模式 +RANGE_TILDE = re.compile(r'(\d)[~～](\d)') + + +def _fix_range_symbol(text: str) -> str: + return RANGE_TILDE.sub(r'\1到\2', text) + + +# ======================================================================== +# 顿号→空格（唱词中并列词用空格分隔） +# ======================================================================== + +def _fix_enumeration_pause(text: str) -> str: + return text.replace("、", " ") + + +# ======================================================================== +# 节目名称书名号补全 +# ======================================================================== + +# 需要带书名号的固定名称（节目名等） +# 格式: (裸名称, 带书名号版本) +BOOK_TITLE_NAMES = [ + ("军事科技", "《军事科技》"), + ("军事报道", "《军事报道》"), +] + + +def _fix_book_titles(text: str) -> str: + for bare, titled in BOOK_TITLE_NAMES: + # 只替换没有被书名号包围的裸名称 + pattern = re.compile(r'(? str: + text = LOST_DECIMAL_BEFORE_UNIT.sub(r'0.\1\2', text) + text = LOST_DECIMAL_AFTER_UNIT.sub(r'\g<1>0.\2', text) + return text + + +# ======================================================================== +# 军事领域高频同音字修正 +# ======================================================================== + +# 格式: (错误写法正则, 正确写法, A稿中应有的验证词) +# 只有当 A 稿中存在正确写法时才替换，避免误改 +HOMOPHONE_PAIRS = [ + # 海军 + ("建制", "舰只", "舰只"), + ("舰手", "舰艏", "舰艏"), + ("舰位", "舰尾", "舰尾"), + ("继承", "击沉", "击沉"), + ("沉默", "沉没", "沉没"), + ("空花弹", "滑翔弹", "滑翔弹"), + ("建支", "舰只", "舰只"), + ("坚支", "舰只", "舰只"), + # 其他 + ("符和", "符合", "符合"), + ("决意", "决议", "决议"), +] + + +def _build_homophone_mapping(script_text: str) -> Dict[str, str]: + mapping = {} + for wrong, correct, verify_word in HOMOPHONE_PAIRS: + if verify_word in script_text: + mapping[wrong] = correct + return mapping + + +def _fix_homophones(text: str, mapping: Dict[str, str]) -> str: + if not mapping: + return text + for wrong, correct in mapping.items(): + text = text.replace(wrong, correct) + return text + + +# ======================================================================== +# 主入口 +# ======================================================================== + +def normalize_terms( + sentences: List[Tuple[int, int, str, int]], + script_text: str, +) -> List[Tuple[int, int, str, int]]: + """ + 对 ASR 句子列表做术语格式化（确定性正则替换，不调 AI）。 + 在 ASR 结果出来后、AI 校对之前调用。 + """ + if not sentences: + return [] + if not script_text: + return list(sentences) + + model_mapping = _build_model_mapping(script_text) + quote_mapping = _build_quote_mapping(script_text) + homophone_mapping = _build_homophone_mapping(script_text) + + if model_mapping: + print(f"[术语格式化] 型号映射 {len(model_mapping)} 条: {list(model_mapping.items())[:5]}") + if quote_mapping: + print(f"[术语格式化] 引号昵称 {len(quote_mapping)} 个: {dict((k, list(v)) for k, v in list(quote_mapping.items())[:5])}") + if homophone_mapping: + print(f"[术语格式化] 同音字映射 {len(homophone_mapping)} 条: {list(homophone_mapping.items())[:5]}") + + result = [] + fix_count = 0 + for bg, ed, text, spk in sentences: + original = text + text = _fix_model_hyphens(text, model_mapping) + text = _fix_weapon_quotes(text, quote_mapping, script_text) + text = _fix_chinese_numbers(text) + text = _fix_range_symbol(text) + text = _fix_enumeration_pause(text) + text = _fix_lost_decimal(text) + text = _fix_homophones(text, homophone_mapping) + text = _fix_book_titles(text) + if text != original: + fix_count += 1 + result.append((bg, ed, text, spk)) + + print(f"[术语格式化] 完成，修正 {fix_count} 句") + return result