res_levis_API/config_env.py

import os
import re
from pathlib import Path
from dotenv import load_dotenv


def _load_dotenv_chain() -> None:
    # Load local .env first
    load_dotenv(override=True)

    # Support shell-style source directives in .env, e.g. ". /data/conf/presence/core.conf"
    local_env = Path(".env")
    if not local_env.exists():
        return

    for raw_line in local_env.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith(". "):
            target = line[2:].strip().strip('"').strip("'")
        elif line.startswith("source "):
            target = line[7:].strip().strip('"').strip("'")
        else:
            continue

        if not target:
            continue
        source_path = Path(target)
        if not source_path.is_absolute():
            source_path = (local_env.parent / source_path).resolve()
        if source_path.exists():
            load_dotenv(dotenv_path=source_path, override=True)

    # Safety fallback for deployments that keep Keycloak vars in core.conf.
    default_core_conf = Path("/data/conf/presence/core.conf")
    if default_core_conf.exists():
        load_dotenv(dotenv_path=default_core_conf, override=True)


_load_dotenv_chain()

_SHELL_VAR_PATTERN = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}|\$([A-Za-z_][A-Za-z0-9_]*)")


def _clean(value: str) -> str:
    if value is None:
        return ""
    text = str(value).strip().strip('"').strip("'")
    if not text:
        return ""

    # Expand ${VAR} and $VAR references using already loaded environment variables.
    def _replace(match: re.Match) -> str:
        var_name = match.group(1) or match.group(2)
        return os.getenv(var_name, "")

    return _SHELL_VAR_PATTERN.sub(_replace, text)


def _has_http_scheme(value: str) -> bool:
    return value.startswith("http://") or value.startswith("https://")


def _ensure_http_scheme(value: str, default_scheme: str = "https") -> str:
    value = _clean(value)
    if not value:
        return ""
    if _has_http_scheme(value):
        return value
    if value.startswith("/"):
        return value
    return f"{default_scheme}://{value}"


def _absolute_or_join(value: str, base_url: str) -> str:
    value = _clean(value)
    if not value:
        return ""
    if _has_http_scheme(value):
        return value
    if value.startswith("/"):
        if base_url:
            return f"{base_url.rstrip('/')}{value}"
        return ""
    return _ensure_http_scheme(value)


def _env_first(*names: str) -> str:
    for name in names:
        value = _clean(os.getenv(name))
        if value:
            return value
    return ""


# Keycloak configuration (look in the .env)
SECRET = _env_first("SECRET", "client_secret")
KEYCLOAK_AUDIENCE = _env_first("KEYCLOAK_AUDIENCE", "keycloak_audience")

_raw_keycloak_server = _env_first("KEYCLOAK_SERVER", "keycloak_server")
KEYCLOAK_SERVER = _ensure_http_scheme(_raw_keycloak_server)

_default_realm = _env_first("KEYCLOAK_REALM", "keycloak_realm") or "API.Server.local"

_raw_keycloak_issuer = _env_first("KEYCLOAK_ISSUER", "keycloak_issuer")
if _raw_keycloak_issuer and "${" not in _raw_keycloak_issuer:
    KEYCLOAK_ISSUER = _absolute_or_join(_raw_keycloak_issuer, KEYCLOAK_SERVER)
elif KEYCLOAK_SERVER:
    KEYCLOAK_ISSUER = f"{KEYCLOAK_SERVER.rstrip('/')}/realms/{_default_realm}"
else:
    KEYCLOAK_ISSUER = ""

_raw_keycloak_protocol = _env_first("KEYCLOAK_PROTOCOL_ENDPOINT", "keycloak_protocol_endpoint")
if _raw_keycloak_protocol and "${" not in _raw_keycloak_protocol:
    KEYCLOAK_PROTOCOL_ENDPOINT = _absolute_or_join(_raw_keycloak_protocol, KEYCLOAK_SERVER)
elif KEYCLOAK_ISSUER:
    KEYCLOAK_PROTOCOL_ENDPOINT = f"{KEYCLOAK_ISSUER.rstrip('/')}/protocol/openid-connect"
else:
    KEYCLOAK_PROTOCOL_ENDPOINT = ""

_raw_jwks = _env_first("KEYCLOAK_JWKS_URL", "keycloak_jwks_url")
if _raw_jwks and "${" not in _raw_jwks:
    KEYCLOAK_JWKS_URL = _absolute_or_join(_raw_jwks, KEYCLOAK_SERVER)
elif KEYCLOAK_PROTOCOL_ENDPOINT:
    KEYCLOAK_JWKS_URL = f"{KEYCLOAK_PROTOCOL_ENDPOINT.rstrip('/')}/certs"
else:
    KEYCLOAK_JWKS_URL = ""

_raw_auth = _env_first("KEYCLOAK_AUTH_URL", "keycloak_auth_url")
if _raw_auth and "${" not in _raw_auth:
    KEYCLOAK_AUTH_URL = _absolute_or_join(_raw_auth, KEYCLOAK_SERVER)
elif KEYCLOAK_PROTOCOL_ENDPOINT:
    KEYCLOAK_AUTH_URL = f"{KEYCLOAK_PROTOCOL_ENDPOINT.rstrip('/')}/auth"
else:
    KEYCLOAK_AUTH_URL = ""

_raw_token = _env_first("KEYCLOAK_TOKEN_URL", "keycloak_token_url")
if _raw_token and "${" not in _raw_token:
    KEYCLOAK_TOKEN_URL = _absolute_or_join(_raw_token, KEYCLOAK_SERVER)
elif KEYCLOAK_PROTOCOL_ENDPOINT:
    KEYCLOAK_TOKEN_URL = f"{KEYCLOAK_PROTOCOL_ENDPOINT.rstrip('/')}/token"
else:
    KEYCLOAK_TOKEN_URL = ""

CORE_API_URL = os.getenv("CORE_API_URL", "http://localhost:1902")

MQTT_HOST = os.getenv("MQTT_HOST", "192.168.1.101")
MQTT_PORT = int(os.getenv("MQTT_PORT", "1883"))
MQTT_TOPIC = os.getenv("MQTT_TOPIC", "#")
MQTT_VERSION = os.getenv("MQTT_VERSION", "mqttv311")
MQTT_STATUS_INTERVAL = int(os.getenv("MQTT_STATUS_INTERVAL", "30"))
MQTT_STALE_AFTER = int(os.getenv("MQTT_STALE_AFTER", "30"))

BLE_AI_INFER_CSV = os.getenv(
    "BLE_AI_INFER_CSV",
    "/data/service/ble-ai-localizer/data/infer/infer.csv",
)

BLE_AI_META_DIR = os.getenv(
    "BLE_AI_META_DIR",
    "/data/service/ble-ai-localizer/data/maps/",
)

BLE_AI_MAPS_DIR = os.getenv(
    "BLE_AI_MAPS_DIR",
    "/data/service/ble-ai-localizer/data/maps",
)