# -*- coding: utf-8 -*- """环境变量解析""" import os import json from copy import deepcopy ENV_MAP = { "TIMEZONE": ("app.timezone",), "STORE_ID": ("app.store_id",), "SCHEMA_OLTP": ("app.schema_oltp",), "SCHEMA_ETL": ("app.schema_etl",), "PG_DSN": ("db.dsn",), "PG_HOST": ("db.host",), "PG_PORT": ("db.port",), "PG_NAME": ("db.name",), "PG_USER": ("db.user",), "PG_PASSWORD": ("db.password",), "API_BASE": ("api.base_url",), "API_TOKEN": ("api.token",), "API_TIMEOUT": ("api.timeout_sec",), "API_PAGE_SIZE": ("api.page_size",), "EXPORT_ROOT": ("io.export_root",), "LOG_ROOT": ("io.log_root",), "OVERLAP_SECONDS": ("run.overlap_seconds",), "WINDOW_BUSY_MIN": ("run.window_minutes.default_busy",), "WINDOW_IDLE_MIN": ("run.window_minutes.default_idle",), } def _deep_set(d, dotted_keys, value): cur = d for k in dotted_keys[:-1]: cur = cur.setdefault(k, {}) cur[dotted_keys[-1]] = value def _coerce_env(v: str): if v is None: return None s = v.strip() if s.lower() in ("true", "false"): return s.lower() == "true" try: if s.isdigit() or (s.startswith("-") and s[1:].isdigit()): return int(s) except Exception: pass if (s.startswith("{") and s.endswith("}")) or (s.startswith("[") and s.endswith("]")): try: return json.loads(s) except Exception: return s return s def load_env_overrides(defaults: dict) -> dict: cfg = deepcopy(defaults) for env_key, dotted in ENV_MAP.items(): val = os.environ.get(env_key) if val is None: continue v2 = _coerce_env(val) for path in dotted: _deep_set(cfg, path.split("."), v2) return cfg