from __future__ import annotations import os import re from dataclasses import dataclass from functools import lru_cache from pathlib import Path from typing import Any from dotenv import load_dotenv BASE_DIR = Path(__file__).resolve().parents[1] load_dotenv(BASE_DIR / ".env") _ENV_PATTERN = re.compile(r"\$\{([A-Z0-9_]+)\}") def expand_env(value: Any) -> Any: if not isinstance(value, str): return value def _replace(match: re.Match[str]) -> str: name = match.group(1) return os.getenv(name, match.group(0)) return _ENV_PATTERN.sub(_replace, value) def parse_csv_env(value: str) -> tuple[str, ...]: return tuple(item.strip() for item in value.split(",") if item.strip()) @dataclass(frozen=True) class Settings: app_name: str dashboard_token: str default_operator: str public_base_url: str provider_name: str anthropic_api_key: str anthropic_model: str ollama_base_url: str ollama_model: str chatbot_bridge_url: str chatbot_admin_token: str chatbot_timeout_seconds: float ssh_host: str ssh_port: int ssh_user: str ssh_key_path: str ssh_use_agent: bool target_repo_root: str allowed_commands_file: Path default_ssh_profile: str auto_execute_low_risk: bool telegram_enabled: bool telegram_bot_token: str telegram_mode: str telegram_poll_interval: float telegram_allowed_chat_ids: tuple[str, ...] telegram_webhook_secret: str telegram_webhook_url: str telegram_api_base_url: str db_path: Path # OpenTelemetry otel_enabled: bool otel_service_name: str otel_otlp_endpoint: str # Langfuse — observabilidad LLM langfuse_enabled: bool langfuse_public_key: str langfuse_secret_key: str langfuse_host: str # Proyecto gestionado (configurable para reutilizar el supervisor en cualquier proyecto) project_name: str project_description: str project_domain: str # Qdrant — vector DB para RAG qdrant_url: str qdrant_api_key: str rag_enabled: bool rag_collection: str rag_embed_model: str # WhatsApp Business API whatsapp_enabled: bool whatsapp_phone_number_id: str whatsapp_access_token: str whatsapp_webhook_verify_token: str whatsapp_app_secret: str whatsapp_allowed_phones: tuple[str, ...] whatsapp_api_version: str def public_summary(self) -> dict[str, Any]: telegram_active = self.telegram_enabled and bool(self.telegram_bot_token) return { "app_name": self.app_name, "project_name": self.project_name, "project_domain": self.project_domain, "provider_name": self.provider_name, "public_base_url": self.public_base_url, "chatbot_bridge_enabled": bool(self.chatbot_bridge_url and self.chatbot_admin_token), "ssh_host": self.ssh_host, "ssh_port": self.ssh_port, "ssh_user": self.ssh_user, "ssh_use_agent": self.ssh_use_agent, "default_ssh_profile": self.default_ssh_profile, "auto_execute_low_risk": self.auto_execute_low_risk, "telegram_enabled": telegram_active, "telegram_mode": self.telegram_mode if telegram_active else "disabled", "telegram_allowed_chats": len(self.telegram_allowed_chat_ids), "telegram_webhook_configured": bool(self.telegram_webhook_url), "langfuse_enabled": self.langfuse_enabled, "langfuse_host": self.langfuse_host, "langfuse_keys_configured": bool(self.langfuse_public_key and self.langfuse_secret_key), } @lru_cache(maxsize=1) def get_settings() -> Settings: db_default = BASE_DIR / "data" / "supervisor.sqlite3" commands_default = BASE_DIR / "config" / "allowed_commands.example.json" telegram_mode = os.getenv("TELEGRAM_MODE", "polling").strip().lower() if telegram_mode not in {"polling", "webhook"}: telegram_mode = "polling" allowed_commands = Path(expand_env(os.getenv("ALLOWED_COMMANDS_FILE", str(commands_default)))) if not allowed_commands.is_absolute(): allowed_commands = (BASE_DIR / allowed_commands).resolve() db_path = Path(expand_env(os.getenv("SUPERVISOR_DB_PATH", str(db_default)))) if not db_path.is_absolute(): db_path = (BASE_DIR / db_path).resolve() return Settings( app_name=os.getenv("SUPERVISOR_APP_NAME", "XZonas Multiagent Supervisor"), dashboard_token=os.getenv("SUPERVISOR_DASHBOARD_TOKEN", "change-me"), default_operator=os.getenv("DEFAULT_OPERATOR", "propietario"), public_base_url=os.getenv("PUBLIC_BASE_URL", "http://127.0.0.1:8787"), provider_name=os.getenv("LLM_PROVIDER", "heuristic").strip().lower(), anthropic_api_key=os.getenv("ANTHROPIC_API_KEY", "").strip(), anthropic_model=os.getenv("ANTHROPIC_MODEL", "claude-sonnet-4-5").strip(), ollama_base_url=os.getenv("OLLAMA_BASE_URL", "http://127.0.0.1:11434").rstrip("/"), ollama_model=os.getenv("OLLAMA_MODEL", "qwen2.5-coder:14b").strip(), chatbot_bridge_url=os.getenv("CHATBOT_BRIDGE_URL", "").strip(), chatbot_admin_token=os.getenv("CHATBOT_ADMIN_TOKEN", "").strip(), chatbot_timeout_seconds=float(os.getenv("CHATBOT_TIMEOUT_SECONDS", "20")), ssh_host=os.getenv("SSH_HOST", "").strip(), ssh_port=int(os.getenv("SSH_PORT", "22")), ssh_user=os.getenv("SSH_USER", "").strip(), ssh_key_path=os.getenv("SSH_KEY_PATH", "").strip(), ssh_use_agent=os.getenv("SSH_USE_AGENT", "false").strip().lower() in {"1", "true", "yes", "si", "on"}, target_repo_root=os.getenv("TARGET_REPO_ROOT", "").strip(), allowed_commands_file=allowed_commands, default_ssh_profile=os.getenv("DEFAULT_SSH_PROFILE", "siteground-prod").strip(), auto_execute_low_risk=os.getenv("AUTO_EXECUTE_LOW_RISK", "false").strip().lower() in {"1", "true", "yes", "si", "on"}, telegram_enabled=os.getenv("TELEGRAM_ENABLED", "false").strip().lower() in {"1", "true", "yes", "si", "on"}, telegram_bot_token=os.getenv("TELEGRAM_BOT_TOKEN", "").strip(), telegram_mode=telegram_mode, telegram_poll_interval=float(os.getenv("TELEGRAM_POLL_INTERVAL", "2.5")), telegram_allowed_chat_ids=parse_csv_env(os.getenv("TELEGRAM_ALLOWED_CHAT_IDS", "")), telegram_webhook_secret=os.getenv("TELEGRAM_WEBHOOK_SECRET", "change-me-telegram-webhook").strip(), telegram_webhook_url=os.getenv("TELEGRAM_WEBHOOK_URL", "").strip(), telegram_api_base_url=os.getenv("TELEGRAM_API_BASE_URL", "https://api.telegram.org").rstrip("/"), db_path=db_path, otel_enabled=os.getenv("OTEL_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"}, otel_service_name=os.getenv("OTEL_SERVICE_NAME", "xzonas-supervisor").strip(), otel_otlp_endpoint=os.getenv("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317").strip(), langfuse_enabled=os.getenv("LANGFUSE_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"}, langfuse_public_key=os.getenv("LANGFUSE_PUBLIC_KEY", "").strip(), langfuse_secret_key=os.getenv("LANGFUSE_SECRET_KEY", "").strip(), langfuse_host=os.getenv("LANGFUSE_HOST", "http://localhost:3100").strip(), project_name=os.getenv("PROJECT_NAME", "xzonasvalencia.com").strip(), project_description=os.getenv("PROJECT_DESCRIPTION", "portal de anuncios adultos en España").strip(), project_domain=os.getenv("PROJECT_DOMAIN", "xzonasvalencia.com").strip(), whatsapp_enabled=os.getenv("WHATSAPP_ENABLED", "false").strip().lower() in {"1", "true", "yes", "si", "on"}, whatsapp_phone_number_id=os.getenv("WHATSAPP_PHONE_NUMBER_ID", os.getenv("WHATSAPP_META_PHONE_ID", "")).strip(), whatsapp_access_token=os.getenv("WHATSAPP_ACCESS_TOKEN", os.getenv("WHATSAPP_META_TOKEN", "")).strip(), whatsapp_webhook_verify_token=os.getenv("WHATSAPP_WEBHOOK_VERIFY_TOKEN", os.getenv("WHATSAPP_VERIFY_TOKEN", "change-me-wa-verify")).strip(), whatsapp_app_secret=os.getenv("WHATSAPP_APP_SECRET", "").strip(), whatsapp_allowed_phones=parse_csv_env(os.getenv("WHATSAPP_ALLOWED_PHONES", "")), whatsapp_api_version=os.getenv("WHATSAPP_API_VERSION", "v20.0").strip(), qdrant_url=os.getenv("QDRANT_URL", "").strip(), qdrant_api_key=os.getenv("QDRANT_API_KEY", "").strip(), rag_enabled=os.getenv("RAG_ENABLED", "false").strip().lower() in {"1", "true", "yes", "on"}, rag_collection=os.getenv("RAG_COLLECTION", "xzonas_docs").strip(), rag_embed_model=os.getenv("RAG_EMBED_MODEL", "BAAI/bge-small-en-v1.5").strip(), )