crypto_okx/lib/instance/runtime_config_lib.py

"""env 运行时覆盖:热生效项优先读 SQLite,再回退 os.environ."""
from __future__ import annotations

import os
from typing import Callable, Optional

from lib.env.env_file_lib import load_env_file_into_environ
from lib.instance.runtime_settings_lib import runtime_get, with_db

ENV_OVERRIDE_PREFIX = "env."


def runtime_env_key(name: str) -> str:
    return ENV_OVERRIDE_PREFIX + name


def get_config(key: str, get_db: Callable, default: Optional[str] = None) -> Optional[str]:
    def _read(conn):
        v = runtime_get(conn, runtime_env_key(key))
        return v

    try:
        v = with_db(get_db, _read)
        if v is not None:
            return v
    except Exception:
        pass
    raw = os.getenv(key)
    if raw is None or raw == "":
        return default
    return raw


def set_config_overrides(get_db: Callable, mapping: dict[str, str]) -> None:
    from lib.instance.runtime_settings_lib import runtime_set_many

    def _write(conn):
        payload = {runtime_env_key(k): str(v) for k, v in mapping.items()}
        runtime_set_many(conn, payload)

    with_db(get_db, _write)


def apply_env_reload(env_path: str, get_db: Callable, changed_keys: list[str], groups: list[dict]) -> dict[str, bool]:
    """写盘后同步 os.environ,并将可热生效项写入 runtime 覆盖."""
    load_env_file_into_environ(env_path)
    hot: dict[str, str] = {}
    field_map = {}
    for group in groups:
        for field in group.get("fields") or []:
            field_map[field["key"]] = field
    for key in changed_keys:
        meta = field_map.get(key) or {}
        # SIM_DEFAULT_MODE 不在 example schema 的 hot 列表时仍应热切
        is_hot = bool(meta.get("hot_reload") and not meta.get("restart_required"))
        if key == "SIM_DEFAULT_MODE":
            is_hot = True
        if is_hot:
            val = os.getenv(key)
            if val is not None:
                hot[key] = val
    if hot:
        set_config_overrides(get_db, hot)
    # 撮合模式:写入 .env 的同时切换运行时 trading.mode(sim|live)
    if "SIM_DEFAULT_MODE" in changed_keys:
        try:
            from lib.sim.mode_lib import set_trading_mode

            raw = (os.getenv("SIM_DEFAULT_MODE") or "").strip().lower()
            if raw in ("sim", "live"):
                set_trading_mode(get_db, raw)
        except Exception:
            pass
    from lib.env.env_schema import updates_need_restart

    return {"restart_required": updates_need_restart(groups, changed_keys)}