env_config.py

#!/usr/bin/env python3
"""Load KEY=VALUE settings from config.env (project root).

CLI flags always win over config.env values.
"""

from __future__ import annotations

import os
import shlex
from pathlib import Path

ROOT = Path(__file__).resolve().parent
DEFAULT_ENV_PATH = ROOT / "config.env"
# Alternate configs (e.g. config.en_pidgin.env) via CLI --config-env or XTTS_CONFIG_ENV.

# Always take these from config.env so a broken shell mirror/token cannot win.
_CONFIG_FORCE_KEYS = frozenset(
    {
        "HF_ENDPOINT",
        "HF_TOKEN",
        "HUGGINGFACE_ACCESS_TOKEN",
        "HUGGING_FACE_HUB_TOKEN",
        "HUGGINGFACE_TOKEN",
        "HF_HUB_DOWNLOAD_TIMEOUT",
        "HF_HUB_ETAG_TIMEOUT",
        "HF_DOWNLOAD_RETRIES",
    }
)


def load_config_env(path: Path | None = None, *, override: bool = False) -> Path | None:
    """Parse config.env into os.environ.

    Returns the path loaded, or None if missing.
    Does not override existing environment variables unless override=True,
    except for `_CONFIG_FORCE_KEYS` (HF endpoint/token/timeouts).
    """
    env_path = (path or DEFAULT_ENV_PATH).resolve()
    if not env_path.is_file():
        return None

    for raw in env_path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[len("export ") :].strip()
        if "=" not in line:
            continue
        key, value = line.split("=", 1)
        key = key.strip()
        value = value.strip()
        if not key:
            continue
        if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'":
            value = value[1:-1]
        # Skip blank assignments (e.g. FOO=) so they don't poison Hub clients
        if value == "":
            continue
        if override or key not in os.environ or key in _CONFIG_FORCE_KEYS:
            os.environ[key] = value
    return env_path


def env_str(name: str, default: str | None = None) -> str | None:
    v = os.getenv(name)
    if v is None or v == "":
        return default
    return v


def env_int(name: str, default: int) -> int:
    v = os.getenv(name)
    if v is None or v == "":
        return default
    return int(v)


def env_float(name: str, default: float) -> float:
    v = os.getenv(name)
    if v is None or v == "":
        return default
    return float(v)


def env_path(name: str, default: str | Path) -> Path:
    v = os.getenv(name)
    return Path(v) if v else Path(default)


def env_list(name: str, default: list[str] | None = None) -> list[str]:
    """Split a config value into a list (whitespace or comma separated)."""
    v = os.getenv(name)
    if v is None or v.strip() == "":
        return list(default or [])
    normalized = v.replace(",", " ")
    parts = shlex.split(normalized)
    return [p for p in parts if p]


def early_config_path_from_argv(argv: list[str] | None = None) -> Path | None:
    """Peek argv for --config-env / --config-env=PATH before full argparse."""
    import sys

    args = argv if argv is not None else sys.argv[1:]
    for i, a in enumerate(args):
        if a == "--config-env" and i + 1 < len(args):
            return Path(args[i + 1])
        if a.startswith("--config-env="):
            return Path(a.split("=", 1)[1])
    return None


def resolve_config_env_path(path: Path | None = None) -> Path:
    """Resolve config file: explicit path > XTTS_CONFIG_ENV > config.env."""
    if path is not None:
        return path.resolve()
    env_name = (os.getenv("XTTS_CONFIG_ENV") or "").strip()
    if env_name:
        return Path(env_name).expanduser().resolve()
    return DEFAULT_ENV_PATH.resolve()


def ensure_config_loaded(path: Path | None = None) -> None:
    """Load config.env (or alternate). If path is None, also checks argv --config-env."""
    if path is None:
        path = early_config_path_from_argv()
    env_path = resolve_config_env_path(path)
    loaded = load_config_env(env_path)
    if loaded:
        print(f"[config] loaded {loaded}")
    else:
        example = ROOT / "config.env.example"
        print(
            f"[config] no config file at {env_path}"
            + (f" (copy from {example.name})" if example.is_file() else "")
        )
    apply_hf_token()


def _sync_hf_hub_endpoint(endpoint: str) -> None:
    """Keep huggingface_hub / datasets in sync if they were already imported."""
    endpoint = endpoint.rstrip("/")
    try:
        import huggingface_hub.constants as hub_constants

        hub_constants.ENDPOINT = endpoint
        hub_constants.HUGGINGFACE_CO_URL_TEMPLATE = endpoint + "/{repo_id}/resolve/{revision}/{filename}"
    except Exception:
        pass
    try:
        import datasets.config as ds_config

        ds_config.HF_ENDPOINT = endpoint
        ds_config.HUB_DATASETS_URL = endpoint + "/datasets/{repo_id}/resolve/{revision}/{path}"
    except Exception:
        pass


def apply_hf_token() -> None:
    """Normalize HF token aliases and sync Hub endpoint.

    HF_ENDPOINT from config.env already wins over the shell via
    `_CONFIG_FORCE_KEYS` — no need to `unset HF_ENDPOINT` before runs.
    """
    endpoint = (os.environ.get("HF_ENDPOINT") or "").strip().rstrip("/")
    if endpoint and not endpoint.startswith(("http://", "https://")):
        del os.environ["HF_ENDPOINT"]
        endpoint = ""
    if not endpoint:
        endpoint = "https://huggingface.co"
        os.environ["HF_ENDPOINT"] = endpoint

    _sync_hf_hub_endpoint(endpoint)
    print(f"[config] HF_ENDPOINT={endpoint}")

    # Longer timeouts help IncompleteRead on slow/unstable links
    for key, default in (
        ("HF_HUB_DOWNLOAD_TIMEOUT", "120"),
        ("HF_HUB_ETAG_TIMEOUT", "60"),
        ("HF_DOWNLOAD_RETRIES", "8"),
    ):
        val = env_str(key, default)
        if val:
            os.environ[key] = val

    token = (
        env_str("HF_TOKEN")
        or env_str("HUGGINGFACE_ACCESS_TOKEN")
        or env_str("HUGGING_FACE_HUB_TOKEN")
        or env_str("HUGGINGFACE_TOKEN")
    )
    if not token:
        print("[config] HF token not set (optional for public repos)")
        return

    os.environ["HF_TOKEN"] = token
    os.environ["HUGGING_FACE_HUB_TOKEN"] = token
    if not os.getenv("HUGGINGFACE_ACCESS_TOKEN"):
        os.environ["HUGGINGFACE_ACCESS_TOKEN"] = token

    print(f"[config] HF token loaded ({token[:6]}…{token[-4:] if len(token) > 10 else '****'})")

    try:
        from huggingface_hub import login

        login(token=token, add_to_git_credential=False)
    except Exception as e:
        print(f"[config] huggingface_hub.login skipped: {e}")