#!/usr/bin/env python3 """ Settled Reach Wiki Sync — wiki frontmatter ↔ systems.db Data flow: - DB owns structured fields (identity, gates, history, economy, factions, culture) - Wiki owns authored prose (supply_dependency, faction_notes, silence_topic, narrative_hook, calibration_note) - Wiki pages are generated from DB (structured infobox) + prose (authored body) This module provides: - import_from_wiki(): parse wiki frontmatter into systems.db (migration/bootstrap) - generate_wiki(): render wiki pages from DB + existing prose - stats(): show completion stats Database: server/data/systems.db Schema: server/data/systems-schema.sql """ import json import re import sqlite3 import sys from pathlib import Path # --------------------------------------------------------------------------- # Paths # --------------------------------------------------------------------------- SCRIPT_DIR = Path(__file__).resolve().parent WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve() SCHEMA_PATH = WORKTREE_ROOT / "server" / "data" / "systems-schema.sql" WIKI_DIR = WORKTREE_ROOT / "wiki" / "star-systems" DB_PATH = WORKTREE_ROOT / "server" / "data" / "systems.db" # --------------------------------------------------------------------------- # DB helpers # --------------------------------------------------------------------------- def get_connection(): conn = sqlite3.connect(str(DB_PATH)) conn.execute("PRAGMA journal_mode=WAL;") conn.execute("PRAGMA foreign_keys=ON;") conn.row_factory = sqlite3.Row return conn def ensure_schema(conn): schema_sql = SCHEMA_PATH.read_text() conn.executescript(schema_sql) # --------------------------------------------------------------------------- # Table → column mapping # --------------------------------------------------------------------------- # Maps each table to its columns (excluding system_id which is always the key) TABLE_COLUMNS = { "star_systems": [ "proper_name", "system_name", "star_type", "spectral_class", "dist_ly", "geographic_sector", "geographic_band", "political_zone", "habitable_planet_count", "inhabited_planet_count", "asteroid_belt", "gas_giant", "habitability_profile", "earth_alignment", "earth_proximity", "earth_tension", "stability_index", "system_volatility", "cultural_corridor", "generation_priority", ], "system_gates": [ "horizon_station", "aperture_count", "gate_connections", "gate_topology", "span_gate_network", ], "system_history": [ "settlement_wave", "founding_motivation", "founding_culture_primary", "founding_culture_secondary", "cultural_persistence", "religious_status", "religious_generation_count", ], "system_economy": [ "economic_tier", "population", "economic_base_primary", "economic_base_secondary", "distribution_index", "imprint_access", ], "system_factions": [ "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", ], "system_culture": [ "cultural_register", "cultural_register_secondary", "ambient_anxiety", "local_pride", "atmospheric_tone", "atmospheric_tone_secondary", "active_situation", "silence_threshold", "primary_archetype", "secondary_archetype", "narrative_notable", ], } # All structured columns (flat list for frontmatter parsing) ALL_STRUCTURED_COLUMNS = set() for cols in TABLE_COLUMNS.values(): ALL_STRUCTURED_COLUMNS.update(cols) ALL_STRUCTURED_COLUMNS.add("system_id") INTEGER_COLUMNS = { "habitable_planet_count", "inhabited_planet_count", "asteroid_belt", "gas_giant", "horizon_station", "aperture_count", "gate_connections", "religious_generation_count", "economic_tier", "population", "stability_index", "narrative_notable", } REAL_COLUMNS = {"dist_ly"} BOOLEAN_COLUMNS = {"asteroid_belt", "gas_giant", "horizon_station", "narrative_notable"} PROSE_SECTIONS = { "supply dependency": "supply_dependency", "faction notes": "faction_notes", "silence topic": "silence_topic", "narrative hook": "narrative_hook", "calibration note": "calibration_note", } # --------------------------------------------------------------------------- # Frontmatter parsing (for migration/bootstrap from existing wiki) # --------------------------------------------------------------------------- def parse_frontmatter(text: str) -> dict: """Extract YAML frontmatter between --- delimiters.""" match = re.match(r"^---\n(.*?\n)---", text, re.DOTALL) if not match: return {} result = {} lines = match.group(1).split("\n") i = 0 while i < len(lines): line = lines[i].strip() i += 1 if not line or line.startswith("#"): continue if ":" not in line: continue key, _, value = line.partition(":") key = key.strip() value = value.strip() # YAML list (historical_events) if key == "historical_events": items = [] if value == "[]": result[key] = items continue while i < len(lines): next_line = lines[i] stripped = next_line.strip() if not stripped or (not next_line.startswith(" ") and not next_line.startswith("\t")): break if stripped.startswith("- "): item = {} item_content = stripped[2:].strip() if ":" in item_content: ik, _, iv = item_content.partition(":") iv = iv.strip() if iv == "~" or iv.lower() == "null": iv = None else: try: iv = int(iv) except (ValueError, TypeError): pass item[ik.strip()] = iv items.append(item) i += 1 elif stripped.startswith("#"): break elif ":" in stripped: ik, _, iv = stripped.partition(":") iv = iv.strip() if iv == "~" or iv.lower() == "null": iv = None else: try: iv = int(iv) except (ValueError, TypeError): pass if items: items[-1][ik.strip()] = iv i += 1 else: break result[key] = items continue # Strip quotes 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 = None else: result[key] = value continue result[key] = value return result def parse_body_sections(text: str) -> dict[str, str]: """Extract authored prose 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 = PROSE_SECTIONS.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 def coerce_value(col, val): """Coerce a value for SQLite insertion.""" if val is None: return None val_str = str(val).strip() if val_str == "" or val_str == "~" or val_str.lower() == "null": return None if col in BOOLEAN_COLUMNS: 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 if col in REAL_COLUMNS: try: return float(val_str) except (ValueError, TypeError): return None return val_str # --------------------------------------------------------------------------- # Import from wiki (migration/bootstrap) # --------------------------------------------------------------------------- def import_from_wiki(): """Parse all wiki frontmatter and populate systems.db.""" if not WIKI_DIR.is_dir(): return {"ok": False, "error": f"Wiki directory not found: {WIKI_DIR}"} conn = get_connection() ensure_schema(conn) warnings = [] imported = 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) sid = fm.get("system_id") if not sid: warnings.append(f"No system_id in {d.name}/index.md") continue try: # Upsert into each table for table, columns in TABLE_COLUMNS.items(): all_cols = ["system_id"] + columns if table == "star_systems": all_cols.append("updated_at") values = [sid] for col in columns: values.append(coerce_value(col, fm.get(col))) if table == "star_systems": values.append(None) # updated_at = DEFAULT placeholders = ", ".join(["?"] * len(all_cols)) col_list = ", ".join(all_cols) if table == "star_systems": update_set = ", ".join( f"{c} = excluded.{c}" for c in columns ) + ", updated_at = datetime('now')" else: update_set = ", ".join( f"{c} = excluded.{c}" for c in columns ) sql = ( f"INSERT INTO {table} ({col_list}) " f"VALUES ({placeholders}) " f"ON CONFLICT(system_id) DO UPDATE SET {update_set}" ) conn.execute(sql, values) # Historical events events = fm.get("historical_events", []) if isinstance(events, list): conn.execute( "DELETE FROM historical_events WHERE system_id = ?", (sid,) ) for idx, event in enumerate(events): if isinstance(event, dict): conn.execute( "INSERT INTO historical_events " "(system_id, event_type, age_years, sort_order) " "VALUES (?, ?, ?, ?)", (sid, event.get("type", ""), event.get("age_years"), idx), ) imported += 1 except sqlite3.Error as exc: warnings.append(f"Error on {sid}: {exc}") conn.commit() conn.close() return { "ok": True, "systems_imported": imported, "warnings": warnings, "summary": f"Imported {imported} systems into {DB_PATH}", } # --------------------------------------------------------------------------- # Wiki page generation (DB → wiki infobox) # --------------------------------------------------------------------------- def _format_population(pop): """Format population with commas.""" if pop is None: return "~" try: return f"{int(pop):,}" except (ValueError, TypeError): return str(pop) def generate_infobox(conn, sid: str) -> str: """Generate the READ-ONLY infobox markdown for a system.""" s = conn.execute("SELECT * FROM star_systems WHERE system_id = ?", (sid,)).fetchone() if not s: return "" g = conn.execute("SELECT * FROM system_gates WHERE system_id = ?", (sid,)).fetchone() h = conn.execute("SELECT * FROM system_history WHERE system_id = ?", (sid,)).fetchone() e = conn.execute("SELECT * FROM system_economy WHERE system_id = ?", (sid,)).fetchone() f = conn.execute("SELECT * FROM system_factions WHERE system_id = ?", (sid,)).fetchone() c = conn.execute("SELECT * FROM system_culture WHERE system_id = ?", (sid,)).fetchone() events = conn.execute( "SELECT event_type, age_years FROM historical_events " "WHERE system_id = ? ORDER BY sort_order", (sid,) ).fetchall() lines = [ "## System Profile", "", "", "| | |", "|---|---|", ] # Star spectral = s["spectral_class"] or s["star_type"] or "~" dist = f"{s['dist_ly']:.1f} ly" if s["dist_ly"] is not None else "~" lines.append(f"| **Star** | {spectral} · {dist} |") # Bodies hab = s["habitable_planet_count"] inh = s["inhabited_planet_count"] if hab is not None or inh is not None: hab_s = f"{hab} habitable" if hab is not None else "~" inh_s = f"{inh} inhabited" if inh is not None else "~" lines.append(f"| **Bodies** | {hab_s} · {inh_s} |") # Gates if g: ap = g["aperture_count"] if g["aperture_count"] is not None else "~" topo = g["gate_topology"] or "~" lines.append(f"| **Gates** | {ap} aperture · {topo} |") # Settlement if h: wave = h["settlement_wave"] or "~" lines.append(f"| **Settlement** | {wave} |") # Historical events if events: ev_parts = [] for ev in events: age = f"{ev['age_years']}y" if ev["age_years"] is not None else "~" ev_parts.append(f"{ev['event_type']} ({age})") lines.append(f"| **History** | {', '.join(ev_parts)} |") # Economy if e and (e["population"] is not None or e["economic_tier"] is not None): pop = _format_population(e["population"]) tier = f"Tier {e['economic_tier']}" if e["economic_tier"] is not None else "~" base_parts = [] if e["economic_base_primary"]: base_parts.append(e["economic_base_primary"]) if e["economic_base_secondary"]: base_parts.append(e["economic_base_secondary"]) base = " / ".join(base_parts) if base_parts else "~" if e["population"] is not None: lines.append(f"| **Population** | {pop} |") lines.append(f"| **Economy** | {tier} · {base} |") # Governance if f and (f["governance_type"] or f["dominant_faction"]): gov = f["governance_type"] or "~" dom = f["dominant_faction"] or "~" lines.append(f"| **Governance** | {gov} · {dom} |") # Culture if c and c["atmospheric_tone"]: tone = c["atmospheric_tone"] if c["atmospheric_tone_secondary"]: tone += f" / {c['atmospheric_tone_secondary']}" lines.append(f"| **Atmosphere** | {tone} |") # Stability stab = s["stability_index"] if stab is not None: vol = s["system_volatility"] or "~" lines.append(f"| **Stability** | {stab} · {vol} |") # Political if s["earth_alignment"]: lines.append(f"| **Earth Alignment** | {s['earth_alignment']} |") lines.append("") return "\n".join(lines) def generate_topology(conn, sid: str, star_map_path: Path) -> str: """Generate the READ-ONLY topology section from star-map.json.""" try: with open(star_map_path) as f: data = json.load(f) except (FileNotFoundError, json.JSONDecodeError): return "## Topology\n\n" # Find hop distance hop = "~" for node in data["nodes"]: if node["system_id"] == sid: hop = node.get("hop_distance_from_gateway", "~") break # Find adjacent systems adjacent = set() for a, b in data["edges"]: if a == sid: adjacent.add(b) elif b == sid: adjacent.add(a) adj_str = ", ".join(sorted(adjacent)) if adjacent else "none" return ( "## Topology\n" "\n" f"- **Hop Distance from Gateway:** {hop}\n" f"- **Adjacent Systems:** {adj_str}\n" ) def generate_wiki_page(conn, sid: str, star_map_path: Path) -> str: """Generate a complete wiki page from DB + existing prose.""" s = conn.execute("SELECT * FROM star_systems WHERE system_id = ?", (sid,)).fetchone() if not s: return "" system_name = s["proper_name"] or s["system_name"] or f"System {sid}" star_type = s["star_type"] or "~" sector = s["geographic_sector"] or "~" # Title block title = f"# {system_name}\n**{sid}** | {star_type}-type | {sector}\n\n---\n\n" # Infobox infobox = generate_infobox(conn, sid) # Topology topology = generate_topology(conn, sid, star_map_path) return title + infobox + "\n---\n\n" + topology def generate_wiki(prose_only_update=False): """Generate/update wiki pages from DB. For each system: - Generates the title + infobox + topology (from DB/star-map) - Preserves existing authored prose sections from the wiki file - Writes the combined result back """ conn = get_connection() star_map_path = WORKTREE_ROOT / "docs" / "design" / "star-map.json" systems = conn.execute("SELECT system_id FROM star_systems ORDER BY system_id").fetchall() updated = 0 created = 0 for row in systems: sid = row["system_id"] dirname = sid.replace(" ", "-") outdir = WIKI_DIR / dirname outfile = outdir / "index.md" # Read existing prose if file exists existing_prose = {} if outfile.exists(): text = outfile.read_text(encoding="utf-8") existing_prose = parse_body_sections(text) # Generate structured parts s = conn.execute("SELECT * FROM star_systems WHERE system_id = ?", (sid,)).fetchone() system_name = s["proper_name"] or s["system_name"] or f"System {sid}" star_type = s["star_type"] or "~" sector = s["geographic_sector"] or "~" infobox = generate_infobox(conn, sid) topology = generate_topology(conn, sid, star_map_path) # Build page page_lines = [ f"# {system_name}", f"**{sid}** | {star_type}-type | {sector}", "", "---", "", infobox, "---", "", ] # Prose sections — preserve existing content or leave placeholder for header, key in PROSE_SECTIONS.items(): section_title = header.title() content = existing_prose.get(key, "") page_lines.append(f"## {section_title}") if content: page_lines.append("") page_lines.append(content) else: comment_map = { "supply_dependency": "", "faction_notes": "", "silence_topic": "", "narrative_hook": "", "calibration_note": "", } page_lines.append(comment_map.get(key, "")) page_lines.append("") # Topology (always last) page_lines.append(topology) outdir.mkdir(parents=True, exist_ok=True) outfile.write_text("\n".join(page_lines), encoding="utf-8") if outfile.exists(): updated += 1 else: created += 1 conn.close() return { "ok": True, "updated": updated, "created": created, "summary": f"Generated {updated + created} wiki pages", } # --------------------------------------------------------------------------- # Stats # --------------------------------------------------------------------------- def stats(): """Show per-table fill rates.""" conn = get_connection() try: row = conn.execute("SELECT COUNT(*) as cnt FROM star_systems").fetchone() total = row["cnt"] if total == 0: print("No systems in database. Run import first.") return {"ok": True, "total": 0} print(f"Star systems: {total}") print(f"Database: {DB_PATH}") print() total_cells = 0 filled_cells = 0 for table, columns in TABLE_COLUMNS.items(): table_filled = 0 table_total = total * len(columns) for col in columns: r = conn.execute( f"SELECT COUNT(*) as cnt FROM {table} WHERE {col} IS NOT NULL" ).fetchone() table_filled += r["cnt"] total_cells += table_total filled_cells += table_filled pct = table_filled / table_total * 100 if table_total > 0 else 0 bar = "#" * int(pct / 5) + "." * (20 - int(pct / 5)) print(f" {table:20s} {table_filled:4d}/{table_total:4d} {bar} {pct:.0f}%") # Historical events ev_count = conn.execute("SELECT COUNT(*) as cnt FROM historical_events").fetchone()["cnt"] ev_systems = conn.execute( "SELECT COUNT(DISTINCT system_id) as cnt FROM historical_events" ).fetchone()["cnt"] print(f" {'historical_events':20s} {ev_count} events across {ev_systems} systems") overall = filled_cells / total_cells * 100 if total_cells > 0 else 0 print(f"\nOverall: {filled_cells}/{total_cells} ({overall:.1f}%)") return {"ok": True, "total": total, "filled": filled_cells, "pct": round(overall, 1)} except sqlite3.OperationalError as exc: print(f"Table not found — run import first: {exc}") return {"ok": False, "error": str(exc)} finally: conn.close() # --------------------------------------------------------------------------- # CLI # --------------------------------------------------------------------------- def main(): help_text = f"""\ Wiki Sync — systems.db ↔ wiki pages Usage: wiki_sync.py import Parse wiki frontmatter into systems.db (bootstrap) wiki_sync.py generate Generate wiki pages from systems.db wiki_sync.py stats Show completion stats wiki_sync.py --help Show this help Database: {DB_PATH} Schema: {SCHEMA_PATH} Wiki: {WIKI_DIR} """ if len(sys.argv) < 2 or sys.argv[1] in ("--help", "-h", "help"): print(help_text) sys.exit(0) cmd = sys.argv[1] if cmd == "import": result = import_from_wiki() print(json.dumps(result, indent=2)) elif cmd == "generate": result = generate_wiki() print(json.dumps(result, indent=2)) elif cmd == "stats": stats() else: print(json.dumps({"ok": False, "error": f"Unknown: {cmd}"}, indent=2)) sys.exit(1) if __name__ == "__main__": main()