diff --git a/servers/tools-pve/asr-pipeline/.dockerignore b/servers/tools-pve/asr-pipeline/.dockerignore index 035fdf7..bf4df38 100644 --- a/servers/tools-pve/asr-pipeline/.dockerignore +++ b/servers/tools-pve/asr-pipeline/.dockerignore @@ -1,7 +1,9 @@ .git -.env -tests -README.md +.venv __pycache__ *.pyc -seed +state/ +cache/ +dry-run/ +logs/ +.env diff --git a/servers/tools-pve/asr-pipeline/.env.example b/servers/tools-pve/asr-pipeline/.env.example index ad631e9..2daba59 100644 --- a/servers/tools-pve/asr-pipeline/.env.example +++ b/servers/tools-pve/asr-pipeline/.env.example @@ -1,11 +1,44 @@ -ZENMUX_API_KEY= -ASR_AGENT_TOKEN= -ASR_PUBLIC_URL=http://192.168.50.49:8788 +# Deployment example only. Do not commit real secrets. +ASR_WORKSPACE=/home/claw/codex-workspace/meeting-asr +ASR_VAULT=/home/claw/obsidian-vault +ASR_STATE_DIR=/home/claw/codex-workspace/meeting-asr/state +ASR_CACHE_DIR=/home/claw/codex-workspace/meeting-asr/cache +ASR_LOG=/home/claw/codex-workspace/meeting-asr/logs/asr-pipeline.log +ASR_ENV_FILE=/home/claw/codex-workspace/meeting-asr/.env +ASR_RECORDINGS_ROOT=/home/claw/codex-workspace/meeting-asr/recordings +MEETING_INBOX_BASE=/home/claw/codex-workspace/meeting-asr/recordings +ASR_CONTROL_DB=/home/claw/codex-workspace/meeting-asr/state/intake.db + +# ASR Provider: qwen3cam | soniox | gemini | openai | grok +ASR_PROVIDER=qwen3cam ASR_DIARIZE_URL=http://192.168.50.100:10099/asr +ASR_VOICEPRINT_ENROLL_URL=http://192.168.50.100:10099/voiceprints/enroll + +# Unified LiteLLM gateway (replaces deprecated ZenMux) +LITELLM_BASE_URL=http://192.168.50.49:4000/v1 +LITELLM_API_KEY= +ASR_GEMINI_MODEL=aistudio/gemini-3.5-transcribe +ASR_OPENAI_MODEL=openai/gpt-4o-transcribe-diarize +ASR_CLEAN_MODEL=aistudio/gemini-flash-lite-latest +ASR_KEYWORD_MODEL=aistudio/gemini-flash-lite-latest +ASR_DRAFT_MODEL=codex/gpt-sol-latest + +# Direct cloud ASR keys (when not routed through LiteLLM) +SONIOX_API_KEY= +SONIOX_MODEL=stt-async-v5 +XAI_API_KEY= +ASR_GROK_MODEL=grok-voice-transcribe-2.0 + +# Knowledge & Obsidian integration HINDSIGHT_API=http://192.168.50.47:8888 HINDSIGHT_BANK=personal-context +ASR_GIT_SYNC=0 + +# Web & Notifications +ASR_PUBLIC_URL=http://192.168.50.49:8788 +ASR_REVIEW_PUBLIC_HOST=192.168.50.145 +ASR_REVIEW_PORT=8788 +ASR_AGENT_TOKEN= +ASR_WEB_PASSWORD= +ASR_NOTIFY=0 ASR_NOTIFY_BACKEND=none -HERMES_WEBHOOK_URL= -HERMES_WEBHOOK_SECRET= -TELEGRAM_BOT_TOKEN= -TELEGRAM_CHAT_ID= diff --git a/servers/tools-pve/asr-pipeline/Dockerfile b/servers/tools-pve/asr-pipeline/Dockerfile index f9711ac..b7b7b2a 100644 --- a/servers/tools-pve/asr-pipeline/Dockerfile +++ b/servers/tools-pve/asr-pipeline/Dockerfile @@ -2,7 +2,7 @@ FROM python:3.13-slim-bookworm@sha256:ed86c82274b3c69b52fb5820f358f0bd7df0b60333 RUN apt-get update \ && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \ - ca-certificates curl tini \ + ca-certificates curl ffmpeg git tini \ && rm -rf /var/lib/apt/lists/* WORKDIR /app @@ -15,6 +15,7 @@ COPY termbank/ ./termbank/ ENV PYTHONUNBUFFERED=1 \ PYTHONDONTWRITEBYTECODE=1 \ PYTHONPATH=/app \ + ASR_DATA_DIR=/data \ ASR_STATE_DIR=/state \ ASR_VAULT=/data/vault \ ASR_TERMBANK_DIR=/app/termbank \ @@ -24,6 +25,11 @@ ENV PYTHONUNBUFFERED=1 \ ASR_DRAFTS_DIR=/data/artifacts/drafts \ ASR_RESULTS_DIR=/data/artifacts/results \ ASR_REVISION_QUEUE_DIR=/state/revision-queue \ + ASR_PROVIDER=qwen3cam \ + LITELLM_BASE_URL=http://192.168.50.49:4000/v1 \ + ASR_CLEAN_MODEL=aistudio/gemini-flash-lite-latest \ + ASR_KEYWORD_MODEL=aistudio/gemini-flash-lite-latest \ + ASR_DRAFT_MODEL=codex/gpt-sol-latest \ ASR_GIT_SYNC=0 \ ASR_NOTIFY=0 \ ASR_REVIEW_HOST=0.0.0.0 \ diff --git a/servers/tools-pve/asr-pipeline/README.md b/servers/tools-pve/asr-pipeline/README.md index 0eef9c7..e09c3f3 100644 --- a/servers/tools-pve/asr-pipeline/README.md +++ b/servers/tools-pve/asr-pipeline/README.md @@ -1,51 +1,55 @@ -# ASR Pipeline on tools-pve +# Meeting ASR -独立的录音转写、精修、人工审核和导出服务。WebUI 与 Agent API 共用 `192.168.50.49:8788`；worker 调用 unRaid 上的 Qwen3-ASR/CAM++ 服务以及 ZenMux 模型 API。 +独立的会议/通话录音转写与整理 Web 应用（`asrservice` + `asrpipe`）。支持本地 GPU 声纹服务（`qwen3cam`）与云端多模型语音识别（`soniox`、`gemini`、`openai`、`grok`），可通过 Arcane 部署到本地 VM（`tools-pve`）或外网 VPS。 -## 数据边界 +## 核心架构 -- `/mnt/truenas/personal/ASR-Pipeline` → 容器 `/data`：原始录音、转写、ASR 原始响应、草稿、正式稿、日志和 SQLite 在线备份。 -- `/opt/asr-pipeline/state` → 容器 `/state`：活动 SQLite、worker heartbeat、revision queue。SQLite WAL 不放在 NFS。 -- `/srv/asr/inbox`：指向 TrueNAS inbox 的宿主机链接，供 `scp`/`sftp` 上传。 -- Secret 只存在于 Arcane Project 环境和 tools-pve 的 `0600` 恢复副本，不进入 Git。 +1. **Web 控制与审核面（`asrservice.web`）** + - 极简零构建服务端渲染 HTML + 原生 JS（深浅色自适应）。 + - 上传录音时可按任务切换 ASR 引擎（`qwen3cam` / `soniox` / `gemini` / `openai` / `grok`）。 + - 任务详情页内置支持 HTTP Range 的音频播放器、失败一键切换模型重试、草稿与纪要 Markdown 在线编辑。 + - `/review/` 提供低置信度词语、关系确认与说话人身份确认（本地 `qwen3cam` 自动注册声纹至 `:10099`，线上多说话人模型直接更新草稿与词库而不依赖 `:10099`）。 + - `/lexicon` 提供热词、同音纠错、人物关系管理与 Obsidian 词库一键同步。 + - `/settings` 展示全部 ASR 引擎、LiteLLM 网关、Obsidian 与 Hindsight 实时就绪状态。 + - 支持可选 `ASR_WEB_PASSWORD` Cookie 鉴权（外网 VPS 部署时自动保护 `/ui/*` 与 `/review/*`）。 -SSH 大文件建议先上传为 `.part`，完成后在同一目录原子改名为受支持扩展名： +2. **后台消费与修订队列（`asrservice.worker`）** + - 监控 `.asr-inbox/`（Web 上传或目录投递）并认领 `intake.db` 任务。 + - 调用统一多引擎层 `asrpipe.providers.transcribe_audio()` 生成带结构化注释头的原始转写。 + - 执行 `asr-refine draft`：结合 SQLite 词库、Obsidian Vault（`People/`、`Companies/`、`Projects/`、`minutes/`、`_glossary.md`）与 Hindsight（`personal-context`）进行分段去口语化、保真检查与会议纪要生成。 + - 监控 `state/revision-queue/` 处理 Web 审核提交的修订重跑请求，确认通过后发布正式纪要至 Obsidian 并可选执行 Git 自动提交推送。 + +3. **语音识别（ASR）与大模型网关边界** + - **LLM 后处理**：彻底移除 ZenMux，统一通过 **LiteLLM**（内网 `http://192.168.50.49:4000/v1` / 外网 `https://api.opfo.xyz/v1`）调用： + - `ASR_CLEAN_MODEL` / `ASR_KEYWORD_MODEL`：默认 `aistudio/gemini-flash-lite-latest` + - `ASR_DRAFT_MODEL`：默认 `codex/gpt-sol-latest` + - **ASR 引擎**： + - `qwen3cam`（内网默认）：对接局域网 GPU `Qwen3-ASR 1.7B + CAM++`（`http://192.168.50.100:10099/asr`），支持声纹特征比对与增量注册。 + - `soniox`（线上高精度首选）：因 LiteLLM 同步 `/v1/audio/transcriptions` 不支持 Soniox 异步 Files API、结构化 `context` 热词注入与逐词 `speaker` 分离且有 180s 超时限制，由 `asrpipe.providers` 通过 `SONIOX_API_KEY` 直连 Soniox `stt-async-v5`（自动注入 SQLite 词库热词与任务背景，完成后自动清理云端临时音频）。 + - `gemini`：通过 LiteLLM 统一网关调用 `aistudio/gemini-3.5-transcribe`（长音频自动 `ffmpeg` 切片）。 + - `openai`：通过 LiteLLM 统一网关调用 `openai/gpt-4o-transcribe-diarize`（支持 `diarized_json` 角色分离与 >23MB 自动压缩切片）。 + - `grok`：通过 `XAI_API_KEY` 调用 xAI `grok-voice-transcribe-2.0`（支持多说话人分离）。 + +## 目录结构 + +- `asrpipe/`：核心转写引擎适配（`providers.py`）、词库召回（`lexicon.py`）、分段去口语化（`segment.py`、`refine.py`）、审核流（`review.py`）与 Obsidian/Hindsight 集成。 +- `asrservice/`：容器化 Web 服务（`web.py`）、后台 Worker（`worker.py`）、SQLite 任务状态机（`db.py`）、Webhook 通知（`notify.py`）与健康检查（`healthcheck.py`）。 +- `recordings/bin/`：录音监听与 CLI 转写脚本（兼容 `MEETING_ASR_BACKEND` 多引擎调度）。 +- `termbank/`：专业词库源数据。 +- `tests/`：核心管线、多引擎契约与 Web 服务单元测试。 +- `Dockerfile` & `compose.yaml`：面向 Arcane 部署的单镜像双角色（`asr-web` + `asr-worker`）容器编排。 + +## 本地开发与测试 ```bash -scp call.m4a tools@192.168.50.49:/srv/asr/inbox/call.m4a.part -ssh tools@192.168.50.49 'mv /srv/asr/inbox/call.m4a.part /srv/asr/inbox/call.m4a' +python3 -m venv .venv +.venv/bin/pip install -e . +ASR_NOTIFY=0 ASR_GIT_SYNC=0 .venv/bin/python -m unittest discover -s tests -v ``` -## Agent API +## Arcane 部署（内网 VM / 外网 VPS） -所有 `/api/v1/*` 路由使用 `Authorization: Bearer $ASR_AGENT_TOKEN`。 - -```bash -curl -X POST http://192.168.50.49:8788/api/v1/jobs \ - -H "Authorization: Bearer $ASR_AGENT_TOKEN" \ - -H 'X-Filename: call.m4a' -H 'X-Title: 项目通话' \ - -H 'X-Context: 参与人和项目背景' --data-binary @call.m4a - -curl -H "Authorization: Bearer $ASR_AGENT_TOKEN" \ - http://192.168.50.49:8788/api/v1/jobs/JOB_ID -``` - -路由： - -- `POST /api/v1/jobs`：上传并创建异步任务。 -- `GET /api/v1/jobs?limit=&offset=&status=`：分页查询。 -- `GET /api/v1/jobs/{id}`：任务详情、事件时间线与产物链接。 -- `GET /api/v1/jobs/{id}/artifacts/{transcript|draft|result}`：读取产物。 -- `GET /api/v1/jobs/{id}/reviews`：待人工确认项。 -- `POST /api/v1/jobs/{id}/reviews/{question_id}`：JSON `{"answer":"..."}`。 -- `POST /api/v1/jobs/{id}/retry`：重试失败任务。 - -`/healthz` 仅表示 Web 进程存活；`/readyz` 同时检查数据库与 worker heartbeat。 - -## 通知 - -`ASR_NOTIFY_BACKEND=none|telegram|hermes`。Telegram 需要 `TELEGRAM_BOT_TOKEN` 与 `TELEGRAM_CHAT_ID`；Hermes 需要一个明确存在的签名 webhook URL 与共享密钥。发送失败只影响提醒，不改变任务结果。 - -## 运维 - -项目由 Arcane Git Sync 管理，自动同步、自动 pull 和自动 redeploy 均关闭。Compose 变更先在 Git 中验证，再对 tools-pve 的准确 Sync 手动同步。活动数据库每小时通过 SQLite backup API 写到 `/data/backups/*.latest`。 +1. 复制 `.env.example` 为 `.env`，按环境配置： + - **内网 VM（`tools-pve`）**：`MEETING_ASR_BACKEND=qwen3cam`（同时可配 `SONIOX_API_KEY` 等以便在 Web UI 按需切换）、`LITELLM_BASE_URL=http://192.168.50.49:4000/v1`、`HINDSIGHT_URL=http://192.168.50.47:8888`。 + - **外网 VPS**：`MEETING_ASR_BACKEND=soniox`（或 `gemini` / `openai` / `grok`）、`LITELLM_BASE_URL=https://api.opfo.xyz/v1`、设置 `ASR_WEB_PASSWORD` 开启 Web 登录保护。 +2. 使用 `compose.yaml` 启动 `asr-web`（默认 `:8788`）与 `asr-worker`。 diff --git a/servers/tools-pve/asr-pipeline/asrpipe/cli.py b/servers/tools-pve/asr-pipeline/asrpipe/cli.py index 83a954e..b5fe096 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/cli.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/cli.py @@ -209,7 +209,9 @@ def cmd_doctor(a): print("credentials") try: k = CFG.api_key() - chk("api key", len(k) > 20, f"len={len(k)} source={'env' if os.environ.get('ANTHROPIC_AUTH_TOKEN') else CFG.env_file.name}") + env_set = any(os.environ.get(x) for x in ( + "LITELLM_API_KEY", "ASR_LLM_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_AUTH_TOKEN")) + chk("api key", len(k) > 10, f"len={len(k)} source={'env' if env_set else CFG.env_file.name}") except Exception as e: chk("api key", False, str(e)) diff --git a/servers/tools-pve/asr-pipeline/asrpipe/config.py b/servers/tools-pve/asr-pipeline/asrpipe/config.py index 6b82465..822f940 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/config.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/config.py @@ -1,15 +1,11 @@ #!/usr/bin/env python3 -"""Configuration for the standalone ASR post-processing pipeline. +"""Configuration for the standalone ASR post-processing pipeline and web service. -This program deliberately has **no dependency on Claude Code, opencode, or any -agent CLI**. It talks to the model provider over plain HTTPS and does its own file -and git handling, so it can run from systemd, cron, or a shell with nothing but -python3 and a network route. +This program has **no dependency on Claude Code, opencode, or any agent CLI**. +It talks to LiteLLM / model providers over plain HTTPS and does its own file +and git handling, so it can run inside Docker (via Arcane), systemd, or a shell. -Everything is overridable by environment variable so the systemd unit stays the -single place where deployment-specific values live. Credentials are read from the -same 600-perm env file the legacy unit uses, so token rotation remains a one-file -change. +Everything is overridable by environment variable. """ from __future__ import annotations @@ -18,8 +14,8 @@ from dataclasses import dataclass, field from pathlib import Path PKG = Path(__file__).resolve().parent -ROOT = PKG.parent # OBSIDIAN/asr-pipeline -OBSIDIAN = ROOT.parent # OBSIDIAN project dir +ROOT = PKG.parent +OBSIDIAN = Path(os.environ.get("ASR_WORKSPACE", str(ROOT))) def _env(name: str, default: str) -> str: @@ -46,60 +42,124 @@ class Config: # ---- paths ---------------------------------------------------------- vault: Path = field(default_factory=lambda: Path( _env("ASR_VAULT", str(Path.home() / "obsidian-vault")))) - obsidian: Path = OBSIDIAN + obsidian: Path = field(default_factory=lambda: Path( + _env("ASR_WORKSPACE", str(ROOT)))) root: Path = ROOT - # ---- models --------------------------------------------------------- - # Cleaning: measured against claude-haiku-4.5 on a 118k-char English-mixed call, - # gemini-3.5-flash-lite tripped the fidelity gate 19/87 times vs haiku's 78/87 - # (haiku's failures were 69x "over-compression" — it behaves like a summarizer - # and deleted terms such as AUTOIMMUNE/ADCS/CD), while being 2x faster and - # 2.9x cheaper. Hence flash-lite is the default cleaner. - clean_model: str = _env("ASR_CLEAN_MODEL", "google/gemini-3.5-flash-lite") - # Drafting: one call, needs judgement and structure. sonnet-5.5 replaces the - # legacy headless opus-4-8. - draft_model: str = _env("ASR_DRAFT_MODEL", "anthropic/claude-sonnet-5.5") - keyword_model: str = _env("ASR_KEYWORD_MODEL", "google/gemini-3.5-flash-lite") + # ---- LiteLLM / LLM routing (ZenMux deprecated) ---------------------- + litellm_base_url: str = field(default_factory=lambda: _env( + "LITELLM_BASE_URL", + _env("ASR_LLM_BASE_URL", "http://192.168.50.49:4000/v1")).rstrip("/")) - endpoint: str = _env("ASR_ENDPOINT", "https://zenmux.ai/api/v1/chat/completions") - # ZenMux returned HTTP 429 at ~12 concurrent requests and was stable at 6. - concurrency: int = _envi("CONCURRENCY", 6) - http_timeout: float = _envf("HTTP_TIMEOUT", 420.0) - max_retries: int = _envi("MAX_RETRIES", 5) + clean_model: str = field(default_factory=lambda: _env( + "ASR_CLEAN_MODEL", "aistudio/gemini-flash-lite-latest")) + draft_model: str = field(default_factory=lambda: _env( + "ASR_DRAFT_MODEL", "codex/gpt-sol-latest")) + keyword_model: str = field(default_factory=lambda: _env( + "ASR_KEYWORD_MODEL", "aistudio/gemini-flash-lite-latest")) + + endpoint: str = field(default_factory=lambda: _env( + "ASR_ENDPOINT", + _env("LITELLM_BASE_URL", + _env("ASR_LLM_BASE_URL", "http://192.168.50.49:4000/v1")).rstrip("/") + + "/chat/completions")) + transcriptions_endpoint: str = field(default_factory=lambda: _env( + "ASR_TRANSCRIPTIONS_ENDPOINT", + _env("LITELLM_BASE_URL", + _env("ASR_LLM_BASE_URL", "http://192.168.50.49:4000/v1")).rstrip("/") + + "/audio/transcriptions")) + + concurrency: int = field(default_factory=lambda: _envi("CONCURRENCY", 6)) + http_timeout: float = field(default_factory=lambda: _envf("HTTP_TIMEOUT", 420.0)) + max_retries: int = field(default_factory=lambda: _envi("MAX_RETRIES", 5)) + + # ---- ASR providers -------------------------------------------------- + asr_provider: str = field(default_factory=lambda: _env( + "ASR_PROVIDER", _env("MEETING_ASR_BACKEND", "qwen3cam")).lower()) + diarize_url: str = field(default_factory=lambda: _env( + "ASR_DIARIZE_URL", + _env("MEETING_DIARIZE_URL", "http://192.168.50.100:10099/asr"))) + voiceprint_enroll_url: str = field(default_factory=lambda: _env( + "ASR_VOICEPRINT_ENROLL_URL", + "http://192.168.50.100:10099/voiceprints/enroll")) + + soniox_api_key: str = field(default_factory=lambda: _env("SONIOX_API_KEY", "")) + soniox_api_base: str = field(default_factory=lambda: _env( + "SONIOX_API_BASE", "https://api.soniox.com/v1").rstrip("/")) + soniox_model: str = field(default_factory=lambda: _env( + "SONIOX_MODEL", "stt-async-v5")) + + gemini_asr_model: str = field(default_factory=lambda: _env( + "ASR_GEMINI_MODEL", "aistudio/gemini-3.5-transcribe")) + openai_asr_model: str = field(default_factory=lambda: _env( + "ASR_OPENAI_MODEL", "openai/gpt-4o-transcribe-diarize")) + + grok_api_key: str = field(default_factory=lambda: _env( + "XAI_API_KEY", _env("GROK_API_KEY", ""))) + grok_api_base: str = field(default_factory=lambda: _env( + "XAI_API_BASE", "https://api.x.ai/v1").rstrip("/")) + grok_asr_model: str = field(default_factory=lambda: _env( + "ASR_GROK_MODEL", "grok-voice-transcribe-2.0")) + + segment_seconds: int = field(default_factory=lambda: _envi( + "SEGMENT_SECONDS", int(_env("MEETING_ASR_SEGMENT_SECONDS", "540")))) # ---- chunking ------------------------------------------------------- - chunk_budget: int = _envi("CHUNK_BUDGET", 1800) - chunk_overlap: int = _envi("CHUNK_OVERLAP", 200) - collapse_noise: bool = _env("ASR_COLLAPSE_NOISE", "1") != "0" + chunk_budget: int = field(default_factory=lambda: _envi("CHUNK_BUDGET", 1800)) + chunk_overlap: int = field(default_factory=lambda: _envi("CHUNK_OVERLAP", 200)) + collapse_noise: bool = field( + default_factory=lambda: _env("ASR_COLLAPSE_NOISE", "1") != "0") # ---- fidelity gate -------------------------------------------------- - gate_enabled: bool = _env("ASR_GATE", "1") != "0" - gate_min_ratio: float = _envf("GATE_MIN_RATIO", 0.55) - gate_max_ratio: float = _envf("GATE_MAX_RATIO", 1.35) + gate_enabled: bool = field( + default_factory=lambda: _env("ASR_GATE", "1") != "0") + gate_min_ratio: float = field( + default_factory=lambda: _envf("GATE_MIN_RATIO", 0.55)) + gate_max_ratio: float = field( + default_factory=lambda: _envf("GATE_MAX_RATIO", 1.35)) # ---- knowledge sources ---------------------------------------------- - use_termbank: bool = _env("ASR_TERMBANK", "1") != "0" - use_hindsight: bool = _env("ASR_HINDSIGHT", "1") != "0" - use_keyword_recall: bool = _env("ASR_KEYWORD_RECALL", "1") != "0" - use_lexicon: bool = _env("ASR_LEXICON", "1") != "0" - use_normalize: bool = _env("ASR_NORMALIZE", "1") != "0" - use_evidence: bool = _env("ASR_EVIDENCE", "1") != "0" - hindsight_api: str = _env("HINDSIGHT_API", "http://192.168.50.47:8888") - hindsight_bank: str = _env("HINDSIGHT_BANK", "personal-context") - hindsight_timeout: float = _envf("HINDSIGHT_TIMEOUT", 12.0) - flag_min_score: float = _envf("FLAG_MIN_SCORE", 0.86) - review_min_score: float = _envf("REVIEW_MIN_SCORE", 0.82) - review_max_score: float = _envf("REVIEW_MAX_SCORE", 0.96) - review_host: str = _env("ASR_REVIEW_HOST", "0.0.0.0") - review_public_host: str = _env("ASR_REVIEW_PUBLIC_HOST", "192.168.50.145") - review_port: int = _envi("REVIEW_PORT", 8765) - review_idle_seconds: int = _envi("REVIEW_IDLE_SECONDS", 2700) - review_max_seconds: int = _envi("REVIEW_MAX_SECONDS", 7200) + use_termbank: bool = field( + default_factory=lambda: _env("ASR_TERMBANK", "1") != "0") + use_hindsight: bool = field( + default_factory=lambda: _env("ASR_HINDSIGHT", "1") != "0") + use_keyword_recall: bool = field( + default_factory=lambda: _env("ASR_KEYWORD_RECALL", "1") != "0") + use_lexicon: bool = field( + default_factory=lambda: _env("ASR_LEXICON", "1") != "0") + use_normalize: bool = field( + default_factory=lambda: _env("ASR_NORMALIZE", "1") != "0") + use_evidence: bool = field( + default_factory=lambda: _env("ASR_EVIDENCE", "1") != "0") + hindsight_api: str = field(default_factory=lambda: _env( + "HINDSIGHT_API", "http://192.168.50.47:8888")) + hindsight_bank: str = field(default_factory=lambda: _env( + "HINDSIGHT_BANK", "personal-context")) + hindsight_timeout: float = field( + default_factory=lambda: _envf("HINDSIGHT_TIMEOUT", 12.0)) + flag_min_score: float = field( + default_factory=lambda: _envf("FLAG_MIN_SCORE", 0.86)) + review_min_score: float = field( + default_factory=lambda: _envf("REVIEW_MIN_SCORE", 0.82)) + review_max_score: float = field( + default_factory=lambda: _envf("REVIEW_MAX_SCORE", 0.96)) + review_host: str = field( + default_factory=lambda: _env("ASR_REVIEW_HOST", "0.0.0.0")) + review_public_host: str = field( + default_factory=lambda: _env("ASR_REVIEW_PUBLIC_HOST", "192.168.50.145")) + review_port: int = field(default_factory=lambda: _envi("REVIEW_PORT", 8765)) + review_idle_seconds: int = field( + default_factory=lambda: _envi("REVIEW_IDLE_SECONDS", 2700)) + review_max_seconds: int = field( + default_factory=lambda: _envi("REVIEW_MAX_SECONDS", 7200)) # ---- behaviour ------------------------------------------------------ - git_sync: bool = _env("ASR_GIT_SYNC", "1") != "0" - notify: bool = _env("ASR_NOTIFY", "1") != "0" - cache_llm: bool = _env("ASR_CACHE", "0") != "0" # off in production + git_sync: bool = field( + default_factory=lambda: _env("ASR_GIT_SYNC", "1") != "0") + notify: bool = field( + default_factory=lambda: _env("ASR_NOTIFY", "1") != "0") + cache_llm: bool = field( + default_factory=lambda: _env("ASR_CACHE", "0") != "0") @property def inbox(self) -> Path: @@ -120,7 +180,6 @@ class Config: @property def pending_index(self) -> Path: - # Legacy compatibility only; the Web queue replaced the Vault draft index. return self.vault / "minutes" / "_drafts" / "_待确认.md" @property @@ -150,15 +209,16 @@ class Config: @property def notifier(self) -> Path: - return self.obsidian / "bin" / "meeting-status-notify.py" + return Path(_env("ASR_NOTIFIER", str(ROOT / "scripts" / "meeting-status-notify.py"))) @property def env_file(self) -> Path: - """600-perm credentials file, shared with the legacy systemd unit.""" - return Path(_env("ASR_ENV_FILE", str(self.obsidian / "bin" / ".asr-refine.env"))) + """600-perm credentials file.""" + return Path(_env("ASR_ENV_FILE", str(ROOT / ".env"))) def api_key(self) -> str: - for k in ("ZENMUX_API_KEY", "ANTHROPIC_AUTH_TOKEN"): + for k in ("LITELLM_API_KEY", "ASR_LLM_API_KEY", "OPENAI_API_KEY", + "ANTHROPIC_AUTH_TOKEN"): v = os.environ.get(k) if v: return v @@ -166,13 +226,15 @@ class Config: if p.exists(): for line in p.read_text(errors="ignore").splitlines(): line = line.strip() - if line.startswith("ANTHROPIC_AUTH_TOKEN="): - val = line.split("=", 1)[1].strip().strip("'\"") - if val: - return val + for prefix in ("LITELLM_API_KEY=", "ASR_LLM_API_KEY=", + "OPENAI_API_KEY=", "ANTHROPIC_AUTH_TOKEN="): + if line.startswith(prefix): + val = line.split("=", 1)[1].strip().strip("'\"") + if val: + return val raise RuntimeError( - "no API key: set ZENMUX_API_KEY/ANTHROPIC_AUTH_TOKEN or provide " - f"{p} with ANTHROPIC_AUTH_TOKEN=…") + "no API key: set LITELLM_API_KEY / ASR_LLM_API_KEY / OPENAI_API_KEY " + f"or provide {p}") CFG = Config() diff --git a/servers/tools-pve/asr-pipeline/asrpipe/llm.py b/servers/tools-pve/asr-pipeline/asrpipe/llm.py index 0206723..a95bbcf 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/llm.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/llm.py @@ -1,19 +1,7 @@ #!/usr/bin/env python3 -"""HTTPS client for the model provider. No agent CLI, no SDK. +"""HTTPS client for LiteLLM / OpenAI-compatible model gateways. No agent CLI, no SDK. -Deliberately built on urllib so the program has zero third-party HTTP -dependencies: it must keep working inside a bare systemd unit years from now. - -Behaviours encoded from measurement rather than guesswork: - * concurrency 6 (HTTP 429 appears at ~12) - * HTTP 402 "Payment Required" is transient on ZenMux — credit seems reserved - against max_tokens, so a burst of large-budget calls can trip it and then - succeed moments later; it gets a much longer backoff than a rate limit - * `temperature` is rejected outright by claude-sonnet-5.5 / opus-5 but accepted by - haiku-4.5, sonnet-4.6, gemini and qwen, so it cannot be inferred from the - vendor prefix; the client self-heals on the 400 and remembers - * an empty completion (reasoning models burning the whole budget on hidden - thoughts) is retried rather than silently propagated as "" +Built on urllib so the program has zero third-party HTTP dependencies. """ from __future__ import annotations @@ -30,14 +18,22 @@ from .config import CFG from .log import log PRICES = { + "aistudio/gemini-flash-lite-latest": (0.3, 2.5), + "vertex/gemini-flash-lite-latest": (0.3, 2.5), + "gemini-flash-lite-latest": (0.3, 2.5), "google/gemini-3.5-flash-lite": (0.3, 2.5), - "google/gemini-3.5-flash": (1.5, 9.0), + "aistudio/gemini-flash-latest": (1.5, 9.0), + "vertex/gemini-flash-latest": (1.5, 9.0), + "aistudio/gemini-pro-latest": (2.5, 12.0), + "codex/gpt-sol-latest": (2.0, 8.0), + "openai/gpt-sol-latest": (2.0, 8.0), + "codex/gpt-luna-latest": (1.5, 6.0), + "xai/grok-latest": (2.0, 10.0), "anthropic/claude-haiku-4.5": (1.0, 5.0), "anthropic/claude-sonnet-5": (2.0, 10.0), "anthropic/claude-sonnet-5.5": (2.0, 10.0), "anthropic/claude-sonnet-4.6": (3.0, 15.0), "anthropic/claude-opus-5": (5.0, 25.0), - "qwen/qwen3.7-plus": (0.1373076, 0.5492304), } _sem = threading.BoundedSemaphore(CFG.concurrency) @@ -45,6 +41,8 @@ _NO_TEMPERATURE: set[str] = { "anthropic/claude-sonnet-5", "anthropic/claude-sonnet-5.5", "anthropic/claude-opus-5", + "codex/gpt-sol-latest", + "codex/gpt-luna-latest", } _lock = threading.Lock() _key: list[str] = [] @@ -143,7 +141,7 @@ def chat(model: str, prompt: str, *, system: str | None = None, except Exception: pass last = f"HTTP{e.code}: {detail[:200]}" - if e.code == 400 and "temperature" in detail: + if e.code == 400 and "temperature" in detail.lower(): with _lock: _NO_TEMPERATURE.add(model) payload = payload_of() @@ -151,13 +149,12 @@ def chat(model: str, prompt: str, *, system: str | None = None, cpath = _cache_path(model, payload) retries += 1 continue - if e.code not in (402, 408, 429, 500, 502, 503, 504): + if e.code not in (408, 429, 500, 502, 503, 504): if usage: usage.failures += 1 raise RuntimeError(f"{model} {last}") from e retries += 1 - delay = (min(20 * (attempt + 1), 120) if e.code == 402 - else min(2 ** attempt + random.uniform(0, 1.5), 45)) + delay = min(2 ** attempt + random.uniform(0, 1.5), 45) log(f"llm retry {attempt+1}/{CFG.max_retries} {model} {last} " f"sleep {delay:.0f}s", level="WARN") time.sleep(delay) @@ -203,8 +200,6 @@ def chat_many(model: str, prompts: list[str], **kw) -> list[str]: except BaseException as e: # noqa: BLE001 - recorded, decided by caller errors.append((i, e)) if errors: - # Partial failure must not silently yield empty chunks: the caller falls - # back to the original text for these indices, which the gate then flags. for i, e in errors: log(f"chunk {i} failed: {e}", level="ERROR") raise PartialFailure(out, [i for i, _ in errors]) diff --git a/servers/tools-pve/asr-pipeline/asrpipe/prompts.py b/servers/tools-pve/asr-pipeline/asrpipe/prompts.py index 5d07d7a..3826fb7 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/prompts.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/prompts.py @@ -166,6 +166,20 @@ SPEAKER_NOTE = { "- 不同 speaker_N 是不同真人（CAM++ 已分），但可能过度聚类（同一人被拆成多个 N），" "按内容合并同一真人。\n" "- frontmatter 的 people 优先填**声纹命名的真人**。"), + "soniox": ( + "转写来自 Soniox 线上分角色模型：\n" + "- `speaker_N` 为自动区分的说话人标签，按内容与背景推断身份并标「疑似」，" + "待人工在 Web 审核中确认。"), + "openai": ( + "转写来自 OpenAI Transcribe 模型：\n" + "- 若含 `speaker_N` 标签，为单场自动区分的说话人，按内容推断身份并标「疑似」；" + "若无说话人标签则按内容推断发言归属。"), + "gemini": ( + "转写来自 Gemini 3.5 Transcribe 模型：\n" + "- 按内容推断发言归属，不要臆造未出现的 speaker 标签。"), + "grok": ( + "转写来自 Grok Voice Transcribe 2.0 分角色模型：\n" + "- `speaker_N` 为自动区分的说话人标签，按内容推断身份并标「疑似」。"), "vibevoice": "转写带 S# 说话人标签，仅作软提示，块间编号不保证一致。", "qwen3-plain": "转写**无说话人信息**，据内容推断发言归属，不要臆造 speaker 标签。", } diff --git a/servers/tools-pve/asr-pipeline/asrpipe/providers.py b/servers/tools-pve/asr-pipeline/asrpipe/providers.py new file mode 100644 index 0000000..f24f7d4 --- /dev/null +++ b/servers/tools-pve/asr-pipeline/asrpipe/providers.py @@ -0,0 +1,844 @@ +#!/usr/bin/env python3 +"""Unified ASR provider abstraction for local GPU (qwen3cam) and online models +(Soniox, Gemini 3.5 Transcribe via LiteLLM, OpenAI Transcribe via LiteLLM, Grok). +""" +from __future__ import annotations + +import json +import mimetypes +import os +import re +import shutil +import subprocess +import tempfile +import time +import urllib.error +import urllib.request +import uuid +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from .config import CFG +from .log import log + +PROVIDERS: dict[str, dict[str, str]] = { + "qwen3cam": { + "label": "内网 Qwen3-ASR 1.7B + CAM++ (分角色 + 声纹)", + "default_model": "qwen3-asr-1.7b-cam++", + "mode": "lan-diarized-voiceprint", + }, + "soniox": { + "label": "Soniox Async v5 (线上分角色 + 词库热词注入)", + "default_model": "stt-async-v5", + "mode": "cloud-diarized", + }, + "gemini": { + "label": "Gemini 3.5 Transcribe (LiteLLM 统一路由)", + "default_model": "aistudio/gemini-3.5-transcribe", + "mode": "litellm-segmented", + }, + "openai": { + "label": "OpenAI GPT-4o Transcribe Diarize (LiteLLM 统一路由)", + "default_model": "openai/gpt-4o-transcribe-diarize", + "mode": "litellm-diarized", + }, + "grok": { + "label": "Grok Voice Transcribe 2.0 (线上分角色)", + "default_model": "grok-voice-transcribe-2.0", + "mode": "cloud-diarized", + }, +} + + +@dataclass +class TranscribeResult: + provider: str + model: str + transcript_text: str + raw_response: Any + diarized: bool = True + voiceprint_session: str = "" + duration_seconds: int = 0 + meta: dict[str, Any] = field(default_factory=dict) + + +def normalize_provider_name(name: str | None) -> str: + raw = (name or CFG.asr_provider or "qwen3cam").strip().lower() + if raw in ("paraformer", "qwen3", "qwen3-cam"): + return "qwen3cam" + if raw in ("gemini-3.5-transcribe", "aistudio/gemini-3.5-transcribe", + "vertex/gemini-3.5-transcribe"): + return "gemini" + if raw in ("openai-diarize", "gpt-4o-transcribe-diarize", "gpt-4o-transcribe", + "gpt-transcribe"): + return "openai" + if raw in ("xai", "grok-voice-transcribe-2.0"): + return "grok" + return raw if raw in PROVIDERS else "qwen3cam" + + +def provider_status() -> list[dict[str, Any]]: + """Return readiness metadata for each ASR provider without exposing secrets.""" + has_litellm = False + try: + has_litellm = bool(CFG.api_key()) + except Exception: + has_litellm = False + return [ + { + "id": "qwen3cam", + "label": PROVIDERS["qwen3cam"]["label"], + "model": PROVIDERS["qwen3cam"]["default_model"], + "ready": bool(CFG.diarize_url), + "endpoint": CFG.diarize_url, + "via": "LAN GPU (:10099)", + }, + { + "id": "soniox", + "label": PROVIDERS["soniox"]["label"], + "model": CFG.soniox_model, + "ready": bool(CFG.soniox_api_key), + "endpoint": CFG.soniox_api_base, + "via": "SONIOX_API_KEY 直连", + }, + { + "id": "gemini", + "label": PROVIDERS["gemini"]["label"], + "model": CFG.gemini_asr_model, + "ready": has_litellm, + "endpoint": CFG.transcriptions_endpoint, + "via": "LiteLLM 统一网关", + }, + { + "id": "openai", + "label": PROVIDERS["openai"]["label"], + "model": CFG.openai_asr_model, + "ready": has_litellm, + "endpoint": CFG.transcriptions_endpoint, + "via": "LiteLLM 统一网关", + }, + { + "id": "grok", + "label": PROVIDERS["grok"]["label"], + "model": CFG.grok_asr_model, + "ready": bool(CFG.grok_api_key), + "endpoint": CFG.grok_api_base + "/stt", + "via": "XAI_API_KEY 直连", + }, + ] + + +def collect_hotwords(limit: int = 120) -> list[str]: + """Collect canonical entity names and glossary terms for ASR context biasing.""" + terms: list[str] = [] + seen: set[str] = set() + + def add(word: str) -> None: + w = word.strip() + if not w or len(w) < 2 or len(w) > 40 or w in seen: + return + if re.match(r"^\d{4}-\d{2}|^_|speaker_", w): + return + seen.add(w) + terms.append(w) + + try: + from . import lexicon + con = lexicon.connect() + for row in con.execute( + "SELECT canonical FROM entity ORDER BY synced_at DESC LIMIT ?", + (limit,)).fetchall(): + add(str(row["canonical"])) + for row in con.execute( + "SELECT DISTINCT canonical FROM alias WHERE confirmed=1 LIMIT ?", + (limit,)).fetchall(): + add(str(row["canonical"])) + con.close() + except Exception: + pass + + if CFG.glossary.exists(): + for raw in CFG.glossary.read_text(errors="ignore").splitlines(): + s = raw.strip() + if not s.startswith(("-", "*")) or re.search(r"[→⇒]|->", s): + continue + s = re.split(r"\s[—–-]\s|[:：]", + re.sub(r"^[-*+\d\.\)\s]+", "", s), maxsplit=1)[0].strip() + s = re.sub(r"\[\[|\]\]", "", s).strip() + add(s) + if len(terms) >= limit: + break + return terms[:limit] + + +def _normalize_speaker_label(raw: Any, mapping: dict[str, str]) -> str: + s = str(raw if raw is not None else "").strip() + if not s or s == "?": + return "speaker_0" + if re.match(r"^[\u4e00-\u9fff]{2,8}$", s): + return s + if re.match(r"^speaker_\d+$", s): + return s + m = re.match(r"^[sS](?:peaker)?[_ -]?(\d+)$", s) + if m: + return f"speaker_{m.group(1)}" + if s.isdigit(): + # Soniox often uses 1-based speaker numbers ("1", "2"); normalize to speaker_0..N-1 + if s not in mapping: + mapping[s] = f"speaker_{len(mapping)}" + return mapping[s] + if s not in mapping: + mapping[s] = f"speaker_{len(mapping)}" + return mapping[s] + + +def format_diarized_transcript( + segments: list[dict[str, Any]], + *, + provider: str, + model: str, + header_title: str, + voiceprint_session: str = "", + matched_names: list[str] | None = None, +) -> str: + if not segments: + raise RuntimeError(f"ASR ({provider}) response contains no segments") + speaker_map: dict[str, str] = {} + turns: list[list[str]] = [] + speakers_seen: set[str] = set() + for seg in segments: + raw_spk = seg.get("speaker") + text = str(seg.get("text") or "").strip() + if not text: + continue + spk = (_normalize_speaker_label(raw_spk, speaker_map) + if provider != "qwen3cam" + else str(raw_spk if raw_spk is not None else "speaker_0")) + speakers_seen.add(spk) + if turns and turns[-1][0] == spk: + turns[-1][1] += text + else: + turns.append([spk, text]) + if not turns: + raise RuntimeError(f"ASR ({provider}) segments contained no text") + + lines = [f""] + if voiceprint_session or provider == "qwen3cam": + lines.append( + f"") + hdr = f"# 转写({header_title})" + if matched_names: + hdr += " | 声纹命名: " + ", ".join(matched_names) + lines.append(hdr) + lines.append("") + lines.extend(f"【{spk}】{txt}\n" for spk, txt in turns) + return "\n".join(lines).strip() + "\n" + + +def transcript_from_qwen3cam_response(payload: dict[str, Any]) -> str: + segments = payload.get("segments") + if not isinstance(segments, list) or not segments: + raise RuntimeError("ASR response contains no segments") + vp = payload.get("voiceprint") or {} + sid = str(vp.get("session_id") or "") + matched = [ + f"{m.get('speaker_local')}={m.get('person_name')}" + for m in vp.get("matches", []) + if m.get("status") == "matched" + ] + return format_diarized_transcript( + segments, + provider="qwen3cam", + model="qwen3-asr-1.7b-cam++", + header_title="Qwen3-ASR 1.7B + CAM++ 分角色 + 声纹", + voiceprint_session=sid, + matched_names=matched, + ) + + +def transcript_from_soniox_response( + payload: dict[str, Any], + *, + model: str = "stt-async-v5", +) -> str: + """Convert Soniox token-level or segment-level diarized response into turns.""" + tokens = payload.get("tokens") + segments: list[dict[str, Any]] = [] + if isinstance(tokens, list) and tokens: + cur_speaker: Any = None + cur_pieces: list[str] = [] + for tok in tokens: + txt = str(tok.get("text") or "") + if not txt: + continue + spk = tok.get("speaker") + if cur_speaker is not None and spk != cur_speaker and cur_pieces: + segments.append({"speaker": cur_speaker, + "text": "".join(cur_pieces).strip()}) + cur_pieces = [] + cur_speaker = spk if spk is not None else (cur_speaker or "1") + cur_pieces.append(txt) + if cur_pieces: + segments.append({"speaker": cur_speaker or "1", + "text": "".join(cur_pieces).strip()}) + elif isinstance(payload.get("segments"), list) and payload["segments"]: + segments = payload["segments"] + elif isinstance(payload.get("text"), str) and payload["text"].strip(): + segments = [{"speaker": "speaker_0", "text": payload["text"].strip()}] + return format_diarized_transcript( + segments, + provider="soniox", + model=model, + header_title=f"Soniox {model} 分角色", + ) + + +def transcript_from_openai_response( + payload: dict[str, Any], + *, + model: str, +) -> str: + """Convert OpenAI / LiteLLM transcription response (diarized or plain) into text.""" + segments = payload.get("segments") + if isinstance(segments, list) and segments and any( + s.get("speaker") is not None for s in segments if isinstance(s, dict)): + return format_diarized_transcript( + segments, + provider="openai", + model=model, + header_title=f"OpenAI {model} 分角色", + ) + text = str(payload.get("text") or payload.get("transcript") or "").strip() + if not text and isinstance(segments, list): + text = "".join(str(s.get("text") or "") for s in segments).strip() + if not text: + raise RuntimeError(f"OpenAI ASR ({model}) returned empty transcript") + if re.search(r"【[^】]+】", text): + return ( + f"\n" + f"# 转写(OpenAI {model})\n\n{text}\n" + ) + return ( + f"\n" + f"--- 第1段 [00:00 - 结束] ---\n{text}\n" + ) + + +def transcript_from_grok_response( + payload: dict[str, Any], + *, + model: str = "grok-voice-transcribe-2.0", +) -> str: + segments = payload.get("segments") or payload.get("utterances") or payload.get("turns") + if isinstance(segments, list) and segments: + norm_segs = [] + for s in segments: + if not isinstance(s, dict): + continue + spk = s.get("speaker") + if spk is None: + spk = s.get("speaker_id") + if spk is None: + spk = "speaker_0" + txt = str(s.get("text") or s.get("transcript") or "").strip() + if txt: + norm_segs.append({"speaker": spk, "text": txt}) + if norm_segs: + return format_diarized_transcript( + norm_segs, + provider="grok", + model=model, + header_title=f"Grok {model} 分角色", + ) + words = payload.get("words") + if isinstance(words, list) and words and any( + isinstance(w, dict) and w.get("speaker") is not None for w in words): + norm_segs = [] + cur_spk = None + buf: list[str] = [] + for w in words: + if not isinstance(w, dict): + continue + txt = str(w.get("word") or w.get("text") or "") + spk = w.get("speaker") + if cur_spk is not None and spk != cur_spk and buf: + norm_segs.append({"speaker": cur_spk, "text": "".join(buf).strip()}) + buf = [] + cur_spk = spk if spk is not None else (cur_spk or "0") + buf.append(txt) + if buf: + norm_segs.append({"speaker": cur_spk or "0", "text": "".join(buf).strip()}) + if norm_segs: + return format_diarized_transcript( + norm_segs, + provider="grok", + model=model, + header_title=f"Grok {model} 分角色", + ) + text = str(payload.get("text") or payload.get("transcript") or "").strip() + if not text: + raise RuntimeError(f"Grok ASR ({model}) returned empty transcript") + return format_diarized_transcript( + [{"speaker": "speaker_0", "text": text}], + provider="grok", + model=model, + header_title=f"Grok {model}", + ) + + +def probe_duration_seconds(audio: Path) -> int: + try: + out = subprocess.check_output([ + "ffprobe", "-v", "error", "-show_entries", "format=duration", + "-of", "default=nk=1:nw=1", str(audio), + ], text=True, timeout=30).strip() + return max(1, int(float(out) + 0.999)) if out else 0 + except Exception: + return 0 + + +def _encode_multipart( + fields: dict[str, str], + file_field: str, + file_path: Path, +) -> tuple[bytes, str]: + boundary = f"----ASRBoundary{uuid.uuid4().hex}" + body = bytearray() + for name, val in fields.items(): + body.extend(f"--{boundary}\r\n".encode()) + body.extend( + f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode()) + body.extend(str(val).encode("utf-8")) + body.extend(b"\r\n") + mime = mimetypes.guess_type(file_path.name)[0] or "application/octet-stream" + body.extend(f"--{boundary}\r\n".encode()) + body.extend( + f'Content-Disposition: form-data; name="{file_field}"; ' + f'filename="{file_path.name}"\r\n'.encode()) + body.extend(f"Content-Type: {mime}\r\n\r\n".encode()) + body.extend(file_path.read_bytes()) + body.extend(f"\r\n--{boundary}--\r\n".encode()) + return bytes(body), f"multipart/form-data; boundary={boundary}" + + +def _http_multipart( + url: str, + *, + fields: dict[str, str], + file_field: str, + file_path: Path, + headers: dict[str, str] | None = None, + timeout: int = 900, +) -> dict[str, Any]: + data, content_type = _encode_multipart(fields, file_field, file_path) + req_headers = {"Content-Type": content_type, **(headers or {})} + req = urllib.request.Request(url, data=data, headers=req_headers, method="POST") + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8", errors="ignore") + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="ignore")[:300] + raise RuntimeError(f"ASR HTTP {exc.code}: {detail}") from exc + try: + return json.loads(raw) + except ValueError as exc: + raise RuntimeError(f"ASR returned invalid JSON: {raw[:200]}") from exc + + +def _http_json( + method: str, + url: str, + *, + payload: dict[str, Any] | None = None, + headers: dict[str, str] | None = None, + timeout: int = 120, +) -> dict[str, Any]: + data = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None + req_headers = {"Accept": "application/json", **(headers or {})} + if data is not None: + req_headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=req_headers, method=method) + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + raw = resp.read().decode("utf-8", errors="ignore") + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="ignore")[:300] + raise RuntimeError(f"ASR HTTP {exc.code}: {detail}") from exc + if not raw.strip(): + return {} + try: + return json.loads(raw) + except ValueError as exc: + raise RuntimeError(f"ASR returned invalid JSON: {raw[:200]}") from exc + + +def _compress_or_split_audio( + audio: Path, + tmpdir: Path, + *, + max_bytes: int = 23 * 1024 * 1024, + segment_seconds: int = 2400, + force_split: bool = False, +) -> list[tuple[Path, int, int]]: + """Return [(segment_path, start_sec, end_sec), ...] suitable for upstream API.""" + duration = probe_duration_seconds(audio) + if not force_split and audio.stat().st_size <= max_bytes and audio.suffix.lower() in { + ".mp3", ".m4a", ".wav", ".ogg", ".webm", ".flac"}: + return [(audio, 0, duration)] + + if not shutil.which("ffmpeg"): + if audio.stat().st_size <= max_bytes: + return [(audio, 0, duration)] + raise RuntimeError("audio exceeds size limit and ffmpeg is not installed") + + if not force_split: + compressed = tmpdir / "compressed.mp3" + subprocess.run([ + "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-i", str(audio), "-ar", "16000", "-ac", "1", "-b:a", "48k", + str(compressed), + ], check=True, timeout=600) + if compressed.exists() and compressed.stat().st_size <= max_bytes: + return [(compressed, 0, duration)] + + total = max(duration, 1) + step = max(60, segment_seconds) + n_segs = (total + step - 1) // step + out: list[tuple[Path, int, int]] = [] + for i in range(n_segs): + start = i * step + end = min(total, start + step) + seg = tmpdir / f"segment_{i + 1:03d}.mp3" + subprocess.run([ + "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-ss", str(start), "-t", str(step), "-i", str(audio), + "-ar", "16000", "-ac", "1", "-b:a", "64k", str(seg), + ], check=True, timeout=600) + out.append((seg, start, end)) + return out + + +def transcribe_qwen3cam(audio: Path, *, model: str = "") -> TranscribeResult: + result = subprocess.run([ + "curl", "-sS", "--fail-with-body", "--max-time", "900", + "-F", f"file=@{audio}", "-F", "language=Chinese", + "-F", "save_voiceprint_session=true", CFG.diarize_url, + ], capture_output=True, timeout=930) + if result.returncode: + detail = result.stderr.decode("utf-8", errors="ignore")[:240] + raise RuntimeError(f"qwen3cam ASR HTTP failed ({result.returncode}): {detail}") + try: + payload = json.loads(result.stdout) + except ValueError as exc: + raise RuntimeError("qwen3cam returned invalid JSON") from exc + text = transcript_from_qwen3cam_response(payload) + vp = payload.get("voiceprint") or {} + return TranscribeResult( + provider="qwen3cam", + model=model or PROVIDERS["qwen3cam"]["default_model"], + transcript_text=text, + raw_response=payload, + diarized=True, + voiceprint_session=str(vp.get("session_id") or ""), + ) + + +def build_soniox_context(title: str = "", user_context: str = "") -> dict[str, Any]: + ctx: dict[str, Any] = {} + general = [{"key": "domain", "value": "Biotech / Investment / Corporate Meeting"}] + if title.strip(): + general.append({"key": "title", "value": title.strip()[:120]}) + ctx["general"] = general + if user_context.strip(): + ctx["text"] = user_context.strip()[:2000] + terms = collect_hotwords(120) + if terms: + ctx["terms"] = terms + return ctx + + +def transcribe_soniox( + audio: Path, + *, + model: str = "", + title: str = "", + user_context: str = "", + poll_interval: float = 3.0, + max_wait_seconds: float = 900.0, +) -> TranscribeResult: + key = CFG.soniox_api_key + if not key: + raise RuntimeError("SONIOX_API_KEY is not configured") + base = CFG.soniox_api_base + chosen_model = model or CFG.soniox_model or "stt-async-v5" + headers = {"Authorization": f"Bearer {key}"} + + file_resp = _http_multipart( + f"{base}/files", + fields={}, + file_field="file", + file_path=audio, + headers=headers, + timeout=600, + ) + file_id = str(file_resp.get("id") or "") + if not file_id: + raise RuntimeError(f"Soniox file upload returned no id: {file_resp}") + + transcription_id = "" + try: + job_payload: dict[str, Any] = { + "model": chosen_model, + "file_id": file_id, + "enable_speaker_diarization": True, + "enable_language_identification": True, + "language_hints": ["zh", "en"], + "context": build_soniox_context(title, user_context), + } + job_resp = _http_json( + "POST", f"{base}/transcriptions", + payload=job_payload, headers=headers, timeout=60, + ) + transcription_id = str(job_resp.get("id") or "") + if not transcription_id: + raise RuntimeError(f"Soniox transcription create returned no id: {job_resp}") + + deadline = time.time() + max_wait_seconds + status_resp: dict[str, Any] = {} + while time.time() < deadline: + status_resp = _http_json( + "GET", f"{base}/transcriptions/{transcription_id}", + headers=headers, timeout=60, + ) + st = str(status_resp.get("status") or "").lower() + if st == "completed": + break + if st in ("error", "failed"): + err = status_resp.get("error_message") or status_resp + raise RuntimeError(f"Soniox transcription failed: {err}") + time.sleep(poll_interval) + else: + raise RuntimeError("Soniox transcription timed out waiting for completion") + + if "tokens" in status_resp or "segments" in status_resp: + transcript_payload = status_resp + else: + transcript_payload = _http_json( + "GET", f"{base}/transcriptions/{transcription_id}/transcript", + headers=headers, timeout=120, + ) + text = transcript_from_soniox_response( + transcript_payload, model=chosen_model) + return TranscribeResult( + provider="soniox", + model=chosen_model, + transcript_text=text, + raw_response=transcript_payload, + diarized=True, + ) + finally: + for cleanup_url in ( + f"{base}/transcriptions/{transcription_id}" if transcription_id else "", + f"{base}/files/{file_id}" if file_id else "", + ): + if cleanup_url: + try: + _http_json("DELETE", cleanup_url, headers=headers, timeout=15) + except Exception: + pass + + +def transcribe_gemini( + audio: Path, + *, + model: str = "", + title: str = "", + user_context: str = "", +) -> TranscribeResult: + chosen_model = model or CFG.gemini_asr_model or "aistudio/gemini-3.5-transcribe" + headers = {"Authorization": f"Bearer {CFG.api_key()}"} + with tempfile.TemporaryDirectory(prefix="asr-gemini-") as tmp: + segments = _compress_or_split_audio( + audio, Path(tmp), + max_bytes=18 * 1024 * 1024, + segment_seconds=CFG.segment_seconds, + force_split=probe_duration_seconds(audio) > CFG.segment_seconds + 60, + ) + responses: list[dict[str, Any]] = [] + parts: list[str] = [f""] + for idx, (seg_path, start, end) in enumerate(segments, 1): + resp = _http_multipart( + CFG.transcriptions_endpoint, + fields={"model": chosen_model, "response_format": "json"}, + file_field="file", + file_path=seg_path, + headers=headers, + timeout=600, + ) + responses.append(resp) + seg_text = str(resp.get("text") or resp.get("transcript") or "").strip() + if not seg_text: + raise RuntimeError(f"Gemini segment {idx} returned empty text") + parts.append( + f"--- 第{idx}段 [{start // 60:02d}:{start % 60:02d} - " + f"{end // 60:02d}:{end % 60:02d}] ---\n{seg_text}\n") + return TranscribeResult( + provider="gemini", + model=chosen_model, + transcript_text="\n\n".join(parts).strip() + "\n", + raw_response={"segments": responses}, + diarized=False, + ) + + +def transcribe_openai( + audio: Path, + *, + model: str = "", + title: str = "", + user_context: str = "", +) -> TranscribeResult: + chosen_model = model or CFG.openai_asr_model or "openai/gpt-4o-transcribe-diarize" + headers = {"Authorization": f"Bearer {CFG.api_key()}"} + is_diarize = "diarize" in chosen_model.lower() + with tempfile.TemporaryDirectory(prefix="asr-openai-") as tmp: + segments = _compress_or_split_audio( + audio, Path(tmp), + max_bytes=23 * 1024 * 1024, + segment_seconds=2400, + force_split=False, + ) + responses: list[dict[str, Any]] = [] + all_diarized_segs: list[dict[str, Any]] = [] + plain_parts: list[str] = [f""] + hotwords = collect_hotwords(80) + prompt = "" + if not is_diarize and hotwords: + prompt = "会议专有名词参考：" + "、".join(hotwords)[:800] + + for idx, (seg_path, start, end) in enumerate(segments, 1): + fields: dict[str, str] = { + "model": chosen_model, + "language": "zh", + "response_format": "diarized_json" if is_diarize else "json", + } + if prompt: + fields["prompt"] = prompt + try: + resp = _http_multipart( + CFG.transcriptions_endpoint, + fields=fields, + file_field="file", + file_path=seg_path, + headers=headers, + timeout=600, + ) + except RuntimeError: + if is_diarize: + # Fallback if gateway enforces response_format=json + fields["response_format"] = "json" + resp = _http_multipart( + CFG.transcriptions_endpoint, + fields=fields, + file_field="file", + file_path=seg_path, + headers=headers, + timeout=600, + ) + else: + raise + responses.append(resp) + if isinstance(resp.get("segments"), list) and resp["segments"]: + all_diarized_segs.extend(resp["segments"]) + else: + txt = str(resp.get("text") or resp.get("transcript") or "").strip() + if txt: + plain_parts.append( + f"--- 第{idx}段 [{start // 60:02d}:{start % 60:02d} - " + f"{end // 60:02d}:{end % 60:02d}] ---\n{txt}\n") + + if all_diarized_segs: + text = transcript_from_openai_response( + {"segments": all_diarized_segs}, model=chosen_model) + diarized = True + elif len(responses) == 1: + text = transcript_from_openai_response(responses[0], model=chosen_model) + diarized = "【" in text + else: + if len(plain_parts) <= 1: + raise RuntimeError(f"OpenAI ASR ({chosen_model}) returned no text") + text = "\n\n".join(plain_parts).strip() + "\n" + diarized = False + + return TranscribeResult( + provider="openai", + model=chosen_model, + transcript_text=text, + raw_response={"segments": responses} if len(responses) > 1 else responses[0], + diarized=diarized, + ) + + +def transcribe_grok( + audio: Path, + *, + model: str = "", + title: str = "", + user_context: str = "", +) -> TranscribeResult: + key = CFG.grok_api_key + if not key: + raise RuntimeError("XAI_API_KEY is not configured") + chosen_model = model or CFG.grok_asr_model or "grok-voice-transcribe-2.0" + headers = {"Authorization": f"Bearer {key}"} + url = f"{CFG.grok_api_base}/stt" + with tempfile.TemporaryDirectory(prefix="asr-grok-") as tmp: + segments = _compress_or_split_audio( + audio, Path(tmp), max_bytes=23 * 1024 * 1024, segment_seconds=2400) + all_texts: list[str] = [] + raw_list: list[dict[str, Any]] = [] + for seg_path, _start, _end in segments: + resp = _http_multipart( + url, + fields={"model": chosen_model, "diarize": "true", "language": "zh"}, + file_field="file", + file_path=seg_path, + headers=headers, + timeout=600, + ) + raw_list.append(resp) + all_texts.append(transcript_from_grok_response(resp, model=chosen_model)) + text = all_texts[0] if len(all_texts) == 1 else "\n\n".join(all_texts) + return TranscribeResult( + provider="grok", + model=chosen_model, + transcript_text=text, + raw_response=raw_list[0] if len(raw_list) == 1 else {"chunks": raw_list}, + diarized=True, + ) + + +def transcribe_audio( + audio: Path, + *, + provider: str | None = None, + model: str = "", + title: str = "", + user_context: str = "", +) -> TranscribeResult: + """Dispatch transcription to the selected provider and return normalized result.""" + prov = normalize_provider_name(provider) + log(f"transcribe start provider={prov} model={model or 'default'} file={audio.name}") + if prov == "qwen3cam": + return transcribe_qwen3cam(audio, model=model) + if prov == "soniox": + return transcribe_soniox(audio, model=model, title=title, user_context=user_context) + if prov == "gemini": + return transcribe_gemini(audio, model=model, title=title, user_context=user_context) + if prov == "openai": + return transcribe_openai(audio, model=model, title=title, user_context=user_context) + if prov == "grok": + return transcribe_grok(audio, model=model, title=title, user_context=user_context) + raise ValueError(f"unsupported ASR provider: {prov}") diff --git a/servers/tools-pve/asr-pipeline/asrpipe/refine.py b/servers/tools-pve/asr-pipeline/asrpipe/refine.py index f2eca49..5d14438 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/refine.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/refine.py @@ -28,6 +28,10 @@ from .log import log SOURCE_DETAIL = { "qwen3cam": "Qwen3-ASR 1.7B + CAM++ 分角色 + 声纹自动命名", + "soniox": "Soniox stt-async-v5 线上分角色转写", + "gemini": "Gemini 3.5 Transcribe (LiteLLM)", + "openai": "OpenAI GPT-4o Transcribe (LiteLLM)", + "grok": "Grok Voice Transcribe 2.0", "vibevoice": "VibeVoice 分角色", "qwen3-plain": "Qwen3-ASR 1.7B", } diff --git a/servers/tools-pve/asr-pipeline/asrpipe/review.py b/servers/tools-pve/asr-pipeline/asrpipe/review.py index ed46e2d..b185026 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/review.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/review.py @@ -185,7 +185,8 @@ def _voiceprint_name(value: str) -> str: def _confirm_voiceprint(draft_path: str, speaker_id: str, value: str) -> bool: - """Persist an explicit speaker identity and immediately enroll its session.""" + """Persist an explicit speaker identity and, when a voiceprint session exists, + enroll it against the acoustic voiceprint service.""" name = _voiceprint_name(value) if not name: return False @@ -196,8 +197,6 @@ def _confirm_voiceprint(draft_path: str, speaker_id: str, value: str) -> bool: return False text = path.read_text(encoding="utf-8", errors="ignore") session_id = _frontmatter_value(text, "voiceprint_session") - if not session_id: - return False fm = re.match(r"(?s)^---\n(.*?)\n---", text) if not fm: return False @@ -205,27 +204,34 @@ def _confirm_voiceprint(draft_path: str, speaker_id: str, value: str) -> bool: speaker_line = rf"(?m)^(\s+{re.escape(speaker_id)}:)\s*.*$" if re.search(speaker_line, block): block = re.sub(speaker_line, rf"\1 {name}", block) + elif re.search(r"(?m)^voiceprint_speakers:\s*(?:\{\})?\s*$", block): + block = re.sub( + r"(?m)^voiceprint_speakers:\s*(?:\{\})?\s*$", + f"voiceprint_speakers:\n {speaker_id}: {name}", block, count=1) elif re.search(r"(?m)^voiceprint_speakers:\s*$", block): block = re.sub( r"(?m)^(voiceprint_speakers:\s*)$", rf"\1\n {speaker_id}: {name}", block, count=1) else: block += f"\nvoiceprint_speakers:\n {speaker_id}: {name}" - if re.search(r"(?m)^voiceprint_enroll:", block): - block = re.sub(r"(?m)^voiceprint_enroll:.*$", - "voiceprint_enroll: true", block) - else: - block += "\nvoiceprint_enroll: true" + if session_id: + if re.search(r"(?m)^voiceprint_enroll:", block): + block = re.sub(r"(?m)^voiceprint_enroll:.*$", + "voiceprint_enroll: true", block) + else: + block += "\nvoiceprint_enroll: true" text = text[:fm.start()] + "---\n" + block + "\n---" + text[fm.end():] from . import vault vault.write_draft_file(path, text) + if not session_id or not CFG.voiceprint_enroll_url: + return False body = json.dumps({ "session_id": session_id, "speaker_id": int(speaker_id.split("_")[1]), "name": name, }).encode() req = urllib.request.Request( - "http://192.168.50.100:10099/voiceprints/enroll", data=body, + CFG.voiceprint_enroll_url, data=body, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=20) as response: diff --git a/servers/tools-pve/asr-pipeline/asrpipe/review_web.py b/servers/tools-pve/asr-pipeline/asrpipe/review_web.py index e14875c..33b2b9e 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/review_web.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/review_web.py @@ -29,6 +29,8 @@ def _url(token: str) -> str: def request_start(job_id: str = "") -> bool: """Ensure control service is running and send a task-scoped review URL.""" + if os.environ.get("ASR_ROLE"): + return True try: r = subprocess.run(["systemctl", "--user", "start", UNIT], capture_output=True, text=True, timeout=15) diff --git a/servers/tools-pve/asr-pipeline/asrpipe/segment.py b/servers/tools-pve/asr-pipeline/asrpipe/segment.py index 4175f6b..e0ed72f 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/segment.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/segment.py @@ -22,6 +22,7 @@ import re from dataclasses import dataclass, field VOICEPRINT_HDR = re.compile(r"^") SEGMENT_BANNER = re.compile(r"^---\s*第\d+段.*---\s*$") TURN_TAG = re.compile(r"^【([^】]+)】\s*(.*)$", re.S) VIBE_TAG = re.compile(r"^\[(\d{1,2}:\d{2})\]\s*(S\d+)\s*[:：]\s*(.*)$", re.S) @@ -95,6 +96,11 @@ class Chunk: def detect_backend(raw: str) -> str: head = raw[:500] + m_prov = PROVIDER_HDR.search(head) + if m_prov: + prov = m_prov.group(1).strip().lower() + if prov in {"qwen3cam", "soniox", "gemini", "openai", "grok", "vibevoice"}: + return prov if "CAM++" in head or TURN_TAG.match(raw.strip().splitlines()[-1] if raw.strip() else ""): return "qwen3cam" if re.search(r"^\[\d{1,2}:\d{2}\]\s*S\d+\s*[:：]", raw, re.M): @@ -108,7 +114,7 @@ def parse(raw: str) -> tuple[list[Turn], dict]: """Split a transcript into turns plus header metadata.""" meta: dict = { "voiceprint_session": "", "voiceprint_named": {}, "banners": [], - "intake": {}, + "intake": {}, "asr_provider": "", "asr_model": "", } lines = raw.splitlines() body_lines = [] @@ -120,6 +126,11 @@ def parse(raw: str) -> tuple[list[Turn], dict]: except (TypeError, ValueError): pass continue + m_prov = PROVIDER_HDR.match(ln) + if m_prov: + meta["asr_provider"] = m_prov.group(1) or "" + meta["asr_model"] = m_prov.group(2) or "" + continue if VOICEPRINT_HDR.match(ln): m = re.search(r"voiceprint_session:\s*(\S+)", ln) if m: diff --git a/servers/tools-pve/asr-pipeline/asrservice/__init__.py b/servers/tools-pve/asr-pipeline/asrservice/__init__.py index a8a2f6f..926906f 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/__init__.py +++ b/servers/tools-pve/asr-pipeline/asrservice/__init__.py @@ -1 +1 @@ -"""Standalone ASR service for web, Agent API, and SSH drop-folder intake.""" +"""Standalone WebUI, REST API, and background worker service for Meeting ASR.""" diff --git a/servers/tools-pve/asr-pipeline/asrservice/config.py b/servers/tools-pve/asr-pipeline/asrservice/config.py index facd811..3eedd87 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/config.py +++ b/servers/tools-pve/asr-pipeline/asrservice/config.py @@ -3,23 +3,27 @@ from __future__ import annotations import os from pathlib import Path +from asrpipe.config import CFG + DATA = Path(os.getenv("ASR_DATA_DIR", "/data")) INBOX = DATA / "inbox" PROCESSING = DATA / "processing" PROCESSED = DATA / "processed" FAILED = DATA / "failed" TRANSCRIPTS = DATA / "transcripts" -STATE = Path(os.getenv("ASR_STATE_DIR", "/state")) -DRAFTS = DATA / "artifacts" / "drafts" -RESULTS = DATA / "artifacts" / "results" +STATE = Path(os.getenv("ASR_STATE_DIR", str(CFG.state_dir))) +DRAFTS = Path(os.getenv("ASR_DRAFTS_DIR", str(DATA / "artifacts" / "drafts"))) +RESULTS = Path(os.getenv("ASR_RESULTS_DIR", str(DATA / "artifacts" / "results"))) LOGS = DATA / "logs" META = STATE / "uploads" -REVISION_QUEUE = STATE / "revision-queue" +REVISION_QUEUE = Path(os.getenv("ASR_REVISION_QUEUE_DIR", str(STATE / "revision-queue"))) DB = STATE / "intake.db" SUPPORTED = {".m4a", ".mp3", ".wav", ".mp4", ".aac", ".ogg", ".opus", ".webm", ".flac"} PUBLIC_URL = os.getenv("ASR_PUBLIC_URL", "http://192.168.50.49:8788").rstrip("/") AGENT_TOKEN = os.getenv("ASR_AGENT_TOKEN", "") -DIARIZE_URL = os.getenv("ASR_DIARIZE_URL", "http://192.168.50.100:10099/asr") +WEB_PASSWORD = os.getenv("ASR_WEB_PASSWORD", "") +DIARIZE_URL = os.getenv("ASR_DIARIZE_URL", CFG.diarize_url) +DEFAULT_PROVIDER = os.getenv("ASR_PROVIDER", CFG.asr_provider) MAX_UPLOAD = int(os.getenv("ASR_UPLOAD_MAX_BYTES", str(4 * 1024**3))) POLL_SECONDS = max(1, int(os.getenv("ASR_POLL_SECONDS", "5"))) STABLE_SECONDS = max(1, int(os.getenv("ASR_STABLE_SECONDS", "10"))) @@ -27,8 +31,13 @@ MAX_ATTEMPTS = max(1, int(os.getenv("ASR_MAX_ATTEMPTS", "3"))) def ensure_dirs() -> None: - for path in (INBOX, PROCESSING, PROCESSED, FAILED, TRANSCRIPTS, DRAFTS, - RESULTS, STATE, LOGS, META, REVISION_QUEUE, DATA / "cache", - DATA / "artifacts" / "raw", DATA / "vault" / "minutes", - DATA / "vault" / "People", DATA / "vault" / "Companies"): + # Keep asrpipe CFG aligned with container/service paths when running under asrservice + CFG.state_dir + for path in ( + INBOX, PROCESSING, PROCESSED, FAILED, TRANSCRIPTS, DRAFTS, + RESULTS, STATE, LOGS, META, REVISION_QUEUE, DATA / "cache", + DATA / "artifacts" / "raw", DATA / "artifacts" / "results", + DATA / "vault" / "minutes", DATA / "vault" / "People", + DATA / "vault" / "Companies", DATA / "vault" / "Projects", + ): path.mkdir(parents=True, exist_ok=True) diff --git a/servers/tools-pve/asr-pipeline/asrservice/db.py b/servers/tools-pve/asr-pipeline/asrservice/db.py index 948e3a1..176a488 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/db.py +++ b/servers/tools-pve/asr-pipeline/asrservice/db.py @@ -1,14 +1,15 @@ from __future__ import annotations -import json import sqlite3 import time from typing import Any -from .config import DB, ensure_dirs +from .config import DB, DEFAULT_PROVIDER, ensure_dirs -STATUSES = {"uploading", "queued", "transcribing", "refining", "review", - "publishing", "complete", "failed"} +STATUSES = { + "uploading", "queued", "transcribing", "refining", "review", + "publishing", "complete", "failed", +} def connect() -> sqlite3.Connection: @@ -26,6 +27,8 @@ def connect() -> sqlite3.Connection: draft TEXT NOT NULL DEFAULT '', result TEXT NOT NULL DEFAULT '', source TEXT NOT NULL DEFAULT 'web', sha256 TEXT NOT NULL DEFAULT '', attempts INTEGER NOT NULL DEFAULT 0, error_code TEXT NOT NULL DEFAULT '', + asr_provider TEXT NOT NULL DEFAULT 'qwen3cam', + asr_model TEXT NOT NULL DEFAULT '', created_at REAL NOT NULL, updated_at REAL NOT NULL ); CREATE TABLE IF NOT EXISTS job_event( @@ -37,9 +40,14 @@ def connect() -> sqlite3.Connection: """) cols = {r[1] for r in con.execute("PRAGMA table_info(intake_job)")} additions = { - "draft": "TEXT NOT NULL DEFAULT ''", "result": "TEXT NOT NULL DEFAULT ''", - "source": "TEXT NOT NULL DEFAULT 'web'", "sha256": "TEXT NOT NULL DEFAULT ''", - "attempts": "INTEGER NOT NULL DEFAULT 0", "error_code": "TEXT NOT NULL DEFAULT ''", + "draft": "TEXT NOT NULL DEFAULT ''", + "result": "TEXT NOT NULL DEFAULT ''", + "source": "TEXT NOT NULL DEFAULT 'web'", + "sha256": "TEXT NOT NULL DEFAULT ''", + "attempts": "INTEGER NOT NULL DEFAULT 0", + "error_code": "TEXT NOT NULL DEFAULT ''", + "asr_provider": f"TEXT NOT NULL DEFAULT '{DEFAULT_PROVIDER}'", + "asr_model": "TEXT NOT NULL DEFAULT ''", } for name, ddl in additions.items(): if name not in cols: @@ -48,23 +56,40 @@ def connect() -> sqlite3.Connection: return con -def create_job(job_id: str, original_name: str, stored_name: str, *, title: str = "", - context: str = "", size: int = 0, source: str = "web") -> dict[str, Any]: +def create_job( + job_id: str, + original_name: str, + stored_name: str, + *, + title: str = "", + context: str = "", + size: int = 0, + source: str = "web", + asr_provider: str = "", + asr_model: str = "", +) -> dict[str, Any]: now = time.time() + prov = (asr_provider or DEFAULT_PROVIDER or "qwen3cam").strip().lower() with connect() as con: - con.execute("INSERT INTO intake_job(id,original_name,stored_name,title,context,size," - "status,source,created_at,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?)", - (job_id, original_name, stored_name, title, context, size, - "queued", source, now, now)) - con.execute("INSERT INTO job_event(job_id,status,detail,created_at) VALUES(?,?,?,?)", - (job_id, "queued", "录音已进入队列", now)) + con.execute( + "INSERT INTO intake_job(id,original_name,stored_name,title,context,size," + "status,source,asr_provider,asr_model,created_at,updated_at) " + "VALUES(?,?,?,?,?,?,?,?,?,?,?,?)", + (job_id, original_name, stored_name, title, context, size, + "queued", source, prov, asr_model, now, now)) + con.execute( + "INSERT INTO job_event(job_id,status,detail,created_at) VALUES(?,?,?,?)", + (job_id, "queued", f"录音已进入队列 (ASR: {prov})", now)) return get_job(job_id) or {} def update_job(job_id: str, status: str, detail: str = "", **fields: Any) -> bool: if status not in STATUSES: raise ValueError(f"invalid status: {status}") - allowed = {"transcript", "draft", "result", "sha256", "attempts", "error_code"} + allowed = { + "transcript", "draft", "result", "sha256", "attempts", + "error_code", "asr_provider", "asr_model", + } changes = {k: v for k, v in fields.items() if k in allowed and v is not None} sets = ["status=?", "updated_at=?", "detail=?"] values: list[Any] = [status, time.time(), detail[:1000]] @@ -75,8 +100,9 @@ def update_job(job_id: str, status: str, detail: str = "", **fields: Any) -> boo with connect() as con: cur = con.execute(f"UPDATE intake_job SET {','.join(sets)} WHERE id=?", values) if cur.rowcount: - con.execute("INSERT INTO job_event(job_id,status,detail,created_at) VALUES(?,?,?,?)", - (job_id, status, detail[:1000], time.time())) + con.execute( + "INSERT INTO job_event(job_id,status,detail,created_at) VALUES(?,?,?,?)", + (job_id, status, detail[:1000], time.time())) return bool(cur.rowcount) @@ -88,8 +114,9 @@ def get_job(job_id: str) -> dict[str, Any] | None: def find_by_stored(stored_name: str) -> dict[str, Any] | None: with connect() as con: - row = con.execute("SELECT * FROM intake_job WHERE stored_name=? ORDER BY created_at DESC LIMIT 1", - (stored_name,)).fetchone() + row = con.execute( + "SELECT * FROM intake_job WHERE stored_name=? ORDER BY created_at DESC LIMIT 1", + (stored_name,)).fetchone() return dict(row) if row else None diff --git a/servers/tools-pve/asr-pipeline/asrservice/healthcheck.py b/servers/tools-pve/asr-pipeline/asrservice/healthcheck.py index 6477a40..e05520f 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/healthcheck.py +++ b/servers/tools-pve/asr-pipeline/asrservice/healthcheck.py @@ -2,4 +2,6 @@ import time from .config import STATE heartbeat = STATE / "worker.heartbeat" -raise SystemExit(0 if heartbeat.exists() and time.time() - heartbeat.stat().st_mtime < 90 else 1) +raise SystemExit( + 0 if heartbeat.exists() and time.time() - heartbeat.stat().st_mtime < 90 else 1 +) diff --git a/servers/tools-pve/asr-pipeline/asrservice/notify.py b/servers/tools-pve/asr-pipeline/asrservice/notify.py index a76e501..772b839 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/notify.py +++ b/servers/tools-pve/asr-pipeline/asrservice/notify.py @@ -15,21 +15,28 @@ def send(status: str, filename: str = "", detail: str = "") -> bool: chat = os.getenv("TELEGRAM_CHAT_ID", "") if not token or not chat: return False - body = parse.urlencode({"chat_id": chat, - "text": "\n".join(x for x in (status, filename, detail) if x)}).encode() - req = request.Request(f"https://api.telegram.org/bot{token}/sendMessage", - data=body, method="POST") + body = parse.urlencode({ + "chat_id": chat, + "text": "\n".join(x for x in (status, filename, detail) if x), + }).encode() + req = request.Request( + f"https://api.telegram.org/bot{token}/sendMessage", + data=body, method="POST") elif backend == "hermes": url = os.getenv("HERMES_WEBHOOK_URL", "") secret = os.getenv("HERMES_WEBHOOK_SECRET", "") if not url or not secret: return False - body = json.dumps({"status": status, "filename": filename, "detail": detail}, - ensure_ascii=False).encode() + body = json.dumps( + {"status": status, "filename": filename, "detail": detail}, + ensure_ascii=False).encode() sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest() - req = request.Request(url, data=body, method="POST", - headers={"Content-Type": "application/json", - "X-Webhook-Signature": sig}) + req = request.Request( + url, data=body, method="POST", + headers={ + "Content-Type": "application/json", + "X-Webhook-Signature": sig, + }) else: return False with request.urlopen(req, timeout=10) as response: diff --git a/servers/tools-pve/asr-pipeline/asrservice/web.py b/servers/tools-pve/asr-pipeline/asrservice/web.py index 28756b4..9a71e75 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/web.py +++ b/servers/tools-pve/asr-pipeline/asrservice/web.py @@ -1,178 +1,955 @@ from __future__ import annotations -import hashlib, hmac, html, json, os, re, time, urllib.parse, uuid +import hashlib +import hmac +import html +import json +import mimetypes +import os +import re +import shutil +import threading +import time +import urllib.parse +import urllib.request +import uuid +from http import cookies from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from asrpipe import review +from asrpipe import graph, lexicon, providers, review, vault +from asrpipe.config import CFG from . import db -from .config import AGENT_TOKEN, DRAFTS, FAILED, INBOX, MAX_UPLOAD, PUBLIC_URL, RESULTS, STATE, SUPPORTED, TRANSCRIPTS, ensure_dirs +from .config import ( + AGENT_TOKEN, DEFAULT_PROVIDER, DRAFTS, FAILED, INBOX, MAX_UPLOAD, + PROCESSED, PROCESSING, PUBLIC_URL, RESULTS, REVISION_QUEUE, STATE, + SUPPORTED, TRANSCRIPTS, WEB_PASSWORD, ensure_dirs, +) -LABELS = {"queued":"等待处理","transcribing":"正在转写","refining":"正在整理","review":"等待确认","publishing":"正在定稿","complete":"已完成","failed":"处理失败"} +LABELS = { + "uploading": "上传中", + "queued": "等待处理", + "transcribing": "正在转写", + "refining": "正在整理", + "review": "等待确认", + "publishing": "正在定稿", + "complete": "已完成", + "failed": "处理失败", +} -def safe(value, limit): - return re.sub(r"\s+", " ", re.sub(r"[\x00-\x1f\x7f]+", " ", value or "")).strip()[:limit] -def stem(value): - return re.sub(r"[^\w\u3400-\u9fff.-]+", "-", safe(value, 80)).strip("-.") or "录音" +def safe(value: str, limit: int) -> str: + return re.sub( + r"\s+", " ", re.sub(r"[\x00-\x1f\x7f]+", " ", value or "") + ).strip()[:limit] -def public(row, detailed=False): - out = {k: row.get(k) for k in ("id","original_name","title","size","status","detail","source","sha256","attempts","error_code","created_at","updated_at")} + +def stem(value: str) -> str: + return ( + re.sub(r"[^\w\u3400-\u9fff.-]+", "-", safe(value, 80)).strip("-.") + or "录音" + ) + + +def find_audio_file(job: dict) -> Path | None: + stored = str(job.get("stored_name") or "") + if not stored: + return None + jid = str(job.get("id") or "")[:8] + base_stem = Path(stored).stem + suffix = Path(stored).suffix + for folder in (PROCESSED, PROCESSING, INBOX, FAILED): + if not folder.exists(): + continue + direct = folder / stored + if direct.is_file(): + return direct + if jid: + alt = folder / f"{base_stem}-{jid}{suffix}" + if alt.is_file(): + return alt + return None + + +def public(row: dict, detailed: bool = False) -> dict: + keys = ( + "id", "original_name", "title", "size", "status", "detail", "source", + "sha256", "attempts", "error_code", "asr_provider", "asr_model", + "created_at", "updated_at", + ) + out = {k: row.get(k) for k in keys} + out["asr_provider"] = out.get("asr_provider") or DEFAULT_PROVIDER out["status_label"] = LABELS.get(row["status"], row["status"]) out["web_url"] = f"{PUBLIC_URL}/jobs/{row['id']}/" out["review_url"] = f"{PUBLIC_URL}/jobs/{row['id']}/review/" - out["artifacts"] = {name:f"/api/v1/jobs/{row['id']}/artifacts/{name}" for name in ("transcript","draft","result") if row.get(name)} + out["artifacts"] = { + name: f"/api/v1/jobs/{row['id']}/artifacts/{name}" + for name in ("transcript", "draft", "result") + if row.get(name) + } if detailed: out["context"] = row.get("context", "") out["events"] = db.events(row["id"]) return out -def pending(job): - found=[] + +def pending(job: dict | None = None) -> list[dict]: + if not job: + return review.pending(200) + found = [] for row in review.pending(1000): - payload=json.loads(row["payload"] or "{}") - if ((job.get("transcript") and row["recording"]==job["transcript"]) - or (job.get("draft") and payload.get("draft_path")==job["draft"])): + payload = json.loads(row["payload"] or "{}") + if ( + (job.get("transcript") and row["recording"] == job["transcript"]) + or (job.get("draft") and payload.get("draft_path") == job["draft"]) + or payload.get("job_id") == job["id"] + ): found.append(row) return found -def shell(title, body, script=""): - return f'''{html.escape(title)}
ASR 录音工作台
{body}
{script}'''.encode() -def home(): - body='''

上传录音

也可通过 SSH 上传到 tools-pve:/srv/asr/inbox，完成后自动处理。

最近任务

自动刷新
读取中…
''' - script='''''' - return shell("ASR 录音工作台",body,script) +def _session_token() -> str: + secret = (WEB_PASSWORD or AGENT_TOKEN or "local").encode("utf-8") + return hmac.new(secret, b"asr-web-session-v1", hashlib.sha256).hexdigest() -def job_page(job): - events=''.join(f"

{html.escape(LABELS.get(e['status'],e['status']))}
{time.strftime('%Y-%m-%d %H:%M:%S',time.localtime(e['created_at']))} · {html.escape(e['detail'])}

" for e in db.events(job["id"])) - links=' · '.join(f'{label}' for n,label in (("transcript","转写"),("draft","草稿"),("result","正式稿")) if job.get(n)) - retry=f'
' if job["status"]=="failed" else "" - review_link=f'

继续人工确认

' if pending(job) else "" - return shell("任务详情",f"

{html.escape(job.get('title') or job['original_name'])}

{html.escape(LABELS.get(job['status'],job['status']))}

{html.escape(job.get('detail',''))}

{links}

{review_link}{retry}

处理时间线

{events}
") -def review_page(job): - rows=pending(job) - if not rows: return shell("人工确认","

当前没有待确认问题

返回任务

") - row=rows[0]; payload=json.loads(row["payload"] or "{}"); choices=payload.get("candidates") or [] - buttons=''.join(f'' for v in choices)+'' - draft=""; path=Path(str(payload.get("draft_path") or "")) - try: path.resolve().relative_to(DRAFTS.resolve()); draft=path.read_text(encoding="utf-8") - except (OSError,ValueError): pass - return shell("人工确认",f"

剩余 {len(rows)} 个问题

{html.escape(row['question'])}

{html.escape(str(payload.get('sentence') or ''))}

查看草稿
{html.escape(draft)}
{buttons}

") +def shell(title: str, body: str, script: str = "") -> bytes: + return f""" + +{html.escape(title)} · Meeting ASR
+ASR 录音工作台 + +
{body}
{script}""".encode("utf-8") + + +def _provider_options(selected: str = "") -> str: + sel = providers.normalize_provider_name(selected or DEFAULT_PROVIDER) + opts = [] + for p in providers.provider_status(): + pid = p["id"] + is_sel = " selected" if pid == sel else "" + status_note = "" if p["ready"] else " (待配置密钥)" + opts.append( + f'' + ) + return "".join(opts) + + +def login_page(err: str = "") -> bytes: + alert = f'

{html.escape(err)}

' if err else "" + body = f"""
+

访问验证

请输入工作台访问口令。

{alert} +
+ + +

+
""" + return shell("登录", body) + + +def home() -> bytes: + options = _provider_options(DEFAULT_PROVIDER) + body = f"""

上传会议 / 通话录音

+

支持 m4a、mp3、wav、mp4、aac、ogg、opus、webm、flac。支持内网 Qwen3-ASR+CAM++ 及线上 Soniox / Gemini 3.5 / OpenAI / Grok。

+
+
+
+
+
+
+
+
+
+ + +

+ +
+

最近任务

+
+
读取中…
""" + script = """""" + return shell("任务工作台", body, script) + + +def _read_artifact_text(job: dict, name: str) -> str: + roots = {"transcript": TRANSCRIPTS, "draft": DRAFTS, "result": RESULTS} + raw_path = str(job.get(name) or "") + if not raw_path: + return "" + path = Path(raw_path) + allowed_roots = [roots[name].resolve(), CFG.minutes.resolve(), CFG.drafts.resolve()] + for root in allowed_roots: + try: + path.resolve().relative_to(root) + if path.is_file(): + return path.read_text(encoding="utf-8", errors="ignore") + except (OSError, ValueError): + continue + return "" + + +def job_page(job: dict) -> bytes: + jid = job["id"] + events_html = "".join( + f"

{html.escape(LABELS.get(e['status'], e['status']))} " + f"{time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(e['created_at']))}" + f"
{html.escape(e['detail'])}

" + for e in db.events(jid) + ) + links = " · ".join( + f'下载{label} (.md/.txt)' + for n, label in (("result", "正式纪要"), ("draft", "精修草稿"), ("transcript", "原始转写")) + if job.get(n) + ) + audio_html = ( + f'' + if find_audio_file(job) + else '

原音频文件不在活动目录

' + ) + pend = pending(job) + review_banner = ( + f'
当前有 {len(pend)} 个问题待你确认 · ' + f'立即前往确认页面 →
' + if pend else "" + ) + options = _provider_options(job.get("asr_provider") or DEFAULT_PROVIDER) + retry_html = f"""
+切换 ASR 模型重跑 / 重新处理该录音 +
+
+
+
+
+

+
""" + + sections = [] + for key, label in (("result", "正式纪要"), ("draft", "精修草稿"), ("transcript", "原始转写")): + content = _read_artifact_text(job, key) + if not content: + continue + open_attr = " open" if key in ("result", "draft") else "" + edit_form = "" + if key in ("draft", "result"): + edit_form = f"""
在线微调 {label} Markdown +
+ + +

""" + sections.append( + f'
' + f'{label}
{html.escape(content)}
{edit_form}
' + ) + + prov_tag = f"{job.get('asr_provider') or DEFAULT_PROVIDER}" + if job.get("asr_model"): + prov_tag += f":{job['asr_model']}" + body = f"""
+

{html.escape(job.get('title') or job['original_name'])}

+ASR: {html.escape(prov_tag)} +来源: {html.escape(str(job.get('source') or 'web'))} +尝试: {int(job.get('attempts') or 0)} 次
+{html.escape(LABELS.get(job['status'], job['status']))} +
+

{html.escape(job.get('detail', ''))}

+{review_banner} +{audio_html} +

{links}

+{retry_html} +
+{''.join(sections)} +

处理时间线

{events_html}
""" + return shell("任务详情", body) + + +def review_page(job: dict | None = None) -> bytes: + rows = pending(job) + back_url = f"/jobs/{job['id']}/" if job else "/" + action_url = f"/jobs/{job['id']}/review/" if job else "/review/" + if not rows: + return shell( + "人工确认", + f'

当前没有待确认问题

' + f'

所有已入队的确认项均已处理完毕。

' + f'

← 返回任务详情

', + ) + row = rows[0] + payload = json.loads(row["payload"] or "{}") + choices = payload.get("candidates") or [] + buttons = "".join( + f'' + for v in choices + ) + buttons += ( + '' + ) + draft = "" + path = Path(str(payload.get("draft_path") or "")) + for root in (DRAFTS, CFG.drafts): + try: + path.resolve().relative_to(root.resolve()) + if path.is_file(): + draft = path.read_text(encoding="utf-8", errors="ignore") + break + except (OSError, ValueError): + continue + kind = str(payload.get("kind") or "") + placeholder = ( + "只输入真实姓名，例如：高瑶" + if kind == "voiceprint" + else ("若背景偏差较大，请在此填写正确背景" + if kind == "background_check" + else "输入正确名称或修改说明") + ) + title_str = (job.get("title") or job["original_name"]) if job else "全局待确认队列" + sentence = str(payload.get("sentence") or "") + source_html = f'
{html.escape(sentence)}
' if sentence else "" + draft_html = ( + f'
' + f'展开查看完整草稿
{html.escape(draft)}
' + if draft else "" + ) + body = f"""
+
{html.escape(title_str)} · 剩余 {len(rows)} 个问题 · #{row['id']} +查看任务详情
+

{html.escape(row['question'])}

+{source_html}{draft_html} +
+ +{buttons} + +
+ + +
+
""" + return shell("人工确认", body) + + +def lexicon_page(msg: str = "") -> bytes: + try: + st = lexicon.stats() + except Exception: + st = {"entities": 0, "with_company": 0, "with_related": 0, + "aliases": 0, "aliases_card": 0, "aliases_learned": 0, "corrections": 0} + try: + gst = graph.stats() + except Exception: + gst = {"relations": 0, "interactions": 0, "documents": 0} + am = lexicon.alias_map() + learned_rows = [] + for alias_key, targets in sorted(am.items()): + for canon, kind, origin, confirmed in targets: + if origin != "card": + learned_rows.append( + f"{html.escape(alias_key)}{html.escape(canon)}" + f"{html.escape(kind)}{html.escape(origin)}" + f"{'已确认' if confirmed else '候选'}" + ) + notice = f'
{html.escape(msg)}
' if msg else "" + table_html = ( + "" + "" + + "".join(learned_rows[:120]) + "
简称 / 误识别词规范全名类型来源状态
" + if learned_rows else '

暂无人工新增映射

' + ) + body = f"""{notice}
+

实体与纠错词库 (SQLite + Obsidian)

+
+

实体库：{st.get('entities', 0)} 个实体（{st.get('with_company', 0)} 含公司归属）， +{st.get('aliases', 0)} 条别名（卡片 {st.get('aliases_card', 0)} + 学习/术语表 {st.get('aliases_learned', 0)}）， +{st.get('corrections', 0)} 条发音纠错；关系图谱：{gst.get('relations', 0)} 条关系， +{gst.get('documents', 0)} 篇已索引纪要。

+

新增确定性别名 / ASR 纠错映射

+
+
+
+
+
+ +

+
+

已学习的别名与发音纠错（前 120 条）

{table_html}
""" + return shell("实体词库", body) + + +def settings_page() -> bytes: + hb_path = STATE / "worker.heartbeat" + hb_ok = hb_path.exists() and (time.time() - hb_path.stat().st_mtime < 90) + hb_text = "运行正常 (Ready)" if hb_ok else "未检测到近期心跳 (Standby / 独立模式)" + prov_rows = [] + for p in providers.provider_status(): + badge = '就绪' if p["ready"] else '未配置密钥' + prov_rows.append( + f"{html.escape(p['id'])}" + f"{html.escape(p['label'])}" + f"{html.escape(p['model'])}" + f"{html.escape(p['via'])}" + f"{badge}" + ) + hs_ok = False + if CFG.use_hindsight and CFG.hindsight_api: + try: + with urllib.request.urlopen(f"{CFG.hindsight_api}/health", timeout=3) as r: + hs_ok = r.status == 200 + except Exception: + hs_ok = False + vault_ok = CFG.vault.is_dir() + body = f"""

系统状态与配置

+ + + + + + + + +
Worker 状态{html.escape(hb_text)}
默认 ASR 通道{html.escape(DEFAULT_PROVIDER)}
后处理 LLM 网关 (LiteLLM){html.escape(CFG.endpoint)}
S1c/S2/S3 清洗与关键词模型{html.escape(CFG.clean_model)}
S4/Revise 纪要成稿模型{html.escape(CFG.draft_model)}
Hindsight 记忆服务{html.escape(CFG.hindsight_api)} (bank: {html.escape(CFG.hindsight_bank)}) · {'在线' if hs_ok else '离线或未连接'}
Obsidian Vault 路径{html.escape(str(CFG.vault))} · {'已挂载' if vault_ok else '未挂载'} · Git 自动同步: {'开启' if CFG.git_sync else '关闭'}
+

语音识别模型 (ASR Providers)

+ +{''.join(prov_rows)}
ID名称默认型号接入方式状态
""" + return shell("系统状态与配置", body) + class Handler(BaseHTTPRequestHandler): - server_version="ASRService/1" - def log_message(self,*_): pass - def send_data(self,data,status=200,ctype="text/html; charset=utf-8",disposition=""): - self.send_response(status); self.send_header("Content-Type",ctype); self.send_header("Cache-Control","no-store"); self.send_header("X-Content-Type-Options","nosniff"); self.send_header("Content-Security-Policy","default-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'") - if disposition:self.send_header("Content-Disposition",disposition) - self.send_header("Content-Length",str(len(data)));self.end_headers();self.wfile.write(data) - def json(self,obj,status=200):self.send_data(json.dumps(obj,ensure_ascii=False).encode(),status,"application/json; charset=utf-8") - def auth(self): - value=self.headers.get("Authorization","") - return bool(AGENT_TOKEN and value.startswith("Bearer ") and hmac.compare_digest(value[7:],AGENT_TOKEN)) - def need_auth(self): - if self.auth():return True - self.json({"error":"unauthorized"},401);return False - def header(self,name,limit):return safe(urllib.parse.unquote(self.headers.get(name,"")),limit) - def upload(self,source,protected): - if protected and not self.auth():return self.json({"error":"unauthorized"},401) - try:length=int(self.headers.get("Content-Length","0")) - except ValueError:length=0 - name=self.header("X-Filename",240);suffix=Path(name).suffix.lower() - if length<=0 or length>MAX_UPLOAD:return self.json({"error":"invalid size"},413) - if suffix not in SUPPORTED:return self.json({"error":"unsupported type"},415) - jid=uuid.uuid4().hex[:24]; title=self.header("X-Title",80); stored=f"{time.strftime('%Y-%m-%d_%H_%M_%S')}-{stem(title or Path(name).stem)}-{jid[:8]}{suffix}"; staging=INBOX/".uploading";staging.mkdir(parents=True,exist_ok=True);part=staging/f"{jid}.part";remaining=length;digest=hashlib.sha256() + server_version = "ASRService/2" + + def log_message(self, *_): + pass + + def send_data( + self, + data: bytes, + status: int = 200, + ctype: str = "text/html; charset=utf-8", + disposition: str = "", + extra_headers: dict[str, str] | None = None, + ) -> None: + self.send_response(status) + self.send_header("Content-Type", ctype) + self.send_header("Cache-Control", "no-store") + self.send_header("X-Content-Type-Options", "nosniff") + self.send_header( + "Content-Security-Policy", + "default-src 'self'; media-src 'self'; style-src 'self' 'unsafe-inline'; script-src 'self' 'unsafe-inline'", + ) + if disposition: + self.send_header("Content-Disposition", disposition) + if extra_headers: + for k, v in extra_headers.items(): + self.send_header(k, v) + self.send_header("Content-Length", str(len(data))) + self.end_headers() + self.wfile.write(data) + + def json(self, obj: object, status: int = 200) -> None: + self.send_data( + json.dumps(obj, ensure_ascii=False).encode("utf-8"), + status, + "application/json; charset=utf-8", + ) + + def auth(self) -> bool: + value = self.headers.get("Authorization", "") + return bool( + AGENT_TOKEN + and value.startswith("Bearer ") + and hmac.compare_digest(value[7:], AGENT_TOKEN) + ) + + def web_authed(self) -> bool: + if not WEB_PASSWORD: + return True + if self.auth(): + return True + raw_cookie = self.headers.get("Cookie", "") + jar = cookies.SimpleCookie() + try: + jar.load(raw_cookie) + except Exception: + return False + morsel = jar.get("asr_session") + return bool(morsel and hmac.compare_digest(morsel.value, _session_token())) + + def need_web_auth(self) -> bool: + if self.web_authed(): + return True + self.send_response(303) + self.send_header("Location", "/login") + self.end_headers() + return False + + def need_auth(self) -> bool: + if self.auth(): + return True + self.json({"error": "unauthorized"}, 401) + return False + + def header(self, name: str, limit: int) -> str: + return safe(urllib.parse.unquote(self.headers.get(name, "")), limit) + + def upload(self, source: str, protected: bool) -> None: + if protected and not self.auth(): + return self.json({"error": "unauthorized"}, 401) + if not protected and not self.web_authed(): + return self.json({"error": "unauthorized"}, 401) + try: + length = int(self.headers.get("Content-Length", "0")) + except ValueError: + length = 0 + name = self.header("X-Filename", 240) + suffix = Path(name).suffix.lower() + if length <= 0 or length > MAX_UPLOAD: + return self.json({"error": "invalid size"}, 413) + if suffix not in SUPPORTED: + return self.json({"error": "unsupported type"}, 415) + jid = uuid.uuid4().hex[:24] + title = self.header("X-Title", 80) + prov = providers.normalize_provider_name( + self.header("X-ASR-Provider", 40) or DEFAULT_PROVIDER + ) + model = self.header("X-ASR-Model", 80) + stored = f"{time.strftime('%Y-%m-%d_%H_%M_%S')}-{stem(title or Path(name).stem)}-{jid[:8]}{suffix}" + staging = INBOX / ".uploading" + staging.mkdir(parents=True, exist_ok=True) + part = staging / f"{jid}.part" + remaining = length + digest = hashlib.sha256() try: with part.open("xb") as out: while remaining: - chunk=self.rfile.read(min(1024*1024,remaining)) - if not chunk:raise EOFError - out.write(chunk);digest.update(chunk);remaining-=len(chunk) - out.flush();os.fsync(out.fileno()) - job=db.create_job(jid,name,stored,title=title,context=self.header("X-Context",2000),size=length,source=source);db.update_job(jid,"queued","录音已落盘，等待 worker",sha256=digest.hexdigest());part.replace(INBOX/stored);self.json(public(job),201) - except Exception:part.unlink(missing_ok=True);self.json({"error":"upload failed"},500) - def artifact(self,job,name,api=False): - if not job:return self.json({"error":"not found"},404) - roots={"transcript":TRANSCRIPTS,"draft":DRAFTS,"result":RESULTS};path=Path(job.get(name) or "") - try:path.resolve().relative_to(roots[name].resolve());data=path.read_bytes() - except (OSError,ValueError):return self.json({"error":"not found"},404) - self.send_data(data,200,"text/markdown; charset=utf-8",f'inline; filename="{path.name}"') - def do_GET(self): - path=urllib.parse.urlsplit(self.path).path - if path in ("/health","/healthz"):return self.json({"ok":True,"service":"asr-web"}) - if path=="/readyz": - try:db.connect().close();ok=(STATE/"worker.heartbeat").exists() and time.time()-(STATE/"worker.heartbeat").stat().st_mtime<90 - except Exception:ok=False - return self.json({"ready":ok},200 if ok else 503) - if path=="/":return self.send_data(home()) - if path=="/docs":return self.send_data(shell("Agent API","

Agent API

使用 Bearer Token。POST /api/v1/jobs 的 body 是原始录音，设置 X-Filename、X-Title、X-Context；GET /api/v1/jobs 与 /api/v1/jobs/{id} 查询任务。

")) - if path=="/ui/jobs":return self.json({"jobs":[public(x) for x in db.list_jobs()]}) - m=re.fullmatch(r"/jobs/([0-9a-f]{24})/",path) - if m: - job=db.get_job(m.group(1));return self.send_data(job_page(job)) if job else self.send_data(b"Not found",404,"text/plain") - m=re.fullmatch(r"/jobs/([0-9a-f]{24})/review/",path) - if m: - job=db.get_job(m.group(1));return self.send_data(review_page(job)) if job else self.send_data(b"Not found",404,"text/plain") - m=re.fullmatch(r"/jobs/([0-9a-f]{24})/artifact/(transcript|draft|result)",path) - if m:return self.artifact(db.get_job(m.group(1)),m.group(2)) - if path=="/api/v1/jobs": - if not self.need_auth():return - q=urllib.parse.parse_qs(urllib.parse.urlsplit(self.path).query);return self.json({"jobs":[public(x) for x in db.list_jobs(int(q.get('limit',[30])[0]),int(q.get('offset',[0])[0]),q.get('status',[''])[0])]}) - m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})",path) - if m: - if not self.need_auth():return - job=db.get_job(m.group(1));return self.json(public(job,True)) if job else self.json({"error":"not found"},404) - m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/artifacts/(transcript|draft|result)",path) - if m: - if not self.need_auth():return - return self.artifact(db.get_job(m.group(1)),m.group(2),True) - m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/reviews",path) - if m: - if not self.need_auth():return - job=db.get_job(m.group(1)) - if not job:return self.json({"error":"not found"},404) - rows=[] - for row in pending(job): - payload=json.loads(row["payload"] or "{}") - rows.append({"id":row["id"],"question":row["question"], - "kind":payload.get("kind",""), - "sentence":payload.get("sentence",""), - "candidates":payload.get("candidates") or []}) - return self.json({"reviews":rows}) - self.send_data(b"Not found",404,"text/plain") - def do_POST(self): - path=urllib.parse.urlsplit(self.path).path - if path=="/ui/upload":return self.upload("web",False) - if path=="/api/v1/jobs":return self.upload("agent",True) - m=re.fullmatch(r"/jobs/([0-9a-f]{24})/review/",path) - if m: - job=db.get_job(m.group(1));length=min(int(self.headers.get("Content-Length","0")),4096);form=urllib.parse.parse_qs(self.rfile.read(length).decode());qid=int(form.get("id",["0"])[0]);answer=(form.get("custom",[""])[0].strip() or form.get("answer",[""])[0].strip()) - if not job or qid not in {x["id"] for x in pending(job)} or not answer:return self.send_data(b"Invalid answer",400,"text/plain") - review.answer(qid,answer[:120]);self.send_response(303);self.send_header("Location",path);self.end_headers();return - m=re.fullmatch(r"/(?:api/v1/)?jobs/([0-9a-f]{24})/retry",path) - if m: - api=path.startswith("/api/") - if api and not self.need_auth():return - job=db.get_job(m.group(1));source=next((p for p in FAILED.iterdir() if job and p.name==job["stored_name"]),None) - if not job or job["status"]!="failed" or not source:return self.json({"error":"cannot retry"},409) - source.replace(INBOX/source.name);db.update_job(job["id"],"queued","重新排队",error_code="") - if api:return self.json({"ok":True},202) - self.send_response(303);self.send_header("Location",f"/jobs/{job['id']}/");self.end_headers();return - m=re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/reviews/(\d+)",path) - if m: - if not self.need_auth():return - job=db.get_job(m.group(1));length=min(int(self.headers.get("Content-Length","0")),4096) - try:answer=str(json.loads(self.rfile.read(length)).get("answer") or "").strip() - except (ValueError,AttributeError):answer="" - qid=int(m.group(2)) - if not job or qid not in {x["id"] for x in pending(job)} or not answer: - return self.json({"error":"invalid review answer"},400) - return self.json(review.answer(qid,answer[:120])) - self.send_data(b"Not found",404,"text/plain") + chunk = self.rfile.read(min(1024 * 1024, remaining)) + if not chunk: + raise EOFError + out.write(chunk) + digest.update(chunk) + remaining -= len(chunk) + out.flush() + os.fsync(out.fileno()) + job = db.create_job( + jid, name, stored, + title=title, + context=self.header("X-Context", 2000), + size=length, + source=source, + asr_provider=prov, + asr_model=model, + ) + db.update_job( + jid, "queued", f"录音已落盘，等待 worker (ASR: {prov})", + sha256=digest.hexdigest(), + asr_provider=prov, + asr_model=model, + ) + part.replace(INBOX / stored) + self.json(public(job), 201) + except Exception: + part.unlink(missing_ok=True) + self.json({"error": "upload failed"}, 500) -def main(): - ensure_dirs();db.connect().close();server=ThreadingHTTPServer(("0.0.0.0",8788),Handler);server.daemon_threads=True;server.serve_forever() -if __name__=="__main__":main() + def artifact(self, job: dict | None, name: str, api: bool = False) -> None: + if not job: + return self.json({"error": "not found"}, 404) + text = _read_artifact_text(job, name) + if not text: + return self.json({"error": "not found"}, 404) + path = Path(str(job.get(name) or f"{name}.md")) + self.send_data( + text.encode("utf-8"), + 200, + "text/markdown; charset=utf-8", + f'inline; filename="{path.name}"', + ) + + def stream_audio(self, job: dict | None) -> None: + if not job: + return self.send_data(b"Not found", 404, "text/plain") + audio_path = find_audio_file(job) + if not audio_path or not audio_path.is_file(): + return self.send_data(b"Audio not found", 404, "text/plain") + mime = mimetypes.guess_type(audio_path.name)[0] or "audio/mpeg" + size = audio_path.stat().st_size + range_hdr = self.headers.get("Range", "") + start, end = 0, size - 1 + status = 200 + if range_hdr.startswith("bytes="): + m = re.match(r"bytes=(\d*)-(\d*)", range_hdr) + if m: + if m.group(1): + start = int(m.group(1)) + if m.group(2): + end = min(size - 1, int(m.group(2))) + if 0 <= start <= end < size: + status = 206 + length = end - start + 1 + self.send_response(status) + self.send_header("Content-Type", mime) + self.send_header("Accept-Ranges", "bytes") + if status == 206: + self.send_header("Content-Range", f"bytes {start}-{end}/{size}") + self.send_header("Content-Length", str(length)) + self.end_headers() + with audio_path.open("rb") as f: + f.seek(start) + rem = length + while rem > 0: + buf = f.read(min(256 * 1024, rem)) + if not buf: + break + self.wfile.write(buf) + rem -= len(buf) + + def do_GET(self) -> None: + parsed = urllib.parse.urlsplit(self.path) + path = parsed.path + if path in ("/health", "/healthz"): + return self.json({"ok": True, "service": "asr-web"}) + if path == "/readyz": + try: + db.connect().close() + hb = STATE / "worker.heartbeat" + ok = hb.exists() and time.time() - hb.stat().st_mtime < 90 + except Exception: + ok = False + return self.json({"ready": ok}, 200 if ok else 503) + if path == "/login": + return self.send_data(login_page()) + if path == "/api/v1/providers": + if not self.need_auth(): + return + return self.json({"providers": providers.provider_status()}) + if path == "/api/v1/jobs": + if not self.need_auth(): + return + q = urllib.parse.parse_qs(parsed.query) + return self.json({ + "jobs": [ + public(x) + for x in db.list_jobs( + int(q.get("limit", [30])[0]), + int(q.get("offset", [0])[0]), + q.get("status", [""])[0], + ) + ] + }) + m = re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})", path) + if m: + if not self.need_auth(): + return + job = db.get_job(m.group(1)) + return self.json(public(job, True)) if job else self.json({"error": "not found"}, 404) + m = re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/artifacts/(transcript|draft|result)", path) + if m: + if not self.need_auth(): + return + return self.artifact(db.get_job(m.group(1)), m.group(2), True) + m = re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/reviews", path) + if m: + if not self.need_auth(): + return + job = db.get_job(m.group(1)) + if not job: + return self.json({"error": "not found"}, 404) + rows = [] + for row in pending(job): + payload = json.loads(row["payload"] or "{}") + rows.append({ + "id": row["id"], + "question": row["question"], + "kind": payload.get("kind", ""), + "sentence": payload.get("sentence", ""), + "candidates": payload.get("candidates") or [], + }) + return self.json({"reviews": rows}) + + # Web UI routes below (check optional web password) + if not self.need_web_auth(): + return + if path == "/": + return self.send_data(home()) + if path == "/review/": + return self.send_data(review_page(None)) + if path == "/lexicon": + return self.send_data(lexicon_page()) + if path == "/settings": + return self.send_data(settings_page()) + if path == "/docs": + return self.send_data(shell( + "Agent API", + "

Agent API

" + "

所有 /api/v1/* 接口使用 Authorization: Bearer $ASR_AGENT_TOKEN。

" + "
", + )) + if path == "/ui/jobs": + q = urllib.parse.parse_qs(parsed.query) + st = q.get("status", [""])[0] + return self.json({"jobs": [public(x) for x in db.list_jobs(50, 0, st)]}) + m = re.fullmatch(r"/jobs/([0-9a-f]{24})/", path) + if m: + job = db.get_job(m.group(1)) + return self.send_data(job_page(job)) if job else self.send_data(b"Not found", 404, "text/plain") + m = re.fullmatch(r"/jobs/([0-9a-f]{24})/audio", path) + if m: + return self.stream_audio(db.get_job(m.group(1))) + m = re.fullmatch(r"/jobs/([0-9a-f]{24})/review/", path) + if m: + job = db.get_job(m.group(1)) + return self.send_data(review_page(job)) if job else self.send_data(b"Not found", 404, "text/plain") + m = re.fullmatch(r"/jobs/([0-9a-f]{24})/artifact/(transcript|draft|result)", path) + if m: + return self.artifact(db.get_job(m.group(1)), m.group(2)) + self.send_data(b"Not found", 404, "text/plain") + + def do_POST(self) -> None: + path = urllib.parse.urlsplit(self.path).path + if path == "/login": + length = min(int(self.headers.get("Content-Length", "0")), 4096) + form = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8", errors="ignore")) + pw = form.get("password", [""])[0] + if WEB_PASSWORD and hmac.compare_digest(pw, WEB_PASSWORD): + self.send_response(303) + self.send_header( + "Set-Cookie", + f"asr_session={_session_token()}; Path=/; HttpOnly; SameSite=Lax; Max-Age=2592000", + ) + self.send_header("Location", "/") + self.end_headers() + return + return self.send_data(login_page("口令不正确"), 401) + if path == "/ui/upload": + return self.upload("web", False) + if path == "/api/v1/jobs": + return self.upload("agent", True) + + m = re.fullmatch(r"/api/v1/jobs/([0-9a-f]{24})/reviews/(\d+)", path) + if m: + if not self.need_auth(): + return + job = db.get_job(m.group(1)) + length = min(int(self.headers.get("Content-Length", "0")), 4096) + try: + answer = str(json.loads(self.rfile.read(length)).get("answer") or "").strip() + except (ValueError, AttributeError): + answer = "" + qid = int(m.group(2)) + if not job or qid not in {x["id"] for x in pending(job)} or not answer: + return self.json({"error": "invalid review answer"}, 400) + return self.json(review.answer(qid, answer[:120])) + + m = re.fullmatch(r"/(?:api/v1/)?jobs/([0-9a-f]{24})/retry", path) + if m: + api = path.startswith("/api/") + if api and not self.need_auth(): + return + if not api and not self.need_web_auth(): + return + job = db.get_job(m.group(1)) + if not job: + return self.json({"error": "not found"}, 404) + source = find_audio_file(job) + if not source or job["status"] in {"transcribing", "refining", "publishing"}: + return self.json({"error": "cannot retry"}, 409) + length = min(int(self.headers.get("Content-Length", "0")), 4096) + raw_body = self.rfile.read(length).decode("utf-8", errors="ignore") if length > 0 else "" + new_prov = job.get("asr_provider") or DEFAULT_PROVIDER + new_model = job.get("asr_model") or "" + if raw_body: + if api and raw_body.strip().startswith("{"): + try: + body_json = json.loads(raw_body) + if body_json.get("provider"): + new_prov = providers.normalize_provider_name(body_json["provider"]) + if "model" in body_json: + new_model = safe(str(body_json.get("model") or ""), 80) + except ValueError: + pass + else: + form = urllib.parse.parse_qs(raw_body) + if form.get("provider"): + new_prov = providers.normalize_provider_name(form["provider"][0]) + if "model" in form: + new_model = safe(form.get("model", [""])[0], 80) + dest = INBOX / job["stored_name"] + if source.resolve() != dest.resolve(): + shutil.copyfile(source, dest) if source.parent == PROCESSED else source.replace(dest) + # Touch mtime slightly in the past so worker picks it up without extra wait + old_ts = time.time() - 60 + os.utime(dest, (old_ts, old_ts)) + db.update_job( + job["id"], "queued", f"重新排队 (ASR: {new_prov})", + error_code="", asr_provider=new_prov, asr_model=new_model, + ) + if api: + return self.json({"ok": True, "asr_provider": new_prov, "asr_model": new_model}, 202) + self.send_response(303) + self.send_header("Location", f"/jobs/{job['id']}/") + self.end_headers() + return + + if not self.need_web_auth(): + return + + if path == "/review/" or re.fullmatch(r"/jobs/([0-9a-f]{24})/review/", path): + m_job = re.fullmatch(r"/jobs/([0-9a-f]{24})/review/", path) + job = db.get_job(m_job.group(1)) if m_job else None + length = min(int(self.headers.get("Content-Length", "0")), 4096) + form = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8", errors="ignore")) + qid = int(form.get("id", ["0"])[0]) + answer = form.get("custom", [""])[0].strip() or form.get("answer", [""])[0].strip() + allowed_ids = {x["id"] for x in pending(job)} + if (m_job and not job) or qid not in allowed_ids or not answer: + return self.send_data(b"Invalid answer", 400, "text/plain") + review.answer(qid, answer[:120]) + self.send_response(303) + self.send_header("Location", path) + self.end_headers() + return + + m_edit = re.fullmatch(r"/jobs/([0-9a-f]{24})/edit", path) + if m_edit: + job = db.get_job(m_edit.group(1)) + if not job: + return self.send_data(b"Not found", 404, "text/plain") + length = min(int(self.headers.get("Content-Length", "0")), 2 * 1024 * 1024) + form = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8", errors="ignore")) + target = form.get("target", [""])[0] + content = form.get("content", [""])[0] + if target in ("draft", "result") and job.get(target) and content.strip(): + tpath = Path(str(job[target])) + vault.write_draft_file(tpath, content) + self.send_response(303) + self.send_header("Location", f"/jobs/{job['id']}/") + self.end_headers() + return + + if path == "/lexicon/learn": + length = min(int(self.headers.get("Content-Length", "0")), 4096) + form = urllib.parse.parse_qs(self.rfile.read(length).decode("utf-8", errors="ignore")) + alias_val = safe(form.get("alias", [""])[0], 40) + canon_val = safe(form.get("canonical", [""])[0], 40) + note_val = safe(form.get("note", [""])[0], 120) + if alias_val and canon_val: + lexicon.learn_alias(alias_val, canon_val, note=note_val) + lexicon.learn_correction(alias_val, canon_val, note=note_val) + return self.send_data(lexicon_page(f"已添加映射：{alias_val} → {canon_val}")) + return self.send_data(lexicon_page("请填写完整的别名与规范名"), 400) + + if path == "/lexicon/sync": + l_res = lexicon.sync_from_vault(force=True) + g_res = graph.sync(force=True) + return self.send_data(lexicon_page(f"同步完成：词库 {l_res}，图谱 {g_res}")) + + self.send_data(b"Not found", 404, "text/plain") + + +def main() -> None: + os.environ.setdefault("ASR_ROLE", "web") + os.environ.setdefault("ASR_REVISION_QUEUE_DIR", str(REVISION_QUEUE)) + ensure_dirs() + db.connect().close() + if os.environ.get("ASR_ROLE", "").lower() == "all": + from . import worker + t = threading.Thread(target=worker.run_loop, name="asr-worker", daemon=True) + t.start() + port = int(os.environ.get("ASR_PORT", os.environ.get("ASR_REVIEW_PORT", "8788"))) + host = os.environ.get("ASR_HOST", "0.0.0.0") + server = ThreadingHTTPServer((host, port), Handler) + server.daemon_threads = True + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/servers/tools-pve/asr-pipeline/asrservice/worker.py b/servers/tools-pve/asr-pipeline/asrservice/worker.py index 63cbb9b..94315e8 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/worker.py +++ b/servers/tools-pve/asr-pipeline/asrservice/worker.py @@ -6,17 +6,17 @@ import json import os import shutil import sqlite3 -import subprocess import time import uuid from pathlib import Path -from asrpipe import refine, review, revise, vault +from asrpipe import graph, lexicon, providers, refine, review, revise, vault from . import db -from .config import (DIARIZE_URL, DRAFTS, FAILED, INBOX, MAX_ATTEMPTS, META, - POLL_SECONDS, PROCESSED, PROCESSING, PUBLIC_URL, - REVISION_QUEUE, STABLE_SECONDS, STATE, SUPPORTED, TRANSCRIPTS, LOGS, - ensure_dirs) +from .config import ( + DATA, DEFAULT_PROVIDER, DRAFTS, FAILED, INBOX, LOGS, MAX_ATTEMPTS, + POLL_SECONDS, PROCESSED, PROCESSING, PUBLIC_URL, REVISION_QUEUE, + STABLE_SECONDS, STATE, SUPPORTED, TRANSCRIPTS, ensure_dirs, +) from .notify import send @@ -35,69 +35,83 @@ def sha256(path: Path) -> str: def transcript_from_response(payload: dict) -> str: - segments = payload.get("segments") - if not isinstance(segments, list) or not segments: - raise RuntimeError("ASR response contains no segments") - vp = payload.get("voiceprint") or {} - sid = str(vp.get("session_id") or "") - matched = [f"{m.get('speaker_local')}={m.get('person_name')}" for m in vp.get("matches", []) - if m.get("status") == "matched"] - speakers = {str(s.get("speaker")) for s in segments if s.get("speaker") is not None} - lines = [f"", - "# 转写(Qwen3-ASR 1.7B + CAM++ 分角色 + 声纹)" - + (" | 声纹命名: " + ", ".join(matched) if matched else ""), ""] - turns: list[list[str]] = [] - for segment in segments: - speaker, text = str(segment.get("speaker", "?")), str(segment.get("text") or "").strip() - if not text: - continue - if turns and turns[-1][0] == speaker: - turns[-1][1] += text - else: - turns.append([speaker, text]) - lines.extend(f"【{speaker}】{text}\n" for speaker, text in turns) - return "\n".join(lines).strip() + "\n" + """Backward-compatible helper for qwen3cam JSON responses.""" + return providers.transcript_from_qwen3cam_response(payload) -def transcribe(audio: Path, job_id: str) -> Path: +def transcribe( + audio: Path, + job_id: str, + *, + provider: str = "", + model: str = "", + title: str = "", + user_context: str = "", +) -> tuple[Path, providers.TranscribeResult]: + res = providers.transcribe_audio( + audio, + provider=provider or DEFAULT_PROVIDER, + model=model, + title=title, + user_context=user_context, + ) response_path = TRANSCRIPTS / f"{job_id}.asr.json" - result = subprocess.run([ - "curl", "-sS", "--fail-with-body", "--max-time", "900", - "-F", f"file=@{audio}", "-F", "language=Chinese", - "-F", "save_voiceprint_session=true", DIARIZE_URL, - ], capture_output=True, timeout=930) - if result.returncode: - raise RuntimeError(f"ASR HTTP failed ({result.returncode})") - response_path.write_bytes(result.stdout) try: - payload = json.loads(result.stdout) - except ValueError as exc: - raise RuntimeError("ASR returned invalid JSON") from exc + response_path.write_text( + json.dumps(res.raw_response, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + except Exception: + pass transcript = TRANSCRIPTS / f"{job_id}.txt" - transcript.write_text(transcript_from_response(payload), encoding="utf-8") - return transcript + transcript.write_text(res.transcript_text, encoding="utf-8") + return transcript, res def process_audio(source: Path, job: dict) -> None: job_id = job["id"] attempts = int(job.get("attempts") or 0) + 1 + prov = providers.normalize_provider_name(job.get("asr_provider") or DEFAULT_PROVIDER) + model = str(job.get("asr_model") or "") digest = sha256(source) - db.update_job(job_id, "transcribing", "正在进行分角色转写", sha256=digest, - attempts=attempts, error_code="") - transcript = transcribe(source, job_id) - intake = {"job_id": job_id, "title": job.get("title", ""), - "context": job.get("context", ""), "source": job.get("source", "")} + prov_label = providers.PROVIDERS.get(prov, {}).get("label", prov) + db.update_job( + job_id, "transcribing", f"正在转写 ({prov_label})", + sha256=digest, attempts=attempts, error_code="", + asr_provider=prov, asr_model=model, + ) + transcript, res = transcribe( + source, job_id, + provider=prov, model=model, + title=str(job.get("title") or ""), + user_context=str(job.get("context") or ""), + ) + intake = { + "job_id": job_id, + "title": job.get("title", ""), + "context": job.get("context", ""), + "source": job.get("source", ""), + "asr_provider": res.provider, + "asr_model": res.model, + } raw = transcript.read_text(encoding="utf-8") raw = f"\n" + raw transcript.write_text(raw, encoding="utf-8") - db.update_job(job_id, "refining", "转写完成，正在生成整理稿", transcript=str(transcript)) - result = refine.refine(raw, date=time.strftime("%Y-%m-%d"), - transcript_path=str(transcript), queue_review=True) + db.update_job( + job_id, "refining", f"转写完成 ({res.provider}:{res.model})，正在生成整理稿", + transcript=str(transcript), asr_provider=res.provider, asr_model=res.model, + ) + result = refine.refine( + raw, date=time.strftime("%Y-%m-%d"), + transcript_path=str(transcript), queue_review=True, + ) draft = DRAFTS / f"{job_id}.md" vault.write_draft_file(draft, result.draft) questions = review.enqueue_draft(str(transcript), draft) - db.update_job(job_id, "review", f"整理稿已生成，{questions} 个问题待确认", - transcript=str(transcript), draft=str(draft)) + db.update_job( + job_id, "review", f"整理稿已生成，{questions} 个问题待确认", + transcript=str(transcript), draft=str(draft), + ) target = PROCESSED / source.name if target.exists(): target = PROCESSED / f"{source.stem}-{job_id[:8]}{source.suffix}" @@ -110,8 +124,11 @@ def claim(path: Path) -> tuple[Path, dict]: job = db.find_by_stored(path.name) if not job: job_id = uuid.uuid4().hex[:24] - job = db.create_job(job_id, path.name, path.name, title=path.stem, - size=path.stat().st_size, source="ssh") + job = db.create_job( + job_id, path.name, path.name, title=path.stem, + size=path.stat().st_size, source="ssh", + asr_provider=DEFAULT_PROVIDER, + ) target = PROCESSING / path.name if target.exists(): target = PROCESSING / f"{path.stem}-{job['id'][:8]}{path.suffix}" @@ -132,10 +149,12 @@ def scan_inbox() -> None: except Exception as exc: attempts = int((db.get_job(job["id"]) or {}).get("attempts") or 0) retry = attempts < MAX_ATTEMPTS - db.update_job(job["id"], "queued" if retry else "failed", - "处理失败，等待重试" if retry else "处理失败", - error_code=type(exc).__name__) - log(f"job={job['id']} failed type={type(exc).__name__}") + db.update_job( + job["id"], "queued" if retry else "failed", + f"处理失败，等待重试 ({exc})" if retry else f"处理失败：{exc}", + error_code=type(exc).__name__, + ) + log(f"job={job['id']} failed type={type(exc).__name__} err={exc}") target = INBOX / source.name if retry else FAILED / source.name if source.exists(): shutil.move(str(source), target) @@ -151,27 +170,41 @@ def process_revisions() -> None: revise.revise(draft) current = db.get_job(job_id) or {} if current.get("status") == "complete": - send("ASR 正式记录已生成", current.get("title") or current.get("original_name", ""), + send("ASR 正式记录已生成", + current.get("title") or current.get("original_name", ""), f"{PUBLIC_URL}/jobs/{job_id}/") elif current.get("status") != "review": - db.update_job(job_id, "review", "定稿预览已生成，等待发布确认", draft=str(draft)) + db.update_job( + job_id, "review", "定稿预览已生成，等待发布确认", + draft=str(draft), + ) marker.unlink(missing_ok=True) except Exception as exc: - log(f"revision marker={marker.name} failed type={type(exc).__name__}") + log(f"revision marker={marker.name} failed type={type(exc).__name__} err={exc}") def recover() -> None: for job in db.pending_recovery(): if job["status"] in {"transcribing", "refining"}: - candidate = next((p for p in PROCESSING.iterdir() if p.name == job["stored_name"]), None) + candidate = next( + (p for p in PROCESSING.iterdir() if p.name == job["stored_name"]), + None, + ) if candidate: shutil.move(str(candidate), INBOX / candidate.name) db.update_job(job["id"], "queued", "worker 重启后重新排队") +def sync_vault_knowledge() -> None: + try: + lexicon.sync_from_vault() + graph.sync() + except Exception as exc: + log(f"vault knowledge sync skipped: {exc}") + + def backup_databases() -> None: - # STATE is local disk; durable copies belong on the NFS data volume. - backup_dir = Path("/data/backups") + backup_dir = DATA / "backups" backup_dir.mkdir(parents=True, exist_ok=True) for name in ("intake.db", "lexicon.db"): source = STATE / name @@ -184,22 +217,34 @@ def backup_databases() -> None: temp.replace(target) -def main() -> None: +def run_loop() -> None: + os.environ.setdefault("ASR_ROLE", "worker") + os.environ.setdefault("ASR_REVISION_QUEUE_DIR", str(REVISION_QUEUE)) ensure_dirs() lock = (STATE / "worker.lock").open("w") fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) recover() + sync_vault_knowledge() (STATE / "worker.heartbeat").write_text(str(time.time())) last_backup = 0.0 + last_vault_sync = time.time() while True: scan_inbox() process_revisions() - if time.time() - last_backup >= 3600: + now = time.time() + if now - last_vault_sync >= 900: + sync_vault_knowledge() + last_vault_sync = now + if now - last_backup >= 3600: backup_databases() - last_backup = time.time() - (STATE / "worker.heartbeat").write_text(str(time.time())) + last_backup = now + (STATE / "worker.heartbeat").write_text(str(now)) time.sleep(POLL_SECONDS) +def main() -> None: + run_loop() + + if __name__ == "__main__": main() diff --git a/servers/tools-pve/asr-pipeline/compose.yaml b/servers/tools-pve/asr-pipeline/compose.yaml index eb1f79e..2f25a5c 100644 --- a/servers/tools-pve/asr-pipeline/compose.yaml +++ b/servers/tools-pve/asr-pipeline/compose.yaml @@ -1,11 +1,24 @@ x-asr-environment: &asr-environment - ZENMUX_API_KEY: ${ZENMUX_API_KEY:-} - ASR_DRAFT_MODEL: ${ASR_DRAFT_MODEL:-anthropic/claude-sonnet-5.5} - ASR_AGENT_TOKEN: ${ASR_AGENT_TOKEN:-} - ASR_PUBLIC_URL: ${ASR_PUBLIC_URL:-http://192.168.50.49:8788} + ASR_PROVIDER: ${ASR_PROVIDER:-qwen3cam} ASR_DIARIZE_URL: ${ASR_DIARIZE_URL:-http://192.168.50.100:10099/asr} + ASR_VOICEPRINT_ENROLL_URL: ${ASR_VOICEPRINT_ENROLL_URL:-http://192.168.50.100:10099/voiceprints/enroll} + LITELLM_BASE_URL: ${LITELLM_BASE_URL:-http://192.168.50.49:4000/v1} + LITELLM_API_KEY: ${LITELLM_API_KEY:-} + ASR_GEMINI_MODEL: ${ASR_GEMINI_MODEL:-aistudio/gemini-3.5-transcribe} + ASR_OPENAI_MODEL: ${ASR_OPENAI_MODEL:-openai/gpt-4o-transcribe-diarize} + SONIOX_API_KEY: ${SONIOX_API_KEY:-} + SONIOX_MODEL: ${SONIOX_MODEL:-stt-async-v5} + XAI_API_KEY: ${XAI_API_KEY:-} + ASR_GROK_MODEL: ${ASR_GROK_MODEL:-grok-voice-transcribe-2.0} + ASR_CLEAN_MODEL: ${ASR_CLEAN_MODEL:-aistudio/gemini-flash-lite-latest} + ASR_KEYWORD_MODEL: ${ASR_KEYWORD_MODEL:-aistudio/gemini-flash-lite-latest} + ASR_DRAFT_MODEL: ${ASR_DRAFT_MODEL:-codex/gpt-sol-latest} + ASR_AGENT_TOKEN: ${ASR_AGENT_TOKEN:-} + ASR_WEB_PASSWORD: ${ASR_WEB_PASSWORD:-} + ASR_PUBLIC_URL: ${ASR_PUBLIC_URL:-http://192.168.50.49:8788} HINDSIGHT_API: ${HINDSIGHT_API:-http://192.168.50.47:8888} HINDSIGHT_BANK: ${HINDSIGHT_BANK:-personal-context} + ASR_GIT_SYNC: ${ASR_GIT_SYNC:-0} ASR_NOTIFY_BACKEND: ${ASR_NOTIFY_BACKEND:-none} HERMES_WEBHOOK_URL: ${HERMES_WEBHOOK_URL:-} HERMES_WEBHOOK_SECRET: ${HERMES_WEBHOOK_SECRET:-} @@ -14,7 +27,7 @@ x-asr-environment: &asr-environment services: asr-web: - image: tools-asr-pipeline:2026-09-18 + image: tools-asr-pipeline:2026-11-02 pull_policy: never build: context: . @@ -39,7 +52,7 @@ services: - no-new-privileges:true asr-worker: - image: tools-asr-pipeline:2026-09-18 + image: tools-asr-pipeline:2026-11-02 pull_policy: never container_name: tools-asr-worker restart: unless-stopped diff --git a/servers/tools-pve/asr-pipeline/requirements.txt b/servers/tools-pve/asr-pipeline/requirements.txt index df58d12..c2cbb05 100644 --- a/servers/tools-pve/asr-pipeline/requirements.txt +++ b/servers/tools-pve/asr-pipeline/requirements.txt @@ -1,3 +1,3 @@ -pypinyin==0.55.0 -rapidfuzz==3.14.1 -numpy==2.3.3 +pypinyin>=0.51,<1 +rapidfuzz>=3,<4 +numpy>=1.26,<3 diff --git a/servers/tools-pve/asr-pipeline/tests/test_service.py b/servers/tools-pve/asr-pipeline/tests/test_service.py index 802a5ea..27c44b5 100644 --- a/servers/tools-pve/asr-pipeline/tests/test_service.py +++ b/servers/tools-pve/asr-pipeline/tests/test_service.py @@ -1,51 +1,197 @@ from __future__ import annotations -import importlib import json import os import tempfile +import threading import unittest +import urllib.request +from http.server import ThreadingHTTPServer from pathlib import Path +from unittest import mock + +TMP = tempfile.TemporaryDirectory() +os.environ["ASR_DATA_DIR"] = str(Path(TMP.name) / "data") +os.environ["ASR_STATE_DIR"] = str(Path(TMP.name) / "state") +os.environ["ASR_VAULT"] = str(Path(TMP.name) / "vault") +os.environ["ASR_RECORDINGS_ROOT"] = str(Path(TMP.name) / "recordings") +os.environ["ASR_AGENT_TOKEN"] = "test-agent-token" +os.environ["ASR_NOTIFY"] = "0" +os.environ["ASR_GIT_SYNC"] = "0" + +from asrpipe import lexicon, providers, review, segment # noqa: E402 +from asrservice import config as svc_cfg, db as svc_db, web as svc_web # noqa: E402 -class ServiceTests(unittest.TestCase): - @classmethod - def setUpClass(cls): - cls.tmp = tempfile.TemporaryDirectory() - os.environ["ASR_DATA_DIR"] = str(Path(cls.tmp.name) / "data") - os.environ["ASR_STATE_DIR"] = str(Path(cls.tmp.name) / "state") - from asrservice import config, db, worker - cls.config, cls.db, cls.worker = config, db, worker +class ProviderAndServiceTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.root = Path(self.tmp.name) + svc_cfg.ensure_dirs() + svc_db.connect().close() - @classmethod - def tearDownClass(cls): - cls.tmp.cleanup() + def tearDown(self): + self.tmp.cleanup() - def test_job_events_follow_updates(self): - row = self.db.create_job("a" * 24, "x.m4a", "x.m4a", source="ssh") - self.assertEqual(row["status"], "queued") - self.db.update_job(row["id"], "transcribing", "started", attempts=1) - current = self.db.get_job(row["id"]) - self.assertEqual(current["attempts"], 1) - self.assertEqual([e["status"] for e in self.db.events(row["id"])], - ["queued", "transcribing"]) + def test_soniox_token_diarization_and_segment_parse(self): + payload = { + "tokens": [ + {"text": "大家", "speaker": "1"}, + {"text": "好，", "speaker": "1"}, + {"text": "今天", "speaker": "1"}, + {"text": "讨论", "speaker": "1"}, + {"text": "预算。", "speaker": "1"}, + {"text": "收到，", "speaker": "2"}, + {"text": "我来", "speaker": "2"}, + {"text": "汇报。", "speaker": "2"}, + ] + } + text = providers.transcript_from_soniox_response(payload, model="stt-async-v5") + parsed_turns, meta = segment.parse(text) + self.assertEqual(meta["asr_provider"], "soniox") + self.assertEqual(meta["asr_model"], "stt-async-v5") + self.assertEqual(len(parsed_turns), 2) + self.assertEqual(parsed_turns[0].speaker, "speaker_0") + self.assertEqual(parsed_turns[0].text, "大家好，今天讨论预算。") + self.assertEqual(parsed_turns[1].speaker, "speaker_1") + self.assertEqual(parsed_turns[1].text, "收到，我来汇报。") - def test_transcript_contract(self): - text = self.worker.transcript_from_response({ - "segments": [ - {"speaker": "speaker_0", "text": "你好。"}, - {"speaker": "speaker_0", "text": "继续。"}, - {"speaker": "speaker_1", "text": "收到。"}, - ], - "voiceprint": {"session_id": "s1", "matches": []}, - }) - self.assertIn("voiceprint_session: s1", text) - self.assertIn("【speaker_0】你好。继续。", text) - self.assertEqual(text.count("【speaker_0】"), 1) + def test_soniox_context_includes_lexicon_and_intake(self): + lexicon.learn_alias("lite网关", "LiteLLM", note="test") + ctx = providers.build_soniox_context( + title="网关评审", + user_context="讨论 LiteLLM 与 Soniox 集成", + ) + self.assertIn("terms", ctx) + self.assertIn("LiteLLM", ctx["terms"]) + self.assertIn("讨论 LiteLLM", ctx.get("text", "")) - def test_transcript_rejects_empty_segments(self): - with self.assertRaises(RuntimeError): - self.worker.transcript_from_response({"segments": []}) + def test_openai_and_grok_diarized_transcriptions(self): + audio = self.root / "sample.mp3" + audio.write_bytes(b"ID3-fake-audio") + + def fake_openai_multipart(url, *, fields, file_field, file_path, headers=None, timeout=900): + return { + "segments": [ + {"speaker": "A", "text": "第一句发言。"}, + {"speaker": "A", "text": "继续补充。"}, + {"speaker": "B", "text": "第二位发言。"}, + ] + } + + with mock.patch("asrpipe.providers._http_multipart", side_effect=fake_openai_multipart), \ + mock.patch("asrpipe.providers.probe_duration_seconds", return_value=30), \ + mock.patch.object(providers.CFG, "api_key", return_value="sk-test"): + res = providers.transcribe_openai( + audio, + title="OpenAI 测试", + model="openai/gpt-4o-transcribe-diarize", + ) + turns, meta = segment.parse(res.transcript_text) + self.assertEqual(meta["asr_provider"], "openai") + self.assertEqual(len(turns), 2) + self.assertIn("第一句发言。继续补充。", turns[0].text) + self.assertEqual(turns[1].speaker, "speaker_1") + + def fake_grok_multipart(url, *, fields, file_field, file_path, headers=None, timeout=900): + return { + "utterances": [ + {"speaker": 0, "text": "Grok 说话人零。"}, + {"speaker": 1, "text": "Grok 说话人一。"}, + ] + } + + with mock.patch("asrpipe.providers._http_multipart", side_effect=fake_grok_multipart), \ + mock.patch("asrpipe.providers.probe_duration_seconds", return_value=30), \ + mock.patch.object(providers.CFG, "grok_api_key", "xai-test"): + gres = providers.transcribe_grok( + audio, + title="Grok 测试", + ) + gturns, gmeta = segment.parse(gres.transcript_text) + self.assertEqual(gmeta["asr_provider"], "grok") + self.assertEqual(len(gturns), 2) + self.assertEqual(gturns[0].speaker, "speaker_0") + self.assertEqual(gturns[1].speaker, "speaker_1") + + def test_online_speaker_confirmation_updates_draft_without_voiceprint_server(self): + drafts_dir = review.CFG.drafts + drafts_dir.mkdir(parents=True, exist_ok=True) + draft = drafts_dir / "online-draft.md" + draft.write_text( + "---\n" + "title: 线上模型会议\n" + "status: draft\n" + "asr_provider: soniox\n" + "voiceprint_session: ''\n" + "voiceprint_speakers: {}\n" + "---\n\n" + "【speaker_1】发言内容\n", + encoding="utf-8", + ) + with mock.patch.object(review.CFG, "voiceprint_enroll_url", "http://127.0.0.1:9/unreachable"): + enrolled = review._confirm_voiceprint(str(draft), "speaker_1", "张三") + self.assertFalse(enrolled) + updated = draft.read_text(encoding="utf-8") + self.assertIn("speaker_1: 张三", updated) + + def test_db_and_web_endpoints(self): + job_id = "0123456789abcdef01234567" + job = svc_db.create_job( + job_id=job_id, + original_name="meeting.m4a", + stored_name=f"20261102-{job_id}.m4a", + title="架构周会", + context="验证 Web API", + size=2048, + asr_provider="soniox", + asr_model="stt-async-v5", + ) + self.assertEqual(job["asr_provider"], "soniox") + self.assertEqual(job["asr_model"], "stt-async-v5") + + svc_web.AGENT_TOKEN = "test-agent-token" + server = ThreadingHTTPServer(("127.0.0.1", 0), svc_web.Handler) + port = server.server_address[1] + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + def fetch(path: str, headers: dict[str, str] | None = None) -> tuple[int, str]: + req = urllib.request.Request( + f"http://127.0.0.1:{port}{path}", + headers=headers or {}, + ) + with urllib.request.urlopen(req, timeout=5) as resp: + return resp.status, resp.read().decode("utf-8") + + status, body = fetch("/health") + self.assertEqual(status, 200) + self.assertEqual(json.loads(body)["ok"], True) + + status, body = fetch( + "/api/v1/providers", + headers={"Authorization": "Bearer test-agent-token"}, + ) + self.assertEqual(status, 200) + prov_data = json.loads(body) + prov_ids = [p["id"] for p in prov_data["providers"]] + self.assertEqual(prov_ids, ["qwen3cam", "soniox", "gemini", "openai", "grok"]) + + status, body = fetch(f"/jobs/{job_id}/") + self.assertEqual(status, 200) + self.assertIn("架构周会", body) + self.assertIn("soniox", body) + + status, body = fetch("/lexicon") + self.assertEqual(status, 200) + self.assertIn("词库", body) + + status, body = fetch("/settings") + self.assertEqual(status, 200) + self.assertIn("LiteLLM", body) + finally: + server.shutdown() + server.server_close() if __name__ == "__main__":