""" Shared configuration for AgentsMeeting bots. All secrets via environment variables. No hardcoded keys. Usage: from src.shared.config import get_bot_config cfg = get_bot_config("xxm") """ import os, json, yaml from typing import Optional # Paths PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) CONFIG_DIR = os.path.join(PROJECT_ROOT, "config", "profiles") def required_env(name: str) -> str: """Get required env var, fail fast if missing.""" v = os.environ.get(name, "") if not v: raise RuntimeError(f"Missing required env var: {name}") return v def optional_env(name: str, default: str = "") -> str: """Get optional env var with fallback.""" return os.environ.get(name, default) class BotConfig: """Single bot's configuration.""" def __init__(self, profile: str): self.profile = profile # Load from config.yaml if exists yaml_path = os.path.join(CONFIG_DIR, profile, "config.yaml") yaml_cfg = {} if os.path.exists(yaml_path): try: with open(yaml_path, "r", encoding="utf-8") as f: raw = yaml.safe_load(f) if raw is not None: yaml_cfg = raw except Exception: pass # Provider configs (env var overrides file config) self.providers = { "volcengine": { "api_key": os.environ.get("VOLCENGINE_KEY") or _nested_get(yaml_cfg, "providers.volcengine.api_key", ""), "base_url": "https://ark.cn-beijing.volces.com/api/coding/v3", }, "ocg_new": { "api_key": os.environ.get("OCG_NEW_KEY") or _nested_get(yaml_cfg, "providers.ocg-new.api_key", ""), "base_url": "https://opencode.ai/zen/go/v1", }, "ocg_old": { "api_key": os.environ.get("OCG_OLD_KEY") or _nested_get(yaml_cfg, "providers.ocg-old.api_key", ""), "base_url": "https://opencode.ai/zen/go/v1", }, } # XMPP config self.jid = os.environ.get(f"{profile.upper()}_JID") or _nested_get(yaml_cfg, "xmpp.jid", f"{profile}@yoin.fun") self.password = os.environ.get(f"{profile.upper()}_PASS") or _nested_get(yaml_cfg, "xmpp.password", "") self.xmpp_host = os.environ.get("XMPP_HOST", "xmpp.yoin.fun") self.xmpp_port = int(os.environ.get("XMPP_PORT", "3021")) self.muc_rooms = (os.environ.get("MUC_ROOMS", "coregroup@conference.yoin.fun")).split(",") # Session config self.session_id = os.environ.get(f"{profile.upper()}_SESSION") or _nested_get(yaml_cfg, "session.id", f"ses_{profile}") # Model config self.model = os.environ.get("DEFAULT_MODEL", "deepseek-v4-flash") self.provider = os.environ.get("DEFAULT_PROVIDER", "volcengine") # API config self.api_timeout = int(os.environ.get("API_TIMEOUT", "60")) self.max_tool_loops = int(os.environ.get("MAX_TOOL_LOOPS", "30")) def _nested_get(d: dict, path: str, default=""): """Get nested dict value by dot-separated path.""" parts = path.split(".") for p in parts: if isinstance(d, dict) and p in d: d = d[p] else: return default return d def get_bot_config(profile: str) -> BotConfig: """Factory: load config for a bot profile.""" return BotConfig(profile)