#!/usr/bin/env python3 """ Settled Reach Wiki Sync — parse wiki frontmatter into SQLite. Wiki pages are the single source of truth for all per-system data. star-map.json only owns topology (edges). All per-node fields come from wiki. This module provides the sync logic. Called by tooling/process-wiki-changes. Usage: python3 wiki_sync.py sync Parse wiki and upsert into SQLite python3 wiki_sync.py stats Show completion stats python3 wiki_sync.py --help Show this help """ import json import re import sqlite3 import sys from pathlib import Path # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- SCRIPT_DIR = Path(__file__).resolve().parent CONFIG_PATH = SCRIPT_DIR / "config.json" WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve() SCHEMA_PATH = WORKTREE_ROOT / "db" / "schema.sql" WIKI_DIR = WORKTREE_ROOT / "wiki" / "star-systems" DB_PATH = (WORKTREE_ROOT / ".." / "settledreach.db").resolve() # --------------------------------------------------------------------------- # Config / DB # --------------------------------------------------------------------------- def load_config(): with open(CONFIG_PATH, "r") as f: cfg = json.load(f) cfg["sqlite_db_resolved"] = str(DB_PATH) return cfg def get_connection(cfg): conn = sqlite3.connect(cfg["sqlite_db_resolved"]) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA foreign_keys=ON;") conn.row_factory = sqlite3.Row return conn # --------------------------------------------------------------------------- # Column definitions (must match db/schema.sql star_systems table) # --------------------------------------------------------------------------- ALL_COLUMNS = [ "system_id", "astronomical_id", "proper_name", "system_name", "star_type", "geographic_sector", "geographic_band", "political_zone", "habitable_planet_count", "inhabited_planet_count", "asteroid_belt", "gas_giant", "habitability_profile", "horizon_station", "aperture_count", "gate_connections", "gate_topology", "span_gate_network", "settlement_wave", "founding_motivation", "founding_culture_primary", "founding_culture_secondary", "cultural_persistence", "historical_event", "historical_event_age_years", "religious_status", "religious_generation_count", "economic_tier", "population", "economic_base_primary", "economic_base_secondary", "distribution_index", "imprint_access", "supply_dependency", "governance_type", "dominant_faction", "primary_fault_line", "secondary_fault_line", "assembly_presence", "commission_presence", "syndic_presence", "syndic_type", "separatist_presence", "separatist_type", "institute_presence", "guardians_presence", "faction_notes", "cultural_register", "cultural_register_secondary", "ambient_anxiety", "local_pride", "atmospheric_tone", "atmospheric_tone_secondary", "active_situation", "silence_threshold", "silence_topic", "earth_alignment", "earth_proximity", "earth_tension", "primary_archetype", "secondary_archetype", "narrative_notable", "narrative_hook", "generation_priority", "stability_index", "system_volatility", "cultural_corridor", "calibration_note", ] INTEGER_COLUMNS = { "habitable_planet_count", "inhabited_planet_count", "horizon_station", "aperture_count", "gate_connections", "historical_event_age_years", "religious_generation_count", } BODY_SECTION_MAP = { "supply dependency": "supply_dependency", "faction notes": "faction_notes", "silence topic": "silence_topic", "narrative hook": "narrative_hook", "calibration note": "calibration_note", } # --------------------------------------------------------------------------- # Parsing # --------------------------------------------------------------------------- def parse_frontmatter(text: str) -> dict[str, str]: """Extract YAML frontmatter between --- delimiters, ignoring comments.""" match = re.match(r"^---\n(.*?\n)---", text, re.DOTALL) if not match: return {} result = {} for line in match.group(1).split("\n"): line = line.strip() if not line or line.startswith("#"): continue if ":" not in line: continue key, _, value = line.partition(":") key = key.strip() value = value.strip() if value.startswith('"') and value.endswith('"'): value = value[1:-1] elif value.startswith("'") and value.endswith("'"): value = value[1:-1] if value == "~" or value.lower() == "null" or value == "": value = "" result[key] = value return result def parse_body_sections(text: str) -> dict[str, str]: """Extract content from ## sections in the markdown body.""" match = re.match(r"^---\n.*?\n---\n", text, re.DOTALL) body = text[match.end():] if match else text sections: dict[str, str] = {} current_key: str | None = None current_lines: list[str] = [] for line in body.split("\n"): if line.startswith("## "): if current_key is not None: sections[current_key] = "\n".join(current_lines).strip() header = line[3:].strip().lower() current_key = BODY_SECTION_MAP.get(header) current_lines = [] elif current_key is not None: if line.strip().startswith(""): continue current_lines.append(line) if current_key is not None: sections[current_key] = "\n".join(current_lines).strip() return sections # --------------------------------------------------------------------------- # Data loading # --------------------------------------------------------------------------- def load_wiki_pages(): """Load all wiki pages. Returns (systems_dict, warnings, body_count).""" if not WIKI_DIR.is_dir(): return {}, [f"Wiki directory not found: {WIKI_DIR}"], 0 systems = {} warnings = [] body_count = 0 for d in sorted(WIKI_DIR.iterdir()): if not d.is_dir(): continue index_file = d / "index.md" if not index_file.exists(): warnings.append(f"Missing index.md in {d.name}") continue text = index_file.read_text(encoding="utf-8") fm = parse_frontmatter(text) body = parse_body_sections(text) sid = fm.get("system_id") if not sid: warnings.append(f"No system_id in {d.name}/index.md") continue record = {} for col in ALL_COLUMNS: if col in fm and fm[col] != "": record[col] = fm[col] elif col in body and body[col] != "": record[col] = body[col] for key in BODY_SECTION_MAP.values(): if key in body and body[key]: body_count += 1 systems[sid] = record return systems, warnings, body_count def coerce_value(col, val): """Coerce a value for SQLite insertion.""" if val is None: return None if isinstance(val, (int, float)): return val val_str = str(val).strip() if val_str == "" or val_str == "~" or val_str.lower() == "null": return None if col == "horizon_station": lower = val_str.lower() if lower in ("true", "yes", "1"): return 1 if lower in ("false", "no", "0"): return 0 return None if col in INTEGER_COLUMNS: try: return int(val_str) except (ValueError, TypeError): return None return val_str # --------------------------------------------------------------------------- # Sync # --------------------------------------------------------------------------- def sync(cfg): """Parse wiki and upsert into star_systems table.""" wiki_systems, warnings, body_count = load_wiki_pages() if not wiki_systems: return {"ok": False, "error": "No wiki pages found", "warnings": warnings} conn = get_connection(cfg) try: schema_sql = SCHEMA_PATH.read_text() conn.executescript(schema_sql) col_list = ", ".join(ALL_COLUMNS) placeholders = ", ".join(["?"] * len(ALL_COLUMNS)) update_set = ", ".join( f"{col} = excluded.{col}" for col in ALL_COLUMNS if col != "system_id" ) upsert_sql = ( f"INSERT INTO star_systems ({col_list}, synced_at) " f"VALUES ({placeholders}, datetime('now')) " f"ON CONFLICT(system_id) DO UPDATE SET " f"{update_set}, synced_at = datetime('now')" ) upserted = 0 for sid in sorted(wiki_systems.keys()): record = wiki_systems[sid] record["system_id"] = sid values = [coerce_value(col, record.get(col)) for col in ALL_COLUMNS] if values[1] is None: # astronomical_id NOT NULL warnings.append(f"Skipping {sid}: missing astronomical_id") continue conn.execute(upsert_sql, values) upserted += 1 conn.commit() return { "ok": True, "systems_synced": upserted, "body_sections_filled": body_count, "warnings": warnings, "summary": f"Synced {upserted} systems from wiki ({body_count} body sections filled)", } except sqlite3.Error as exc: conn.rollback() return {"ok": False, "error": str(exc), "warnings": warnings} finally: conn.close() def stats(cfg): """Show per-column fill rates.""" conn = get_connection(cfg) try: row = conn.execute("SELECT COUNT(*) as cnt FROM star_systems").fetchone() total = row["cnt"] if total == 0: return {"ok": True, "summary": "No systems in database. Run sync first."} fill = {} for col in ALL_COLUMNS: r = conn.execute( f"SELECT COUNT(*) as cnt FROM star_systems WHERE {col} IS NOT NULL AND {col} != ''" ).fetchone() fill[col] = r["cnt"] total_cells = total * len(ALL_COLUMNS) filled = sum(fill.values()) pct = filled / total_cells * 100 print(f"Star systems: {total}") print(f"Completion: {filled}/{total_cells} ({pct:.1f}%)") print() for col in ALL_COLUMNS: c = fill[col] p = c / total * 100 bar = "#" * int(p / 5) + "." * (20 - int(p / 5)) mark = "+" if p == 100 else " " print(f" {mark} {col:35s} {c:3d}/{total} {bar} {p:.0f}%") return {"ok": True, "total": total, "filled": filled, "pct": round(pct, 1)} except sqlite3.OperationalError as exc: return {"ok": False, "error": f"Table not found — run sync first: {exc}"} finally: conn.close() # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- HELP_TEXT = f"""\ Wiki Sync — wiki frontmatter -> SQLite Usage: wiki_sync.py sync Parse wiki and upsert into SQLite wiki_sync.py stats Show completion stats wiki_sync.py --help Show this help Source: {WIKI_DIR} Schema: {SCHEMA_PATH} DB: {DB_PATH} """ def main(): if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"): print(HELP_TEXT) sys.exit(0) cmd = sys.argv[1] try: cfg = load_config() except (FileNotFoundError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": f"Config error: {exc}"}, indent=2)) sys.exit(1) if cmd == "sync": result = sync(cfg) print(json.dumps(result, indent=2)) elif cmd == "stats": result = stats(cfg) if not result.get("ok"): print(json.dumps(result, indent=2)) else: result = {"ok": False, "error": f"Unknown: {cmd}"} print(json.dumps(result, indent=2)) sys.exit(0 if result.get("ok") else 1) if __name__ == "__main__": main()