doco: v2说话人分段模式 — ASR说话人分离+大block拆分+三维动画解说识别
- asr_adapter: 新增roleType=1说话人分离参数，新增parse_order_result_with_speaker()，write_asr_result自动输出asr_v2_timed_spk.txt - fusion_align: 新增speaker-aware alignment v2流程（_annotate_b_lines_with_speakers区间匹配、_detect_speaker_blocks、SYSTEM_PROMPT_SPEAKER_ALIGN大block拆分prompt、_build_broadcast_segments支持block内多段拆分） - cli: 兼容v1/v2 stats字典 - 新增convert_to_md.py（20期融合A稿docx转md+YAML frontmatter） - backup_before_spk/: 修改前代码备份
This commit is contained in:
simonkoson
@@ -0,0 +1,352 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
讯飞 ASR 适配层
|
||||
=================================================
|
||||
来源: demo 跑通的 xfyun_asr_standard.py
|
||||
改动: 凭证从环境变量读取,不再硬编码
|
||||
|
||||
接口: https://raasr.xfyun.cn/v2/api/upload / getResult
|
||||
签名: signa = base64(HmacSHA1(MD5(appid + ts), secretKey))
|
||||
|
||||
特性:
|
||||
- 支持热词列表(hotWord),提升专业术语识别率
|
||||
- 支持军事领域参数(pd=mil)
|
||||
- 支持顺滑+口语规整(输出更接近书面语)
|
||||
- 默认语种 cn(中文普通话),免费包标配
|
||||
|
||||
凭证来源: 环境变量
|
||||
- XFYUN_APP_ID
|
||||
- XFYUN_SECRET_KEY
|
||||
"""
|
||||
|
||||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
import wave
|
||||
from pathlib import Path
|
||||
from urllib.parse import quote
|
||||
from typing import List, Tuple, Optional
|
||||
|
||||
import requests
|
||||
|
||||
# ========================================================================
|
||||
# 凭证 — 优先加载 doco/.env(与 llm.py 相同方式)
|
||||
# ========================================================================
|
||||
|
||||
try:
|
||||
from dotenv import load_dotenv
|
||||
|
||||
_doco_env = Path(__file__).resolve().parent.parent.parent / ".env" # doco/.env
|
||||
if _doco_env.exists():
|
||||
load_dotenv(str(_doco_env), override=True)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
APP_ID = os.environ.get("XFYUN_APP_ID", "").strip()
|
||||
SECRET_KEY = os.environ.get("XFYUN_SECRET_KEY", "").strip()
|
||||
|
||||
if not APP_ID or not SECRET_KEY:
|
||||
print("[ASR 配置错误] XFYUN_APP_ID 或 XFYUN_SECRET_KEY 未配置或为空", file=sys.stderr)
|
||||
print("[ASR 配置错误] 请在 doco/.env 中设置这两个环境变量", file=sys.stderr)
|
||||
print("[ASR 配置错误] 格式: XFYUN_APP_ID=你的appid / XFYUN_SECRET_KEY=你的secret", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 接口配置
|
||||
# ========================================================================
|
||||
|
||||
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" # 口语规整
|
||||
|
||||
# 轮询配置
|
||||
POLL_INTERVAL_SECONDS = 30
|
||||
MAX_WAIT_MINUTES = 30
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 热词列表(每期节目调用前从 A 稿提取)
|
||||
# ========================================================================
|
||||
|
||||
def get_hot_words(episode_id: str) -> List[str]:
|
||||
"""
|
||||
读取 programs/<episode_id>/本期热词表.txt，
|
||||
按 "|" 切分、strip、去空去重，返回 List[str]。
|
||||
文件缺失返回 [] 并 stderr 警告(不退出)。
|
||||
"""
|
||||
from pathlib import Path as _Path
|
||||
|
||||
# doco 项目根 = doco/src/doco/asr_adapter.py → 上3级到达 doco/
|
||||
_project_root = _Path(__file__).resolve().parent.parent.parent
|
||||
hotwords_file = _project_root / "programs" / episode_id / "本期热词表.txt"
|
||||
|
||||
if not hotwords_file.exists():
|
||||
print(f"[ASR 热词] 未找到热词表: {hotwords_file}，热词跳过", file=sys.stderr)
|
||||
return []
|
||||
|
||||
try:
|
||||
raw = hotwords_file.read_text(encoding="utf-8")
|
||||
except Exception as e:
|
||||
print(f"[ASR 热词] 读取热词表失败: {e}", file=sys.stderr)
|
||||
return []
|
||||
|
||||
# 按 | 切分、strip、过滤空字符串、去重(保持顺序)
|
||||
words: List[str] = []
|
||||
seen: set = set()
|
||||
for token in raw.split("|"):
|
||||
w = token.strip()
|
||||
if w and w not in seen:
|
||||
seen.add(w)
|
||||
words.append(w)
|
||||
|
||||
return words
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 签名+工具
|
||||
# ========================================================================
|
||||
|
||||
|
||||
def make_signa(app_id: str, secret_key: str, ts: str) -> str:
|
||||
"""
|
||||
讯飞老版签名:signa = base64(HmacSHA1(MD5(appid + ts), secretKey))
|
||||
"""
|
||||
base_string = (app_id + ts).encode("utf-8")
|
||||
md5_str = hashlib.md5(base_string).hexdigest() # 32位小写hex
|
||||
mac = hmac.new(
|
||||
secret_key.encode("utf-8"),
|
||||
md5_str.encode("utf-8"),
|
||||
digestmod=hashlib.sha1,
|
||||
)
|
||||
signa = base64.b64encode(mac.digest()).decode("utf-8")
|
||||
return signa
|
||||
|
||||
|
||||
def get_audio_duration_ms(filepath: str) -> int:
|
||||
"""获取音频时长(毫秒)。WAV用内置,MP3用mutagen。"""
|
||||
ext = os.path.splitext(filepath)[1].lower()
|
||||
|
||||
if ext == ".wav":
|
||||
with wave.open(filepath, "rb") as wf:
|
||||
n_frames = wf.getnframes()
|
||||
sample_rate = wf.getframerate()
|
||||
duration_ms = int(round(n_frames / sample_rate * 1000))
|
||||
return duration_ms
|
||||
|
||||
if ext == ".mp3":
|
||||
try:
|
||||
from mutagen.mp3 import MP3
|
||||
return int(MP3(filepath).info.length * 1000)
|
||||
except ImportError:
|
||||
return 0
|
||||
|
||||
raise ValueError(f"不支持的音频格式: {ext}")
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 上传
|
||||
# ========================================================================
|
||||
|
||||
|
||||
def upload_audio(
|
||||
filepath: str,
|
||||
hot_words: Optional[List[str]] = None,
|
||||
) -> str:
|
||||
"""上传音频,返回 orderId"""
|
||||
if not os.path.exists(filepath):
|
||||
raise FileNotFoundError(f"音频文件不存在: {filepath}")
|
||||
|
||||
if not APP_ID or not SECRET_KEY:
|
||||
raise ValueError("请先设置 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)
|
||||
|
||||
# 构建URL参数
|
||||
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,
|
||||
}
|
||||
|
||||
# 热词,用 | 分隔
|
||||
if hot_words:
|
||||
hot_word_str = "|".join(hot_words)
|
||||
params["hotWord"] = hot_word_str
|
||||
|
||||
url_parts = [f"{quote(k, safe='')}={quote(str(v), safe='')}" for k, v in params.items()]
|
||||
url = f"{UPLOAD_URL}?{'&'.join(url_parts)}"
|
||||
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
with open(filepath, "rb") as f:
|
||||
audio_bytes = f.read()
|
||||
|
||||
resp = requests.post(url, headers=headers, 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"]
|
||||
return order_id
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 查询结果
|
||||
# ========================================================================
|
||||
|
||||
|
||||
def query_result(order_id: str) -> dict:
|
||||
"""单次查询"""
|
||||
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)
|
||||
return resp.json()
|
||||
|
||||
|
||||
def poll_until_done(order_id: str) -> dict:
|
||||
"""轮询直到完成"""
|
||||
start_time = time.time()
|
||||
while True:
|
||||
elapsed = time.time() - start_time
|
||||
if elapsed > MAX_WAIT_MINUTES * 60:
|
||||
raise TimeoutError(f"超过 {MAX_WAIT_MINUTES} 分钟未完成")
|
||||
|
||||
data = query_result(order_id)
|
||||
order_info = data.get("content", {}).get("orderInfo", {})
|
||||
status = order_info.get("status")
|
||||
fail_type = order_info.get("failType", 0)
|
||||
|
||||
if status == 4:
|
||||
return data
|
||||
if status == -1:
|
||||
raise RuntimeError(f"转写失败: failType={fail_type}, 数据: {data}")
|
||||
|
||||
time.sleep(POLL_INTERVAL_SECONDS)
|
||||
|
||||
|
||||
# ========================================================================
|
||||
# 结果解析
|
||||
# ========================================================================
|
||||
|
||||
|
||||
def parse_order_result(order_result_str: str) -> List[Tuple[int, int, str]]:
|
||||
"""
|
||||
解析嵌套JSON,返回 [(sentence_start_ms, sentence_end_ms, text), ...]
|
||||
"""
|
||||
if not order_result_str:
|
||||
return []
|
||||
|
||||
cleaned = re.sub(r"\\\\", r"\\", order_result_str)
|
||||
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))
|
||||
|
||||
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))
|
||||
|
||||
return sentences
|
||||
|
||||
|
||||
def format_timestamp(ms: int) -> str:
|
||||
"""毫秒转 [Nm Ns] 格式"""
|
||||
total_sec = ms // 1000
|
||||
return f"{total_sec // 60}m{total_sec % 60}s"
|
||||
|
||||
|
||||
def transcribe(
|
||||
audio_path: str,
|
||||
hot_words: Optional[List[str]] = None,
|
||||
) -> Tuple[List[Tuple[int, int, str]], str]:
|
||||
"""
|
||||
完整转写流程:上传 → 轮询 → 解析
|
||||
返回 (sentences, raw_order_result_json_str)
|
||||
- sentences: [(start_ms, end_ms, text), ...]
|
||||
- raw_order_result_json_str: 讯飞原始 orderResult 字段原文(用于断点续跑落盘)
|
||||
"""
|
||||
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_order_result(order_result_str)
|
||||
return sentences, order_result_str
|
||||
|
||||
|
||||
def write_asr_result(
|
||||
sentences: List[Tuple[int, int, str]],
|
||||
output_dir: str,
|
||||
raw_order_result: str = "",
|
||||
) -> Tuple[str, str]:
|
||||
"""
|
||||
将 ASR 结果写入文件
|
||||
返回 (timed_txt_path, raw_json_path)
|
||||
"""
|
||||
os.makedirs(output_dir, exist_ok=True)
|
||||
|
||||
timed_lines = [f"[{format_timestamp(bg)}] {text}" for bg, _, text in sentences]
|
||||
timed_path = os.path.join(output_dir, "asr_v2_timed.txt")
|
||||
with open(timed_path, "w", encoding="utf-8") as f:
|
||||
f.write("\n".join(timed_lines))
|
||||
|
||||
raw_path = os.path.join(output_dir, "asr_result_raw.json")
|
||||
with open(raw_path, "w", encoding="utf-8") as f:
|
||||
if raw_order_result:
|
||||
f.write(raw_order_result)
|
||||
else:
|
||||
# 没有原始数据时写空对象(兼容旧调用)
|
||||
f.write("{}")
|
||||
|
||||
return timed_path, raw_path
|
||||
Reference in New Issue
Block a user