#!/usr/bin/env python3 """Configuration for the standalone ASR post-processing pipeline and web service. 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. """ from __future__ import annotations import os from dataclasses import dataclass, field from pathlib import Path PKG = Path(__file__).resolve().parent ROOT = PKG.parent OBSIDIAN = Path(os.environ.get("ASR_WORKSPACE", str(ROOT))) def _env(name: str, default: str) -> str: v = os.environ.get(name) return v if v not in (None, "") else default def _envf(name: str, default: float) -> float: try: return float(os.environ["ASR_" + name]) except (KeyError, ValueError): return default def _envi(name: str, default: int) -> int: try: return int(os.environ["ASR_" + name]) except (KeyError, ValueError): return default @dataclass class Config: # ---- paths ---------------------------------------------------------- vault: Path = field(default_factory=lambda: Path( _env("ASR_VAULT", str(Path.home() / "obsidian-vault")))) obsidian: Path = field(default_factory=lambda: Path( _env("ASR_WORKSPACE", str(ROOT)))) root: Path = ROOT # ---- 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("/")) 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 = 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 = 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 = 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 = 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: return self.vault / ".asr-inbox" @property def state(self) -> Path: return self.inbox / ".state" @property def drafts(self) -> Path: """Private rough drafts. They are never part of the Obsidian repository.""" return Path(_env("ASR_DRAFTS_DIR", str(self.state_dir / "drafts"))) @property def minutes(self) -> Path: return Path(_env("ASR_RESULTS_DIR", str(self.vault / "minutes"))) @property def pending_index(self) -> Path: return self.vault / "minutes" / "_drafts" / "_待确认.md" @property def glossary(self) -> Path: return self.vault / "minutes" / "_glossary.md" @property def termbank_dir(self) -> Path: return Path(_env("ASR_TERMBANK_DIR", str(self.root / "termbank"))) @property def cache_dir(self) -> Path: return Path(_env("ASR_CACHE_DIR", str(self.root / "cache"))) @property def state_dir(self) -> Path: """Program-owned persistent state. Never inside the vault.""" return Path(_env("ASR_STATE_DIR", str(self.root / "state"))) @property def lexicon_db(self) -> Path: return Path(_env("ASR_LEXICON_DB", str(self.state_dir / "lexicon.db"))) @property def log_file(self) -> Path: return Path(_env("ASR_LOG", str(self.obsidian / "logs" / "asr-pipeline.log"))) @property def notifier(self) -> Path: return Path(_env("ASR_NOTIFIER", str(ROOT / "scripts" / "meeting-status-notify.py"))) @property def env_file(self) -> Path: """600-perm credentials file.""" return Path(_env("ASR_ENV_FILE", str(ROOT / ".env"))) def api_key(self) -> str: 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 p = self.env_file if p.exists(): for line in p.read_text(errors="ignore").splitlines(): line = line.strip() 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 LITELLM_API_KEY / ASR_LLM_API_KEY / OPENAI_API_KEY " f"or provide {p}") CFG = Config()