's'

src/live_config/__init__.py

@@ -0,0 +1,8 @@
+from .loader import business_config, get_config_root, load_hub_config_snapshot, services_config_list
+
+__all__ = [
+    "get_config_root",
+    "load_hub_config_snapshot",
+    "services_config_list",
+    "business_config",
+]

src/live_config/loader.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Any
+
+
+def get_config_root() -> Path:
+    env = os.environ.get("LIVE_CONFIG_ROOT", "").strip()
+    if env:
+        return Path(env)
+    if Path("/opt/live/config/system.json").is_file():
+        return Path("/opt/live/config")
+    # 仓库内开发默认值
+    base = Path(__file__).resolve().parent.parent.parent
+    return base / "live-platform" / "config" / "defaults"
+
+
+def _read_json(path: Path) -> dict[str, Any]:
+    if not path.is_file():
+        return {}
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError:
+        return {}
+
+
+def services_config_list() -> list[dict[str, Any]]:
+    root = get_config_root() / "services"
+    if not root.is_dir():
+        return []
+    out: list[dict[str, Any]] = []
+    for f in sorted(root.glob("*.json")):
+        data = _read_json(f)
+        if data:
+            data.setdefault("id", f.stem)
+            out.append(data)
+    return out
+
+
+def business_config(module_id: str) -> dict[str, Any]:
+    return _read_json(get_config_root() / "business" / f"{module_id}.json")
+
+
+def load_hub_config_snapshot() -> dict[str, Any]:
+    root = get_config_root()
+    return {
+        "config_root": str(root),
+        "system": _read_json(root / "system.json"),
+        "network": _read_json(root / "network.json"),
+        "services": services_config_list(),
+        "business": {
+            "douyinyoutube": business_config("douyinyoutube"),
+        },
+    }