from __future__ import annotations import json import os from dataclasses import dataclass, field @dataclass class LingmaAccount: username: str password: str # Optional: pre-captured Lingma session to skip Playwright auto-login. # Either inline base64 of a tar.gz bundle, or a path on disk holding the # same. Inline wins if both are set. session_bundle_b64: str = "" session_bundle_file: str = "" @dataclass class Settings: host: str port: int api_keys: list[str] metrics_token: str log_level: str gateway_max_in_flight: int gateway_queue_timeout_sec: float lingma_bin: str lingma_work_dir: str lingma_socket_port: int lingma_startup_timeout: int lingma_rpc_timeout: int default_model: str default_ask_mode: str dedicated_domain_url: str auto_login_enabled: bool auto_login_headless: bool auto_login_timeout: int auto_login_max_retry: int accounts: list[LingmaAccount] = field(default_factory=list) instance_count: int = 1 session_reuse_enabled: bool = True session_cache_max_entries: int = 256 session_cache_ttl_sec: float = 1800.0 def _bool_env(name: str, default: bool) -> bool: raw = os.getenv(name) if raw is None: return default return raw.strip().lower() in {"1", "true", "yes", "on"} def _parse_accounts(raw: str) -> list[LingmaAccount]: """Parse LINGMA_ACCOUNTS. Accepted formats: - JSON array: `[{"username":"u1","password":"p1"},{"username":"u2","password":"p2"}]` - CSV: `u1:p1,u2:p2` - Newlines: `u1:p1\nu2:p2` Whitespace around entries is trimmed. Empty entries are ignored. Passwords containing ':' are supported (only the first ':' is the separator). """ raw = (raw or "").strip() if not raw: return [] if raw.startswith("["): try: data = json.loads(raw) except Exception: return [] out: list[LingmaAccount] = [] if isinstance(data, list): for item in data: if isinstance(item, dict): u = str(item.get("username", "")).strip() p = str(item.get("password", "")).strip() bundle = str(item.get("session_bundle", "")).strip() bundle_file = str(item.get("session_bundle_file", "")).strip() # Username/password become optional when a bundle is supplied: # Playwright login is only needed if there's no pre-captured session. if (u and p) or bundle or bundle_file: out.append( LingmaAccount( username=u, password=p, session_bundle_b64=bundle, session_bundle_file=bundle_file, ) ) return out out: list[LingmaAccount] = [] for entry in raw.replace("\n", ",").split(","): entry = entry.strip() if not entry or ":" not in entry: continue u, p = entry.split(":", 1) u, p = u.strip(), p.strip() if u and p: out.append(LingmaAccount(u, p)) return out def load_settings() -> Settings: keys_raw = os.getenv("API_KEYS", "") api_keys = [k.strip() for k in keys_raw.split(",") if k.strip()] work_dir = os.getenv( "LINGMA_WORK_DIR", "/app/data/.lingma/vscode/sharedClientCache", ) accounts = _parse_accounts(os.getenv("LINGMA_ACCOUNTS", "")) # LINGMA_SESSION_BUNDLE / LINGMA_SESSION_BUNDLE_FILE are singleton envs # that attach a session to the first account (or implicitly create one # when neither LINGMA_ACCOUNTS nor LINGMA_USERNAME is provided -- common # "I just want to skip Playwright" case). fallback_bundle = os.getenv("LINGMA_SESSION_BUNDLE", "").strip() fallback_bundle_file = os.getenv("LINGMA_SESSION_BUNDLE_FILE", "").strip() if not accounts: u = os.getenv("LINGMA_USERNAME", "").strip() p = os.getenv("LINGMA_PASSWORD", "").strip() if u and p: accounts.append(LingmaAccount(u, p)) elif fallback_bundle or fallback_bundle_file: # Bundle-only login: no creds needed. accounts.append(LingmaAccount(username="", password="")) if accounts and (fallback_bundle or fallback_bundle_file): # Only fill on account[0] if it doesn't already carry one (accounts # loaded from LINGMA_ACCOUNTS JSON may have per-entry bundles). if not accounts[0].session_bundle_b64 and not accounts[0].session_bundle_file: accounts[0].session_bundle_b64 = fallback_bundle accounts[0].session_bundle_file = fallback_bundle_file explicit_count = os.getenv("LINGMA_INSTANCE_COUNT", "").strip() if explicit_count: try: instance_count = max(1, int(explicit_count)) except ValueError: instance_count = len(accounts) or 1 else: instance_count = max(1, len(accounts)) if accounts else 1 return Settings( host=os.getenv("HOST", "0.0.0.0"), port=int(os.getenv("PORT", "8317")), api_keys=api_keys, metrics_token=os.getenv("METRICS_TOKEN", "").strip(), log_level=os.getenv("LOG_LEVEL", "INFO").strip() or "INFO", gateway_max_in_flight=int(os.getenv("GATEWAY_MAX_IN_FLIGHT", "4")), gateway_queue_timeout_sec=float(os.getenv("GATEWAY_QUEUE_TIMEOUT_SEC", "30")), lingma_bin=os.getenv("LINGMA_BIN", "/app/data/bin/Lingma"), lingma_work_dir=work_dir, lingma_socket_port=int(os.getenv("LINGMA_SOCKET_PORT", "36510")), lingma_startup_timeout=int(os.getenv("LINGMA_STARTUP_TIMEOUT", "40")), lingma_rpc_timeout=int(os.getenv("LINGMA_RPC_TIMEOUT", "30")), default_model=os.getenv("DEFAULT_MODEL", "org_auto"), default_ask_mode=os.getenv("DEFAULT_ASK_MODE", "chat"), dedicated_domain_url=os.getenv("DEDICATED_DOMAIN_URL", "").strip(), auto_login_enabled=_bool_env("AUTO_LOGIN_ENABLED", True), auto_login_headless=_bool_env("AUTO_LOGIN_HEADLESS", True), auto_login_timeout=int(os.getenv("AUTO_LOGIN_TIMEOUT", "180")), auto_login_max_retry=int(os.getenv("AUTO_LOGIN_MAX_RETRY", "2")), accounts=accounts, instance_count=instance_count, session_reuse_enabled=_bool_env("SESSION_REUSE_ENABLED", True), session_cache_max_entries=int(os.getenv("SESSION_CACHE_MAX_ENTRIES", "256")), session_cache_ttl_sec=float(os.getenv("SESSION_CACHE_TTL_SEC", "1800")), )