diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..88df2ae --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,48 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + branches: [main] + +jobs: + lint: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + - run: pip install ruff + - run: ruff check app/ --ignore E501 + - run: ruff format --check app/ + + test: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + - name: Install lightweight test dependencies (no GPU packages) + run: pip install pytest pytest-cov fastapi httpx numpy aiofiles starlette python-multipart + - name: Run tests + run: | + pytest tests/test_security.py tests/test_voiceprint_db.py tests/test_job_service.py \ + -v --tb=short --no-header + + security-scan: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Install pip-audit + run: pip install pip-audit + - name: Run pip-audit + run: pip-audit -r requirements.txt --ignore-vuln PYSEC-2022-42969 || true + # || true: 不阻断构建，但在 Summary 中显示漏洞报告 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index f424b1b..9f51abe 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -3,6 +3,9 @@ name: Build and publish Docker image on: release: types: [published] + push: + tags: + - 'v*' workflow_dispatch: jobs: @@ -26,14 +29,24 @@ jobs: username: ${{ github.actor }} password: ${{ secrets.GITHUB_TOKEN }} + - name: Log in to Docker Hub + uses: docker/login-action@v3 + with: + username: ${{ secrets.DOCKERHUB_USERNAME }} + password: ${{ secrets.DOCKERHUB_TOKEN }} + - name: Compute image tags id: tags run: | - IMAGE="ghcr.io/${{ github.repository }}" - TAGS="$IMAGE:latest" - if [ "${{ github.event_name }}" = "release" ]; then - VERSION="${GITHUB_REF#refs/tags/}" - TAGS="$TAGS,$IMAGE:$VERSION" + GHCR_IMAGE="ghcr.io/${{ github.repository }}" + DOCKERHUB_IMAGE="${{ secrets.DOCKERHUB_USERNAME }}/voscript" + TAGS="$GHCR_IMAGE:latest,$DOCKERHUB_IMAGE:latest" + # Tag-triggered builds (release or push-tag) get the version tag too. + # Strip leading "v" for Docker Hub convention (0.7.0), keep raw ref for GHCR. + if [ "${{ github.event_name }}" = "release" ] || [ "${{ github.event_name }}" = "push" ]; then + RAW_VERSION="${GITHUB_REF#refs/tags/}" + STRIPPED_VERSION="${RAW_VERSION#v}" + TAGS="$TAGS,$GHCR_IMAGE:$RAW_VERSION,$DOCKERHUB_IMAGE:$STRIPPED_VERSION" fi echo "tags=$TAGS" >> "$GITHUB_OUTPUT" diff --git a/.gitignore b/.gitignore index 66374a4..0bc50c0 100644 --- a/.gitignore +++ b/.gitignore @@ -32,3 +32,7 @@ logs/ node_modules/ tmp/ .claude/ + +# Code review intermediate artifacts +.full-review/ +.full-review-archive/ diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..07c254a --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,89 @@ +# 贡献指南 + +感谢你考虑向 voscript 贡献！ + +## 开始之前 + +- 提 issue 之前先搜一下是否已有相同问题 +- 大的改动建议先开 issue 讨论设计方向 + +## 开发环境 + +```bash +# 安装依赖（需要 Python 3.11+） +pip install fastapi uvicorn pytest pytest-asyncio aiofiles httpx starlette python-multipart + +# 运行测试 +pytest tests/ -v + +# 启动本地服务（需要 GPU + 模型文件） +cd app && uvicorn main:app --reload --port 8780 +``` + +## 提 PR 须知 + +1. **Fork → 新建分支 → PR**，分支命名建议：`feat/xxx`、`fix/xxx`、`docs/xxx` +2. **测试**：新功能必须附带对应测试（`tests/` 目录），所有测试必须通过 +3. **文档**：影响 API 或行为的改动需同步更新 `doc/` 中的相关文档 +4. **提交信息**：用英文小写动词开头，例如 `fix: handle zero-length audio` / `feat: add speaker rename API` + +## 代码风格 + +- Python：PEP 8，函数命名 `snake_case`，类型注解尽量完整 +- 不要引入 print 调试语句，用 `logger.debug()` +- 安全敏感代码（鉴权、文件路径）改动务必在 PR 描述中说明 + +## 报告安全问题 + +**不要开公开 issue**。请直接发 email 或通过 GitHub Security Advisory 私下报告。 +详见 [SECURITY.md](./SECURITY.md)。 + +## 行为准则 + +友善、建设性地交流。维护者有权关闭不遵守基本礼仪的讨论。 + +--- + +# Contributing (English) + +Thank you for contributing to voscript! + +## Before You Start + +- Search existing issues before filing a new one +- For large changes, open an issue first to discuss design direction + +## Development Setup + +```bash +# Requires Python 3.11+ +pip install fastapi uvicorn pytest pytest-asyncio aiofiles httpx starlette python-multipart + +# Run tests +pytest tests/ -v + +# Start local server (requires GPU + model files) +cd app && uvicorn main:app --reload --port 8780 +``` + +## Pull Request Guidelines + +1. **Fork → branch → PR** — suggested branch names: `feat/xxx`, `fix/xxx`, `docs/xxx` +2. **Tests**: new features must include tests; all tests must pass +3. **Docs**: changes to API behavior must update the relevant files in `doc/` +4. **Commit messages**: lowercase verb prefix, e.g. `fix: handle zero-length audio` + +## Code Style + +- Python: PEP 8, `snake_case` for functions, type annotations where practical +- Use `logger.debug()` instead of `print` +- Security-sensitive changes (auth, file paths) must be explained in the PR description + +## Reporting Security Issues + +**Do not open a public issue.** Use GitHub Security Advisory or email privately. +See [SECURITY.md](./SECURITY.md). + +## Code of Conduct + +Be kind and constructive. Maintainers may close discussions that violate basic courtesy norms. diff --git a/README.en.md b/README.en.md index 05714ff..262c896 100644 --- a/README.en.md +++ b/README.en.md @@ -1,93 +1,92 @@ -# VoScript +
+ +# 🎙️ VoScript [简体中文](./README.md) | **English** -Self-hosted GPU transcription service — **meeting transcripts that remember -their speakers.** A small HTTP API that turns audio into timestamped text -labelled with speaker names, and that auto-recognizes returning voices -across recordings. + + CI + +License: Apache 2.0 +Docker ready -``` -Audio ──► faster-whisper large-v3 (transcription) - ──► pyannote 3.1 (speaker diarization) - ──► DeepFilterNet / noisereduce (optional denoising) - ──► WeSpeaker ResNet34 (speaker embeddings) - ──► VoiceprintDB (cosine match vs. enrolled speakers) - ──► timestamped text with identified speaker names -``` +**Meeting recordings → transcripts with real speaker names. Self-hosted, GPU-powered, remembers every voice.** -What sets it apart from a plain whisper wrapper: the **persistent -voiceprint library**. Enroll a speaker once, and from then on every -recording they appear in gets their real name automatically. +[Quickstart](./doc/quickstart.en.md) · [API Reference](./doc/api.en.md) · [Security](./doc/security.en.md) · [Benchmarks](./doc/benchmarks.en.md) · [Changelog](./doc/changelog.en.md) -> Example consumer: [OpenPlaud(Maple)](https://github.com/MapleEve/openplaud) -> uses voscript as the backend for meeting recordings. voscript itself is -> just an HTTP service — any client that can POST multipart audio works. +
-## Documentation +--- -All detailed docs live in [`doc/`](./doc/). Chinese is the default, every -page has an English counterpart: +You have a meeting recording with six people. You want to know who said what. Whisper gives you a wall of text. pyannote can split it into "Speaker A / Speaker B / Speaker C" — but it doesn't know who anyone is. You still have to label every recording by hand. -| Topic | 中文 | English | -| --- | --- | --- | -| Quickstart | [quickstart.zh.md](./doc/quickstart.zh.md) | [quickstart.en.md](./doc/quickstart.en.md) | -| API reference | [api.zh.md](./doc/api.zh.md) | [api.en.md](./doc/api.en.md) | -| **Install guide for AI agents** | [ai-install.zh.md](./doc/ai-install.zh.md) | [ai-install.en.md](./doc/ai-install.en.md) | -| **Usage guide for AI agents** | [ai-usage.zh.md](./doc/ai-usage.zh.md) | [ai-usage.en.md](./doc/ai-usage.en.md) | -| Security policy | [security.zh.md](./doc/security.zh.md) | [security.en.md](./doc/security.en.md) | -| Benchmarks (real-audio wall clock + resource usage) | [benchmarks.zh.md](./doc/benchmarks.zh.md) | [benchmarks.en.md](./doc/benchmarks.en.md) | -| Changelog | [changelog.zh.md](./doc/changelog.zh.md) | [changelog.en.md](./doc/changelog.en.md) | - -First-time deployers: start with the [Quickstart](./doc/quickstart.en.md). -AI agents integrating the API: read the [AI usage guide](./doc/ai-usage.en.md). -AI agents deploying the service for a user: read the -[AI install guide](./doc/ai-install.en.md). - -## Features +VoScript fixes that: **enroll a voice once, and it gets automatically identified in every future recording**. Not "Speaker 2" — "Maple". -- **Async job pipeline**: `queued → converting → denoising (optional) → transcribing → identifying → completed` -- **Chinese + multilingual transcription** (WhisperX + faster-whisper large-v3, **word-level timestamps** via forced alignment; omit `language` to auto-detect — Mandarin audio outputs Simplified Chinese) -- **Speaker diarization** (pyannote 3.1) + **WeSpeaker ResNet34** embeddings -- **Adaptive voiceprint threshold**: `VOICEPRINT_THRESHOLD` (default 0.75) is the base; the actual threshold relaxes per-speaker based on intra-cluster std of enrolled embeddings — fixed −0.05 for 1 sample, `min(3×std, 0.10)` for 2+, floor at 0.60. Lifted recall from 50% to 70% on 10 real recordings with zero false positives -- **Optional denoising with SNR gate**: `DENOISE_MODEL` (`none` | `deepfilternet` | `noisereduce`); `DENOISE_SNR_THRESHOLD` (default 10.0 dB) — audio above this SNR is considered clean and skipped automatically, preventing DeepFilterNet from degrading already-clean recordings -- **AS-norm voiceprint scoring**: at startup, automatically builds an impostor cohort from existing transcription embeddings and applies Adaptive Score Normalization — eliminates speaker-dependent baseline bias, ~15–30% relative EER improvement -- **Persistent voiceprints**: enroll once, auto-match across future recordings. sqlite + sqlite-vec under the hood — top-k nearest-neighbour search scales to thousands of speakers -- **File hash deduplication**: submitting the same file twice returns the existing result immediately, skipping Whisper GPU inference -- **Stable HTTP contract**: `/api/transcribe`, `/api/jobs/{id}`, `/api/voiceprints*`, etc. — any HTTP client works -- **Container runs as non-root**; all `/api/*` routes accept optional Bearer / `X-API-Key` auth (constant-time compare); uploads capped by `MAX_UPLOAD_BYTES`; voiceprint DB is concurrency-safe with atomic writes — full hardening list in [`doc/security.en.md`](./doc/security.en.md) -- Minimal built-in web UI at `/` for manual testing +``` +Audio ──► faster-whisper large-v3 transcription + word-level timestamps + ──► pyannote 3.1 speaker diarization + ──► WeSpeaker ResNet34 speaker embeddings + ──► VoiceprintDB (AS-norm) match against enrolled voices + ──► timestamped transcript with real speaker names +``` ## 30-second start -```bash -git clone https://github.com/MapleEve/voscript.git -cd voscript - -cp .env.example .env -# edit .env — at minimum set HF_TOKEN and API_KEY +> **Security**: set a strong `API_KEY` in `.env` before exposing this on any network. Without it, anyone can delete your voiceprint library or trigger GPU jobs. +```bash +git clone https://github.com/MapleEve/voscript.git && cd voscript +cp .env.example .env # at minimum: HF_TOKEN and API_KEY docker compose up -d --build curl -sf http://localhost:8780/healthz ``` -Full steps + troubleshooting in [`doc/quickstart.en.md`](./doc/quickstart.en.md). +Full setup + troubleshooting → [`doc/quickstart.en.md`](./doc/quickstart.en.md) + +## Features + +- **Persistent voiceprint library** — enroll once, auto-match across all future recordings. sqlite + sqlite-vec under the hood, top-k nearest-neighbour search, scales to thousands of speakers +- **AS-norm scoring** — builds an impostor cohort from existing transcription embeddings at startup; eliminates speaker-dependent baseline bias, ~15–30% relative EER improvement +- **Adaptive threshold** — each speaker's match threshold relaxes dynamically based on enrollment variance; lifted recall from 50% to 70% on 10 real recordings with zero false positives +- **Speaker cluster consolidation** — when diarization splits one person into multiple clusters, they're automatically merged to a single label +- **Word-level timestamps** — WhisperX forced alignment, every word precisely timed +- **Optional denoising with SNR gate** — DeepFilterNet / noisereduce; audio above the SNR threshold is treated as clean and skipped automatically (prevents degrading already-clean recordings) +- **File hash deduplication** — submitting the same file twice returns the existing result immediately, no GPU re-run +- **Job persistence** — completed transcriptions remain accessible after restart +- **Ngram dedup** — `no_repeat_ngram_size` parameter suppresses repetitive filler words in the transcript +- **Plain HTTP contract** — any client that can send multipart/form-data works, no framework lock-in + +Security: path traversal protection, non-root container, upload size cap, constant-time auth, atomic writes — full list in [`doc/security.en.md`](./doc/security.en.md) + +## Integration + +It's a plain HTTP service. Two config values and you're done: + +- **Transcription base URL**: `http://:8780` +- **API key**: the `API_KEY` you set in `.env` + +[BetterAINote](https://github.com/MapleEve/openplaud) connects this way. Any other client works the same. Full API contract → [`doc/api.en.md`](./doc/api.en.md) -## How to integrate +## Documentation + +| Topic | 中文 | English | +| --- | --- | --- | +| Quickstart | [quickstart.zh.md](./doc/quickstart.zh.md) | [quickstart.en.md](./doc/quickstart.en.md) | +| API reference | [api.zh.md](./doc/api.zh.md) | [api.en.md](./doc/api.en.md) | +| Install guide for AI agents | [ai-install.zh.md](./doc/ai-install.zh.md) | [ai-install.en.md](./doc/ai-install.en.md) | +| Usage guide for AI agents | [ai-usage.zh.md](./doc/ai-usage.zh.md) | [ai-usage.en.md](./doc/ai-usage.en.md) | +| Security policy | [security.zh.md](./doc/security.zh.md) | [security.en.md](./doc/security.en.md) | +| Benchmarks | [benchmarks.zh.md](./doc/benchmarks.zh.md) | [benchmarks.en.md](./doc/benchmarks.en.md) | +| Changelog | [changelog.zh.md](./doc/changelog.zh.md) | [changelog.en.md](./doc/changelog.en.md) | -voscript is a plain HTTP service — no specific client is required. Anything -that can send `multipart/form-data` works (curl, axios, requests, browser -uploads, …). +## Contributing -A typical integration — OpenPlaud(Maple), under Settings → Transcription: +PRs welcome — read [CONTRIBUTING.md](./CONTRIBUTING.md) first. -- **Private transcription base URL**: `http://:8780` -- **Private transcription API key**: the same `API_KEY` as in `.env` +## Star History -After that its worker routes every recording through this service. If -you're writing your own client, the full contract + error table lives in -[`doc/api.en.md`](./doc/api.en.md). +[![Star History Chart](https://api.star-history.com/svg?repos=MapleEve/voscript&type=date)](https://www.star-history.com/#MapleEve/voscript&type=date) ## License -MIT — see [LICENSE](./LICENSE). +Apache 2.0 — [LICENSE](./LICENSE) diff --git a/README.md b/README.md index 1523ada..ebae389 100644 --- a/README.md +++ b/README.md @@ -1,85 +1,90 @@ -# VoScript +
+ +# VoScript 🎙️ **简体中文** | [English](./README.en.md) -自托管的 GPU 转录服务——**带说话人记忆的会议逐字稿**。一套简单的 HTTP API， -把音频转成带说话人名字的文字，同一个人再次出现会自动识别。 +CI +License +Docker -``` -音频 ──► faster-whisper large-v3 （转录） - ──► pyannote 3.1 （说话人分离） - ──► DeepFilterNet / noisereduce（可选降噪） - ──► WeSpeaker ResNet34 （声纹提取） - ──► VoiceprintDB （与已注册声纹做余弦匹配） - ──► 带时间戳和已识别说话人姓名的文本 -``` +**会议录音 → 逐字稿，带真名说话人标签。自托管，GPU 驱动，记得住每个人的声音。** -与"纯 whisper 包装"的区别：**持久化声纹库**。登记过一次，之后所有录音里这个 -人都会被自动贴上真名，不需要每次人工贴标签。 +[快速上手](./doc/quickstart.zh.md) · [API 参考](./doc/api.zh.md) · [安全策略](./doc/security.zh.md) · [Benchmarks](./doc/benchmarks.zh.md) · [更新日志](./doc/changelog.zh.md) -> 用例参考：[OpenPlaud(Maple)](https://github.com/MapleEve/openplaud) 把本服务作为 -> 会议录音的后端——把这个 repo 当作标准 HTTP 服务对接即可，不限特定客户端。 +
-## 文档 +--- -所有详细文档都在 [`doc/`](./doc/)，默认中文，每一份都有对应英文： +开完会，录音里有六个人，你想知道谁说了什么。Whisper 只给你一段文字，pyannote 能告诉你"说话人A/说话人B"，但它不认识人——每次还是得手动贴名字。 -| 主题 | 中文 | English | -| --- | --- | --- | -| 快速安装 | [quickstart.zh.md](./doc/quickstart.zh.md) | [quickstart.en.md](./doc/quickstart.en.md) | -| API 参考 | [api.zh.md](./doc/api.zh.md) | [api.en.md](./doc/api.en.md) | -| **给 AI 的安装部署指南** | [ai-install.zh.md](./doc/ai-install.zh.md) | [ai-install.en.md](./doc/ai-install.en.md) | -| **给 AI 的接口使用指南** | [ai-usage.zh.md](./doc/ai-usage.zh.md) | [ai-usage.en.md](./doc/ai-usage.en.md) | -| 安全策略 | [security.zh.md](./doc/security.zh.md) | [security.en.md](./doc/security.en.md) | -| Benchmarks（真实音频耗时 + 资源占用） | [benchmarks.zh.md](./doc/benchmarks.zh.md) | [benchmarks.en.md](./doc/benchmarks.en.md) | -| 更新日志 | [changelog.zh.md](./doc/changelog.zh.md) | [changelog.en.md](./doc/changelog.en.md) | - -人第一次部署 → [快速安装](./doc/quickstart.zh.md)； -AI agent 帮用户部署 → [给 AI 的安装部署指南](./doc/ai-install.zh.md)； -AI agent 调用接口 → [给 AI 的接口使用指南](./doc/ai-usage.zh.md)。 - -## 功能 +VoScript 解决的就是这个：**登记一次声纹，之后所有录音里这个人都会被自动识别出来**。不是"说话人2"，是"Maple"。 -- **异步任务流水线**：`queued → converting → denoising（可选）→ transcribing → identifying → completed` -- **中文 + 多语种转录**（WhisperX + faster-whisper large-v3，**带词级时间戳**的 forced alignment；`language` 省略时自动检测语言，普通话音频输出简体中文） -- **说话人分离**（pyannote 3.1）+ **WeSpeaker ResNet34** 声纹提取 -- **自适应声纹阈值**：`VOICEPRINT_THRESHOLD`（默认 0.75）作为基准，实际阈值按每位说话人已注册向量的簇内标准差动态宽松：1 条样本固定宽松 −0.05，2 条及以上按 `min(3×std, 0.10)` 宽松，最低不低于 0.60。在 10 条真实录音上召回率从 50% 提升至 70%，零误识别 -- **可选降噪 + SNR 门控**：`DENOISE_MODEL`（`none` | `deepfilternet` | `noisereduce`），`DENOISE_SNR_THRESHOLD`（默认 10.0 dB）——高于此 SNR 的录音视为干净音频自动跳过，防止 DeepFilterNet 劣化已清晰的录音 -- **AS-norm 声纹评分**：启动时自动从已有转录的声纹 embedding 构建 impostor cohort，用自适应分数归一化（AS-norm）替代原始余弦，消除说话人依赖的基准偏差，相对 EER 降低 15–30% -- **持久化声纹**：一次登记，后续录音自动识别。底层 sqlite + sqlite-vec，top-k 近邻搜索 O(log N)，上千个声纹毫无压力 -- **文件哈希去重**：相同文件重复提交时直接返回已有结果，不再重跑 Whisper GPU 推理 -- **稳定的 HTTP 合同**：`/api/transcribe`、`/api/jobs/{id}`、`/api/voiceprints*` 等，任何 HTTP 客户端都能接入 -- **容器以非 root 用户运行**；所有 `/api/*` 路由支持可选 Bearer / `X-API-Key` 鉴权（常量时间对比）；上传有 `MAX_UPLOAD_BYTES` 上限；声纹库并发安全、原子写入——完整硬化清单见 [`doc/security.zh.md`](./doc/security.zh.md) -- `/` 自带一个轻量 Web UI，方便单独测试 +``` +音频 ──► faster-whisper large-v3 转录 + 词级时间戳 + ──► pyannote 3.1 说话人分离 + ──► WeSpeaker ResNet34 声纹提取 + ──► VoiceprintDB (AS-norm) 与已注册声纹匹配 + ──► 带时间戳 + 真名的逐字稿 +``` ## 30 秒上手 -```bash -git clone https://github.com/MapleEve/voscript.git -cd voscript - -cp .env.example .env -# 编辑 .env —— 至少要填 HF_TOKEN 和 API_KEY +> **安全警告**：生产环境或公网暴露前**必须**在 `.env` 里设置 `API_KEY`，否则任何人都能删你的声纹库、触发 GPU 任务。 +```bash +git clone https://github.com/MapleEve/voscript.git && cd voscript +cp .env.example .env # 至少填 HF_TOKEN 和 API_KEY docker compose up -d --build curl -sf http://localhost:8780/healthz ``` -完整步骤 + 排障清单看 [`doc/quickstart.zh.md`](./doc/quickstart.zh.md)。 +完整步骤 + 排障清单 → [`doc/quickstart.zh.md`](./doc/quickstart.zh.md) + +## 功能 + +- **持久化声纹库** — 登记一次，后续所有录音自动识别。底层 sqlite + sqlite-vec，top-k 近邻，上千声纹毫无压力 +- **AS-norm 评分** — 启动时自动从历史转录构建 impostor cohort，消除说话人依赖的基准偏差，相对 EER 降低 15–30% +- **自适应阈值** — 每位说话人的实际识别阈值根据注册样本的方差动态宽松，10 条真实录音召回率从 50% → 70%，零误识别 +- **说话人聚类合并** — 同一个人被分出多个聚类时自动合并为一个标签 +- **词级时间戳** — WhisperX forced alignment，每个词都有精确时间 +- **可选降噪 + SNR 门控** — DeepFilterNet / noisereduce，SNR 高于阈值的录音自动跳过（防止对干净音频劣化） +- **文件哈希去重** — 相同文件重复提交直接返回已有结果，不重跑 GPU +- **任务持久化** — 重启后已完成任务仍可访问 +- **ngram 去重** — `no_repeat_ngram_size` 参数抑制转录中的口语重复（比如"就是就是就是"） +- **纯 HTTP 合同** — 任何能发 multipart/form-data 的客户端都能接入，不绑定特定框架 + +安全相关：路径遍历防护、非 root 容器、上传大小限制、常量时间鉴权、原子写入……完整清单 → [`doc/security.zh.md`](./doc/security.zh.md) + +## 接入 + +就是个普通 HTTP 服务，没有特殊依赖。配两个值就行： + +- **转录服务地址**：`http://<主机>:8780` +- **API Key**：`.env` 里设的那个 `API_KEY` + +[BetterAINote](https://github.com/MapleEve/openplaud) 就是这样接的，其它客户端一样。完整接口合同 → [`doc/api.zh.md`](./doc/api.zh.md) -## 怎么接入 +## 文档 + +| 主题 | 中文 | English | +| --- | --- | --- | +| 快速安装 | [quickstart.zh.md](./doc/quickstart.zh.md) | [quickstart.en.md](./doc/quickstart.en.md) | +| API 参考 | [api.zh.md](./doc/api.zh.md) | [api.en.md](./doc/api.en.md) | +| 给 AI 的安装指南 | [ai-install.zh.md](./doc/ai-install.zh.md) | [ai-install.en.md](./doc/ai-install.en.md) | +| 给 AI 的接口指南 | [ai-usage.zh.md](./doc/ai-usage.zh.md) | [ai-usage.en.md](./doc/ai-usage.en.md) | +| 安全策略 | [security.zh.md](./doc/security.zh.md) | [security.en.md](./doc/security.en.md) | +| Benchmarks | [benchmarks.zh.md](./doc/benchmarks.zh.md) | [benchmarks.en.md](./doc/benchmarks.en.md) | +| 更新日志 | [changelog.zh.md](./doc/changelog.zh.md) | [changelog.en.md](./doc/changelog.en.md) | -voscript 就是个普通的 HTTP 服务，没有特定客户端的强依赖。任何能发 -`multipart/form-data` 的东西都能用（curl、axios、requests、网页上传框……）。 +## 贡献 -一个典型对接示例——OpenPlaud(Maple) 的"设置 → 转录"里配： +欢迎 PR，请先读 [CONTRIBUTING.md](./CONTRIBUTING.md)。 -- **Private transcription base URL**：`http://<主机>:8780` -- **Private transcription API key**：跟 `.env` 里的 `API_KEY` 一致 +## Star History -之后它的 worker 会自动把每条录音都丢给这个服务。想自己写客户端的话，看 -[`doc/api.zh.md`](./doc/api.zh.md) 的完整合同 + 错误码表。 +[![Star History Chart](https://api.star-history.com/svg?repos=MapleEve/voscript&type=date)](https://www.star-history.com/#MapleEve/voscript&type=date) ## License -MIT —— 看 [LICENSE](./LICENSE)。 +Apache 2.0 — [LICENSE](./LICENSE) diff --git a/app/Dockerfile b/app/Dockerfile index 4b635d4..208be7e 100755 --- a/app/Dockerfile +++ b/app/Dockerfile @@ -1,3 +1,8 @@ +# TODO: pin digest — 运行： +# docker pull pytorch/pytorch:2.4.1-cuda12.4-cudnn9-runtime +# docker inspect pytorch/pytorch:2.4.1-cuda12.4-cudnn9-runtime --format='{{index .RepoDigests 0}}' +# 然后改为：FROM pytorch/pytorch:2.4.1-cuda12.4-cudnn9-runtime@sha256: +# 不 pin digest 的话 mutable tag 可能导致构建不可重现（对应评审 CD-H6 / BP-M6）。 FROM pytorch/pytorch:2.4.1-cuda12.4-cudnn9-runtime ENV DEBIAN_FRONTEND=noninteractive @@ -24,6 +29,8 @@ RUN apt-get update && apt-get install -y --no-install-recommends \ ffmpeg \ libsndfile1 \ git \ + # TODO: 待确认 fake_git hack 删除后可移除此依赖（评审 BP-L5）。 + # DeepFilterNet 在初始化时可能探测 `git`，若已完全去除该 hack，可删除此行。 && rm -rf /var/lib/apt/lists/* # Non-root runtime user. An RCE inside the container only owns the service's @@ -43,7 +50,6 @@ RUN if [ -n "$PIP_INDEX_URL" ]; then \ fi COPY --chown=app:app . . -RUN chown -R app:app /app # HuggingFace cache lives under the app user's reach, not /root. ENV HF_HOME=/cache \ diff --git a/app/api/__init__.py b/app/api/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/deps.py b/app/api/deps.py new file mode 100644 index 0000000..3de8871 --- /dev/null +++ b/app/api/deps.py @@ -0,0 +1,32 @@ +"""FastAPI dependency callables shared across all routers.""" + +import hmac + +from fastapi import Header, HTTPException, Request + +from config import API_KEY + + +async def verify_api_key(x_api_key: str | None = Header(None)) -> None: + """Dependency that enforces API-key authentication. + + Raises HTTP 403 when a key is configured but the supplied value does not + match. When no key is configured (open mode) the dependency is a no-op. + + Note: the middleware in main.py handles the Bearer-token path and path + allow-listing; this dependency is the fallback for router-level auth. + """ + if API_KEY is None: + return # open mode — no check needed + if not x_api_key or not hmac.compare_digest(x_api_key, API_KEY): + raise HTTPException(403, "Invalid API key") + + +def get_db(request: Request): + """Return the VoiceprintDB instance stored on app.state.""" + return request.app.state.db + + +def get_pipeline(request: Request): + """Return the TranscriptionPipeline instance stored on app.state.""" + return request.app.state.pipeline diff --git a/app/api/routers/__init__.py b/app/api/routers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/api/routers/health.py b/app/api/routers/health.py new file mode 100644 index 0000000..bff662e --- /dev/null +++ b/app/api/routers/health.py @@ -0,0 +1,17 @@ +"""Health-check endpoint.""" + +from fastapi import APIRouter +from fastapi.responses import HTMLResponse +from pathlib import Path + +router = APIRouter() + + +@router.get("/healthz") +async def healthz(): + return {"ok": True} + + +@router.get("/", response_class=HTMLResponse) +async def index(): + return Path("static/index.html").read_text(encoding="utf-8") diff --git a/app/api/routers/transcriptions.py b/app/api/routers/transcriptions.py new file mode 100644 index 0000000..ec04148 --- /dev/null +++ b/app/api/routers/transcriptions.py @@ -0,0 +1,343 @@ +"""Transcription endpoints. + +Covers: + POST /api/transcribe + GET /api/jobs/{job_id} + GET /api/transcriptions + GET /api/transcriptions/{tr_id} + GET /api/transcriptions/{tr_id}/audio + PUT /api/transcriptions/{tr_id}/segments/{seg_id}/speaker + GET /api/export/{tr_id} +""" + +import json +import logging +import uuid +from datetime import datetime +from pathlib import PurePosixPath +from threading import Thread +from typing import Annotated + +from fastapi import APIRouter, File, Form, HTTPException +from fastapi import Path as FPath +from fastapi import Request, UploadFile +from fastapi.responses import FileResponse, PlainTextResponse + +from api.deps import get_db, get_pipeline +from config import MAX_UPLOAD_BYTES, TRANSCRIPTIONS_DIR, UPLOAD_CHUNK, UPLOADS_DIR +from services.audio_service import ( + lookup_hash, + safe_log_filename, + safe_tr_dir, + save_upload_and_hash, +) +from services.job_service import jobs, run_transcription + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _format_srt_time(seconds: float) -> str: + # [CQ-M13] 防御 None / NaN / 负秒——SRT 不允许负时间戳，NaN 会导致 int() 抛异常。 + if seconds is None or seconds != seconds: # NaN 自身不等于自身 + seconds = 0.0 + seconds = max(0.0, float(seconds)) + h = int(seconds // 3600) + m = int((seconds % 3600) // 60) + s = int(seconds % 60) + ms = int((seconds % 1) * 1000) + return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" + + +def _format_timestamp(seconds: float) -> str: + if seconds is None or seconds != seconds: + seconds = 0.0 + seconds = max(0.0, float(seconds)) + m = int(seconds // 60) + s = int(seconds % 60) + return f"{m:02d}:{s:02d}" + + +# --------------------------------------------------------------------------- +# Routes +# --------------------------------------------------------------------------- + + +@router.post("/transcribe") +async def transcribe( + request: Request, + file: UploadFile = File(...), + language: str = Form(None), + min_speakers: int = Form(0), + max_speakers: int = Form(0), + denoise_model: str = Form("none"), + snr_threshold: float = Form(None), + no_repeat_ngram_size: str = Form("0"), +): + try: + no_repeat_ngram_size = int(no_repeat_ngram_size) + except (ValueError, TypeError): + raise HTTPException( + status_code=422, + detail=[ + { + "loc": ["body", "no_repeat_ngram_size"], + "msg": "value is not a valid integer", + "type": "type_error.integer", + } + ], + ) + pipeline = get_pipeline(request) + voiceprint_db = get_db(request) + + # Normalise empty string to None so pipeline treats it as auto-detect. + language = language.strip() if language else None + + job_id = f"tr_{datetime.now():%Y%m%d_%H%M%S}_{uuid.uuid4().hex[:6]}" + + safe_filename = PurePosixPath(file.filename or "upload").name or "upload" + # Strip control chars before using the name in paths/logs — PurePosixPath.name + # preserves newlines and ANSI escapes which would otherwise enable log injection. + safe_filename = safe_log_filename(safe_filename) or "upload" + save_path = UPLOADS_DIR / f"{job_id}_{safe_filename}" + + # PERF-C2: async write + streaming SHA-256 — no event-loop blockage on large uploads. + try: + _size, file_hash = await save_upload_and_hash( + file, save_path, MAX_UPLOAD_BYTES, UPLOAD_CHUNK + ) + except ValueError as exc: + save_path.unlink(missing_ok=True) + raise HTTPException(413, str(exc)) from exc + + # Dedup: if identical audio was already transcribed, return existing result. + existing_id = lookup_hash(file_hash) + if existing_id: + save_path.unlink(missing_ok=True) + logger.info( + "Dedup hit: %s already transcribed as %s", safe_filename, existing_id + ) + return {"id": existing_id, "status": "completed", "deduplicated": True} + + jobs[job_id] = { + "status": "queued", + "filename": safe_filename, + "created_at": datetime.now().isoformat(), + } + # CD-C3: daemon=True ensures this thread does not prevent the process from + # exiting on SIGTERM — the OS will clean up in-progress transcriptions on + # shutdown rather than hanging indefinitely waiting for the thread to finish. + thread = Thread( + target=run_transcription, + args=( + job_id, + save_path, + language, + min_speakers, + max_speakers, + pipeline, + voiceprint_db, + denoise_model, + snr_threshold, + file_hash, + no_repeat_ngram_size if no_repeat_ngram_size >= 3 else 0, + ), + daemon=True, + ) + thread.start() + + return {"id": job_id, "status": "queued"} + + +@router.get("/jobs/{job_id}") +async def get_job( + job_id: Annotated[str, FPath(pattern=r"^tr_[A-Za-z0-9_-]{1,64}$")], +): + if job_id in jobs: + job = jobs[job_id] + resp = {"id": job_id, "status": job["status"], "filename": job.get("filename")} + if job["status"] == "completed": + resp["result"] = job["result"] + elif job["status"] == "failed": + resp["error"] = job.get("error") + return resp + + # AR-C2 fallback: process restarted — try reading persisted status.json. + status_path = TRANSCRIPTIONS_DIR / job_id / "status.json" + result_path = TRANSCRIPTIONS_DIR / job_id / "result.json" + + if status_path.exists(): + try: + status_data = json.loads(status_path.read_text()) + except Exception: + raise HTTPException(404, "Job not found") + + current_status = status_data.get("status") + + if current_status == "completed" and result_path.exists(): + try: + result = json.loads(result_path.read_text(encoding="utf-8")) + except Exception: + result = None + return { + "id": job_id, + "status": "completed", + "filename": status_data.get("filename"), + "result": result, + } + + if current_status not in ("completed", "failed"): + # In-progress status persisted by a previous process that no longer + # owns this job — treat as a restart failure. + return { + "id": job_id, + "status": "failed", + "error": "Process restarted while job was in progress", + "filename": status_data.get("filename"), + } + + return { + "id": job_id, + "status": current_status, + "error": status_data.get("error"), + "filename": status_data.get("filename"), + } + + raise HTTPException(404, "Job not found") + + +@router.get("/transcriptions") +async def list_transcriptions(): + results = [] + for tr_dir in sorted(TRANSCRIPTIONS_DIR.iterdir(), reverse=True): + if not tr_dir.is_dir(): + continue + result_file = tr_dir / "result.json" + if result_file.exists(): + try: + data = json.loads(result_file.read_text(encoding="utf-8")) + results.append( + { + "id": data["id"], + "filename": data["filename"], + "created_at": data["created_at"], + "segment_count": len(data["segments"]), + "speaker_count": len(data.get("unique_speakers", [])), + } + ) + except Exception as exc: + logger.warning( + "Skipping corrupt result.json in %s: %s", tr_dir.name, exc + ) + return results + + +@router.get("/transcriptions/{tr_id}") +async def get_transcription( + tr_id: Annotated[str, FPath(pattern=r"^tr_[A-Za-z0-9_-]{1,64}$")], +): + result_file = safe_tr_dir(tr_id) / "result.json" + if not result_file.exists(): + raise HTTPException(404, "Transcription not found") + return json.loads(result_file.read_text(encoding="utf-8")) + + +@router.get("/transcriptions/{tr_id}/audio") +async def download_audio( + tr_id: Annotated[str, FPath(pattern=r"^tr_[A-Za-z0-9_-]{1,64}$")], +): + """Return the original uploaded audio file for this transcription.""" + result_file = safe_tr_dir(tr_id) / "result.json" + if not result_file.exists(): + raise HTTPException(404, "Transcription not found") + data = json.loads(result_file.read_text(encoding="utf-8")) + audio_file = UPLOADS_DIR / data["filename"] + if not audio_file.exists(): + raise HTTPException(404, "Original audio file not found") + return FileResponse(audio_file, filename=data["filename"]) + + +@router.put("/transcriptions/{tr_id}/segments/{seg_id}/speaker") +async def reassign_speaker( + tr_id: Annotated[str, FPath(pattern=r"^tr_[A-Za-z0-9_-]{1,64}$")], + seg_id: int, + speaker_name: str = Form(...), + speaker_id: str = Form(None), +): + """Reassign a segment to a different speaker and optionally enroll the voiceprint.""" + result_file = safe_tr_dir(tr_id) / "result.json" + if not result_file.exists(): + raise HTTPException(404, "Transcription not found") + data = json.loads(result_file.read_text(encoding="utf-8")) + + seg = next((s for s in data["segments"] if s["id"] == seg_id), None) + if seg is None: + raise HTTPException(404, "Segment not found") + + seg["speaker_name"] = speaker_name + if speaker_id: + seg["speaker_id"] = speaker_id + + # [CQ-H7] 同步更新 speaker_map，保持人工纠错在整条记录内一致。 + # 原 segment 可能引用一个 speaker_label（如 "SPEAKER_01"），我们在 speaker_map + # 的对应条目上更新 matched_name / matched_id，而不是改 key。 + spk_label = seg.get("speaker_label") + speaker_map = data.get("speaker_map") or {} + if spk_label and spk_label in speaker_map: + speaker_map[spk_label]["matched_name"] = speaker_name + if speaker_id: + speaker_map[spk_label]["matched_id"] = speaker_id + data["speaker_map"] = speaker_map + + result_file.write_text( + json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8" + ) + return {"ok": True} + + +@router.get("/export/{tr_id}") +async def export_transcription( + tr_id: Annotated[str, FPath(pattern=r"^tr_[A-Za-z0-9_-]{1,64}$")], + format: str = "srt", +): + result_file = safe_tr_dir(tr_id) / "result.json" + if not result_file.exists(): + raise HTTPException(404, "Transcription not found") + data = json.loads(result_file.read_text(encoding="utf-8")) + segments = data["segments"] + + if format == "srt": + lines = [] + for i, seg in enumerate(segments, 1): + start = _format_srt_time(seg["start"]) + end = _format_srt_time(seg["end"]) + lines.append( + f"{i}\n{start} --> {end}\n[{seg['speaker_name']}] {seg['text']}\n" + ) + return PlainTextResponse( + "\n".join(lines), + media_type="text/srt", + headers={"Content-Disposition": f'attachment; filename="{tr_id}.srt"'}, + ) + elif format == "txt": + lines = [] + for seg in segments: + ts = _format_timestamp(seg["start"]) + lines.append(f"[{ts}] {seg['speaker_name']}: {seg['text']}") + return PlainTextResponse( + "\n".join(lines), + media_type="text/plain", + headers={"Content-Disposition": f'attachment; filename="{tr_id}.txt"'}, + ) + elif format == "json": + return FileResponse( + result_file, media_type="application/json", filename=f"{tr_id}.json" + ) + else: + raise HTTPException(400, "Unsupported format. Use: srt, txt, json") diff --git a/app/api/routers/voiceprints.py b/app/api/routers/voiceprints.py new file mode 100644 index 0000000..5809e9e --- /dev/null +++ b/app/api/routers/voiceprints.py @@ -0,0 +1,94 @@ +"""Voiceprint management endpoints. + +All routes under /api/voiceprints/*. +""" + +import logging + +from fastapi import APIRouter, Form, HTTPException, Request + +from api.deps import get_db +from config import TRANSCRIPTIONS_DIR +from services.audio_service import safe_speaker_label, safe_tr_dir + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/api") + + +@router.post("/voiceprints/enroll") +async def enroll_speaker( + request: Request, + tr_id: str = Form(...), + speaker_label: str = Form(...), + speaker_name: str = Form(...), + speaker_id: str = Form(None), +): + """Enroll or update a voiceprint from a transcription's speaker embedding.""" + import numpy as np + + voiceprint_db = get_db(request) + + # SEC-C2: validate both tr_id and speaker_label before building any path. + safe_label = safe_speaker_label(speaker_label) + emb_path = safe_tr_dir(tr_id) / f"emb_{safe_label}.npy" + if not emb_path.exists(): + raise HTTPException(404, "Embedding not found for this speaker label") + # SEC-C1: allow_pickle=False prevents arbitrary code execution via + # a crafted .npy file that embeds a pickle payload (CVSS 9.1). + embedding = np.load(emb_path, allow_pickle=False) + + if speaker_id and voiceprint_db.get_speaker(speaker_id): + voiceprint_db.update_speaker(speaker_id, embedding, name=speaker_name) + return {"action": "updated", "speaker_id": speaker_id} + else: + new_id = voiceprint_db.add_speaker(speaker_name, embedding) + return {"action": "created", "speaker_id": new_id} + + +@router.get("/voiceprints") +async def list_voiceprints(request: Request): + return get_db(request).list_speakers() + + +@router.post("/voiceprints/rebuild-cohort") +async def rebuild_cohort(request: Request): + """Rebuild the AS-norm cohort from all processed transcriptions.""" + voiceprint_db = get_db(request) + cohort_path = TRANSCRIPTIONS_DIR / "asnorm_cohort.npy" + n = voiceprint_db.build_cohort_from_transcriptions( + str(TRANSCRIPTIONS_DIR), save_path=str(cohort_path) + ) + # [CQ-M10] 报告跳过/损坏的文件数，让调用方看到 cohort 的实际覆盖情况 + skipped = getattr(voiceprint_db, "last_cohort_skipped", 0) + return { + "cohort_size": n, + "skipped": skipped, + "saved_to": str(cohort_path), + } + + +@router.get("/voiceprints/{speaker_id}") +async def get_voiceprint(speaker_id: str, request: Request): + speaker = get_db(request).get_speaker(speaker_id) + if not speaker: + raise HTTPException(404, "Speaker not found") + return speaker + + +@router.delete("/voiceprints/{speaker_id}") +async def delete_voiceprint(speaker_id: str, request: Request): + try: + get_db(request).delete_speaker(speaker_id) + except ValueError as e: + raise HTTPException(404, str(e)) + return {"ok": True} + + +@router.put("/voiceprints/{speaker_id}/name") +async def rename_voiceprint(speaker_id: str, request: Request, name: str = Form(...)): + try: + get_db(request).rename_speaker(speaker_id, name) + except ValueError as e: + raise HTTPException(404, str(e)) + return {"ok": True} diff --git a/app/config.py b/app/config.py new file mode 100644 index 0000000..5208bfc --- /dev/null +++ b/app/config.py @@ -0,0 +1,101 @@ +"""Centralised configuration — all os.getenv() calls live here. + +Modules import from this file rather than calling os.getenv() directly so +that environment-variable names are defined in exactly one place and are +easy to audit. +""" + +import os +from pathlib import Path + + +def _env_float(name: str, default: float) -> float: + try: + return float(os.getenv(name, str(default))) + except ValueError: + return default + + +# --------------------------------------------------------------------------- +# Directory layout +# --------------------------------------------------------------------------- + +DATA_DIR: Path = Path(os.getenv("DATA_DIR", "/data")) +TRANSCRIPTIONS_DIR: Path = DATA_DIR / "transcriptions" +UPLOADS_DIR: Path = DATA_DIR / "uploads" +VOICEPRINTS_DIR: Path = DATA_DIR / "voiceprints" +MODELS_DIR: Path = Path(os.getenv("MODELS_DIR", "/models")) + +# --------------------------------------------------------------------------- +# Auth / CORS +# --------------------------------------------------------------------------- + +API_KEY: str | None = (os.getenv("API_KEY") or "").strip() or None + +# SEC-C3: allow operators to explicitly acknowledge running without auth. +# When ALLOW_NO_AUTH=1 the warning is suppressed; the service still runs open. +ALLOW_NO_AUTH: bool = os.getenv("ALLOW_NO_AUTH", "0") == "1" + +CORS_ORIGINS: str = os.getenv("CORS_ALLOW_ORIGINS", "*").strip() + +# --------------------------------------------------------------------------- +# Upload limits +# --------------------------------------------------------------------------- + +# Cap how much any single upload can occupy on disk. Whisper + pyannote +# comfortably handle 2 GB of audio (~20 h @ typical bitrates); anything +# beyond that is either a mistake or an attempt to exhaust storage. +MAX_UPLOAD_BYTES: int = int(os.getenv("MAX_UPLOAD_BYTES", str(2 * 1024 * 1024 * 1024))) +UPLOAD_CHUNK: int = 1 << 20 # 1 MiB + +# --------------------------------------------------------------------------- +# Model / inference settings +# --------------------------------------------------------------------------- + +WHISPER_MODEL: str = os.getenv("WHISPER_MODEL", "large-v3") +HF_TOKEN: str | None = os.getenv("HF_TOKEN") +DEVICE: str = os.getenv("DEVICE", "cuda") +LANGUAGE: str = os.getenv("LANGUAGE", "") + +# --------------------------------------------------------------------------- +# Denoising +# --------------------------------------------------------------------------- + +DENOISE_MODEL: str = os.getenv("DENOISE_MODEL", "none").strip().lower() + +# SNR threshold (dB) below which DeepFilterNet is applied. +# Audio estimated at or above this level is considered clean and skipped, +# matching the A/B finding that DF hurts high-quality recordings (e.g. PLAUD Pin). +DENOISE_SNR_THRESHOLD: float = _env_float("DENOISE_SNR_THRESHOLD", 10.0) + +# --------------------------------------------------------------------------- +# Speaker identification +# --------------------------------------------------------------------------- + +# Base cosine-similarity threshold for voiceprint identify(). The actual +# threshold per candidate is adaptive — see voiceprint_db.identify's docstring +# for the per-speaker relaxation rules. +VOICEPRINT_THRESHOLD: float = _env_float("VOICEPRINT_THRESHOLD", 0.75) + +# --------------------------------------------------------------------------- +# Misc +# --------------------------------------------------------------------------- + +FFMPEG_TIMEOUT_SEC: int = int(os.getenv("FFMPEG_TIMEOUT_SEC", "1800")) +JOBS_MAX_CACHE: int = int(os.getenv("JOBS_MAX_CACHE", "200")) + +# Paths that must stay open even when API_KEY auth is enabled. "/" is the +# bundled web UI (browsers can't attach a Bearer header to a direct +# navigation — the UI's own fetch() calls to /api/* still carry the key). +# /static/* serves the UI's assets. /healthz is a liveness probe. /docs +# /redoc /openapi.json are FastAPI's auto docs. +PUBLIC_EXACT_PATHS: frozenset = frozenset( + { + "/", + "/healthz", + "/docs", + "/redoc", + "/openapi.json", + } +) +PUBLIC_PATH_PREFIXES: tuple = ("/static/",) diff --git a/app/main.py b/app/main.py index cf0fff5..c65d61e 100755 --- a/app/main.py +++ b/app/main.py @@ -1,26 +1,29 @@ """FastAPI service for voice transcription with speaker identification.""" -import hashlib import hmac -import json -import os -import subprocess -import uuid import logging -from datetime import datetime -from pathlib import Path, PurePosixPath -from threading import Thread +from contextlib import asynccontextmanager -from fastapi import FastAPI, File, Form, UploadFile, HTTPException, Request -from fastapi.responses import ( - FileResponse, - HTMLResponse, - JSONResponse, - PlainTextResponse, -) +from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import JSONResponse from fastapi.staticfiles import StaticFiles +from api.routers import health, transcriptions, voiceprints +from services.job_service import recover_orphan_jobs +from config import ( + ALLOW_NO_AUTH, + API_KEY, + CORS_ORIGINS, + HF_TOKEN, + PUBLIC_EXACT_PATHS, + PUBLIC_PATH_PREFIXES, + TRANSCRIPTIONS_DIR, + UPLOADS_DIR, + VOICEPRINTS_DIR, + WHISPER_MODEL, + DEVICE, +) from pipeline import TranscriptionPipeline from voiceprint_db import VoiceprintDB @@ -29,201 +32,96 @@ ) logger = logging.getLogger(__name__) -DATA_DIR = Path(os.getenv("DATA_DIR", "/data")) -TRANSCRIPTIONS_DIR = DATA_DIR / "transcriptions" -UPLOADS_DIR = DATA_DIR / "uploads" -VOICEPRINTS_DIR = DATA_DIR / "voiceprints" - - -# Base cosine-similarity threshold for voiceprint identify(). The actual -# threshold per candidate is adaptive — see voiceprint_db.identify's docstring -# for the per-speaker relaxation rules. -def _env_float(name: str, default: float) -> float: - try: - return float(os.getenv(name, str(default))) - except ValueError: - return default - - -VOICEPRINT_THRESHOLD = _env_float("VOICEPRINT_THRESHOLD", 0.75) - -DENOISE_MODEL = os.getenv("DENOISE_MODEL", "none").strip().lower() - -# SNR threshold (dB) below which DeepFilterNet is applied. -# Audio estimated at or above this level is considered clean and skipped, -# matching the A/B finding that DF hurts high-quality recordings (e.g. PLAUD Pin). -DENOISE_SNR_THRESHOLD = _env_float("DENOISE_SNR_THRESHOLD", 10.0) - -# Lazy module-level handle so DeepFilterNet loads once at first use. -_df_model = None -_df_state = None - - -def _load_deepfilternet(): - global _df_model, _df_state - if _df_model is None: - import os, shutil - - if not shutil.which("git"): - fake = "/tmp/_fake_git" - if not os.path.exists(fake): - with open(fake, "w") as f: - f.write("#!/bin/sh\necho unknown\nexit 0\n") - os.chmod(fake, 0o755) - os.environ["PATH"] = "/tmp:" + os.environ.get("PATH", "") - import df as _df_pkg - - _df_model, _df_state, _ = _df_pkg.init_df() - logger.info("DeepFilterNet model loaded") - return _df_model, _df_state - - -def _estimate_snr(wav_path: Path) -> float: - """Estimate signal-to-noise ratio (dB) using a simple energy-based heuristic. - - Strategy: divide the audio into short frames, compute per-frame RMS energy, - then treat the bottom 20 % of frame energies as the noise floor and the top - 80 % as the speech signal. SNR = 10 * log10(speech_power / noise_power). - This is intentionally lightweight — no VAD model, no STFT — so it adds - negligible latency before deciding whether to invoke DeepFilterNet. - """ - import math - import torchaudio +# --------------------------------------------------------------------------- +# Lifespan: startup / teardown +# --------------------------------------------------------------------------- - waveform, sr = torchaudio.load(str(wav_path)) - # Flatten to mono - if waveform.shape[0] > 1: - waveform = waveform.mean(dim=0, keepdim=True) - waveform = waveform.squeeze(0) # shape: (num_samples,) - # 30 ms frames - frame_len = max(1, int(sr * 0.03)) - num_frames = len(waveform) // frame_len - if num_frames < 5: - # Too short to estimate reliably — assume clean - return float("inf") +@asynccontextmanager +async def lifespan(app: FastAPI): + # Ensure data directories exist + for d in [TRANSCRIPTIONS_DIR, UPLOADS_DIR, VOICEPRINTS_DIR]: + d.mkdir(parents=True, exist_ok=True) - frames = waveform[: num_frames * frame_len].reshape(num_frames, frame_len) - frame_rms = frames.pow(2).mean(dim=1).sqrt() # shape: (num_frames,) + # AR-C2: mark any in-progress jobs from a previous process as failed so + # frontend polls receive a definitive terminal state on restart. + recover_orphan_jobs() - sorted_rms, _ = frame_rms.sort() - noise_cutoff = max(1, int(num_frames * 0.20)) - noise_rms = sorted_rms[:noise_cutoff].mean().item() - speech_rms = sorted_rms[noise_cutoff:].mean().item() - - if noise_rms < 1e-9: - return float("inf") # Silent noise floor — effectively infinite SNR - - snr_db = 10.0 * math.log10((speech_rms / noise_rms) ** 2) - return snr_db - - -def _maybe_denoise( - wav_path: Path, model: str = None, snr_threshold: float = None -) -> Path: - """Return denoised WAV path if DENOISE_MODEL is set; otherwise return wav_path unchanged.""" - effective_model = (model or DENOISE_MODEL).strip().lower() - if effective_model == "none": - return wav_path - - threshold = snr_threshold if snr_threshold is not None else DENOISE_SNR_THRESHOLD - out_path = wav_path.with_suffix(".denoised.wav") - - if effective_model == "deepfilternet": - import torch, torchaudio + # Initialise voiceprint DB and AS-norm cohort + db = VoiceprintDB(str(VOICEPRINTS_DIR)) + try: + _cohort_path = TRANSCRIPTIONS_DIR / "asnorm_cohort.npy" + if _cohort_path.exists(): + db.load_cohort(str(_cohort_path)) + logger.info("AS-norm cohort loaded from %s", _cohort_path) + else: + _n = db.build_cohort_from_transcriptions( + str(TRANSCRIPTIONS_DIR), save_path=str(_cohort_path) + ) + logger.info("AS-norm cohort built: %d embeddings", _n) + except Exception as exc: + logger.warning( + "AS-norm cohort init failed (identify will use raw cosine): %s", exc + ) + app.state.db = db - snr_db = _estimate_snr(wav_path) - if snr_db >= threshold: - logger.info("DeepFilterNet skipped (SNR=%.1fdB, clean audio)", snr_db) - return wav_path + # Initialise transcription pipeline + app.state.pipeline = TranscriptionPipeline(WHISPER_MODEL, DEVICE, HF_TOKEN) - logger.info( - "DeepFilterNet applying (SNR=%.1fdB < %.1fdB threshold)", - snr_db, - threshold, + # Auth mode warning + if API_KEY is None and not ALLOW_NO_AUTH: + logger.warning( + "API_KEY is not set. Service is OPEN to all requests. " + "Set API_KEY env var or set ALLOW_NO_AUTH=1 to suppress this warning." ) - model, df_state = _load_deepfilternet() - import df as _df_pkg - - audio, sr = torchaudio.load(str(wav_path)) - if sr != df_state.sr(): - audio = torchaudio.functional.resample(audio, sr, df_state.sr()) - audio = audio.contiguous() - with torch.backends.cudnn.flags(enabled=False): - enhanced = _df_pkg.enhance(model, df_state, audio) - torchaudio.save( - str(out_path), - enhanced.unsqueeze(0) if enhanced.dim() == 1 else enhanced, - df_state.sr(), + elif API_KEY is None: + logger.warning( + "API_KEY is not set and ALLOW_NO_AUTH=1. " + "The service is accepting unauthenticated requests intentionally." ) - logger.info("DeepFilterNet: denoised %s → %s", wav_path.name, out_path.name) - - elif effective_model == "noisereduce": - import numpy as np, soundfile as sf, noisereduce as nr - - data, sr = sf.read(str(wav_path), dtype="float32") - reduced = nr.reduce_noise(y=data, sr=sr, stationary=True) - sf.write(str(out_path), reduced, sr) - logger.info("noisereduce: denoised %s → %s", wav_path.name, out_path.name) - else: - logger.warning("Unknown DENOISE_MODEL=%r — skipping denoising", effective_model) - return wav_path - - return out_path - - -API_KEY = (os.getenv("API_KEY") or "").strip() or None -# Paths that must stay open even when API_KEY auth is enabled. "/" is the -# bundled web UI (browsers can't attach a Bearer header to a direct -# navigation — the UI's own fetch() calls to /api/* still carry the key). -# /static/* serves the UI's assets. /healthz is a liveness probe. /docs -# /redoc /openapi.json are FastAPI's auto docs. -# We match exact strings for everything except /static/ to avoid a -# startswith("/docs") bypass like /docsXYZ. -PUBLIC_EXACT_PATHS = { - "/", - "/healthz", - "/docs", - "/redoc", - "/openapi.json", -} -PUBLIC_PATH_PREFIXES = ("/static/",) + logger.info("API_KEY auth enabled for /api/* and / (Bearer or X-API-Key).") -# Cap how much any single upload can occupy on disk. Whisper + pyannote -# comfortably handle 2 GB of audio (~20 h @ typical bitrates); anything -# beyond that is either a mistake or an attempt to exhaust storage. -MAX_UPLOAD_BYTES = int(os.getenv("MAX_UPLOAD_BYTES", str(2 * 1024 * 1024 * 1024))) -UPLOAD_CHUNK = 1 << 20 # 1 MiB + yield + # No teardown required; daemon threads finish on process exit. -for d in [TRANSCRIPTIONS_DIR, UPLOADS_DIR, VOICEPRINTS_DIR]: - d.mkdir(parents=True, exist_ok=True) -if API_KEY is None: - logger.warning( - "API_KEY is not set. The service is accepting unauthenticated requests. " - "Do not expose this port to untrusted networks." - ) -else: - logger.info("API_KEY auth enabled for /api/* and / (Bearer or X-API-Key).") +# --------------------------------------------------------------------------- +# Application +# --------------------------------------------------------------------------- -app = FastAPI(title="Voice Transcribe", version="1.0.0") +app = FastAPI(title="VoScript", version="0.7.0", lifespan=lifespan) -_cors_origins_env = os.getenv("CORS_ALLOW_ORIGINS", "*").strip() -_cors_origins = [o.strip() for o in _cors_origins_env.split(",") if o.strip()] or ["*"] +# CORS +_cors_origins = [o.strip() for o in CORS_ORIGINS.split(",") if o.strip()] or ["*"] app.add_middleware( CORSMiddleware, allow_origins=_cors_origins, allow_credentials=False, allow_methods=["*"], - allow_headers=["*"], + allow_headers=[ + "Authorization", + "Content-Type", + "X-API-Key", + "X-Request-Id", + ], expose_headers=["*"], ) app.mount("/static", StaticFiles(directory="static"), name="static") +@app.middleware("http") +async def add_security_headers(request: Request, call_next): + response = await call_next(request) + response.headers.setdefault("X-Content-Type-Options", "nosniff") + response.headers.setdefault("X-Frame-Options", "DENY") + response.headers.setdefault("Referrer-Policy", "strict-origin-when-cross-origin") + response.headers.setdefault("X-XSS-Protection", "1; mode=block") + return response + + @app.middleware("http") async def require_api_key(request: Request, call_next): if API_KEY is None: @@ -259,505 +157,10 @@ async def require_api_key(request: Request, call_next): return await call_next(request) -@app.get("/healthz") -async def healthz(): - return {"ok": True} - - -pipeline = TranscriptionPipeline() -voiceprint_db = VoiceprintDB(str(VOICEPRINTS_DIR)) - -# Auto-build or load AS-norm cohort from existing transcriptions. This lets -# identify() use normalized scores instead of raw cosine against speaker- -# dependent baselines. Failure is non-fatal — we fall back to raw cosine. -try: - _cohort_path = TRANSCRIPTIONS_DIR / "asnorm_cohort.npy" - if _cohort_path.exists(): - voiceprint_db.load_cohort(str(_cohort_path)) - logger.info("AS-norm cohort loaded from %s", _cohort_path) - else: - _n = voiceprint_db.build_cohort_from_transcriptions( - str(TRANSCRIPTIONS_DIR), save_path=str(_cohort_path) - ) - logger.info("AS-norm cohort built: %d embeddings", _n) -except Exception as _exc: - logger.warning( - "AS-norm cohort init failed (identify will use raw cosine): %s", _exc - ) - -# In-memory job status -jobs: dict[str, dict] = {} - -# Serialise GPU work: only one transcription runs at a time. -# Concurrent HTTP uploads are fine; they queue here before touching the GPU. -import threading as _threading - -_gpu_sem = _threading.Semaphore(1) - - -def _convert_to_wav(input_path: Path) -> Path: - """Convert any audio format to 16 kHz mono WAV via ffmpeg. - - We shell out to ffmpeg directly instead of using pydub because pydub's - mediainfo_json() raises KeyError('codec_type') on newer ffmpeg output - for some Opus/container combinations (see jiaaro/pydub#638). ffmpeg - itself handles every format faster-whisper / pyannote ingest, so this - is the simpler and more robust path. - """ - wav_path = input_path.with_suffix(".wav") - if input_path.suffix.lower() == ".wav": - return input_path - # "--" closes ffmpeg's option parsing so a filename like `-foo.mp4` - # can't be interpreted as a flag. Defense in depth — the upload path - # already strips client-side directory components and prefixes the - # job_id, so input_path always starts with /data/uploads/tr_... - subprocess.run( - [ - "ffmpeg", - "-y", - "-v", - "error", - "-i", - str(input_path), - "-ar", - "16000", - "-ac", - "1", - "-f", - "wav", - "--", - str(wav_path), - ], - check=True, - ) - return wav_path - - -_HASH_INDEX_FILE = TRANSCRIPTIONS_DIR / "hash_index.json" -_hash_index_lock = __import__("threading").Lock() - - -def _compute_file_hash(path: Path) -> str: - sha256 = hashlib.sha256() - with open(path, "rb") as f: - while chunk := f.read(1 << 20): - sha256.update(chunk) - return sha256.hexdigest() - - -def _lookup_hash(file_hash: str) -> str | None: - """Return existing tr_id if hash is already transcribed and result exists.""" - with _hash_index_lock: - if not _HASH_INDEX_FILE.exists(): - return None - index = json.loads(_HASH_INDEX_FILE.read_text()) - tr_id = index.get(file_hash) - if tr_id and (TRANSCRIPTIONS_DIR / tr_id / "result.json").exists(): - return tr_id - return None - - -def _register_hash(file_hash: str, tr_id: str) -> None: - with _hash_index_lock: - index = ( - json.loads(_HASH_INDEX_FILE.read_text()) - if _HASH_INDEX_FILE.exists() - else {} - ) - index[file_hash] = tr_id - _HASH_INDEX_FILE.write_text(json.dumps(index, indent=2)) - - -def _run_transcription( - job_id: str, - audio_path: Path, - language: str, - min_speakers: int, - max_speakers: int, - denoise_model: str = None, - snr_threshold: float = None, - file_hash: str = None, -): - """Background transcription worker.""" - try: - jobs[job_id]["status"] = "converting" - wav_path = _convert_to_wav(audio_path) - - jobs[job_id]["status"] = "queued" - with _gpu_sem: - jobs[job_id]["status"] = ( - "denoising" - if (denoise_model or DENOISE_MODEL) != "none" - else "transcribing" - ) - clean_path = _maybe_denoise(wav_path, denoise_model, snr_threshold) - - # DF peaks at ~15 GB reserved in PyTorch's CUDA cache. - # ctranslate2 (Whisper) calls cudaMalloc directly and sees the OS - # free memory — not PyTorch's allocator pool — so it OOMs unless we - # explicitly flush the cache before Whisper cold-loads. - try: - import torch as _torch - import gc as _gc - - _gc.collect() - if _torch.cuda.is_available(): - _torch.cuda.empty_cache() - except Exception: - pass - - jobs[job_id]["status"] = "transcribing" - result = pipeline.process( - str(clean_path), - raw_audio_path=str(wav_path), - language=language, - min_speakers=min_speakers or None, - max_speakers=max_speakers or None, - ) - - # Release cached CUDA memory so the next queued job has headroom - try: - import torch as _torch - import gc as _gc - - _gc.collect() - if _torch.cuda.is_available(): - _torch.cuda.empty_cache() - except Exception: - pass - - # Match speakers against voiceprint DB - jobs[job_id]["status"] = "identifying" - speaker_map = {} - for spk_label, embedding in result["speaker_embeddings"].items(): - spk_id, spk_name, sim = voiceprint_db.identify( - embedding, threshold=VOICEPRINT_THRESHOLD - ) - speaker_map[spk_label] = { - "matched_id": spk_id, - "matched_name": spk_name or spk_label, - "similarity": round(sim, 4), - "embedding_key": spk_label, - } - - # Build final segments - segments = [] - for i, seg in enumerate(result["segments"]): - spk_label = seg["speaker"] - match = speaker_map.get(spk_label, {}) - out = { - "id": i, - "start": seg["start"], - "end": seg["end"], - "text": seg["text"], - "speaker_label": spk_label, - "speaker_id": match.get("matched_id"), - "speaker_name": match.get("matched_name", spk_label), - "similarity": match.get("similarity", 0), - } - # Forward word-level timestamps when forced alignment produced them - # (0.3.0+). Absent when the language has no alignment model or - # alignment failed — clients must treat the key as optional. - if seg.get("words"): - out["words"] = seg["words"] - segments.append(out) - - # Save transcription result - effective_denoise = (denoise_model or DENOISE_MODEL).strip().lower() - effective_snr = ( - snr_threshold if snr_threshold is not None else DENOISE_SNR_THRESHOLD - ) - tr = { - "id": job_id, - "filename": audio_path.name, - "created_at": datetime.now().isoformat(), - "status": "completed", - "language": language, - "segments": segments, - "speaker_map": speaker_map, - "unique_speakers": result["unique_speakers"], - "params": { - "language": language or "auto", - "denoise_model": effective_denoise, - "snr_threshold": effective_snr, - "voiceprint_threshold": VOICEPRINT_THRESHOLD, - "min_speakers": min_speakers, - "max_speakers": max_speakers, - }, - } - - tr_dir = TRANSCRIPTIONS_DIR / job_id - tr_dir.mkdir(exist_ok=True) - (tr_dir / "result.json").write_text( - json.dumps(tr, ensure_ascii=False, indent=2), encoding="utf-8" - ) - - # Save raw embeddings for later enrollment - import numpy as np - - for spk_label, emb in result["speaker_embeddings"].items(): - np.save(tr_dir / f"emb_{spk_label}.npy", emb) - - if file_hash: - _register_hash(file_hash, job_id) - - jobs[job_id]["status"] = "completed" - jobs[job_id]["result"] = tr - logger.info( - "Job %s completed: %d segments, %d speakers", - job_id, - len(segments), - len(speaker_map), - ) - - except Exception as e: - logger.exception("Job %s failed", job_id) - jobs[job_id]["status"] = "failed" - jobs[job_id]["error"] = str(e) - - -# --- Routes --- - - -@app.get("/", response_class=HTMLResponse) -async def index(): - return (Path("static/index.html")).read_text(encoding="utf-8") - - -@app.post("/api/transcribe") -async def transcribe( - file: UploadFile = File(...), - language: str = Form(None), - min_speakers: int = Form(0), - max_speakers: int = Form(0), - denoise_model: str = Form("none"), - snr_threshold: float = Form(None), -): - # Normalise empty string to None so pipeline treats it as auto-detect. - language = language.strip() if language else None - - job_id = f"tr_{datetime.now():%Y%m%d_%H%M%S}_{uuid.uuid4().hex[:6]}" - - safe_filename = PurePosixPath(file.filename or "upload").name or "upload" - save_path = UPLOADS_DIR / f"{job_id}_{safe_filename}" - - size = 0 - with open(save_path, "wb") as f: - while True: - chunk = file.file.read(UPLOAD_CHUNK) - if not chunk: - break - size += len(chunk) - if size > MAX_UPLOAD_BYTES: - f.close() - save_path.unlink(missing_ok=True) - raise HTTPException( - 413, - f"Upload exceeds MAX_UPLOAD_BYTES ({MAX_UPLOAD_BYTES} bytes)", - ) - f.write(chunk) - - # Dedup: if identical audio was already transcribed, return existing result. - file_hash = _compute_file_hash(save_path) - existing_id = _lookup_hash(file_hash) - if existing_id: - save_path.unlink(missing_ok=True) - logger.info( - "Dedup hit: %s already transcribed as %s", safe_filename, existing_id - ) - return {"id": existing_id, "status": "completed", "deduplicated": True} - - jobs[job_id] = { - "status": "queued", - "filename": safe_filename, - "created_at": datetime.now().isoformat(), - } - thread = Thread( - target=_run_transcription, - args=( - job_id, - save_path, - language, - min_speakers, - max_speakers, - denoise_model, - snr_threshold, - file_hash, - ), - ) - thread.start() - - return {"id": job_id, "status": "queued"} - - -@app.get("/api/jobs/{job_id}") -async def get_job(job_id: str): - if job_id not in jobs: - raise HTTPException(404, "Job not found") - job = jobs[job_id] - resp = {"id": job_id, "status": job["status"], "filename": job.get("filename")} - if job["status"] == "completed": - resp["result"] = job["result"] - elif job["status"] == "failed": - resp["error"] = job.get("error") - return resp - - -@app.get("/api/transcriptions") -async def list_transcriptions(): - results = [] - for tr_dir in sorted(TRANSCRIPTIONS_DIR.iterdir(), reverse=True): - result_file = tr_dir / "result.json" - if result_file.exists(): - data = json.loads(result_file.read_text(encoding="utf-8")) - results.append( - { - "id": data["id"], - "filename": data["filename"], - "created_at": data["created_at"], - "segment_count": len(data["segments"]), - "speaker_count": len(data.get("unique_speakers", [])), - } - ) - return results - - -@app.get("/api/transcriptions/{tr_id}") -async def get_transcription(tr_id: str): - result_file = TRANSCRIPTIONS_DIR / tr_id / "result.json" - if not result_file.exists(): - raise HTTPException(404, "Transcription not found") - return json.loads(result_file.read_text(encoding="utf-8")) - - -@app.put("/api/transcriptions/{tr_id}/segments/{seg_id}/speaker") -async def reassign_speaker( - tr_id: str, seg_id: int, speaker_name: str = Form(...), speaker_id: str = Form(None) -): - """Reassign a segment to a different speaker and optionally enroll the voiceprint.""" - result_file = TRANSCRIPTIONS_DIR / tr_id / "result.json" - if not result_file.exists(): - raise HTTPException(404, "Transcription not found") - data = json.loads(result_file.read_text(encoding="utf-8")) - - seg = next((s for s in data["segments"] if s["id"] == seg_id), None) - if seg is None: - raise HTTPException(404, "Segment not found") - - seg["speaker_name"] = speaker_name - if speaker_id: - seg["speaker_id"] = speaker_id - - result_file.write_text( - json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8" - ) - return {"ok": True} - - -@app.post("/api/voiceprints/enroll") -async def enroll_speaker( - tr_id: str = Form(...), - speaker_label: str = Form(...), - speaker_name: str = Form(...), - speaker_id: str = Form(None), -): - """Enroll or update a voiceprint from a transcription's speaker embedding.""" - import numpy as np - - emb_path = TRANSCRIPTIONS_DIR / tr_id / f"emb_{speaker_label}.npy" - if not emb_path.exists(): - raise HTTPException(404, "Embedding not found for this speaker label") - embedding = np.load(emb_path) - - if speaker_id and voiceprint_db.get_speaker(speaker_id): - voiceprint_db.update_speaker(speaker_id, embedding, name=speaker_name) - return {"action": "updated", "speaker_id": speaker_id} - else: - new_id = voiceprint_db.add_speaker(speaker_name, embedding) - return {"action": "created", "speaker_id": new_id} - - -@app.get("/api/voiceprints") -async def list_voiceprints(): - return voiceprint_db.list_speakers() - - -@app.post("/api/voiceprints/rebuild-cohort") -async def rebuild_cohort(): - """Rebuild the AS-norm cohort from all processed transcriptions.""" - cohort_path = TRANSCRIPTIONS_DIR / "asnorm_cohort.npy" - n = voiceprint_db.build_cohort_from_transcriptions( - str(TRANSCRIPTIONS_DIR), save_path=str(cohort_path) - ) - return {"cohort_size": n, "saved_to": str(cohort_path)} - - -@app.delete("/api/voiceprints/{speaker_id}") -async def delete_voiceprint(speaker_id: str): - try: - voiceprint_db.delete_speaker(speaker_id) - except ValueError as e: - raise HTTPException(404, str(e)) - return {"ok": True} - - -@app.put("/api/voiceprints/{speaker_id}/name") -async def rename_voiceprint(speaker_id: str, name: str = Form(...)): - try: - voiceprint_db.rename_speaker(speaker_id, name) - except ValueError as e: - raise HTTPException(404, str(e)) - return {"ok": True} - - -@app.get("/api/export/{tr_id}") -async def export_transcription(tr_id: str, format: str = "srt"): - result_file = TRANSCRIPTIONS_DIR / tr_id / "result.json" - if not result_file.exists(): - raise HTTPException(404, "Transcription not found") - data = json.loads(result_file.read_text(encoding="utf-8")) - segments = data["segments"] - - if format == "srt": - lines = [] - for i, seg in enumerate(segments, 1): - start = _format_srt_time(seg["start"]) - end = _format_srt_time(seg["end"]) - lines.append( - f"{i}\n{start} --> {end}\n[{seg['speaker_name']}] {seg['text']}\n" - ) - return PlainTextResponse( - "\n".join(lines), - media_type="text/srt", - headers={"Content-Disposition": f"attachment; filename={tr_id}.srt"}, - ) - elif format == "txt": - lines = [] - for seg in segments: - ts = _format_timestamp(seg["start"]) - lines.append(f"[{ts}] {seg['speaker_name']}: {seg['text']}") - return PlainTextResponse( - "\n".join(lines), - media_type="text/plain", - headers={"Content-Disposition": f"attachment; filename={tr_id}.txt"}, - ) - elif format == "json": - return FileResponse( - result_file, media_type="application/json", filename=f"{tr_id}.json" - ) - else: - raise HTTPException(400, "Unsupported format. Use: srt, txt, json") - - -def _format_srt_time(seconds: float) -> str: - h = int(seconds // 3600) - m = int((seconds % 3600) // 60) - s = int(seconds % 60) - ms = int((seconds % 1) * 1000) - return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}" - +# --------------------------------------------------------------------------- +# Routers +# --------------------------------------------------------------------------- -def _format_timestamp(seconds: float) -> str: - m = int(seconds // 60) - s = int(seconds % 60) - return f"{m:02d}:{s:02d}" +app.include_router(health.router) +app.include_router(transcriptions.router) +app.include_router(voiceprints.router) diff --git a/app/pipeline.py b/app/pipeline.py index bff48f7..5761cec 100755 --- a/app/pipeline.py +++ b/app/pipeline.py @@ -11,12 +11,19 @@ import os import logging +from pathlib import Path + import numpy as np import torch import torchaudio logger = logging.getLogger(__name__) +# WeSpeaker ResNet34 推荐输入 ≥1.5s；过短的 chunk 嵌入方差显著放大会污染 speaker_avg。 +# 上限避免超长 chunk 带来的显存浪费。两者均可通过环境变量覆盖。 +MIN_EMBED_DURATION = float(os.getenv("MIN_EMBED_DURATION", "1.5")) +MAX_EMBED_DURATION = float(os.getenv("MAX_EMBED_DURATION", "10.0")) + class TranscriptionPipeline: def __init__( @@ -44,7 +51,7 @@ def whisper(self): decoupled from the transcriber. """ if self._whisper is None: - from pathlib import Path + # faster_whisper 按需 lazy import，避免在不使用 whisper 的进程里加载 GPU 库 from faster_whisper import WhisperModel compute_type = "float16" if self.device == "cuda" else "int8" @@ -104,7 +111,9 @@ def embedding_model(self): self._embedding_model = Inference(model, window="whole") return self._embedding_model - def transcribe(self, audio_path: str, language: str = None) -> dict: + def transcribe( + self, audio_path: str, language: str = None, no_repeat_ngram_size: int = None + ) -> dict: """Run faster-whisper and return a whisperx-compatible result dict. whisperx.align expects ``{"segments": [...], "language": "..."}`` with @@ -124,14 +133,16 @@ def transcribe(self, audio_path: str, language: str = None) -> dict: lang_arg or "auto", ) - segments_iter, info = self.whisper.transcribe( - audio_path, + whisper_kwargs = dict( language=lang_arg, beam_size=5, vad_filter=True, vad_parameters=dict(min_silence_duration_ms=500), initial_prompt=initial_prompt, ) + if no_repeat_ngram_size and no_repeat_ngram_size >= 3: + whisper_kwargs["no_repeat_ngram_size"] = no_repeat_ngram_size + segments_iter, info = self.whisper.transcribe(audio_path, **whisper_kwargs) segments = [ { "start": round(float(s.start), 3), @@ -178,22 +189,59 @@ def extract_speaker_embeddings( WeSpeaker ResNet34 produces ~256-dim embeddings (vs ECAPA-TDNN 192-dim). The downstream VoiceprintDB is dim-agnostic and infers the dimension on first insert, so no other changes are required. + + PERF-H1: segments are loaded on-demand via torchaudio.load(frame_offset, + num_frames) instead of loading the entire file into memory. A 2-hour + WAV at 16 kHz mono is ~900 MB–2 GB; with segment-level loading the peak + allocation per iteration is bounded by MAX_EMBED_DURATION * sr * 4 bytes + (~640 KB at 16 kHz / 10 s), a >1000x reduction for long recordings. """ - waveform, sr = torchaudio.load(audio_path) - if sr != 16000: - waveform = torchaudio.functional.resample(waveform, sr, 16000) - sr = 16000 - if waveform.shape[0] > 1: - waveform = waveform.mean(dim=0, keepdim=True) + # Obtain file metadata without decoding audio data (torchaudio >= 0.9). + info = torchaudio.info(audio_path) + native_sr = info.sample_rate + target_sr = 16000 + + min_samples = int(MIN_EMBED_DURATION * native_sr) + max_samples = int(MAX_EMBED_DURATION * native_sr) speaker_segments: dict[str, list] = {} for t in turns: spk = t["speaker"] - start_sample = int(t["start"] * sr) - end_sample = int(t["end"] * sr) - chunk = waveform[:, start_sample:end_sample] - if chunk.shape[1] < sr: # skip segments shorter than 1s + start_sample = int(t["start"] * native_sr) + end_sample = int(t["end"] * native_sr) + num_frames = end_sample - start_sample + + if num_frames < min_samples: continue + # 截断过长 chunk 以控制显存占用 + if num_frames > max_samples: + num_frames = max_samples + + # Load only the required segment — no whole-file decode. + try: + chunk, chunk_sr = torchaudio.load( + audio_path, + frame_offset=start_sample, + num_frames=num_frames, + ) + except Exception as e: + logger.warning( + "Failed to load segment %s [%d:%d]: %s", + spk, + start_sample, + end_sample, + e, + ) + continue + + # Resample to 16 kHz when the file's native rate differs. + if chunk_sr != target_sr: + chunk = torchaudio.functional.resample(chunk, chunk_sr, target_sr) + + # Downmix multi-channel audio to mono. + if chunk.shape[0] > 1: + chunk = chunk.mean(dim=0, keepdim=True) + speaker_segments.setdefault(spk, []).append(chunk) embeddings = {} @@ -205,7 +253,7 @@ def extract_speaker_embeddings( # Inference.__call__ accepts a dict with waveform (1, T) tensor # and sample_rate; window="whole" returns one ndarray per chunk. emb = self.embedding_model( - {"waveform": chunk.to(self.device), "sample_rate": 16000} + {"waveform": chunk.to(self.device), "sample_rate": target_sr} ) emb_list.append(np.asarray(emb)) if emb_list: @@ -338,6 +386,7 @@ def process( language: str = None, min_speakers: int = None, max_speakers: int = None, + no_repeat_ngram_size: int = None, ) -> dict: """Full pipeline: transcribe → diarize → forced-align → extract embeddings. @@ -348,7 +397,9 @@ def process( embed_path = raw_audio_path or audio_path logger.info("Starting transcription: %s", audio_path) - transcription_result = self.transcribe(audio_path, language=language) + transcription_result = self.transcribe( + audio_path, language=language, no_repeat_ngram_size=no_repeat_ngram_size + ) logger.info( "Transcription done: %d segments", len(transcription_result.get("segments", [])), diff --git a/app/services/__init__.py b/app/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/app/services/audio_service.py b/app/services/audio_service.py new file mode 100644 index 0000000..a56b6ea --- /dev/null +++ b/app/services/audio_service.py @@ -0,0 +1,330 @@ +"""Audio processing utilities. + +Covers format conversion (ffmpeg), content-addressed deduplication via a +SHA-256 hash index, and optional noise reduction (DeepFilterNet / noisereduce). +""" + +import fcntl +import hashlib +import json +import logging +import re +import subprocess +import threading +from pathlib import Path + +from fastapi import HTTPException + +from config import ( + DENOISE_MODEL, + DENOISE_SNR_THRESHOLD, + FFMPEG_TIMEOUT_SEC, + TRANSCRIPTIONS_DIR, +) + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Input-validation helpers (SEC-C2 / BP-C2) +# --------------------------------------------------------------------------- + +_TR_ID_RE = re.compile(r"^tr_[A-Za-z0-9_-]{1,64}$") +_SPEAKER_LABEL_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$") +_CTRL_CHAR_RE = re.compile(r"[\x00-\x1f\x7f]") + + +def safe_log_filename(name: str | None) -> str: + """Strip control chars (incl. CR/LF, ANSI escapes) from user-supplied names + before writing them to logs, so attackers can't inject fake log lines. + """ + if not name: + return "" + return _CTRL_CHAR_RE.sub("?", name) + + +def safe_tr_dir(tr_id: str) -> Path: + """Validate tr_id and return the transcription directory path. + + Raises HTTPException(400) if tr_id contains path traversal characters. + """ + if not _TR_ID_RE.match(tr_id): + raise HTTPException(400, f"Invalid transcription ID format: {tr_id!r}") + path = (TRANSCRIPTIONS_DIR / tr_id).resolve() + if not str(path).startswith(str(TRANSCRIPTIONS_DIR.resolve())): + raise HTTPException(400, "Path traversal detected") + return path + + +def safe_speaker_label(label: str) -> str: + """Validate speaker_label to prevent path traversal via filename injection.""" + if not _SPEAKER_LABEL_RE.match(label): + raise HTTPException(400, f"Invalid speaker label: {label!r}") + return label + + +# --------------------------------------------------------------------------- +# ffmpeg conversion +# --------------------------------------------------------------------------- + + +def convert_to_wav(input_path: Path) -> Path: + """Convert any audio format to 16 kHz mono WAV via ffmpeg. + + We shell out to ffmpeg directly instead of using pydub because pydub's + mediainfo_json() raises KeyError('codec_type') on newer ffmpeg output + for some Opus/container combinations (see jiaaro/pydub#638). ffmpeg + itself handles every format faster-whisper / pyannote ingest, so this + is the simpler and more robust path. + """ + wav_path = input_path.with_suffix(".wav") + if input_path.suffix.lower() == ".wav": + return input_path + # "--" closes ffmpeg's option parsing so a filename like `-foo.mp4` + # can't be interpreted as a flag. Defense in depth — the upload path + # already strips client-side directory components and prefixes the + # job_id, so input_path always starts with /data/uploads/tr_... + try: + subprocess.run( + [ + "ffmpeg", + "-y", + "-v", + "error", + "-i", + str(input_path), + "-ar", + "16000", + "-ac", + "1", + "-f", + "wav", + "--", + str(wav_path), + ], + check=True, + timeout=FFMPEG_TIMEOUT_SEC, + ) + except subprocess.TimeoutExpired: + wav_path.unlink(missing_ok=True) + logger.error( + "ffmpeg timed out after %ds on %s", FFMPEG_TIMEOUT_SEC, input_path.name + ) + raise HTTPException(504, f"ffmpeg timed out after {FFMPEG_TIMEOUT_SEC}s") + return wav_path + + +# --------------------------------------------------------------------------- +# Content-addressed hash index +# --------------------------------------------------------------------------- + +_HASH_INDEX_FILE = TRANSCRIPTIONS_DIR / "hash_index.json" +# CQ-H5: threading.Lock only works within a single process. Replace with an +# fcntl-based file lock so multiple uvicorn workers can safely share the index. +_hash_index_thread_lock = threading.Lock() # intra-process guard (belt) + + +def _with_file_lock(path: Path, func): + """Execute *func* while holding an exclusive fcntl lock on *path*.lock. + + Falls back to the in-process threading lock on platforms without fcntl + (e.g. Windows). The thread lock is always acquired first so that two + threads in the same process don't race through the fcntl acquire. + """ + lock_path = str(path) + ".lock" + with _hash_index_thread_lock: + try: + with open(lock_path, "w") as lock_f: + try: + fcntl.flock(lock_f.fileno(), fcntl.LOCK_EX) + return func() + finally: + fcntl.flock(lock_f.fileno(), fcntl.LOCK_UN) + except (AttributeError, OSError): + # fcntl unavailable (Windows) or lock file can't be opened — the + # thread lock we already hold is sufficient for single-process use. + return func() + + +def compute_file_hash(path: Path) -> str: + sha256 = hashlib.sha256() + with open(path, "rb") as f: + while chunk := f.read(1 << 20): + sha256.update(chunk) + return sha256.hexdigest() + + +async def save_upload_and_hash( + file, save_path: Path, max_bytes: int, chunk_size: int +) -> tuple[int, str]: + """Save an uploaded UploadFile to *save_path* using async I/O and compute + its SHA-256 digest on the fly, avoiding a second full-file read. + + Returns (total_bytes_written, hex_digest). + + Raises ValueError when the upload exceeds *max_bytes* — the caller is + responsible for unlinking *save_path* and returning HTTP 413. + + PERF-C2: replaces the former synchronous open()+write() loop and the + separate compute_file_hash() call, both of which blocked the asyncio + event loop for several seconds on large files. + """ + import aiofiles + + sha256 = hashlib.sha256() + size = 0 + async with aiofiles.open(save_path, "wb") as f: + while True: + chunk = await file.read(chunk_size) + if not chunk: + break + size += len(chunk) + if size > max_bytes: + raise ValueError(f"Upload exceeds MAX_UPLOAD_BYTES ({max_bytes} bytes)") + await f.write(chunk) + sha256.update(chunk) + return size, sha256.hexdigest() + + +def lookup_hash(file_hash: str) -> str | None: + """Return existing tr_id if hash is already transcribed and result exists.""" + + def _do(): + if not _HASH_INDEX_FILE.exists(): + return None + return json.loads(_HASH_INDEX_FILE.read_text()).get(file_hash) + + tr_id = _with_file_lock(_HASH_INDEX_FILE, _do) + if tr_id and (TRANSCRIPTIONS_DIR / tr_id / "result.json").exists(): + return tr_id + return None + + +def register_hash(file_hash: str, tr_id: str) -> None: + def _do(): + index = ( + json.loads(_HASH_INDEX_FILE.read_text()) + if _HASH_INDEX_FILE.exists() + else {} + ) + index[file_hash] = tr_id + _HASH_INDEX_FILE.write_text(json.dumps(index, indent=2)) + + _with_file_lock(_HASH_INDEX_FILE, _do) + + +# --------------------------------------------------------------------------- +# DeepFilterNet / noise reduction +# --------------------------------------------------------------------------- + +# Lazy module-level handle so DeepFilterNet loads once at first use. +_df_model = None +_df_state = None + + +def _load_deepfilternet(): + global _df_model, _df_state + if _df_model is None: + import df as _df_pkg + + _df_model, _df_state, _ = _df_pkg.init_df() + logger.info("DeepFilterNet model loaded") + return _df_model, _df_state + + +def _estimate_snr(wav_path: Path) -> float: + """Estimate signal-to-noise ratio (dB) using a simple energy-based heuristic. + + Strategy: divide the audio into short frames, compute per-frame RMS energy, + then treat the bottom 20 % of frame energies as the noise floor and the top + 80 % as the speech signal. SNR = 10 * log10(speech_power / noise_power). + + This is intentionally lightweight — no VAD model, no STFT — so it adds + negligible latency before deciding whether to invoke DeepFilterNet. + """ + import math + + import torchaudio + + waveform, sr = torchaudio.load(str(wav_path)) + # Flatten to mono + if waveform.shape[0] > 1: + waveform = waveform.mean(dim=0, keepdim=True) + waveform = waveform.squeeze(0) # shape: (num_samples,) + + # 30 ms frames + frame_len = max(1, int(sr * 0.03)) + num_frames = len(waveform) // frame_len + if num_frames < 5: + # Too short to estimate reliably — assume clean + return float("inf") + + frames = waveform[: num_frames * frame_len].reshape(num_frames, frame_len) + frame_rms = frames.pow(2).mean(dim=1).sqrt() # shape: (num_frames,) + + sorted_rms, _ = frame_rms.sort() + noise_cutoff = max(1, int(num_frames * 0.20)) + noise_rms = sorted_rms[:noise_cutoff].mean().item() + speech_rms = sorted_rms[noise_cutoff:].mean().item() + + if noise_rms < 1e-9: + return float("inf") # Silent noise floor — effectively infinite SNR + + snr_db = 10.0 * math.log10((speech_rms / noise_rms) ** 2) + return snr_db + + +def maybe_denoise( + wav_path: Path, model: str = None, snr_threshold: float = None +) -> Path: + """Return denoised WAV path if DENOISE_MODEL is set; otherwise return wav_path unchanged.""" + effective_model = (model or DENOISE_MODEL).strip().lower() + if effective_model == "none": + return wav_path + + threshold = snr_threshold if snr_threshold is not None else DENOISE_SNR_THRESHOLD + out_path = wav_path.with_suffix(".denoised.wav") + + if effective_model == "deepfilternet": + import torch + import torchaudio + + snr_db = _estimate_snr(wav_path) + if snr_db >= threshold: + logger.info("DeepFilterNet skipped (SNR=%.1fdB, clean audio)", snr_db) + return wav_path + + logger.info( + "DeepFilterNet applying (SNR=%.1fdB < %.1fdB threshold)", + snr_db, + threshold, + ) + df_model, df_state = _load_deepfilternet() + import df as _df_pkg + + audio, sr = torchaudio.load(str(wav_path)) + if sr != df_state.sr(): + audio = torchaudio.functional.resample(audio, sr, df_state.sr()) + audio = audio.contiguous() + with torch.backends.cudnn.flags(enabled=False): + enhanced = _df_pkg.enhance(df_model, df_state, audio) + torchaudio.save( + str(out_path), + enhanced.unsqueeze(0) if enhanced.dim() == 1 else enhanced, + df_state.sr(), + ) + logger.info("DeepFilterNet: denoised %s → %s", wav_path.name, out_path.name) + + elif effective_model == "noisereduce": + import soundfile as sf + import noisereduce as nr + + data, sr = sf.read(str(wav_path), dtype="float32") + reduced = nr.reduce_noise(y=data, sr=sr, stationary=True) + sf.write(str(out_path), reduced, sr) + logger.info("noisereduce: denoised %s → %s", wav_path.name, out_path.name) + + else: + logger.warning("Unknown DENOISE_MODEL=%r — skipping denoising", effective_model) + return wav_path + + return out_path diff --git a/app/services/job_service.py b/app/services/job_service.py new file mode 100644 index 0000000..08673d4 --- /dev/null +++ b/app/services/job_service.py @@ -0,0 +1,383 @@ +"""Job management and background transcription worker. + +Owns: +- _LRUJobsDict: bounded thread-safe LRU store for job states +- jobs: the singleton in-memory job registry +- _gpu_sem: semaphore that serialises GPU access to one transcription at a time +- run_transcription: the background worker function +""" + +import json +import logging +import threading +from collections import OrderedDict +from datetime import datetime +from pathlib import Path + +from config import ( + DENOISE_MODEL, + DENOISE_SNR_THRESHOLD, + JOBS_MAX_CACHE, + TRANSCRIPTIONS_DIR, + VOICEPRINT_THRESHOLD, +) +from services.audio_service import convert_to_wav, maybe_denoise, register_hash + +logger = logging.getLogger(__name__) + +# CQ-C1: counter used to periodically rebuild AS-norm cohort inside the +# transcription worker so it becomes active without requiring a server restart. +_cohort_rebuild_counter: dict = {} + + +# --------------------------------------------------------------------------- +# Status persistence helpers (AR-C2) +# --------------------------------------------------------------------------- + + +def _write_status( + job_id: str, + status: str, + error: str | None = None, + filename: str | None = None, +) -> None: + """Write job status to disk for persistence across process restarts.""" + status_path = TRANSCRIPTIONS_DIR / job_id / "status.json" + status_path.parent.mkdir(parents=True, exist_ok=True) + try: + payload = { + "status": status, + "updated_at": datetime.now().isoformat(), + "error": error, + } + if filename is not None: + payload["filename"] = filename + status_path.write_text(json.dumps(payload)) + except Exception as exc: + logger.warning("Failed to write status.json for %s: %s", job_id, exc) + + +def recover_orphan_jobs() -> None: + """Mark any in-progress jobs as failed if the process was restarted. + + Called once during application lifespan startup so that frontend polls + receive a definitive terminal state instead of hanging on stale + 'transcribing'/'queued' statuses written by a previous process. + """ + try: + for status_path in TRANSCRIPTIONS_DIR.glob("*/status.json"): + try: + data = json.loads(status_path.read_text()) + if data.get("status") not in ("completed", "failed"): + data["status"] = "failed" + data["error"] = "Process restarted while job was in progress" + data["updated_at"] = datetime.now().isoformat() + status_path.write_text(json.dumps(data)) + logger.info( + "AR-C2: marked orphan job %s as failed", + status_path.parent.name, + ) + except Exception as exc: + logger.warning( + "AR-C2: could not recover orphan job at %s: %s", status_path, exc + ) + except Exception as exc: + logger.warning("AR-C2: orphan job recovery scan failed: %s", exc) + + +# --------------------------------------------------------------------------- +# Bounded LRU job store (CQ-H2 / PERF-C1) +# --------------------------------------------------------------------------- + + +class _LRUJobsDict: + """Thread-safe LRU dict for job states with bounded size.""" + + def __init__(self, maxsize: int = 200): + self._d: OrderedDict = OrderedDict() + self._lock = threading.Lock() + self._maxsize = maxsize + + def __setitem__(self, key, value): + with self._lock: + if key in self._d: + self._d.move_to_end(key) + self._d[key] = value + if len(self._d) > self._maxsize: + self._d.popitem(last=False) + + def __getitem__(self, key): + with self._lock: + return self._d[key] + + def __contains__(self, key): + with self._lock: + return key in self._d + + def get(self, key, default=None): + with self._lock: + return self._d.get(key, default) + + +# In-memory job status — bounded LRU (CQ-H2 / PERF-C1) +jobs: _LRUJobsDict = _LRUJobsDict(maxsize=JOBS_MAX_CACHE) + +# Serialise GPU work: only one transcription runs at a time. +# Concurrent HTTP uploads are fine; they queue here before touching the GPU. +_gpu_sem = threading.Semaphore(1) + + +# --------------------------------------------------------------------------- +# Background worker +# --------------------------------------------------------------------------- + + +def run_transcription( + job_id: str, + audio_path: Path, + language: str, + min_speakers: int, + max_speakers: int, + pipeline, + voiceprint_db, + denoise_model: str = None, + snr_threshold: float = None, + file_hash: str = None, + no_repeat_ngram_size: int = 0, +): + """Background transcription worker. + + Accepts *pipeline* and *voiceprint_db* as explicit arguments (injected by + the route handler from app.state) to avoid global-state coupling and make + the function testable in isolation. + """ + # Track intermediate files so they can be cleaned up on both success and + # failure. Initialise to audio_path so the cleanup guard (path != audio_path) + # is safe even if an exception fires before the variables are reassigned. + wav_path: Path = audio_path + clean_path: Path = audio_path + try: + jobs[job_id]["status"] = "converting" + _write_status(job_id, "converting", filename=audio_path.name) + wav_path = convert_to_wav(audio_path) + + jobs[job_id]["status"] = "queued" + _write_status(job_id, "queued") + with _gpu_sem: + _intermediate = ( + "denoising" + if (denoise_model or DENOISE_MODEL) != "none" + else "transcribing" + ) + jobs[job_id]["status"] = _intermediate + _write_status(job_id, _intermediate) + clean_path = maybe_denoise(wav_path, denoise_model, snr_threshold) + + # DF peaks at ~15 GB reserved in PyTorch's CUDA cache. + # ctranslate2 (Whisper) calls cudaMalloc directly and sees the OS + # free memory — not PyTorch's allocator pool — so it OOMs unless we + # explicitly flush the cache before Whisper cold-loads. + try: + import gc as _gc + + import torch as _torch + + _gc.collect() + if _torch.cuda.is_available(): + _torch.cuda.empty_cache() + except Exception as exc: + logger.warning("pre-whisper CUDA cache flush failed: %s", exc) + + jobs[job_id]["status"] = "transcribing" + _write_status(job_id, "transcribing") + result = pipeline.process( + str(clean_path), + raw_audio_path=str(wav_path), + language=language, + min_speakers=min_speakers or None, + max_speakers=max_speakers or None, + no_repeat_ngram_size=no_repeat_ngram_size or None, + ) + + # Release cached CUDA memory so the next queued job has headroom + try: + import gc as _gc + + import torch as _torch + + _gc.collect() + if _torch.cuda.is_available(): + _torch.cuda.empty_cache() + except Exception as exc: + logger.warning("post-pipeline CUDA cache flush failed: %s", exc) + + # Delete intermediate files — keep only the original uploaded file. + # clean_path is the denoised WAV (may equal wav_path if denoising was skipped). + # wav_path is the converted WAV (may equal audio_path if input was already WAV). + if clean_path != wav_path: + clean_path.unlink(missing_ok=True) + if wav_path != audio_path: + wav_path.unlink(missing_ok=True) + + # Match speakers against voiceprint DB + jobs[job_id]["status"] = "identifying" + _write_status(job_id, "identifying") + speaker_map = {} + for spk_label, embedding in result["speaker_embeddings"].items(): + spk_id, spk_name, sim = voiceprint_db.identify( + embedding, threshold=VOICEPRINT_THRESHOLD + ) + speaker_map[spk_label] = { + "matched_id": spk_id, + "matched_name": spk_name or spk_label, + "similarity": round(sim, 4), + "embedding_key": spk_label, + } + + # [CQ-H6] 若所有 turn 均短于 MIN_EMBED_DURATION，embeddings 为空 → 不产生 speaker_map。 + # 记录明确 warning，让前端可以区分"无可登记 speaker"并避免传 'undefined' 字符串。 + warning = None + if not speaker_map: + warning = "no_speakers_detected" + logger.warning( + "Job %s produced no speaker embeddings (all turns < min duration)", + job_id, + ) + + # Consolidate multiple diarization clusters that resolved to the same + # enrolled speaker. Pick the cluster with the highest similarity as the + # canonical label; remap all others to it so one person appears under a + # single label rather than as separate SPEAKER_XX entries. + _id_to_clusters: dict = {} + for _lbl, _info in speaker_map.items(): + _mid = _info["matched_id"] + if _mid is not None: + _id_to_clusters.setdefault(_mid, []).append((_lbl, _info["similarity"])) + + _cluster_remap: dict[str, str] = {} + for _mid, _cluster_list in _id_to_clusters.items(): + _cluster_list.sort(key=lambda x: x[1], reverse=True) + _canonical_lbl = _cluster_list[0][0] + for _lbl, _ in _cluster_list[1:]: + _cluster_remap[_lbl] = _canonical_lbl + logger.info( + "Job %s: merged cluster %s → %s (same enrolled speaker %s)", + job_id, + _lbl, + _canonical_lbl, + _mid, + ) + + # Build final segments with remapped speaker labels + segments = [] + for i, seg in enumerate(result["segments"]): + spk_label = seg["speaker"] + canonical_label = _cluster_remap.get(spk_label, spk_label) + match = speaker_map.get(canonical_label, speaker_map.get(spk_label, {})) + out = { + "id": i, + "start": seg["start"], + "end": seg["end"], + "text": seg["text"], + "speaker_label": canonical_label, + "speaker_id": match.get("matched_id"), + "speaker_name": match.get("matched_name", canonical_label), + "similarity": match.get("similarity", 0), + } + # Forward word-level timestamps when forced alignment produced them + # (0.3.0+). Absent when the language has no alignment model or + # alignment failed — clients must treat the key as optional. + if seg.get("words"): + out["words"] = seg["words"] + segments.append(out) + + # Derive unique_speakers from resolved speaker names (ordered by first + # appearance in the transcript, deduplicated). Enrolled speakers appear + # under their enrolled name; unidentified clusters keep their raw label. + _seen_spk: set = set() + resolved_unique_speakers: list = [] + for seg in segments: + name = seg["speaker_name"] + if name not in _seen_spk: + _seen_spk.add(name) + resolved_unique_speakers.append(name) + + # Save transcription result + effective_denoise = (denoise_model or DENOISE_MODEL).strip().lower() + effective_snr = ( + snr_threshold if snr_threshold is not None else DENOISE_SNR_THRESHOLD + ) + tr = { + "id": job_id, + "filename": audio_path.name, + "created_at": datetime.now().isoformat(), + "status": "completed", + "language": language, + "segments": segments, + "speaker_map": speaker_map, + "unique_speakers": resolved_unique_speakers, + "params": { + "language": language or "auto", + "denoise_model": effective_denoise, + "snr_threshold": effective_snr, + "voiceprint_threshold": VOICEPRINT_THRESHOLD, + "min_speakers": min_speakers, + "max_speakers": max_speakers, + "no_repeat_ngram_size": no_repeat_ngram_size or 0, + }, + } + if warning is not None: + tr["warning"] = warning + + tr_dir = TRANSCRIPTIONS_DIR / job_id + tr_dir.mkdir(exist_ok=True) + (tr_dir / "result.json").write_text( + json.dumps(tr, ensure_ascii=False, indent=2), encoding="utf-8" + ) + + # Save raw embeddings for later enrollment + import numpy as np + + for spk_label, emb in result["speaker_embeddings"].items(): + np.save(tr_dir / f"emb_{spk_label}.npy", emb) + + if file_hash: + register_hash(file_hash, job_id) + + # CQ-C1: After each successful transcription, check if AS-norm cohort + # should be rebuilt. Every 10th job (or when cohort is absent) we rebuild + # so that newly enrolled speakers contribute to normalization without + # requiring a server restart. + try: + _cohort_rebuild_counter[0] = _cohort_rebuild_counter.get(0, 0) + 1 + if voiceprint_db.cohort_size == 0 or _cohort_rebuild_counter[0] % 10 == 0: + voiceprint_db.build_cohort_from_transcriptions(str(TRANSCRIPTIONS_DIR)) + logger.info( + "AS-norm cohort rebuilt: size=%d", voiceprint_db.cohort_size + ) + except Exception as exc: + logger.warning("cohort rebuild failed: %s", exc) + + jobs[job_id]["status"] = "completed" + jobs[job_id]["result"] = tr + _write_status(job_id, "completed") + logger.info( + "Job %s completed: %d segments, %d speakers", + job_id, + len(segments), + len(speaker_map), + ) + + except Exception as e: + logger.exception("Job %s failed", job_id) + jobs[job_id]["status"] = "failed" + jobs[job_id]["error"] = str(e) + _write_status(job_id, "failed", error=str(e)) + # Best-effort cleanup of intermediate files on failure. + try: + if clean_path != wav_path: + clean_path.unlink(missing_ok=True) + if wav_path != audio_path: + wav_path.unlink(missing_ok=True) + except Exception: + pass diff --git a/app/static/index.html b/app/static/index.html index 6a46f95..4c44902 100755 --- a/app/static/index.html +++ b/app/static/index.html @@ -3,8 +3,13 @@ + voscript +