feat: CCA 唱词助手子项目 v3 — 脚本版流水线完成
新增 cca/ 子项目：编导A稿+人声音频 → ASR+AI校对+AI折行 → 5段SRT字幕。 - 讯飞录音文件转写标准版（热词注入） - DeepSeek AI校对（严格纪律：只改错别字/术语/填充词，不润色） - DeepSeek AI折行（语义断句，≤14字/行） - 节目结构自动切分（导视/正片×3/预告） - 绝对时间戳SRT输出（大洋系统兼容） Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
simonkoson
@@ -0,0 +1,7 @@
|
||||
# 讯飞录音文件转写标准版（不是大模型版）
|
||||
# 控制台: https://console.xfyun.cn/
|
||||
XFYUN_APP_ID=your_app_id
|
||||
XFYUN_SECRET_KEY=your_secret_key
|
||||
|
||||
# DeepSeek V4 Pro（用于专有名词提取 + 校对）
|
||||
DEEPSEEK_API_KEY=your_deepseek_api_key
|
||||
@@ -0,0 +1,7 @@
|
||||
.env
|
||||
output*/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
*.mp3
|
||||
*.wav
|
||||
pr/
|
||||
+147
@@ -0,0 +1,147 @@
|
||||
# 项目协作主控文件 (cca/CLAUDE.md)
|
||||
|
||||
---
|
||||
|
||||
## 🔖 状态栏 (STATUS — 每次结束 session 前必须更新这三行)
|
||||
|
||||
- **最后更新**：Claude Code（动手开发）｜ 2026-07-04
|
||||
- **当前状态一句话**：脚本版流水线 v3 完成（绝对时间戳+严格校对不润色），等制片人周一大洋系统验证 SRT 导入效果。
|
||||
- **下一个动手的人从这里开始**：读完本文件，运行 `python cca_pipeline.py --help` 了解用法。凭证需填入 `cca/.env`（XFYUN + DEEPSEEK）。可选下一步：用 `--audio` 跑真实 ASR（带热词注入）看是否比缓存版效果更好。
|
||||
|
||||
---
|
||||
|
||||
## 🤖 给 AI 的工作约定 (READ FIRST)
|
||||
|
||||
- 开始任何工作前，先读完本文件，特别是「状态栏」「交接备注」「关键决策」三节。
|
||||
- 完成一段工作后**增量更新**：「已完成」追加一条、「待办」勾掉或新增、必要时更新「关键决策」，最后改「状态栏」三行。不要重写整份文件，不要删历史，只追加/勾选。
|
||||
- 涉及发送、发布、删除、改权限、付款等不可逆操作前，先在对话里跟制片人确认。
|
||||
- **本项目着急用**——优先出可用版本，不追求完美架构。先跑通再优化。
|
||||
- **沟通**：全程简体中文；制片人不懂代码，解释说人话；给明确推荐不反复抛选项。
|
||||
|
||||
---
|
||||
|
||||
## 1. 项目概览
|
||||
|
||||
- **项目名**：CCA 唱词助手（ChangCi Assist）— 《军事科技》栏目编导/责编用的唱词字幕生成工具
|
||||
- **目标**：把编导 A 稿 + 粗编人声音频 → 经过 ASR + AI 校对 + 编导审稿 → 最终输出符合大洋系统的 SRT 字幕文件
|
||||
- **使用角色**：编导（提供 A 稿+音频、审稿确认）、责编（拿 SRT 拉到大洋上线）
|
||||
- **部署计划**：先独立部署在 lanhao 配音 2.0 网站测试，成熟后吸收进 TPS 主项目作为子功能
|
||||
|
||||
### 核心工作流
|
||||
|
||||
```
|
||||
编导A稿 ──→ AI提取专有名词词典
|
||||
↓
|
||||
粗编人声音频 + 词典 ──→ 讯飞ASR(录音文件转写标准版) ──→ 带时间戳ASR稿
|
||||
↓
|
||||
ASR稿 ⊕ A稿 比对
|
||||
↓
|
||||
┌─ AI自动修正(明显错别字)
|
||||
└─ 编导审稿台(拿不准的差异)
|
||||
↓
|
||||
B稿(定稿)
|
||||
↓
|
||||
按拍词规则折行 + 时间戳
|
||||
↓
|
||||
SRT文件(大洋格式)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 2. 技术栈与运行方式
|
||||
|
||||
- **语言**：Python（与主项目 backend 对齐）
|
||||
- **前端**：Web 界面（编导审稿台需要双屏对比 UI），技术选型待定（可复用主项目 React 或轻量方案）
|
||||
- **ASR**：讯飞开放平台 录音文件转写标准版（已有 API Key，与 doco 共用同一套讯飞凭证）
|
||||
- **AI**：LLM 用于两处——① 从 A 稿提取专有名词词典；② ASR 稿校对（的地得、引号、错别字）
|
||||
- **输出格式**：SRT（大洋系统兼容格式，有样本参照）
|
||||
- **美术风格**：参考 `ai-labeling/example/`（功能区划.jpg、幅面参考.jpg、页面风格.webp），与 TPS 主项目 UI 风格对齐
|
||||
- **凭证**：子项目自己 `.env`，不动主项目 `backend/.env`
|
||||
|
||||
---
|
||||
|
||||
## 3. 当前进度
|
||||
|
||||
- **已完成至**：脚本版流水线 v3 跑通——绝对时间戳 + 严格校对（只改错别字/术语/填充词，不润色不调序）+ AI折行 + 5段SRT。
|
||||
- **正在做**：无，等制片人周一在大洋系统验证 SRT 导入效果。
|
||||
- **卡点/待解**：无硬卡点。可选优化：用 `--audio` 跑真实 ASR（热词注入生效）看是否在转写层就避免"建制→舰只"类错误。
|
||||
|
||||
---
|
||||
|
||||
## 4. 已完成（只追加，最新在上）
|
||||
|
||||
- [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（主入口串联全流程）。本地测试全部通过。
|
||||
- [2026-07-04] 子项目立项：目录结构、CLAUDE.md、Brief、主项目寄存条建立。
|
||||
|
||||
---
|
||||
|
||||
## 5. 待办（按优先级）
|
||||
|
||||
- [x] ~~PRD / 业务规则确认~~ → 已在对话中完成（2026-07-04）
|
||||
- [x] ~~脚本版流水线~~ → v3 完成（绝对时间戳+严格校对），等大洋验证
|
||||
- [x] ~~AI 校对层~~ → 已实现（两层防线：热词注入+DeepSeek 严格校对）
|
||||
- [ ] **大洋系统验证**：周一导入 SRT 测试兼容性
|
||||
- [ ] **热词注入真实 ASR 测试**：用 `--audio` 跑完整流水线（非缓存），验证热词在转写层的效果
|
||||
- [ ] **编导审稿台**：双屏对比 UI、差异高亮、编导确认/手改交互（第二步）
|
||||
- [ ] **部署至 lanhao 配音 2.0**：先跑起来测试
|
||||
|
||||
---
|
||||
|
||||
## 6. 关键决策（跨 session 最易丢）
|
||||
|
||||
- [2026-07-04] **ASR 稿是权威底稿**：最终 SRT 必须和音频对应，ASR 稿（时间戳+内容）是基准。A 稿只提供专有名词写法和上下文参考，不能用 A 稿覆盖 ASR 的内容结构。
|
||||
- [2026-07-04] **讯飞用录音文件转写标准版**（与 doco 一样，不要大模型版）。热词偏置注入专有名词词典。
|
||||
- [2026-07-04] **编导审稿台是必要环节**：AI 拿不准的差异必须过编导确认，不能全自动。尊重编导创作主权（继承主项目设计哲学）。
|
||||
- [2026-07-04] **先独立部署再吸收**：先在 lanhao 配音 2.0 上跑通测试，成熟后并入 TPS 主项目。不等主项目进度。
|
||||
- [2026-07-04] **SRT 格式有样本**：data/ 下 3 个 SRT 是人工拍词导出的真实样本，作为格式参照。
|
||||
- [2026-07-04] **折行（拍词）规则**：
|
||||
- A. 每行 ≤ 14 字
|
||||
- B. ASR 中 >2 秒空白 → 插入空白行（屏幕清字幕）
|
||||
- C. 按语义断句（不机械凑满 14 字），符合阅读习惯
|
||||
- D. 去掉逗号/句号/感叹号等标点，只保留引号
|
||||
- [2026-07-04] **输出结构（5 个 SRT）**：
|
||||
- 导视 1 个 + 正片 3 个 + 下期预告 1 个
|
||||
- 正片 3 个拆分依据：按时长大致均分，优先在角色转换处（解说词→专家采访）切分
|
||||
- 5 个 SRT 时间戳连续拼接 = 音频总长
|
||||
- [2026-07-04] **节目结构标志词**：
|
||||
- 导视结束标志：听到"本期《军事科技》……"
|
||||
- 正片开始标志：听到"各位观众你们好，我是主持人蓝皓"
|
||||
- 正片结束标志：听到"好了观众朋友们，感谢您关注国防军事频道军事科技……"
|
||||
- 正片之后是下期预告（无固定话术）
|
||||
- [2026-07-04] **正片拆 3 个 SRT 的原因**：大洋字幕系统不稳定容易出错，责编提的需求。
|
||||
- [2026-07-04] **音频是纯人声**：已分离好的干净音频（无 BGM/音效），无需人声分离。
|
||||
- [2026-07-04] **先出脚本版**：命令行跑通流水线，审稿台第二步做。
|
||||
- [2026-07-04] **讯飞代码从 doco 复用**：`doco/src/doco/asr_adapter.py` 有完整签名/上传/轮询/解析逻辑，直接适配。
|
||||
- [2026-07-04] **SRT 用绝对时间戳**：每段 SRT 的时间戳是音频中的真实位置（不从 00:00:00 开始），方便在大洋时间线直接对位。制片人实测确认比相对时间戳好用。
|
||||
- [2026-07-04] **AI 校对严格纪律**：只允许改三类——① 错别字/同音字 ② 术语格式（F-15J）③ 口语填充词删除。绝不润色、绝不调序、绝不替换实词。ASR 是已录音频的转写，改不了内容。
|
||||
- [2026-07-04] **两层 ASR 纠错防线**：第一层=热词注入（预防，让讯飞在转写时就认对专有名词）；第二层=AI 校对（修正，用 A 稿上下文判断同音字）。两层互补。
|
||||
- [2026-07-04] **LLM 选型已定**：校对+折行+热词提取统一用 DeepSeek（deepseek-chat），性价比最优。
|
||||
|
||||
---
|
||||
|
||||
## 7. ⏩ 交接备注
|
||||
|
||||
- **立项背景**：编导剪完节目后需要给责编一份 SRT 唱词字幕文件。传统流程是责编拿折行稿到大洋系统里 1:1 实时"拍词"（播放视频、听到对应句子拍空格记录时间戳），效率极低。CCA 用 ASR 自动化这个过程。
|
||||
- **"拍词"术语解释**：折行稿（去标点、按规则断行的文稿）+ 时间戳对位 = 拍词。传统靠人工实时听拍，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/`（绝对时间戳+严格校对，当前最优）。
|
||||
- **运行命令示例**：
|
||||
- 从缓存跑（调试校对/折行）：`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`
|
||||
- **时间压力**：这两天要出可用版本，功能不复杂但要快。
|
||||
|
||||
---
|
||||
|
||||
## 8. 待确认 / 开放问题
|
||||
|
||||
- [x] ~~拍词规则~~ → 已确认（见关键决策）
|
||||
- [x] ~~大洋 SRT 样本文件~~ → data/ 下已有 3 个真实样本
|
||||
- [x] ~~音频格式~~ → 纯人声 MP3，无需预处理
|
||||
- [x] ~~LLM 选型~~ → DeepSeek（deepseek-chat），已验证效果好、价格低
|
||||
- [ ] 前端审稿台技术选型（第二步再定）
|
||||
- [ ] 大洋系统 SRT 导入兼容性（周一验证）
|
||||
@@ -0,0 +1,152 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
CCA 唱词助手 — 脚本版流水线入口
|
||||
|
||||
用法:
|
||||
# 完整流程: A稿热词 + ASR + AI校对 + AI折行 → 5个SRT
|
||||
python cca_pipeline.py --audio data/xxx.mp3 --script data/xxx.docx
|
||||
|
||||
# 手动指定热词
|
||||
python cca_pipeline.py --audio data/xxx.mp3 --hotwords "热词1|热词2"
|
||||
|
||||
# 跳过ASR，从缓存处理（调试折行/校对用）
|
||||
python cca_pipeline.py --asr-cache output/asr_raw.json --script data/xxx.docx
|
||||
|
||||
# 省token模式（不用AI折行和校对）
|
||||
python cca_pipeline.py --asr-cache output/asr_raw.json --no-ai
|
||||
|
||||
流水线:
|
||||
A稿 → 热词提取 → 音频+热词 → 讯飞ASR → AI校对 → 节目结构切分 → AI折行 → 5个SRT
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
||||
|
||||
from asr_client import transcribe, parse_result
|
||||
from line_breaker import process_sentences
|
||||
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
|
||||
from hotword_extractor import extract_hotwords
|
||||
from ai_proofreader import proofread_batch
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="CCA 唱词助手 - 自动生成拍词 SRT")
|
||||
parser.add_argument("--audio", type=str, help="音频文件路径 (mp3/wav)")
|
||||
parser.add_argument("--script", type=str, help="A稿路径 (.docx/.txt)，用于热词提取和AI校对")
|
||||
parser.add_argument("--hotwords", type=str, default="", help="手动指定热词，用|分隔（与--script可叠加）")
|
||||
parser.add_argument("--asr-cache", type=str, help="ASR 缓存 JSON 路径（跳过ASR调用）")
|
||||
parser.add_argument("--output-dir", type=str, default="output", help="输出目录 (默认: output/)")
|
||||
parser.add_argument("--no-ai", action="store_true", help="不使用AI折行和校对（省token）")
|
||||
parser.add_argument("--no-proofread", action="store_true", help="跳过AI校对（只省校对的token）")
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.audio and not args.asr_cache:
|
||||
parser.error("必须提供 --audio 或 --asr-cache")
|
||||
|
||||
output_dir = Path(args.output_dir)
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
use_ai = not args.no_ai
|
||||
|
||||
# ====== Step 0: 热词提取（从A稿）======
|
||||
hot_words = []
|
||||
script_text = ""
|
||||
|
||||
if args.hotwords:
|
||||
hot_words = [w.strip() for w in args.hotwords.split("|") if w.strip()]
|
||||
|
||||
if args.script:
|
||||
print(f"[流水线] 从A稿提取热词: {args.script}")
|
||||
script_hot = extract_hotwords(args.script, use_ai=use_ai)
|
||||
# 合并手动热词和A稿热词
|
||||
seen = set(hot_words)
|
||||
for w in script_hot:
|
||||
if w not in seen:
|
||||
hot_words.append(w)
|
||||
seen.add(w)
|
||||
|
||||
# 读取A稿全文（校对用）
|
||||
ext = os.path.splitext(args.script)[1].lower()
|
||||
if ext == ".docx":
|
||||
from hotword_extractor import read_docx_text
|
||||
script_text = read_docx_text(args.script)
|
||||
else:
|
||||
from hotword_extractor import read_text_file
|
||||
script_text = read_text_file(args.script)
|
||||
|
||||
if hot_words:
|
||||
print(f"[流水线] 热词共 {len(hot_words)} 个: {', '.join(hot_words[:10])}...")
|
||||
# 保存热词列表
|
||||
hotwords_path = output_dir / "hotwords.txt"
|
||||
with open(hotwords_path, "w", encoding="utf-8") as f:
|
||||
f.write("|".join(hot_words))
|
||||
|
||||
# ====== Step 1: ASR 转写 ======
|
||||
if args.asr_cache:
|
||||
print(f"[流水线] 从缓存加载 ASR 结果: {args.asr_cache}")
|
||||
with open(args.asr_cache, "r", encoding="utf-8") as f:
|
||||
raw_json = f.read()
|
||||
sentences = parse_result(raw_json)
|
||||
else:
|
||||
print(f"[流水线] 开始 ASR 转写: {args.audio}")
|
||||
sentences, raw_json = transcribe(args.audio, hot_words=hot_words if hot_words else None)
|
||||
|
||||
cache_path = output_dir / "asr_raw.json"
|
||||
with open(cache_path, "w", encoding="utf-8") as f:
|
||||
f.write(raw_json)
|
||||
print(f"[流水线] ASR 原始结果已缓存: {cache_path}")
|
||||
|
||||
print(f"[流水线] ASR 共 {len(sentences)} 句")
|
||||
|
||||
# ====== Step 2: AI 校对 ======
|
||||
if use_ai and not args.no_proofread and script_text:
|
||||
print("[流水线] AI 校对中 (DeepSeek)...")
|
||||
sentences = proofread_batch(sentences, script_text)
|
||||
elif not script_text and not args.no_proofread and use_ai:
|
||||
print("[流水线] 未提供A稿(--script)，跳过AI校对")
|
||||
|
||||
# ====== Step 3: 节目结构切分 ======
|
||||
print("[流水线] 切分节目结构...")
|
||||
segments = split_into_segments(sentences)
|
||||
print(f"[流水线] 切分结果: {[name for name, _ in segments]}")
|
||||
|
||||
# ====== Step 4: 折行 + 生成 SRT ======
|
||||
if use_ai:
|
||||
print("[流水线] 使用 AI 折行 (DeepSeek)...")
|
||||
else:
|
||||
print("[流水线] 使用机械折行规则...")
|
||||
|
||||
for seg_name, seg_sentences in segments:
|
||||
if not seg_sentences:
|
||||
print(f" [{seg_name}] 空段，跳过")
|
||||
continue
|
||||
|
||||
if use_ai:
|
||||
subtitle_lines = process_sentences_with_ai(seg_sentences)
|
||||
else:
|
||||
subtitle_lines = process_sentences(seg_sentences)
|
||||
|
||||
seg_offset = subtitle_lines[0][0] if subtitle_lines else 0
|
||||
seg_end = subtitle_lines[-1][1] if subtitle_lines else 0
|
||||
seg_duration = seg_end - seg_offset
|
||||
|
||||
srt_filename = f"{seg_name}.srt"
|
||||
srt_path = output_dir / srt_filename
|
||||
write_srt(subtitle_lines, str(srt_path)) # 绝对时间戳，方便在时间线上对位
|
||||
print(f" [{seg_name}] 时长 {ms_to_srt_time(seg_duration)}, 在音频中的位置: {ms_to_srt_time(seg_offset)} ~ {ms_to_srt_time(seg_end)}")
|
||||
|
||||
# ====== 完成 ======
|
||||
print(f"\n[流水线] 完成! 输出目录: {output_dir}/")
|
||||
print("[流水线] 生成的文件:")
|
||||
for f in sorted(output_dir.glob("*.srt")):
|
||||
print(f" {f.name}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+1292
File diff suppressed because it is too large
Load Diff
+1252
File diff suppressed because it is too large
Load Diff
+1112
File diff suppressed because it is too large
Load Diff
Binary file not shown.
@@ -0,0 +1,129 @@
|
||||
# CCA 唱词助手子项目 Brief
|
||||
|
||||
> 主项目 → 子项目的"交接宪法"：红线、技术栈、出入口接口
|
||||
> 起草日期：2026-07-04
|
||||
> 状态：主项目签发，子项目内部不修改
|
||||
|
||||
---
|
||||
|
||||
## 一、为什么做
|
||||
|
||||
**痛点**：《军事科技》每期节目剪辑完成后，编导需要制作一份唱词字幕（SRT 文件）交给责编拉到大洋系统上线。目前纯手工操作——对照 A 稿听音频逐句打字幕，费时且容易出错。
|
||||
|
||||
**升级目标**：编导只需提供 A 稿 + 粗编人声音频，系统自动走 ASR + AI 校对 + 编导审稿确认，最终输出大洋系统可直接使用的 SRT 文件。
|
||||
|
||||
**紧迫性**：切实解决当前痛点，功能不复杂，两天内出可用版本。先独立部署测试，成熟后并入 TPS 主项目。
|
||||
|
||||
---
|
||||
|
||||
## 二、做什么（功能边界）
|
||||
|
||||
| 模块 | 简述 |
|
||||
|---|---|
|
||||
| 专有名词提取 | AI 从 A 稿中摘取军事专有名词（型号、人名、机构等），形成热词词典 |
|
||||
| 讯飞 ASR 转写 | 音频 + 词典 → 带时间戳的 ASR 稿 |
|
||||
| AI 校对比对 | ASR 稿 ⊕ A 稿比对，自动修正明显错误，标记存疑差异 |
|
||||
| 编导审稿台 | 双屏对比 UI，差异高亮，编导点击确认/手动修改 |
|
||||
| SRT 生成 | B 稿按拍词规则折行 + 时间戳 → 大洋格式 SRT |
|
||||
|
||||
**具体怎么做，子项目内部讨论，本 Brief 不预设方案。**
|
||||
|
||||
---
|
||||
|
||||
## 三、怎么用（目标流程）
|
||||
|
||||
1. 编导剪完节目，手头有 A 稿（docx/txt）+ 粗编纯人声音频
|
||||
2. 编导上传 A 稿和音频到 CCA 系统
|
||||
3. 系统自动：提取专有名词 → 讯飞 ASR → AI 校对 → 生成初版对比
|
||||
4. 编导进入审稿台：查看差异、确认/修改
|
||||
5. 系统输出 SRT 文件
|
||||
6. 责编拿 SRT 拉到大洋系统形成唱词字幕
|
||||
|
||||
---
|
||||
|
||||
## 四、不做什么（红线）
|
||||
|
||||
- ❌ **不替代编导审稿**：AI 拿不准的必须过编导确认，不全自动出 SRT
|
||||
- ❌ **不改 A 稿内容**：A 稿只用于提取专有名词和提供参照，不回写
|
||||
- ❌ **不动主项目 backend 代码 / schema**（在吸收进主项目之前）
|
||||
- ❌ **不复用主项目 backend/.env 的凭证**（子项目自己 .env）
|
||||
- ❌ **不用讯飞大模型版 ASR**（用录音文件转写标准版，与 doco 一致）
|
||||
- ❌ **不做视频处理**（输入是已分离的纯人声音频，不是视频）
|
||||
- ❌ **不做内容创作**（AI 只校对不创作，尊重 ASR 底稿 + 编导意见）
|
||||
|
||||
---
|
||||
|
||||
## 五、技术栈约束
|
||||
|
||||
- **运行环境**：Python 3.x + Web 前端（审稿台）
|
||||
- **ASR**：讯飞开放平台 录音文件转写标准版（热词偏置注入）
|
||||
- **AI/LLM**：用于专有名词提取 + 校对（具体模型子项目讨论，省 token 优先）
|
||||
- **输出格式**：SRT（大洋系统兼容，有样本）
|
||||
- **部署**：先部署 lanhao 配音 2.0 网站测试，后续并入 TPS
|
||||
- **美术风格**：参考 `ai-labeling/example/` 目录下的设计稿（`功能区划.jpg`、`幅面参考.jpg`、`页面风格.webp`），CCA 的 UI 风格与 TPS 主项目保持一致
|
||||
|
||||
---
|
||||
|
||||
## 六、交付什么（出口接口）
|
||||
|
||||
### 6.1 并入架构（同一服务器、同一进程）
|
||||
|
||||
CCA 最终**不是独立微服务**，而是代码合并进 TPS 同一套服务：
|
||||
|
||||
- **后端**：CCA 的 API 路由并入 `backend/app/api/`（如 `cca.py`），业务逻辑并入 `backend/app/services/`（如 `cca_service.py`），共用同一个 FastAPI 进程
|
||||
- **前端**：CCA 页面并入 `frontend/src/pages/`（如 `CCA/` 目录），作为 TPS React 应用的一个路由页面（`/cca`）
|
||||
- **同一台服务器**：不存在跨服务调用，共用 TPS 的 PostgreSQL、同一个 `.env`
|
||||
|
||||
### 6.2 开发阶段的独立形态
|
||||
|
||||
并入前的开发测试阶段，CCA 以独立可运行的 Web 应用存在：
|
||||
- 独立的后端（FastAPI 或轻量框架）+ 前端（审稿台 UI）
|
||||
- 可单独部署到 lanhao 配音 2.0 网站跑通验证
|
||||
- 代码结构从一开始就按"能干净并入 TPS"设计（目录命名、API 风格、前端组件规范对齐主项目）
|
||||
|
||||
### 6.3 具体交付清单
|
||||
|
||||
| 交付物 | 说明 |
|
||||
|---|---|
|
||||
| CCA 后端 API | 文件上传（A稿+音频）、ASR 任务触发/轮询、校对结果、审稿确认、SRT 下载 |
|
||||
| 讯飞 ASR 适配层 | 热词注入 + 转写 + 解析，可被主项目其他模块复用（与 doco 逻辑同源） |
|
||||
| 专有名词提取服务 | AI 从 A 稿提取军事专名 → 热词词典 |
|
||||
| AI 校对比对服务 | ASR 稿 ⊕ A 稿差异标记（自动修正 + 存疑清单） |
|
||||
| 编导审稿台前端 | 双屏对比 UI + 差异高亮 + 确认/手改交互 |
|
||||
| 拍词折行引擎 | 按规则把 B 稿折行 + 结合时间戳生成 SRT |
|
||||
| SRT 输出 | 大洋系统兼容格式（有样本对标） |
|
||||
|
||||
### 6.4 配置与凭证
|
||||
- 开发阶段：cca 自己的 `.env`（讯飞 Key + LLM Key）
|
||||
- 并入后：凭证合并进 TPS 主项目 `backend/.env`，讯飞 Key 可与 doco 共用
|
||||
- 主项目 `docs/api_credentials_inventory.md` 登记 CCA 凭证元信息
|
||||
|
||||
### 6.5 不接受
|
||||
- 并入前，子项目自己改主项目 backend 代码、schema、迁移
|
||||
- 未经主项目审批引入主项目没有的技术栈
|
||||
- 并入时引入与主项目冲突的依赖版本
|
||||
|
||||
---
|
||||
|
||||
## 七、入口接口（子项目要知道的主项目现状）
|
||||
|
||||
- **主项目已有讯飞 ASR 经验**：doco 子项目已跑通讯飞录音文件转写标准版，签名/上传/轮询/解析全套逻辑可复用
|
||||
- **主项目有 `api_credentials_inventory.md`**：凭证元信息登记表
|
||||
- **主项目有跨子项目协作规则**：PRD 版本管理等，见 `跨子项目协作规则.md`
|
||||
|
||||
---
|
||||
|
||||
## 八、制片人待提供
|
||||
|
||||
- [ ] 拍词规则（责编拍词的具体 markdown 规则文档）
|
||||
- [ ] 大洋 SRT 样本文件（一份能用的真实样本）
|
||||
- [ ] 粗编音频样本（一期的真实音频，用于开发测试）
|
||||
- [ ] A 稿样本（对应的编导 A 稿）
|
||||
|
||||
---
|
||||
|
||||
## 九、本 Brief 自身的修订规则
|
||||
|
||||
- 本 Brief 不在子项目内修改
|
||||
- 红线/技术栈如有变更，**主项目这边发新版**
|
||||
- 子项目可在自己的 chat 里建议调整，调整动作只能在主项目发生
|
||||
@@ -0,0 +1,5 @@
|
||||
requests>=2.28
|
||||
python-dotenv>=1.0
|
||||
mutagen>=1.47
|
||||
openai>=1.0
|
||||
python-docx>=1.0
|
||||
@@ -0,0 +1,224 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
AI 折行引擎 — 用 DeepSeek 对 ASR 长句做语义折行
|
||||
|
||||
对于 ≤14 字的句子直接输出，>14 字的句子批量发给 AI 折行。
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
_env_path = Path(__file__).resolve().parent.parent / ".env"
|
||||
if _env_path.exists():
|
||||
load_dotenv(str(_env_path), override=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
from openai import OpenAI
|
||||
|
||||
DEEPSEEK_API_KEY = os.environ.get("DEEPSEEK_API_KEY", "").strip()
|
||||
DEEPSEEK_BASE_URL = os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com").strip()
|
||||
DEEPSEEK_MODEL = os.environ.get("DEEPSEEK_MODEL", "deepseek-chat").strip()
|
||||
|
||||
MAX_CHARS = 14
|
||||
MAX_CHARS_SOFT = 16
|
||||
SILENCE_THRESHOLD_MS = 2000
|
||||
|
||||
SYSTEM_PROMPT = """你是电视节目唱词字幕的折行助手。你的任务是将一段文字按照以下规则折成多行：
|
||||
|
||||
规则：
|
||||
1. 每行最多14个字（中文字符、英文字母、数字各算1个字）
|
||||
2. 去掉逗号、句号、感叹号、问号、顿号、分号、冒号、省略号等标点，只保留引号（""''「」）和书名号（《》）
|
||||
3. 折行要符合语义和阅读习惯，不能把词语切断
|
||||
4. 每行不一定要凑满14字，可以是5字、8字、10字等，关键是语义完整
|
||||
5. 保持原文内容不变，不增不减不改字
|
||||
|
||||
输出格式：每行一句，不加序号，不加标点（引号和书名号除外）。"""
|
||||
|
||||
USER_PROMPT_TEMPLATE = """请将以下文字折行（每行≤14字，去标点保引号，按语义断句）：
|
||||
|
||||
{text}"""
|
||||
|
||||
BATCH_USER_PROMPT = """请将以下编号文字逐条折行（每行≤14字，去标点保引号，按语义断句）。
|
||||
每条之间用空行分隔，保持编号对应。
|
||||
|
||||
{numbered_texts}"""
|
||||
|
||||
|
||||
def _create_client() -> OpenAI:
|
||||
if not DEEPSEEK_API_KEY:
|
||||
raise ValueError("请在 .env 中设置 DEEPSEEK_API_KEY")
|
||||
return OpenAI(api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL)
|
||||
|
||||
|
||||
def ai_break_single(text: str, client: OpenAI) -> List[str]:
|
||||
"""单句 AI 折行"""
|
||||
resp = client.chat.completions.create(
|
||||
model=DEEPSEEK_MODEL,
|
||||
messages=[
|
||||
{"role": "system", "content": SYSTEM_PROMPT},
|
||||
{"role": "user", "content": USER_PROMPT_TEMPLATE.format(text=text)},
|
||||
],
|
||||
temperature=0.1,
|
||||
max_tokens=500,
|
||||
)
|
||||
result = resp.choices[0].message.content.strip()
|
||||
lines = [l.strip() for l in result.split("\n") if l.strip()]
|
||||
return lines
|
||||
|
||||
|
||||
def ai_break_batch(texts: List[str], client: OpenAI) -> List[List[str]]:
|
||||
"""
|
||||
批量 AI 折行（减少 API 调用次数）
|
||||
每批最多 20 条，避免输出过长出错
|
||||
"""
|
||||
if not texts:
|
||||
return []
|
||||
|
||||
numbered = "\n".join(f"[{i+1}] {t}" for i, t in enumerate(texts))
|
||||
resp = client.chat.completions.create(
|
||||
model=DEEPSEEK_MODEL,
|
||||
messages=[
|
||||
{"role": "system", "content": SYSTEM_PROMPT},
|
||||
{"role": "user", "content": BATCH_USER_PROMPT.format(numbered_texts=numbered)},
|
||||
],
|
||||
temperature=0.1,
|
||||
max_tokens=3000,
|
||||
)
|
||||
result = resp.choices[0].message.content.strip()
|
||||
|
||||
# 解析结果：按空行或编号分隔
|
||||
all_results = []
|
||||
current_lines = []
|
||||
|
||||
for line in result.split("\n"):
|
||||
line = line.strip()
|
||||
# 检测新编号开头 [N] 或纯空行作为分隔
|
||||
if not line:
|
||||
if current_lines:
|
||||
all_results.append(current_lines)
|
||||
current_lines = []
|
||||
continue
|
||||
|
||||
# 去掉可能的编号前缀
|
||||
import re
|
||||
cleaned = re.sub(r'^\[\d+\]\s*', '', line)
|
||||
if cleaned:
|
||||
current_lines.append(cleaned)
|
||||
|
||||
if current_lines:
|
||||
all_results.append(current_lines)
|
||||
|
||||
# 如果解析结果数量不匹配，回退到逐条处理
|
||||
if len(all_results) != len(texts):
|
||||
print(f"[AI折行] 批量解析不匹配 (期望{len(texts)}条，得到{len(all_results)}条)，回退逐条处理")
|
||||
all_results = []
|
||||
for text in texts:
|
||||
lines = ai_break_single(text, client)
|
||||
all_results.append(lines)
|
||||
|
||||
return all_results
|
||||
|
||||
|
||||
def process_sentences_with_ai(
|
||||
sentences: List[Tuple[int, int, str, int]],
|
||||
batch_size: int = 15,
|
||||
) -> List[Tuple[int, int, str]]:
|
||||
"""
|
||||
用 AI 折行处理 ASR 句子列表。
|
||||
|
||||
输入: [(start_ms, end_ms, text, speaker_id), ...]
|
||||
输出: [(start_ms, end_ms, text), ...]
|
||||
|
||||
策略:
|
||||
- ≤14 字：直接输出（去标点）
|
||||
- >14 字：批量调 AI 折行
|
||||
- 句间 >2秒：插入空白行
|
||||
"""
|
||||
from line_breaker import clean_punctuation
|
||||
|
||||
if not sentences:
|
||||
return []
|
||||
|
||||
client = _create_client()
|
||||
result = []
|
||||
|
||||
# 先收集需要 AI 折行的句子索引
|
||||
needs_ai = [] # (original_index, text)
|
||||
for i, (bg, ed, text, spk) in enumerate(sentences):
|
||||
cleaned = clean_punctuation(text)
|
||||
if len(cleaned) > MAX_CHARS:
|
||||
needs_ai.append((i, cleaned))
|
||||
|
||||
# 批量调 AI
|
||||
ai_results = {} # index -> [lines]
|
||||
if needs_ai:
|
||||
print(f"[AI折行] 共 {len(needs_ai)} 句需要 AI 折行...")
|
||||
for batch_start in range(0, len(needs_ai), batch_size):
|
||||
batch = needs_ai[batch_start:batch_start + batch_size]
|
||||
batch_texts = [t for _, t in batch]
|
||||
batch_indices = [idx for idx, _ in batch]
|
||||
|
||||
print(f"[AI折行] 处理第 {batch_start+1}-{batch_start+len(batch)} 条...")
|
||||
try:
|
||||
broken = ai_break_batch(batch_texts, client)
|
||||
for idx, lines in zip(batch_indices, broken):
|
||||
ai_results[idx] = lines
|
||||
except Exception as e:
|
||||
print(f"[AI折行] 批量失败: {e}，回退逐条处理")
|
||||
for idx, text in batch:
|
||||
try:
|
||||
lines = ai_break_single(text, client)
|
||||
ai_results[idx] = lines
|
||||
except Exception as e2:
|
||||
print(f"[AI折行] 第{idx}句失败: {e2}，使用机械切分")
|
||||
from line_breaker import break_sentence
|
||||
ai_results[idx] = break_sentence(text)
|
||||
|
||||
# 组装最终结果
|
||||
for i, (bg, ed, text, spk) in enumerate(sentences):
|
||||
# 检查空白
|
||||
if i > 0:
|
||||
prev_ed = sentences[i - 1][1]
|
||||
gap = bg - prev_ed
|
||||
if gap > SILENCE_THRESHOLD_MS:
|
||||
result.append((prev_ed, bg, ""))
|
||||
|
||||
cleaned = clean_punctuation(text)
|
||||
if not cleaned.strip():
|
||||
continue
|
||||
|
||||
if i in ai_results:
|
||||
lines = ai_results[i]
|
||||
else:
|
||||
lines = [cleaned]
|
||||
|
||||
# 后处理：AI 偶尔返回超长行，强制二次切分
|
||||
from line_breaker import break_sentence
|
||||
final_lines = []
|
||||
for line in lines:
|
||||
if len(line) > MAX_CHARS_SOFT:
|
||||
final_lines.extend(break_sentence(line))
|
||||
else:
|
||||
final_lines.append(line)
|
||||
lines = final_lines
|
||||
|
||||
# 为子行分配时间戳
|
||||
total_chars = sum(len(l) for l in lines)
|
||||
duration = ed - bg
|
||||
current_ms = bg
|
||||
|
||||
for line in lines:
|
||||
if not line.strip():
|
||||
continue
|
||||
line_duration = int(duration * len(line) / total_chars) if total_chars > 0 else 0
|
||||
line_end = min(current_ms + line_duration, ed)
|
||||
result.append((current_ms, line_end, line))
|
||||
current_ms = line_end
|
||||
|
||||
return result
|
||||
@@ -0,0 +1,163 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
AI 校对器 — ASR 稿与 A 稿比对 + 上下文纠错
|
||||
|
||||
解决的核心问题:
|
||||
- ASR 同音字误识别（"建制"→"舰只"、"舰手"→"舰艏"）
|
||||
- 军事术语规范化（"f15j"→"F-15J"）
|
||||
- 的/地/得纠错
|
||||
- 去除口语填充词（"嗯""那个""就是说"）
|
||||
|
||||
策略:
|
||||
- 将 ASR 全文 + A 稿全文一起发给 DeepSeek
|
||||
- AI 结合节目主题和上下文做纠错
|
||||
- 返回修正后的句子列表 + 修改说明
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Tuple, Dict
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
_env_path = Path(__file__).resolve().parent.parent / ".env"
|
||||
if _env_path.exists():
|
||||
load_dotenv(str(_env_path), override=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
PROOFREAD_SYSTEM_PROMPT = """你是电视军事节目《军事科技》的字幕校对专家。你将收到两份材料：
|
||||
1. **ASR稿**：语音识别的转写结果，带有时间编号，是字幕的基础
|
||||
2. **A稿**：编导写的节目文稿（仅包含解说词，不包含专家采访的具体内容）
|
||||
|
||||
你的任务是校对 ASR 稿中的**语音识别错误**。
|
||||
|
||||
**铁律（违反任何一条都算失败）：**
|
||||
- ASR稿是已经录好的音频的转写，内容不能改——**绝不润色语句、绝不调整语序、绝不增删实词**
|
||||
- 只修三类问题：① 错别字/同音字 ② 术语格式 ③ 口语填充词
|
||||
- 除这三类外的一切文字，原封不动照抄，一个字都不能动
|
||||
- A稿只用来判断"这个词在本期节目的语境下应该是哪个字"，不能把ASR稿往A稿的措辞上靠
|
||||
|
||||
**允许修的三类：**
|
||||
1. **同音字/错别字**（ASR听错的字）：如"建制"→"舰只"、"舰手"→"舰艏"、"继承"→"击沉"、"空花弹"→"滑翔弹"
|
||||
2. **术语格式**：英文型号大小写+连字符（"f15j"→"F-15J"、"v22"→"V-22"、"rq四"→"RQ-4"）
|
||||
3. **口语填充词删除**：只删"嗯""呃""唉""啊""呢""那个""就是说""这个"这类纯填充词。如果"这个"后面紧跟名词作指示代词（"这个导弹"），保留不删
|
||||
|
||||
**绝对不许做的（哪怕你觉得改了更好也不许）：**
|
||||
- 不许调整语序（"它在性质上就是"不许改成"它本质上就是"）
|
||||
- 不许替换实词（"不是那么特别的顺利"不许改成"不太顺利"）
|
||||
- 不许增删标点来改变句子结构
|
||||
- 不许把口语化表达改成书面语
|
||||
- 不许根据A稿的措辞替换ASR中意思相同但用词不同的表达
|
||||
|
||||
**输出格式：**
|
||||
JSON数组，每个元素：{"id": 编号, "original": "原文", "corrected": "修正后", "changes": "修改说明（无修改写空字符串）"}
|
||||
只输出JSON，不要其他内容。"""
|
||||
|
||||
|
||||
PROOFREAD_USER_TEMPLATE = """**A稿（节目文稿，仅供参考）：**
|
||||
{script_text}
|
||||
|
||||
**ASR稿（需要校对，请逐条检查）：**
|
||||
{asr_text}"""
|
||||
|
||||
|
||||
def _create_client():
|
||||
api_key = os.environ.get("DEEPSEEK_API_KEY", "").strip()
|
||||
if not api_key:
|
||||
raise ValueError("请在 .env 中设置 DEEPSEEK_API_KEY")
|
||||
from openai import OpenAI
|
||||
return OpenAI(
|
||||
api_key=api_key,
|
||||
base_url=os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
|
||||
)
|
||||
|
||||
|
||||
def proofread_batch(
|
||||
asr_sentences: List[Tuple[int, int, str, int]],
|
||||
script_text: str,
|
||||
batch_size: int = 30,
|
||||
) -> List[Tuple[int, int, str, int]]:
|
||||
"""
|
||||
对 ASR 句子列表做 AI 校对。
|
||||
|
||||
输入:
|
||||
asr_sentences: [(start_ms, end_ms, text, speaker_id), ...]
|
||||
script_text: A稿全文
|
||||
batch_size: 每批处理的句子数
|
||||
|
||||
返回:
|
||||
校对后的句子列表，格式同输入
|
||||
"""
|
||||
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) # 浅拷贝
|
||||
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)
|
||||
|
||||
# 构建 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"[校对] 处理第 {batch_start+1}-{batch_end} 句...")
|
||||
|
||||
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,
|
||||
)
|
||||
|
||||
result_text = resp.choices[0].message.content.strip()
|
||||
|
||||
# 尝试解析 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()
|
||||
|
||||
corrections = json.loads(result_text)
|
||||
|
||||
# 应用修正
|
||||
for item in corrections:
|
||||
idx = item.get("id", 0) - 1 # 编号从1开始
|
||||
corrected = item.get("corrected", "")
|
||||
changes = item.get("changes", "")
|
||||
|
||||
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})")
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
print(f"[校对] JSON解析失败，跳过本批: {e}", file=sys.stderr)
|
||||
except Exception as e:
|
||||
print(f"[校对] 出错: {e}", file=sys.stderr)
|
||||
|
||||
print(f"[校对] 完成，共修正 {total_changes} 处")
|
||||
return corrected_sentences
|
||||
@@ -0,0 +1,229 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
讯飞 ASR 客户端 — 适配自 doco/src/doco/asr_adapter.py
|
||||
录音文件转写标准版: https://raasr.xfyun.cn/v2/api
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import wave
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
from typing import List, Tuple, Optional
|
||||
|
||||
import requests
|
||||
|
||||
# ========================================================================
|
||||
# 凭证
|
||||
# ========================================================================
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
_env_path = Path(__file__).resolve().parent.parent / ".env"
|
||||
if _env_path.exists():
|
||||
load_dotenv(str(_env_path), override=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
APP_ID = os.environ.get("XFYUN_APP_ID", "").strip()
|
||||
SECRET_KEY = os.environ.get("XFYUN_SECRET_KEY", "").strip()
|
||||
|
||||
# ========================================================================
|
||||
# 接口配置
|
||||
# ========================================================================
|
||||
|
||||
HOST = "https://raasr.xfyun.cn/v2/api"
|
||||
UPLOAD_URL = HOST + "/upload"
|
||||
RESULT_URL = HOST + "/getResult"
|
||||
|
||||
LANGUAGE = "cn"
|
||||
PD = "mil"
|
||||
ENG_SMOOTHPROC = "true"
|
||||
ENG_COLLOQPROC = "true"
|
||||
ROLE_TYPE = "1"
|
||||
ROLE_NUM = "0"
|
||||
|
||||
POLL_INTERVAL_SECONDS = 30
|
||||
MAX_WAIT_MINUTES = 30
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 签名
|
||||
# ========================================================================
|
||||
|
||||
def _make_signa(app_id: str, secret_key: str, ts: str) -> str:
|
||||
base_string = (app_id + ts).encode("utf-8")
|
||||
md5_str = hashlib.md5(base_string).hexdigest()
|
||||
mac = hmac.new(
|
||||
secret_key.encode("utf-8"),
|
||||
md5_str.encode("utf-8"),
|
||||
digestmod=hashlib.sha1,
|
||||
)
|
||||
return base64.b64encode(mac.digest()).decode("utf-8")
|
||||
|
||||
|
||||
def _get_audio_duration_ms(filepath: str) -> int:
|
||||
ext = os.path.splitext(filepath)[1].lower()
|
||||
if ext == ".wav":
|
||||
with wave.open(filepath, "rb") as wf:
|
||||
return int(round(wf.getnframes() / wf.getframerate() * 1000))
|
||||
if ext == ".mp3":
|
||||
try:
|
||||
from mutagen.mp3 import MP3
|
||||
return int(MP3(filepath).info.length * 1000)
|
||||
except ImportError:
|
||||
print("[警告] 需要 mutagen 库来读取 MP3 时长: pip install mutagen", file=sys.stderr)
|
||||
return 0
|
||||
raise ValueError(f"不支持的音频格式: {ext}")
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 上传
|
||||
# ========================================================================
|
||||
|
||||
def upload_audio(filepath: str, hot_words: Optional[List[str]] = None) -> str:
|
||||
if not APP_ID or not SECRET_KEY:
|
||||
raise ValueError("请先在 .env 中设置 XFYUN_APP_ID 和 XFYUN_SECRET_KEY")
|
||||
|
||||
file_size = os.path.getsize(filepath)
|
||||
file_name = os.path.basename(filepath)
|
||||
duration_ms = _get_audio_duration_ms(filepath)
|
||||
ts = str(int(time.time()))
|
||||
signa = _make_signa(APP_ID, SECRET_KEY, ts)
|
||||
|
||||
params = {
|
||||
"appId": APP_ID,
|
||||
"signa": signa,
|
||||
"ts": ts,
|
||||
"fileSize": str(file_size),
|
||||
"fileName": file_name,
|
||||
"duration": str(duration_ms),
|
||||
"language": LANGUAGE,
|
||||
"pd": PD,
|
||||
"eng_smoothproc": ENG_SMOOTHPROC,
|
||||
"eng_colloqproc": ENG_COLLOQPROC,
|
||||
"roleType": ROLE_TYPE,
|
||||
"roleNum": ROLE_NUM,
|
||||
}
|
||||
|
||||
if hot_words:
|
||||
params["hotWord"] = "|".join(hot_words[:200])
|
||||
|
||||
url_parts = [f"{quote(k, safe='')}={quote(str(v), safe='')}" for k, v in params.items()]
|
||||
url = f"{UPLOAD_URL}?{'&'.join(url_parts)}"
|
||||
|
||||
with open(filepath, "rb") as f:
|
||||
audio_bytes = f.read()
|
||||
|
||||
print(f"[ASR] 上传音频: {file_name} ({file_size/1024/1024:.1f}MB)")
|
||||
resp = requests.post(url, headers={"Content-Type": "application/json"}, data=audio_bytes, timeout=300)
|
||||
data = resp.json()
|
||||
|
||||
if data.get("code") != "000000":
|
||||
raise RuntimeError(f"上传失败: code={data.get('code')}, desc={data.get('descInfo')}")
|
||||
|
||||
order_id = data["content"]["orderId"]
|
||||
print(f"[ASR] 上传成功, orderId={order_id}")
|
||||
return order_id
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 轮询
|
||||
# ========================================================================
|
||||
|
||||
def poll_until_done(order_id: str) -> dict:
|
||||
start_time = time.time()
|
||||
print(f"[ASR] 开始轮询 (每{POLL_INTERVAL_SECONDS}秒)...")
|
||||
while True:
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed > MAX_WAIT_MINUTES * 60:
|
||||
raise TimeoutError(f"超过 {MAX_WAIT_MINUTES} 分钟未完成")
|
||||
|
||||
ts = str(int(time.time()))
|
||||
signa = _make_signa(APP_ID, SECRET_KEY, ts)
|
||||
params = {
|
||||
"appId": APP_ID,
|
||||
"signa": signa,
|
||||
"ts": ts,
|
||||
"orderId": order_id,
|
||||
"resultType": "transfer",
|
||||
}
|
||||
url_parts = [f"{quote(k, safe='')}={quote(str(v), safe='')}" for k, v in params.items()]
|
||||
url = f"{RESULT_URL}?{'&'.join(url_parts)}"
|
||||
|
||||
resp = requests.post(url, timeout=30)
|
||||
data = resp.json()
|
||||
order_info = data.get("content", {}).get("orderInfo", {})
|
||||
status = order_info.get("status")
|
||||
|
||||
if status == 4:
|
||||
print(f"[ASR] 转写完成 (耗时{int(elapsed)}秒)")
|
||||
return data
|
||||
if status == -1:
|
||||
raise RuntimeError(f"转写失败: {data}")
|
||||
|
||||
print(f"[ASR] 等待中... ({int(elapsed)}秒)")
|
||||
time.sleep(POLL_INTERVAL_SECONDS)
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 解析
|
||||
# ========================================================================
|
||||
|
||||
def parse_result(order_result_str: str) -> List[Tuple[int, int, str, int]]:
|
||||
"""
|
||||
解析讯飞返回结果
|
||||
返回 [(start_ms, end_ms, text, speaker_id), ...]
|
||||
"""
|
||||
if not order_result_str:
|
||||
return []
|
||||
|
||||
cleaned = order_result_str.replace("\\\\", "\\")
|
||||
outer = json.loads(cleaned)
|
||||
|
||||
sentences = []
|
||||
for item in outer.get("lattice", []):
|
||||
inner_str = item.get("json_1best", "")
|
||||
if not inner_str:
|
||||
continue
|
||||
inner = json.loads(inner_str)
|
||||
st = inner.get("st", {})
|
||||
bg = int(st.get("bg", 0))
|
||||
ed = int(st.get("ed", 0))
|
||||
rl = int(st.get("rl", 0))
|
||||
|
||||
words = []
|
||||
for rt in st.get("rt", []):
|
||||
for ws in rt.get("ws", []):
|
||||
for cw in ws.get("cw", []):
|
||||
w = cw.get("w", "").strip()
|
||||
wp = cw.get("wp", "n")
|
||||
if w and wp != "g":
|
||||
words.append(w)
|
||||
sentence = "".join(words).strip()
|
||||
if sentence:
|
||||
sentences.append((bg, ed, sentence, rl))
|
||||
|
||||
return sentences
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 完整流程
|
||||
# ========================================================================
|
||||
|
||||
def transcribe(audio_path: str, hot_words: Optional[List[str]] = None) -> Tuple[List[Tuple[int, int, str, int]], str]:
|
||||
"""
|
||||
完整转写: 上传 → 轮询 → 解析
|
||||
返回 (sentences, raw_json_str)
|
||||
sentences: [(start_ms, end_ms, text, speaker_id), ...]
|
||||
"""
|
||||
order_id = upload_audio(audio_path, hot_words=hot_words)
|
||||
result_data = poll_until_done(order_id)
|
||||
order_result_str = result_data["content"]["orderResult"]
|
||||
sentences = parse_result(order_result_str)
|
||||
return sentences, order_result_str
|
||||
@@ -0,0 +1,210 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
热词提取器 — 从 A 稿中提取军事专有名词，注入讯飞 ASR 热词
|
||||
|
||||
两种模式:
|
||||
1. 规则提取（不消耗 API）: 匹配 A 稿中的武器型号、专有名词等
|
||||
2. AI 提取（消耗 DeepSeek API）: 让 AI 理解全文后提取专业术语
|
||||
|
||||
热词用途: 注入讯飞 ASR 的 hotWord 参数，提升领域识别准确率
|
||||
讯飞限制: 最多 200 个热词，每个 2-16 字，用 | 分隔
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import List, Optional
|
||||
|
||||
# ========================================================================
|
||||
# 规则提取
|
||||
# ========================================================================
|
||||
|
||||
# 常见军事装备型号模式
|
||||
MILITARY_PATTERNS = [
|
||||
# 武器型号: F-35B, F-15J, RQ-4, V-22 等
|
||||
re.compile(r'[A-Z]{1,3}-?\d{1,4}[A-Z]?(?:/[A-Z])?'),
|
||||
# 中文+数字型号: 12式, 25式, 17式
|
||||
re.compile(r'\d{1,3}式'),
|
||||
# 导弹/武器名称中的英文缩写
|
||||
re.compile(r'(?:MK|ESSM|LM|OPY|FCS)-?\d*[A-Z]*'),
|
||||
]
|
||||
|
||||
# 军事领域高频词（手动维护，补充 ASR 容易错的同音词）
|
||||
MILITARY_VOCAB = [
|
||||
# 海军
|
||||
"舰只", "舰艇", "舰载机", "护卫舰", "驱逐舰", "航空母舰", "航母",
|
||||
"出云级", "出云号", "日向级", "日向号", "加贺号", "最上级",
|
||||
"伊势号", "满载排水量", "飞行甲板", "舰艏", "舰艉", "舰宽",
|
||||
"垂直发射系统", "近防系统", "反潜", "扫雷",
|
||||
# 空军
|
||||
"战斗机", "隐身战斗机", "舰载机", "无人机", "旋翼机",
|
||||
"航空自卫队", "航空宇宙自卫队",
|
||||
# 陆军/导弹
|
||||
"巡航导弹", "反舰导弹", "高超音速", "滑翔弹",
|
||||
"战斧", "陆上自卫队", "海上自卫队",
|
||||
"岸基", "弹径", "弹体", "射程", "马赫数",
|
||||
# 通用军事
|
||||
"防卫省", "自卫队", "专守防卫", "和平宪法",
|
||||
"军备", "军售", "军费", "军工", "军舰",
|
||||
"进攻性", "防御性", "远程打击", "精确打击",
|
||||
"作战编队", "态势感知", "火力演习",
|
||||
# 人名/地名
|
||||
"蓝皓", "熊本县", "南鸟岛", "东富士",
|
||||
# 节目相关
|
||||
"军事科技", "国防军事频道",
|
||||
]
|
||||
|
||||
|
||||
def extract_by_rules(text: str) -> List[str]:
|
||||
"""用正则从文本中提取军事术语"""
|
||||
found = set()
|
||||
|
||||
# 正则匹配
|
||||
for pattern in MILITARY_PATTERNS:
|
||||
for match in pattern.finditer(text):
|
||||
word = match.group().strip()
|
||||
if 2 <= len(word) <= 16:
|
||||
found.add(word)
|
||||
|
||||
# 固定词表匹配（只加文本中确实出现的词）
|
||||
for word in MILITARY_VOCAB:
|
||||
if word in text:
|
||||
found.add(word)
|
||||
|
||||
return sorted(found)
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# AI 提取
|
||||
# ========================================================================
|
||||
|
||||
AI_EXTRACT_PROMPT = """你是军事节目的专有名词提取助手。请从以下节目文稿中提取所有军事专有名词和术语。
|
||||
|
||||
提取范围：
|
||||
1. 武器装备型号（如 F-35B、12式反舰导弹、战斧巡航导弹）
|
||||
2. 军事单位/部队名称（如 航空自卫队、陆上自卫队）
|
||||
3. 军舰/飞机名称（如 出云号、日向级）
|
||||
4. 军事术语（如 垂直发射系统、高超音速滑翔弹、专守防卫）
|
||||
5. 人名、地名（如 蓝皓、熊本县、南鸟岛）
|
||||
6. 容易被语音识别混淆的词（如 "舰只"容易被识别为"建制"，"舰艏"容易被识别为"舰手"）
|
||||
|
||||
输出格式：每行一个词，不加序号，不加解释。每个词 2-16 字。"""
|
||||
|
||||
|
||||
def extract_by_ai(text: str) -> List[str]:
|
||||
"""用 DeepSeek 从文本中提取专有名词"""
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
_env_path = Path(__file__).resolve().parent.parent / ".env"
|
||||
if _env_path.exists():
|
||||
load_dotenv(str(_env_path), override=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
api_key = os.environ.get("DEEPSEEK_API_KEY", "").strip()
|
||||
if not api_key:
|
||||
print("[热词] DeepSeek 未配置，跳过 AI 提取", file=sys.stderr)
|
||||
return []
|
||||
|
||||
from openai import OpenAI
|
||||
client = OpenAI(
|
||||
api_key=api_key,
|
||||
base_url=os.environ.get("DEEPSEEK_BASE_URL", "https://api.deepseek.com"),
|
||||
)
|
||||
|
||||
# 文稿太长时截取前6000字
|
||||
truncated = text[:6000] if len(text) > 6000 else text
|
||||
|
||||
resp = client.chat.completions.create(
|
||||
model=os.environ.get("DEEPSEEK_MODEL", "deepseek-chat"),
|
||||
messages=[
|
||||
{"role": "system", "content": AI_EXTRACT_PROMPT},
|
||||
{"role": "user", "content": truncated},
|
||||
],
|
||||
temperature=0.1,
|
||||
max_tokens=2000,
|
||||
)
|
||||
|
||||
result = resp.choices[0].message.content.strip()
|
||||
words = []
|
||||
for line in result.split("\n"):
|
||||
word = line.strip().strip("-").strip("·").strip("•").strip()
|
||||
if word and 2 <= len(word) <= 16:
|
||||
words.append(word)
|
||||
|
||||
return words
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 读取 A 稿
|
||||
# ========================================================================
|
||||
|
||||
def read_docx_text(docx_path: str) -> str:
|
||||
"""从 docx 文件中提取纯文本"""
|
||||
try:
|
||||
from docx import Document
|
||||
doc = Document(docx_path)
|
||||
return "\n".join(p.text for p in doc.paragraphs if p.text.strip())
|
||||
except ImportError:
|
||||
print("[热词] 需要 python-docx 库: pip install python-docx", file=sys.stderr)
|
||||
return ""
|
||||
|
||||
|
||||
def read_text_file(path: str) -> str:
|
||||
"""读取 txt 文件"""
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
return f.read()
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 主入口
|
||||
# ========================================================================
|
||||
|
||||
def extract_hotwords(
|
||||
script_path: str,
|
||||
use_ai: bool = True,
|
||||
max_words: int = 200,
|
||||
) -> List[str]:
|
||||
"""
|
||||
从 A 稿提取热词列表
|
||||
|
||||
script_path: A 稿路径 (.docx 或 .txt)
|
||||
use_ai: 是否使用 AI 提取（默认 True）
|
||||
max_words: 最大热词数（讯飞限制 200）
|
||||
"""
|
||||
ext = os.path.splitext(script_path)[1].lower()
|
||||
if ext == ".docx":
|
||||
text = read_docx_text(script_path)
|
||||
elif ext in (".txt", ".md"):
|
||||
text = read_text_file(script_path)
|
||||
else:
|
||||
print(f"[热词] 不支持的文件格式: {ext}", file=sys.stderr)
|
||||
return []
|
||||
|
||||
if not text:
|
||||
return []
|
||||
|
||||
# 规则提取（免费）
|
||||
rule_words = extract_by_rules(text)
|
||||
print(f"[热词] 规则提取: {len(rule_words)} 个")
|
||||
|
||||
# AI 提取（可选）
|
||||
ai_words = []
|
||||
if use_ai:
|
||||
print("[热词] AI 提取中...")
|
||||
ai_words = extract_by_ai(text)
|
||||
print(f"[热词] AI 提取: {len(ai_words)} 个")
|
||||
|
||||
# 合并去重
|
||||
seen = set()
|
||||
merged = []
|
||||
for word in ai_words + rule_words: # AI 结果优先
|
||||
if word not in seen:
|
||||
seen.add(word)
|
||||
merged.append(word)
|
||||
|
||||
# 截取前 max_words 个
|
||||
result = merged[:max_words]
|
||||
print(f"[热词] 最终热词: {len(result)} 个")
|
||||
return result
|
||||
@@ -0,0 +1,156 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
折行引擎 — 将 ASR 句子按拍词规则折成 ≤14 字/行的字幕行
|
||||
|
||||
规则:
|
||||
A. 每行 ≤ 14 字
|
||||
B. ASR 中 >2 秒空白 → 插入空白行
|
||||
C. 按语义断句（不机械凑满 14 字）
|
||||
D. 去掉逗号/句号/感叹号/问号等标点，只保留引号
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import List, Tuple
|
||||
|
||||
MAX_CHARS_PER_LINE = 14
|
||||
MAX_CHARS_SOFT = 16 # 找不到好断点时允许的最大宽容值
|
||||
SILENCE_THRESHOLD_MS = 2000 # >2秒空白插入空行
|
||||
|
||||
# 保留的标点（引号类）
|
||||
KEEP_PUNCTUATION = set('""''「」『』《》')
|
||||
|
||||
# 需要去掉的标点
|
||||
REMOVE_PUNCTUATION = re.compile(r'[，。！？、；：…—·\,\.\!\?\;\:]')
|
||||
|
||||
# 语义断句的优先切分点（按优先级排序）
|
||||
BREAK_PATTERNS = [
|
||||
re.compile(r'(?<=[。！？])'),
|
||||
re.compile(r'(?<=[，、；：])'),
|
||||
re.compile(r'(?<=》)'),
|
||||
re.compile('(?<=[”’』」])'), # 右引号后: " ' 』 」
|
||||
]
|
||||
|
||||
|
||||
def clean_punctuation(text: str) -> str:
|
||||
"""去掉标点，保留引号类"""
|
||||
result = []
|
||||
for ch in text:
|
||||
if ch in KEEP_PUNCTUATION:
|
||||
result.append(ch)
|
||||
elif REMOVE_PUNCTUATION.match(ch):
|
||||
continue
|
||||
else:
|
||||
result.append(ch)
|
||||
return "".join(result)
|
||||
|
||||
|
||||
def break_sentence(text: str) -> List[str]:
|
||||
"""
|
||||
将一个句子按语义折行，每行 ≤ MAX_CHARS_PER_LINE 字。
|
||||
先尝试在自然断句点切分，如果不行就硬切。
|
||||
"""
|
||||
if len(text) <= MAX_CHARS_PER_LINE:
|
||||
return [text] if text.strip() else []
|
||||
|
||||
# 14-16字且找不到好断点时，允许不切（人工拍词也偶尔允许略超）
|
||||
if len(text) <= MAX_CHARS_SOFT:
|
||||
# 只有在有明显语义断点时才切
|
||||
for pattern in BREAK_PATTERNS:
|
||||
matches = list(pattern.finditer(text))
|
||||
if matches:
|
||||
pos = matches[-1].end()
|
||||
if 3 <= pos <= len(text) - 3:
|
||||
return [text[:pos].strip(), text[pos:].strip()]
|
||||
return [text]
|
||||
|
||||
lines = []
|
||||
remaining = text
|
||||
|
||||
while len(remaining) > MAX_CHARS_PER_LINE:
|
||||
# 在前14字范围内找最佳切分点（从后往前找）
|
||||
best_pos = -1
|
||||
window = remaining[:MAX_CHARS_PER_LINE]
|
||||
|
||||
# 尝试在语义点切分
|
||||
for pattern in BREAK_PATTERNS:
|
||||
matches = list(pattern.finditer(window))
|
||||
if matches:
|
||||
pos = matches[-1].end()
|
||||
if 3 <= pos <= MAX_CHARS_PER_LINE:
|
||||
best_pos = pos
|
||||
break
|
||||
|
||||
if best_pos == -1:
|
||||
# 没有好的语义切分点，尝试在常见虚词前切
|
||||
for i in range(min(MAX_CHARS_PER_LINE, len(remaining)) - 1, 2, -1):
|
||||
ch = remaining[i]
|
||||
if ch in "的了是在和与而但又或则也还却并且从向把被让给":
|
||||
best_pos = i
|
||||
break
|
||||
|
||||
if best_pos == -1:
|
||||
# 实在找不到，硬切在14字
|
||||
best_pos = MAX_CHARS_PER_LINE
|
||||
|
||||
line = remaining[:best_pos].strip()
|
||||
if line:
|
||||
lines.append(line)
|
||||
remaining = remaining[best_pos:].strip()
|
||||
|
||||
if remaining.strip():
|
||||
lines.append(remaining.strip())
|
||||
|
||||
return lines
|
||||
|
||||
|
||||
def process_sentences(
|
||||
sentences: List[Tuple[int, int, str, int]],
|
||||
) -> List[Tuple[int, int, str]]:
|
||||
"""
|
||||
将 ASR 句子列表处理为折行后的字幕行列表。
|
||||
|
||||
输入: [(start_ms, end_ms, text, speaker_id), ...]
|
||||
输出: [(start_ms, end_ms, text), ...] 其中 text="" 表示空白行
|
||||
|
||||
处理逻辑:
|
||||
1. 检测句子间空白 >2秒 → 插入空白行
|
||||
2. 清理标点
|
||||
3. 按规则折行
|
||||
4. 为折行后的子行分配时间戳（按字数比例）
|
||||
"""
|
||||
if not sentences:
|
||||
return []
|
||||
|
||||
result = []
|
||||
|
||||
for i, (bg, ed, text, _spk) in enumerate(sentences):
|
||||
# 检查与前一句的空白
|
||||
if i > 0:
|
||||
prev_ed = sentences[i - 1][1]
|
||||
gap = bg - prev_ed
|
||||
if gap > SILENCE_THRESHOLD_MS:
|
||||
# 插入空白行，占据空白时段
|
||||
result.append((prev_ed, bg, ""))
|
||||
|
||||
# 清理标点
|
||||
cleaned = clean_punctuation(text)
|
||||
if not cleaned.strip():
|
||||
continue
|
||||
|
||||
# 折行
|
||||
lines = break_sentence(cleaned)
|
||||
if not lines:
|
||||
continue
|
||||
|
||||
# 为每个子行按字数比例分配时间戳
|
||||
total_chars = sum(len(l) for l in lines)
|
||||
duration = ed - bg
|
||||
current_ms = bg
|
||||
|
||||
for line in lines:
|
||||
line_duration = int(duration * len(line) / total_chars) if total_chars > 0 else 0
|
||||
line_end = min(current_ms + line_duration, ed)
|
||||
result.append((current_ms, line_end, line))
|
||||
current_ms = line_end
|
||||
|
||||
return result
|
||||
@@ -0,0 +1,179 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
节目结构切分器 — 将 ASR 结果按节目结构拆分为 5 段
|
||||
|
||||
结构: 导视 + 正片(3段) + 下期预告
|
||||
标志词:
|
||||
- 导视结束 / 正片开始: "各位观众你们好" 或 "我是主持人蓝皓"
|
||||
- 正片结束: "好了观众朋友们" 或 "感谢您.*关注.*军事科技"
|
||||
- 正片之后 = 下期预告
|
||||
|
||||
正片拆3段: 按时长大致均分，优先在角色转换处（speaker_id 变化）切分
|
||||
"""
|
||||
|
||||
import re
|
||||
from typing import List, Tuple, Optional
|
||||
|
||||
# 标志词模式
|
||||
PATTERN_SHOW_START = re.compile(r"各位观众你们好|我是主持人蓝皓")
|
||||
PATTERN_SHOW_END = re.compile(r"好了观众朋友们|感谢您.*关注.*军事科技|感谢您持续关注")
|
||||
|
||||
|
||||
def find_segment_boundaries(
|
||||
sentences: List[Tuple[int, int, str, int]],
|
||||
) -> Tuple[int, int]:
|
||||
"""
|
||||
找到正片开始和结束的句子索引。
|
||||
返回 (show_start_idx, show_end_idx)
|
||||
- show_start_idx: "各位观众你们好"所在句子的索引
|
||||
- show_end_idx: "好了观众朋友们"所在句子的索引
|
||||
"""
|
||||
show_start_idx = 0
|
||||
show_end_idx = len(sentences) - 1
|
||||
|
||||
for i, (_, _, text, _) in enumerate(sentences):
|
||||
if PATTERN_SHOW_START.search(text):
|
||||
show_start_idx = i
|
||||
break
|
||||
|
||||
for i in range(len(sentences) - 1, -1, -1):
|
||||
_, _, text, _ = sentences[i]
|
||||
if PATTERN_SHOW_END.search(text):
|
||||
show_end_idx = i
|
||||
break
|
||||
|
||||
return show_start_idx, show_end_idx
|
||||
|
||||
|
||||
def split_show_into_three(
|
||||
sentences: List[Tuple[int, int, str, int]],
|
||||
start_idx: int,
|
||||
end_idx: int,
|
||||
) -> Tuple[int, int]:
|
||||
"""
|
||||
将正片（start_idx 到 end_idx）拆成 3 段。
|
||||
返回两个切分点索引 (split1_idx, split2_idx)
|
||||
|
||||
策略: 按时长三等分，然后在附近找 speaker_id 变化的位置。
|
||||
"""
|
||||
if end_idx - start_idx < 6:
|
||||
# 太短了，均分
|
||||
third = (end_idx - start_idx) // 3
|
||||
return start_idx + third, start_idx + 2 * third
|
||||
|
||||
show_start_ms = sentences[start_idx][0]
|
||||
show_end_ms = sentences[end_idx][1]
|
||||
total_duration = show_end_ms - show_start_ms
|
||||
|
||||
target1_ms = show_start_ms + total_duration // 3
|
||||
target2_ms = show_start_ms + 2 * total_duration // 3
|
||||
|
||||
split1 = _find_best_split(sentences, start_idx, end_idx, target1_ms)
|
||||
split2 = _find_best_split(sentences, split1 + 1, end_idx, target2_ms)
|
||||
|
||||
# 确保 split2 > split1
|
||||
if split2 <= split1:
|
||||
split2 = split1 + (end_idx - split1) // 2
|
||||
|
||||
return split1, split2
|
||||
|
||||
|
||||
def _find_best_split(
|
||||
sentences: List[Tuple[int, int, str, int]],
|
||||
range_start: int,
|
||||
range_end: int,
|
||||
target_ms: int,
|
||||
search_window: int = 15,
|
||||
) -> int:
|
||||
"""
|
||||
在 target_ms 附近（±search_window 句）找最佳切分点。
|
||||
优先找 speaker_id 变化的位置，其次找 >2秒 空白。
|
||||
"""
|
||||
# 先找到时间上最接近 target_ms 的句子
|
||||
closest_idx = range_start
|
||||
min_diff = abs(sentences[range_start][0] - target_ms)
|
||||
for i in range(range_start, min(range_end + 1, len(sentences))):
|
||||
diff = abs(sentences[i][0] - target_ms)
|
||||
if diff < min_diff:
|
||||
min_diff = diff
|
||||
closest_idx = i
|
||||
|
||||
# 在附近找 speaker 变化点
|
||||
search_lo = max(range_start + 1, closest_idx - search_window)
|
||||
search_hi = min(range_end, closest_idx + search_window)
|
||||
|
||||
best_idx = closest_idx
|
||||
best_score = 0
|
||||
|
||||
for i in range(search_lo, search_hi):
|
||||
score = 0
|
||||
# speaker 变化加分
|
||||
if sentences[i][3] != sentences[i - 1][3] and sentences[i][3] != 0:
|
||||
score += 10
|
||||
# 空白间隔加分
|
||||
gap = sentences[i][0] - sentences[i - 1][1]
|
||||
if gap > 2000:
|
||||
score += 5
|
||||
elif gap > 1000:
|
||||
score += 2
|
||||
# 离目标越近加分
|
||||
time_diff = abs(sentences[i][0] - target_ms)
|
||||
time_score = max(0, 5 - time_diff / 10000)
|
||||
score += time_score
|
||||
|
||||
if score > best_score:
|
||||
best_score = score
|
||||
best_idx = i
|
||||
|
||||
return best_idx
|
||||
|
||||
|
||||
def split_into_segments(
|
||||
sentences: List[Tuple[int, int, str, int]],
|
||||
) -> List[Tuple[str, List[Tuple[int, int, str, int]]]]:
|
||||
"""
|
||||
将全部 ASR 句子拆分为 5 段。
|
||||
返回: [("导视", [...]), ("正片1", [...]), ("正片2", [...]), ("正片3", [...]), ("预告", [...])]
|
||||
|
||||
如果找不到标志词，则只输出单段。
|
||||
"""
|
||||
if not sentences:
|
||||
return [("正片1", [])]
|
||||
|
||||
show_start_idx, show_end_idx = find_segment_boundaries(sentences)
|
||||
|
||||
# 导视: 0 到 show_start_idx-1
|
||||
intro = sentences[:show_start_idx] if show_start_idx > 0 else []
|
||||
|
||||
# 正片: show_start_idx 到 show_end_idx
|
||||
show_sentences = sentences[show_start_idx:show_end_idx + 1]
|
||||
|
||||
# 预告: show_end_idx+1 到末尾
|
||||
trailer = sentences[show_end_idx + 1:] if show_end_idx < len(sentences) - 1 else []
|
||||
|
||||
# 正片拆3段
|
||||
if len(show_sentences) > 6:
|
||||
split1, split2 = split_show_into_three(sentences, show_start_idx, show_end_idx)
|
||||
# 转为相对于 show_sentences 的索引
|
||||
rel_split1 = split1 - show_start_idx
|
||||
rel_split2 = split2 - show_start_idx
|
||||
show_part1 = show_sentences[:rel_split1]
|
||||
show_part2 = show_sentences[rel_split1:rel_split2]
|
||||
show_part3 = show_sentences[rel_split2:]
|
||||
else:
|
||||
show_part1 = show_sentences
|
||||
show_part2 = []
|
||||
show_part3 = []
|
||||
|
||||
segments = []
|
||||
if intro:
|
||||
segments.append(("导视", intro))
|
||||
segments.append(("正片1", show_part1))
|
||||
if show_part2:
|
||||
segments.append(("正片2", show_part2))
|
||||
if show_part3:
|
||||
segments.append(("正片3", show_part3))
|
||||
if trailer:
|
||||
segments.append(("预告", trailer))
|
||||
|
||||
return segments
|
||||
@@ -0,0 +1,49 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
SRT 生成器 — 将折行后的字幕行列表写成大洋系统兼容的 SRT 文件
|
||||
|
||||
格式参照 data/ 下的真实样本:
|
||||
- 序号从 1 开始
|
||||
- 时间格式: HH:MM:SS,mmm --> HH:MM:SS,mmm
|
||||
- 每条字幕一行文字（不多行）
|
||||
- 空白行（屏幕清字幕）写为空内容
|
||||
- 条目之间用空行分隔
|
||||
"""
|
||||
|
||||
from typing import List, Tuple
|
||||
|
||||
|
||||
def ms_to_srt_time(ms: int) -> str:
|
||||
"""毫秒 → SRT 时间格式 HH:MM:SS,mmm"""
|
||||
if ms < 0:
|
||||
ms = 0
|
||||
hours = ms // 3600000
|
||||
minutes = (ms % 3600000) // 60000
|
||||
seconds = (ms % 60000) // 1000
|
||||
millis = ms % 1000
|
||||
return f"{hours:02d}:{minutes:02d}:{seconds:02d},{millis:03d}"
|
||||
|
||||
|
||||
def write_srt(
|
||||
subtitle_lines: List[Tuple[int, int, str]],
|
||||
output_path: str,
|
||||
time_offset: int = 0,
|
||||
) -> None:
|
||||
"""
|
||||
写入 SRT 文件
|
||||
|
||||
subtitle_lines: [(start_ms, end_ms, text), ...] text="" 表示空白行
|
||||
output_path: 输出文件路径
|
||||
time_offset: 时间偏移（用于正片拆分后各段从0开始计时的情况，这里不用，保持绝对时间）
|
||||
"""
|
||||
with open(output_path, "w", encoding="utf-8") as f:
|
||||
for idx, (start_ms, end_ms, text) in enumerate(subtitle_lines, 1):
|
||||
start = ms_to_srt_time(start_ms - time_offset)
|
||||
end = ms_to_srt_time(end_ms - time_offset)
|
||||
f.write(f"{idx}\n")
|
||||
f.write(f"{start} --> {end}\n")
|
||||
# 空白行写一个空格（参照样本中的做法）
|
||||
f.write(f"{text if text else ' '}\n")
|
||||
f.write("\n")
|
||||
|
||||
print(f"[SRT] 已写入: {output_path} ({len(subtitle_lines)} 条)")
|
||||
@@ -0,0 +1,136 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
本地测试 — 不需要讯飞凭证
|
||||
用模拟的 ASR 数据测试折行引擎和 SRT 生成
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent / "src"))
|
||||
|
||||
from line_breaker import break_sentence, clean_punctuation, process_sentences
|
||||
from srt_writer import write_srt, ms_to_srt_time
|
||||
from segment_splitter import split_into_segments, find_segment_boundaries
|
||||
|
||||
|
||||
def test_clean_punctuation():
|
||||
print("=== 测试标点清理 ===")
|
||||
cases = [
|
||||
('对于战后日本而言，重新武装最大的限制不是工业能力，而是政治约束。', '对于战后日本而言重新武装最大的限制不是工业能力而是政治约束'),
|
||||
('这是“和平宪法”的约束', '这是“和平宪法”的约束'),
|
||||
('你好！世界？', '你好世界'),
|
||||
]
|
||||
for input_text, expected in cases:
|
||||
result = clean_punctuation(input_text)
|
||||
status = "✓" if result == expected else "✗"
|
||||
print(f" {status} 输入: {input_text}")
|
||||
print(f" 结果: {result}")
|
||||
if result != expected:
|
||||
print(f" 期望: {expected}")
|
||||
print()
|
||||
|
||||
|
||||
def test_break_sentence():
|
||||
print("=== 测试折行 ===")
|
||||
cases = [
|
||||
"对于战后日本而言重新武装最大的限制不是工业能力而是政治约束",
|
||||
"各位观众你们好欢迎收看军事科技我是主持人蓝皓",
|
||||
"F-35B隐身战斗机",
|
||||
"日本通过从美国引进F-35战机和战斧巡航导弹等先进装备迅速获得了第五代战机远程精确打击能力",
|
||||
]
|
||||
for text in cases:
|
||||
lines = break_sentence(text)
|
||||
print(f" 输入 ({len(text)}字): {text}")
|
||||
for line in lines:
|
||||
print(f" → [{len(line):2d}字] {line}")
|
||||
# 验证每行不超过14字
|
||||
for line in lines:
|
||||
assert len(line) <= 14, f"超过14字: {line} ({len(line)}字)"
|
||||
print()
|
||||
|
||||
|
||||
def test_process_sentences():
|
||||
print("=== 测试完整折行流程 ===")
|
||||
# 模拟 ASR 句子（参照 data/1.srt 内容）
|
||||
sentences = [
|
||||
(8520, 8840, "", 0), # 空句
|
||||
(8880, 9880, "各位观众你们好", 1),
|
||||
(9920, 11320, "欢迎收看军事科技", 1),
|
||||
(11360, 12280, "我是主持人蓝皓", 1),
|
||||
(12320, 13120, " ", 0), # 空白
|
||||
(13160, 14400, "日本国会参议院", 1),
|
||||
(14440, 16000, "今天上午6月26日表决通过了", 1),
|
||||
(16040, 18480, "防卫省设置法修正案等法案", 1),
|
||||
# 模拟 >2秒空白
|
||||
(22000, 24320, "这是日本自1954年", 1),
|
||||
(24360, 29040, "成立陆上海上航空自卫队以来", 1),
|
||||
]
|
||||
|
||||
result = process_sentences(sentences)
|
||||
print(f" 输入 {len(sentences)} 句 → 输出 {len(result)} 行")
|
||||
for start, end, text in result:
|
||||
time_str = f"{ms_to_srt_time(start)} --> {ms_to_srt_time(end)}"
|
||||
print(f" {time_str} {text if text else '[空白]'}")
|
||||
print()
|
||||
|
||||
|
||||
def test_segment_detection():
|
||||
print("=== 测试节目结构识别 ===")
|
||||
sentences = [
|
||||
(0, 5000, "本期军事科技将为您介绍", 1),
|
||||
(5000, 8000, "日本军备的发展历程", 1),
|
||||
(8880, 9880, "各位观众你们好", 1),
|
||||
(9920, 11320, "欢迎收看军事科技", 1),
|
||||
(11360, 12280, "我是主持人蓝皓", 1),
|
||||
(13000, 50000, "正片内容一", 1),
|
||||
(50000, 100000, "正片内容二", 2),
|
||||
(100000, 150000, "正片内容三", 1),
|
||||
(150000, 200000, "正片内容四", 2),
|
||||
(200000, 250000, "正片内容五", 1),
|
||||
(250000, 300000, "正片内容六", 2),
|
||||
(300000, 310000, "好了观众朋友们感谢您持续关注国防军事频道军事科技", 1),
|
||||
(310000, 320000, "下期我们将继续关注", 1),
|
||||
(320000, 330000, "某某话题", 1),
|
||||
]
|
||||
|
||||
start_idx, end_idx = find_segment_boundaries(sentences)
|
||||
print(f" 正片开始: 第{start_idx}句 = {sentences[start_idx][2]}")
|
||||
print(f" 正片结束: 第{end_idx}句 = {sentences[end_idx][2]}")
|
||||
|
||||
segments = split_into_segments(sentences)
|
||||
for name, seg in segments:
|
||||
if seg:
|
||||
print(f" [{name}] {len(seg)}句, 时间 {ms_to_srt_time(seg[0][0])} ~ {ms_to_srt_time(seg[-1][1])}")
|
||||
else:
|
||||
print(f" [{name}] 空")
|
||||
print()
|
||||
|
||||
|
||||
def test_srt_output():
|
||||
print("=== 测试 SRT 输出 ===")
|
||||
subtitle_lines = [
|
||||
(8880, 9880, "各位观众你们好"),
|
||||
(9920, 11320, "欢迎收看《军事科技》"),
|
||||
(11360, 12280, "我是主持人蓝皓"),
|
||||
(12320, 13120, ""), # 空白行
|
||||
(13160, 14400, "日本国会参议院"),
|
||||
]
|
||||
|
||||
output_path = Path(__file__).parent / "output" / "test.srt"
|
||||
output_path.parent.mkdir(exist_ok=True)
|
||||
write_srt(subtitle_lines, str(output_path))
|
||||
|
||||
# 读回来检查
|
||||
content = output_path.read_text(encoding="utf-8")
|
||||
print(content[:500])
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
test_clean_punctuation()
|
||||
test_break_sentence()
|
||||
test_process_sentences()
|
||||
test_segment_detection()
|
||||
test_srt_output()
|
||||
print("=== 全部测试完成 ===")
|
||||
Reference in New Issue
Block a user