from pathlib import Path import re path = Path("grafo_d_server.py") text = path.read_text(encoding="utf-8") # Asegurar imports necesarios. text = text.replace( "from urllib.parse import urlparse\n", "from urllib.parse import urlparse, parse_qs\n" ) text = text.replace( "from urllib.parse import urlparse, parse_qs, parse_qs\n", "from urllib.parse import urlparse, parse_qs\n" ) if "from urllib.error import HTTPError, URLError" not in text: text = text.replace( "from urllib.request import Request, urlopen\n", "from urllib.request import Request, urlopen\nfrom urllib.error import HTTPError, URLError\n" ) HANDLER = r'''class Handler(SimpleHTTPRequestHandler): def _send_json(self, obj, code=200): data = json.dumps(obj, ensure_ascii=False, indent=2).encode("utf-8") self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _payload(self): n = int(self.headers.get("Content-Length", "0")) return json.loads(self.rfile.read(n).decode("utf-8")) if n else {} def do_GET(self): parsed_url = urlparse(self.path) path = parsed_url.path query = parse_qs(parsed_url.query) try: if path in ("/api/autosave/status", "/api/persistence/status"): result = persistence_status_sql() self._send_json(result) return if path == "/api/apis": self._send_json(list_apis()) return if path == "/api/apis/status": graph = "" if query.get("graph"): graph = query.get("graph", [""])[0] result = api_status_sql(graph=graph) self._send_json(result) return db = load_db() if path == "/api/db": self._send_json(db) return if path == "/api/session/save": result = save_session_sql() self._send_json(result) return if path == "/api/events": self._send_json(db.get("events", [])) return if path == "/api/graphs": self._send_json(db.get("graphs", [])) return if path == "/api/alerts": self._send_json(db.get("alerts", [])) return if path == "/api/health": graph_events = [ e for e in db.get("events", []) if str(e.get("row")) != "__AXIS__" ] self._send_json({ "ok": True, "db": str(DB), "events": len(graph_events), "apis": len(list_apis()), "sqlite_integrity": self.sqlite_integrity() }) return except Exception as e: self._send_json({"ok": False, "error": str(e)}, 500) return return super().do_GET() def sqlite_integrity(self): with connect_db() as con: return con.execute("PRAGMA integrity_check").fetchone()[0] def do_POST(self): path = urlparse(self.path).path try: payload = self._payload() if path == "/api/ui-state": result = save_ui_state_payload_sql(payload) render_frontend() self._send_json(result) return if path == "/api/events": result = insert_manual_event_sql(payload) render_frontend() self._send_json({"ok": True, **result}) return if path == "/api/ingest": result = insert_event_sql(payload, allow_upsert=False) render_frontend() self._send_json({"ok": True, **result}) return if path == "/api/folders": result = create_folder_sql(payload) render_frontend() self._send_json(result) return if path == "/api/graphs": result = create_graph_sql(payload) render_frontend() self._send_json(result) return if path == "/api/apis": api = upsert_api(payload) self._send_json({"ok": True, "api": api}) return if path.startswith("/api/apis/") and path.endswith("/sync"): api_id = path.split("/")[-2] result = sync_api(api_id) self._send_json(result) return if path == "/api/render": render_frontend() self._send_json({"ok": True, "rendered": True}) return if path in ("/api/autosave/save", "/api/session/save"): result = autosave_session_sql(payload if "payload" in locals() else {}) self._send_json(result) return self._send_json({"ok": False, "error": "unknown endpoint"}, 404) except ValueError as e: code = 409 if "duplicate" in str(e) else 400 self._send_json({"ok": False, "error": str(e)}, code) except Exception as e: self._send_json({"ok": False, "error": str(e)}, 500) def do_PATCH(self): path = urlparse(self.path).path try: payload = self._payload() if path.startswith("/api/events/"): event_id = path.rsplit("/", 1)[-1] event = update_event_sql(event_id, payload) if not event: self._send_json({"ok": False, "error": "event not found"}, 404) return render_frontend() self._send_json({"ok": True, "event": event}) return if path.startswith("/api/apis/"): api_id = path.rsplit("/", 1)[-1] payload["id"] = api_id api = upsert_api(payload) self._send_json({"ok": True, "api": api}) return self._send_json({"ok": False, "error": "unknown endpoint"}, 404) except Exception as e: self._send_json({"ok": False, "error": str(e)}, 500) def do_DELETE(self): path = urlparse(self.path).path try: if path.startswith("/api/events/"): event_id = path.rsplit("/", 1)[-1] if not delete_event_sql(event_id): self._send_json({"ok": False, "error": "event not found"}, 404) return render_frontend() self._send_json({"ok": True, "deleted": event_id}) return if path.startswith("/api/folders/"): from urllib.parse import unquote folder_name = unquote(path.rsplit("/", 1)[-1]) result = delete_empty_folders_sql([folder_name]) render_frontend() self._send_json(result) return if path.startswith("/api/graphs/"): row_key = path.rsplit("/", 1)[-1] result = delete_graphs_sql([row_key], vacuum=True) render_frontend() self._send_json(result) return if path.startswith("/api/apis/"): api_id = path.rsplit("/", 1)[-1] with connect_db() as con: con.execute("BEGIN IMMEDIATE") cur_logs = con.execute( "DELETE FROM api_ingest_log WHERE api_id = ?", (api_id,) ) cur = con.execute( "DELETE FROM api_registry WHERE id = ?", (api_id,) ) con.commit() render_frontend() self._send_json({ "ok": True, "deleted": api_id, "deleted_apis": cur.rowcount, "deleted_api_logs": cur_logs.rowcount }) return self._send_json({"ok": False, "error": "unknown endpoint"}, 404) except Exception as e: self._send_json({"ok": False, "error": str(e)}, 500) ''' pattern = re.compile( r'\nclass Handler\(SimpleHTTPRequestHandler\):.*?\n\nif __name__ == "__main__":', re.S ) if not pattern.search(text): raise SystemExit("ERROR: no pude localizar class Handler hasta if __name__ == '__main__'.") text = pattern.sub("\n" + HANDLER + '\n\nif __name__ == "__main__":', text) path.write_text(text, encoding="utf-8") print("Handler V55 reparado correctamente.")