import json import sqlite3 from pathlib import Path ROOT = Path(__file__).resolve().parent SQLITE_DB = ROOT / "DATABASE" / "GRAFO_D_DATABASE.sqlite" def _connect(): con = sqlite3.connect(SQLITE_DB) con.row_factory = sqlite3.Row con.execute("PRAGMA foreign_keys = ON") return con def _row_value(row, *names, default=None): """ Lector tolerante para sqlite3.Row. Permite soportar nombres canónicos y alias sin romper bases antiguas. """ if row is None: return default keys = set(row.keys()) for name in names: if name in keys: value = row[name] if value is not None: return value return default def _clean_text(value): if value is None: return "" return str(value) def _load_api_registry(cur): """ Carga tolerante de api_registry. Prioridad: 1. Si existe raw_json válido, se usa como base. 2. Se mezclan columnas reales de la tabla para que el frontend/backend pueda leer id, endpoint, enabled, status, target_graph, etc. """ api_registry = [] try: cols = [ r["name"] for r in cur.execute("PRAGMA table_info(api_registry)").fetchall() ] if not cols: return api_registry rows = cur.execute( "SELECT " + ", ".join(cols) + " FROM api_registry ORDER BY id" ).fetchall() for row in rows: item = {} raw_json = "" if "raw_json" in cols: raw_json = row["raw_json"] or "" if raw_json: try: parsed = json.loads(raw_json) if isinstance(parsed, dict): item.update(parsed) except Exception: pass for col in cols: value = row[col] if value is not None: item[col] = value 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.setdefault("target_graph", "") item.setdefault("status", "unknown") item["enabled"] = bool(item.get("enabled")) api_registry.append(item) except sqlite3.OperationalError: return [] return api_registry def _merge_dict_preserve_canonical(canonical, persisted): """ Mezcla metadata persistida desde ui_state sin perder datos canónicos de SQLite. Regla: - Si SQLite trae PLACE / WHO / BRIEF, esos valores son fuente principal. - Si ui_state trae campos adicionales, se conservan. - Si ui_state trae valores vacíos, no borra valores canónicos. """ if not isinstance(canonical, dict): canonical = {} if not isinstance(persisted, dict): return canonical merged = dict(canonical) for key, persisted_value in persisted.items(): if not isinstance(persisted_value, dict): merged[key] = persisted_value continue canonical_value = merged.get(key) if not isinstance(canonical_value, dict): canonical_value = {} item = dict(canonical_value) for field, value in persisted_value.items(): if value not in (None, ""): item[field] = value else: item.setdefault(field, value) merged[key] = item return merged def load_database(): con = _connect() cur = con.cursor() rows = cur.execute(""" SELECT e.*, g.visible_name AS graph_visible_name, g.row_key AS graph_row_key, f.name AS folder_name FROM events e JOIN graphs g ON g.id = e.graph_id JOIN folders f ON f.id = g.folder_id ORDER BY g.id, COALESCE(e.date_year, 999999), COALESCE(e.date_month, 99), COALESCE(e.date_day, 99), e.sort_index, e.id """).fetchall() events = [] event_meta = {} for r in rows: start_col = _row_value(r, "start_col", default=None) if start_col is None: start_col = _row_value(r, "sort_index", default=1) end_col = _row_value(r, "end_col", default=None) if end_col is None: end_col = start_col duration = _row_value(r, "duration", default=None) if duration is None: duration = 1 row_key = _clean_text(_row_value(r, "row_key", "graph_row_key", default="")) event_name = _clean_text(_row_value(r, "event_name", "name", default="Untitled event")) legacy_id = _clean_text(_row_value(r, "legacy_id", default="")) or f"{row_key}:{r['id']}" date_text = _clean_text(_row_value(r, "date_text", "date", default="")) date_year = _row_value(r, "date_year", "year", default=None) date_month = _row_value(r, "date_month", "month", default=None) date_day = _row_value(r, "date_day", "day", default=None) place = _clean_text(_row_value(r, "place", "PLACE", default="")) who = _clean_text(_row_value(r, "who", "WHO", "actor", default="")) brief_description = _clean_text( _row_value( r, "brief_description", "brief", "BRIEF", "description", default="" ) ) source_link = _clean_text(_row_value(r, "source_link", "url", "source", default="")) api_text = _clean_text(_row_value(r, "api_text", "api", default="")) folder_name = _clean_text(_row_value(r, "folder_name", default="Uncategorized")) sort_index = _row_value(r, "sort_index", default=start_col) event = { "id": legacy_id, "db_id": r["id"], "row": row_key, "row_key": row_key, "name": event_name, "event_name": event_name, "start_col": int(start_col or 1), "end_col": int(end_col or start_col or 1), "duration": int(duration or 1), "sort_index": int(sort_index or start_col or 1), # V55: campos canónicos visibles para frontend. "date": date_text, "date_text": date_text, "date_year": date_year, "date_month": date_month, "date_day": date_day, "place": place, "PLACE": place, "who": who, "WHO": who, "description": brief_description, "brief": brief_description, "BRIEF": brief_description, "brief_description": brief_description, "source_link": source_link, "api": api_text, "api_text": api_text, "folder": folder_name, "group": folder_name, } if date_year is not None: event["year"] = date_year if date_month is not None: event["month"] = date_month if date_day is not None: event["day"] = date_day events.append(event) key = f"{row_key}:{event['start_col']}:{event['end_col']}:{event_name}" meta = { "id": legacy_id, "db_id": r["id"], "group": folder_name, "row": row_key, "row_key": row_key, "name": event_name, "event_name": event_name, "date": date_text, "date_text": date_text, "year": date_year, "month": date_month, "day": date_day, "place": place, "PLACE": place, "who": who, "WHO": who, "description": brief_description, "brief": brief_description, "BRIEF": brief_description, "brief_description": brief_description, "source_link": source_link, "api": api_text, "api_text": api_text, } event_meta[key] = meta event["_meta"] = meta folders = cur.execute(""" SELECT name, is_open FROM folders ORDER BY id """).fetchall() graphs = cur.execute(""" SELECT g.row_key, g.visible_name, f.name AS folder_name FROM graphs g JOIN folders f ON f.id = g.folder_id ORDER BY g.id """).fetchall() api_registry = _load_api_registry(cur) canonical_graph_labels = { "PAYLOAD HIST": "PAYLOAD HIST", "ISS": "ISS" } graph_folder = { g["row_key"]: g["folder_name"] for g in graphs } open_state = { f["name"]: bool(f["is_open"]) for f in folders } years = sorted({ int(e["year"]) for e in events if "year" in e and e["year"] is not None }) width_by_year = {} for year in years: max_count = 1 row_keys = sorted({ str(e.get("row")) for e in events if e.get("year") == year }) for row_key in row_keys: count = sum( 1 for e in events if e.get("year") == year and str(e.get("row")) == row_key ) max_count = max(max_count, count) width_by_year[year] = max_count start_by_year = {} col = 2 for year in years: start_by_year[year] = col col += width_by_year[year] for year in years: per_row = {} year_events = [ e for e in events if e.get("year") == year ] year_events.sort(key=lambda e: ( str(e.get("row", "")), int(e.get("date_month") or e.get("month") or 99), int(e.get("date_day") or e.get("day") or 99), int(e.get("sort_index") or e.get("start_col") or 0), str(e.get("name", "")) )) for e in year_events: row_key = str(e.get("row")) idx = per_row.get(row_key, 0) target_col = start_by_year[year] + idx e["start_col"] = target_col e["end_col"] = target_col e["duration"] = 1 per_row[row_key] = idx + 1 axis_events = [] for year in years: start = start_by_year[year] end = start + width_by_year[year] - 1 axis_events.append({ "id": f"AXIS_{year}", "row": "__AXIS__", "row_key": "__AXIS__", "name": str(year), "event_name": str(year), "start_col": start, "end_col": end, "duration": width_by_year[year], "year": year, "date": str(year), "date_text": str(year), "place": "", "who": "", "description": "", "brief": "", "brief_description": "", "source_link": "", "api": "", "api_text": "", }) # Reconstruye metadata después del remapeo de columnas. # El frontend usa row:start_col:end_col:name como id estable. rebuilt_event_meta = {} for e in events: meta = e.pop("_meta", None) if not meta: continue start_col = e.get("start_col") end_col = e.get("end_col", start_col) name = e.get("name", "") row_key = e.get("row", "") meta["start_col"] = start_col meta["end_col"] = end_col rebuilt_event_meta[f"{row_key}:{start_col}:{end_col}:{name}"] = meta event_meta = rebuilt_event_meta data = { "schema": "grafo_d_sqlite_adapter", "version": "55-front-fields-api-ready", "engine_version": "existing-engine-unchanged", "events": axis_events + events, "categories": [], "metadata": { "status": "sqlite_canonical", "active_graph": None, "schema_note": "Runtime data loaded from canonical SQLite through adapter.", "timeline_layout": "engine_owned", "v55_note": "Events now expose place/who/brief fields directly to frontend." }, "graphs": [ { "id": "sqlite_canonical", "name": "SQLite Canonical", "type": "sqlite" } ], "ui_state": { "graph_labels": { g["row_key"]: canonical_graph_labels.get(g["row_key"], g["visible_name"]) for g in graphs }, "folders": { "folders": [f["name"] for f in folders], "graphFolder": graph_folder, "open": open_state, "deletedRows": [] }, "event_meta": event_meta, "folder_meta": {}, "graph_set_cards": {} }, "api_registry": api_registry } # Overlay de estado UI persistido en SQLite. # V55: event_meta se fusiona, no reemplaza, para no perder PLACE / WHO / BRIEF # provenientes de la tabla canónica events. try: ui_rows = cur.execute(""" SELECT key, value_json FROM ui_state ORDER BY key """).fetchall() for ui in ui_rows: key = ui["key"] try: value = json.loads(ui["value_json"]) except Exception: continue if key == "event_meta": data["ui_state"]["event_meta"] = _merge_dict_preserve_canonical( data["ui_state"].get("event_meta", {}), value ) elif key == "folders": if isinstance(value, dict): value.setdefault("folders", data["ui_state"]["folders"].get("folders", [])) value.setdefault("graphFolder", data["ui_state"]["folders"].get("graphFolder", {})) value.setdefault("open", data["ui_state"]["folders"].get("open", {})) value.setdefault("deletedRows", []) data["ui_state"]["folders"] = value else: data.setdefault("ui_state", {})[key] = value except sqlite3.OperationalError: pass con.close() return data def save_database(data): raise RuntimeError("Direct JSON save is disabled. Use SQLite write operations.") if __name__ == "__main__": print(json.dumps(load_database(), indent=2, ensure_ascii=False))