feat: CCA v6 腾讯云部署 + 审稿台（含查找替换）
- deploy/cca_route.py: Flask 蓝图（6个API端点），WAV自动转MP3 - deploy/cca.html: 4步单页流程（上传→处理→审稿→下载），查找替换(Ctrl+H) - src/term_normalizer.py: 新增正则层（同音字/引号/书名号/小数点/波浪号） - src/ai_proofreader.py: speaker角色识别+专家段增强Prompt+的地得加强 - src/ai_line_breaker.py: 引号不跨屏+极短行合并+短句合并间隔放宽 - cca_pipeline.py: Step 2.5 校对后二次正则兜底 - 已部署至 http://101.42.29.217/cca.html Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
simonkoson
+313
-3
@@ -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
|
||||
|
||||
+176
-65
@@ -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
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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'(?<!“)' + re.escape(target) + r'(?!”)'
|
||||
)
|
||||
text = pattern.sub(replacement, text)
|
||||
elif not suffix:
|
||||
# 单独出现（无后缀），但要避免替换那些在 A 稿中不带引号的后缀组合
|
||||
# 用负向前瞻排除不该加引号的后缀
|
||||
exclude_chars = "".join(bare_suffixes - {""}) if bare_suffixes else ""
|
||||
if exclude_chars:
|
||||
lookahead = f'(?![{re.escape(exclude_chars)}])'
|
||||
else:
|
||||
lookahead = ''
|
||||
pattern = re.compile(
|
||||
r'(?<!“)(?<!《)' + re.escape(word) + lookahead + r'(?!”)(?!》)'
|
||||
)
|
||||
text = pattern.sub(f'“{word}”', text)
|
||||
return text
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 中文数字修复
|
||||
# ========================================================================
|
||||
|
||||
CHINESE_NUM_FIXES = [
|
||||
(re.compile(r'数10([年架艘枚门辆台套件个发种类])'), r'数十\1'),
|
||||
(re.compile(r'数100([年架艘枚门辆台套件个发种类])'), r'数百\1'),
|
||||
(re.compile(r'数1000([年架艘枚门辆台套件个发种类])'), r'数千\1'),
|
||||
(re.compile(r'几10([年架艘枚门辆台套件个发种类])'), r'几十\1'),
|
||||
(re.compile(r'几100([年架艘枚门辆台套件个发种类])'), r'几百\1'),
|
||||
]
|
||||
|
||||
|
||||
def _fix_chinese_numbers(text: str) -> 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'(?<!《)' + re.escape(bare) + r'(?!》)')
|
||||
text = pattern.sub(titled, text)
|
||||
return text
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 小数点丢失修复（09马赫→0.9马赫 等）
|
||||
# ========================================================================
|
||||
|
||||
# 匹配丢失小数点的情况：
|
||||
# 1. "09马赫" → "0.9马赫"（数字在单位前）
|
||||
# 2. "马赫数09" → "马赫数0.9"（数字在单位后）
|
||||
# 3. 通用：非正常的 0+单个数字 紧跟/紧接单位
|
||||
LOST_DECIMAL_BEFORE_UNIT = re.compile(r'(?<!\d)0(\d)(\s*(?:马赫|倍|秒|米|千米|公里))')
|
||||
LOST_DECIMAL_AFTER_UNIT = re.compile(r'(马赫数|倍数|速度约)0(\d)(?!\d)')
|
||||
|
||||
|
||||
def _fix_lost_decimal(text: str) -> 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
|
||||
Reference in New Issue
Block a user