HTTPS
https://git.nageyqma.icu/nageyqma/audio_to_text.git
SSH
git@git.nageyqma.icu:nageyqma/audio_to_text.git
HTTPS 链接可直接 git clone；推送代码请使用 SSH。
仅展示默认分支 main 最近 100 条提交。
| feat: 本地课堂录音批量转文字（FunASR + faster-whisper 混合引擎） - 中文主导录音走 FunASR SeACo-Paraformer（原生热词偏置），英/法等走 faster-whisper
- 按文件自动探测语言并路由引擎，支持 --language/--engine/--model/--dry-run/--download-only
- 断点续传：输出 md 比音频新则跳过；单文件失败不中断批次
- run.sh 依赖未就绪时自动执行 setup.sh（装 uv/venv/CPU torch + 预下载约 3GB 模型）
- .gitignore 排除模型缓存、虚拟环境、录音与转写数据（隐私与大体积） 查看变更.gitignore | 26 +++++++ README.md | 100 ++++++++++++++++++++++++ hotwords.txt | 500 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ input_audio/.gitkeep | 0 output_text/.gitkeep | 0 prompt.md | 405 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 7 ++ run.sh | 12 +++ setup.sh | 57 ++++++++++++++ transcribe.py | 542 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 10 files changed, 1649 insertions(+) @@ -0,0 +1,26 @@ +# ---- 大体积 / 自动生成，不入版本库 ---- +# Python 虚拟环境（约 1.8GB，setup.sh 自动重建） +.venv/ +# 模型缓存（约 3GB，首次运行自动下载） +models/ +# Python 字节码 +__pycache__/ +*.py[cod] + +# ---- 个人数据：录音与转写结果 ---- +# 分享仓库时不应包含真实课堂录音及其转写文本（隐私 + 体积） +# 目录本身通过 .gitkeep 保留，需要单独共享时自行拷贝 +input_audio/* +!input_audio/.gitkeep +output_text/* +!output_text/.gitkeep + +# ---- 内部工作目录 ---- +.zcode/ +.qoder/ +work/ +.venv.bak/ + +# ---- 编辑器 / 系统 ---- +.DS_Store +*.swp @@ -0,0 +1,100 @@ +# 本地课堂录音批量转文字（混合引擎） + +把课堂录音（iOS 语音备忘录的 m4a 等）在**本地**批量转成 Markdown 纯文本，事后自己整理成笔记和考点。 +不联网调用任何云端 API；中文主导的录音走 **FunASR SeACo-Paraformer**（中文最准 + 原生土木术语热词）， +英语/法语等走 **faster-whisper**。按整个文件自动路由引擎，全程 CPU 推理，无需 GPU、无需 sudo。 + +## 用法（三步） + +```bash +./setup.sh # ① 首次配置：装依赖 + 预下载约 3GB 模型（联网一次，之后离线可用） +# ② 把录音文件丢进 input_audio/ +./run.sh # ③ 转写：output_text/ 里出现同名 .md 文件 +``` + +> 也可以跳过 ① 直接 `./run.sh`：检测到依赖未就绪（如刚克隆本仓库）会**自动执行 +> `setup.sh`，装完接着转写**；`requirements.txt`/`setup.sh` 有更新时同样会自动补装。 +> `models/` 被删除也没关系，转写时按需自动重新下载。 + +## CLI 参数 + +| 命令 | 作用 | +|---|---| +| `./run.sh` | 全自动：探测语言 → 路由引擎 → 转写 | +| `./run.sh --language zh` | 强制语言：`zh`→Paraformer；`en`/`fr`/…→Whisper | +| `./run.sh --engine paraformer` | 强制引擎：`paraformer` 或 `whisper`（优先级高于 `--language`） | +| `./run.sh --model large-v3` | 换更准的 Whisper 模型（默认 `large-v3-turbo`，更准但更慢） | +| `./run.sh --dry-run` | 只解码+探测+打印每个文件路由到哪个引擎（依据），不转写 | +| `./run.sh --download-only` | 只预下载/校验模型到 `models/`，可重复运行 | + +## 特性 + +- **断点续传**：`output_text/<同名>.md` 已存在且比音频新 → 自动跳过。重新录制/替换音频后自动重转。 +- **失败隔离**：单个文件损坏/转写失败不中断批次，结束打印「成功 N / 跳过 N / 失败 N」及原因。 +- **热词纠错**：Paraformer 原生词级偏置，土木术语识别更准（见下节）。 +- **长录音**：VAD 自动切分，几小时的录音无需手动分段。 +- **输出格式**：`# 文件名` + 正文（无时间戳），UTF-8 纯文本 Markdown，可直接整理。 + +## 添加课程专属热词 + +编辑 `hotwords.txt`，**一行一个词**，保存即生效（下次运行自动加载）。 +`#` 开头的行为注释/分组标题，会被自动忽略，可按课程分区管理。 +Paraformer 用全部热词做词级偏置；Whisper 的 initial_prompt 取前 50 个。 +课程里新出现的术语（老师名字、教材名词、规范编号等）都值得加进去。 +现有分组：基础土木词 + 桥梁工程 / 结构鉴定和加固 / 基础工程 / 土木防灾减灾 + 通用易错词。 +若发现某个热词「抢词」（如把别的词错误识别成它），直接删掉该行再重转即可。 + +## 想更准 / 想强制引擎 + +- 更准（慢）：`./run.sh --model large-v3` +- 整段中文但夹杂英文术语的录音，若被误路由到 Whisper：`./run.sh --language zh` +- 反之，英语课里想走 Paraformer 是不支持的（Paraformer 只适合中文主导）； + 英语/法语主导请用默认路由或 `--engine whisper`。 + +## 网络问题备选 + +HuggingFace 下载慢/失败时，先设置国内镜像再重跑： + +```bash +export HF_ENDPOINT=https://hf-mirror.com +./setup.sh +``` + +FunASR（Paraformer/VAD/标点）走 ModelScope（国内直连，一般无需镜像）。 + +## 将来有 NVIDIA GPU + +```bash +uv pip install torch torchaudio # 换回默认源（CUDA 版） +A2T_DEVICE=cuda ./run.sh # 代码保留了 device 参数位，环境变量即可切换 +``` + +（同时把 `setup.sh` 里的 `--index-url https://download.pytorch.org/whl/cpu` 去掉重装 torch。） + +## 分享 / 克隆 + +仓库**不含**模型（约 3GB）、虚拟环境（约 1.8GB）以及录音/转写数据（隐私，见 `.gitignore`）， +克隆后直接 `./run.sh` 即可：自动安装依赖、下载模型，然后转写。 + +## 目录结构 + +``` +audio_to_text/ +├── setup.sh # 首次一键配置（幂等；--check 供 run.sh 校验就绪状态） +├── run.sh # 日常入口（依赖未就绪时自动先执行 setup.sh） +├── transcribe.py # 主程序 +├── hotwords.txt # 土木工程热词表（一行一个，即改即生效） +├── requirements.txt +├── input_audio/ # ← 录音放这里（内容不入库） +├── output_text/ # ← 转写结果在这里（同名 .md，内容不入库） +├── models/ # 模型缓存（不入库；可整体删除，运行时自动重新下载） +└── .venv/ # 项目内 Python 3.12 环境（不入库；不动系统 Python） +``` + +## 常见问题 + +- **内存**：两套引擎严格分时加载（先 Whisper 组 → 释放 → Paraformer 组），15GB 内存机器可放心用。 +- **m4a 解码失败**：优先 PyAV 直接解码；异常时自动回退 `ffmpeg` 转码，仍失败则该文件记为失败并继续其余文件。 +- **重复转写**：只想强制重转某个文件，删掉 `output_text/` 里对应的 `.md` 再运行即可（或 `touch` 音频文件）。 +- **转写质量评估**：本次验证用的是 piper TTS 合成音；真实课堂录音的准确率请以实际录音为准， + 遇到专业词识别不准时优先往 `hotwords.txt` 加词。 @@ -0,0 +1,500 @@ +混凝土 +钢筋混凝土 +预应力混凝土 +水泥 +砂浆 +骨料 +外加剂 +水灰比 +坍落度 +和易性 +强度等级 +立方体抗压强度 +养护 +收缩 +徐变 +碳化 +钢筋锈蚀 +保护层厚度 +箍筋 +纵筋 +弯起钢筋 +架立筋 +分布筋 +锚固长度 +搭接长度 +受弯承载力 +斜截面 +最小配筋率 +张拉 +锚具 +孔道灌浆 +先张法 +后张法 +无粘结预应力 +弯矩 +剪力 +轴力 +扭矩 +挠度 +应力 +应变 +弹性模量 +泊松比 +惯性矩 +剪切模量 +屈曲 +疲劳 +蠕变 +超静定 +桁架 +支座反力 +力法 +位移法 +型钢 +焊缝 +角焊缝 +高强螺栓 +铆钉 +节点板 +加劲肋 +防火涂料 +土力学 +孔隙比 +密实度 +含水率 +液限 +塑限 +塑性指数 +压缩模量 +固结 +渗透系数 +抗剪强度 +黏聚力 +内摩擦角 +土压力 +主动土压力 +被动土压力 +朗肯 +库仑 +地基承载力 +地基处理 +换填 +夯实 +桩基 +预制桩 +灌注桩 +摩擦桩 +端承桩 +承台 +沉降 +不均匀沉降 +基坑 +支护 +地下连续墙 +降水 +边坡稳定 +挡土墙 +模板工程 +钢筋工程 +混凝土浇筑 +振捣 +施工缝 +后浇带 +脚手架 +塔吊 +起重机 +隐蔽工程 +监理 +网络计划 +关键线路 +工程量清单 +概算 +预算 +招投标 +桥梁 +桥墩 +桥台 +主梁 +支座 +伸缩缝 +桥面铺装 +涵洞 +隧道 +衬砌 +围岩 +新奥法 +锚杆 +喷射混凝土 +超前支护 +路基 +路面 +基层 +面层 +沥青混凝土 +水泥混凝土路面 +级配碎石 +压实度 +弯沉 +边沟 +工程测量 +水准测量 +水准仪 +全站仪 +经纬仪 +导线测量 +高程 +放样 +控制点 +沉降观测 + +# ==================================================================== +# 四门在修专业课针对性热词（2026-09-18 添加，# 开头的行为注释，不参与热词） +# 依据：已转写课堂录音中实际出现的识别错误 + 各课程标准术语。 +# ==================================================================== + +# ---------- 通用易错词（四门课高频且实际转写中反复出错） ---------- +地基 +基础 +上部结构 +下部结构 +剪切 +失稳 +验算 +构造要求 +承载力 +裂缝 +构件 +黏性土 +粉土 +砂土 +粒径 +均布荷载 +局部荷载 +集中荷载 +计算简图 +参数化设计 +比萨斜塔 + +# ---------- 桥梁工程 ---------- +桥梁工程 +简支梁 +简支梁桥 +悬臂梁 +连续梁桥 +梁桥 +桥跨结构 +桥面板 +行车道板 +桥面系 +桥面连续 +现浇 +装配式 +单向板 +双向板 +悬臂板 +铰接 +铰接悬臂板 +铰缝 +企口 +钢板连接 +湿接缝 +空心板 +箱梁 +横隔梁 +盖梁 +墩台 +搭板 +锥坡 +护坡 +跨径 +计算跨径 +跨中 +有效分布宽度 +有效工作宽度 +分布宽度 +着地宽度 +车轮荷载 +车辆荷载 +车道荷载 +人群荷载 +轴重 +荷载组合 +恒载 +活载 +冲击系数 +影响线 +横向分布系数 +抗扭刚度 +扭转 +弹性支承 +最不利 +配筋 +配筋率 +裂缝宽度 +承载能力极限状态 +正常使用极限状态 +预拱度 +铺装层 +公路桥梁设计通用规范 + +# ---------- 结构鉴定和加固 ---------- +结构鉴定 +加固 +加固改造 +可靠性鉴定 +抗震鉴定 +危房 +回弹法 +回弹仪 +钻芯法 +超声回弹综合法 +碳纤维 +碳纤维布 +粘钢 +植筋 +加大截面 +灌浆 +注浆 +纠偏 +纠倾 +掏土 +迫降 +顶升 +托换 +基础托换 +锚杆静压桩 +树根桩 +残余变形 +倾斜观测 +露筋 +蜂窝麻面 +空鼓 +剥落 +锈胀 +抗倾覆 +倾覆 +整体剪切破坏 +局部剪切破坏 +冲剪破坏 +冲切破坏 +塑性区 +滑动面 +隆起 +松砂 +摩尔库仑 +三七灰土 +灰土 +生石灰 +熟石灰 +麻刀 +湿陷性 +湿陷性黄土 +膨胀土 +自由膨胀率 +季节性冻土 +冻胀 +冻结深度 +切向冻胀力 +淤泥质土 + +# ---------- 基础工程 ---------- +基础工程 +基础埋深 +埋置深度 +持力层 +下卧层 +软弱下卧层 +地基模型 +文克勒 +弹性半空间 +基底反力 +基底压力 +附加应力 +自重应力 +应力扩散角 +分层总和法 +角点法 +压缩系数 +固结度 +有效应力 +孔隙水压力 +静止土压力 +超固结 +先期固结压力 +太沙基 +沉降差 +差异沉降 +沉降速率 +倾斜率 +浅基础 +深基础 +筏板基础 +筏形基础 +箱形基础 +独立基础 +条形基础 +扩展基础 +刚性基础 +柔性基础 +沉井 +倒梁法 +地基承载力特征值 +特征值 +临塑荷载 +极限荷载 +比例界限 +载荷试验 +静力触探 +标准贯入试验 +勘察 +验槽 +桩基础 +单桩承载力 +桩侧摩阻力 +桩端阻力 +负摩阻力 +群桩 +预应力管桩 +管桩 +钢管桩 +钻孔灌注桩 +人工挖孔桩 +沉管灌注桩 +静压桩 +沉桩 +试桩 +静载荷试验 +桩身完整性 +断桩 +缩颈 +钢筋笼 +泥浆护壁 +水下混凝土 +嵌岩桩 +基岩 +复合地基 +褥垫层 +桩土应力比 +置换率 +水泥粉煤灰碎石桩 +水泥土搅拌桩 +深层搅拌法 +旋喷桩 +砂石桩 +振冲法 +强夯法 +垫层 +堆载预压 +真空预压 +砂井 +塑料排水板 +土工格栅 +土工布 +加筋土 +基坑支护 +支护结构 +围护结构 +地下水位 +潜水 +承压水 +井点降水 +止水帷幕 +土钉墙 +排桩 +冠梁 +内支撑 +钢支撑 +钢板桩 +双排桩 +逆作法 +放坡 +抗滑桩 +抗隆起 +泄水孔 +排水沟 +截水沟 +盲沟 +反滤层 +流砂 +管涌 +坝基 + +# ---------- 土木防灾减灾 ---------- +防灾减灾 +抗震设防 +设防烈度 +烈度 +震级 +震中 +地震作用 +地震动 +场地类别 +特征周期 +反应谱 +阻尼 +阻尼比 +自振周期 +振型 +时程分析 +底部剪力法 +多遇地震 +罕遇地震 +延性 +塑性铰 +强柱弱梁 +强剪弱弯 +强节点弱构件 +轴压比 +层间位移角 +薄弱层 +刚度 +侧向刚度 +平面不规则 +竖向不规则 +隔震 +消能减震 +阻尼器 +黏滞阻尼器 +屈曲约束支撑 +铅芯橡胶支座 +防震缝 +非结构构件 +生命线工程 +震害 +抗震加固 +砂土液化 +液化判别 +震陷 +地面塌陷 +采空区 +岩溶 +土洞 +泥石流 +崩塌 +滑坡 +风灾 +台风 +基本风压 +风振 +脉动风 +顺风向 +横风向 +涡激振动 +抖振 +颤振 +驰振 +风洞试验 +大跨空间结构 +高耸结构 +耐火极限 +防火分区 +结构抗火 +连续倒塌 +韧性 +易损性 +风险评估 +唐山地震 +汶川地震 +建筑抗震设计规范 +建筑地基基础设计规范 @@ -0,0 +1,405 @@ +# 施工任务：本地课堂录音批量转文字（混合引擎） + +> **本文件是一份自包含的施工说明书**，写给在全新对话中执行它的编码 agent。请完整读完再动手。 +> 文中「环境事实」一节已在 2026-09-09 实际探测验证过，**无需重复探测**。 +> 架构方向已由用户拍板，**不要更改整体方向**；某一步失败时按第 7 节的兜底方案处理。 +> 本任务涉及：安装 uv（用户级）、在项目内建 venv、下载约 3GB 模型、长时间本地转写。全程不需要 sudo，不做任何系统级安装，不使用云端 API。 + +--- + +## 0. 项目目标 + +用户是土木工程专业的学生/从业者，需要把**每天大量的课堂录音**（主要来自 iOS 内置「语音备忘录」的 m4a）在本地转成文字，事后人工整理成笔记和考点。 + +- **质量要求**：整体语义正确即可，用户容忍个别词的小瑕疵；但希望**土木工程专业术语**尽量识别准确。 +- **语言构成**：中英文混合，偶尔出现法语。中文占主体的录音用 Paraformer（中文最准 + 原生热词机制），英语/法语占主体的用 Whisper（多语言强项）。 +- **最终使用形态（务必保证）**：把音频文件丢进 `input_audio/` → 运行 `./run.sh` → `output_text/` 里出现**同名**的 `.md` 文件（纯文本 Markdown，无时间戳）。 +- **零系统安装**：所有依赖装在项目文件夹内（`.venv/`），模型缓存在项目内（`models/`），不动系统 Python、不装全局包。 + +## 1. 环境事实（已探测，直接采信） + +| 项目 | 值 | +|---|---| +| 项目目录 | `/media/data/usrdir/agent/zcode/audio_to_text`（已有本文件，其余为空） | +| 系统 | Arch Linux，zsh，内核 7.x | +| Python | 仅 `/usr/bin/python3` = **3.14.7**，**无 pip**、无 uv/conda/pyenv | +| GPU | **无**（无 NVIDIA 驱动/设备），CPU：i7-13700H，20 线程 | +| 内存 | 15GB（可用约 7.3GB）——两套引擎不要同时常驻，按组分时加载 | +| 磁盘 | `/media/data` 剩余 535GB，充足 | +| ffmpeg | 有，`/usr/sbin/ffmpeg`，版本 9.0.1（兜底转码可用，但主路径不需要） | +| 网络 | modelscope.cn / huggingface.co / pypi.org 均可达 | + +**推论**：系统 Python 3.14 太新且无 pip，直接用会有 wheel 兼容风险 → 用 **uv** 在项目内建 **Python 3.12** 的 venv（uv 会自动下载托管版 CPython 3.12，无需系统权限）。 + +## 2. 架构设计（定案） + +混合引擎，**按文件路由**（不做句级混切，保持语义连贯和实现简单）： + +``` +扫描 input_audio/ 全部音频 + └─ 对每个文件： + 1. 解码 → 16kHz float32 单声道 numpy（faster_whisper.audio.decode_audio，PyAV 自带解码，m4a/AAC 直接支持） + 2. 语言探测：取开头 45s + 时长中点 45s 两个切片（文件短于 90s 则只用全片）， + 用 faster-whisper transcribe(language=None) 读 info.language / info.language_probability，两切片概率取平均 + 3. 路由： + - 用户强制 --engine → 用指定引擎 + - 用户强制 --language zh → Paraformer；--language en/fr/… → Whisper（language 传入） + - 否则自动：zh 概率 > 0.6 且为最高 → **Paraformer**；否则 → **Whisper**（language 用探测结果） + └─ 按引擎分组： + - 先处理 **Whisper 组**（探测时 turbo 模型已加载，直接复用） + - 处理完 `del model; gc.collect()` 释放，再加载 **Paraformer 组** + └─ 每个文件写 output_text/<原文件名(去扩展名)>.md，UTF-8 +``` + +### 引擎一：FunASR SeACo-Paraformer（中文主导） + +```python +from funasr import AutoModel +model = AutoModel( + model="iic/speech_seaco_paraformer_large_asr_nat-zh-cn-16k-common-vocab8404-pytorch", + vad_model="iic/speech_fsmn_vad_zh-cn-16k-common-pytorch", + vad_kwargs={"max_single_segment_time": 30000}, + punc_model="iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch", + device="cpu", + cache_dir="models/funasr", # 模型缓存在项目内 + disable_update=True, # 关掉启动时联网检查 +) +res = model.generate( + input=audio_np, # 支持 numpy 输入；若该版本不接受 ndarray，见第 7 节兜底 + fs=16000, + batch_size_s=300, # 长音频按 300s 一批 + hotword=" ".join(hotwords), # 热词：读 hotwords.txt，空格拼接传入（SeACo 原生热词机制） +) +text = res[0]["text"] +``` + +- VAD 会自动切分几小时的录音；标点模型自动加标点。 +- **热词是 Paraformer 相对 Whisper 的核心优势**（真正的词级偏置，土木术语纠正效果比 Whisper 的 initial_prompt 强）。 + +### 引擎二：faster-whisper（英/法/其他） + +```python +from faster_whisper import WhisperModel, BatchedInferencePipeline +model = WhisperModel( + "large-v3-turbo", # 默认；--model 可覆盖为 large-v3（更准更慢）等 + device="cpu", compute_type="int8", + cpu_threads=16, # 20 线程机器留一点给系统 + download_root="models/faster-whisper", +) +pipe = BatchedInferencePipeline(model=model) # 批处理提速 +segments, info = pipe.transcribe( + audio_np, + language=None, # 或探测/强制的语言码 (zh/en/fr) + vad_filter=True, + vad_parameters={"min_silence_duration_ms": 500}, + beam_size=5, + initial_prompt=whisper_prompt, # 见下；若 Batched 版不支持该参数，见第 7 节兜底 + condition_on_previous_text=True, +) +text = "".join(s.text for s in segments) +``` + +- `whisper_prompt`：`"这是一堂土木工程专业课，可能涉及以下术语："` + 热词表前 50 个词顿号连接（控制在约 200 token 内，超长会被截断）。仅中文相关时价值大，对 en/fr 无害。 +- 语言探测也用这个模型实例（transcribe 前 45s 切片即可拿到 info），**探测和转写复用同一实例**。 + +### 输出格式 + +`output_text/<stem>.md`，UTF-8： + +```markdown +# <原文件名（不含扩展名）> + +<转写正文> +``` + +- Whisper：把连续约 6 个 segment 合并为一个自然段，段间空行。 +- Paraformer：`res[0]["text"]` 本身就是带标点的整段文本，直接写入（可按句号每 5~8 句插一个空行分段）。 + +### CLI 规格 + +```bash +./run.sh # 全自动：探测语言 → 路由引擎 +./run.sh --language zh # 强制语言（zh→Paraformer；en/fr→Whisper） +./run.sh --engine whisper # 强制引擎：paraformer | whisper +./run.sh --model large-v3 # 覆盖 Whisper 模型（默认 large-v3-turbo） +./run.sh --dry-run # 只打印每个文件将路由到哪个引擎，不转写（调路由用） +``` + +### 断点续传（重要，日常增量使用依赖它） + +若 `output_text/<stem>.md` 已存在且其 mtime **晚于**音频文件 mtime → 跳过。重新录制/替换音频后会自动重转。单文件转写失败不中断批次，结束打印汇总（成功 N / 跳过 N / 失败 N 及失败原因）。 + +## 3. 需要创建的文件（全部在项目根目录） + +``` +audio_to_text/ +├── setup.sh # 首次一键配置（幂等，可重复运行） +├── run.sh # 日常入口 +├── transcribe.py # 主程序（--download-only / --dry-run / 转写） +├── hotwords.txt # 土木工程热词表（内容见第 4 节，直接照抄） +├── requirements.txt +├── README.md +├── .gitignore # .venv/ models/ work/ __pycache__/ （可选但建议） +├── input_audio/ # setup.sh 创建 +├── output_text/ # setup.sh 创建 +└── models/ # 模型缓存（funasr/ 与 faster-whisper/ 两个子目录） +``` + +### setup.sh 要点 + +```bash +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "$0")" +# 1) uv：没有则装（用户级，装到 ~/.local/bin，无需 sudo） +command -v uv >/dev/null || curl -LsSf https://astral.sh/uv/install.sh | sh +export PATH="$HOME/.local/bin:$PATH" +# 2) 项目内 venv（Python 3.12；uv 自动下载托管版 CPython） +[ -d .venv ] || uv venv --python 3.12 .venv +source .venv/bin/activate +# 3) 先装 CPU 版 torch（必须用 CPU 源，否则会拉 2GB+ 的 CUDA 包；本机无 GPU） +uv pip install torch torchaudio --index-url https://download.pytorch.org/whl/cpu +# 4) 其余依赖 +uv pip install -r requirements.txt +# 5) 建目录 + 预下载全部模型（联网一次，之后可离线） +mkdir -p input_audio output_text models +python transcribe.py --download-only +echo "配置完成。以后：把音频放进 input_audio/ 然后运行 ./run.sh" +``` + +### requirements.txt + +``` +numpy==1.26.4 +faster-whisper>=1.0.0 +funasr>=1.1.0 +modelscope>=1.15 +``` + +> numpy 锁 1.26.4：funasr 及其依赖对 numpy 2.x 兼容性不稳。torch/torchaudio 由 setup.sh 单独装（CPU 源），不写进 requirements。 + +### run.sh + +```bash +#!/usr/bin/env bash +set -euo pipefail +cd "$(dirname "$0")" +source .venv/bin/activate +exec python transcribe.py "$@" +``` + +### transcribe.py 实现要点 + +- 结构：`parse_args` → `collect_audio_files("input_audio")` → `resume 过滤` → `decode + probe + route` → `按组转写` → `写输出` → `汇总打印`。 +- 支持的输入扩展名：`.m4a .mp3 .wav .flac .aac .ogg .mp4 .wma .amr`（m4a 是重点，iOS 语音备忘录 = AAC in m4a 容器，务必用真实 m4a 测试）。 +- `--download-only`：依次实例化 Whisper 模型（触发 HF 下载到 `models/faster-whisper`）和 FunASR AutoModel（触发 ModelScope 下载到 `models/funasr`），加载成功即释放退出，打印各模型缓存路径。 +- `--dry-run`：只做解码+探测+路由，打印每条 `文件名 → 引擎(依据)`，不加载 FunASR、不转写。 +- 解码兜底：`faster_whisper.audio.decode_audio(path, sampling_rate=16000, mono=True)` 为主；异常时退到 `ffmpeg -y -i <in> -ac 1 -ar 16000 <tmp>.wav` 再读 wav（`wave`/`soundfile` 均可）。 +- 日志：每个文件开始/结束/耗时/路由结果打印一行；全程中文输出。 +- 文件名可能含空格/中文：subprocess 一律用参数列表形式（`shell=False`），路径用 pathlib。 + +### README.md 要点 + +- 三行用法（setup → 丢音频 → run.sh）、各 CLI 参数说明 +- 如何往 `hotwords.txt` 加课程专属词（一行一个，改完即生效） +- 想更准：`./run.sh --model large-v3`；想强制引擎/语言：`--engine` / `--language` +- HuggingFace 下载慢/失败的备选：`export HF_ENDPOINT=https://hf-mirror.com` 后重跑 setup +- 以后若加了 NVIDIA GPU：`setup.sh` 中 torch 换回默认源并把 `device` 改 `cuda` 即可（代码里保留 device 参数位） + +## 4. hotwords.txt 内容（直接照抄，一行一个） + +```text +混凝土 +钢筋混凝土 +预应力混凝土 +水泥 +砂浆 +骨料 +外加剂 +水灰比 +坍落度 +和易性 +强度等级 +立方体抗压强度 +养护 +收缩 +徐变 +碳化 +钢筋锈蚀 +保护层厚度 +箍筋 +纵筋 +弯起钢筋 +架立筋 +分布筋 +锚固长度 +搭接长度 +受弯承载力 +斜截面 +最小配筋率 +张拉 +锚具 +孔道灌浆 +先张法 +后张法 +无粘结预应力 +弯矩 +剪力 +轴力 +扭矩 +挠度 +应力 +应变 +弹性模量 +泊松比 +惯性矩 +剪切模量 +屈曲 +疲劳 +蠕变 +超静定 +桁架 +支座反力 +力法 +位移法 +型钢 +焊缝 +角焊缝 +高强螺栓 +铆钉 +节点板 +加劲肋 +防火涂料 +土力学 +孔隙比 +密实度 +含水率 +液限 +塑限 +塑性指数 +压缩模量 +固结 +渗透系数 +抗剪强度 +黏聚力 +内摩擦角 +土压力 +主动土压力 +被动土压力 +朗肯 +库仑 +地基承载力 +地基处理 +换填 +夯实 +桩基 +预制桩 +灌注桩 +摩擦桩 +端承桩 +承台 +沉降 +不均匀沉降 +基坑 +支护 +地下连续墙 +降水 +边坡稳定 +挡土墙 +模板工程 +钢筋工程 +混凝土浇筑 +振捣 +施工缝 +后浇带 +脚手架 +塔吊 +起重机 +隐蔽工程 +监理 +网络计划 +关键线路 +工程量清单 +概算 +预算 +招投标 +桥梁 +桥墩 +桥台 +主梁 +支座 +伸缩缝 +桥面铺装 +涵洞 +隧道 +衬砌 +围岩 +新奥法 +锚杆 +喷射混凝土 +超前支护 +路基 +路面 +基层 +面层 +沥青混凝土 +水泥混凝土路面 +级配碎石 +压实度 +弯沉 +边沟 +工程测量 +水准测量 +水准仪 +全站仪 +经纬仪 +导线测量 +高程 +放样 +控制点 +沉降观测 +``` + +## 5. 实施步骤 + +1. 写 `requirements.txt` / `hotwords.txt` / `setup.sh` / `run.sh` / `.gitignore` → `chmod +x` 两个 sh +2. 跑 `./setup.sh`（下载约 3GB：whisper-turbo ≈1.6GB 来自 HF；paraformer+vad+punc ≈1.2GB 来自 ModelScope，国内很快） +3. 写 `transcribe.py`（含 `--download-only`、`--dry-run`） +4. 验证（见第 6 节） +5. 写 `README.md` +6. 向用户报告使用方法 + +## 6. 验证与验收标准 + +**路由与流程测试**（不依赖真实课堂录音）： +1. `./run.sh --dry-run` 对空目录输出 0 个文件不报错。 +2. 造测试音频：若有 `espeak-ng`/`espeak`，合成三句话（中文土木句如"预应力混凝土梁的箍筋间距应满足计算要求"、一句英语、一句法语）→ 合并为一个 wav → `ffmpeg -c:a aac` 转成 **m4a** 放进 `input_audio/`。若没有 espeak，就合成一段 30s 音调 m4a（此时只验证机械流程，不验证识别质量），并明确告知用户需要真实录音来评估质量。 +3. `./run.sh --dry-run`：确认探测/路由输出合理（多语言混合音频路由到哪个引擎都要有依据打印）。 +4. `./run.sh --engine paraformer --language zh` 等强制组合各跑一次，确认参数生效。 +5. 完整跑一次：确认 `output_text/<同名>.md` 生成、UTF-8、内容非空；espeak 合成音需能识别出接近原文的句子。 +6. **断点续传**：立即重跑，确认全部跳过；`touch input_audio/<文件>` 改 mtime 后重跑，确认重转。 +7. 单文件故意损坏（截断的 m4a）：确认报错被捕获、其余文件继续、汇总正确。 + +**验收清单**：以上全过 + `--download-only` 幂等 + 全程未用 sudo/未装系统包 + 项目外无残留（uv 本体在 `~/.local/bin` 属用户级，允许）。 + +## 7. 已知坑与兜底 + +| 坑 | 兜底 | +|---|---| +| FunASR 某版本不接受 ndarray 输入 | 改为传临时 wav 路径给 `model.generate(input=wav_path)` | +| `BatchedInferencePipeline.transcribe` 不认 `initial_prompt` | 对 Whisper 组退回普通 `model.transcribe(..., vad_filter=True)`（慢一点但功能全） | +| HF 下载慢/断 | `export HF_ENDPOINT=https://hf-mirror.com`；ModelScope 一般无需镜像 | +| 内存吃紧 | 两引擎严格分时加载（Whisper 组 → 释放 → Paraformer 组）；`batch_size` 降到 4 | +| torch 装成 CUDA 巨包 | 必须用 `--index-url https://download.pytorch.org/whl/cpu` | +| funasr 启动联网检查卡住 | `disable_update=True` | +| numpy 2.x 与 funasr 不兼容 | 锁 `numpy==1.26.4` | +| uv 安装脚本被网络策略拦截 | 从 `https://github.com/astral-sh/uv/releases` 手动下载 x86_64-unknown-linux-gnu tar 包解压到 `~/.local/bin` | +| Whisper 对中文输出繁体 | `initial_prompt` 开头加"以下是普通话的句子。"（已知有效做法）；或结果做 `opencc` 简化（可选，不强制） | +| 长录音（2h+） | VAD 自动切分，无需额外逻辑；确保 batch_size_s=300 生效 | + +## 8. 边界（本次不做） + +- 不做说话人分离（课堂以老师单声道为主） +- 不做时间戳输出（用户明确不要） +- 不做云端 API、不做 GPU 支持（代码留 device 参数位即可） +- 不初始化 git、不提交 @@ -0,0 +1,7 @@ +# numpy 锁 1.26.4：funasr 及其依赖对 numpy 2.x 兼容性不稳 +numpy==1.26.4 +faster-whisper>=1.0.0 +funasr>=1.1.0 +modelscope>=1.15 +# uv 创建的 venv 默认不带 setuptools；funasr 生态部分依赖运行时仍 import pkg_resources +setuptools @@ -0,0 +1,12 @@ +#!/usr/bin/env bash +# 日常入口：把音频放进 input_audio/ 后运行 ./run.sh [参数] +# 依赖/模型未就绪时（如刚克隆本仓库），自动先执行 ./setup.sh 完成安装与下载，再继续转写。 +set -euo pipefail +cd "$(dirname "$0")" + +if ! bash setup.sh --check; then + echo "==> 依赖未就绪（或配置已更新），自动执行 ./setup.sh（首次需联网，约 3GB）…" + bash setup.sh +fi +source .venv/bin/activate +exec python transcribe.py "$@" @@ -0,0 +1,57 @@ +#!/usr/bin/env bash +# 首次一键配置（幂等，可重复运行）： +# 装 uv（用户级）→ 项目内 venv（Python 3.12）→ CPU 版 torch → 其余依赖 → 预下载模型 +# 全程无需 sudo，不动系统 Python。 +# +# 用法： +# ./setup.sh 完整配置（依赖 + 约 3GB 模型） +# ./setup.sh --check 只校验是否已配置（供 run.sh 自动引导），就绪退出码 0，否则 1 +set -euo pipefail +cd "$(dirname "$0")" + +MARKER=".venv/.a2t-ready" + +# 配置指纹：依赖清单或本安装脚本内容变化 → 指纹不匹配 → 视为未配置，自动重装 +stamp() { sha256sum requirements.txt setup.sh 2>/dev/null | sha256sum | cut -d' ' -f1; } + +if [ "${1:-}" = "--check" ]; then + if [ -x .venv/bin/python ] && [ -f "$MARKER" ] && [ "$(cat "$MARKER")" = "$(stamp)" ]; then + exit 0 + fi + exit 1 +fi + +# 1) uv：没有则装（用户级，装到 ~/.local/bin，无需 sudo） +if ! command -v uv >/dev/null 2>&1; then + echo "==> 安装 uv（用户级，~/.local/bin）…" + curl -LsSf https://astral.sh/uv/install.sh | sh +fi +export PATH="$HOME/.local/bin:$PATH" +if ! command -v uv >/dev/null 2>&1; then + echo "uv 安装失败。可从 https://github.com/astral-sh/uv/releases 手动下载" >&2 + echo "x86_64-unknown-linux-gnu tar 包，解压出 uv 放入 ~/.local/bin 后重跑本脚本。" >&2 + exit 1 +fi + +# 2) 项目内 venv（Python 3.12；uv 自动下载托管版 CPython，不动系统 Python） +if [ ! -x .venv/bin/python ]; then + echo "==> 创建项目内 venv（Python 3.12）…" + uv venv --python 3.12 .venv +fi +source .venv/bin/activate + +# 3) 先装 CPU 版 torch（必须用 CPU 源，否则会拉 2GB+ 的 CUDA 包；本机无 GPU） +echo "==> 安装 CPU 版 torch / torchaudio（download.pytorch.org 的 cpu 源）…" +uv pip install torch torchaudio --index-url https://download.pytorch.org/whl/cpu + +# 4) 其余依赖 +echo "==> 安装其余依赖（faster-whisper / funasr / modelscope …）…" +uv pip install -r requirements.txt + +# 5) 建目录 + 预下载全部模型（联网一次，之后可离线转写） +mkdir -p input_audio output_text models +python transcribe.py --download-only + +printf '%s\n' "$(stamp)" > "$MARKER" +echo +echo "配置完成。以后：把音频放进 input_audio/ 然后运行 ./run.sh" @@ -0,0 +1,542 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""本地课堂录音批量转文字（混合引擎：FunASR SeACo-Paraformer + faster-whisper）。 + +中文主导的录音路由到 FunASR SeACo-Paraformer（原生热词机制）， +英语/法语等其他语言路由到 faster-whisper。按文件整体路由，不做句级混切。 +所有推理均在本地完成，不使用云端 API。 + +用法： + ./run.sh # 全自动：探测语言 → 路由引擎 → 转写 + ./run.sh --language zh # 强制语言（zh→Paraformer；en/fr/…→Whisper） + ./run.sh --engine whisper # 强制引擎：paraformer | whisper + ./run.sh --model large-v3 # 覆盖 Whisper 模型（默认 large-v3-turbo） + ./run.sh --dry-run # 只解码+探测+打印路由结果，不转写 + ./run.sh --download-only # 只预下载全部模型到 models/ 后退出 +""" + +from __future__ import annotations + +import argparse +import gc +import inspect +import os +import re +import shutil +import subprocess +import sys +import tempfile +import time +import wave +from pathlib import Path + +import numpy as np + +# ---------------------------------------------------------------- 路径与常量 + +PROJECT_DIR = Path(__file__).resolve().parent +INPUT_DIR = PROJECT_DIR / "input_audio" +OUTPUT_DIR = PROJECT_DIR / "output_text" +MODELS_DIR = PROJECT_DIR / "models" +FUNASR_CACHE_DIR = MODELS_DIR / "funasr" +WHISPER_CACHE_DIR = MODELS_DIR / "faster-whisper" + +# funasr(≥1.2) 不再读取 AutoModel 的 cache_dir 参数，统一走 modelscope 缓存， +# 只认 MODELSCOPE_CACHE 环境变量 —— 必须在 import funasr 之前设置，保证模型缓存在项目内 +os.environ["MODELSCOPE_CACHE"] = str(FUNASR_CACHE_DIR) + +SUPPORTED_EXTS = {".m4a", ".mp3", ".wav", ".flac", ".aac", ".ogg", ".mp4", ".wma", ".amr"} + +PARAFORMER_MODEL = "iic/speech_seaco_paraformer_large_asr_nat-zh-cn-16k-common-vocab8404-pytorch" +VAD_MODEL = "iic/speech_fsmn_vad_zh-cn-16k-common-pytorch" +PUNC_MODEL = "iic/punc_ct-transformer_zh-cn-common-vocab272727-pytorch" + +WHISPER_MODEL_DEFAULT = "large-v3-turbo" +WHISPER_CPU_THREADS = 16 # 20 线程机器留一点给系统 +WHISPER_BATCH_SIZE = 8 # 15GB 内存的保守批大小 +WHISPER_PROMPT_HOTWORDS = 50 # initial_prompt 携带的热词个数（超长会被截断） + +SAMPLE_RATE = 16000 +PROBE_SLICE_SECONDS = 45.0 # 语言探测切片长度 +ZH_PROB_THRESHOLD = 0.6 # zh 平均概率超过该值且为最高才走 Paraformer +SEGMENTS_PER_PARAGRAPH = 6 # Whisper 每 ~6 个 segment 合并成一个自然段 +SENTENCES_PER_PARAGRAPH = 6 # Paraformer 每 ~6 句插一个空行 + +# 将来有 NVIDIA GPU：export A2T_DEVICE=cuda（代码保留 device 参数位） +DEVICE = os.environ.get("A2T_DEVICE", "cpu") +COMPUTE_TYPE = "int8" if DEVICE == "cpu" else "float16" + +WS_RE = re.compile(r"[ \t\r\n\f\v]+") +SENT_RE = re.compile(r"[^。！？；]*[。！？；]+|[^。！？；]+$") + + +def log(msg: str) -> None: + print(msg, flush=True) + + +def short_err(exc: BaseException, limit: int = 150) -> str: + return re.sub(r"\s+", " ", str(exc))[:limit] or type(exc).__name__ + + +# ---------------------------------------------------------------- CLI + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser( + prog="run.sh", + description="本地课堂录音批量转文字（中文主导→Paraformer，英/法等→Whisper）", + ) + p.add_argument("--language", metavar="CODE", default=None, + help="强制语言：zh→Paraformer；en/fr/…→Whisper（语言码传给 Whisper）") + p.add_argument("--engine", choices=("paraformer", "whisper"), default=None, + help="强制指定转写引擎") + p.add_argument("--model", default=WHISPER_MODEL_DEFAULT, + help=f"Whisper 模型（默认 {WHISPER_MODEL_DEFAULT}）") + p.add_argument("--dry-run", action="store_true", + help="只解码+探测+打印路由结果，不加载 FunASR、不转写") + p.add_argument("--download-only", action="store_true", + help="只预下载/校验全部模型到 models/ 后退出") + args = p.parse_args() + if args.language: + args.language = args.language.strip().lower() + args.lang_primary = args.language.split("-")[0] + else: + args.lang_primary = None + return args + + +# ---------------------------------------------------------------- 基础工具 + +def collect_audio_files() -> list[Path]: + if not INPUT_DIR.is_dir(): + return [] + return [p for p in sorted(INPUT_DIR.iterdir()) + if p.is_file() and p.suffix.lower() in SUPPORTED_EXTS] + + +def output_is_fresh(audio: Path) -> bool: + """断点续传：输出 md 存在且 mtime 晚于音频 mtime → 视为最新，跳过。""" + out = OUTPUT_DIR / (audio.stem + ".md") + return out.exists() and out.stat().st_mtime > audio.stat().st_mtime + + +def load_hotwords() -> list[str]: + p = PROJECT_DIR / "hotwords.txt" + if not p.exists(): + return [] + # 「#」开头的行为注释/分组标题，不参与热词 + return [ln.strip() for ln in p.read_text(encoding="utf-8").splitlines() + if ln.strip() and not ln.strip().startswith("#")] + + +def decode_audio(path: Path, offset: float = 0.0, duration: float | None = None) -> np.ndarray: + """解码为 16kHz float32 单声道。主路径 PyAV（faster-whisper 自带），异常回退 ffmpeg。""" + from faster_whisper.audio import decode_audio as fw_decode + try: + # 该实现签名只有 (input_file, sampling_rate, split_stereo)，默认即输出单声道 float32 + audio = fw_decode(str(path), sampling_rate=SAMPLE_RATE) + except Exception as exc: + log(f" [解码] PyAV 失败（{type(exc).__name__}: {short_err(exc)}），回退 ffmpeg 转码 …") + audio = decode_audio_ffmpeg(path) + if offset > 0 or duration is not None: + i0 = max(0, int(offset * SAMPLE_RATE)) + i1 = len(audio) if duration is None else max(i0, min(len(audio), int((offset + duration) * SAMPLE_RATE))) + audio = audio[i0:i1] + return audio + + +def decode_audio_ffmpeg(path: Path) -> np.ndarray: + ffmpeg = shutil.which("ffmpeg") + if not ffmpeg: + raise RuntimeError("PyAV 解码失败，且系统中没有 ffmpeg 可作兜底") + fd, name = tempfile.mkstemp(suffix=".wav", prefix="a2t_") + os.close(fd) + tmp = Path(name) + try: + # 文件名可能含空格/中文：一律参数列表形式，shell=False + proc = subprocess.run( + [ffmpeg, "-y", "-i", str(path), "-ac", "1", "-ar", str(SAMPLE_RATE), str(tmp)], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + ) + if proc.returncode != 0: + raise RuntimeError(f"ffmpeg 兜底转码也失败（返回码 {proc.returncode}），文件可能损坏或为空") + with wave.open(str(tmp), "rb") as w: + if w.getframerate() != SAMPLE_RATE or w.getnchannels() != 1: + raise RuntimeError("ffmpeg 输出 wav 参数异常") + data = w.readframes(w.getnframes()) + return np.frombuffer(data, dtype=np.int16).astype(np.float32) / 32768.0 + finally: + tmp.unlink(missing_ok=True) + + +def write_temp_wav(audio: np.ndarray) -> Path: + """float32 [-1,1] 单声道 16k → 临时 16bit PCM wav（FunASR ndarray 兜底用）。""" + fd, name = tempfile.mkstemp(suffix=".wav", prefix="a2t_") + os.close(fd) + pcm = np.clip(audio, -1.0, 1.0) + pcm = (pcm * 32767.0).astype(np.int16) + with wave.open(name, "wb") as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(SAMPLE_RATE) + w.writeframes(pcm.tobytes()) + return Path(name) + + +# ---------------------------------------------------------------- 模型加载 + +def dir_has_files(d: Path) -> bool: + return d.is_dir() and any(d.iterdir()) + + +def load_whisper(model_name: str): + from faster_whisper import WhisperModel + if not dir_has_files(WHISPER_CACHE_DIR): + log(f" 首次使用：本地无缓存，正在从 HuggingFace 下载 {model_name}" + f"（体积较大，仅此一次，之后离线可用）…") + log(" 下载慢/失败可设置国内镜像后重试：export HF_ENDPOINT=https://hf-mirror.com") + return WhisperModel( + model_name, + device=DEVICE, + compute_type=COMPUTE_TYPE, + cpu_threads=WHISPER_CPU_THREADS, + download_root=str(WHISPER_CACHE_DIR), + ) + + +def make_batched_pipeline(model): + from faster_whisper import BatchedInferencePipeline + return BatchedInferencePipeline(model=model) + + +def load_paraformer(): + import torch + torch.set_num_threads(WHISPER_CPU_THREADS) + if not dir_has_files(FUNASR_CACHE_DIR): + log(" 首次使用：本地无缓存，正在从 ModelScope 下载 SeACo-Paraformer/VAD/标点" + "（约 1.3GB，仅此一次，之后离线可用）…") + from funasr import AutoModel + return AutoModel( + model=PARAFORMER_MODEL, + vad_model=VAD_MODEL, + vad_kwargs={"max_single_segment_time": 30000}, + punc_model=PUNC_MODEL, + device=DEVICE, + cache_dir=str(FUNASR_CACHE_DIR), + disable_update=True, + ) + + +def release(*objs) -> None: + for o in objs: + del o + gc.collect() + + +# ---------------------------------------------------------------- 语言探测与路由 + +def probe_slices(path: Path) -> list[np.ndarray]: + """探测切片：开头 45s + 时长中点 45s；文件短于 90s 则只用全片（单次完整解码）。""" + audio = decode_audio(path) + total_s = len(audio) / SAMPLE_RATE + if total_s <= PROBE_SLICE_SECONDS * 2: + return [audio] + first_len = int(PROBE_SLICE_SECONDS * SAMPLE_RATE) + mid_start = max(0, len(audio) // 2 - first_len // 2) + return [audio[:first_len], audio[mid_start:mid_start + first_len]] + + +def probe_language(whisper_model, audio_slices: list[np.ndarray]) -> tuple[str, float]: + """对每个切片 transcribe(language=None) 读 info.language/概率，各语言概率取平均。""" + per_lang: dict[str, list[float]] = {} + for idx, sl in enumerate(audio_slices, 1): + _segments, info = whisper_model.transcribe(sl, language=None, beam_size=5) + lang = info.language or "unknown" + prob = float(info.language_probability or 0.0) + per_lang.setdefault(lang, []).append(prob) + log(f" [探测] 切片 {idx}/{len(audio_slices)}：{lang}（p={prob:.2f}）") + avg = {lang: sum(v) / len(v) for lang, v in per_lang.items()} + best = max(avg, key=lambda k: avg[k]) + return best, avg[best] + + +def forced_route(args: argparse.Namespace) -> tuple[str, str | None, str]: + """强制模式下的路由：--engine 优先于 --language。返回 (引擎, whisper语言码, 依据)。""" + if args.engine and args.language: + eng = args.engine + basis = f"强制 --engine={eng} + --language={args.language}" + elif args.engine: + eng = args.engine + basis = f"强制 --engine={eng}" + else: + eng = "paraformer" if args.lang_primary == "zh" else "whisper" + basis = f"强制 --language={args.language}" + lang = args.lang_primary if (eng == "whisper" and args.language) else None + return eng, lang, basis + + +def auto_route(detected: str, prob: float) -> tuple[str, str]: + if detected.startswith("zh") and prob > ZH_PROB_THRESHOLD: + return "paraformer", f"自动：探测到 {detected}（平均p={prob:.2f} > {ZH_PROB_THRESHOLD}）" + return "whisper", f"自动：探测到 {detected}（平均p={prob:.2f}）" + + +# ---------------------------------------------------------------- 转写 + +def build_whisper_prompt(hotwords: list[str], language: str | None) -> str | None: + if not hotwords: + return None + terms = "、".join(hotwords[:WHISPER_PROMPT_HOTWORDS]) + prompt = f"这是一堂土木工程专业课，可能涉及以下术语：{terms}" + if language and language.startswith("zh"): + prompt = "以下是普通话的句子。" + prompt # 防止中文输出成繁体 + return prompt + + +def _accepted_kwargs(func) -> set[str]: + try: + params = inspect.signature(func).parameters + except (TypeError, ValueError): + return set() + return {n for n, v in params.items() + if v.kind in (v.POSITIONAL_OR_KEYWORD, v.KEYWORD_ONLY)} + + +def whisper_transcribe(model, pipe, audio: np.ndarray, language: str | None, initial_prompt: str | None): + """优先 BatchedInferencePipeline（快）；不支持 initial_prompt 时按兜底方案退回普通 transcribe。""" + want = { + "language": language, + "beam_size": 5, + "vad_filter": True, + "vad_parameters": {"min_silence_duration_ms": 500}, + "condition_on_previous_text": True, + "initial_prompt": initial_prompt, + "batch_size": WHISPER_BATCH_SIZE, + } + candidates = [] + if pipe is not None and (not initial_prompt or "initial_prompt" in _accepted_kwargs(pipe.transcribe)): + candidates.append(pipe.transcribe) + candidates.append(model.transcribe) + last_err: TypeError | None = None + for fn in candidates: + accepted = _accepted_kwargs(fn) + kwargs = {k: v for k, v in want.items() if v is not None and k in accepted} + try: + return fn(audio, **kwargs) + except TypeError as exc: # 该实现不认某个参数 → 换下一个候选 + last_err = exc + raise RuntimeError(f"transcribe 调用失败：{last_err}") + + +def merge_whisper_segments(segments) -> str: + """连续约 6 个 segment 合并为一个自然段，段间空行。""" + paragraphs: list[str] = [] + buf: list[str] = [] + for i, seg in enumerate(segments, 1): + buf.append(seg.text) + if i % 50 == 0: + log(f" … 已转写 {i} 段") + if i % SEGMENTS_PER_PARAGRAPH == 0: + paragraphs.append(WS_RE.sub(" ", "".join(buf)).strip()) + buf = [] + if buf: + paragraphs.append(WS_RE.sub(" ", "".join(buf)).strip()) + return "\n\n".join(p for p in paragraphs if p) + + +def paraformer_transcribe(am, audio: np.ndarray, hotword_str: str) -> str: + gen_kwargs = {"fs": SAMPLE_RATE, "batch_size_s": 300} + if hotword_str: + gen_kwargs["hotword"] = hotword_str + try: + res = am.generate(input=audio, **gen_kwargs) + except Exception as exc: + log(f" [Paraformer] ndarray 输入不被接受（{type(exc).__name__}: {short_err(exc)}），" + f"按兜底方案改传临时 wav 路径重试") + wav = write_temp_wav(audio) + try: + res = am.generate(input=str(wav), **gen_kwargs) + finally: + wav.unlink(missing_ok=True) + if not res or not isinstance(res[0], dict) or "text" not in res[0]: + raise RuntimeError(f"FunASR 返回结果异常：{short_err(Exception(repr(res)))}") + return str(res[0]["text"]) + + +def paraformer_paragraphs(text: str) -> str: + """整段带标点文本按句号等断句，每 ~6 句插一个空行。""" + sents = [s for s in SENT_RE.findall(text) if s.strip()] + if len(sents) <= SENTENCES_PER_PARAGRAPH: + return text.strip() + return "\n\n".join( + "".join(sents[i:i + SENTENCES_PER_PARAGRAPH]).strip() + for i in range(0, len(sents), SENTENCES_PER_PARAGRAPH) + ) + + +def write_output(audio: Path, body: str) -> Path: + out = OUTPUT_DIR / (audio.stem + ".md") + out.write_text(f"# {audio.stem}\n\n{body}\n", encoding="utf-8") + return out + + +# ---------------------------------------------------------------- 子命令 + +def cmd_download_only(args: argparse.Namespace) -> int: + log(f"==> 预下载/校验 Whisper 模型：{args.model}（来自 HuggingFace）…") + m = load_whisper(args.model) + log(f" 缓存目录：{WHISPER_CACHE_DIR}") + del m + gc.collect() + log("==> 预下载/校验 FunASR 模型：SeACo-Paraformer + VAD + 标点（来自 ModelScope）…") + am = load_paraformer() + log(f" 缓存目录：{FUNASR_CACHE_DIR}") + del am + gc.collect() + log("全部模型就绪，之后可离线转写。") + return 0 + + +# ---------------------------------------------------------------- 主流程 + +def main() -> int: + args = parse_args() + OUTPUT_DIR.mkdir(parents=True, exist_ok=True) + INPUT_DIR.mkdir(parents=True, exist_ok=True) + + if args.download_only: + return cmd_download_only(args) + + files = collect_audio_files() + if not files: + log("input_audio/ 中没有音频文件" + f"（支持：{' '.join(sorted(SUPPORTED_EXTS))}）。把录音放进去后再运行。") + return 0 + log(f"在 input_audio/ 找到 {len(files)} 个音频文件。") + + todo: list[Path] = [] + skipped = 0 + for f in files: + if output_is_fresh(f): + skipped += 1 + log(f" 跳过（输出已存在且比音频新）：{f.name}") + else: + todo.append(f) + if not todo: + log(f"==== 汇总 ====") + log(f"成功 0 / 跳过 {skipped} / 失败 0（全部文件均为最新）") + return 0 + + hotwords = load_hotwords() + log(f"热词：{len(hotwords)} 个（hotwords.txt）") + + forced = bool(args.engine or args.language) + whisper_model = None + whisper_items: list[tuple[Path, str | None]] = [] # (音频, whisper 语言码或 None=自动) + para_items: list[Path] = [] + ok_n = fail_n = 0 + failures: list[tuple[str, str]] = [] + + # ---------- 阶段一：解码 + 语言探测 + 路由 ---------- + if forced: + for f in todo: + eng, lang, basis = forced_route(args) + log(f" 路由：{f.name} → {eng}（{basis}）") + if eng == "whisper": + whisper_items.append((f, lang)) + else: + para_items.append(f) + else: + log(f"==> 加载 Whisper（{args.model}）：先做语言探测，之后复用同一实例转写 …") + whisper_model = load_whisper(args.model) + for f in todo: + log(f" 探测：{f.name}") + try: + slices = probe_slices(f) + detected, prob = probe_language(whisper_model, slices) + del slices + except Exception as exc: + fail_n += 1 + failures.append((f.name, f"解码/探测失败：{short_err(exc, 200)}")) + log(f" 路由：{f.name} → 失败（解码/探测失败：{short_err(exc)}）") + continue + eng, basis = auto_route(detected, prob) + log(f" 路由：{f.name} → {eng}（{basis}）") + if eng == "whisper": + whisper_items.append((f, detected)) + else: + para_items.append(f) + + if args.dry_run: + log("==== dry-run 结束（未转写）====") + log(f"Whisper 组 {len(whisper_items)} 个 / Paraformer 组 {len(para_items)} 个 / 跳过 {skipped} 个") + return 0 + + # ---------- 阶段二：Whisper 组（探测时模型已加载，直接复用） ---------- + pipe = None + if whisper_items: + if whisper_model is None: + log(f"==> 加载 Whisper（{args.model}）…") + whisper_model = load_whisper(args.model) + try: + pipe = make_batched_pipeline(whisper_model) + except Exception: + pipe = None # 旧版无 BatchedInferencePipeline → 直接用普通 transcribe + for f, lang in whisper_items: + t0 = time.monotonic() + log(f" [Whisper] 开始：{f.name}（language={lang or '自动'}）") + try: + audio = decode_audio(f) + prompt = build_whisper_prompt(hotwords, lang) + segments, _info = whisper_transcribe(whisper_model, pipe, audio, lang, prompt) + body = merge_whisper_segments(segments) + if not body.strip(): + raise RuntimeError("转写结果为空（可能是纯静音或音频异常）") + out = write_output(f, body) + log(f" [Whisper] 完成：{f.name} → {out.name}（耗时 {time.monotonic() - t0:.1f}s）") + ok_n += 1 + except Exception as exc: + fail_n += 1 + failures.append((f.name, f"{type(exc).__name__}: {short_err(exc, 300)}")) + log(f" [Whisper] 失败：{f.name}（{short_err(exc, 200)}）") + + # 两引擎不同时常驻：先彻底释放 Whisper，再加载 Paraformer + if whisper_model is not None: + del whisper_model, pipe + whisper_model = None + gc.collect() + log("==> 已释放 Whisper 模型") + + # ---------- 阶段三：Paraformer 组 ---------- + if para_items: + log("==> 加载 FunASR（SeACo-Paraformer + VAD + 标点）…") + am = load_paraformer() + hotword_str = " ".join(hotwords) + for f in para_items: + t0 = time.monotonic() + log(f" [Paraformer] 开始：{f.name}") + try: + audio = decode_audio(f) + text = paraformer_transcribe(am, audio, hotword_str) + body = paraformer_paragraphs(text) + if not body.strip(): + raise RuntimeError("转写结果为空（可能是纯静音或音频异常）") + out = write_output(f, body) + log(f" [Paraformer] 完成：{f.name} → {out.name}（耗时 {time.monotonic() - t0:.1f}s）") + ok_n += 1 + except Exception as exc: + fail_n += 1 + failures.append((f.name, f"{type(exc).__name__}: {short_err(exc, 300)}")) + log(f" [Paraformer] 失败：{f.name}（{short_err(exc, 200)}）") + del am + gc.collect() + + # ---------- 汇总 ---------- + log("==== 汇总 ====") + log(f"成功 {ok_n} / 跳过 {skipped} / 失败 {fail_n}") + for name, reason in failures: + log(f" ✗ {name}：{reason}") + return 1 if fail_n else 0 + + +if __name__ == "__main__": + sys.exit(main()) |