"""Environment-driven application configuration. Two model backends are supported: Cloud model — a remote OpenAI-compatible API (e.g. gpt-5.4 via a hosted endpoint). Used as the powerful multimodal backbone for tasks like audio transcription. Local model — a self-hosted model served via SGLang, vLLM, or any OpenAI-compatible server (e.g. Qwen3.5-2B on localhost). Used for high-throughput image recognition / classification. Both backends expose ``/v1/chat/completions``; the only difference is the base URL, API key, and model name. """ from __future__ import annotations import logging import os from dataclasses import dataclass from urllib.parse import unquote, urlsplit log = logging.getLogger(__name__) @dataclass(frozen=True) class Config: server_host: str server_port: int # Auth: YesCaptcha clientKey client_key: str | None # ── Cloud model (remote API) ── cloud_base_url: str cloud_api_key: str cloud_model: str # ── Local model (self-hosted via SGLang / vLLM) ── local_base_url: str local_api_key: str local_model: str captcha_retries: int captcha_timeout: int # Playwright browser browser_headless: bool browser_timeout: int # seconds # WARP SOCKS5 proxy for clean IP egress (e.g. socks5://user:pass@host:1080) warp_proxy: str | None # ── Convenience aliases (backward-compat) ── @property def captcha_base_url(self) -> str: return self.cloud_base_url @property def captcha_api_key(self) -> str: return self.cloud_api_key @property def captcha_model(self) -> str: return self.cloud_model @property def captcha_multimodal_model(self) -> str: return self.local_model @property def warp_proxy_playwright(self) -> dict[str, str] | None: if not self.warp_proxy: return None parsed = urlsplit(self.warp_proxy) if not parsed.scheme or not parsed.hostname or not parsed.port: return None if parsed.scheme.startswith("socks"): if parsed.username or parsed.password: log.warning( "Skipping Playwright proxy because Chromium does not support authenticated SOCKS5 proxies" ) return None # socks5h is treated identically to socks5 by Chromium (which always # resolves DNS through the proxy for SOCKS), but Chromium only accepts # "socks5" as the scheme – strip the "h" suffix if present. pw_scheme = "socks5" if parsed.scheme in ("socks5", "socks5h") else parsed.scheme return {"server": f"{pw_scheme}://{parsed.hostname}:{parsed.port}"} proxy: dict[str, str] = { "server": f"{parsed.scheme}://{parsed.hostname}:{parsed.port}" } if parsed.username: proxy["username"] = unquote(parsed.username) if parsed.password: proxy["password"] = unquote(parsed.password) return proxy def load_config() -> Config: return Config( server_host=os.environ.get("SERVER_HOST", "0.0.0.0"), server_port=int(os.environ.get("SERVER_PORT", "8000")), client_key=os.environ.get("CLIENT_KEY", "").strip() or None, # Cloud model cloud_base_url=os.environ.get( "CLOUD_BASE_URL", os.environ.get("CAPTCHA_BASE_URL", "https://your-openai-compatible-endpoint/v1"), ), cloud_api_key=os.environ.get( "CLOUD_API_KEY", os.environ.get("CAPTCHA_API_KEY", ""), ), cloud_model=os.environ.get( "CLOUD_MODEL", os.environ.get("CAPTCHA_MODEL", "gpt-5.4"), ), # Local model local_base_url=os.environ.get( "LOCAL_BASE_URL", os.environ.get("CAPTCHA_BASE_URL", "http://localhost:30000/v1"), ), local_api_key=os.environ.get( "LOCAL_API_KEY", os.environ.get("CAPTCHA_API_KEY", "EMPTY"), ), local_model=os.environ.get( "LOCAL_MODEL", os.environ.get("CAPTCHA_MULTIMODAL_MODEL", "Qwen/Qwen3.5-2B"), ), captcha_retries=int(os.environ.get("CAPTCHA_RETRIES", "3")), captcha_timeout=int(os.environ.get("CAPTCHA_TIMEOUT", "30")), browser_headless=os.environ.get("BROWSER_HEADLESS", "true").strip().lower() in {"1", "true", "yes"}, browser_timeout=int(os.environ.get("BROWSER_TIMEOUT", "30")), warp_proxy=os.environ.get("WARP_PROXY", "").strip() or None, ) config = load_config()