backend/server_runtime.py

from __future__ import annotations

import os
import sys
from pathlib import Path
from typing import Any, Callable, Mapping, MutableMapping, Protocol

from dotenv import load_dotenv

from backend.runtime_utils import (
    can_bind_tcp_port,
    find_free_tcp_port,
    prepare_runtime_environment,
)

DEFAULT_HUGGINGFACE_PORT = 7860
ADMIN_TOKEN_ENV_KEY = "ADMIN_TOKEN"


class LoggerLike(Protocol):
    def warning(self, msg: str, *args: Any, **kwargs: Any) -> None: ...


def is_huggingface_space(env: Mapping[str, str] | None = None) -> bool:
    runtime_env = env or os.environ
    return bool(runtime_env.get("SPACE_ID") or runtime_env.get("SPACE_HOST"))


def bootstrap_runtime_port(
    env: MutableMapping[str, str] | None = None,
    *,
    huggingface_port: int = DEFAULT_HUGGINGFACE_PORT,
) -> None:
    runtime_env = env or os.environ
    if is_huggingface_space(runtime_env) and not runtime_env.get("PORT", "").strip():
        runtime_env["PORT"] = str(huggingface_port)


def load_runtime_env(
    env_file: Path,
    *,
    override: bool = False,
) -> bool:
    if not env_file.exists():
        return False
    load_dotenv(env_file, override=override)
    return True


def is_admin_token_configured(
    env: Mapping[str, str] | None = None,
    *,
    token_env_key: str = ADMIN_TOKEN_ENV_KEY,
) -> bool:
    runtime_env = os.environ if env is None else env
    return bool(runtime_env.get(token_env_key, "").strip())


def require_admin_token_configured(
    env: Mapping[str, str] | None = None,
    *,
    token_env_key: str = ADMIN_TOKEN_ENV_KEY,
) -> None:
    if is_admin_token_configured(env, token_env_key=token_env_key):
        return
    raise RuntimeError(
        "ADMIN_TOKEN chua duoc cau hinh. Tu choi khoi dong runtime de tranh fallback auth mac dinh."
    )


def load_fastapi_app(
    project_root: Path,
    *,
    module_name: str = "backend.main",
    attr_name: str = "app",
) -> Any:
    prepare_runtime_environment(project_root)
    project_root_str = str(project_root)
    if project_root_str not in sys.path:
        sys.path.insert(0, project_root_str)

    module = __import__(module_name, fromlist=[attr_name])
    return getattr(module, attr_name)


def resolve_server_port(
    host: str,
    *,
    env: MutableMapping[str, str] | None = None,
    logger: LoggerLike | None = None,
    huggingface_port: int = DEFAULT_HUGGINGFACE_PORT,
    port_checker: Callable[[str, int], bool] = can_bind_tcp_port,
    free_port_finder: Callable[[str], int] = find_free_tcp_port,
) -> int:
    runtime_env = env or os.environ
    raw_port = runtime_env.get("PORT", "").strip()

    if raw_port:
        try:
            configured_port = int(raw_port)
        except ValueError as exc:
            raise ValueError(f"Invalid PORT value: {raw_port}") from exc

        if is_huggingface_space(runtime_env) or port_checker(host, configured_port):
            return configured_port

        fallback_port = free_port_finder(host)
        runtime_env["PORT"] = str(fallback_port)
        if logger is not None:
            logger.warning(
                "PORT %s dang ban tren %s, tu dong chuyen sang cong %s",
                configured_port,
                host,
                fallback_port,
            )
        return fallback_port

    if is_huggingface_space(runtime_env):
        runtime_env["PORT"] = str(huggingface_port)
        return huggingface_port

    fallback_port = free_port_finder(host)
    runtime_env["PORT"] = str(fallback_port)
    return fallback_port