dekun
b733e551a0
Add scripts/normalize_ambiguous_unicode.py; fix corrupted patch_instance_theme_templates.py. Preserves curly quotes in string literals; removes Git homoglyph warnings on .env.example. Co-authored-by: Cursor <cursoragent@cursor.com>
63 lines
1.9 KiB
Python
63 lines
1.9 KiB
Python
"""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 {}
|
|
if meta.get("hot_reload") and not meta.get("restart_required"):
|
|
val = os.getenv(key)
|
|
if val is not None:
|
|
hot[key] = val
|
|
if hot:
|
|
set_config_overrides(get_db, hot)
|
|
from lib.env.env_schema import updates_need_restart
|
|
|
|
return {"restart_required": updates_need_restart(groups, changed_keys)}
|