scripts/render-config.py

import argparse
import json
import os
import re
import sys
import urllib.request
from pathlib import Path
from typing import Any


USABLE_MODELS_DATASET = "alchoholpad/litellm-usable-models"
USABLE_MODEL_FILES = [
    ("usable-models.json", "chat"),
    ("usable-image-models.json", "image"),
    ("usable-vision-models.json", "vision"),
]


ENV_REF_RE = re.compile(r"os\.environ/([A-Za-z0-9_]+)")
NUMBERED_ENV_SLOT_RE = re.compile(r"_\d+$")
OPTIONAL_ENV_REFS = {
    "CLOUDFLARE_ACCOUNT_ID",
    "DATABASE_URL",
    "LITELLM_MASTER_KEY",
    "MODAL_API_BASE",
}
API_PROVIDER_ENVS = {
    "AGENTROUTER_API_KEY",
    "AIMLAPI_API_KEY",
    "ASSEMBLYAI_API_KEY",
    "BYTEZ_API_KEY",
    "CEREBRAS_API_KEY",
    "CHINAWHAPI_API_KEY",
    "CLOUDFLARE_API_TOKEN",
    "COHERE_API_KEY",
    "CONSOLECLOUD_API_KEY",
    "CREATEANYTHING_API_KEY",
    "DEEPGRAM_API_KEY",
    "DEEPSEEK_API_KEY",
    "DISCORD_TOKEN",
    "EDENAI_API_KEY",
    "ELECTRONHUB_API_KEY",
    "ELEVENLABS_API_KEY",
    "EXA_API_KEY",
    "GEMINI_API_KEY",
    "GENLABS_API_KEY",
    "GITHUB_API_KEY",
    "GROQ_API_KEY",
    "HUGGINGFACE_API_KEY",
    "HUGGINGFACE_API_KEY_1",
    "HUGGINGFACE_API_KEY_2",
    "IMAGEAPI_API_KEY",
    "INFERENCE_SH_API_KEY",
    "JINA_AI_API_KEY",
    "MAPLEFLOW_API_KEY",
    "MISTRAL_API_KEY",
    "MODAL_API_KEY",
    "MODELSLAB_API_KEY",
    "OLLAMA_API_KEY",
    "OPENAI_API_KEY",
    "OPENROUTER_API_KEY",
    "OPENROUTER_API_KEY_1",
    "OPENROUTER_API_KEY_2",
    "OPENROUTER_API_KEY_3",
    "OPENROUTER_API_KEY_4",
    "PLATFORMCOMFY_API_KEY",
    "POLLINATIONS_API_KEY",
    "POLLINATIONS_API_KEY_1",
    "PUTER_API_KEY",
    "SAMBANOVA_API_KEY",
    "STABLEHORDE_API_KEY",
    "STABLEHORDE_API_KEY_1",
    "TAVILY_API_KEY",
    "TWELVELABS_API_KEY",
    "VERCEL_AI_GATEWAY_API_KEY",
    "VOIDAI_API_KEY",
    "WORLDLABS_API_KEY",
    "XIAOMI_MIMO_API_KEY",
    "YOU_API_KEY",
}

PASS_THROUGH_ENDPOINTS: list[dict[str, Any]] = [
    {
        "path": "/genlabs",
        "target": "https://api.genlabs.dev/deca/v1",
        "include_subpath": True,
        "headers": {"Authorization": "Bearer os.environ/GENLABS_API_KEY"},
    },
    {
        "path": "/inference-sh",
        "target": "https://api.inference.sh",
        "include_subpath": True,
        "headers": {"Authorization": "Bearer os.environ/INFERENCE_SH_API_KEY"},
    },
    {
        "path": "/tavily",
        "target": "https://api.tavily.com",
        "include_subpath": True,
        "headers": {"Authorization": "Bearer os.environ/TAVILY_API_KEY"},
    },
    {
        "path": "/exa",
        "target": "https://api.exa.ai",
        "include_subpath": True,
        "headers": {"x-api-key": "os.environ/EXA_API_KEY"},
    },
    {
        "path": "/discord",
        "target": "https://api.zukijourney.com/v1",
        "include_subpath": True,
        "headers": {"Authorization": "Bearer os.environ/DISCORD_TOKEN"},
    },
    {
        "path": "/worldlabs",
        "target": "https://api.worldlabs.ai",
        "include_subpath": True,
        "headers": {"WLT-Api-Key": "os.environ/WORLDLABS_API_KEY"},
    },
    {
        "path": "/twelvelabs",
        "target": "https://api.twelvelabs.io",
        "include_subpath": True,
        "headers": {"x-api-key": "os.environ/TWELVELABS_API_KEY"},
    },
    {
        "path": "/stablehorde",
        "target": "https://aihorde.net/api",
        "include_subpath": True,
        "headers": {"apikey": "os.environ/STABLEHORDE_API_KEY"},
    },
    {
        "path": "/you",
        "target": "https://ydc-index.io",
        "include_subpath": True,
        "headers": {"X-API-Key": "os.environ/YOU_API_KEY"},
    },
    {
        "path": "/modal",
        "target": "os.environ/MODAL_API_BASE",
        "include_subpath": True,
        "headers": {"Authorization": "Bearer os.environ/MODAL_API_KEY"},
    },
]


def env(name: str) -> str:
    return os.environ.get(name, "").strip()


def unique(values: list[str]) -> list[str]:
    seen: set[str] = set()
    out: list[str] = []
    for value in values:
        if value in seen:
            continue
        seen.add(value)
        out.append(value)
    return out


def env_variant_names(
    base: str,
    *,
    include_numbered: bool = True,
    include_named: bool = True,
) -> list[str]:
    names: list[str] = []
    if env(base):
        names.append(base)

    if include_numbered:
        for index in range(1, 11):
            name = f"{base}_{index}"
            if env(name):
                names.append(name)

    if include_named and not NUMBERED_ENV_SLOT_RE.search(base):
        prefix = f"{base}_"
        for name in sorted(os.environ):
            if not name.startswith(prefix):
                continue
            suffix = name[len(prefix) :]
            if not suffix:
                continue
            if suffix.isdigit() and not include_numbered:
                continue
            if env(name):
                names.append(name)

    return unique(names)


def env_names(base: str) -> list[str]:
    return env_variant_names(base)


SECRET_TO_ENV = {
    "openrouter_ai": "OPENROUTER_API_KEY",
    "aistudio_google_com": "GEMINI_API_KEY",
    "huggingface_co": "HUGGINGFACE_API_KEY",
    "pollinations_ai": "POLLINATIONS_API_KEY",
    "discord_com": "DISCORD_TOKEN",
    "admin_mistral_ai": "MISTRAL_API_KEY",
    "aimlapi_com": "AIMLAPI_API_KEY",
    "api_stablehorde_net": "STABLEHORDE_API_KEY",
    "app_edenai_run": "EDENAI_API_KEY",
    "app_genlabs_dev": "GENLABS_API_KEY",
    "assemblyai_com": "ASSEMBLYAI_API_KEY",
    "chinawhapi_com": "CHINAWHAPI_API_KEY",
    "cloud_cerebras_ai": "CEREBRAS_API_KEY",
    "cloud_sambanova_ai": "SAMBANOVA_API_KEY",
    "console_deepgram_com": "DEEPGRAM_API_KEY",
    "console_groq_com": "GROQ_API_KEY",
    "dash_cloudflare_com": "CLOUDFLARE_API_TOKEN",
    "dashboard_cohere_com": "COHERE_API_KEY",
    "elevenlabs_io": "ELEVENLABS_API_KEY",
    "github_com": "GITHUB_API_KEY",
    "jina_ai": "JINA_AI_API_KEY",
    "imageapi_org": "IMAGEAPI_API_KEY",
    "mapleflow_io": "MAPLEFLOW_API_KEY",
    "modal_com": "MODAL_API_KEY",
    "modelslab_com": "MODELSLAB_API_KEY",
    "ollama_com": "OLLAMA_API_KEY",
    "pawan_krd": "Pawan_Krd",
    "platform_deepseek_com": "DEEPSEEK_API_KEY",
    "platform_openai_com": "OPENAI_API_KEY",
    "playground_electronhub_ai": "ELECTRONHUB_API_KEY",
    "stablehorde_net": "STABLEHORDE_API_KEY",
    "vercel_com": "VERCEL_AI_GATEWAY_API_KEY",
    "voidai_app": "VOIDAI_API_KEY",
    "xiaomimimo_com": "XIAOMI_MIMO_API_KEY",
}


def _load_domain_email_secrets():
    """Map service_domain_email HF secrets to expected env var names."""
    for name, value in list(os.environ.items()):
        if not value or name.startswith("HF_"):
            continue
        for prefix, env_var in SECRET_TO_ENV.items():
            if name.startswith(prefix + "_") and name != env_var:
                if not os.environ.get(env_var):
                    os.environ[env_var] = value
                break


def catalog_env_names(api_key_env: str) -> list[str]:
    if NUMBERED_ENV_SLOT_RE.search(api_key_env):
        return [api_key_env] if env(api_key_env) else []
    names = env_variant_names(api_key_env, include_numbered=False)
    # Also find domain_email suffixed vars (e.g. aistudio_google_com_fahadbinhussain001_gmail_com)
    prefix = api_key_env + "_"
    for name in sorted(os.environ):
        if name.startswith(prefix) and name != api_key_env and env(name):
            if name not in names:
                names.append(name)
    return unique(names)


def download_usable_models() -> set[str]:
    """Download usable model IDs from the HF dataset."""
    hf_token = os.environ.get("HF_TOKEN", "").strip()
    usable_ids: set[str] = set()

    for filename, label in USABLE_MODEL_FILES:
        url = f"https://huggingface.co/datasets/{USABLE_MODELS_DATASET}/resolve/main/{filename}"
        headers = {"User-Agent": "litellm-render-config/1.0"}
        if hf_token:
            headers["Authorization"] = f"Bearer {hf_token}"
        try:
            req = urllib.request.Request(url, headers=headers)
            with urllib.request.urlopen(req, timeout=30) as resp:
                data = json.loads(resp.read().decode())
                ids = data.get("usable_model_ids", [])
                usable_ids.update(ids)
                print(f"  Loaded {len(ids)} usable {label} models from dataset", file=sys.stderr)
        except Exception as e:
            print(f"  Warning: could not fetch {filename}: {e}", file=sys.stderr)

    return usable_ids


def load_usable_models(path: Path | None) -> set[str]:
    """Load usable model IDs from a local file or HF dataset."""
    if path and path.exists():
        data = json.loads(path.read_text(encoding="utf-8"))
        ids = data.get("usable_model_ids", [])
        if isinstance(ids, str):
            ids = ids.split()
        print(f"Loaded {len(ids)} usable models from {path}", file=sys.stderr)
        return set(ids)

    print("No local usable models file, fetching from HF dataset...", file=sys.stderr)
    return download_usable_models()


def load_secrets(path: Path) -> int:
    payload = json.loads(path.read_text(encoding="utf-8-sig"))
    if not isinstance(payload, dict):
        raise ValueError(f"{path} must contain a JSON object.")

    loaded = 0
    for name, value in payload.items():
        if not isinstance(name, str) or not isinstance(value, str) or not value:
            continue
        os.environ[name] = value
        loaded += 1
    return loaded


def yaml_quote(value: str) -> str:
    escaped = value.replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def render_scalar(value: Any) -> str:
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    if value is None:
        return "null"
    return yaml_quote(str(value))


def render_yaml(value: Any, indent: int = 0) -> list[str]:
    prefix = " " * indent
    if isinstance(value, dict):
        if not value:
            return [f"{prefix}{{}}"]
        lines: list[str] = []
        for key, child in value.items():
            if isinstance(child, (dict, list)):
                lines.append(f"{prefix}{key}:")
                lines.extend(render_yaml(child, indent + 2))
            else:
                lines.append(f"{prefix}{key}: {render_scalar(child)}")
        return lines
    if isinstance(value, list):
        if not value:
            return [f"{prefix}[]"]
        lines = []
        for item in value:
            if isinstance(item, dict):
                if not item:
                    lines.append(f"{prefix}- {{}}")
                    continue
                first = True
                for key, child in item.items():
                    marker = "- " if first else "  "
                    if isinstance(child, (dict, list)):
                        lines.append(f"{prefix}{marker}{key}:")
                        lines.extend(render_yaml(child, indent + 4))
                    else:
                        lines.append(f"{prefix}{marker}{key}: {render_scalar(child)}")
                    first = False
            elif isinstance(item, list):
                lines.append(f"{prefix}-")
                lines.extend(render_yaml(item, indent + 2))
            else:
                lines.append(f"{prefix}- {render_scalar(item)}")
        return lines
    return [f"{prefix}{render_scalar(value)}"]


def add_model(
    models: list[dict[str, Any]],
    alias: str,
    model: str,
    model_info: dict[str, Any] | None = None,
    **params: str,
) -> None:
    litellm_params = {"model": model}
    litellm_params.update({key: value for key, value in params.items() if value is not None})
    entry: dict[str, Any] = {"model_name": alias, "litellm_params": litellm_params}
    if model_info:
        entry["model_info"] = model_info
    models.append(entry)


def suffixed(alias: str, index: int, total: int) -> str:
    return alias if total == 1 else f"{alias}-{index}"


def build_legacy_models() -> list[dict[str, Any]]:
    models: list[dict[str, Any]] = []

    gemini_keys = env_names("GEMINI_API_KEY") + env_names("CONSOLECLOUD_API_KEY")
    for index, key_name in enumerate(gemini_keys, start=1):
        suffix_total = len(gemini_keys)
        add_model(
            models,
            suffixed("gemini-flash", index, suffix_total),
            "gemini/gemini-2.5-flash",
            api_key=f"os.environ/{key_name}",
        )
        add_model(
            models,
            suffixed("gemini-flash-lite", index, suffix_total),
            "gemini/gemini-2.5-flash-lite",
            api_key=f"os.environ/{key_name}",
        )
        add_model(
            models,
            suffixed("gemini-pro", index, suffix_total),
            "gemini/gemini-2.5-pro",
            api_key=f"os.environ/{key_name}",
        )

    openrouter_base = env("OPENROUTER_API_BASE_URL") or "https://openrouter.ai/api/v1"
    openrouter_keys = env_names("OPENROUTER_API_KEY")
    for index, key_name in enumerate(openrouter_keys, start=1):
        add_model(
            models,
            suffixed("openrouter-auto", index, len(openrouter_keys)),
            "openrouter/auto",
            api_key=f"os.environ/{key_name}",
            api_base=openrouter_base,
        )

    openai_keys = env_names("OPENAI_API_KEY")
    for index, key_name in enumerate(openai_keys, start=1):
        add_model(
            models,
            suffixed("openai-fast", index, len(openai_keys)),
            "openai/gpt-4o-mini",
            api_key=f"os.environ/{key_name}",
        )

    anthropic_keys = env_names("ANTHROPIC_API_KEY")
    for index, key_name in enumerate(anthropic_keys, start=1):
        add_model(
            models,
            suffixed("claude-haiku", index, len(anthropic_keys)),
            "anthropic/claude-3-5-haiku-latest",
            api_key=f"os.environ/{key_name}",
        )

    custom_base = env("CUSTOM_OPENAI_API_BASE")
    if custom_base:
        custom_alias = env("CUSTOM_OPENAI_ALIAS") or "custom-openai"
        custom_model = env("CUSTOM_OPENAI_MODEL") or "gpt-4o-mini"
        params = {"api_base": custom_base}
        if env("CUSTOM_OPENAI_API_KEY"):
            params["api_key"] = "os.environ/CUSTOM_OPENAI_API_KEY"
        add_model(models, custom_alias, f"openai/{custom_model}", **params)

    return models


def default_catalog_path(template_path: Path) -> Path:
    return template_path.resolve().parent / "model-catalog.json"


def load_model_catalog(path: Path, usable_ids: set[str] | None = None) -> list[dict[str, Any]]:
    catalog = json.loads(path.read_text(encoding="utf-8"))
    if catalog.get("version") != 1:
        raise ValueError(f"Unsupported model catalog version in {path}")

    # Build a set of usable suffixes for matching (strip provider prefix from usable IDs)
    usable_suffixes: set[str] | None = None
    if usable_ids is not None:
        usable_suffixes = set()
        for uid in usable_ids:
            if "/" in uid:
                usable_suffixes.add(uid.split("/", 1)[1])
            else:
                usable_suffixes.add(uid)
        usable_suffixes.update(usable_ids)

    models: list[dict[str, Any]] = []
    seen_models: set[str] = set()  # Track added model IDs to deduplicate
    for group in catalog.get("groups", []):
        api_key_env = group.get("api_key_env")
        literal_api_key = group.get("literal_api_key")
        env_slots: list[str | None]
        if api_key_env:
            env_slots = catalog_env_names(str(api_key_env))
            if not env_slots:
                continue
        else:
            env_slots = [None]

        for env_slot in env_slots:
            params = dict(group.get("params") or {})
            if env_slot:
                params["api_key"] = f"os.environ/{env_slot}"
            elif literal_api_key is not None:
                params["api_key"] = literal_api_key

            for suffix in group.get("suffixes", []):
                if isinstance(suffix, dict):
                    alias_suffix = suffix["alias"]
                    model_suffix = suffix["model"]
                else:
                    alias_suffix = str(suffix)
                    model_suffix = str(suffix)

                # Filter by usable models if list is provided
                if usable_suffixes is not None:
                    full_model_id = f"{group['model_prefix']}/{model_suffix}"
                    if (full_model_id not in usable_suffixes and
                            model_suffix not in usable_suffixes):
                        continue

                # Deduplicate: skip if this model ID was already added
                model_id = f"{group['model_prefix']}/{model_suffix}"
                if model_id in seen_models:
                    continue
                seen_models.add(model_id)

                add_model(
                    models,
                    f"{group['alias_prefix']}/{alias_suffix}",
                    f"{group['model_prefix']}/{model_suffix}",
                    model_info=group.get("model_info"),
                    **params,
                )

    return models


def render_models(models: list[dict[str, Any]]) -> str:
    if not models:
        return "  []"
    return "\n".join(render_yaml(models, indent=2))


def render_general_settings() -> str:
    settings: dict[str, Any] = {
        "pass_through_endpoints": PASS_THROUGH_ENDPOINTS,
    }
    if env("LITELLM_MASTER_KEY"):
        settings = {"master_key": "os.environ/LITELLM_MASTER_KEY", **settings}
    if env("DATABASE_URL"):
        settings["database_url"] = "os.environ/DATABASE_URL"

    return "\n".join(render_yaml(settings, indent=2))


def render_template(template: str, models: list[dict[str, Any]]) -> str:
    rendered = template
    if "__AUTO_MODEL_LIST__" in rendered:
        rendered = rendered.replace("__AUTO_MODEL_LIST__", render_models(models))
    if "__GENERAL_SETTINGS__" in rendered:
        rendered = rendered.replace("__GENERAL_SETTINGS__", render_general_settings())
    return rendered


def env_refs(text: str) -> set[str]:
    return set(ENV_REF_RE.findall(text))


def is_api_provider_env_ref(name: str) -> bool:
    if name in API_PROVIDER_ENVS:
        return True
    for base in API_PROVIDER_ENVS:
        if not NUMBERED_ENV_SLOT_RE.search(base) and name.startswith(f"{base}_"):
            return True
    return False


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("template", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument(
        "--catalog",
        type=Path,
        help="Model catalog JSON to expand into model_list. Defaults to config/model-catalog.json beside the template.",
    )
    parser.add_argument(
        "--usable-models",
        type=Path,
        default=None,
        help="Path to usable-models.json. If not provided, fetches from HF dataset. Use --no-usable-models to disable filtering.",
    )
    parser.add_argument(
        "--no-usable-models",
        action="store_true",
        help="Disable usable models filtering (load all models from catalog).",
    )
    parser.add_argument(
        "--include-legacy-aliases",
        action="store_true",
        help="Also add the old short aliases such as gemini-flash and openai-fast.",
    )
    parser.add_argument(
        "--secrets",
        type=Path,
        default=Path(os.environ["LITELLM_SECRETS_FILE"]) if os.environ.get("LITELLM_SECRETS_FILE") else None,
        help="Optional local JSON secret file. Values are loaded only into this process.",
    )
    parser.add_argument("--strict-env", action="store_true")
    parser.add_argument("--summary-json", action="store_true")
    args = parser.parse_args()

    secrets_loaded = 0
    if args.secrets:
        secrets_loaded = load_secrets(args.secrets)

    _load_domain_email_secrets()

    # Load usable models filter
    usable_ids: set[str] | None = None
    if not args.no_usable_models:
        usable_ids = load_usable_models(args.usable_models)

    catalog_path = args.catalog or default_catalog_path(args.template)
    models = load_model_catalog(catalog_path, usable_ids=usable_ids)
    if args.include_legacy_aliases:
        models.extend(build_legacy_models())

    template = args.template.read_text(encoding="utf-8")
    rendered = render_template(template, models)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(rendered, encoding="utf-8")

    refs = env_refs(rendered)
    present = {name for name in refs if env(name)}
    missing_required = sorted(refs - present - OPTIONAL_ENV_REFS)
    missing_optional = sorted((refs - present) & OPTIONAL_ENV_REFS)
    api_refs = sorted(name for name in refs if is_api_provider_env_ref(name))
    api_present = sorted(set(api_refs) & present)
    missing_api_refs = sorted(set(api_refs) - present)
    summary = {
        "template": str(args.template),
        "catalog": str(catalog_path),
        "output": str(args.output),
        "secretsLoaded": secrets_loaded,
        "usableModelsFilter": usable_ids is not None,
        "usableModelCount": len(usable_ids) if usable_ids else 0,
        "models": len(models),
        "apiProviderEnvRefs": len(api_refs),
        "apiProviderEnvRefsPresent": len(api_present),
        "missingApiProviderEnvRefs": missing_api_refs,
        "envRefs": len(refs),
        "envRefsPresent": len(present),
        "missingRequired": missing_required,
        "missingOptional": missing_optional,
    }

    if args.summary_json:
        print(json.dumps(summary, indent=2), file=sys.stderr)
    else:
        filter_info = f", usable filter={len(usable_ids)}" if usable_ids else ""
        print(
            "Rendered LiteLLM config "
            f"({len(models)} models{filter_info}, {len(api_refs)} API provider env refs, "
            f"{len(api_present)} API provider env refs present, {len(refs)} total env refs, "
            f"{len(missing_required)} required missing).",
            file=sys.stderr,
        )

    if args.strict_env and missing_required:
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())