from __future__ import annotations import asyncio import logging import os import tempfile from datetime import datetime, timezone from typing import Any import httpx from .config import Settings from .orchestrator import MultiAgentOrchestrator from .storage import SupervisorStore logger = logging.getLogger(__name__) TELEGRAM_QUICK_COMMANDS: dict[str, dict[str, str]] = { "git_status": {"text": "estado git", "label": "estado del despliegue"}, "tail_php_errorlog": {"text": "errores php", "label": "últimos errores PHP"}, "composer_lint": {"text": "lint php", "label": "lint PHP"}, "phpstan": {"text": "phpstan", "label": "PHPStan"}, "psalm": {"text": "psalm", "label": "Psalm"}, "npm_minify": {"text": "minificar assets", "label": "minify custom"}, "npm_build_grid": {"text": "grid-core", "label": "build grid-core"}, "cache_flush": {"text": "vaciar cache", "label": "vaciar OPcache"}, } TELEGRAM_MARKETING_PRESETS: dict[str, str] = { "reactivacion": "Reactiva anunciantes inactivos con consentimiento WhatsApp, prioriza Valencia centro y municipios cercanos, y prepara segmentación + copy + riesgos.", "captacion": "Diseña campaña de captación de anunciantes premium en Valencia, con secuencia por WhatsApp, Telegram y email, priorizando respuesta rápida y objeciones frecuentes.", "copy": "Prepara tres variantes de copy comercial para WhatsApp orientadas a reactivación de anunciantes fríos, con tono directo, elegante y medible.", "oferta": "Propón una campaña test A/B con oferta de reactivación para anunciantes dormidos, incluyendo segmentación, métricas, riesgos y siguiente experimento.", } AUTOMATION_LABELS: dict[str, str] = { "ops_daily": "Ops diario", "marketing_daily": "Marketing diario", "seo_weekly": "SEO semanal", "stack_health": "Salud stack", "reactivacion_anunciantes": "Reactivación CRM", "seguridad_semanal": "Seguridad semanal", "git_code_analysis": "CodeAgent git", "whatsapp_reactivacion": "Campaña WA reactivación", } class TelegramBotService: def __init__(self, settings: Settings, store: SupervisorStore, orchestrator: MultiAgentOrchestrator) -> None: self.settings = settings self.store = store self.orchestrator = orchestrator self._poller_task: asyncio.Task[None] | None = None self._client: httpx.AsyncClient | None = None self._scheduler: Any | None = None # Callable registrado en main.py al startup para enviar campañas WA supervisor self._wa_campaign_sender: Any = None @property def enabled(self) -> bool: return self.settings.telegram_enabled and bool(self.settings.telegram_bot_token) @property def allowed_chat_ids(self) -> set[str]: return set(self.settings.telegram_allowed_chat_ids) async def start(self) -> None: if not self.enabled: logger.info("Telegram deshabilitado o sin token; no se inicia el bot") return if self.settings.telegram_mode == "polling": await self._delete_webhook() if self._poller_task is None or self._poller_task.done(): self._poller_task = asyncio.create_task(self._polling_loop(), name="telegram-poller") logger.info("Telegram polling iniciado") return if self.settings.telegram_mode == "webhook": await self._ensure_webhook() logger.info("Telegram webhook preparado") async def stop(self) -> None: if self._poller_task is not None: self._poller_task.cancel() try: await self._poller_task except asyncio.CancelledError: pass self._poller_task = None if self._client is not None: await self._client.aclose() self._client = None # ------------------------------------------------------------------ # # Métodos públicos para notificaciones programadas (scheduler/agentes) # # ------------------------------------------------------------------ # def attach_scheduler(self, scheduler: Any) -> None: """Referencia opcional al scheduler para consultas de estado en /digest.""" self._scheduler = scheduler def set_wa_campaign_sender(self, sender: Any) -> None: """Registra el callable async que ejecuta el envío real de una campaña WA supervisor. Firma esperada: async (proposal_id: str) -> dict[str, Any] Se inyecta desde main.py en startup para evitar acoplamiento circular. """ self._wa_campaign_sender = sender async def broadcast_to_admins(self, text: str, proposals: list[dict[str, Any]]) -> None: """Envía un mensaje a todos los chats autorizados con botones inline opcionales.""" if not self.enabled or not self.allowed_chat_ids: return keyboard: dict[str, Any] | None = None if proposals: rows = [] for proposal in proposals[:4]: pid = proposal.get("id", "") title = str(proposal.get("title") or "Propuesta")[:28] rows.append([ {"text": f"✅ {title}", "callback_data": f"approve:{pid}"}, {"text": "❌", "callback_data": f"reject:{pid}"}, ]) if self.settings.public_base_url: rows.append([{"text": "🖥️ Dashboard", "url": self.settings.public_base_url}]) keyboard = {"inline_keyboard": rows} for chat_id in self.allowed_chat_ids: try: await self._send_message(chat_id, text, reply_markup=keyboard) except Exception as exc: logger.warning( "broadcast_to_admins: fallo enviando a %s: %s", chat_id, self._sanitize_telegram_error(str(exc)), ) async def handle_update(self, update: dict[str, Any]) -> None: if not self.enabled: return await self._process_update(update) async def _get_client(self) -> httpx.AsyncClient: if self._client is None: self._client = httpx.AsyncClient(timeout=30) return self._client def _api_url(self, method: str) -> str: return f"{self.settings.telegram_api_base_url}/bot{self.settings.telegram_bot_token}/{method}" async def _api_call(self, method: str, payload: dict[str, Any] | None = None) -> Any: client = await self._get_client() try: response = await client.post(self._api_url(method), json=payload or {}) except httpx.HTTPError as exc: raise RuntimeError(self._sanitize_telegram_error(str(exc))) from exc try: data = response.json() except ValueError as exc: raise RuntimeError(f"Telegram API devolvió una respuesta no JSON en {method}") from exc if response.status_code >= 400: description = self._telegram_error_description(data) or response.text.strip() or f"HTTP {response.status_code}" raise RuntimeError(self._sanitize_telegram_error(f"Telegram API error en {method}: {description}")) if not isinstance(data, dict): raise RuntimeError(f"Telegram API devolvió un formato inesperado en {method}") if not data.get("ok"): description = self._telegram_error_description(data) or f"Telegram API devolvió error en {method}" raise RuntimeError(self._sanitize_telegram_error(description)) return data.get("result") async def _delete_webhook(self) -> None: try: # drop_pending_updates=True descarta callbacks/mensajes acumulados mientras el # bot estaba parado, evitando el spam de "query is too old" al arrancar. await self._api_call("deleteWebhook", {"drop_pending_updates": True}) except Exception as exc: logger.warning("No se pudo eliminar webhook de Telegram antes del polling: %s", self._sanitize_telegram_error(str(exc))) async def _ensure_webhook(self) -> None: if not self.settings.telegram_webhook_url: logger.warning("TELEGRAM_MODE=webhook pero TELEGRAM_WEBHOOK_URL está vacío") return payload = { "url": self.settings.telegram_webhook_url, "allowed_updates": ["message", "callback_query"], "drop_pending_updates": False, } try: await self._api_call("setWebhook", payload) except Exception as exc: logger.warning("No se pudo registrar el webhook de Telegram: %s", self._sanitize_telegram_error(str(exc))) async def _polling_loop(self) -> None: offset = int(self.store.get_runtime_state("telegram.offset", 0) or 0) _conflict_backoff = 15.0 # segundos de espera iniciales ante Conflict _conflict_count = 0 while True: try: result = await self._api_call( "getUpdates", { "timeout": 20, "offset": offset, "allowed_updates": ["message", "callback_query", "voice"], }, ) _conflict_count = 0 # reset backoff tras éxito updates = result if isinstance(result, list) else [] for update in updates: await self._process_update(update) offset = max(offset, int(update.get("update_id", 0)) + 1) self.store.set_runtime_state("telegram.offset", offset) if not updates: await asyncio.sleep(self.settings.telegram_poll_interval) except asyncio.CancelledError: raise except Exception as exc: err_str = self._sanitize_telegram_error(str(exc)) if "Conflict" in err_str: _conflict_count += 1 # Backoff exponencial: 15s → 30s → 60s (máx), para no inundar los logs wait = min(_conflict_backoff * (2 ** (_conflict_count - 1)), 60.0) logger.error( "Telegram Conflict: otra instancia del bot está activa (intento #%d). " "Detén el proceso local (uvicorn fuera de Docker) para resolver. " "Reintentando en %.0fs.", _conflict_count, wait, ) await asyncio.sleep(wait) else: logger.warning("Error en polling de Telegram: %s", err_str) await asyncio.sleep(max(self.settings.telegram_poll_interval, 2.0)) async def _process_update(self, update: dict[str, Any]) -> None: if "callback_query" in update: await self._process_callback_query(update["callback_query"]) return if "message" in update: await self._process_message(update["message"]) async def _process_message(self, message: dict[str, Any]) -> None: chat = message.get("chat") or {} chat_id = str(chat.get("id", "")).strip() text = str(message.get("text") or "").strip() actor = self._actor_from_message(message) if not chat_id: return # Mensaje de voz: transcribir con faster-whisper voice = message.get("voice") if voice and not text: if self._is_chat_allowed(chat_id): await self._handle_voice_message(chat_id, voice, actor) return if not text: return if text.startswith("/whoami"): await self._handle_whoami(chat_id) return if text.startswith("/start") or text.startswith("/menu"): await self._show_main_menu(chat_id, actor) return if not self._is_chat_allowed(chat_id): await self._send_message(chat_id, self._unauthorized_text(chat_id)) return self.store.add_audit("telegram.message.received", {"chat_id": chat_id, "actor": actor, "text": text[:500]}) if text.startswith("/help"): await self._send_message(chat_id, self._help_text(), reply_markup=self._main_menu_keyboard()) return if text.startswith("/status"): await self._send_message(chat_id, self._status_text(), reply_markup=self._main_menu_keyboard()) return if text.startswith("/metricas"): await self._send_chatbot_overview(chat_id) return if text.startswith("/campanas"): await self._send_chatbot_campaigns(chat_id) return if text.startswith("/campana "): campaign_ref = text.split(None, 1)[1].strip() await self._send_chatbot_campaign_detail(chat_id, campaign_ref) return if text.startswith("/waestado"): await self._send_whatsapp_hub(chat_id) return if text.startswith("/playbookwa"): await self._send_whatsapp_playbook(chat_id) return if text.startswith("/consentwa "): await self._handle_whatsapp_consent_command(chat_id, text, actor) return if text.startswith("/campanawa "): await self._handle_whatsapp_campaign_command(chat_id, text, actor) return if text.startswith("/enviarcampana "): await self._handle_send_campaign_command(chat_id, text, actor) return if text.startswith("/enviarwa "): await self._handle_send_wa_proposal_command(chat_id, text, actor) return if text.startswith("/marketingmenu"): await self._send_marketing_menu(chat_id) return if text.startswith("/marketing "): brief = text.split(None, 1)[1].strip() await self._prepare_marketing_and_notify(chat_id, brief, actor) return if text.startswith("/autonomia") or text.startswith("/automatizaciones"): await self._send_automation_hub(chat_id) return if text.startswith("/digest"): await self._send_digest(chat_id) return if text.startswith("/analytics"): await self._send_analytics_summary(chat_id) return if text.startswith("/trazas"): await self._send_langfuse_status(chat_id) return if text.startswith("/audit"): await self._send_audit_log(chat_id) return if text.startswith("/automatizacion"): await self._handle_automation_command(chat_id, text, actor) return if text.startswith("/aprendizaje"): await self._send_learning_status(chat_id) return if text.startswith("/modoaprendizaje"): mode = text.split(None, 1)[1].strip() if len(text.split(None, 1)) > 1 else "" if not mode: await self._send_learning_status(chat_id) return await self._set_learning_mode_and_notify(chat_id, mode, actor) return if text.startswith("/feedback "): await self._handle_feedback_command(chat_id, text, actor) return if text.startswith("/ops"): await self._send_ops_menu(chat_id) return if text.startswith("/memorias"): await self._send_recent_memories(chat_id) return if text.startswith("/seguridad"): await self._send_security_status(chat_id) return if text.startswith("/escanear"): await self._send_security_scan(chat_id, actor) return if text.startswith("/ssh"): parts = text.split(None, 2) sub = parts[1].strip().lower() if len(parts) > 1 else "" if sub == "test": await self._send_ssh_test(chat_id) elif sub == "comandos" or sub == "lista" or sub == "": await self._send_ssh_commands_menu(chat_id) elif sub == "proponer" and len(parts) > 2: await self._send_ssh_propose(chat_id, parts[2].strip(), actor, "") else: # /ssh directamente await self._send_ssh_propose(chat_id, sub, actor, "") return if text.startswith("/informe"): kind = text.split(None, 1)[1].strip().lower() if len(text.split(None, 1)) > 1 else "estado" await self._send_report_document(chat_id, kind, actor) return if text.startswith("/stackia"): await self._send_ai_stack(chat_id) return if text.startswith("/dashboard"): await self._send_message(chat_id, self._dashboard_text(), reply_markup=self._main_menu_keyboard()) return if text.startswith("/propuestas"): await self._send_pending_proposals(chat_id) return if text.startswith("/estudios"): await self._send_recent_studies(chat_id) return if text.startswith("/aprobar "): proposal_id = text.split(None, 1)[1].strip() await self._approve_and_notify(chat_id, proposal_id, actor, note="Aprobado desde Telegram") return if text.startswith("/rechazar "): proposal_id = text.split(None, 1)[1].strip() await self._reject_and_notify(chat_id, proposal_id, actor, note="Rechazado desde Telegram") return if text.startswith("/memorizar "): study_id = text.split(None, 1)[1].strip() await self._memorize_and_notify(chat_id, study_id, actor, note="Memoria promovida desde Telegram") return # Indicar que se está procesando antes de llamar al orquestador (puede tardar varios segundos) await self._send_chat_action(chat_id, "typing") result = await self.orchestrator.create_command_plan(text, actor, "telegram") study = result["study"] proposals = result["proposals"] await self._send_message( chat_id, self._study_text(study), reply_markup=self._study_keyboard(study["id"]), ) for proposal in proposals: await self._send_message( chat_id, self._proposal_text(proposal), reply_markup=self._proposal_keyboard(proposal), ) # Teclado de navegación rápida al final si hay propuestas pendientes pending_count = sum(1 for p in proposals if p.get("status") == "pending") if pending_count > 0: nav_keyboard = {"inline_keyboard": [ [ {"text": f"📋 Ver {pending_count} pendiente{'s' if pending_count!=1 else ''}", "callback_data": "menu:proposals"}, {"text": "🎛️ Menú", "callback_data": "menu:main"}, ] ]} await self._send_message( chat_id, f"_Revisa y aprueba {'las propuestas' if pending_count > 1 else 'la propuesta'} para ejecutar la acción._", reply_markup=nav_keyboard, ) async def _process_callback_query(self, callback_query: dict[str, Any]) -> None: callback_id = str(callback_query.get("id") or "") data = str(callback_query.get("data") or "").strip() message = callback_query.get("message") or {} chat = message.get("chat") or {} chat_id = str(chat.get("id", "")).strip() actor = self._actor_from_message(callback_query) if not callback_id or not data or not chat_id: return if not self._is_chat_allowed(chat_id): await self._answer_callback(callback_id, "Chat no autorizado", show_alert=True) await self._send_message(chat_id, self._unauthorized_text(chat_id)) return parts = data.split(":") action = parts[0].strip().lower() if action == "approve" and len(parts) >= 2 and parts[1].strip(): target_id = parts[1].strip() await self._answer_callback(callback_id, "Aprobando propuesta…") await self._approve_and_notify(chat_id, target_id, actor, note="Aprobado desde botón Telegram") return if action == "reject" and len(parts) >= 2 and parts[1].strip(): target_id = parts[1].strip() await self._answer_callback(callback_id, "Rechazando propuesta…") await self._reject_and_notify(chat_id, target_id, actor, note="Rechazado desde botón Telegram") return if action == "memorize" and len(parts) >= 2 and parts[1].strip(): target_id = parts[1].strip() await self._answer_callback(callback_id, "Promoviendo estudio a memoria…") await self._memorize_and_notify(chat_id, target_id, actor, note="Memorizado desde botón Telegram") return if action == "menu" and len(parts) >= 2: await self._answer_callback(callback_id, "Abriendo menú…") await self._dispatch_menu_action(chat_id, parts[1].strip().lower(), actor) return if action == "autorule" and len(parts) >= 3: mode = parts[1].strip().lower() rule_key = parts[2].strip() if mode == "run": await self._answer_callback(callback_id, "Lanzando automatización…") await self._run_automation_rule_and_notify(chat_id, rule_key, actor) return if mode == "toggle": await self._answer_callback(callback_id, "Cambiando automatización…") await self._toggle_automation_rule_and_notify(chat_id, rule_key, actor) return if action == "mode" and len(parts) >= 2: await self._answer_callback(callback_id, "Cambiando modo de aprendizaje…") await self._set_learning_mode_and_notify(chat_id, parts[1].strip().lower(), actor) return if action == "quickcmd" and len(parts) >= 2: await self._answer_callback(callback_id, "Preparando acción rápida…") await self._run_quick_command(chat_id, parts[1].strip(), actor) return if action == "quickmk" and len(parts) >= 2: await self._answer_callback(callback_id, "Preparando marketing…") await self._run_marketing_preset(chat_id, parts[1].strip(), actor) return if action == "report" and len(parts) >= 2: await self._answer_callback(callback_id, "Generando informe…") await self._send_report_document(chat_id, parts[1].strip(), actor) return if action == "sshp" and len(parts) >= 2: await self._answer_callback(callback_id, "Creando propuesta SSH…") await self._send_ssh_propose(chat_id, parts[1].strip(), actor, "") return if action == "ssh": if len(parts) >= 2 and parts[1] == "test": await self._answer_callback(callback_id, "Probando SSH…") await self._send_ssh_test(chat_id) else: await self._answer_callback(callback_id, "Cargando…") await self._send_ssh_commands_menu(chat_id) return if action == "feedback" and len(parts) >= 3: proposal_id = parts[1].strip() score = parts[2].strip() await self._answer_callback(callback_id, "Registrando feedback…") await self._record_feedback_and_notify(chat_id, actor, proposal_id, score) return await self._answer_callback(callback_id, "Acción no soportada", show_alert=True) async def _approve_and_notify(self, chat_id: str, proposal_id: str, actor: str, note: str | None = None) -> None: try: proposal = await self.orchestrator.approve_proposal(proposal_id, actor, note) execution_log = self._trim_text(proposal.get("execution_log") or "", 1600) message = [ f"✅ Propuesta aprobada: {proposal.get('title', 'Propuesta')}", f"ID: {proposal_id}", f"Estado final: {proposal.get('status', 'approved')}", ] if execution_log: message.extend(["", "Salida resumida:", execution_log]) await self._send_message(chat_id, "\n".join(message), reply_markup=self._proposal_result_keyboard(proposal)) except KeyError: await self._send_message(chat_id, f"No encuentro la propuesta {proposal_id}.") except Exception as exc: await self._send_message(chat_id, f"La aprobación de {proposal_id} ha fallado: {exc}") async def _reject_and_notify(self, chat_id: str, proposal_id: str, actor: str, note: str | None = None) -> None: try: proposal = self.orchestrator.reject_proposal(proposal_id, actor, note) await self._send_message( chat_id, f"❌ Propuesta rechazada: {proposal.get('title', 'Propuesta')}\nID: {proposal_id}\nEstado final: {proposal.get('status', 'rejected')}", ) except KeyError: await self._send_message(chat_id, f"No encuentro la propuesta {proposal_id}.") except Exception as exc: await self._send_message(chat_id, f"El rechazo de {proposal_id} ha fallado: {exc}") async def _memorize_and_notify(self, chat_id: str, study_id: str, actor: str, note: str | None = None) -> None: try: memory = self.orchestrator.memorize_study(study_id, actor, note) await self._send_message( chat_id, f"🧠 *Estudio promovido a memoria*\n" f"`{study_id}`\n" f"Asunto: {memory.get('subject', 'Memoria aprobada')}", reply_markup=self._main_menu_keyboard(), ) except KeyError: await self._send_message(chat_id, f"No encuentro el estudio `{study_id}`.", reply_markup=self._main_menu_keyboard()) except Exception as exc: await self._send_message(chat_id, f"No he podido memorizar `{study_id}`: {exc}") async def _send_pending_proposals(self, chat_id: str) -> None: proposals = self.store.list_proposals(status="pending", limit=8) if not proposals: await self._send_message(chat_id, "✅ No hay propuestas pendientes ahora mismo.", reply_markup=self._main_menu_keyboard()) return urgency = "🔴 urgente" if len(proposals) >= 5 else "🟡 pendientes" await self._send_message( chat_id, f"📌 {urgency}: *{len(proposals)} propuesta{'s' if len(proposals)!=1 else ''}* esperando revisión.", ) for proposal in proposals: await self._send_message(chat_id, self._proposal_text(proposal), reply_markup=self._proposal_keyboard(proposal)) async def _send_recent_studies(self, chat_id: str) -> None: studies = self.store.list_studies(limit=5) if not studies: await self._send_message(chat_id, "No hay estudios todavía.", reply_markup=self._main_menu_keyboard()) return lines = ["📚 Últimos estudios:"] for study in studies: lines.append( f"- {study['id'][:8]} · {study['summary']} · canal {study.get('channel', 'desconocido')}" ) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._main_menu_keyboard()) async def _send_digest(self, chat_id: str) -> None: """Resumen ejecutivo de las últimas 24 h: automatizaciones, propuestas, errores.""" counts = self.store.count_summary() automation_runs = self.store.list_automation_runs(limit=8) pending_proposals = self.store.list_proposals(status="pending", limit=6) recent_memories = self.store.list_memories(limit=3) sched = scheduler.get_status() if (scheduler := getattr(self, "_scheduler", None)) else {} sep = "✦━━━━━━━━━━━━━━━━━✦" run_lines: list[str] = [] for run in automation_runs: icon = "✅" if run.get("status") in {"planned", "executed", "success"} else "❌" rule = run.get("rule_key", "?")[:28] ts = str(run.get("created_at", ""))[:16] run_lines.append(f" {icon} `{rule}` · {ts}") prop_lines: list[str] = [] risk_icons = {"low": "🟢", "medium": "🟡", "high": "🔴"} for prop in pending_proposals: ri = risk_icons.get(prop.get("risk_level", ""), "⚪") prop_lines.append(f" {ri} {prop.get('title', 'Propuesta')[:42]}") mem_lines: list[str] = [] for mem in recent_memories: mem_lines.append(f" 🧠 {mem.get('subject', 'Memoria')[:48]}") sched_status = "🟢 activo" if sched.get("running") else "⚪ detenido" sections = [ "📅 *Digest del supervisor*", sep, "📊 *Contadores globales*", f" Estudios: {counts.get('studies', 0)} | Propuestas: {counts.get('proposals', 0)}", f" Pendientes: {counts.get('pending', 0)} | Ejecutadas: {counts.get('executed', 0)}", f" Memorias: {counts.get('memories', 0)} | Scheduler: {sched_status}", f" Automatizaciones: {counts.get('automation_enabled', 0)}/{counts.get('automation_rules', 0)} activas", ] if run_lines: sections += [sep, "🤖 *Últimas ejecuciones automáticas*"] + run_lines if prop_lines: sections += [sep, "📌 *Propuestas pendientes de revisión*"] + prop_lines if mem_lines: sections += [sep, "🧠 *Memorias recientes*"] + mem_lines sections += [sep, "Usa /propuestas para revisar · /autonomia para reglas · /audit para trazabilidad"] await self._send_message(chat_id, "\n".join(sections), reply_markup=self._automation_menu_keyboard()) async def _send_langfuse_status(self, chat_id: str) -> None: """Estado de Langfuse — observabilidad LLM desde Telegram.""" from . import langfuse_client as _lf info = _lf.get_summary() sep = "━━━━━━━━━━━━━━━━━━━" status_icon = "✅" if info["enabled"] and info["connected"] else ("⚠️" if info["package_available"] else "❌") if info["enabled"] and info["connected"]: status_text = "Activo y conectado" elif info["enabled"]: status_text = "Habilitado pero no conectado" else: status_text = "Desactivado \\(LANGFUSE\\_ENABLED\\=false\\)" pkg_text = "✅ instalado" if info["package_available"] else "❌ no instalado" host_safe = self.settings.langfuse_host.replace(".", "\\.").replace("-", "\\-").replace(":", "\\:") lines = [ "🔍 *Langfuse — Observabilidad LLM*", sep, f" Estado: {status_icon} {status_text}", f" Paquete: {pkg_text}", f" Host: `{host_safe}`", sep, ] if not info["enabled"]: lines += [ "📌 *Para activar self\\-hosted:*", "1\\. `docker compose \\-f docker\\-compose\\.langfuse\\.yml up \\-d`", "2\\. Abre http://localhost:3100 → crea proyecto → copia claves", "3\\. Pon en `.env`: `LANGFUSE\\_ENABLED\\=true` \\+ claves", "4\\. Reinicia el supervisor", ] else: lines += [ "📊 *Qué se observa:*", "\\- Cada llamada LLM: prompt, output, tokens, latencia", "\\- Trazas: plan de órdenes, marketing, chat directo", "\\- Scores de feedback del operador", sep, f" UI: {self.settings.langfuse_host}", ] lines.append(sep) lines.append("Usa /stackia para estado completo del stack IA") await self._send_message(chat_id, "\n".join(lines), reply_markup=self._main_menu_keyboard()) async def _send_analytics_summary(self, chat_id: str) -> None: """Resumen de analytics: visitas, interacciones y eventos del proyecto.""" data = self.store.count_analytics_summary() sep = "━━━━━━━━━━━━━━━━━━━" lines = [ "📈 *Analytics del proyecto*", sep, f" Total eventos registrados: `{data['total']}`", f" Hoy: `{data['today']}` | Últimos 7 días: `{data['this_week']}`", ] if data.get("by_type"): lines += [sep, "🏷️ *Por tipo de evento*"] for item in data["by_type"][:8]: lines.append(f" `{item['event_type']}` → {item['count']}") if data.get("top_pages"): lines += [sep, "📄 *Páginas / módulos más activos*"] for item in data["top_pages"][:5]: page = item["page"][:40] if item["page"] else "—" lines.append(f" `{page}` → {item['count']}") if data.get("events_7d"): lines += [sep, "📅 *Evolución últimos 7 días*"] for day_item in data["events_7d"]: bar = "▓" * min(int(day_item["count"]), 20) lines.append(f" `{day_item['date']}` {bar} {day_item['count']}") if data["total"] == 0: lines += [ sep, "⚠️ Sin eventos todavía\\. Para empezar a registrar visitas e interacciones, envía eventos desde el portal PHP a `POST /api/analytics/track`\\.", ] lines += [sep, "Usa /digest para resumen general · /audit para trazabilidad"] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._main_menu_keyboard()) async def _send_audit_log(self, chat_id: str) -> None: """Muestra las últimas 15 entradas del audit_log.""" entries = self.store.list_audit_log(limit=15) if not entries: await self._send_message( chat_id, "📋 No hay entradas en el audit log todavía.", reply_markup=self._main_menu_keyboard(), ) return sep = "━━━━━━━━━━━━━━━━━━━" lines = ["🔍 *Auditoría del sistema* \\(últimas 15\\)", sep] for e in entries: ts = str(e.get("created_at", ""))[:16] evt = e.get("event_type", "?") lines.append(f"`{ts}` · `{evt}`") lines.append(sep) lines.append("Usa /digest para resumen general · /status para estado del sistema") await self._send_message(chat_id, "\n".join(lines), reply_markup=self._main_menu_keyboard()) async def _send_chatbot_overview(self, chat_id: str) -> None: try: overview = await self.orchestrator.get_chatbot_overview() except Exception as exc: await self._send_message(chat_id, f"No he podido leer las métricas del chatbot: {exc}") return metrics = overview.get("metrics") or {} actions = overview.get("suggested_actions") or [] lines = [ "📈 Métricas del chatbot", "", f"Campañas totales: {metrics.get('campaigns_total', 0)}", f"Campañas activas: {metrics.get('campaigns_active', 0)}", f"Mensajes 24h: {metrics.get('messages_24h', 0)}", f"Consentimientos activos: {metrics.get('consents_active', 0)}", f"Saldo total créditos: {metrics.get('credits_balance_total', 0)}", f"Anunciantes con créditos: {metrics.get('advertisers_with_credits', 0)}", ] if actions: lines.extend(["", "Siguientes acciones sugeridas:"]) lines.extend(f"- {item}" for item in actions[:4]) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._metrics_menu_keyboard()) async def _send_chatbot_campaigns(self, chat_id: str) -> None: try: payload = await self.orchestrator.list_chatbot_campaigns(limit=8) except Exception as exc: await self._send_message(chat_id, f"No he podido leer campañas del chatbot: {exc}") return campaigns = payload.get("campaigns") or [] if not campaigns: await self._send_message(chat_id, "No veo campañas recientes en el chatbot ahora mismo.", reply_markup=self._campaigns_menu_keyboard()) return lines = ["📣 Campañas recientes del chatbot", ""] for item in campaigns[:8]: lines.append( f"- {item.get('campaign_ref', 'n/a')} · {item.get('name', 'Campaña')} · {item.get('status', 'n/a')} · {item.get('channel', 'n/a')} · enviados {item.get('sent_count', 0)}/{item.get('target_total', 0)}" ) lines.extend([ "", "Para ver una campaña concreta: /campana source:id", "Ejemplo: /campana bulk:12", ]) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._campaigns_menu_keyboard()) async def _send_chatbot_campaign_detail(self, chat_id: str, campaign_ref: str) -> None: source = None raw_id = campaign_ref if ":" in campaign_ref: source, raw_id = campaign_ref.split(":", 1) source = source.strip() or None try: campaign_id = int(raw_id.strip()) except ValueError: await self._send_message(chat_id, "Formato inválido. Usa /campana source:id o /campana 123") return try: detail = await self.orchestrator.get_chatbot_campaign(campaign_id, source) except Exception as exc: await self._send_message(chat_id, f"No he podido leer esa campaña: {exc}") return campaign = detail.get("campaign") or {} lines = [ "🧾 Detalle de campaña", "", f"Ref: {campaign.get('campaign_ref', campaign_ref)}", f"Nombre: {campaign.get('name', 'Campaña')}", f"Canal: {campaign.get('channel', 'n/a')}", f"Estado: {campaign.get('status', 'n/a')}", f"Objetivo: {campaign.get('target_total', 0)}", f"Enviados: {campaign.get('sent_count', 0)}", f"Entregados: {campaign.get('delivered_count', 0)}", f"Fallidos: {campaign.get('failed_count', 0)}", f"Respuestas: {campaign.get('responded_count', 0)}", ] preview = campaign.get('message_preview') if preview: lines.extend(["", f"Mensaje: {preview}"]) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._campaigns_menu_keyboard()) async def _prepare_marketing_and_notify(self, chat_id: str, brief: str, actor: str) -> None: if len(brief.strip()) < 3: await self._send_message(chat_id, "Escribe un brief algo más concreto. Ejemplo: /marketing reactiva anunciantes fríos por WhatsApp con foco Valencia") return try: result = await self.orchestrator.create_marketing_plan(brief, actor, "telegram-marketing") except Exception as exc: await self._send_message(chat_id, f"No he podido preparar el plan de marketing: {exc}") return study = result["study"] proposals = result["proposals"] await self._send_message(chat_id, self._study_text(study), reply_markup=self._study_keyboard(study["id"])) for proposal in proposals: await self._send_message( chat_id, self._proposal_text(proposal), reply_markup=self._proposal_keyboard(proposal), ) async def _show_main_menu(self, chat_id: str, actor: str | None = None) -> None: await self._send_message(chat_id, self._main_menu_text(actor), reply_markup=self._main_menu_keyboard()) async def _dispatch_menu_action(self, chat_id: str, section: str, actor: str) -> None: if section == "main": await self._show_main_menu(chat_id, actor) return if section == "status": await self._send_message(chat_id, self._status_text(), reply_markup=self._main_menu_keyboard()) return if section == "metrics": await self._send_chatbot_overview(chat_id) return if section == "campaigns": await self._send_chatbot_campaigns(chat_id) return if section == "whatsapp": await self._send_whatsapp_hub(chat_id) return if section == "marketing": await self._send_marketing_menu(chat_id) return if section == "automation": await self._send_automation_hub(chat_id) return if section == "learning": await self._send_learning_status(chat_id) return if section == "feedback": await self._send_recent_feedback(chat_id) return if section == "ops": await self._send_ops_menu(chat_id) return if section == "proposals": await self._send_pending_proposals(chat_id) return if section == "studies": await self._send_recent_studies(chat_id) return if section == "memories": await self._send_recent_memories(chat_id) return if section == "security": await self._send_security_status(chat_id) return if section == "stack": await self._send_ai_stack(chat_id) return if section == "reports": await self._send_reports_menu(chat_id) return if section == "ssh": await self._send_ssh_commands_menu(chat_id) return if section == "help": await self._send_message(chat_id, self._help_text(), reply_markup=self._main_menu_keyboard()) return await self._send_message(chat_id, "No reconozco esa sección del menú.", reply_markup=self._main_menu_keyboard()) async def _send_marketing_menu(self, chat_id: str) -> None: await self._send_message(chat_id, self._marketing_menu_text(), reply_markup=self._marketing_menu_keyboard()) async def _send_automation_hub(self, chat_id: str) -> None: state = self.orchestrator.get_autonomy_state() rules = state.get("rules") or [] recent_runs = state.get("recent_runs") or [] lines = [ "🤖 Automatizaciones supervisadas", "", f"Activas: {state.get('enabled', 0)}/{state.get('total', 0)}", "", "Estas rutinas no ejecutan cambios por su cuenta: generan estudios y propuestas para revisión.", "", "Reglas disponibles:", ] for rule in rules: status = "🟢" if rule.get("enabled") else "⚪" lines.append( f"- {status} {rule.get('rule_key')} · {rule.get('title')} · {rule.get('schedule_hint')}" ) if recent_runs: lines.extend(["", "Últimas ejecuciones:"]) for run in recent_runs[:4]: lines.append( f"- {run.get('rule_key')} · {run.get('status')} · {run.get('created_at')}" ) lines.extend( [ "", "Comando manual:", "/automatizacion ", ] ) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._automation_menu_keyboard()) async def _send_whatsapp_hub(self, chat_id: str) -> None: try: overview = await self.orchestrator.get_chatbot_overview() except Exception as exc: await self._send_message(chat_id, f"No he podido leer el estado de WhatsApp: {exc}", reply_markup=self._whatsapp_menu_keyboard()) return metrics = overview.get("metrics") or {} capabilities = overview.get("capabilities") or {} recent_campaigns = overview.get("recent_campaigns") or [] lines = [ "📲 WhatsApp, consentimiento y campañas reales", "", f"Canal habilitado: {'sí' if capabilities.get('whatsapp_enabled') else 'no'}", f"Credenciales Meta: {'listas' if capabilities.get('whatsapp_configured') else 'pendientes'}", f"Consentimientos activos: {metrics.get('consents_active', 0)}", f"Campañas activas: {metrics.get('campaigns_active', 0)}", f"Mensajes 24h: {metrics.get('messages_24h', 0)}", "", "Comandos reales:", "/consentwa | | | | [advertiser_id]", "/campanawa | | | | [YYYY-MM-DD HH:MM:SS]", "/enviarcampana o /enviarcampana marketing:id", "/playbookwa", ] if recent_campaigns: lines.extend(["", "Últimas campañas:"]) for item in recent_campaigns[:4]: lines.append( f"- {item.get('campaign_ref', 'n/a')} · {item.get('status', 'n/a')} · {item.get('sent_count', 0)}/{item.get('target_total', 0)}" ) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._whatsapp_menu_keyboard()) async def _send_whatsapp_playbook(self, chat_id: str) -> None: text = ( "📋 Playbook comercial WhatsApp\n\n" "1. Captura consentimiento explícito y guarda prueba.\n" "2. Empieza con un piloto pequeño (5-20 contactos), nunca con toda la base.\n" "3. Usa mensaje corto, claro y con salida fácil.\n" "4. Revisa métricas y errores antes de escalar.\n" "5. Solo después arma campañas mayores.\n\n" "Secuencia sugerida:\n" "- consentimiento\n" "- campaña draft\n" "- revisión de campaña\n" "- /enviarcampana \n" "- revisar /metricas y /campanas\n\n" "Ejemplo de consentimiento:\n" "/consentwa +34600111222 | Marta | formulario_web | https://xzonasvalencia.com/landing-optin | 123\n\n" "Ejemplo de campaña:\n" "/campanawa 123 | Reactivación marzo | Hola {nombre}, tengo una propuesta nueva para ti en XZonas. Si no quieres recibir más mensajes, me lo dices y te doy de baja. | +34600111222:Marta,+34600333444:Lucía" ) await self._send_message(chat_id, text, reply_markup=self._whatsapp_menu_keyboard()) async def _send_ops_menu(self, chat_id: str) -> None: await self._send_message(chat_id, self._ops_menu_text(), reply_markup=self._ops_menu_keyboard()) async def _send_recent_memories(self, chat_id: str) -> None: memories = self.store.list_memories(limit=6) if not memories: await self._send_message(chat_id, "No hay memorias aprobadas todavía.", reply_markup=self._learning_menu_keyboard()) return lines = ["🧠 Memorias recientes", ""] for memory in memories[:6]: lines.append(f"- {memory['subject']} · {memory['actor']} · {memory['created_at']}") await self._send_message(chat_id, "\n".join(lines), reply_markup=self._learning_menu_keyboard()) async def _send_learning_status(self, chat_id: str) -> None: state = self.orchestrator.get_learning_state() summary = state.get("summary") or {} lines = [ "🧠 Estado de aprendizaje", "", f"Modo actual: {state.get('mode', 'semi')}", f"Feedback total: {summary.get('total', 0)}", f"Puntuación media: {summary.get('avg_score', 0)}", f"Feedback positivo: {summary.get('positive', 0)}", f"Feedback negativo: {summary.get('negative', 0)}", "", "Modos:", "- manual: nada se promueve solo", "- semi: aprendes por feedback, pero memorizas tú", "- auto: feedback positivo puede promover memorias automáticamente", ] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._learning_menu_keyboard()) async def _send_recent_feedback(self, chat_id: str) -> None: feedback_items = self.orchestrator.get_learning_state().get("recent_feedback") or [] if not feedback_items: await self._send_message(chat_id, "Todavía no hay feedback registrado para el aprendizaje.", reply_markup=self._learning_menu_keyboard()) return lines = ["📝 Feedback reciente", ""] for item in feedback_items[:8]: lines.append( f"- score {item.get('score')} · {item.get('label')} · propuesta {item.get('proposal_id') or 'n/a'} · {item.get('created_at')}" ) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._learning_menu_keyboard()) async def _set_learning_mode_and_notify(self, chat_id: str, mode: str, actor: str) -> None: try: state = self.orchestrator.set_learning_mode(mode, actor) await self._send_message( chat_id, f"🧠 Modo de aprendizaje actualizado a: {state.get('mode', mode)}", reply_markup=self._learning_menu_keyboard(), ) except Exception as exc: await self._send_message(chat_id, f"No he podido cambiar el modo de aprendizaje: {exc}", reply_markup=self._learning_menu_keyboard()) async def _handle_feedback_command(self, chat_id: str, text: str, actor: str) -> None: parts = text.split(None, 3) if len(parts) < 3: await self._send_message(chat_id, "Uso: /feedback <1-5> [nota]", reply_markup=self._learning_menu_keyboard()) return proposal_id = parts[1].strip() score = parts[2].strip() note = parts[3].strip() if len(parts) > 3 else "" await self._record_feedback_and_notify(chat_id, actor, proposal_id, score, note) async def _record_feedback_and_notify(self, chat_id: str, actor: str, proposal_id: str, score: str, note: str = "") -> None: try: numeric_score = int(score) except ValueError: await self._send_message(chat_id, "La puntuación debe ser un número entre 1 y 5.", reply_markup=self._learning_menu_keyboard()) return try: result = self.orchestrator.record_learning_feedback( actor=actor, score=numeric_score, note=note, proposal_id=proposal_id, source="telegram", ) except Exception as exc: await self._send_message(chat_id, f"No he podido registrar el feedback: {exc}", reply_markup=self._learning_menu_keyboard()) return feedback = result.get("feedback") or {} memory = result.get("memory") lines = [ "🧪 Feedback registrado", "", f"Propuesta: {proposal_id}", f"Puntuación: {feedback.get('score', numeric_score)}/5", f"Etiqueta: {feedback.get('label', 'n/a')}", f"Modo: {result.get('mode', 'semi')}", ] if note: lines.extend(["", f"Nota: {note}"]) if memory: lines.extend(["", f"🧠 Memoria auto-promovida: {memory.get('subject', 'Memoria aprobada')}"]) elif result.get("mode") == "semi" and numeric_score >= 4: lines.extend(["", "Sugerencia: como el feedback es positivo, puedes promover el estudio a memoria si quieres consolidarlo."]) await self._send_message(chat_id, "\n".join(lines), reply_markup=self._learning_menu_keyboard()) async def _handle_automation_command(self, chat_id: str, text: str, actor: str) -> None: parts = text.split(None, 2) if len(parts) < 3: await self._send_message( chat_id, "Uso: /automatizacion ", reply_markup=self._automation_menu_keyboard(), ) return rule_key = parts[1].strip() command = parts[2].strip().lower() if command == "run": await self._run_automation_rule_and_notify(chat_id, rule_key, actor) return if command in {"on", "off"}: await self._set_automation_rule_enabled_and_notify(chat_id, rule_key, command == "on", actor) return await self._send_message(chat_id, "Acción inválida. Usa run, on u off.", reply_markup=self._automation_menu_keyboard()) async def _set_automation_rule_enabled_and_notify(self, chat_id: str, rule_key: str, enabled: bool, actor: str) -> None: try: rule = self.orchestrator.set_automation_rule_enabled(rule_key, enabled, actor) state = "activada" if rule.get("enabled") else "desactivada" await self._send_message( chat_id, f"🤖 Automatización {state}: {rule.get('title', rule_key)}", reply_markup=self._automation_menu_keyboard(), ) except KeyError: await self._send_message(chat_id, f"No encuentro la automatización {rule_key}.", reply_markup=self._automation_menu_keyboard()) except Exception as exc: await self._send_message(chat_id, f"No he podido cambiar {rule_key}: {exc}", reply_markup=self._automation_menu_keyboard()) async def _toggle_automation_rule_and_notify(self, chat_id: str, rule_key: str, actor: str) -> None: rules = {rule["rule_key"]: rule for rule in self.orchestrator.list_automation_rules()} rule = rules.get(rule_key) if rule is None: await self._send_message(chat_id, f"No encuentro la automatización {rule_key}.", reply_markup=self._automation_menu_keyboard()) return await self._set_automation_rule_enabled_and_notify(chat_id, rule_key, not bool(rule.get("enabled")), actor) async def _run_automation_rule_and_notify(self, chat_id: str, rule_key: str, actor: str) -> None: try: payload = await self.orchestrator.run_automation_rule(rule_key, actor) except KeyError: await self._send_message(chat_id, f"No encuentro la automatización {rule_key}.", reply_markup=self._automation_menu_keyboard()) return except Exception as exc: await self._send_message(chat_id, f"La automatización {rule_key} ha fallado: {exc}", reply_markup=self._automation_menu_keyboard()) return result = payload.get("result") or {} rule = payload.get("rule") or {} study = result.get("study") or {} proposals = result.get("proposals") or [] await self._send_message( chat_id, f"🤖 Automatización ejecutada: {rule.get('title', rule_key)}\nCanal: automation:{rule_key}", reply_markup=self._automation_menu_keyboard(), ) if study: await self._send_message(chat_id, self._study_text(study), reply_markup=self._study_keyboard(study["id"])) for proposal in proposals: await self._send_message(chat_id, self._proposal_text(proposal), reply_markup=self._proposal_keyboard(proposal)) async def _run_quick_command(self, chat_id: str, command_key: str, actor: str) -> None: quick = TELEGRAM_QUICK_COMMANDS.get(command_key) if not quick: await self._send_message(chat_id, "No reconozco esa acción rápida.", reply_markup=self._ops_menu_keyboard()) return result = await self.orchestrator.create_command_plan(quick["text"], actor, "telegram-quick") study = result["study"] proposals = result["proposals"] await self._send_message(chat_id, self._study_text(study), reply_markup=self._study_keyboard(study["id"])) for proposal in proposals: await self._send_message(chat_id, self._proposal_text(proposal), reply_markup=self._proposal_keyboard(proposal)) async def _run_marketing_preset(self, chat_id: str, preset: str, actor: str) -> None: brief = TELEGRAM_MARKETING_PRESETS.get(preset) if not brief: await self._send_message(chat_id, "No reconozco ese preset de marketing.", reply_markup=self._marketing_menu_keyboard()) return await self._prepare_marketing_and_notify(chat_id, brief, actor) async def _send_security_status(self, chat_id: str) -> None: from . import security_scanner as _scanner last = self.store.get_latest_security_scan() if last: scan_line = ( f"Último escaneo: {last['started_at'][:19]}\n" f" Estado: {last['status']}\n" f" Vulnerabilidades: {last['total_vulnerabilities']} " f"(🔴{last['critical']} 🟠{last['high']} 🟡{last['medium']} 🟢{last['low']})" ) else: scan_line = "Último escaneo: ninguno — usa /escanear" lines = [ "🔐 Estado de seguridad", "", f"Chat autorizado: {'sí' if self._is_chat_allowed(chat_id) else 'no'}", f"Chats permitidos: {len(self.allowed_chat_ids)}", f"SSH profile: {self.settings.default_ssh_profile}", f"SSH vía agent: {'sí' if self.settings.ssh_use_agent else 'no'}", "Shell arbitraria: bloqueada", "Ejecución técnica: solo allowlist + aprobación", "Autoejecución low-risk: desactivada" if not self.settings.auto_execute_low_risk else "Autoejecución low-risk: activada", f"Modo aprendizaje: {self.orchestrator.get_learning_mode()}", "", scan_line, ] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._security_menu_keyboard()) async def _send_security_scan(self, chat_id: str, actor: str) -> None: """Lanza pip-audit + trivy y reporta el resultado en Telegram.""" from . import security_scanner as _scanner await self._send_message(chat_id, "🔍 Iniciando escaneo de seguridad...") try: result = await _scanner.run_full_scan(actor=f"telegram:{actor}") self.store.save_security_scan(result) except Exception as exc: await self._send_message(chat_id, f"❌ Error en el escaneo: {exc}") return status_icon = {"clean": "✅", "vulnerabilities_found": "⚠️", "error": "❌"}.get(result.get("status", ""), "ℹ️") tools = ", ".join(result.get("tools_used", [])) or "ninguna" lines = [ f"{status_icon} Escaneo completado — {result.get('status', '?')}", f"Herramientas: {tools}", f"Total CVE: {result.get('total_vulnerabilities', 0)}", f" 🔴 Crítico: {result.get('critical', 0)}", f" 🟠 Alto: {result.get('high', 0)}", f" 🟡 Medio: {result.get('medium', 0)}", f" 🟢 Bajo: {result.get('low', 0)}", "", result.get("summary", ""), ] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._security_menu_keyboard()) async def _send_ssh_test(self, chat_id: str) -> None: """Prueba la conectividad SSH y envía el resultado al chat.""" await self._send_message(chat_id, "\ud83d\udd0c Probando conexi\u00f3n SSH\u2026") try: result = await self._api_call("GET", "/api/ssh/test") if result.get("ok"): output = result.get("output", "") text = ( f"\u2705 *SSH conectado correctamente*\n" f"`{result.get('user')}@{result.get('host')}:{result.get('port')}`\n\n" f"```\n{output[:600]}\n```" ) else: err = result.get("error") or result.get("stderr", "Error desconocido") text = f"\u274c *SSH fall\u00f3*\n`{err[:400]}`" await self._send_message(chat_id, text) except Exception as exc: await self._send_message(chat_id, f"\u274c Error al probar SSH: {exc}") async def _send_ssh_commands_menu(self, chat_id: str) -> None: """Env\u00eda la lista de comandos SSH con botones para proponer cada uno.""" try: data = await self._api_call("GET", "/api/ssh/commands") except Exception as exc: await self._send_message(chat_id, f"\u274c No pude cargar los comandos SSH: {exc}") return cmds = data.get("commands") or [] risk_map = {"low": "\ud83d\udfe2", "medium": "\ud83d\udfe1", "high": "\ud83d\udd34"} header = ( f"\ud83d\udda5\ufe0f *Biblioteca SSH* \u2014 `{data.get('user')}@{data.get('host')}:{data.get('port')}`\n" f"{data.get('total_commands', 0)} comandos disponibles\n\n" "Pulsa un bot\u00f3n para proponer un comando para aprobaci\u00f3n\\.\n" "Tambi\u00e9n puedes usar `/ssh ` directamente\\." ) # Agrupar los botones de 2 en 2 buttons: list[list[dict[str, str]]] = [] row: list[dict[str, str]] = [] visible = [c for c in cmds if not c["key"].startswith("_")] for c in visible: icon = risk_map.get(c["risk"], "\u26aa") row.append({"text": f"{icon} {c['key']}", "callback_data": f"sshp:{c['key']}"}) if len(row) == 2: buttons.append(row) row = [] if row: buttons.append(row) buttons.append([ {"text": "\ud83d\udd0c Test conexi\u00f3n", "callback_data": "ssh:test"}, {"text": "\u2b05\ufe0f Men\u00fa", "callback_data": "menu:main"}, ]) await self._send_message(chat_id, header, reply_markup={"inline_keyboard": buttons}) async def _send_ssh_propose(self, chat_id: str, command_key: str, actor: str, reason: str) -> None: """Crea una propuesta SSH directa sin pasar por el motor de estudio.""" if not command_key: await self._send_ssh_commands_menu(chat_id) return try: result = await self._api_call( "POST", "/api/ssh/propose", payload={"command_key": command_key, "actor": actor, "reason": reason}, ) from_reason = f"\n_Motivo_: {reason}" if reason else "" text = ( f"\ud83d\udce4 *Propuesta SSH creada*\n" f"Comando: `{command_key}`\n" f"Riesgo: `{result.get('risk', '?')}`{from_reason}\n\n" f"{result.get('message', '')}" ) await self._send_message(chat_id, text) except Exception as exc: await self._send_message(chat_id, f"\u274c Error al proponer comando SSH `{command_key}`: {exc}\n\nUsa `/ssh comandos` para ver los disponibles\\.") async def _send_reports_menu(self, chat_id: str) -> None: """Menú de informes .md descargables desde Telegram.""" text = ( "📄 *Informes descargables*\n\n" "Recibes el informe como fichero `.md` adjunto\\. Úsalo en tu editor, Obsidian, Notion o lo que necesites\\.\n\n" "*Disponibles:*\n" "`/informe estado` — estado completo del sistema\n" "`/informe propuestas` — todas las propuestas pendientes\n" "`/informe estudios` — últimos 12 estudios generados\n" "`/informe seguridad` — último escaneo CVE con detalle\n" "`/informe digest` — resumen ejecutivo del día\n" "`/informe memorias` — memorias aprobadas\n" "`/informe automatizaciones` — reglas e historial de ejecuciones" ) keyboard = {"inline_keyboard": [ [ {"text": "📊 Estado", "callback_data": "report:estado"}, {"text": "📋 Propuestas", "callback_data": "report:propuestas"}, ], [ {"text": "📚 Estudios", "callback_data": "report:estudios"}, {"text": "🔐 Seguridad", "callback_data": "report:seguridad"}, ], [ {"text": "📅 Digest", "callback_data": "report:digest"}, {"text": "🧠 Memorias", "callback_data": "report:memorias"}, ], [ {"text": "⚙️ Automatizaciones", "callback_data": "report:automatizaciones"}, ], [{"text": "⬅️ Menú", "callback_data": "menu:main"}], ]} await self._send_message(chat_id, text, reply_markup=keyboard) async def _send_ai_stack(self, chat_id: str) -> None: await self._send_message(chat_id, self._ai_stack_text(), reply_markup=self._stack_menu_keyboard()) async def _handle_whatsapp_consent_command(self, chat_id: str, text: str, actor: str) -> None: body = text.split(None, 1)[1].strip() if len(text.split(None, 1)) > 1 else "" fields = [item.strip() for item in body.split("|")] if len(fields) < 4: await self._send_message( chat_id, "Uso: /consentwa | | | | [advertiser_id]", reply_markup=self._whatsapp_menu_keyboard(), ) return advertiser_id = None if len(fields) >= 5 and fields[4]: try: advertiser_id = int(fields[4]) except ValueError: await self._send_message(chat_id, "El advertiser_id debe ser numérico.", reply_markup=self._whatsapp_menu_keyboard()) return try: result = await self.orchestrator.create_chatbot_consent( actor=actor, phone=fields[0], name=fields[1] or None, source=fields[2] or "supervisor", proof=fields[3] or None, advertiser_id=advertiser_id, ) except Exception as exc: await self._send_message(chat_id, f"No he podido registrar el consentimiento: {exc}", reply_markup=self._whatsapp_menu_keyboard()) return consent = result.get("consent") or {} lines = [ "✅ Consentimiento WhatsApp registrado", "", f"Creado ahora: {'sí' if result.get('created') else 'no, ya existía'}", f"Teléfono: {consent.get('phone', fields[0])}", f"Nombre: {consent.get('name', fields[1] or 'n/a')}", f"Fuente: {consent.get('source', fields[2])}", f"Advertiser ID: {consent.get('advertiser_id', advertiser_id or 'n/a')}", ] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._whatsapp_menu_keyboard()) async def _handle_whatsapp_campaign_command(self, chat_id: str, text: str, actor: str) -> None: body = text.split(None, 1)[1].strip() if len(text.split(None, 1)) > 1 else "" fields = [item.strip() for item in body.split("|")] if len(fields) < 4: await self._send_message( chat_id, "Uso: /campanawa | | | | [YYYY-MM-DD HH:MM:SS]", reply_markup=self._whatsapp_menu_keyboard(), ) return try: advertiser_id = int(fields[0]) except ValueError: await self._send_message(chat_id, "El advertiser_id debe ser numérico.", reply_markup=self._whatsapp_menu_keyboard()) return targets: list[dict[str, Any]] = [] for raw_target in [item.strip() for item in fields[3].split(",") if item.strip()]: phone = raw_target name = None if ":" in raw_target: phone, name = raw_target.split(":", 1) targets.append({"phone": phone.strip(), "name": (name or "").strip() or None}) if not targets: await self._send_message(chat_id, "No he detectado targets válidos. Usa coma para separar y opcionalmente :nombre.", reply_markup=self._whatsapp_menu_keyboard()) return scheduled_at = fields[4] if len(fields) >= 5 and fields[4] else None try: result = await self.orchestrator.create_chatbot_campaign( actor=actor, advertiser_id=advertiser_id, name=fields[1], message_template=fields[2], description="Creada desde Telegram", scheduled_at=scheduled_at, targets=targets, ) except Exception as exc: await self._send_message(chat_id, f"No he podido crear la campaña WhatsApp: {exc}", reply_markup=self._whatsapp_menu_keyboard()) return campaign = result.get("campaign") or {} lines = [ "📣 Campaña WhatsApp creada", "", f"Ref: {campaign.get('campaign_ref', 'n/a')}", f"Nombre: {campaign.get('name', fields[1])}", f"Estado: {campaign.get('status', 'draft')}", f"Targets creados: {result.get('created_targets', 0)}", f"Targets descartados: {len(result.get('skipped_targets') or [])}", "", f"Para armar el envío: /enviarcampana {campaign.get('id', 'ID')}", ] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._whatsapp_menu_keyboard()) async def _handle_send_campaign_command(self, chat_id: str, text: str, actor: str) -> None: raw_ref = text.split(None, 1)[1].strip() if len(text.split(None, 1)) > 1 else "" source = None raw_id = raw_ref if ":" in raw_ref: source, raw_id = raw_ref.split(":", 1) source = source.strip() or None try: campaign_id = int(raw_id.strip()) except ValueError: await self._send_message(chat_id, "Usa /enviarcampana o /enviarcampana marketing:id", reply_markup=self._whatsapp_menu_keyboard()) return try: result = await self.orchestrator.send_chatbot_campaign(actor=actor, campaign_id=campaign_id, source=source, immediate=True) except Exception as exc: await self._send_message(chat_id, f"No he podido armar la campaña para envío: {exc}", reply_markup=self._whatsapp_menu_keyboard()) return campaign = result.get("campaign") or {} lines = [ "🚀 Campaña armada para envío", "", f"Ref: {campaign.get('campaign_ref', raw_ref)}", f"Estado: {campaign.get('status', 'processing')}", f"Pendientes: {result.get('pending_targets', 0)}", f"Modo: {result.get('mode', 'immediate')}", "", str(result.get('message') or 'Ahora el cron del chatbot debe recoger los targets pendientes.'), ] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._campaigns_menu_keyboard()) async def _handle_send_wa_proposal_command(self, chat_id: str, text: str, actor: str) -> None: # noqa: ARG002 """Envía una propuesta de tipo whatsapp_campaign aprobada. Uso: /enviarwa Requiere que la propuesta esté en estado 'approved' y sea de tipo whatsapp_campaign. """ proposal_id = text.split(None, 1)[1].strip() if len(text.split(None, 1)) > 1 else "" if not proposal_id: await self._send_message( chat_id, "Uso: /enviarwa \n\nVer propuestas aprobadas: /propuestas", reply_markup=self._whatsapp_menu_keyboard(), ) return if self._wa_campaign_sender is None: await self._send_message(chat_id, "⚠️ Envío de campañas WA no disponible en esta instancia.", reply_markup=self._whatsapp_menu_keyboard()) return await self._send_message(chat_id, f"⏳ Iniciando envío de campaña WA para propuesta `{proposal_id[:8]}`…") try: result = await self._wa_campaign_sender(proposal_id) except KeyError as exc: await self._send_message(chat_id, f"❌ Propuesta no encontrada: {exc}", reply_markup=self._whatsapp_menu_keyboard()) return except ValueError as exc: await self._send_message(chat_id, f"❌ No se puede enviar: {exc}", reply_markup=self._whatsapp_menu_keyboard()) return except Exception as exc: logger.exception("telegram.enviarwa.error", extra={"proposal_id": proposal_id, "error": str(exc)}) await self._send_message(chat_id, f"❌ Error inesperado: {exc}", reply_markup=self._whatsapp_menu_keyboard()) return sent = result.get("sent", 0) failed = result.get("failed", 0) lines = [ "✅ Campaña WhatsApp enviada", "", f"Propuesta: {proposal_id[:8]}", f"Enviados: {sent}", f"Errores: {failed}", f"Resumen: {result.get('summary', '')}", ] await self._send_message(chat_id, "\n".join(lines), reply_markup=self._whatsapp_menu_keyboard()) async def _handle_whoami(self, chat_id: str) -> None: allowed = self._is_chat_allowed(chat_id) status = "autorizado" if allowed else "todavía no autorizado" message = ( f"👤 Este chat_id es: {chat_id}\n" f"Estado actual: {status}.\n\n" f"Si quieres autorizar este chat, añade TELEGRAM_ALLOWED_CHAT_IDS={chat_id} en tu .env del supervisor." ) await self._send_message(chat_id, message) def _is_chat_allowed(self, chat_id: str) -> bool: return bool(self.allowed_chat_ids) and chat_id in self.allowed_chat_ids def _actor_from_message(self, payload: dict[str, Any]) -> str: user = payload.get("from") or {} username = str(user.get("username") or "").strip() first_name = str(user.get("first_name") or "").strip() if username: return f"telegram:{username}" if first_name: return f"telegram:{first_name}" return f"telegram:{self.settings.default_operator}" def _start_text(self, chat_id: str) -> str: return ( "🤖 XZonas Multiagent Supervisor\n\n" f"Chat ID detectado: {chat_id}\n" f"Canal Telegram: {'activo' if self.enabled else 'desactivado'}\n\n" "Usa /menu para abrir el centro de mando, /whoami para ver tu chat_id y /dashboard si quieres abrir el entorno gráfico." ) def _unauthorized_text(self, chat_id: str) -> str: return ( "⛔ Este chat no está autorizado para operar el supervisor.\n\n" f"Tu chat_id es {chat_id}. Añádelo a TELEGRAM_ALLOWED_CHAT_IDS en el .env y reinicia el servicio.\n" "Mientras tanto, puedes usar /whoami para confirmar el identificador correcto." ) def _help_text(self) -> str: sep = "━━━━━━━━━━━━━━━━━━━" return ( "📖 *Comandos disponibles*\n" f"{sep}\n" "🖥️ *Sistema*\n" "/menu — centro de mando principal\n" "/status — estado completo del supervisor\n" "/digest — resumen ejecutivo de las últimas 24 h\n" "/analytics — visitas, interacciones y tracking del proyecto\n" "/audit — auditoría de eventos internos\n" "/dashboard — enlace al panel web\n" "/whoami — ver tu chat\\_id\n" f"{sep}\n" "🎤 *Entrada de voz*\n" "Envía un mensaje de voz y el bot lo transcribirá automáticamente con faster\\-whisper \\(local, sin coste\\)\\.\n" "Máximo 2 minutos\\. Idioma principal: español\\.\n" f"{sep}\n" "🤖 *Automatización*\n" "/autonomia — estado de todas las reglas activas\n" "/automatizaciones — alias de /autonomia\n" "/automatizacion \\ \\ — controlar una regla\n" f"{sep}\n" "📣 *Marketing*\n" "/metricas — métricas del chatbot\n" "/campanas — campañas recientes\n" "/campana \\ — detalle de campaña\n" "/waestado — estado WhatsApp\n" "/playbookwa — playbook comercial WhatsApp\n" "/marketingmenu — menú de marketing\n" "/marketing \\ — preparar plan con IA\n" "/consentwa \\|\\|\\|\\|\\[id\\] — registrar consentimiento\n" "/campanawa \\ — crear campaña real con consentimientos\n" "/enviarcampana \\ — lanzar campaña al worker\n" f"{sep}\n" "🛠️ *Operaciones*\n" "/ops — menú de operaciones SSH\n" "/seguridad — guardarraíles del sistema\n" "/stackia — stack IA recomendado\n" "/trazas — estado Langfuse \\(observabilidad LLM\\)\n" f"{sep}\n" "🧠 *Aprendizaje*\n" "/aprendizaje — estado del aprendizaje\n" "/modoaprendizaje \\ — cambiar modo\n" "/feedback \\ \\<1\\-5\\> \\[nota\\] — valorar propuesta\n" "/memorias — memorias aprobadas recientes\n" f"{sep}\n" "📊 *Seguimiento*\n" "/propuestas — propuestas pendientes\n" "/aprobar \\ — aprobar propuesta\n" "/rechazar \\ — rechazar propuesta\n" "/estudios — últimos estudios generados\n" "/memorizar \\ — memorizar estudio\n" f"{sep}\n" "🖥️ *SSH — Biblioteca de comandos*\n" "/ssh — lista de comandos permitidos\n" "/ssh test — prueba conexión SSH al servidor\n" "/ssh \\ — proponer comando \\(ej: `/ssh git\\_status`\\)\n" "_Los comandos requieren tu aprobación antes de ejecutarse_\n" f"{sep}\n" "📄 *Informes descargables \\(.md\\)*\n" "/informe \\ — descarga informe como fichero adjunto\n" " tipos: `estado` `propuestas` `estudios` `seguridad` `digest` `memorias` `automatizaciones`\n" f"{sep}\n" "_Cualquier texto libre genera estudio \\+ propuestas automáticamente\\._" ) def _status_text(self) -> str: counts = self.store.count_summary() telegram_mode = self.settings.telegram_mode if self.enabled else "disabled" learning_mode = self.orchestrator.get_learning_mode() bridge_ok = bool(self.settings.chatbot_bridge_url and self.settings.chatbot_admin_token) sep = "━━━━━━━━━━━━━━━━━━━" ollama_extra = "" if self.settings.provider_name == "ollama": ollama_extra = f"\n Modelo: `{self.settings.ollama_model}`" # Indicadores de estado para pendientes pending = counts.get('pending', 0) pending_indicator = "🔴 urgente" if pending > 10 else ("🟡 hay pendientes" if pending > 0 else "🟢 al día") # Trivy disponible try: import shutil as _shutil trivy_ok = bool(_shutil.which("trivy") or __import__("pathlib").Path(r"C:\tools\trivy\trivy.exe").exists()) except Exception: trivy_ok = False now_utc = datetime.now(timezone.utc).strftime("%d %b %Y %H:%M UTC") return ( "📊 *Estado del supervisor*\n" f"`{now_utc}`\n" f"{sep}\n" "🧠 *IA y aprendizaje*\n" f" Proveedor: `{self.orchestrator.provider.name}`{ollama_extra}\n" f" Modo aprendizaje: `{learning_mode}`\n" f"{sep}\n" "⚙️ *Automatización*\n" f" Reglas activas: `{counts['automation_enabled']}/{counts['automation_rules']}`\n" f" Scheduler: {'🟢 activo' if counts.get('automation_enabled', 0) > 0 else '⚪ sin reglas activas'}\n" f"{sep}\n" "📊 *Contadores*\n" f" Estudios: `{counts['studies']}` Propuestas: `{counts['proposals']}`\n" f" Pendientes: `{pending}` — {pending_indicator}\n" f" Ejecutadas: `{counts['executed']}` Memorias: `{counts['memories']}`\n" f"{sep}\n" "🌐 *Integraciones*\n" f" Telegram: `{telegram_mode}` 🎤 voz activa\n" f" Bridge chatbot: {'✅ activo' if bridge_ok else '❌ no configurado'}\n" f" Perfil SSH: `{self.settings.default_ssh_profile}`\n" f" Trivy: {'✅ disponible' if trivy_ok else '⚠️ no instalado'}\n" f" Dashboard: {self.settings.public_base_url or '_no configurado_'}\n" f"{sep}\n" "_Usa /digest para resumen 24 h · /audit para trazabilidad_" ) def _dashboard_text(self) -> str: return ( "🖥️ Entorno gráfico del supervisor\n\n" f"Abre este enlace en navegador: {self.settings.public_base_url}\n\n" "Tu flujo principal ahora puede ser Telegram-first; el dashboard queda como consola secundaria y visual." ) def _main_menu_text(self, actor: str | None = None) -> str: counts = self.store.count_summary() bridge_ok = bool(self.settings.chatbot_bridge_url and self.settings.chatbot_admin_token) pending = counts.get('pending', 0) pending_icon = "🔴" if pending > 5 else ("🟡" if pending > 0 else "🟢") learn = self.orchestrator.get_learning_mode() learn_icon = {"manual": "🔵", "semi": "🟡", "auto": "🔴"}.get(learn, "⚪") sep = "─" * 18 now_utc = datetime.now(timezone.utc).strftime("%d %b %Y %H:%M UTC") return ( "🎛️ *Centro de mando*\n" f"`{now_utc}`\n" f"{sep}\n" f"👤 Operador: `{actor or self.settings.default_operator}`\n" f"🧠 IA: `{self.orchestrator.provider.name}` {learn_icon} modo `{learn}`\n" f"⚙️ Automatizaciones: `{counts['automation_enabled']}/{counts['automation_rules']}` activas\n" f"📌 Pendientes: {pending_icon} `{pending}` · Ejecutadas: `{counts['executed']}`\n" f"📚 Estudios: `{counts['studies']}` · 🧠 Memorias: `{counts['memories']}`\n" f"🌐 Bridge: {'✅' if bridge_ok else '❌ no configurado'}\n" f"{sep}\n" "_Texto libre → estudio \\+ propuestas · 🎤 Voz también funciona_" ) def _marketing_menu_text(self) -> str: return ( "🚀 Marketing y campañas\n\n" "Aquí preparas acciones comerciales sin lanzar nada a ciegas.\n" "Los presets te crean estudios y propuestas listas para revisar desde Telegram.\n" "Si quieres tocar campaña real + consentimientos, entra en WhatsApp o usa /waestado." ) def _ops_menu_text(self) -> str: return ( "🛠️ Operaciones rápidas\n\n" "Estas acciones generan estudios o propuestas seguras sobre allowlist SSH.\n" "Nada de shell libre ni comandos creativos de madrugada." ) def _ai_stack_text(self) -> str: return ( "🧪 Stack IA recomendado (open source / self-hosted)\n\n" "1. Ollama → inferencia local y API /api/chat para el supervisor.\n" "2. Open WebUI → laboratorio visual y operación self-hosted con Ollama.\n" "3. Qdrant → vector DB para RAG/memoria semántica futura; ojo, por defecto local no trae auth.\n" "4. Mem0 OSS → capa de memoria adaptativa sobre LLM + vector store + SQLite.\n" "5. GitHub Copilot Chat + Serena → capa experta de ingeniería en VS Code; no daemon directo del servidor.\n\n" "Modelo actual sugerido para este supervisor: kimi-k2.5:cloud sobre Ollama mientras decides si quieres tirar de modelo cloud o descargar uno local pesado." ) def _main_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "📊 Estado", "callback_data": "menu:status"}, {"text": "📈 Métricas", "callback_data": "menu:metrics"}, ], [ {"text": "📣 Campañas", "callback_data": "menu:campaigns"}, {"text": "📲 WhatsApp", "callback_data": "menu:whatsapp"}, ], [ {"text": "🚀 Marketing", "callback_data": "menu:marketing"}, {"text": "📌 Propuestas", "callback_data": "menu:proposals"}, ], [ {"text": "📚 Estudios", "callback_data": "menu:studies"}, {"text": "🧠 Aprendizaje", "callback_data": "menu:learning"}, ], [ {"text": "🤖 Automatizaciones", "callback_data": "menu:automation"}, {"text": "🛠️ Ops", "callback_data": "menu:ops"}, ], [ {"text": "🔐 Seguridad", "callback_data": "menu:security"}, {"text": "🧪 Stack IA", "callback_data": "menu:stack"}, {"text": "🧠 Memorias", "callback_data": "menu:memories"}, ], [ {"text": "❓ Ayuda", "callback_data": "menu:help"}, {"text": "📄 Informes .md", "callback_data": "menu:reports"}, ], [ {"text": "🖥️ SSH", "callback_data": "ssh:menu"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _metrics_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "🔄 Refrescar métricas", "callback_data": "menu:metrics"}, {"text": "📣 Ver campañas", "callback_data": "menu:campaigns"}, ], [ {"text": "🚀 Marketing", "callback_data": "menu:marketing"}, {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _campaigns_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "📈 Métricas", "callback_data": "menu:metrics"}, {"text": "🚀 Marketing", "callback_data": "menu:marketing"}, ], [ {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _marketing_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "♻️ Reactivación", "callback_data": "quickmk:reactivacion"}, {"text": "🎯 Captación", "callback_data": "quickmk:captacion"}, ], [ {"text": "💬 Copy WA", "callback_data": "quickmk:copy"}, {"text": "🧪 Oferta test", "callback_data": "quickmk:oferta"}, ], [ {"text": "📲 WhatsApp real", "callback_data": "menu:whatsapp"}, {"text": "📈 Métricas", "callback_data": "menu:metrics"}, ], [ {"text": "📣 Campañas", "callback_data": "menu:campaigns"}, {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _whatsapp_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "🧾 Estado WA", "callback_data": "menu:whatsapp"}, {"text": "📈 Métricas", "callback_data": "menu:metrics"}, ], [ {"text": "📣 Campañas", "callback_data": "menu:campaigns"}, {"text": "🚀 Marketing", "callback_data": "menu:marketing"}, ], [ {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _ops_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "🌐 Despliegue", "callback_data": "quickcmd:git_status"}, {"text": "📜 Logs PHP", "callback_data": "quickcmd:tail_php_errorlog"}, ], [ {"text": "🧪 Lint PHP", "callback_data": "quickcmd:composer_lint"}, {"text": "🔎 PHPStan", "callback_data": "quickcmd:phpstan"}, ], [ {"text": "📐 Psalm", "callback_data": "quickcmd:psalm"}, {"text": "🧩 Grid", "callback_data": "quickcmd:npm_build_grid"}, ], [ {"text": "🗜️ Minify", "callback_data": "quickcmd:npm_minify"}, {"text": "♻️ OPcache", "callback_data": "quickcmd:cache_flush"}, ], [ {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _automation_menu_keyboard(self) -> dict[str, Any]: rule_map = {rule["rule_key"]: rule for rule in self.orchestrator.list_automation_rules()} ordered_keys = ["ops_daily", "marketing_daily", "seo_weekly", "stack_health", "reactivacion_anunciantes", "seguridad_semanal"] rows: list[list[dict[str, str]]] = [] for rule_key in ordered_keys: rule = rule_map.get(rule_key) if rule is None: continue label = AUTOMATION_LABELS.get(rule_key, rule_key) status = "🟢" if rule.get("enabled") else "⚪" rows.append( [ {"text": f"▶️ {label}", "callback_data": f"autorule:run:{rule_key}"}, {"text": f"{status} {label}", "callback_data": f"autorule:toggle:{rule_key}"}, ] ) rows.append( [ {"text": "🔄 Estado", "callback_data": "menu:automation"}, {"text": "⬅️ Menú", "callback_data": "menu:main"}, ] ) rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _learning_menu_keyboard(self) -> dict[str, Any]: mode = self.orchestrator.get_learning_mode() rows = [ [ {"text": ("🟢 " if mode == "manual" else "⚪ ") + "Manual", "callback_data": "mode:manual"}, {"text": ("🟡 " if mode == "semi" else "⚪ ") + "Semi", "callback_data": "mode:semi"}, {"text": ("🔴 " if mode == "auto" else "⚪ ") + "Auto", "callback_data": "mode:auto"}, ], [ {"text": "🧾 Estado", "callback_data": "menu:learning"}, {"text": "📝 Feedback", "callback_data": "menu:feedback"}, ], [ {"text": "🧠 Memorias", "callback_data": "menu:memories"}, {"text": "🧪 Stack IA", "callback_data": "menu:stack"}, ], [ {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _security_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "🛠️ Ops", "callback_data": "menu:ops"}, {"text": "🧠 Aprendizaje", "callback_data": "menu:learning"}, ], [ {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _stack_menu_keyboard(self) -> dict[str, Any]: rows = [ [ {"text": "🧠 Aprendizaje", "callback_data": "menu:learning"}, {"text": "🚀 Marketing", "callback_data": "menu:marketing"}, ], [ {"text": "⬅️ Menú", "callback_data": "menu:main"}, ], ] rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _proposal_result_keyboard(self, proposal: dict[str, Any]) -> dict[str, Any]: rows = [ [ {"text": "👍 Útil", "callback_data": f"feedback:{proposal['id']}:5"}, {"text": "👌 Bien", "callback_data": f"feedback:{proposal['id']}:4"}, {"text": "🛠️ Mejorar", "callback_data": f"feedback:{proposal['id']}:2"}, ] ] study_id = proposal.get("study_id") if study_id: rows.append([{"text": "🧠 Memorizar estudio", "callback_data": f"memorize:{study_id}"}]) rows.append([{"text": "🎛️ Menú", "callback_data": "menu:main"}]) rows.extend(self._dashboard_rows()) return {"inline_keyboard": rows} def _study_text(self, study: dict[str, Any]) -> str: agents = ', '.join(study.get('recommended_agents', [])) or 'sin clasificar' sep = "\u2500" * 16 analysis = self._trim_text(study.get('analysis', '') or '', 900) return ( "📊 *Estudio generado*\n" f"`{study['id']}`\n" f"{sep}\n" f"📝 {study['summary']}\n" f"{sep}\n" f"🤖 Agentes: `{agents}`\n" f"📡 Canal: `{study.get('channel', 'desconocido')}`\n\n" f"{analysis}" ) def _proposal_text(self, proposal: dict[str, Any]) -> str: risk = proposal.get('risk_level', 'medium') risk_icon = {'low': '🟢', 'medium': '🟡', 'high': '🔴', 'critical': '🚨'}.get(risk, '⚪') status = proposal.get('status', 'pending') status_icon = {'pending': '⏳', 'approved': '✅', 'rejected': '❌', 'executed': '🚀'}.get(status, '❔') payload = proposal.get("payload") or {} payload_line = ", ".join(f"`{k}`={v}" for k, v in list(payload.items())[:4]) or "—" sep = "\u2500" * 16 return ( f"📌 *Propuesta* {status_icon}\n" f"`{proposal['id']}`\n" f"{sep}\n" f"*{proposal['title']}*\n\n" f"{proposal.get('summary', '')}\n" f"{sep}\n" f"🤖 `{proposal.get('agent_name', '?')}` · {risk_icon} `{risk}` · `{proposal.get('action_type','?')}`\n" f"📦 {payload_line}" ) def _proposal_keyboard(self, proposal: dict[str, Any]) -> dict[str, Any] | None: if proposal.get("status") != "pending": return self._proposal_result_keyboard(proposal) keyboard = [ [ {"text": "✅ Aprobar", "callback_data": f"approve:{proposal['id']}"}, {"text": "❌ Rechazar", "callback_data": f"reject:{proposal['id']}"}, ] ] keyboard.append([{"text": "🎛️ Menú", "callback_data": "menu:main"}]) keyboard.extend(self._dashboard_rows()) return {"inline_keyboard": keyboard} def _study_keyboard(self, study_id: str) -> dict[str, Any] | None: keyboard = [[{"text": "🧠 Promover a memoria", "callback_data": f"memorize:{study_id}"}]] keyboard.append([{"text": "🎛️ Menú", "callback_data": "menu:main"}]) keyboard.extend(self._dashboard_rows()) return {"inline_keyboard": keyboard} def _dashboard_keyboard(self) -> dict[str, Any] | None: rows = self._dashboard_rows() if not rows: return None return {"inline_keyboard": rows} def _dashboard_rows(self) -> list[list[dict[str, Any]]]: if not self.settings.public_base_url: return [] return [[{"text": "🖥️ Abrir dashboard", "url": self.settings.public_base_url}]] async def _send_document(self, chat_id: str, content: bytes, filename: str, caption: str = "") -> None: """Envía bytes como fichero adjunto (sendDocument multipart).""" client = await self._get_client() data: dict[str, Any] = {"chat_id": chat_id} if caption: data["caption"] = caption[:1024] try: response = await client.post( self._api_url("sendDocument"), data=data, files={"document": (filename, content, "text/plain; charset=utf-8")}, ) except httpx.HTTPError as exc: raise RuntimeError(self._sanitize_telegram_error(str(exc))) from exc try: resp_data = response.json() except ValueError as exc: raise RuntimeError("Telegram: respuesta no JSON en sendDocument") from exc if not resp_data.get("ok"): desc = self._telegram_error_description(resp_data) or f"HTTP {response.status_code}" raise RuntimeError(self._sanitize_telegram_error(f"sendDocument: {desc}")) async def _send_report_document(self, chat_id: str, kind: str, actor: str) -> None: # noqa: C901 """Genera un informe .md según el tipo solicitado y lo envía como documento.""" kind_aliases: dict[str, str] = { "estado": "estado", "status": "estado", "propuestas": "propuestas", "proposals": "propuestas", "estudios": "estudios", "studies": "estudios", "seguridad": "seguridad", "security": "seguridad", "digest": "digest", "memorias": "memorias", "memories": "memorias", "automatizaciones": "automatizaciones", "auto": "automatizaciones", } normalized = kind_aliases.get(kind.lower(), "estado") now_str = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M") filename = f"supervisor-{normalized}-{stamp}.md" await self._send_message(chat_id, f"📄 Generando `{filename}`…") risk_map = {"low": "🟢 bajo", "medium": "🟡 medio", "high": "🔴 alto"} lines: list[str] = [] if normalized == "estado": counts = self.store.count_summary() bridge_ok = bool(self.settings.chatbot_bridge_url and self.settings.chatbot_admin_token) lines = [ f"# Estado del Supervisor — {now_str}", "", "## Sistema", f"- Proveedor IA: `{self.orchestrator.provider.name}`", f"- Modelo: `{self.settings.ollama_model}`", f"- Modo aprendizaje: `{self.orchestrator.get_learning_mode()}`", f"- Telegram: `{self.settings.telegram_mode}`", f"- Bridge chatbot: {'✅ activo' if bridge_ok else '❌ no configurado'}", f"- Perfil SSH: `{self.settings.default_ssh_profile}`", f"- Dashboard: {self.settings.public_base_url}", "", "## Contadores", "| Métrica | Valor |", "|---------|-------||", f"| Estudios | {counts['studies']} |", f"| Propuestas totales | {counts['proposals']} |", f"| Pendientes | {counts['pending']} |", f"| Ejecutadas | {counts['executed']} |", f"| Rechazadas | {counts.get('rejected', 0)} |", f"| Memorias | {counts['memories']} |", "", "## Automatizaciones", f"- Reglas activas: {counts['automation_enabled']}/{counts['automation_rules']}", f"- Ejecuciones registradas: {counts.get('automation_runs', 0)}", ] elif normalized == "propuestas": pending = self.store.list_proposals(status="pending", limit=60) recent = self.store.list_proposals(limit=20) lines = [ f"# Propuestas — {now_str}", "", f"Total pendientes: **{len(pending)}**", "", "## Pendientes de aprobación", "", ] for p in pending: lines += [ f"### {p.get('title', 'Propuesta')}", f"- **ID**: `{p['id']}`", f"- **Agente**: {p.get('agent_name', '?')}", f"- **Riesgo**: {risk_map.get(p.get('risk_level', ''), '⚪')}", f"- **Tipo**: {p.get('action_type', '?')}", f"- **Creada**: {str(p.get('created_at', ''))[:16]}", "", f"> {p.get('summary', '')}", "", ] lines += ["---", "", "## Últimas 20 propuestas (todos los estados)", ""] for p in recent: lines.append( f"- [{p.get('status','?').upper()}] **{p.get('title','?')}** — " f"{p.get('agent_name','?')} — {str(p.get('created_at',''))[:10]}" ) elif normalized == "estudios": studies = self.store.list_studies(limit=12) lines = [f"# Últimos Estudios — {now_str}", ""] for s in studies: agents = ", ".join(s.get("recommended_agents") or []) or "sin clasificar" lines += [ f"## {str(s.get('summary', 'Estudio'))[:80]}", f"- **ID**: `{s['id']}`", f"- **Canal**: {s.get('channel', '?')}", f"- **Agentes**: {agents}", f"- **Fecha**: {str(s.get('created_at', ''))[:16]}", "", str(s.get("analysis", ""))[:1200], "", "---", "", ] elif normalized == "seguridad": last = self.store.get_latest_security_scan() lines = [f"# Informe de Seguridad — {now_str}", ""] if not last: lines += ["_Sin escaneos disponibles. Usa `/escanear` para lanzar uno._"] else: status_map = {"clean": "LIMPIO ✅", "vulnerabilities_found": "VULNERABILIDADES ⚠️", "error": "ERROR ❌"} lines += [ f"## Escaneo: {str(last.get('started_at', ''))[:16]}", f"- **Estado**: {status_map.get(last.get('status', ''), last.get('status', '?'))}", f"- **Herramientas**: {', '.join(last.get('tools_used', []))}", "", "## Vulnerabilidades", "| Severidad | Nº |", "|-----------|----||", f"| 🔴 Crítico | {last.get('critical', 0)} |", f"| 🟠 Alto | {last.get('high', 0)} |", f"| 🟡 Medio | {last.get('medium', 0)} |", f"| 🟢 Bajo | {last.get('low', 0)} |", f"| **Total CVE** | **{last.get('total_vulnerabilities', 0)}** |", "", f"> {last.get('summary', '')}", ] pip_vulns = ((last.get("results") or {}).get("pip_audit") or {}).get("vulnerabilities") or [] if pip_vulns: lines += ["", "## Detalle CVE", ""] for v in pip_vulns: lines += [ f"### {v.get('vuln_id','?')} — `{v.get('package','?')}` {v.get('version','')}", f"- **Severidad**: {v.get('severity','unknown')}", f"- **Fix**: {', '.join(v.get('fix_versions') or []) or 'pendiente de release'}", f"- **Aliases**: {v.get('aliases','')}", "", f"> {str(v.get('description', ''))[:600]}", "", ] elif normalized == "digest": counts = self.store.count_summary() runs = self.store.list_automation_runs(limit=12) pending = self.store.list_proposals(status="pending", limit=20) mems = self.store.list_memories(limit=6) lines = [ f"# Digest del Supervisor — {now_str}", "", "## Contadores globales", "| Métrica | Valor |", "|---------|-------||", f"| Estudios | {counts.get('studies',0)} |", f"| Propuestas | {counts.get('proposals',0)} |", f"| Pendientes | {counts.get('pending',0)} |", f"| Ejecutadas | {counts.get('executed',0)} |", f"| Memorias | {counts.get('memories',0)} |", f"| Reglas activas | {counts.get('automation_enabled',0)}/{counts.get('automation_rules',0)} |", ] if runs: lines += ["", "## Últimas ejecuciones", ""] for r in runs: icon = "✅" if r.get("status") in {"planned","executed","success"} else "❌" lines.append(f"- {icon} `{r.get('rule_key','?')}` — {str(r.get('created_at',''))[:16]} — {r.get('status','?')}") if pending: lines += ["", "## Propuestas pendientes", ""] for p in pending[:15]: ri = risk_map.get(p.get("risk_level",""), "⚪") lines.append(f"- {ri} **{p.get('title','?')}** — `{p.get('agent_name','?')}` — {str(p.get('created_at',''))[:10]}") if mems: lines += ["", "## Memorias recientes", ""] for m in mems: lines.append(f"- 🧠 {m.get('subject','?')} — {str(m.get('created_at',''))[:10]}") elif normalized == "memorias": mems = self.store.list_memories(limit=50) lines = [ f"# Memorias — {now_str}", "", f"Total memorias aprobadas: **{len(mems)}**", "", ] for m in mems: lines += [ f"## {m.get('subject','Memoria')}", f"- **Actor**: {m.get('actor','?')}", f"- **Estudio origen**: `{str(m.get('study_id','?'))[:8]}`", f"- **Fecha**: {str(m.get('created_at',''))[:16]}", "", "---", "", ] elif normalized == "automatizaciones": rules = self.orchestrator.list_automation_rules() runs = self.store.list_automation_runs(limit=20) lines = [f"# Automatizaciones — {now_str}", "", "## Reglas configuradas", ""] for r in rules: status = "🟢 ACTIVA" if r.get("enabled") else "⚪ inactiva" lines += [ f"### {r.get('title', r.get('rule_key','?'))}", f"- **Clave**: `{r.get('rule_key','?')}`", f"- **Estado**: {status} | **Frecuencia**: {r.get('schedule_hint','?')}", f"- **Última ejecución**: {str(r.get('last_run_at','nunca'))[:16]} | Estado: {r.get('last_run_status','—')}", "", f"> {r.get('description','')}", "", ] if runs: lines += ["---", "", "## Historial de ejecuciones (últimas 20)", ""] for run in runs: icon = "✅" if run.get("status") in {"planned","executed","success"} else "❌" lines.append( f"- {icon} `{run.get('rule_key','?')}` — " f"{str(run.get('created_at',''))[:16]} — " f"{run.get('actor','?')} — {run.get('status','?')}" ) else: lines = [f"# Estado del Supervisor — {now_str}", "", "Tipo de informe no reconocido: usa estado, propuestas, estudios, seguridad, digest, memorias o automatizaciones."] content = "\n".join(lines).encode("utf-8") try: await self._send_document( chat_id, content, filename, caption=f"📄 Informe `{normalized}` generado el {now_str}", ) except Exception as exc: await self._send_message(chat_id, f"❌ No pude enviar el documento: {exc}") async def _send_message(self, chat_id: str, text: str, reply_markup: dict[str, Any] | None = None, parse_mode: str | None = None) -> None: chunks = self._split_text(text) for index, chunk in enumerate(chunks): payload: dict[str, Any] = { "chat_id": chat_id, "text": chunk, "disable_web_page_preview": True, } if parse_mode: payload["parse_mode"] = parse_mode if reply_markup and index == len(chunks) - 1: payload["reply_markup"] = reply_markup await self._api_call("sendMessage", payload) async def _send_chat_action(self, chat_id: str, action: str = "typing") -> None: """Envía 'typing…' o cualquier acción de chat a Telegram (fuego y olvido).""" try: await self._api_call("sendChatAction", {"chat_id": chat_id, "action": action}) except Exception: pass # No crítico si falla async def _answer_callback(self, callback_id: str, text: str, show_alert: bool = False) -> None: try: await self._api_call( "answerCallbackQuery", { "callback_query_id": callback_id, "text": text, "show_alert": show_alert, }, ) except Exception as exc: safe_error = self._sanitize_telegram_error(str(exc)) if self._is_stale_callback_error(safe_error): logger.debug("Callback de Telegram expirado o inválido; se ignora: %s", safe_error) else: logger.warning("No se pudo responder callback de Telegram: %s", safe_error) def _telegram_error_description(self, payload: Any) -> str: if not isinstance(payload, dict): return "" return str(payload.get("description") or payload.get("error") or "").strip() # ------------------------------------------------------------------ # # Voz — faster-whisper # # ------------------------------------------------------------------ # async def _handle_voice_message(self, chat_id: str, voice: dict[str, Any], actor: str) -> None: """Descarga el mensaje de voz, lo transcribe con faster-whisper y lo procesa como texto.""" file_id = voice.get("file_id", "") duration = voice.get("duration", 0) if not file_id: await self._send_message(chat_id, "❌ No encontré el archivo de voz.") return if duration > 120: await self._send_message(chat_id, "⚠️ El audio supera los 2 minutos. Envía un mensaje de voz más corto.") return # Respuesta inmediata para no parecer colgado mientras descarga+transcribe await self._send_message(chat_id, "🎤 Procesando tu mensaje de voz\u2026 un momento.") await self._send_chat_action(chat_id, "typing") tmp_path: str = "" try: # 1. Obtener URL de descarga file_info = await self._api_call("getFile", {"file_id": file_id}) file_path = file_info.get("file_path", "") if isinstance(file_info, dict) else "" if not file_path: await self._send_message(chat_id, "❌ No pude obtener la ruta del archivo de voz.") return base = self.settings.telegram_api_base_url.rstrip("/") # getFile devuelve un path relativo; la URL de descarga usa el dominio de la API api_domain = "https://api.telegram.org" download_url = f"{api_domain}/file/bot{self.settings.telegram_bot_token}/{file_path}" # 2. Descargar a fichero temporal client = await self._get_client() async with client.stream("GET", download_url, timeout=30) as resp: with tempfile.NamedTemporaryFile(suffix=".ogg", delete=False) as tmp: tmp_path = tmp.name async for chunk in resp.aiter_bytes(chunk_size=8192): tmp.write(chunk) # 3. Transcribir en hilo (operación bloqueante) transcript = await asyncio.to_thread(self._transcribe_audio, tmp_path) except Exception as exc: logger.warning("Error procesando voz: %s", exc) await self._send_message(chat_id, "❌ Error al procesar el audio. Inténtalo en texto.") return finally: try: if tmp_path and os.path.exists(tmp_path): os.unlink(tmp_path) except Exception: pass if not transcript: await self._send_message(chat_id, "❌ No pude entender el audio. ¿Puedes repetirlo en texto?") return # 4. Confirmar transcripción + procesar como texto normal await self._send_message(chat_id, f"🎤 *Transcripción:* _{transcript}_", parse_mode="Markdown") self.store.add_audit("telegram.voice.transcribed", {"chat_id": chat_id, "actor": actor, "transcript": transcript[:500]}) # Reutilizar el flujo exacto que procesa texto libre result = await self.orchestrator.create_command_plan(transcript, actor, "telegram") study = result["study"] proposals = result["proposals"] await self._send_message( chat_id, self._study_text(study), reply_markup=self._study_keyboard(study["id"]), ) for proposal in proposals: await self._send_message( chat_id, self._proposal_text(proposal), reply_markup=self._proposal_keyboard(proposal), ) @staticmethod def _transcribe_audio(file_path: str) -> str: """Ejecutar en hilo de trabajo (bloqueante). Devuelve la transcripción o cadena vacía.""" try: from faster_whisper import WhisperModel # type: ignore[import-not-found] except ImportError: logger.warning("faster-whisper no instalado. Instala con: pip install faster-whisper") return "" try: model = WhisperModel("small", device="cpu", compute_type="int8") segments, _ = model.transcribe(file_path, language="es") return " ".join(seg.text for seg in segments).strip() except Exception as exc: logger.warning("faster-whisper transcribe error: %s", exc) return "" def _sanitize_telegram_error(self, text: str) -> str: clean = str(text or "") token = self.settings.telegram_bot_token if token: clean = clean.replace(token, "***TELEGRAM_BOT_TOKEN***") return clean def _is_stale_callback_error(self, text: str) -> bool: lowered = str(text or "").lower() return any( marker in lowered for marker in ( "query is too old", "query id is invalid", "query_id_invalid", "response timeout expired", ) ) def _split_text(self, text: str, limit: int = 3800) -> list[str]: clean = text.strip() if len(clean) <= limit: return [clean] chunks: list[str] = [] remaining = clean while len(remaining) > limit: split_at = remaining.rfind("\n\n", 0, limit) if split_at < 120: split_at = limit chunks.append(remaining[:split_at].strip()) remaining = remaining[split_at:].strip() if remaining: chunks.append(remaining) return chunks def _trim_text(self, text: str, limit: int) -> str: clean = str(text or "").strip() if len(clean) <= limit: return clean return clean[: limit - 1].rstrip() + "…"