99 lines
3.7 KiB
Python
99 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List
|
|
|
|
from core.settings import AGENT_MD_TEMPLATES_FILE, TOPIC_PRESETS_TEMPLATES_FILE
|
|
|
|
TEMPLATE_KEYS = ("agents_md", "soul_md", "user_md", "tools_md", "identity_md")
|
|
|
|
|
|
def _load_json_object(path: Path, *, label: str) -> Dict[str, Any]:
|
|
import json
|
|
|
|
target = Path(path).resolve()
|
|
if not target.is_file():
|
|
raise RuntimeError(
|
|
f"Missing required {label} file: {target}. "
|
|
"Please restore the tracked files under data/templates before starting the backend."
|
|
)
|
|
try:
|
|
with target.open("r", encoding="utf-8") as file:
|
|
data = json.load(file)
|
|
except Exception as exc:
|
|
raise RuntimeError(f"Invalid JSON in {label} file: {target}") from exc
|
|
if not isinstance(data, dict):
|
|
raise RuntimeError(f"{label} file must contain a JSON object: {target}")
|
|
return data
|
|
|
|
|
|
def _normalize_md_text(value: Any) -> str:
|
|
return str(value or "").replace("\r\n", "\n").strip()
|
|
|
|
|
|
def _write_json_atomic(path: str, payload: Dict[str, Any]) -> None:
|
|
import json
|
|
import os
|
|
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
tmp_path = f"{path}.tmp"
|
|
with open(tmp_path, "w", encoding="utf-8") as file:
|
|
json.dump(payload, file, ensure_ascii=False, indent=2)
|
|
os.replace(tmp_path, path)
|
|
|
|
|
|
def get_agent_md_templates() -> Dict[str, str]:
|
|
raw = _load_json_object(AGENT_MD_TEMPLATES_FILE, label="agent templates")
|
|
missing_keys = [key for key in TEMPLATE_KEYS if key not in raw]
|
|
if missing_keys:
|
|
raise RuntimeError(
|
|
"Agent template file is missing required keys: "
|
|
f"{', '.join(missing_keys)}. File: {Path(AGENT_MD_TEMPLATES_FILE).resolve()}"
|
|
)
|
|
return {key: _normalize_md_text(raw.get(key)) for key in TEMPLATE_KEYS}
|
|
|
|
|
|
def get_topic_presets() -> Dict[str, Any]:
|
|
raw = _load_json_object(TOPIC_PRESETS_TEMPLATES_FILE, label="topic presets")
|
|
presets = raw.get("presets")
|
|
if not isinstance(presets, list):
|
|
raise RuntimeError(
|
|
f"Topic presets file must contain a presets array: {Path(TOPIC_PRESETS_TEMPLATES_FILE).resolve()}"
|
|
)
|
|
invalid_rows = [index for index, row in enumerate(presets) if not isinstance(row, dict)]
|
|
if invalid_rows:
|
|
raise RuntimeError(
|
|
"Topic presets file contains non-object entries at indexes: "
|
|
f"{', '.join(str(index) for index in invalid_rows)}. "
|
|
f"File: {Path(TOPIC_PRESETS_TEMPLATES_FILE).resolve()}"
|
|
)
|
|
return {"presets": [dict(row) for row in presets]}
|
|
|
|
|
|
def update_agent_md_templates(raw: Dict[str, Any]) -> Dict[str, str]:
|
|
payload = {key: _normalize_md_text(raw.get(key)) for key in TEMPLATE_KEYS}
|
|
_write_json_atomic(str(AGENT_MD_TEMPLATES_FILE), payload)
|
|
return payload
|
|
|
|
|
|
def update_topic_presets(raw: Dict[str, Any]) -> Dict[str, Any]:
|
|
presets = raw.get("presets") if isinstance(raw, dict) else None
|
|
if not isinstance(presets, list):
|
|
raise ValueError("topic_presets.presets must be an array")
|
|
invalid_rows = [index for index, row in enumerate(presets) if not isinstance(row, dict)]
|
|
if invalid_rows:
|
|
raise ValueError(
|
|
"topic_presets.presets must contain objects only; invalid indexes: "
|
|
+ ", ".join(str(index) for index in invalid_rows)
|
|
)
|
|
payload: Dict[str, List[Dict[str, Any]]] = {"presets": [dict(row) for row in presets]}
|
|
_write_json_atomic(str(TOPIC_PRESETS_TEMPLATES_FILE), payload)
|
|
return payload
|
|
|
|
|
|
def get_agent_template_value(key: str) -> str:
|
|
templates = get_agent_md_templates()
|
|
if key not in templates:
|
|
raise KeyError(f"Unknown agent template key: {key}")
|
|
return templates[key]
|