#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""本地转写通话录音：whisper medium + ECAPA 说话人聚类（两人）。
全程本机运行，仅从 PyPI / openaipublic / huggingface 下载安装包与模型权重，录音不会上传。
"""
import json
import os
import re
import subprocess
import sys
import numpy as np
BASE = "/workspace/iso-v6/main"
AUDIO = os.path.join(BASE, "材料", "通话录音_20261013_1032.m4a")
WORK = os.path.join(BASE, ".work")
OUT_TRANSCRIPT = os.path.join(BASE, "材料", "通话录音_20261013_1032_转写.txt")
OUT_REVIEW = os.path.join(BASE, "材料", "通话录音_20261013_1032_复核清单.txt")
os.makedirs(WORK, exist_ok=True)
WAV = os.path.join(WORK, "call_16k.wav")
INITIAL_PROMPT = "以下是普通话的对话内容。"
def fmt_ts(sec, with_sec=True):
m, s = divmod(int(round(sec)), 60)
h, m = divmod(m, 60)
return f"{h:d}:{m:02d}:{s:02d}" if h else f"{m:02d}:{s:02d}"
def join_tokens(tokens):
out = ""
for t in tokens:
t = t.strip()
if not t:
continue
if out and re.search(r"[0-9A-Za-z]$", out) and re.match(r"^[0-9A-Za-z]", t):
out += " "
out += t
return out
# ---------- 阶段 1：whisper 转写 ----------
import whisper
print("[1/4] 加载 whisper medium 并转写（CPU）...", flush=True)
model = whisper.load_model("medium")
result = model.transcribe(
AUDIO,
language="zh",
initial_prompt=INITIAL_PROMPT,
word_timestamps=True,
fp16=False,
verbose=False,
)
with open(os.path.join(WORK, "whisper_raw.json"), "w", encoding="utf-8") as f:
json.dump(result, f, ensure_ascii=False, indent=1)
words = [] # (start, end, token)
for seg in result["segments"]:
for w in seg.get("words") or []:
words.append((float(w["start"]), float(w["end"]), w["word"]))
words.sort(key=lambda x: x[0])
print(f" 转写得到 {len(result['segments'])} 段 / {len(words)} 词", flush=True)
# ---------- 阶段 2：说话人嵌入 + 二聚类 ----------
print("[2/4] 提取说话人嵌入（ECAPA）...", flush=True)
if os.path.exists(WAV):
os.remove(WAV)
subprocess.run(
["ffmpeg", "-y", "-v", "error", "-i", AUDIO, "-ac", "1", "-ar", "16000", WAV],
check=True,
)
import torch
import torchaudio # noqa: F401 (确保后端可用)
import soundfile as sf
wav, sr = sf.read(WAV, dtype="float32")
assert sr == 16000
wav_t = torch.from_numpy(wav).unsqueeze(0)
from speechbrain.inference.speaker import EncoderClassifier
classifier = EncoderClassifier.from_hparams(
source="speechbrain/spkrec-ecapa-voxceleb",
savedir=os.path.join(WORK, "ecapa"),
run_opts={"device": "cpu"},
)
def embed(chunk_wav):
with torch.no_grad():
emb = classifier.encode_batch(chunk_wav)
v = emb.squeeze().numpy().astype(np.float64)
return v / (np.linalg.norm(v) + 1e-9)
# 按 whisper 词级时间戳切块：在停顿 / 句末标点处切，块长 0.8~5 秒，
# 快速问答也要切开（问答间常无静音，但有句末标点）
chunks = []
cur = []
def close_cur():
global cur
if cur:
chunks.append((cur[0][0], cur[-1][1]))
cur = []
for st, en, tok in words:
if cur:
gap = st - cur[-1][1]
dur = cur[-1][1] - cur[0][0]
endpunct = bool(re.search(r"[。？！?!，,；；]$", cur[-1][2].strip()))
if gap >= 0.18 or (endpunct and dur >= 1.2) or dur >= 5.0:
close_cur()
cur.append((st, en, tok))
close_cur()
# 合并过短的块（<0.8s）：紧跟前块（停顿小）则并入前块，否则独立成块
merged = []
for c in chunks:
if merged and (c[1] - c[0]) < 0.8 and c[0] - merged[-1][1] < 0.25:
merged[-1] = (merged[-1][0], c[1])
else:
merged.append(list(c))
chunks = [tuple(c) for c in merged]
embs = []
for st, en in chunks:
i0, i1 = int(st * 16000), int(en * 16000)
embs.append(embed(wav_t[:, i0:i1]))
embs = np.stack(embs)
from sklearn.cluster import KMeans
km = KMeans(n_clusters=2, n_init=50, random_state=0).fit(embs)
labels = km.labels_.astype(int)
# 仅纠正“孤立短块”：前后块同标签、自身短且与两个质心距离接近时才翻转。
# 不做多数平滑/全局解码——那会把一问一答的快速交替抹平。
cents = km.cluster_centers_ / np.linalg.norm(km.cluster_centers_, axis=1, keepdims=True)
dcos = 1.0 - embs @ cents.T # [n块, 2质心] 余弦距离
margin = np.abs(dcos[:, 0] - dcos[:, 1])
for i in range(1, len(labels) - 1):
dur = chunks[i][1] - chunks[i][0]
if labels[i - 1] == labels[i + 1] != labels[i] and dur < 1.0 and margin[i] < 0.3:
labels[i] = labels[i - 1]
if os.environ.get("DEBUG_SPK"):
for idx, (st, en) in enumerate(chunks):
txt = join_tokens([t for s, e, t in words if s >= st - 1e-6 and e <= en + 1e-6])
print(f" chunk {idx:2d} [{st:6.2f}-{en:6.2f}] spk{labels[idx]} {txt}", file=sys.stderr)
print(f" {len(chunks)} 个语音块完成聚类", flush=True)
# ---------- 阶段 3：词→说话人→对话轮次 ----------
def word_speaker(st, en):
for idx, (cs, ce) in enumerate(chunks):
if st < ce and en > cs: # 重叠即归属
return labels[idx]
return labels[int(np.argmin([min(abs(st - cs), abs(st - ce)) for cs, ce in chunks]))]
turns = []
for st, en, tok in words:
spk = word_speaker(st, en)
if turns and turns[-1]["spk"] == spk and st - turns[-1]["end"] <= 2.0:
turns[-1]["tokens"].append(tok)
turns[-1]["end"] = max(turns[-1]["end"], en)
else:
turns.append({"spk": spk, "start": st, "end": en, "tokens": [tok]})
first_seen = {}
for t in turns:
first_seen.setdefault(t["spk"], t["start"])
order = sorted(first_seen, key=lambda s: first_seen[s])
spk_name = {s: f"说话人{chr(ord('A') + order.index(s))}" for s in order}
for t in turns:
t["text"] = join_tokens(t["tokens"])
# whisper 分段置信度 → 标注到对应轮次
low_conf = []
for seg in result["segments"]:
bad = []
if seg.get("avg_logprob", 0) < -0.8:
bad.append("模型置信度低")
if seg.get("no_speech_prob", 0) > 0.5:
bad.append("疑似噪声/无语音")
if seg.get("compression_ratio", 1) > 2.4:
bad.append("疑似重复（可能为幻觉）")
if bad:
low_conf.append((float(seg["start"]), float(seg["end"]), bad))
for t in turns:
t["flags_conf"] = sorted(
{r for ls, le, rs in low_conf if t["start"] < le and t["end"] > ls for r in rs}
)
# ---------- 阶段 4：复核清单规则（金额/日期/人名） ----------
CN_NUM = r"[零一二两三四五六七八九十百千万亿]+"
AR_NUM = r"\d[\d,]*(?:\.\d+)?"
AMOUNT_RE = [
(re.compile(r"[¥￥]\s*" + AR_NUM), "金额"),
(re.compile(AR_NUM + r"\s*(?:万|千|百)(?:\s*(?:多|来|左右|块钱|块|元))?"), "金额"),
(re.compile(AR_NUM + r"\s*(?:块钱|块|元|毛)"), "金额"),
(re.compile(CN_NUM + r"(?:万|千|百)(?:" + CN_NUM + r")?(?:元|块|多)?"), "金额"),
(re.compile(r"(?:一分|两分|三分|四分|五分|半分|[0-9.]+\s*分)(?:利|息)"), "金额(利率)"),
(re.compile(r"\b[1-9]\d{2,}\b(?![年月日号时])"), "金额(裸数字，需人工核对)"),
(re.compile(r"\d{11}"), "电话号码"),
]
DATE_RE = [
(re.compile(r"\d{1,4}\s*年(?:初|中|底|前|后)?"), "日期"),
(re.compile(r"(?:\d{1,2}|[零一二两三四五六七八九十]+)\s*月(?:初|中|底)?"), "日期"),
(re.compile(r"(?:\d{1,2}|[零一二两三四五六七八九十]+)\s*[日号](?:之前|以前|前后|左右)?"), "日期"),
(re.compile(r"星期[一二三四五六日天]|周[一二三四五六日天](?:末)?"), "日期"),
(re.compile(r"今年|去年|前年|明年|后年|上个月|下个月|这个月|那个月|月初|月底|年底|年初|年中|前天|昨天|今天|明天|后天|过了年|开年|年底前|春节|清明|五一|端午|中秋|国庆|十一"), "日期(相对)"),
(re.compile(r"\d{1,2}\s*[点:：]\s*\d{0,2}(?:分|半)?|\d{1,2}\s*点半"), "时间点"),
]
# 已知人名/机构来自案卷其他材料，仅用于标注复核位置，不写入转写内容
KNOWN_ENT = re.compile(
r"闫国栋|国栋|祁晓棠|晓棠|卢建平|建平|老卢|小卢|林可|江蕴|筑岸|筑岸装饰|晓棠茶行|茶行|转塘|商会|西湖区|文三路"
)
GENERIC_NAME_RE = [
(re.compile(r"[\u4e00-\u9fa5]{1,3}(?:总|工|老师|律师|经理|老板娘|老板|哥|姐|叔|姨|先生|女士)"), "人名(称呼)"),
(re.compile(r"(?:我|他|她|它|那|这)(?:们)?(?:叫|姓)[\u4e00-\u9fa5]{1,3}"), "人名(自称/称谓)"),
(re.compile(r"[王李张刘陈杨黄赵周吴徐孙马朱胡郭何高林罗郑梁谢宋唐许韩冯邓曹彭曾肖田董袁潘于蒋蔡余杜叶程苏魏吕丁任沈姚卢姜崔钟谭陆汪范金石廖贾夏韦付方白邹孟熊秦邱江尹薛闫段雷侯龙史陶黎贺顾毛郝龚邵万钱严覃武戴莫孔向汤][\u4e00-\u9fa5]{1,2}(?=[，。！？、；：\s]|$)"), "人名(疑似)"),
]
def find_flags(text):
hits = []
for rex, why in AMOUNT_RE + DATE_RE + GENERIC_NAME_RE:
for m in rex.finditer(text):
hits.append((m.start(), why, m.group()))
for m in KNOWN_ENT.finditer(text):
hits.append((m.start(), "人名/机构(案卷已知)", m.group()))
hits.sort()
dedup, seen_span = [], []
for pos, why, g in hits:
if any(s <= pos < e for s, e in seen_span):
continue
if why.startswith("人名(疑似)") and g.startswith("叶") and pos > 0 and text[pos - 1] == "茶":
continue # “茶叶钱”之类不是人名
seen_span.append((pos, pos + len(g)))
dedup.append((why, g))
return dedup
# 结合案卷上下文判断的疑似误听（不改转写内容，仅提示回听）
SUSPECTS = [
(re.compile(r"七姐"), "疑为“祁姐”（案卷来访人姓祁）"),
(re.compile(r"在对几个地方"), "疑为“核对几个地方”"),
(re.compile(r"名钱"), "疑为“名茶”"),
(re.compile(r"元件"), "疑为“原件”"),
(re.compile(r"水资"), "疑为“茶水渍”"),
]
# ---------- 输出 ----------
print("[3/4] 写转写文件...", flush=True)
lines = []
lines.append("通话录音转写")
lines.append(f"录音文件：通话录音_20261013_1032.m4a（时长 {fmt_ts(89.856)}）")
lines.append("转写方式：whisper medium，本机运行；提示词：\u201c" + INITIAL_PROMPT + "\u201d")
lines.append("说话人：自动聚类区分为 2 人；说话人A/B 按先后开口顺序编号，不代表真实身份。")
lines.append("备注（依据内容的推断，非录音原文）：00:00 处说话人A 自我介绍为江律师助理“林可”，")
lines.append("说话人B 应为来访人。")
lines.append("以下为逐句原样转写，未做增删改写。")
lines.append("")
for t in turns:
lines.append(f"[{fmt_ts(t['start'])}–{fmt_ts(t['end'])}] {spk_name[t['spk']]}：{t['text']}")
with open(OUT_TRANSCRIPT, "w", encoding="utf-8") as f:
f.write("\n".join(lines) + "\n")
print("[4/4] 生成复核清单...", flush=True)
review = []
for i, t in enumerate(turns):
reasons = [(w, g) for w, g in find_flags(t["text"])]
for r in t["flags_conf"]:
reasons.append((r, "—"))
for rex, hint in SUSPECTS:
if m := rex.search(t["text"]):
reasons.append((f"疑似误听：{hint}", m.group()))
if reasons:
review.append((t, reasons))
rlines = []
rlines.append("复核清单 — 通话录音_20261013_1032.m4a")
rlines.append(f"建议回听 {len(review)} 处（含金额 / 日期 / 时间点 / 人名 / 模型没把握 / 疑似误听的句子）。")
rlines.append('说明：标注“裸数字/疑似”为规则自动匹配，宁多勿漏；“疑似误听”是结合案卷材料的判断，')
rlines.append("转写正文未做任何改动，回听后自行取舍。")
rlines.append("")
for i, (t, reasons) in enumerate(review, 1):
rlines.append(f"{i}. [{fmt_ts(t['start'])}–{fmt_ts(t['end'])}] {spk_name[t['spk']]}")
rlines.append(f" 原文：{t['text']}")
seen = []
for w, g in reasons:
item = f"{w}（“{g}”）" if g != "—" else w
if item not in seen:
seen.append(item)
rlines.append(" 原因：" + "；".join(seen))
rlines.append("")
with open(OUT_REVIEW, "w", encoding="utf-8") as f:
f.write("\n".join(rlines) + "\n")
print("DONE", flush=True)
print(json.dumps({"turns": len(turns), "review": len(review)}, ensure_ascii=False))