simonkoson
1c3963d17c
- 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/: 修改前代码备份
353 lines
11 KiB
Python
353 lines
11 KiB
Python
# -*- 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
|