from __future__ import annotations import json import sqlite3 import threading import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any def utc_now() -> str: return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z") class SupervisorStore: def __init__(self, db_path: Path) -> None: self.db_path = Path(db_path) self.db_path.parent.mkdir(parents=True, exist_ok=True) self._lock = threading.Lock() self.init_db() def _connect(self) -> sqlite3.Connection: conn = sqlite3.connect(self.db_path, check_same_thread=False) conn.row_factory = sqlite3.Row return conn def init_db(self) -> None: with self._lock, self._connect() as conn: conn.executescript( """ CREATE TABLE IF NOT EXISTS studies ( id TEXT PRIMARY KEY, request_text TEXT NOT NULL, summary TEXT NOT NULL, analysis TEXT NOT NULL, recommended_agents TEXT NOT NULL, actor TEXT NOT NULL, channel TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS proposals ( id TEXT PRIMARY KEY, study_id TEXT NOT NULL, agent_name TEXT NOT NULL, title TEXT NOT NULL, summary TEXT NOT NULL, action_type TEXT NOT NULL, payload TEXT NOT NULL, risk_level TEXT NOT NULL, status TEXT NOT NULL, created_at TEXT NOT NULL, decided_at TEXT, decided_by TEXT, decision_note TEXT, execution_log TEXT, FOREIGN KEY(study_id) REFERENCES studies(id) ); CREATE TABLE IF NOT EXISTS memories ( id INTEGER PRIMARY KEY AUTOINCREMENT, study_id TEXT NOT NULL, subject TEXT NOT NULL, note TEXT NOT NULL, actor TEXT NOT NULL, created_at TEXT NOT NULL, FOREIGN KEY(study_id) REFERENCES studies(id) ); CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL, payload TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS runtime_state ( key TEXT PRIMARY KEY, value TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS training_feedback ( id INTEGER PRIMARY KEY AUTOINCREMENT, proposal_id TEXT, study_id TEXT, actor TEXT NOT NULL, score INTEGER NOT NULL, label TEXT NOT NULL, note TEXT NOT NULL, source TEXT NOT NULL, created_at TEXT NOT NULL, FOREIGN KEY(proposal_id) REFERENCES proposals(id), FOREIGN KEY(study_id) REFERENCES studies(id) ); CREATE TABLE IF NOT EXISTS automation_rules ( rule_key TEXT PRIMARY KEY, title TEXT NOT NULL, description TEXT NOT NULL, prompt_text TEXT NOT NULL, plan_kind TEXT NOT NULL, schedule_hint TEXT NOT NULL, enabled INTEGER NOT NULL, requires_approval INTEGER NOT NULL, last_run_at TEXT, last_run_status TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS automation_runs ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_key TEXT NOT NULL, actor TEXT NOT NULL, status TEXT NOT NULL, study_id TEXT, summary TEXT NOT NULL, created_at TEXT NOT NULL, FOREIGN KEY(rule_key) REFERENCES automation_rules(rule_key) ); CREATE TABLE IF NOT EXISTS inbound_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL, source TEXT NOT NULL, payload TEXT NOT NULL, processed INTEGER NOT NULL DEFAULT 0, study_id TEXT, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS sse_feed ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL, payload TEXT NOT NULL, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS analytics_events ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL, page TEXT NOT NULL DEFAULT '', source TEXT NOT NULL DEFAULT '', actor TEXT NOT NULL DEFAULT '', ip_hash TEXT NOT NULL DEFAULT '', meta TEXT NOT NULL DEFAULT '{}', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS security_scans ( id INTEGER PRIMARY KEY AUTOINCREMENT, scan_id TEXT NOT NULL UNIQUE, actor TEXT NOT NULL, status TEXT NOT NULL, total_vulnerabilities INTEGER NOT NULL DEFAULT 0, critical INTEGER NOT NULL DEFAULT 0, high INTEGER NOT NULL DEFAULT 0, medium INTEGER NOT NULL DEFAULT 0, low INTEGER NOT NULL DEFAULT 0, tools_used TEXT NOT NULL DEFAULT '[]', summary TEXT NOT NULL DEFAULT '', results TEXT NOT NULL DEFAULT '{}', started_at TEXT NOT NULL, finished_at TEXT NOT NULL, created_at TEXT NOT NULL ); """ ) def _loads(self, value: str | None, default: Any) -> Any: if not value: return default try: return json.loads(value) except json.JSONDecodeError: return default def _study_from_row(self, row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return { "id": row["id"], "request_text": row["request_text"], "summary": row["summary"], "analysis": row["analysis"], "recommended_agents": self._loads(row["recommended_agents"], []), "actor": row["actor"], "channel": row["channel"], "created_at": row["created_at"], } def _proposal_from_row(self, row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return { "id": row["id"], "study_id": row["study_id"], "agent_name": row["agent_name"], "title": row["title"], "summary": row["summary"], "action_type": row["action_type"], "payload": self._loads(row["payload"], {}), "risk_level": row["risk_level"], "status": row["status"], "created_at": row["created_at"], "decided_at": row["decided_at"], "decided_by": row["decided_by"], "decision_note": row["decision_note"], "execution_log": row["execution_log"], } def _memory_from_row(self, row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return { "id": row["id"], "study_id": row["study_id"], "subject": row["subject"], "note": row["note"], "actor": row["actor"], "created_at": row["created_at"], } def _feedback_from_row(self, row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return { "id": row["id"], "proposal_id": row["proposal_id"], "study_id": row["study_id"], "actor": row["actor"], "score": int(row["score"]), "label": row["label"], "note": row["note"], "source": row["source"], "created_at": row["created_at"], } def _automation_rule_from_row(self, row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return { "rule_key": row["rule_key"], "title": row["title"], "description": row["description"], "prompt_text": row["prompt_text"], "plan_kind": row["plan_kind"], "schedule_hint": row["schedule_hint"], "enabled": bool(row["enabled"]), "requires_approval": bool(row["requires_approval"]), "last_run_at": row["last_run_at"], "last_run_status": row["last_run_status"], "created_at": row["created_at"], "updated_at": row["updated_at"], } def _automation_run_from_row(self, row: sqlite3.Row | None) -> dict[str, Any] | None: if row is None: return None return { "id": int(row["id"]), "rule_key": row["rule_key"], "actor": row["actor"], "status": row["status"], "study_id": row["study_id"], "summary": row["summary"], "created_at": row["created_at"], } def create_study( self, request_text: str, summary: str, analysis: str, recommended_agents: list[str], actor: str, channel: str, ) -> dict[str, Any]: study_id = uuid.uuid4().hex created_at = utc_now() with self._lock, self._connect() as conn: conn.execute( """ INSERT INTO studies (id, request_text, summary, analysis, recommended_agents, actor, channel, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( study_id, request_text, summary, analysis, json.dumps(recommended_agents, ensure_ascii=False), actor, channel, created_at, ), ) return self.get_study(study_id) # type: ignore[return-value] def get_study(self, study_id: str) -> dict[str, Any] | None: with self._connect() as conn: row = conn.execute("SELECT * FROM studies WHERE id = ?", (study_id,)).fetchone() return self._study_from_row(row) def list_studies(self, limit: int = 50) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT * FROM studies ORDER BY datetime(created_at) DESC LIMIT ?", (limit,), ).fetchall() studies: list[dict[str, Any]] = [] for row in rows: study = self._study_from_row(row) if study is not None: studies.append(study) return studies def create_proposal( self, study_id: str, agent_name: str, title: str, summary: str, action_type: str, payload: dict[str, Any], risk_level: str, status: str = "pending", ) -> dict[str, Any]: proposal_id = uuid.uuid4().hex created_at = utc_now() with self._lock, self._connect() as conn: conn.execute( """ INSERT INTO proposals ( id, study_id, agent_name, title, summary, action_type, payload, risk_level, status, created_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( proposal_id, study_id, agent_name, title, summary, action_type, json.dumps(payload, ensure_ascii=False), risk_level, status, created_at, ), ) return self.get_proposal(proposal_id) # type: ignore[return-value] def get_proposal(self, proposal_id: str) -> dict[str, Any] | None: with self._connect() as conn: row = conn.execute("SELECT * FROM proposals WHERE id = ?", (proposal_id,)).fetchone() return self._proposal_from_row(row) def list_proposals(self, status: str | None = None, limit: int = 100) -> list[dict[str, Any]]: with self._connect() as conn: if status: rows = conn.execute( "SELECT * FROM proposals WHERE status = ? ORDER BY datetime(created_at) DESC LIMIT ?", (status, limit), ).fetchall() else: rows = conn.execute( "SELECT * FROM proposals ORDER BY datetime(created_at) DESC LIMIT ?", (limit,), ).fetchall() proposals: list[dict[str, Any]] = [] for row in rows: proposal = self._proposal_from_row(row) if proposal is not None: proposals.append(proposal) return proposals def decide_proposal(self, proposal_id: str, status: str, actor: str, note: str | None = None) -> dict[str, Any] | None: with self._lock, self._connect() as conn: conn.execute( """ UPDATE proposals SET status = ?, decided_at = ?, decided_by = ?, decision_note = ? WHERE id = ? """, (status, utc_now(), actor, note, proposal_id), ) return self.get_proposal(proposal_id) def set_proposal_execution(self, proposal_id: str, status: str, execution_log: str) -> dict[str, Any] | None: with self._lock, self._connect() as conn: conn.execute( "UPDATE proposals SET status = ?, execution_log = ? WHERE id = ?", (status, execution_log, proposal_id), ) return self.get_proposal(proposal_id) def create_memory_from_study(self, study_id: str, actor: str, note: str | None = None) -> dict[str, Any] | None: study = self.get_study(study_id) if study is None: return None existing = self.get_memory_for_study(study_id) if existing is not None: return existing subject = (study["request_text"] or "Aprendizaje aprobado").strip()[:120] body = note or (study["summary"] + "\n\n" + study["analysis"]).strip() created_at = utc_now() with self._lock, self._connect() as conn: cursor = conn.execute( "INSERT INTO memories (study_id, subject, note, actor, created_at) VALUES (?, ?, ?, ?, ?)", (study_id, subject, body, actor, created_at), ) memory_id = cursor.lastrowid with self._connect() as conn: row = conn.execute("SELECT * FROM memories WHERE id = ?", (memory_id,)).fetchone() return self._memory_from_row(row) def get_memory_for_study(self, study_id: str) -> dict[str, Any] | None: with self._connect() as conn: row = conn.execute( "SELECT * FROM memories WHERE study_id = ? ORDER BY datetime(created_at) DESC LIMIT 1", (study_id,), ).fetchone() return self._memory_from_row(row) def has_memory_for_study(self, study_id: str) -> bool: return self.get_memory_for_study(study_id) is not None def list_memories(self, limit: int = 50) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT * FROM memories ORDER BY datetime(created_at) DESC LIMIT ?", (limit,), ).fetchall() memories: list[dict[str, Any]] = [] for row in rows: memory = self._memory_from_row(row) if memory is not None: memories.append(memory) return memories def create_feedback( self, actor: str, score: int, label: str, note: str, source: str, proposal_id: str | None = None, study_id: str | None = None, ) -> dict[str, Any]: created_at = utc_now() with self._lock, self._connect() as conn: cursor = conn.execute( """ INSERT INTO training_feedback (proposal_id, study_id, actor, score, label, note, source, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, (proposal_id, study_id, actor, score, label, note, source, created_at), ) feedback_id = cursor.lastrowid row = conn.execute("SELECT * FROM training_feedback WHERE id = ?", (feedback_id,)).fetchone() return self._feedback_from_row(row) # type: ignore[return-value] def list_feedback(self, limit: int = 20) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT * FROM training_feedback ORDER BY datetime(created_at) DESC LIMIT ?", (limit,), ).fetchall() feedback_items: list[dict[str, Any]] = [] for row in rows: feedback = self._feedback_from_row(row) if feedback is not None: feedback_items.append(feedback) return feedback_items def feedback_summary(self) -> dict[str, Any]: with self._connect() as conn: total = conn.execute("SELECT COUNT(*) FROM training_feedback").fetchone()[0] avg = conn.execute("SELECT COALESCE(AVG(score), 0) FROM training_feedback").fetchone()[0] positive = conn.execute("SELECT COUNT(*) FROM training_feedback WHERE score >= 4").fetchone()[0] negative = conn.execute("SELECT COUNT(*) FROM training_feedback WHERE score <= 2").fetchone()[0] return { "total": int(total), "avg_score": round(float(avg or 0), 2), "positive": int(positive), "negative": int(negative), } def seed_automation_rules(self, rules: list[dict[str, Any]]) -> None: now = utc_now() with self._lock, self._connect() as conn: for rule in rules: conn.execute( """ INSERT INTO automation_rules ( rule_key, title, description, prompt_text, plan_kind, schedule_hint, enabled, requires_approval, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(rule_key) DO UPDATE SET title = excluded.title, description = excluded.description, prompt_text = excluded.prompt_text, plan_kind = excluded.plan_kind, schedule_hint = excluded.schedule_hint, requires_approval = excluded.requires_approval, updated_at = excluded.updated_at """, ( str(rule["rule_key"]), str(rule["title"]), str(rule["description"]), str(rule["prompt_text"]), str(rule["plan_kind"]), str(rule["schedule_hint"]), 1 if bool(rule.get("enabled", False)) else 0, 1 if bool(rule.get("requires_approval", True)) else 0, now, now, ), ) def get_automation_rule(self, rule_key: str) -> dict[str, Any] | None: with self._connect() as conn: row = conn.execute( "SELECT * FROM automation_rules WHERE rule_key = ?", (rule_key,), ).fetchone() return self._automation_rule_from_row(row) def list_automation_rules(self, limit: int = 50) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT * FROM automation_rules ORDER BY rule_key ASC LIMIT ?", (limit,), ).fetchall() rules: list[dict[str, Any]] = [] for row in rows: rule = self._automation_rule_from_row(row) if rule is not None: rules.append(rule) return rules def set_automation_rule_enabled(self, rule_key: str, enabled: bool) -> dict[str, Any] | None: with self._lock, self._connect() as conn: conn.execute( "UPDATE automation_rules SET enabled = ?, updated_at = ? WHERE rule_key = ?", (1 if enabled else 0, utc_now(), rule_key), ) return self.get_automation_rule(rule_key) def record_automation_rule_run(self, rule_key: str, status: str) -> dict[str, Any] | None: with self._lock, self._connect() as conn: conn.execute( "UPDATE automation_rules SET last_run_at = ?, last_run_status = ?, updated_at = ? WHERE rule_key = ?", (utc_now(), status, utc_now(), rule_key), ) return self.get_automation_rule(rule_key) def add_automation_run( self, rule_key: str, actor: str, status: str, summary: str, study_id: str | None = None, ) -> dict[str, Any]: created_at = utc_now() with self._lock, self._connect() as conn: cursor = conn.execute( """ INSERT INTO automation_runs (rule_key, actor, status, study_id, summary, created_at) VALUES (?, ?, ?, ?, ?, ?) """, (rule_key, actor, status, study_id, summary, created_at), ) run_id = cursor.lastrowid row = conn.execute("SELECT * FROM automation_runs WHERE id = ?", (run_id,)).fetchone() return self._automation_run_from_row(row) # type: ignore[return-value] def list_automation_runs(self, limit: int = 20) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT * FROM automation_runs ORDER BY datetime(created_at) DESC LIMIT ?", (limit,), ).fetchall() runs: list[dict[str, Any]] = [] for row in rows: run = self._automation_run_from_row(row) if run is not None: runs.append(run) return runs def add_audit(self, event_type: str, payload: dict[str, Any]) -> None: with self._lock, self._connect() as conn: conn.execute( "INSERT INTO audit_log (event_type, payload, created_at) VALUES (?, ?, ?)", (event_type, json.dumps(payload, ensure_ascii=False), utc_now()), ) def list_audit_log(self, limit: int = 50) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT id, event_type, payload, created_at FROM audit_log ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [ { "id": r["id"], "event_type": r["event_type"], "payload": self._loads(r["payload"], {}), "created_at": r["created_at"], } for r in rows ] def get_runtime_state(self, key: str, default: Any = None) -> Any: with self._connect() as conn: row = conn.execute("SELECT value FROM runtime_state WHERE key = ?", (key,)).fetchone() if row is None: return default return self._loads(row["value"], default) def set_runtime_state(self, key: str, value: Any) -> None: with self._lock, self._connect() as conn: conn.execute( """ INSERT INTO runtime_state (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = excluded.updated_at """, (key, json.dumps(value, ensure_ascii=False), utc_now()), ) def count_summary(self) -> dict[str, int]: with self._connect() as conn: studies = conn.execute("SELECT COUNT(*) FROM studies").fetchone()[0] proposals = conn.execute("SELECT COUNT(*) FROM proposals").fetchone()[0] pending = conn.execute("SELECT COUNT(*) FROM proposals WHERE status = 'pending'").fetchone()[0] executed = conn.execute("SELECT COUNT(*) FROM proposals WHERE status = 'executed'").fetchone()[0] memories = conn.execute("SELECT COUNT(*) FROM memories").fetchone()[0] automation_rules = conn.execute("SELECT COUNT(*) FROM automation_rules").fetchone()[0] automation_enabled = conn.execute("SELECT COUNT(*) FROM automation_rules WHERE enabled = 1").fetchone()[0] automation_runs = conn.execute("SELECT COUNT(*) FROM automation_runs").fetchone()[0] analytics_total = conn.execute("SELECT COUNT(*) FROM analytics_events").fetchone()[0] rejected = conn.execute("SELECT COUNT(*) FROM proposals WHERE status = 'rejected'").fetchone()[0] return { "studies": int(studies), "proposals": int(proposals), "pending": int(pending), "executed": int(executed), "rejected": int(rejected), "memories": int(memories), "automation_rules": int(automation_rules), "automation_enabled": int(automation_enabled), "automation_runs": int(automation_runs), "analytics_total": int(analytics_total), } # ------------------------------------------------------------------ # # Eventos inbound (PHP portal → supervisor) # # ------------------------------------------------------------------ # def add_inbound_event( self, event_type: str, source: str, payload: dict[str, Any], ) -> int: with self._lock, self._connect() as conn: cursor = conn.execute( "INSERT INTO inbound_events (event_type, source, payload, processed, created_at) VALUES (?, ?, ?, 0, ?)", (event_type, source, json.dumps(payload, ensure_ascii=False), utc_now()), ) return int(cursor.lastrowid) def mark_inbound_event_processed(self, event_id: int, study_id: str | None = None) -> None: with self._lock, self._connect() as conn: conn.execute( "UPDATE inbound_events SET processed = 1, study_id = ? WHERE id = ?", (study_id, event_id), ) def list_inbound_events(self, limit: int = 30, unprocessed_only: bool = False) -> list[dict[str, Any]]: with self._connect() as conn: if unprocessed_only: rows = conn.execute( "SELECT * FROM inbound_events WHERE processed = 0 ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() else: rows = conn.execute( "SELECT * FROM inbound_events ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() result = [] for row in rows: result.append({ "id": row["id"], "event_type": row["event_type"], "source": row["source"], "payload": self._loads(row["payload"], {}), "processed": bool(row["processed"]), "study_id": row["study_id"], "created_at": row["created_at"], }) return result # ------------------------------------------------------------------ # # Analytics de visitas e interacciones # # ------------------------------------------------------------------ # def add_analytics_event( self, event_type: str, page: str = "", source: str = "", actor: str = "anonymous", ip_hash: str = "", meta: dict[str, Any] | None = None, ) -> int: with self._lock, self._connect() as conn: cursor = conn.execute( "INSERT INTO analytics_events (event_type, page, source, actor, ip_hash, meta, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", (event_type, page, source, actor, ip_hash, json.dumps(meta or {}, ensure_ascii=False), utc_now()), ) return int(cursor.lastrowid) def count_analytics_summary(self) -> dict[str, Any]: """Totales, breakdown por tipo y páginas top para el panel de analytics.""" with self._connect() as conn: total = conn.execute("SELECT COUNT(*) FROM analytics_events").fetchone()[0] today = conn.execute( "SELECT COUNT(*) FROM analytics_events WHERE substr(created_at,1,10) = date('now')" ).fetchone()[0] this_week = conn.execute( "SELECT COUNT(*) FROM analytics_events WHERE substr(created_at,1,10) >= date('now', '-7 days')" ).fetchone()[0] by_type_rows = conn.execute( "SELECT event_type, COUNT(*) AS cnt FROM analytics_events GROUP BY event_type ORDER BY cnt DESC LIMIT 10" ).fetchall() top_pages_rows = conn.execute( "SELECT page, COUNT(*) AS cnt FROM analytics_events WHERE page != '' GROUP BY page ORDER BY cnt DESC LIMIT 6" ).fetchall() events_7d_rows = conn.execute( """ SELECT substr(created_at,1,10) AS day, COUNT(*) AS cnt FROM analytics_events WHERE substr(created_at,1,10) >= date('now', '-7 days') GROUP BY day ORDER BY day ASC """ ).fetchall() return { "total": int(total), "today": int(today), "this_week": int(this_week), "by_type": [{"event_type": r["event_type"], "count": r["cnt"]} for r in by_type_rows], "top_pages": [{"page": r["page"], "count": r["cnt"]} for r in top_pages_rows], "events_7d": [{"date": r["day"], "count": r["cnt"]} for r in events_7d_rows], } def list_analytics_events(self, limit: int = 30) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT id, event_type, page, source, actor, created_at FROM analytics_events ORDER BY id DESC LIMIT ?", (limit,), ).fetchall() return [ { "id": row["id"], "event_type": row["event_type"], "page": row["page"], "source": row["source"], "actor": row["actor"], "created_at": row["created_at"], } for row in rows ] # ------------------------------------------------------------------ # # SSE feed (eventos en tiempo real para dashboard) # # ------------------------------------------------------------------ # def push_sse_event(self, event_type: str, payload: dict[str, Any]) -> None: with self._lock, self._connect() as conn: conn.execute( "INSERT INTO sse_feed (event_type, payload, created_at) VALUES (?, ?, ?)", (event_type, json.dumps(payload, ensure_ascii=False), utc_now()), ) def poll_sse_events(self, after_id: int = 0, limit: int = 20) -> list[dict[str, Any]]: with self._connect() as conn: rows = conn.execute( "SELECT * FROM sse_feed WHERE id > ? ORDER BY id ASC LIMIT ?", (after_id, limit), ).fetchall() return [ { "id": row["id"], "event_type": row["event_type"], "payload": self._loads(row["payload"], {}), "created_at": row["created_at"], } for row in rows ] # ------------------------------------------------------------------ # # Transfer de aprendizaje — export / import de memorias # # ------------------------------------------------------------------ # def export_memories(self) -> dict[str, Any]: """Serializa memorias y feedback para transferir aprendizaje a otra instancia.""" with self._connect() as conn: mem_rows = conn.execute( "SELECT subject, note, actor, created_at FROM memories ORDER BY id ASC" ).fetchall() fb_rows = conn.execute( "SELECT actor, score, label, note, source, created_at FROM training_feedback ORDER BY id ASC" ).fetchall() return { "version": 1, "exported_at": utc_now(), "memories": [dict(r) for r in mem_rows], "training_feedback": [dict(r) for r in fb_rows], } def import_memories( self, data: dict[str, Any], actor: str = "import", ) -> dict[str, Any]: """Importa un dump generado por export_memories(). Siempre añade, nunca sobreescribe.""" _IMPORT_STUDY_ID = "import-placeholder" memories: list[dict[str, Any]] = data.get("memories", []) feedback: list[dict[str, Any]] = data.get("training_feedback", []) now = utc_now() imported_mem = 0 imported_fb = 0 with self._lock, self._connect() as conn: for m in memories: subject = str(m.get("subject", ""))[:500] note = str(m.get("note", ""))[:2000] if not subject or not note: continue conn.execute( "INSERT INTO memories (study_id, subject, note, actor, created_at) VALUES (?, ?, ?, ?, ?)", (_IMPORT_STUDY_ID, subject, note, actor, m.get("created_at", now)), ) imported_mem += 1 for fb in feedback: try: score = int(fb.get("score", 0)) except (TypeError, ValueError): score = 0 conn.execute( "INSERT INTO training_feedback " "(proposal_id, study_id, actor, score, label, note, source, created_at) " "VALUES (NULL, NULL, ?, ?, ?, ?, ?, ?)", ( actor, score, str(fb.get("label", ""))[:100], str(fb.get("note", ""))[:500], str(fb.get("source", "import"))[:50], fb.get("created_at", now), ), ) imported_fb += 1 return {"imported_memories": imported_mem, "imported_feedback": imported_fb} # ------------------------------------------------------------------ # # Escaneos de seguridad (pip-audit + trivy) # # ------------------------------------------------------------------ # def save_security_scan(self, scan: dict[str, Any]) -> None: """Persiste el resultado de un escaneo de seguridad completo.""" with self._lock, self._connect() as conn: conn.execute( """INSERT OR REPLACE INTO security_scans (scan_id, actor, status, total_vulnerabilities, critical, high, medium, low, tools_used, summary, results, started_at, finished_at, created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)""", ( scan["scan_id"], scan.get("actor", "sistema"), scan.get("status", "unknown"), scan.get("total_vulnerabilities", 0), scan.get("critical", 0), scan.get("high", 0), scan.get("medium", 0), scan.get("low", 0), json.dumps(scan.get("tools_used", []), ensure_ascii=False), scan.get("summary", "")[:2000], json.dumps(scan.get("results", {}), ensure_ascii=False), scan.get("started_at", utc_now()), scan.get("finished_at", utc_now()), utc_now(), ), ) def list_security_scans(self, limit: int = 20) -> list[dict[str, Any]]: """Devuelve los últimos N escaneos, sin el payload JSON completo de resultados.""" with self._connect() as conn: rows = conn.execute( """SELECT id, scan_id, actor, status, total_vulnerabilities, critical, high, medium, low, tools_used, summary, started_at, finished_at, created_at FROM security_scans ORDER BY id DESC LIMIT ?""", (limit,), ).fetchall() return [ { "id": row["id"], "scan_id": row["scan_id"], "actor": row["actor"], "status": row["status"], "total_vulnerabilities": row["total_vulnerabilities"], "critical": row["critical"], "high": row["high"], "medium": row["medium"], "low": row["low"], "tools_used": self._loads(row["tools_used"], []), "summary": row["summary"], "started_at": row["started_at"], "finished_at": row["finished_at"], "created_at": row["created_at"], } for row in rows ] def get_latest_security_scan(self) -> dict[str, Any] | None: """Devuelve el escaneo más reciente con resultados completos.""" with self._connect() as conn: row = conn.execute( "SELECT * FROM security_scans ORDER BY id DESC LIMIT 1" ).fetchone() if not row: return None return { "id": row["id"], "scan_id": row["scan_id"], "actor": row["actor"], "status": row["status"], "total_vulnerabilities": row["total_vulnerabilities"], "critical": row["critical"], "high": row["high"], "medium": row["medium"], "low": row["low"], "tools_used": self._loads(row["tools_used"], []), "summary": row["summary"], "results": self._loads(row["results"], {}), "started_at": row["started_at"], "finished_at": row["finished_at"], "created_at": row["created_at"], }