core-python/maris_core/utils/env.py

"""Vides mainīgo helperi Maris runtime noklusējumiem."""

from __future__ import annotations

import os
import re

MARIS_MODEL_NAMESPACE = os.getenv("MARIS_MODEL_NAMESPACE", "MarisUK/")
HF_REPO_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$")


def get_env(name: str, default: str | None = None) -> str | None:
    value = os.getenv(name)
    if value is None:
        return default
    normalized = value.strip()
    if normalized:
        return normalized
    return default


def require_env(name: str) -> str:
    value = get_env(name)
    if value is None:
        raise RuntimeError(f"Trūkst obligātais vides mainīgais: {name}")
    return value


def get_env_or_default(name: str, default: str) -> str:
    return get_env(name, default) or default


def get_env_any(*names: str, default: str | None = None) -> str | None:
    for name in names:
        value = get_env(name)
        if value is not None:
            return value
    return default


def get_env_any_or_default(*names: str, default: str) -> str:
    return get_env_any(*names, default=default) or default


def get_hf_token() -> str | None:
    return get_env_any(
        "HF_TOKEN",
        "HUGGING_FACE_HUB_TOKEN",
        "HUGGINGFACEHUB_API_TOKEN",
        "HUGGINGFACE_API_KEY",
        "MARIS_REPO_TOKEN",
        "MARIS_TOKEN",
    )


def validate_maris_repo(repo_id: str, source: str, *, label: str = "repozitorijs") -> str:
    normalized = validate_hf_repo_id(repo_id, source, label=label)
    if not normalized.startswith(MARIS_MODEL_NAMESPACE):
        raise RuntimeError(f"{source} jānorāda uz Maris AI {label} ({MARIS_MODEL_NAMESPACE}...).")
    return normalized


def validate_hf_repo_id(repo_id: str, source: str, *, label: str = "repozitorijs") -> str:
    normalized = repo_id.strip()
    if not normalized:
        raise RuntimeError(f"Trūkst Maris AI {label} konfigurācija: {source}")
    if not HF_REPO_ID_PATTERN.fullmatch(normalized):
        raise RuntimeError(f"{source} jābūt owner/name formātā.")
    return normalized


def validate_maris_model(model_id: str, source: str) -> str:
    return validate_maris_repo(model_id, source, label="modeli")


def validate_hf_model(model_id: str, source: str) -> str:
    return validate_hf_repo_id(model_id, source, label="modeli")


def get_maris_model(*names: str, default: str | None = None) -> str:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        if default is None:
            raise RuntimeError(f"Trūkst Maris AI modeļa konfigurācija: {source}")
        value = default
    if value is None:
        raise RuntimeError(f"Trūkst Maris AI modeļa konfigurācija: {source}")
    return validate_maris_model(value, source)


def get_optional_maris_model(*names: str) -> str | None:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        return None
    return validate_maris_model(value, source)


def get_hf_model(*names: str, default: str | None = None) -> str:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        if default is None:
            raise RuntimeError(f"Trūkst Hugging Face modeļa konfigurācija: {source}")
        value = default
    return validate_hf_model(value, source)


def get_optional_hf_model(*names: str) -> str | None:
    source = ", ".join(names)
    value = get_env_any(*names)
    if value is None:
        return None
    return validate_hf_model(value, source)