#!/usr/bin/env python3 from __future__ import annotations import hashlib import json import re import sqlite3 import sys import unicodedata from datetime import datetime, timezone from pathlib import Path ROOT = Path(__file__).resolve().parent DB = ROOT / "DATABASE" / "GRAFO_D_DATABASE.sqlite" IMPORT_DIR = ROOT / "GRAPH_IMPORT" INBOX_FILE = IMPORT_DIR / "graph_inbox.txt" IMPORTED_DIR = IMPORT_DIR / "imported" REJECTED_DIR = IMPORT_DIR / "rejected" TARGET_FOLDER = "Uncategorized" FIELD_ALIASES = { "graph_name": { "GRAPH NAME", "GRAPH", "GRAFO", "NOMBRE DEL GRAFO", "NOMBRE GRAFO", "TITULO DEL GRAFO", "TÍTULO DEL GRAFO", "GRAPH TITLE", "TITLE", "TITULO", "TÍTULO", }, "name": { "NAME", "NOMBRE", "EVENT NAME", "NOMBRE DEL EVENTO", "EVENTO", "TITLE", "TITULO", "TÍTULO", }, "date": { "DATE", "FECHA", "YEAR", "AÑO", "ANO", "TIME", "TIMESTAMP", "WHEN", "CUANDO", "CUÁNDO", }, "place": { "PLACE", "LUGAR", "LOCATION", "UBICACION", "UBICACIÓN", "SITE", "REGION", "REGIÓN", }, "who": { "WHO", "QUIEN", "QUIÉN", "ACTOR", "ACTORS", "AUTORES", "AUTHOR", "AUTHORS", "PERSONA", "PERSONAS", "ORGANIZATION", "ORGANIZACION", "ORGANIZACIÓN", }, "brief_description": { "BRIEF DESCRIPTION", "DESCRIPTION", "DESCRIPCION", "DESCRIPCIÓN", "RESUMEN", "BRIEF", "ABSTRACT", "NOTES", "NOTAS", "DETALLE", "DETAIL", }, "source_link": { "LINK", "LINKS", "URL", "URLS", "SOURCE", "SOURCES", "FUENTE", "FUENTES", "REFERENCIA", "REFERENCIAS", "BIBLIOGRAFIA", "BIBLIOGRAFÍA", }, } _ALIAS_TO_FIELD = {} def _norm_label(value: str) -> str: value = str(value or "") value = unicodedata.normalize("NFKD", value) value = "".join(ch for ch in value if not unicodedata.combining(ch)) value = value.upper() value = re.sub(r"[^A-Z0-9]+", " ", value) return re.sub(r"\s+", " ", value).strip() for field, aliases in FIELD_ALIASES.items(): for alias in aliases: _ALIAS_TO_FIELD[_norm_label(alias)] = field def now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def ensure_import_dirs() -> None: IMPORT_DIR.mkdir(exist_ok=True) IMPORTED_DIR.mkdir(exist_ok=True) REJECTED_DIR.mkdir(exist_ok=True) if not INBOX_FILE.exists(): INBOX_FILE.write_text("", encoding="utf-8") def connect_db(db_path: Path = DB) -> sqlite3.Connection: con = sqlite3.connect(str(db_path), 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 clean_value(value: str) -> str: value = str(value or "").replace("\r\n", "\n").replace("\r", "\n") lines = [] for line in value.split("\n"): stripped = line.strip() if stripped.startswith("```") or stripped == "***": continue lines.append(line) value = "\n".join(lines).strip() value = re.sub(r"\n{3,}", "\n\n", value) return value.strip(" \t\n`") def is_separator(line: str) -> bool: return bool(re.match(r"^\s*(?:\*{3,}|-{3,}|_{3,})\s*$", line or "")) def label_from_line(line: str): raw = str(line or "").strip() if not raw or is_separator(raw): return None, "" probe = re.sub(r"^\s*#{1,6}\s*", "", raw).strip() if re.match(r"^(NODO|NODE)\s*[-#:]?\s*\d*\b", _norm_label(probe)): return None, "" probe = probe.strip("*").strip() inline = "" if ":" in probe: left, right = probe.split(":", 1) label_part = left inline = right.strip() else: label_part = probe label_part = label_part.strip("*").strip() return _ALIAS_TO_FIELD.get(_norm_label(label_part)), inline def first_nonempty_line(value: str) -> str: for line in str(value or "").splitlines(): line = clean_value(line) if line: return line return "" def read_value_after(lines, start: int) -> str: collected = [] for j in range(start, len(lines)): line = lines[j] if is_separator(line): if collected: break continue field, _ = label_from_line(line) if field is not None: break if re.match(r"^\s*#{1,6}\s+(NODO|NODE)\b", line, flags=re.I): break collected.append(line) return clean_value("\n".join(collected)) def extract_fields(block: str) -> dict: lines = block.replace("\r\n", "\n").replace("\r", "\n").split("\n") fields = {} i = 0 while i < len(lines): field, inline = label_from_line(lines[i]) if field is None: i += 1 continue value_lines = [] if inline: value_lines.append(inline) j = i + 1 while j < len(lines): if is_separator(lines[j]): if value_lines: break j += 1 continue next_field, _ = label_from_line(lines[j]) if next_field is not None: break if re.match(r"^\s*#{1,6}\s+(NODO|NODE)\b", lines[j], flags=re.I): break value_lines.append(lines[j]) j += 1 value = clean_value("\n".join(value_lines)) if value: if field == "source_link": urls = re.findall(r"https?://\S+", value) fields[field] = "\n".join(urls) if urls else value else: fields[field] = value i = max(j, i + 1) return fields def extract_graph_name(text: str) -> str: fields = extract_fields(text) if fields.get("graph_name"): return first_nonempty_line(fields["graph_name"]) lines = text.replace("\r\n", "\n").replace("\r", "\n").split("\n") for i, line in enumerate(lines): field, inline = label_from_line(line) if field == "graph_name": if inline: return clean_value(inline).splitlines()[0].strip() value = read_value_after(lines, i + 1) if value: return first_nonempty_line(value) for line in lines[:30]: s = re.sub(r"^\s*#{1,6}\s*", "", line).strip() s = s.strip("*").strip() if not s or is_separator(s): continue if _norm_label(s) in _ALIAS_TO_FIELD: continue if re.match(r"^(NODO|NODE)\b", s, flags=re.I): continue if len(s) <= 120: return s return "GRAPH_INBOX" def split_node_blocks(text: str) -> list: text = text.replace("\r\n", "\n").replace("\r", "\n") node_re = re.compile(r"(?im)^\s*(?:#{1,6}\s*)?(?:NODO|NODE)\s*[-#:]?\s*\d*\b.*$") matches = list(node_re.finditer(text)) if matches: blocks = [] for idx, match in enumerate(matches): start = match.start() end = matches[idx + 1].start() if idx + 1 < len(matches) else len(text) block = text[start:end].strip() if block: blocks.append(block) return blocks chunks = re.split(r"(?m)^\s*(?:\*{3,}|-{3,}|_{3,})\s*$", text) candidates = [] for chunk in chunks: fields = extract_fields(chunk) evidence = sum(1 for k in ("name", "date", "place", "who", "brief_description") if fields.get(k)) if evidence >= 2 and fields.get("date"): candidates.append(chunk.strip()) return [c for c in candidates if c] def parse_date(value: str): raw = clean_value(value) raw = re.sub(r"\[[^\]]+\]", "", raw).strip() raw = raw.replace("−", "-") raw = re.sub(r"\s+", " ", raw) if not raw: raise ValueError("date is required") def check(y, m=None, d=None): y = int(y) if y == 0 or y < -999999 or y > 999999: raise ValueError(f"year out of range: {y}") if m is None: return str(y), y, None, None m = int(m) if m < 1 or m > 12: raise ValueError(f"month out of range: {m}") if d is None: return f"{y}-{m:02d}" if y < 0 else f"{y:04d}-{m:02d}", y, m, None d = int(d) if d < 1 or d > 31: raise ValueError(f"day out of range: {d}") return f"{y}-{m:02d}-{d:02d}" if y < 0 else f"{y:04d}-{m:02d}-{d:02d}", y, m, d m = re.search(r"^\s*(-?\d{1,6})\s*[-/]\s*(\d{1,2})\s*[-/]\s*(\d{1,2})\s*$", raw) if m: return check(int(m.group(1)), int(m.group(2)), int(m.group(3))) m = re.search(r"^\s*(-?\d{1,6})\s*[-/]\s*(\d{1,2})\s*$", raw) if m: return check(int(m.group(1)), int(m.group(2)), None) m = re.search(r"^\s*(\d{1,2})\s*[-/]\s*(\d{1,2})\s*[-/]\s*(-?\d{1,6})\s*$", raw) if m: return check(int(m.group(3)), int(m.group(2)), int(m.group(1))) m = re.search(r"^\s*(-?\d{1,6})\s*$", raw) if m: return check(int(m.group(1)), None, None) m = re.search(r"(? str: text = unicodedata.normalize("NFKD", str(name or "")) text = "".join(ch for ch in text if not unicodedata.combining(ch)) text = text.upper() text = re.sub(r"[^A-Z0-9]+", " ", text) text = re.sub(r"\s+", " ", text).strip() return (text or "GRAPH_INBOX")[:80] def fallback_description(block: str) -> str: lines = [] for line in block.splitlines(): if label_from_line(line)[0] is not None: continue if re.match(r"^\s*#{1,6}\s+(NODO|NODE)\b", line, flags=re.I): continue if re.search(r"https?://", line): continue if is_separator(line) or line.strip().startswith("```"): continue lines.append(line) return clean_value("\n".join(lines)) def parse_graph_text(text: str): text = str(text or "").strip() if not text: raise ValueError("graph_inbox.txt is empty") graph_name = extract_graph_name(text) blocks = split_node_blocks(text) if not blocks: raise ValueError("no node/event blocks detected") nodes = [] warnings = [] for idx, block in enumerate(blocks, start=1): fields = extract_fields(block) urls = re.findall(r"https?://\S+", block) if urls and not fields.get("source_link"): fields["source_link"] = "\n".join(urls) name = first_nonempty_line(fields.get("name", "")) if not name: headings = re.findall(r"(?m)^\s*#{1,6}\s+(.+?)\s*$", block) for h in headings: if not re.match(r"^(NODO|NODE)\b", h.strip(), flags=re.I): name = h.strip() break date_raw = first_nonempty_line(fields.get("date", "")) if not name or not date_raw: warnings.append(f"node {idx} skipped: missing name or date") continue try: date_text, date_year, date_month, date_day = parse_date(date_raw) except Exception as exc: warnings.append(f"node {idx} skipped: invalid date {date_raw!r}: {exc}") continue brief = fields.get("brief_description", "") or fallback_description(block) nodes.append({ "index": len(nodes) + 1, "event_name": name, "date_text": date_text, "date_year": date_year, "date_month": date_month, "date_day": date_day, "place": fields.get("place", ""), "who": fields.get("who", ""), "brief_description": brief, "source_link": fields.get("source_link", ""), "api_text": "graph-inbox-txt", }) if not nodes: detail = "; ".join(warnings[-5:]) if warnings else "no valid nodes" raise ValueError(f"no valid nodes parsed: {detail}") return graph_name, nodes, warnings def ensure_folder(con, name: str) -> int: ts = now_iso() row = con.execute("SELECT id FROM folders WHERE name = ?", (name,)).fetchone() if row: con.execute("UPDATE folders SET is_open=1, updated_at=? WHERE id=?", (ts, int(row["id"]))) return int(row["id"]) cur = con.execute( "INSERT INTO folders(name, is_open, created_at, updated_at) VALUES (?, 1, ?, ?)", (name, ts, ts), ) return int(cur.lastrowid) def unique_row_key(con, base: str) -> str: base = row_key_from_name(base) candidate = base n = 2 while con.execute("SELECT id FROM graphs WHERE row_key = ?", (candidate,)).fetchone(): candidate = f"{base}_{n}" n += 1 return candidate def sync_ui_state(con, row_key: str, folder_name: str) -> None: ts = now_iso() row = con.execute("SELECT value_json FROM ui_state WHERE key='folders'").fetchone() if row: try: state = json.loads(row["value_json"] or "{}") 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 folder_name not in state["folders"]: state["folders"].insert(0, 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), ) def update_audit(con) -> None: try: ts = now_iso() graph_count = con.execute("SELECT COUNT(*) FROM graphs").fetchone()[0] event_count = con.execute("SELECT COUNT(*) FROM events").fetchone()[0] folder_count = con.execute("SELECT COUNT(*) FROM folders").fetchone()[0] api_count = con.execute("SELECT COUNT(*) FROM api_registry").fetchone()[0] integrity = con.execute("PRAGMA integrity_check").fetchone()[0] row = con.execute("SELECT id, revision FROM persistence_audit ORDER BY id DESC LIMIT 1").fetchone() if row: con.execute( """ UPDATE persistence_audit SET revision=?, dirty=0, last_operation=?, last_status=?, last_error=NULL, sqlite_integrity=?, graph_count=?, event_count=?, folder_count=?, api_count=?, updated_at=? WHERE id=? """, ( int(row["revision"] or 0) + 1, "import-graph-inbox", "ok", integrity, graph_count, event_count, folder_count, api_count, ts, int(row["id"]), ), ) except Exception: pass def preview_graph_inbox(inbox_path: Path = INBOX_FILE) -> dict: """ Read GRAPH_IMPORT/graph_inbox.txt without writing to SQLite. Returns graph name, detected node count, node names and parser warnings. """ ensure_import_dirs() inbox_path = Path(inbox_path) if not inbox_path.exists(): inbox_path.write_text("", encoding="utf-8") return { "ok": True, "preview": True, "importable": False, "message": "GRAPH_IMPORT/graph_inbox.txt is empty", "file": str(inbox_path), "graph_name": None, "nodes_detected": 0, "node_names": [], "warnings": [], } text = inbox_path.read_text(encoding="utf-8", errors="replace") if not text.strip(): return { "ok": True, "preview": True, "importable": False, "message": "GRAPH_IMPORT/graph_inbox.txt is empty", "file": str(inbox_path), "graph_name": None, "nodes_detected": 0, "node_names": [], "warnings": [], } try: graph_name, nodes, warnings = parse_graph_text(text) except Exception as exc: return { "ok": False, "preview": True, "importable": False, "error": str(exc), "file": str(inbox_path), "graph_name": None, "nodes_detected": 0, "node_names": [], "warnings": [], } return { "ok": True, "preview": True, "importable": True, "message": "Graph inbox preview completed", "file": str(inbox_path), "graph_name": graph_name, "nodes_detected": len(nodes), "node_names": [str(n.get("event_name") or "") for n in nodes], "warnings": warnings, } def import_graph_inbox(db_path: Path = DB, inbox_path: Path = INBOX_FILE) -> dict: ensure_import_dirs() inbox_path = Path(inbox_path) if not inbox_path.exists(): inbox_path.write_text("", encoding="utf-8") return { "ok": True, "imported": False, "message": f"created empty inbox: {inbox_path}", "file": str(inbox_path), } text = inbox_path.read_text(encoding="utf-8", errors="replace") if not text.strip(): return { "ok": True, "imported": False, "message": "GRAPH_IMPORT/graph_inbox.txt is empty", "file": str(inbox_path), } digest = hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest()[:12] try: graph_name, nodes, warnings = parse_graph_text(text) except Exception as exc: ts_file = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") rejected = REJECTED_DIR / f"{ts_file}_rejected_{digest}.txt" rejected.write_text(text, encoding="utf-8") return { "ok": False, "imported": False, "error": str(exc), "rejected_copy": str(rejected), "file": str(inbox_path), } ts = now_iso() with connect_db(Path(db_path)) as con: con.execute("BEGIN IMMEDIATE") folder_id = ensure_folder(con, TARGET_FOLDER) row_key = unique_row_key(con, graph_name) cur = con.execute( """ INSERT INTO graphs(row_key, visible_name, folder_id, color, created_at, updated_at) VALUES (?, ?, ?, NULL, ?, ?) """, (row_key, graph_name, folder_id, ts, ts), ) graph_id = int(cur.lastrowid) for i, node in enumerate(nodes, start=1): legacy_id = f"INBOX:{row_key}:{i}:{digest}" 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, str(node["event_name"]), str(node["date_text"]), int(node["date_year"]), node.get("date_month"), node.get("date_day"), i, i, 1, str(node.get("place") or ""), str(node.get("who") or ""), str(node.get("brief_description") or ""), str(node.get("source_link") or ""), str(node.get("api_text") or "graph-inbox-txt"), i, ts, ts, ), ) sync_ui_state(con, row_key, TARGET_FOLDER) update_audit(con) con.commit() ts_file = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") safe_key = re.sub(r"[^A-Z0-9_ -]+", "", row_key).replace(" ", "_")[:80] or "GRAPH" imported_copy = IMPORTED_DIR / f"{ts_file}_{safe_key}_{digest}.txt" imported_copy.write_text(text, encoding="utf-8") inbox_path.write_text("", encoding="utf-8") return { "ok": True, "imported": True, "graph": { "row_key": row_key, "visible_name": graph_name, "folder": TARGET_FOLDER, }, "nodes_imported": len(nodes), "warnings": warnings, "archived_copy": str(imported_copy), "inbox_cleared": True, } def main(argv) -> int: ensure_import_dirs() if len(argv) > 1 and argv[1] == "--status": print(json.dumps({ "ok": True, "inbox": str(INBOX_FILE), "exists": INBOX_FILE.exists(), "bytes": INBOX_FILE.stat().st_size if INBOX_FILE.exists() else 0, "target_folder": TARGET_FOLDER, }, ensure_ascii=False, indent=2)) return 0 if len(argv) > 1 and argv[1] == "--dry-run": text = INBOX_FILE.read_text(encoding="utf-8", errors="replace") graph_name, nodes, warnings = parse_graph_text(text) print(json.dumps({ "ok": True, "dry_run": True, "graph_name": graph_name, "nodes_detected": len(nodes), "warnings": warnings, }, ensure_ascii=False, indent=2)) return 0 result = import_graph_inbox() print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 if result.get("ok") else 2 if __name__ == "__main__": raise SystemExit(main(sys.argv))