#!/usr/bin/env python3 import hashlib import json import sqlite3 import subprocess import sys import threading import time import uuid from datetime import datetime, timezone from pathlib import Path from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler from urllib.parse import urlparse, parse_qs from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError from grafo_d_sqlite_adapter import load_database ROOT = Path(__file__).resolve().parent DB = ROOT / "DATABASE" / "GRAFO_D_DATABASE.sqlite" RENDER = ROOT / "render_excel_like.py" ALLOWED_UI_KEYS = {"folders", "event_meta", "folder_meta", "graph_set_cards"} API_SYNC_TIMEOUT_SECONDS = 15 _api_worker_started = False _render_lock = threading.Lock() def now_iso(): return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) def connect_db(): con = sqlite3.connect(DB, timeout=30) con.row_factory = sqlite3.Row con.execute("PRAGMA foreign_keys = ON") con.execute("PRAGMA journal_mode = WAL") con.execute("PRAGMA busy_timeout = 5000") return con def ensure_runtime_schema(): """Idempotent schema extension inside the same canonical SQLite database.""" with connect_db() as con: con.execute(""" CREATE TABLE IF NOT EXISTS ui_state ( key TEXT PRIMARY KEY, value_json TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ) """) con.execute(""" CREATE TABLE IF NOT EXISTS api_ingest_log ( id INTEGER PRIMARY KEY, api_id TEXT NOT NULL, event_legacy_id TEXT NOT NULL, action TEXT NOT NULL, created_at TEXT NOT NULL, UNIQUE(api_id, event_legacy_id) ) """) repair_malformed_api_events(con) con.commit() def load_db(): ensure_runtime_schema() return load_database() def render_frontend(block=False): def _run(): if not RENDER.exists(): return with _render_lock: subprocess.run([sys.executable, str(RENDER)], cwd=str(ROOT), check=False) if block: _run() else: threading.Thread(target=_run, daemon=True).start() def epoch_to_utc_parts(value): """Return (iso_datetime, yyyy-mm-dd, year, month, day) for valid Unix epoch seconds/milliseconds.""" if value is None: return None raw = str(value).strip() if not raw: return None try: number = float(raw) except Exception: return None # Accept seconds or milliseconds; reject ordinary years and non-epoch identifiers. if number > 10_000_000_000: number = number / 1000.0 if number < 946684800 or number > 4102444800: # 2000-01-01 .. 2100-01-01 UTC return None dt = datetime.fromtimestamp(number, tz=timezone.utc) return ( dt.strftime("%Y-%m-%dT%H:%M:%SZ"), dt.strftime("%Y-%m-%d"), dt.year, dt.month, dt.day, ) def first_present(mapping, keys, default=""): for key in keys: if isinstance(mapping, dict) and mapping.get(key) not in (None, ""): return mapping.get(key) return default def extract_lat_lon(item): """Extract latitude/longitude from common API-center and ISS payload shapes.""" if not isinstance(item, dict): return None, None candidates = [item] for key in ("iss_position", "position", "location", "coordinates", "geo"): val = item.get(key) if isinstance(val, dict): candidates.append(val) lat_keys = ("latitude", "lat") lon_keys = ("longitude", "lon", "lng", "long") for src in candidates: lat = first_present(src, lat_keys, None) lon = first_present(src, lon_keys, None) if lat is not None and lon is not None: return str(lat), str(lon) coords = item.get("coordinates") if isinstance(coords, (list, tuple)) and len(coords) >= 2: # GeoJSON normally stores [longitude, latitude]. return str(coords[1]), str(coords[0]) return None, None def build_place(item, lat, lon): explicit = first_present(item, ("place", "location_name", "city", "country", "region"), "") if explicit: return str(explicit) loc = item.get("location") if isinstance(item, dict) else None if isinstance(loc, str) and loc.strip(): return loc.strip() if lat is not None and lon is not None: return f"Low Earth Orbit / lat {lat}, lon {lon}" return "" def is_iss_payload(api_row, item): api_name = str(api_row.get("name") or api_row.get("id") or "").upper() return "ISS" in api_name or (isinstance(item, dict) and "iss_position" in item) def repair_malformed_api_events(con): """ Remove malformed generic ISS rows produced by the earlier API mapper. They are identifiable by an epoch value parsed as a year plus empty place. Valid historical ISS_POS_* rows are not touched. """ con.execute( """ DELETE FROM events WHERE COALESCE(place, '') = '' AND COALESCE(date_year, 0) > 9999 AND ( UPPER(COALESCE(event_name, '')) LIKE '%ISS-LIVE-LOCATION%' OR UPPER(COALESCE(api_text, '')) LIKE '%ISS-LIVE-LOCATION%' OR UPPER(COALESCE(who, '')) = 'ISS-LIVE-LOCATION' ) """ ) def parse_event_date(event): raw = str(event.get("date") or event.get("date_text") or event.get("timestamp") or event.get("time") or event.get("year") or "").strip() import re epoch = epoch_to_utc_parts(raw) if epoch: iso_datetime, date_text, year, month, day = epoch event["datetime"] = event.get("datetime") or iso_datetime event["date"] = date_text event["year"] = year event["month"] = month event["day"] = day return year, month, day m = re.match(r"^(-?\d{1,6})(?:-(\d{1,2}))?(?:-(\d{1,2}))?$", raw) if m: year = int(m.group(1)) month = int(m.group(2)) if m.group(2) is not None else 1 day = int(m.group(3)) if m.group(3) is not None else 1 else: nums = [int(x) for x in re.findall(r"-?\d+", raw)] if not nums: name_nums = [int(x) for x in re.findall(r"-?\d+", str(event.get("name", "")))] nums = name_nums[:1] year = nums[0] if len(nums) >= 1 else None month = nums[1] if len(nums) >= 2 else 1 day = nums[2] if len(nums) >= 3 else 1 if year is not None: event["year"] = year event["month"] = month event["day"] = day if not event.get("date"): event["date"] = f"{year:04d}-{month:02d}-{day:02d}" if year > 0 else f"{year}-{month:02d}-{day:02d}" return year, month, day def normalize_event(payload): event = dict(payload or {}) event.setdefault("id", event.get("event_id") or event.get("legacy_id") or str(uuid.uuid4())) event.setdefault("name", event.get("title") or event.get("label") or event.get("event_name") or "Untitled event") event.setdefault("row", event.get("graph") or event.get("graph_id") or event.get("row_key") or "default") parse_event_date(event) event.setdefault("start_col", event.get("start") or 1) event.setdefault("end_col", event.get("end") or event["start_col"]) event["start_col"] = int(event["start_col"]) event["end_col"] = int(event["end_col"]) event["updated_at"] = now_iso() event.setdefault("created_at", event["updated_at"]) return event def get_or_create_folder(con, name): folder_name = str(name or "Uncategorized").strip() or "Uncategorized" ts = now_iso() row = con.execute("SELECT id FROM folders WHERE name = ?", (folder_name,)).fetchone() if row: return int(row["id"]) cur = con.execute( "INSERT INTO folders(name, is_open, created_at, updated_at) VALUES (?, 1, ?, ?)", (folder_name, ts, ts), ) return int(cur.lastrowid) def get_or_create_graph(con, row_key, visible_name=None, folder_name="Uncategorized"): key = str(row_key or "default").strip() or "default" ts = now_iso() row = con.execute("SELECT id FROM graphs WHERE row_key = ?", (key,)).fetchone() if row: return int(row["id"]) folder_id = get_or_create_folder(con, folder_name) cur = con.execute( """ INSERT INTO graphs(row_key, visible_name, folder_id, color, created_at, updated_at) VALUES (?, ?, ?, NULL, ?, ?) """, (key, str(visible_name or key), folder_id, ts, ts), ) return int(cur.lastrowid) def max_sort_index(con, graph_id): row = con.execute("SELECT COALESCE(MAX(sort_index), 0) AS n FROM events WHERE graph_id = ?", (graph_id,)).fetchone() return int(row["n"] or 0) def event_to_sql_values(con, event): row_key = str(event.get("row") or event.get("row_key") or event.get("graph") or "default") graph_id = get_or_create_graph(con, row_key, row_key, event.get("group") or event.get("folder") or "Uncategorized") year, month, day = parse_event_date(event) date_text = event.get("date") or event.get("date_text") if not date_text and year is not None: date_text = f"{year:04d}-{int(month or 1):02d}-{int(day or 1):02d}" start_col = int(event.get("start_col") or event.get("start") or max_sort_index(con, graph_id) + 1) end_col = int(event.get("end_col") or event.get("end") or start_col) duration = int(event.get("duration") or max(1, end_col - start_col + 1)) sort_index = int(event.get("sort_index") or start_col) ts = now_iso() return { "legacy_id": str(event.get("id") or event.get("legacy_id") or uuid.uuid4()), "graph_id": graph_id, "row_key": row_key, "event_name": str(event.get("name") or event.get("event_name") or "Untitled event"), "date_text": str(date_text or ""), "date_year": year, "date_month": month, "date_day": day, "start_col": start_col, "end_col": end_col, "duration": duration, "place": str(event.get("place") or event.get("PLACE") or ""), "who": str(event.get("who") or event.get("WHO") or event.get("actor") or ""), "brief_description": str( event.get("brief_description") or event.get("brief") or event.get("BRIEF") or event.get("description") or "" ), "source_link": str(event.get("source_link") or event.get("url") or ""), "api_text": str(event.get("api") or event.get("api_text") or ""), "sort_index": sort_index, "updated_at": ts, } def smart_parse_date(value): """ Deterministic smart date parser for manual UI/API events. Accepted: - -600 - -600-01 - -600-01-01 - 1989 - 1989-01 - 1989-01-01 - 01/01/1989 - 1/1/1989 - 01-01-1989 - 1989/01/01 - June 1989 - 4 June 1989 - Junio 1989 - 4 Junio 1989 """ import re from datetime import date raw = str(value or "").strip() if not raw: raise ValueError("date is required") txt = raw.lower().strip() txt = txt.replace(",", " ") txt = re.sub(r"\s+", " ", txt) months = { "jan": 1, "january": 1, "ene": 1, "enero": 1, "feb": 2, "february": 2, "febrero": 2, "mar": 3, "march": 3, "marzo": 3, "apr": 4, "april": 4, "abr": 4, "abril": 4, "may": 5, "mayo": 5, "jun": 6, "june": 6, "junio": 6, "jul": 7, "july": 7, "julio": 7, "aug": 8, "august": 8, "ago": 8, "agosto": 8, "sep": 9, "sept": 9, "september": 9, "septiembre": 9, "oct": 10, "october": 10, "octubre": 10, "nov": 11, "november": 11, "noviembre": 11, "dec": 12, "december": 12, "dic": 12, "diciembre": 12, } def format_year(y): y = int(y) return f"{y:04d}" if y > 0 else str(y) def is_leap_year(y): y = int(y) return (y % 4 == 0) and (y % 100 != 0 or y % 400 == 0) def check(y, m=None, d=None): y = int(y) if y == 0 or y < -999999 or y > 999999: raise ValueError("year out of range") if m is None: return str(y), y, None, None m = int(m) if m < 1 or m > 12: raise ValueError("month out of range") if d is None: return f"{format_year(y)}-{m:02d}", y, m, None d = int(d) if y > 0: date(y, m, d) else: month_days = [31, 29 if is_leap_year(y) else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31] if d < 1 or d > month_days[m - 1]: raise ValueError("day out of range") return f"{format_year(y)}-{m:02d}-{d:02d}", y, m, d m = re.fullmatch(r"(-?\d{1,6})", txt) if m: return check(m.group(1)) m = re.fullmatch(r"(-?\d{1,6})[-/](\d{1,2})", txt) if m: return check(m.group(1), m.group(2)) m = re.fullmatch(r"(-?\d{1,6})[-/](\d{1,2})[-/](\d{1,2})", txt) if m: return check(m.group(1), m.group(2), m.group(3)) m = re.fullmatch(r"(\d{1,2})[-/](\d{1,2})[-/](-?\d{1,6})", txt) if m: return check(m.group(3), m.group(2), m.group(1)) m = re.fullmatch(r"([a-záéíóúñ]+)\s+(-?\d{1,6})", txt) if m and m.group(1) in months: return check(m.group(2), months[m.group(1)]) m = re.fullmatch(r"(\d{1,2})\s+([a-záéíóúñ]+)\s+(-?\d{1,6})", txt) if m and m.group(2) in months: return check(m.group(3), months[m.group(2)], m.group(1)) m = re.fullmatch(r"([a-záéíóúñ]+)\s+(\d{1,2})\s+(-?\d{1,6})", txt) if m and m.group(1) in months: return check(m.group(3), months[m.group(1)], m.group(2)) raise ValueError("unsupported date format: " + raw) def insert_manual_event_sql(payload): """ Insert one manually-created event into canonical SQLite using smart date parsing. This is used by POST /api/events. """ ensure_runtime_schema() data = dict(payload or {}) row_key = str( data.get("row_key") or data.get("row") or data.get("graph") or data.get("graph_row") or "" ).strip() if not row_key: raise ValueError("row_key/row/graph is required") event_name = str( data.get("event_name") or data.get("name") or data.get("title") or "" ).strip() if not event_name: raise ValueError("event name is required") raw_date = str(data.get("date") or data.get("date_text") or "").strip() date_text, date_year, date_month, date_day = smart_parse_date(raw_date) place = str(data.get("place") or data.get("PLACE") or "").strip() who = str(data.get("who") or data.get("WHO") or "").strip() brief_description = str( data.get("brief_description") or data.get("brief") or data.get("BRIEF") or data.get("description") or "" ).strip() source_link = str(data.get("source_link") or data.get("url") or "").strip() api_text = str(data.get("api_text") or data.get("source") or "manual-ui").strip() ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") graph = con.execute( "SELECT id, visible_name FROM graphs WHERE row_key = ?", (row_key,), ).fetchone() if graph: graph_id = int(graph["id"]) else: graph_id = get_or_create_graph(con, row_key, row_key, "Uncategorized") sort_row = con.execute( "SELECT COALESCE(MAX(sort_index), 0) + 1 AS next_sort FROM events WHERE graph_id = ?", (graph_id,), ).fetchone() sort_index = int(sort_row["next_sort"] or 1) legacy_id = f"{row_key}:{event_name}:{date_text}:{sort_index}" cur = con.execute( """ INSERT INTO events( legacy_id, graph_id, row_key, event_name, date_text, date_year, date_month, date_day, start_col, end_col, duration, place, who, brief_description, source_link, api_text, sort_index, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( legacy_id, graph_id, row_key, event_name, date_text, date_year, date_month, date_day, 1, 1, 1, place, who, brief_description, source_link, api_text, sort_index, ts, ts, ), ) con.commit() return { "ok": True, "id": cur.lastrowid, "row_key": row_key, "event_name": event_name, "date_text": date_text, "date_year": date_year, "date_month": date_month, "date_day": date_day, } def update_event_card_sql(payload): """ Update canonical event fields from the right-side technical card. Matching policy: 1. legacy_id/id if present. 2. row_key + event_name + date_text if legacy id is unavailable. 3. row_key + event_name fallback. This updates SQLite canonical columns: event_name, date_text/year/month/day, place, who, brief_description, source_link, api_text. """ ensure_runtime_schema() data = dict(payload or {}) legacy_id = str(data.get("id") or data.get("legacy_id") or "").strip() row_key = str(data.get("row_key") or data.get("row") or data.get("graph") or "").strip() event_name = str(data.get("event_name") or data.get("name") or data.get("title") or "").strip() raw_date = str(data.get("date") or data.get("date_text") or "").strip() date_text = None date_year = None date_month = None date_day = None if raw_date: date_text, date_year, date_month, date_day = smart_parse_date(raw_date) place = str(data.get("place") or "").strip() who = str(data.get("who") or "").strip() brief_description = str( data.get("brief_description") or data.get("description") or "" ).strip() source_link = str(data.get("source_link") or data.get("url") or "").strip() api_text = str(data.get("api_text") or data.get("api") or "").strip() ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") row = None if legacy_id: row = con.execute( "SELECT id, legacy_id, row_key, event_name, date_text FROM events WHERE legacy_id = ?", (legacy_id,), ).fetchone() if row is None and row_key and event_name and date_text: row = con.execute( """ SELECT id, legacy_id, row_key, event_name, date_text FROM events WHERE row_key = ? AND event_name = ? AND date_text = ? ORDER BY id DESC LIMIT 1 """, (row_key, event_name, date_text), ).fetchone() if row is None and row_key and event_name: row = con.execute( """ SELECT id, legacy_id, row_key, event_name, date_text FROM events WHERE row_key = ? AND event_name = ? ORDER BY id DESC LIMIT 1 """, (row_key, event_name), ).fetchone() if row is None: raise ValueError("event not found for canonical update") event_id = int(row["id"]) current = con.execute( """ SELECT event_name, date_text, date_year, date_month, date_day, place, who, brief_description, source_link, api_text FROM events WHERE id = ? """, (event_id,), ).fetchone() new_event_name = event_name or current["event_name"] new_date_text = date_text if date_text is not None else current["date_text"] new_date_year = date_year if date_text is not None else current["date_year"] new_date_month = date_month if date_text is not None else current["date_month"] new_date_day = date_day if date_text is not None else current["date_day"] con.execute( """ UPDATE events SET event_name = ?, date_text = ?, date_year = ?, date_month = ?, date_day = ?, place = ?, who = ?, brief_description = ?, source_link = ?, api_text = ?, updated_at = ? WHERE id = ? """, ( new_event_name, new_date_text, new_date_year, new_date_month, new_date_day, place, who, brief_description, source_link, api_text or current["api_text"], ts, event_id, ), ) saved = con.execute( """ SELECT id, legacy_id, row_key, event_name, date_text, date_year, date_month, date_day, place, who, brief_description, source_link, api_text FROM events WHERE id = ? """, (event_id,), ).fetchone() con.commit() try: record_persistence_audit("event-card-save", dirty=0, status="ok") except Exception: pass return { "ok": True, "event": dict(saved), } def insert_event_sql(event, allow_upsert=False): ensure_runtime_schema() norm = normalize_event(event) with connect_db() as con: con.execute("BEGIN IMMEDIATE") values = event_to_sql_values(con, norm) existing = con.execute( "SELECT id FROM events WHERE legacy_id = ?", (values["legacy_id"],) ).fetchone() if existing and not allow_upsert: con.rollback() raise ValueError("duplicate event id") if existing: con.execute( """ UPDATE events SET graph_id=:graph_id, row_key=:row_key, event_name=:event_name, date_text=:date_text, date_year=:date_year, date_month=:date_month, date_day=:date_day, start_col=:start_col, end_col=:end_col, duration=:duration, place=:place, who=:who, brief_description=:brief_description, source_link=:source_link, api_text=:api_text, sort_index=:sort_index, updated_at=:updated_at WHERE legacy_id=:legacy_id """, values, ) action = "updated" else: values["created_at"] = norm.get("created_at") or now_iso() con.execute( """ INSERT INTO events( legacy_id, graph_id, row_key, event_name, date_text, date_year, date_month, date_day, start_col, end_col, duration, place, who, brief_description, source_link, api_text, sort_index, created_at, updated_at ) VALUES( :legacy_id, :graph_id, :row_key, :event_name, :date_text, :date_year, :date_month, :date_day, :start_col, :end_col, :duration, :place, :who, :brief_description, :source_link, :api_text, :sort_index, :created_at, :updated_at ) """, values, ) action = "inserted" con.commit() return {"action": action, "event": norm} def update_event_sql(event_id, patch): ensure_runtime_schema() with connect_db() as con: con.execute("BEGIN IMMEDIATE") row = con.execute( "SELECT * FROM events WHERE legacy_id = ? OR CAST(id AS TEXT) = ?", (str(event_id), str(event_id)), ).fetchone() if not row: con.rollback() return None current = { "id": row["legacy_id"] or str(row["id"]), "row": row["row_key"], "name": row["event_name"], "date": row["date_text"] or "", "year": row["date_year"], "month": row["date_month"], "day": row["date_day"], "start_col": row["start_col"], "end_col": row["end_col"], "duration": row["duration"], "place": row["place"], "who": row["who"], "description": row["brief_description"], "source_link": row["source_link"], "api_text": row["api_text"], "sort_index": row["sort_index"], } current.update(patch or {}) values = event_to_sql_values(con, normalize_event(current)) values["db_id"] = row["id"] con.execute( """ UPDATE events SET graph_id=:graph_id, row_key=:row_key, event_name=:event_name, date_text=:date_text, date_year=:date_year, date_month=:date_month, date_day=:date_day, start_col=:start_col, end_col=:end_col, duration=:duration, place=:place, who=:who, brief_description=:brief_description, source_link=:source_link, api_text=:api_text, sort_index=:sort_index, updated_at=:updated_at WHERE id=:db_id """, values, ) con.commit() return current def delete_event_sql(event_id): ensure_runtime_schema() with connect_db() as con: con.execute("BEGIN IMMEDIATE") cur = con.execute( "DELETE FROM events WHERE legacy_id = ? OR CAST(id AS TEXT) = ?", (str(event_id), str(event_id)), ) con.commit() return cur.rowcount > 0 def cleanup_ui_state_after_graph_delete(con, deleted_rows): if not deleted_rows: return deleted = {str(x) for x in deleted_rows} rows = con.execute("SELECT key, value_json FROM ui_state").fetchall() for row in rows: key = row["key"] try: value = json.loads(row["value_json"]) except Exception: continue changed = False if key == "folders" and isinstance(value, dict): for k in ("deletedRows", "deleteRows"): if isinstance(value.get(k), list): new_list = [x for x in value[k] if str(x) not in deleted] if new_list != value[k]: value[k] = new_list changed = True if isinstance(value.get("graphFolder"), dict): for graph_key in list(value["graphFolder"].keys()): if str(graph_key) in deleted: del value["graphFolder"][graph_key] changed = True if key in ("event_meta", "graph_set_cards", "folder_meta") and isinstance(value, dict): for meta_key in list(value.keys()): if any(str(meta_key).startswith(d + ":") or str(meta_key) == d for d in deleted): del value[meta_key] changed = True if changed: con.execute( "UPDATE ui_state SET value_json = ?, updated_at = ? WHERE key = ?", (json.dumps(value, ensure_ascii=False, sort_keys=True), now_iso(), key), ) def delete_empty_folders_sql(folder_names): """ Physically delete explicitly selected empty folders from SQLite. Rules: - Only requested folders are evaluated. - Uncategorized is protected. - Non-empty folders are rejected. - ui_state is cleaned only for deleted folders. """ ensure_runtime_schema() from urllib.parse import unquote names = [] for x in folder_names or []: name = unquote(str(x or "")).strip() if name and name not in names: names.append(name) deleted = [] skipped = [] ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") for name in names: if name == "Uncategorized": skipped.append({"folder": name, "reason": "protected"}) continue row = con.execute( "SELECT id, name FROM folders WHERE name = ?", (name,), ).fetchone() if not row: skipped.append({"folder": name, "reason": "not found"}) continue graph_count = con.execute( "SELECT COUNT(*) FROM graphs WHERE folder_id = ?", (row["id"],), ).fetchone()[0] if graph_count: skipped.append({ "folder": name, "reason": "not empty", "graphs": graph_count, }) continue con.execute("DELETE FROM folders WHERE id = ?", (row["id"],)) deleted.append(name) row = con.execute( "SELECT value_json FROM ui_state WHERE key='folders'" ).fetchone() if row: try: state = json.loads(row["value_json"]) except Exception: state = {} else: state = {} if isinstance(state, dict): deleted_set = set(deleted) state["folders"] = [ f for f in state.get("folders", []) if str(f) not in deleted_set ] open_state = state.get("open", {}) if isinstance(open_state, dict): for f in list(open_state.keys()): if f in deleted_set: del open_state[f] state["open"] = open_state graph_folder = state.get("graphFolder", {}) if isinstance(graph_folder, dict): for row_key, folder in list(graph_folder.items()): if folder in deleted_set: graph_folder[row_key] = "Uncategorized" state["graphFolder"] = graph_folder con.execute( """ INSERT INTO ui_state(key, value_json, created_at, updated_at) VALUES ('folders', ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value_json=excluded.value_json, updated_at=excluded.updated_at """, (json.dumps(state, ensure_ascii=False, sort_keys=True), ts, ts), ) con.commit() try: record_persistence_audit("delete-empty-folder", dirty=0, status="ok") except Exception: pass return { "ok": True, "requested": names, "deleted": deleted, "skipped": skipped, } def delete_graphs_sql(row_keys, vacuum=True): ensure_runtime_schema() if isinstance(row_keys, str): row_keys = [row_keys] row_keys = [str(x).strip() for x in (row_keys or []) if str(x).strip()] if not row_keys: return { "ok": True, "deleted_graphs": 0, "deleted_events": 0, "deleted_apis": 0, "deleted_api_logs": 0, "rows": [], } deleted_graphs = 0 deleted_events = 0 deleted_apis = 0 deleted_api_logs = 0 physically_deleted_rows = [] with connect_db() as con: con.execute("BEGIN IMMEDIATE") for row_key in row_keys: graph = con.execute( "SELECT id, row_key FROM graphs WHERE row_key = ?", (row_key,), ).fetchone() if not graph: continue graph_id = int(graph["id"]) api_rows = con.execute( "SELECT id FROM api_registry WHERE target_graph = ? OR name = ?", (row_key, row_key), ).fetchall() api_ids = [str(r["id"]) for r in api_rows] for api_id in api_ids: cur = con.execute( "DELETE FROM api_ingest_log WHERE api_id = ?", (api_id,), ) deleted_api_logs += cur.rowcount for api_id in api_ids: cur = con.execute( "DELETE FROM api_registry WHERE id = ?", (api_id,), ) deleted_apis += cur.rowcount cur = con.execute( "DELETE FROM events WHERE graph_id = ? OR row_key = ?", (graph_id, row_key), ) deleted_events += cur.rowcount cur = con.execute( "DELETE FROM graphs WHERE id = ?", (graph_id,), ) deleted_graphs += cur.rowcount if cur.rowcount: physically_deleted_rows.append(row_key) cleanup_ui_state_after_graph_delete(con, physically_deleted_rows) con.commit() if vacuum: with connect_db() as con: con.execute("VACUUM") return { "ok": True, "deleted_graphs": deleted_graphs, "deleted_events": deleted_events, "deleted_apis": deleted_apis, "deleted_api_logs": deleted_api_logs, "rows": physically_deleted_rows, } def slugify_graph_key(name): raw = str(name or "").strip() if not raw: raw = "NEW GRAPH" out = [] prev_us = False for ch in raw.upper(): if ch.isalnum(): out.append(ch) prev_us = False else: if not prev_us: out.append("_") prev_us = True return ("".join(out).strip("_") or "NEW_GRAPH") def create_folder_sql(payload): """ Create a folder physically in SQLite. Canonical policy: - folders table is the source of truth. - ui_state is synchronized for UI open/visibility state. - no graph and no event is created. """ ensure_runtime_schema() data = dict(payload or {}) name = str(data.get("name") or data.get("folder") or "").strip() if not name: raise ValueError("folder name is required") ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") con.execute( """ INSERT OR IGNORE INTO folders(name, is_open, created_at, updated_at) VALUES (?, 1, ?, ?) """, (name, ts, ts), ) con.execute( """ UPDATE folders SET is_open = 1, updated_at = ? WHERE name = ? """, (ts, name), ) row = con.execute( "SELECT value_json FROM ui_state WHERE key='folders'" ).fetchone() if row: try: state = json.loads(row["value_json"]) except Exception: state = {} else: state = {} if not isinstance(state, dict): state = {} state.setdefault("folders", []) state.setdefault("graphFolder", {}) state.setdefault("open", {}) state.setdefault("deletedRows", []) state.setdefault("deleteRows", []) if "Uncategorized" not in state["folders"]: state["folders"].insert(0, "Uncategorized") if name not in state["folders"]: state["folders"].append(name) state["open"][name] = True con.execute( """ INSERT INTO ui_state(key, value_json, created_at, updated_at) VALUES ('folders', ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value_json=excluded.value_json, updated_at=excluded.updated_at """, (json.dumps(state, ensure_ascii=False, sort_keys=True), ts, ts), ) con.commit() try: record_persistence_audit("create-folder", dirty=0, status="ok") except Exception: pass return {"ok": True, "folder": name} def sync_ui_folders_to_sql(): """ Convert folders present in ui_state into canonical rows in folders. This makes Save meaningful for folders that were created by older UI code before POST /api/folders existed. """ ensure_runtime_schema() ts = now_iso() with connect_db() as con: row = con.execute( "SELECT value_json FROM ui_state WHERE key='folders'" ).fetchone() if not row: return {"ok": True, "synced": 0} try: state = json.loads(row["value_json"]) except Exception: return {"ok": False, "synced": 0, "error": "invalid ui_state folders json"} if not isinstance(state, dict): return {"ok": False, "synced": 0, "error": "ui_state folders is not object"} folders = state.get("folders", []) if not isinstance(folders, list): return {"ok": False, "synced": 0, "error": "ui_state folders list missing"} synced = 0 con.execute("BEGIN IMMEDIATE") for name in folders: name = str(name or "").strip() if not name: continue before = con.execute( "SELECT id FROM folders WHERE name = ?", (name,), ).fetchone() con.execute( """ INSERT OR IGNORE INTO folders(name, is_open, created_at, updated_at) VALUES (?, 1, ?, ?) """, (name, ts, ts), ) con.execute( "UPDATE folders SET is_open = 1, updated_at = ? WHERE name = ?", (ts, name), ) if before is None: synced += 1 con.commit() return {"ok": True, "synced": synced} def create_graph_sql(payload): """ Create a graph/container physically in SQLite. Policy: - A graph is a container/row. - Creating a graph must NOT create timeline events. - Timeline components are created later by manual event insertion or API ingest. """ ensure_runtime_schema() data = dict(payload or {}) name = str(data.get("name") or data.get("visible_name") or "").strip() if not name: raise ValueError("graph name is required") folder_name = str(data.get("folder") or "Uncategorized").strip() or "Uncategorized" row_key = str(data.get("row_key") or slugify_graph_key(name)).strip() ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") base = row_key candidate = base n = 2 while con.execute("SELECT id FROM graphs WHERE row_key = ?", (candidate,)).fetchone(): candidate = f"{base}_{n}" n += 1 row_key = candidate graph_id = get_or_create_graph(con, row_key, name, folder_name) row = con.execute( "SELECT value_json FROM ui_state WHERE key = 'folders'" ).fetchone() if row: try: state = json.loads(row["value_json"]) except Exception: state = {} else: state = {} state.setdefault("folders", []) state.setdefault("graphFolder", {}) state.setdefault("open", {}) state.setdefault("deletedRows", []) state.setdefault("deleteRows", []) if folder_name not in state["folders"]: state["folders"].append(folder_name) state["graphFolder"][row_key] = folder_name state["open"][folder_name] = True state["deletedRows"] = [x for x in state.get("deletedRows", []) if str(x) != row_key] state["deleteRows"] = [] con.execute( """ INSERT INTO ui_state(key, value_json, created_at, updated_at) VALUES ('folders', ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value_json=excluded.value_json, updated_at=excluded.updated_at """, (json.dumps(state, ensure_ascii=False, sort_keys=True), ts, ts), ) con.commit() return { "ok": True, "graph": { "id": graph_id, "row_key": row_key, "visible_name": name, "folder": folder_name, "events_created": 0, }, } def sync_event_meta_to_events_sql(): """ Synchronize UI technical-card metadata into canonical events. Source: - ui_state key = 'event_meta' Target: - events.place - events.who - events.brief_description - events.source_link - events.event_name - events.date_text/date_year/date_month/date_day This function does not create events. It only updates existing events. Matching uses: 1. event_meta key format row:start:end:name when available. 2. saved group/name/date fields. 3. row_key + event_name fallback. """ ensure_runtime_schema() import json updated = 0 skipped = 0 ts = now_iso() with connect_db() as con: row = con.execute( "SELECT value_json FROM ui_state WHERE key='event_meta'" ).fetchone() if not row: return {"updated": 0, "skipped": 0, "reason": "no event_meta"} try: meta = json.loads(row["value_json"] or "{}") except Exception as exc: return {"updated": 0, "skipped": 0, "error": str(exc)} if not isinstance(meta, dict): return {"updated": 0, "skipped": 0, "reason": "event_meta is not object"} con.execute("BEGIN IMMEDIATE") for key, value in meta.items(): if not isinstance(value, dict): skipped += 1 continue key_s = str(key or "") parts = key_s.split(":", 3) key_row = parts[0].strip() if len(parts) >= 1 else "" key_start = None key_end = None key_name = "" if len(parts) >= 4: try: key_start = int(parts[1]) key_end = int(parts[2]) except Exception: key_start = None key_end = None key_name = parts[3].strip() row_key = str( value.get("row") or value.get("row_key") or key_row or value.get("group") or "" ).strip() event_name = str( value.get("name") or value.get("event_name") or key_name or "" ).strip() raw_date = str( value.get("date") or value.get("date_text") or "" ).strip() place = str(value.get("place") or value.get("PLACE") or "").strip() who = str(value.get("who") or value.get("WHO") or "").strip() brief_description = str( value.get("brief_description") or value.get("brief") or value.get("BRIEF") or value.get("description") or "" ).strip() source_link = str(value.get("source_link") or value.get("url") or "").strip() if not row_key or not event_name: skipped += 1 continue target = None if key_start is not None and key_end is not None: target = con.execute( """ SELECT id, event_name, date_text, date_year, date_month, date_day FROM events WHERE row_key = ? AND start_col = ? AND end_col = ? AND event_name = ? ORDER BY id DESC LIMIT 1 """, (row_key, key_start, key_end, event_name), ).fetchone() if target is None and raw_date: target = con.execute( """ SELECT id, event_name, date_text, date_year, date_month, date_day FROM events WHERE row_key = ? AND event_name = ? AND date_text = ? ORDER BY id DESC LIMIT 1 """, (row_key, event_name, raw_date), ).fetchone() if target is None: target = con.execute( """ SELECT id, event_name, date_text, date_year, date_month, date_day FROM events WHERE row_key = ? AND event_name = ? ORDER BY id DESC LIMIT 1 """, (row_key, event_name), ).fetchone() if target is None: skipped += 1 continue date_text = target["date_text"] date_year = target["date_year"] date_month = target["date_month"] date_day = target["date_day"] if raw_date: try: date_text, date_year, date_month, date_day = smart_parse_date(raw_date) except Exception: pass con.execute( """ UPDATE events SET event_name = ?, date_text = ?, date_year = ?, date_month = ?, date_day = ?, place = ?, who = ?, brief_description = ?, source_link = ?, updated_at = ? WHERE id = ? """, ( event_name, date_text, date_year, date_month, date_day, place, who, brief_description, source_link, ts, int(target["id"]), ), ) updated += 1 con.commit() return {"updated": updated, "skipped": skipped} def save_session_sql(): """ Explicit session save endpoint. Operational policy: - Does not create data. - Does not delete data. - Forces frontend render from canonical SQLite. - Verifies SQLite integrity. - Returns current object counts. Most operations already persist transactionally. This endpoint is the explicit user-facing Save barrier. """ ensure_runtime_schema() event_meta_sync = sync_event_meta_to_events_sql() sync_ui_folders_to_sql() with connect_db() as con: integrity = con.execute("PRAGMA integrity_check").fetchone()[0] graph_count = con.execute("SELECT COUNT(*) FROM graphs").fetchone()[0] event_count = con.execute("SELECT COUNT(*) FROM events").fetchone()[0] api_count = 0 try: api_count = con.execute("SELECT COUNT(*) FROM api_registry").fetchone()[0] except Exception: api_count = 0 render_frontend() return { "ok": integrity == "ok", "sqlite_integrity": integrity, "graphs": graph_count, "events": event_count, "apis": api_count, } def ensure_persistence_audit_schema(): ensure_runtime_schema() with connect_db() as con: con.execute( """ CREATE TABLE IF NOT EXISTS persistence_audit( id INTEGER PRIMARY KEY CHECK(id = 1), revision INTEGER NOT NULL DEFAULT 0, dirty INTEGER NOT NULL DEFAULT 0, last_operation TEXT, last_status TEXT, last_error TEXT, sqlite_integrity TEXT, graph_count INTEGER DEFAULT 0, event_count INTEGER DEFAULT 0, folder_count INTEGER DEFAULT 0, api_count INTEGER DEFAULT 0, updated_at TEXT ) """ ) con.execute( """ INSERT OR IGNORE INTO persistence_audit( id, revision, dirty, last_operation, last_status, sqlite_integrity, updated_at ) VALUES (1, 0, 0, 'init', 'ok', 'unknown', ?) """, (now_iso(),), ) con.commit() def persistence_counts_sql(con): graph_count = con.execute("SELECT COUNT(*) FROM graphs").fetchone()[0] event_count = con.execute("SELECT COUNT(*) FROM events").fetchone()[0] try: folder_count = con.execute("SELECT COUNT(*) FROM folders").fetchone()[0] except Exception: folder_count = 0 try: api_count = con.execute("SELECT COUNT(*) FROM api_registry").fetchone()[0] except Exception: api_count = 0 return graph_count, event_count, folder_count, api_count def record_persistence_audit(operation="autosave", dirty=0, status="ok", error=""): ensure_persistence_audit_schema() with connect_db() as con: integrity = con.execute("PRAGMA integrity_check").fetchone()[0] graph_count, event_count, folder_count, api_count = persistence_counts_sql(con) ok = integrity == "ok" and status == "ok" ts = now_iso() con.execute( """ UPDATE persistence_audit SET revision = revision + 1, dirty = ?, last_operation = ?, last_status = ?, last_error = ?, sqlite_integrity = ?, graph_count = ?, event_count = ?, folder_count = ?, api_count = ?, updated_at = ? WHERE id = 1 """, ( 1 if dirty else 0, operation, "ok" if ok else "error", str(error or ""), integrity, graph_count, event_count, folder_count, api_count, ts, ), ) con.commit() return { "ok": ok, "dirty": bool(dirty), "operation": operation, "status": "ok" if ok else "error", "error": str(error or ""), "sqlite_integrity": integrity, "graphs": graph_count, "events": event_count, "folders": folder_count, "apis": api_count, "updated_at": ts, } def persistence_status_sql(): ensure_persistence_audit_schema() with connect_db() as con: integrity = con.execute("PRAGMA integrity_check").fetchone()[0] graph_count, event_count, folder_count, api_count = persistence_counts_sql(con) row = con.execute("SELECT * FROM persistence_audit WHERE id = 1").fetchone() data = dict(row) if row else {} return { "ok": integrity == "ok" and data.get("last_status", "ok") == "ok", "dirty": bool(data.get("dirty", 0)), "revision": data.get("revision", 0), "last_operation": data.get("last_operation", ""), "last_status": data.get("last_status", ""), "last_error": data.get("last_error", ""), "sqlite_integrity": integrity, "graphs": graph_count, "events": event_count, "folders": folder_count, "apis": api_count, "updated_at": data.get("updated_at", ""), } def save_ui_state_payload_sql(payload): """ Persist current UI folder state into SQLite. This is required for autosave of folders, graph-folder mappings, open/closed folder state, and deleted-row state. """ ensure_runtime_schema() data = dict(payload or {}) key = str(data.get("key") or data.get("state_key") or "folders").strip() or "folders" if "value" in data: value = data["value"] elif "data" in data: value = data["data"] elif "value_json" in data: value = data["value_json"] else: value = data if isinstance(value, str): try: value = json.loads(value) except Exception: value = {"raw": value} if not isinstance(value, dict): value = {"value": value} ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") if key == "folders": value.setdefault("folders", []) value.setdefault("graphFolder", {}) value.setdefault("open", {}) value.setdefault("deletedRows", []) value.setdefault("deleteRows", []) for folder_name in value.get("folders", []) or []: folder_name = str(folder_name).strip() if not folder_name: continue folder_id = get_or_create_folder(con, folder_name) try: con.execute( "UPDATE folders SET is_open = ?, updated_at = ? WHERE id = ?", (1 if value.get("open", {}).get(folder_name, True) else 0, ts, folder_id), ) except Exception: pass for row_key, folder_name in (value.get("graphFolder", {}) or {}).items(): row_key = str(row_key).strip() folder_name = str(folder_name).strip() or "Uncategorized" exists = con.execute( "SELECT id FROM graphs WHERE row_key = ?", (row_key,), ).fetchone() if not exists: continue folder_id = get_or_create_folder(con, folder_name) con.execute( "UPDATE graphs SET folder_id = ?, updated_at = ? WHERE row_key = ?", (folder_id, ts, row_key), ) con.execute( """ INSERT INTO ui_state(key, value_json, created_at, updated_at) VALUES (?, ?, ?, ?) ON CONFLICT(key) DO UPDATE SET value_json=excluded.value_json, updated_at=excluded.updated_at """, (key, json.dumps(value, ensure_ascii=False, sort_keys=True), ts, ts), ) con.commit() record_persistence_audit("ui-state", dirty=0, status="ok") return {"ok": True, "key": key, "saved": True} def autosave_session_sql(payload=None): """ Autosave barrier. It first persists UI state when present, then verifies SQLite and regenerates the HTML. It does not fabricate application data. """ payload = dict(payload or {}) try: if payload.get("ui_state"): save_ui_state_payload_sql({ "key": "folders", "value": payload.get("ui_state"), }) render_frontend() return record_persistence_audit("autosave", dirty=0, status="ok") except Exception as exc: return record_persistence_audit( "autosave", dirty=1, status="error", error=str(exc), ) def save_session_sql(): """ Manual Save button uses the same real autosave verification path. """ return autosave_session_sql({}) def list_apis(): """ Schema-tolerant API registry reader. It reads the actual api_registry columns present in SQLite and fills missing operational fields with safe defaults. This prevents /api/health from failing when older databases do not yet contain last_status, last_latency_ms, last_run_at, etc. """ ensure_runtime_schema() wanted_defaults = { "last_status": None, "last_latency_ms": None, "last_run_at": None, "last_inserted": 0, "last_updated": 0, "last_skipped": 0, "status": "unknown", "raw_json": "", "created_at": "", "updated_at": "", "enabled": True, "method": "GET", "target_graph": "", "endpoint": "", "name": "", } with connect_db() as con: cols = [ r["name"] for r in con.execute("PRAGMA table_info(api_registry)").fetchall() ] if not cols: return [] select_cols = ", ".join(cols) rows = con.execute( f"SELECT {select_cols} FROM api_registry" ).fetchall() out = [] for r in rows: item = dict(r) for key, default in wanted_defaults.items(): item.setdefault(key, default) if "id" not in item: item["id"] = item.get("name") or item.get("endpoint") or "api" item["enabled"] = bool(item.get("enabled")) out.append(item) out.sort(key=lambda x: (str(x.get("name") or ""), str(x.get("id") or ""))) return out def upsert_api(payload): ensure_runtime_schema() data = dict(payload or {}) api_id = str(data.get("id") or f"API_{uuid.uuid4().hex[:12]}") name = str(data.get("name") or api_id) enabled = 1 if data.get("enabled", True) else 0 target_graph = str(data.get("target_graph") or data.get("graph") or "") method = str(data.get("method") or "GET").upper() endpoint = str(data.get("endpoint") or "") interval = int(data.get("interval_seconds") or 300) status = str(data.get("status") or "registered") raw = { "id": api_id, "name": name, "enabled": bool(enabled), "target_graph": target_graph, "method": method, "endpoint": endpoint, "interval_seconds": interval, "status": status, **data, } ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") con.execute( """ INSERT INTO api_registry(id, name, enabled, target_graph, method, endpoint, interval_seconds, status, raw_json, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, enabled=excluded.enabled, target_graph=excluded.target_graph, method=excluded.method, endpoint=excluded.endpoint, interval_seconds=excluded.interval_seconds, status=excluded.status, raw_json=excluded.raw_json, updated_at=excluded.updated_at """, (api_id, name, enabled, target_graph, method, endpoint, interval, status, json.dumps(raw, ensure_ascii=False, sort_keys=True), ts, ts), ) if target_graph: get_or_create_graph(con, target_graph, target_graph, "Uncategorized") con.commit() return raw # --- V55 REAL API STATUS / CONNECTIVITY LAYER --- def api_registry_columns(): try: with connect_db() as con: return [ r["name"] for r in con.execute("PRAGMA table_info(api_registry)").fetchall() ] except Exception: return [] def parse_json_object(value): if not value: return {} if isinstance(value, dict): return value if isinstance(value, str): try: parsed = json.loads(value) return parsed if isinstance(parsed, dict) else {} except Exception: return {} return {} def normalize_api_row(api): """ Normaliza una fila de api_registry mezclando columnas reales y raw_json. Esto permite soportar bases viejas y nuevas. """ item = {} if isinstance(api, sqlite3.Row): api = dict(api) if isinstance(api, dict): raw = parse_json_object(api.get("raw_json")) item.update(raw) item.update(api) item.setdefault("id", item.get("name") or item.get("endpoint") or "api") item.setdefault("name", item.get("id")) item.setdefault("enabled", True) item.setdefault("method", "GET") item.setdefault("endpoint", item.get("url") or "") item.setdefault("target_graph", item.get("graph") or item.get("row") or "") item.setdefault("status", item.get("last_status") or "unknown") item.setdefault("last_latency_ms", None) item.setdefault("last_run_at", "") item.setdefault("last_error", "") item["enabled"] = bool(item.get("enabled")) return item def set_api_status(api_id, **fields): """ Actualiza estado runtime de una API de forma tolerante al esquema. Si algunas columnas no existen, las ignora. También intenta fusionar raw_json cuando existe. """ ensure_runtime_schema() api_id = str(api_id) cols = api_registry_columns() if not cols: return {"ok": False, "error": "api_registry table has no columns"} ts = now_iso() with connect_db() as con: con.execute("BEGIN IMMEDIATE") row = con.execute( "SELECT * FROM api_registry WHERE CAST(id AS TEXT) = ? OR name = ? LIMIT 1", (api_id, api_id), ).fetchone() if not row: con.rollback() return {"ok": False, "error": "api not found", "api_id": api_id} direct = {} for key, value in fields.items(): if key in cols: direct[key] = value if "updated_at" in cols: direct["updated_at"] = ts if direct: assignments = ", ".join([f"{k} = ?" for k in direct.keys()]) params = list(direct.values()) + [row["id"]] con.execute( f"UPDATE api_registry SET {assignments} WHERE id = ?", params, ) if "raw_json" in cols: raw = parse_json_object(row["raw_json"] if "raw_json" in row.keys() else "") raw.update(fields) raw["updated_at"] = ts con.execute( "UPDATE api_registry SET raw_json = ? WHERE id = ?", (json.dumps(raw, ensure_ascii=False, sort_keys=True), row["id"]), ) con.commit() return {"ok": True, "api_id": api_id, "updated": True} def request_headers_for_api(api): headers = { "User-Agent": "GRAFO-D-V55-API-SENSOR/1.0", "Accept": "application/json, text/plain, */*", } raw_headers = ( api.get("headers_json") or api.get("headers") or api.get("request_headers") or "" ) parsed = parse_json_object(raw_headers) for key, value in parsed.items(): if key and value is not None: headers[str(key)] = str(value) return headers def endpoint_for_api(api): return str( api.get("endpoint") or api.get("url") or api.get("uri") or "" ).strip() def safe_api_method(api): method = str(api.get("method") or "GET").upper().strip() # El sensor no debe ejecutar métodos con efectos laterales. if method not in ("GET", "HEAD"): return "GET" return method def check_api_connectivity(api, timeout=5): """ Verifica conectividad real de una API registrada. Estados: - active: HTTP 2xx - degraded: HTTP 3xx o respuesta rara pero alcanzable - inactive: timeout, error de red, HTTP 4xx/5xx - unknown: API deshabilitada o sin endpoint """ api = normalize_api_row(api) api_id = str(api.get("id") or api.get("name") or "api") name = str(api.get("name") or api_id) endpoint = endpoint_for_api(api) base = { "id": api_id, "name": name, "enabled": bool(api.get("enabled")), "target_graph": str(api.get("target_graph") or api.get("graph") or api.get("row") or ""), "endpoint": endpoint, "method": safe_api_method(api), "status": "unknown", "runtime_status": "unknown", "http_status": None, "latency_ms": None, "last_checked": now_iso(), "error": None, } if not api.get("enabled"): base["status"] = "unknown" base["runtime_status"] = "unknown" base["error"] = "API disabled" return base if not endpoint: base["status"] = "unknown" base["runtime_status"] = "unknown" base["error"] = "Missing endpoint" return base parsed = urlparse(endpoint) if parsed.scheme not in ("http", "https"): base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = "Unsupported endpoint scheme" return base started = time.perf_counter() try: req = Request( endpoint, headers=request_headers_for_api(api), method=safe_api_method(api), ) with urlopen(req, timeout=timeout) as res: latency = int((time.perf_counter() - started) * 1000) http_status = int(getattr(res, "status", 0) or res.getcode() or 0) base["http_status"] = http_status base["latency_ms"] = latency if 200 <= http_status < 300: base["status"] = "active" base["runtime_status"] = "active" base["error"] = None elif 300 <= http_status < 400: base["status"] = "degraded" base["runtime_status"] = "degraded" base["error"] = f"Redirect/status {http_status}" else: base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = f"HTTP {http_status}" except HTTPError as exc: latency = int((time.perf_counter() - started) * 1000) base["http_status"] = int(getattr(exc, "code", 0) or 0) base["latency_ms"] = latency base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = f"HTTPError {base['http_status']}" except URLError as exc: latency = int((time.perf_counter() - started) * 1000) base["latency_ms"] = latency base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = str(getattr(exc, "reason", exc)) except Exception as exc: latency = int((time.perf_counter() - started) * 1000) base["latency_ms"] = latency base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = str(exc) try: set_api_status( api_id, status=base["status"], last_status=base["http_status"], last_latency_ms=base["latency_ms"], last_run_at=base["last_checked"], last_error=base["error"] or "", last_checked=base["last_checked"], runtime_status=base["runtime_status"], ) except Exception: pass return base def api_matches_graph(api, graph): if not graph: return True graph = str(graph).strip() candidates = { str(api.get("target_graph") or ""), str(api.get("graph") or ""), str(api.get("row") or ""), str(api.get("name") or ""), str(api.get("id") or ""), } return graph in candidates def api_status_sql(graph=None): """ Endpoint backend para el sensor real: GET /api/apis/status GET /api/apis/status?graph=ISS """ ensure_runtime_schema() apis = [ normalize_api_row(api) for api in list_apis() ] selected = [ api for api in apis if api_matches_graph(api, graph) ] results = [ check_api_connectivity(api) for api in selected ] if not results: overall = "unknown" elif any(x["status"] == "inactive" for x in results): overall = "inactive" elif any(x["status"] == "degraded" for x in results): overall = "degraded" elif any(x["status"] == "unknown" for x in results): overall = "unknown" else: overall = "active" return { "ok": True, "graph": graph, "status": overall, "checked_at": now_iso(), "count": len(results), "apis": results, } def fetch_api_json(api, timeout=API_SYNC_TIMEOUT_SECONDS): """ Descarga JSON desde la API. Se usa por sync_api. """ api = normalize_api_row(api) endpoint = endpoint_for_api(api) if not endpoint: raise ValueError("API endpoint is required") parsed = urlparse(endpoint) if parsed.scheme not in ("http", "https"): raise ValueError("Unsupported endpoint scheme") req = Request( endpoint, headers=request_headers_for_api(api), method=safe_api_method(api), ) with urlopen(req, timeout=timeout) as res: raw = res.read(1024 * 1024) text = raw.decode("utf-8", errors="replace") try: return json.loads(text) except Exception: return {"text": text} def payload_items(payload): """ Convierte respuestas comunes de APIs en una lista de items. """ if isinstance(payload, list): return payload if isinstance(payload, dict): for key in ("events", "items", "results", "data", "payload"): val = payload.get(key) if isinstance(val, list): return val return [payload] return [{"value": payload}] def event_from_api_item(api, item, index=0): """ Mapeo tolerante API → evento canónico GRAFO_D. """ api = normalize_api_row(api) if not isinstance(item, dict): item = {"value": item} target_graph = str( api.get("target_graph") or api.get("graph") or api.get("row") or api.get("name") or "API" ) api_name = str(api.get("name") or api.get("id") or "API") lat, lon = extract_lat_lon(item) raw_time = first_present( item, ("datetime", "date", "date_text", "timestamp", "time", "year"), "", ) epoch = epoch_to_utc_parts(raw_time) if epoch: iso_datetime, date_text, year, month, day = epoch name = f"{api_name} position {iso_datetime}" event_date = date_text else: name = str( first_present( item, ("name", "title", "event_name", "label"), f"{api_name} event {index + 1}", ) ) event_date = str(raw_time or now_iso()[:10]) if is_iss_payload(api, item): if epoch: name = f"ISS position {epoch[0]}" elif "timestamp" in item: name = f"ISS position {item.get('timestamp')}" place = build_place(item, lat, lon) description = str( first_present( item, ("description", "brief", "brief_description", "summary", "text", "message"), "", ) ) if not description: try: description = json.dumps(item, ensure_ascii=False, sort_keys=True) except Exception: description = str(item) source_id = str( first_present( item, ("id", "uuid", "event_id", "legacy_id"), hashlib.sha1( json.dumps(item, ensure_ascii=False, sort_keys=True).encode("utf-8") ).hexdigest()[:16], ) ) return { "id": f"{api.get('id') or api_name}:{source_id}", "row": target_graph, "row_key": target_graph, "name": name, "event_name": name, "date": event_date, "date_text": event_date, "place": place, "who": str(first_present(item, ("who", "actor", "source"), api_name)), "description": description, "brief_description": description, "source_link": endpoint_for_api(api), "api": api_name, "api_text": api_name, } def sync_api(api_id): """ Sincronización tolerante de una API registrada. - Verifica conectividad real. - Si responde, intenta leer JSON. - Normaliza items a eventos. - Inserta o actualiza eventos en SQLite. """ ensure_runtime_schema() api_id = str(api_id) apis = [ normalize_api_row(api) for api in list_apis() ] api = None for candidate in apis: if str(candidate.get("id")) == api_id or str(candidate.get("name")) == api_id: api = candidate break if not api: return { "ok": False, "api_id": api_id, "error": "API not found", } status = check_api_connectivity(api, timeout=API_SYNC_TIMEOUT_SECONDS) if status["status"] not in ("active", "degraded"): return { "ok": False, "api_id": api_id, "status": status, "inserted": 0, "updated": 0, "skipped": 0, "error": status.get("error") or "API not reachable", } payload = fetch_api_json(api, timeout=API_SYNC_TIMEOUT_SECONDS) items = payload_items(payload) inserted = 0 updated = 0 skipped = 0 for index, item in enumerate(items): try: event = event_from_api_item(api, item, index=index) result = insert_event_sql(event, allow_upsert=True) action = result.get("action") if action == "inserted": inserted += 1 elif action == "updated": updated += 1 else: skipped += 1 try: with connect_db() as con: con.execute( """ INSERT OR IGNORE INTO api_ingest_log( api_id, event_legacy_id, action, created_at ) VALUES (?, ?, ?, ?) """, ( str(api.get("id") or api_id), str(event.get("id")), str(action or "upsert"), now_iso(), ), ) con.commit() except Exception: pass except Exception: skipped += 1 try: set_api_status( api_id, last_inserted=inserted, last_updated=updated, last_skipped=skipped, status="active", last_run_at=now_iso(), ) except Exception: pass render_frontend() return { "ok": True, "api_id": api_id, "status": status, "inserted": inserted, "updated": updated, "skipped": skipped, } # --- END V55 REAL API STATUS / CONNECTIVITY LAYER --- def api_poll_worker(): while True: try: apis = list_apis() now = time.time() for api in apis: if not api.get("enabled"): continue endpoint = api.get("endpoint") if not endpoint: continue interval = int(api.get("interval_seconds") or 300) last_run = api.get("last_run_epoch") due = True if isinstance(last_run, (int, float)): due = now - float(last_run) >= interval # Persist an epoch marker only when attempting a poll to prevent tight retry loops. if due: set_api_status(api["id"], last_run_epoch=now, status=api.get("status") or "scheduled") try: sync_api(api["id"]) except Exception: pass except Exception: pass time.sleep(5) def start_api_worker_once(): global _api_worker_started if _api_worker_started: return _api_worker_started = True t = threading.Thread(target=api_poll_worker, daemon=True) t.start() 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) # --- V55B API PROBE FALLBACK OVERRIDE --- # Robust backend probe for /api/apis/status. # It first tries urllib and falls back to local curl when urllib times out. def _v55b_probe_with_urllib(api, timeout=15): api = normalize_api_row(api) endpoint = endpoint_for_api(api) if not endpoint: raise ValueError("Missing endpoint") parsed = urlparse(endpoint) if parsed.scheme not in ("http", "https"): raise ValueError("Unsupported endpoint scheme") started = time.perf_counter() req = Request( endpoint, headers=request_headers_for_api(api), method=safe_api_method(api), ) with urlopen(req, timeout=timeout) as res: # Leer poco para forzar recepción real sin bloquear con respuestas grandes. try: res.read(4096) except Exception: pass latency = int((time.perf_counter() - started) * 1000) http_status = int(getattr(res, "status", 0) or res.getcode() or 0) return { "ok": 200 <= http_status < 300, "http_status": http_status, "latency_ms": latency, "error": None if 200 <= http_status < 300 else f"HTTP {http_status}", "transport": "urllib", } def _v55b_probe_with_curl(api, timeout=15): api = normalize_api_row(api) endpoint = endpoint_for_api(api) if not endpoint: raise ValueError("Missing endpoint") parsed = urlparse(endpoint) if parsed.scheme not in ("http", "https"): raise ValueError("Unsupported endpoint scheme") started = time.perf_counter() cmd = [ "curl", "-L", "-sS", "-o", "/dev/null", "-w", "%{http_code} %{time_total}", "--max-time", str(int(timeout)), endpoint, ] proc = subprocess.run( cmd, cwd=str(ROOT), text=True, capture_output=True, timeout=timeout + 3, ) latency = int((time.perf_counter() - started) * 1000) if proc.returncode != 0: err = (proc.stderr or proc.stdout or "").strip() raise RuntimeError(err or f"curl returned {proc.returncode}") parts = (proc.stdout or "").strip().split() http_status = int(parts[0]) if parts and parts[0].isdigit() else 0 return { "ok": 200 <= http_status < 300, "http_status": http_status, "latency_ms": latency, "error": None if 200 <= http_status < 300 else f"HTTP {http_status}", "transport": "curl", } def check_api_connectivity(api, timeout=15): """ V55B robust real API connectivity check. Fixes false inactive state when urllib times out but curl succeeds locally. """ api = normalize_api_row(api) api_id = str(api.get("id") or api.get("name") or "api") name = str(api.get("name") or api_id) endpoint = endpoint_for_api(api) base = { "id": api_id, "name": name, "enabled": bool(api.get("enabled")), "target_graph": str(api.get("target_graph") or api.get("graph") or api.get("row") or ""), "endpoint": endpoint, "method": safe_api_method(api), "status": "unknown", "runtime_status": "unknown", "http_status": None, "latency_ms": None, "last_checked": now_iso(), "error": None, "transport": None, } if not api.get("enabled"): base["status"] = "unknown" base["runtime_status"] = "unknown" base["error"] = "API disabled" return base if not endpoint: base["status"] = "unknown" base["runtime_status"] = "unknown" base["error"] = "Missing endpoint" return base errors = [] probe = None try: probe = _v55b_probe_with_urllib(api, timeout=timeout) except Exception as exc: errors.append("urllib: " + str(exc)) try: probe = _v55b_probe_with_curl(api, timeout=timeout) except Exception as exc2: errors.append("curl: " + str(exc2)) if probe: base["http_status"] = probe.get("http_status") base["latency_ms"] = probe.get("latency_ms") base["transport"] = probe.get("transport") if probe.get("ok"): base["status"] = "active" base["runtime_status"] = "active" base["error"] = None else: status_code = int(probe.get("http_status") or 0) if 300 <= status_code < 400: base["status"] = "degraded" base["runtime_status"] = "degraded" base["error"] = probe.get("error") or f"HTTP {status_code}" else: base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = probe.get("error") or f"HTTP {status_code}" else: base["status"] = "inactive" base["runtime_status"] = "inactive" base["error"] = " | ".join(errors) if errors else "probe failed" try: set_api_status( api_id, status=base["status"], last_status=base["http_status"], last_latency_ms=base["latency_ms"], last_run_at=base["last_checked"], last_error=base["error"] or "", last_checked=base["last_checked"], runtime_status=base["runtime_status"], ) except Exception: pass return base def api_status_sql(graph=None): """ V55B endpoint backend: GET /api/apis/status GET /api/apis/status?graph=ISS """ ensure_runtime_schema() apis = [ normalize_api_row(api) for api in list_apis() ] selected = [ api for api in apis if api_matches_graph(api, graph) ] results = [ check_api_connectivity(api, timeout=15) for api in selected ] if not results: overall = "unknown" elif any(x["status"] == "inactive" for x in results): overall = "inactive" elif any(x["status"] == "degraded" for x in results): overall = "degraded" elif any(x["status"] == "unknown" for x in results): overall = "unknown" else: overall = "active" return { "ok": True, "graph": graph, "status": overall, "checked_at": now_iso(), "count": len(results), "apis": results, } # --- END V55B API PROBE FALLBACK OVERRIDE --- if __name__ == "__main__": port = int(sys.argv[1]) if len(sys.argv) > 1 else 8765 ensure_runtime_schema() render_frontend(block=True) start_api_worker_once() print(f"GRAFO D API server: http://127.0.0.1:{port}/outputs/timeline_excel_like.html?v=sqlite_server") ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()