lingma-openai-gateway/app/config.py

from __future__ import annotations

import json
import os
from dataclasses import dataclass, field



def _csv_env(raw: str) -> list[str]:
    return [item.strip() for item in (raw or "").replace("\n", ",").split(",") if item.strip()]


@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
    admin_token: str
    metrics_public: bool
    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
    tool_forward_enabled: bool = False
    tool_allowlist: list[str] = field(default_factory=list)


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(),
        admin_token=os.getenv("ADMIN_TOKEN", "").strip(),
        metrics_public=_bool_env("METRICS_PUBLIC", False),
        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")),
        tool_forward_enabled=_bool_env("TOOL_FORWARD_ENABLED", True),
        tool_allowlist=_csv_env(os.getenv("TOOL_ALLOWLIST", "")),
    )