import json import os from typing import Any, Callable, Dict from fastapi import HTTPException from sqlmodel import Session, select from models.bot import BotInstance class SystemService: def __init__( self, *, engine: Any, cache: Any, database_engine: str, redis_enabled: bool, redis_url: str, redis_prefix: str, agent_md_templates_file: str, topic_presets_templates_file: str, default_soul_md: str, default_agents_md: str, default_user_md: str, default_tools_md: str, default_identity_md: str, topic_preset_templates: Any, get_default_system_timezone: Callable[[], str], load_agent_md_templates: Callable[[], Dict[str, Any]], load_topic_presets_template: Callable[[], Dict[str, Any]], get_platform_settings_snapshot: Callable[[], Any], get_speech_runtime_settings: Callable[[], Dict[str, Any]], ) -> None: self._engine = engine self._cache = cache self._database_engine = database_engine self._redis_enabled = redis_enabled self._redis_url = redis_url self._redis_prefix = redis_prefix self._agent_md_templates_file = agent_md_templates_file self._topic_presets_templates_file = topic_presets_templates_file self._default_soul_md = default_soul_md self._default_agents_md = default_agents_md self._default_user_md = default_user_md self._default_tools_md = default_tools_md self._default_identity_md = default_identity_md self._topic_preset_templates = topic_preset_templates self._get_default_system_timezone = get_default_system_timezone self._load_agent_md_templates = load_agent_md_templates self._load_topic_presets_template = load_topic_presets_template self._get_platform_settings_snapshot = get_platform_settings_snapshot self._get_speech_runtime_settings = get_speech_runtime_settings @staticmethod def _write_json_atomic(path: str, payload: Dict[str, Any]) -> None: os.makedirs(os.path.dirname(path), exist_ok=True) tmp = f"{path}.tmp" with open(tmp, "w", encoding="utf-8") as file: json.dump(payload, file, ensure_ascii=False, indent=2) os.replace(tmp, path) def get_system_defaults(self) -> Dict[str, Any]: md_templates = self._load_agent_md_templates() topic_presets = self._load_topic_presets_template() platform_settings = self._get_platform_settings_snapshot() speech_settings = self._get_speech_runtime_settings() return { "templates": { "soul_md": md_templates.get("soul_md") or self._default_soul_md, "agents_md": md_templates.get("agents_md") or self._default_agents_md, "user_md": md_templates.get("user_md") or self._default_user_md, "tools_md": md_templates.get("tools_md") or self._default_tools_md, "identity_md": md_templates.get("identity_md") or self._default_identity_md, }, "limits": { "upload_max_mb": platform_settings.upload_max_mb, }, "workspace": { "download_extensions": list(platform_settings.workspace_download_extensions), "allowed_attachment_extensions": list(platform_settings.allowed_attachment_extensions), }, "bot": { "system_timezone": self._get_default_system_timezone(), }, "loading_page": platform_settings.loading_page.model_dump(), "chat": { "pull_page_size": platform_settings.chat_pull_page_size, "page_size": platform_settings.page_size, "command_auto_unlock_seconds": platform_settings.command_auto_unlock_seconds, }, "topic_presets": topic_presets.get("presets") or self._topic_preset_templates, "speech": { "enabled": speech_settings["enabled"], "model": speech_settings["model"], "device": speech_settings["device"], "max_audio_seconds": speech_settings["max_audio_seconds"], "default_language": speech_settings["default_language"], }, } def get_system_templates(self) -> Dict[str, Any]: return { "agent_md_templates": self._load_agent_md_templates(), "topic_presets": self._load_topic_presets_template(), } def update_system_templates(self, *, payload: Any) -> Dict[str, Any]: if payload.agent_md_templates is not None: sanitized_agent: Dict[str, str] = {} for key in ("agents_md", "soul_md", "user_md", "tools_md", "identity_md"): sanitized_agent[key] = str(payload.agent_md_templates.get(key, "") or "").replace("\r\n", "\n") self._write_json_atomic(str(self._agent_md_templates_file), sanitized_agent) if payload.topic_presets is not None: presets = payload.topic_presets.get("presets") if isinstance(payload.topic_presets, dict) else None if presets is None: normalized_topic: Dict[str, Any] = {"presets": []} elif isinstance(presets, list): normalized_topic = {"presets": [dict(row) for row in presets if isinstance(row, dict)]} else: raise HTTPException(status_code=400, detail="topic_presets.presets must be an array") self._write_json_atomic(str(self._topic_presets_templates_file), normalized_topic) return { "status": "ok", "agent_md_templates": self._load_agent_md_templates(), "topic_presets": self._load_topic_presets_template(), } def get_health(self) -> Dict[str, Any]: try: with Session(self._engine) as session: session.exec(select(BotInstance).limit(1)).first() return {"status": "ok", "database": self._database_engine} except Exception as exc: raise HTTPException(status_code=503, detail=f"database check failed: {exc}") def get_cache_health(self) -> Dict[str, Any]: redis_url = str(self._redis_url or "").strip() configured = bool(self._redis_enabled and redis_url) client_enabled = bool(getattr(self._cache, "enabled", False)) reachable = bool(self._cache.ping()) if client_enabled else False status = "ok" if configured and not reachable: status = "degraded" return { "status": status, "cache": { "configured": configured, "enabled": client_enabled, "reachable": reachable, "prefix": self._redis_prefix, }, }