res_levis_API/config_env.py

import os
from dotenv import load_dotenv

load_dotenv()


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


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("/") and base_url:
        return f"{base_url.rstrip('/')}{value}"
    return _ensure_http_scheme(value)


# Keycloak configuration (look in the .env)
SECRET = os.getenv("SECRET")
KEYCLOAK_AUDIENCE = os.getenv("KEYCLOAK_AUDIENCE")

_raw_keycloak_server = _clean(os.getenv("KEYCLOAK_SERVER"))
KEYCLOAK_SERVER = _ensure_http_scheme(_raw_keycloak_server)

_default_realm = _clean(os.getenv("KEYCLOAK_REALM")) or "API.Server.local"

_raw_keycloak_issuer = _clean(os.getenv("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 = _clean(os.getenv("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 = _clean(os.getenv("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 = _clean(os.getenv("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 = _clean(os.getenv("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",
)