From 56e03051885f4ce63313a2853bc9933d6d0c282b Mon Sep 17 00:00:00 2001 From: Codex Date: Mon, 2 Nov 2026 02:57:42 -0800 Subject: [PATCH] fix(tools-pve/asr-pipeline): mount gitea ssh key for Obsidian vault sync and fix provider source_detail --- servers/tools-pve/asr-pipeline/Dockerfile | 2 +- .../asr-pipeline/asrpipe/providers.py | 110 +++++++++++++----- .../tools-pve/asr-pipeline/asrpipe/refine.py | 8 +- .../tools-pve/asr-pipeline/asrpipe/revise.py | 24 ++-- .../tools-pve/asr-pipeline/asrpipe/segment.py | 6 +- .../tools-pve/asr-pipeline/asrpipe/vault.py | 33 +++++- .../tools-pve/asr-pipeline/asrservice/web.py | 3 +- .../asr-pipeline/asrservice/worker.py | 6 +- servers/tools-pve/asr-pipeline/compose.yaml | 2 + .../asr-pipeline/tests/test_service.py | 7 +- 10 files changed, 156 insertions(+), 45 deletions(-) diff --git a/servers/tools-pve/asr-pipeline/Dockerfile b/servers/tools-pve/asr-pipeline/Dockerfile index b7b7b2a..b725478 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 ffmpeg git tini \ + ca-certificates curl ffmpeg git openssh-client tini \ && rm -rf /var/lib/apt/lists/* WORKDIR /app diff --git a/servers/tools-pve/asr-pipeline/asrpipe/providers.py b/servers/tools-pve/asr-pipeline/asrpipe/providers.py index f24f7d4..2295e14 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/providers.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/providers.py @@ -16,12 +16,18 @@ import urllib.error import urllib.request import uuid from dataclasses import dataclass, field +import http.client from pathlib import Path from typing import Any from .config import CFG from .log import log + +class EmptyAudioError(RuntimeError): + """Raised when an audio file contains silence/noise or produces zero ASR speech tokens.""" + pass + PROVIDERS: dict[str, dict[str, str]] = { "qwen3cam": { "label": "内网 Qwen3-ASR 1.7B + CAM++ (分角色 + 声纹)", @@ -203,7 +209,9 @@ def format_diarized_transcript( matched_names: list[str] | None = None, ) -> str: if not segments: - raise RuntimeError(f"ASR ({provider}) response contains no segments") + raise EmptyAudioError( + f"ASR ({provider}) 未识别到有效语音片段（音频可能为静音、设备未拾音或音量过低）" + ) speaker_map: dict[str, str] = {} turns: list[list[str]] = [] speakers_seen: set[str] = set() @@ -221,7 +229,9 @@ def format_diarized_transcript( else: turns.append([spk, text]) if not turns: - raise RuntimeError(f"ASR ({provider}) segments contained no text") + raise EmptyAudioError( + f"ASR ({provider}) 未识别到任何有效文字（音频可能为静音或底噪）" + ) lines = [f""] if voiceprint_session or provider == "qwen3cam": @@ -240,7 +250,7 @@ def format_diarized_transcript( 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") + raise EmptyAudioError("ASR (qwen3cam) 未识别到有效语音片段（音频可能为静音、设备未拾音或音量过低）") vp = payload.get("voiceprint") or {} sid = str(vp.get("session_id") or "") matched = [ @@ -431,16 +441,33 @@ def _http_multipart( file_path: Path, headers: dict[str, str] | None = None, timeout: int = 900, + retries: int = 3, ) -> 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 + last_exc: Exception | None = None + for attempt in range(1, max(1, retries) + 1): + 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") + break + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="ignore")[:300] + if exc.code in (502, 503, 504) and attempt < retries: + last_exc = exc + time.sleep(2 * attempt) + continue + raise RuntimeError(f"ASR HTTP {exc.code}: {detail}") from exc + except (urllib.error.URLError, http.client.HTTPException, ConnectionError, TimeoutError, OSError) as exc: + last_exc = exc + if attempt < retries: + log(f"multipart upload transient error (attempt {attempt}/{retries}): {exc}") + time.sleep(2 * attempt) + continue + raise RuntimeError(f"ASR upload connection error after {retries} attempts: {exc}") from exc + else: + raise RuntimeError(f"ASR upload failed: {last_exc}") try: return json.loads(raw) except ValueError as exc: @@ -454,18 +481,34 @@ def _http_json( payload: dict[str, Any] | None = None, headers: dict[str, str] | None = None, timeout: int = 120, + retries: int = 3, ) -> 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 + last_exc: Exception | None = None + for attempt in range(1, max(1, retries) + 1): + 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") + break + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="ignore")[:300] + if exc.code in (502, 503, 504) and attempt < retries: + last_exc = exc + time.sleep(1.5 * attempt) + continue + raise RuntimeError(f"ASR HTTP {exc.code}: {detail}") from exc + except (urllib.error.URLError, http.client.HTTPException, ConnectionError, TimeoutError, OSError) as exc: + last_exc = exc + if attempt < retries: + time.sleep(1.5 * attempt) + continue + raise RuntimeError(f"ASR HTTP request failed after {retries} attempts: {exc}") from exc + else: + raise RuntimeError(f"ASR HTTP request failed: {last_exc}") if not raw.strip(): return {} try: @@ -575,14 +618,29 @@ def transcribe_soniox( 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, - ) + with tempfile.TemporaryDirectory(prefix="asr-soniox-") as tmp: + upload_path = audio + if audio.exists() and audio.stat().st_size > 8 * 1024 * 1024 and shutil.which("ffmpeg"): + compressed = Path(tmp) / "soniox_upload.mp3" + try: + 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 0 < compressed.stat().st_size < audio.stat().st_size: + upload_path = compressed + except Exception as exc: + log(f"soniox audio pre-compress skipped: {exc}") + + file_resp = _http_multipart( + f"{base}/files", + fields={}, + file_field="file", + file_path=upload_path, + 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}") @@ -622,7 +680,7 @@ def transcribe_soniox( else: raise RuntimeError("Soniox transcription timed out waiting for completion") - if "tokens" in status_resp or "segments" in status_resp: + if status_resp.get("tokens") or status_resp.get("segments"): transcript_payload = status_resp else: transcript_payload = _http_json( diff --git a/servers/tools-pve/asr-pipeline/asrpipe/refine.py b/servers/tools-pve/asr-pipeline/asrpipe/refine.py index 5d14438..fd07fe4 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/refine.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/refine.py @@ -377,9 +377,13 @@ def write_draft(cleaned: str, meta: dict, context: str, date: str, if ctx: ctx = "本场背景（来自既有笔记与长期记忆；引用其中内容时必须按上面第 3 条加 〔背景〕）：\n" \ + context - detail = SOURCE_DETAIL.get(meta.get("backend", ""), "ASR") + backend = meta.get("backend") or meta.get("asr_provider") or "" + model_name = meta.get("asr_model") or "" + detail = SOURCE_DETAIL.get(backend, "ASR") + if model_name and backend in ("soniox", "gemini", "openai", "grok") and model_name not in detail: + detail = f"{detail} ({model_name})" prompt = prompts.DRAFT_PROMPT.format( - speaker_note=prompts.SPEAKER_NOTE.get(meta.get("backend", ""), ""), + speaker_note=prompts.SPEAKER_NOTE.get(backend, ""), context=ctx, date=date, source_detail=detail, transcript=transcript_path, extra_fm=extra, body=cleaned) return chat(CFG.draft_model, prompt, max_tokens=12000, usage=usage) diff --git a/servers/tools-pve/asr-pipeline/asrpipe/revise.py b/servers/tools-pve/asr-pipeline/asrpipe/revise.py index 4bc11d1..f1eb345 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/revise.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/revise.py @@ -54,12 +54,14 @@ def _safe_title(markdown: str) -> str: return title[:72] or "通话记录" -def _formal_path(markdown: str) -> Path: +def _formal_path(markdown: str, root: Path | None = None) -> Path: + target_dir = root or CFG.minutes + target_dir.mkdir(parents=True, exist_ok=True) date = _field(markdown, "date") or time.strftime("%Y-%m-%d") - base = CFG.minutes / f"{date}-{_safe_title(markdown)}.md" + base = target_dir / f"{date}-{_safe_title(markdown)}.md" transcript = _field(markdown, "transcript") - if CFG.minutes.exists() and transcript: - for path in CFG.minutes.glob("*.md"): + if target_dir.exists() and transcript: + for path in target_dir.glob("*.md"): try: if _field(path.read_text(errors="ignore"), "transcript") == transcript: return path @@ -256,13 +258,21 @@ def publish(path: Path) -> bool: destination = _formal_path(out) vault.write_draft_file(destination, out) + vault_minutes_dir = CFG.vault / "minutes" + vault_destination = ( + destination + if destination.parent.resolve() == vault_minutes_dir.resolve() + else (vault_minutes_dir / destination.name) + ) + if vault_destination != destination: + vault.write_draft_file(vault_destination, out) date = _field(out, "date") or time.strftime("%Y-%m-%d") - touched = [destination] + touched = [vault_destination] for name, kind in _answered_card_requests(path): - card = _create_card(name, kind, destination, date) + card = _create_card(name, kind, vault_destination, date) if card: touched.append(card) - state = vault.sync(touched, f"minutes: {destination.stem}（Web 确认后正式发布）") + state = vault.sync(touched, f"minutes: {vault_destination.stem}（Web 确认后正式发布）") if state == "error": raise RuntimeError("formal minute commit failed") diff --git a/servers/tools-pve/asr-pipeline/asrpipe/segment.py b/servers/tools-pve/asr-pipeline/asrpipe/segment.py index e0ed72f..a2ff994 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/segment.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/segment.py @@ -22,7 +22,7 @@ import re from dataclasses import dataclass, field VOICEPRINT_HDR = re.compile(r"^") +PROVIDER_HDR = re.compile(r"^", re.M) 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) @@ -148,8 +148,10 @@ def parse(raw: str) -> tuple[list[Turn], dict]: continue body_lines.append(ln) - backend = detect_backend(raw) + backend = meta["asr_provider"] or str(meta["intake"].get("asr_provider") or "") or detect_backend(raw) meta["backend"] = backend + if not meta["asr_model"] and meta["intake"].get("asr_model"): + meta["asr_model"] = str(meta["intake"]["asr_model"]) turns: list[Turn] = [] if backend == "vibevoice": diff --git a/servers/tools-pve/asr-pipeline/asrpipe/vault.py b/servers/tools-pve/asr-pipeline/asrpipe/vault.py index 2a57b76..a1aa790 100644 --- a/servers/tools-pve/asr-pipeline/asrpipe/vault.py +++ b/servers/tools-pve/asr-pipeline/asrpipe/vault.py @@ -90,8 +90,26 @@ def update_pending(date: str, title: str, path: Path) -> None: def _git(*args, check=False) -> subprocess.CompletedProcess: - return subprocess.run(["git", *args], cwd=CFG.vault, capture_output=True, - text=True, timeout=180, check=check) + return subprocess.run( + ["git", "-c", f"safe.directory={CFG.vault}", *args], + cwd=CFG.vault, capture_output=True, + text=True, timeout=180, check=check, + ) + + +def pull() -> str: + """Pull latest remote changes into CFG.vault when git_sync is enabled.""" + if not CFG.git_sync or not (CFG.vault / ".git").exists(): + return "disabled" + try: + r = _git("pull", "--rebase", "--autostash", "-q") + if r.returncode != 0: + log(f"git pull failed: {r.stderr.strip()[:200]}", level="WARN") + return "error" + return "pulled" + except Exception as e: + log(f"git pull error: {e}", level="WARN") + return "error" def sync(paths: list[Path], message: str) -> str: @@ -99,9 +117,18 @@ def sync(paths: list[Path], message: str) -> str: if not CFG.git_sync: return "disabled" try: - rel = [str(p.relative_to(CFG.vault)) for p in paths if p.exists()] + vault_root = CFG.vault.resolve() + rel = [] + for p in paths: + if not p.exists(): + continue + try: + rel.append(str(p.resolve().relative_to(vault_root))) + except ValueError: + continue if not rel: return "noop" + _git("pull", "--rebase", "--autostash", "-q") _git("add", "--", *rel) if _git("diff", "--cached", "--quiet").returncode == 0: log("nothing staged (likely already committed by the auto-sync cron)") diff --git a/servers/tools-pve/asr-pipeline/asrservice/web.py b/servers/tools-pve/asr-pipeline/asrservice/web.py index 46acf21..cf0d5fa 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/web.py +++ b/servers/tools-pve/asr-pipeline/asrservice/web.py @@ -1808,9 +1808,10 @@ class Handler(BaseHTTPRequestHandler): return self.redirect("/lexicon?tab=relations&msg=" + urllib.parse.quote(f"已删除关系：{old_s} {old_p} {old_o}")) if path == "/lexicon/sync": + p_res = vault.pull() l_res = lexicon.sync_from_vault(force=True) g_res = graph.sync(force=True) - return self.redirect("/lexicon?msg=" + urllib.parse.quote(f"同步完成：词库 {l_res}，图谱 {g_res}")) + return self.redirect("/lexicon?msg=" + urllib.parse.quote(f"同步完成（Git:{p_res}）：词库 {l_res}，图谱 {g_res}")) # ---- SQLite Settings Save ---- if path == "/settings": diff --git a/servers/tools-pve/asr-pipeline/asrservice/worker.py b/servers/tools-pve/asr-pipeline/asrservice/worker.py index 167f30d..c632364 100644 --- a/servers/tools-pve/asr-pipeline/asrservice/worker.py +++ b/servers/tools-pve/asr-pipeline/asrservice/worker.py @@ -191,13 +191,14 @@ def scan_inbox() -> None: process_audio(source, job) except Exception as exc: attempts = int((db.get_job(job["id"]) or {}).get("attempts") or 0) - retry = attempts < MAX_ATTEMPTS + is_empty_audio = isinstance(exc, providers.EmptyAudioError) + retry = (attempts < MAX_ATTEMPTS) and not is_empty_audio 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}") + log(f"job={job['id']} failed type={type(exc).__name__} retry={retry} err={exc}") target = INBOX / source.name if retry else FAILED / source.name if source.exists(): shutil.move(str(source), target) @@ -255,6 +256,7 @@ def recover() -> None: def sync_vault_knowledge() -> None: try: + vault.pull() lexicon.sync_from_vault() graph.sync() except Exception as exc: diff --git a/servers/tools-pve/asr-pipeline/compose.yaml b/servers/tools-pve/asr-pipeline/compose.yaml index 2f25a5c..9fbd911 100644 --- a/servers/tools-pve/asr-pipeline/compose.yaml +++ b/servers/tools-pve/asr-pipeline/compose.yaml @@ -42,6 +42,7 @@ services: volumes: - /mnt/truenas/personal/ASR-Pipeline:/data - /opt/asr-pipeline/state:/state + - /opt/asr-pipeline/ssh:/root/.ssh:ro healthcheck: test: ["CMD", "curl", "-fsS", "http://127.0.0.1:8788/health"] interval: 30s @@ -63,6 +64,7 @@ services: volumes: - /mnt/truenas/personal/ASR-Pipeline:/data - /opt/asr-pipeline/state:/state + - /opt/asr-pipeline/ssh:/root/.ssh:ro healthcheck: test: ["CMD", "python", "-m", "asrservice.healthcheck"] interval: 30s diff --git a/servers/tools-pve/asr-pipeline/tests/test_service.py b/servers/tools-pve/asr-pipeline/tests/test_service.py index 569b3cc..c18ec62 100644 --- a/servers/tools-pve/asr-pipeline/tests/test_service.py +++ b/servers/tools-pve/asr-pipeline/tests/test_service.py @@ -71,15 +71,20 @@ class ProviderAndServiceTests(unittest.TestCase): ] } text = providers.transcript_from_soniox_response(payload, model="stt-async-v5") - parsed_turns, meta = segment.parse(text) + with_intake = '\n' + text + parsed_turns, meta = segment.parse(with_intake) self.assertEqual(meta["asr_provider"], "soniox") self.assertEqual(meta["asr_model"], "stt-async-v5") + self.assertEqual(meta["backend"], "soniox") 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, "收到，我来汇报。") + with self.assertRaises(providers.EmptyAudioError): + providers.transcript_from_soniox_response({"tokens": [], "text": ""}, model="stt-async-v5") + def test_soniox_context_includes_lexicon_and_intake(self): lexicon.learn_alias("lite网关", "LiteLLM", note="test") ctx = providers.build_soniox_context(