jpmschweitzerandClaude Opus 4.6
e5911838c0
Update wiki_sync.py and process-wiki-system-changes to remove astronomical_id references. Add migrate-s-to-gj.py migration script. Update generate-stubs.py and validate-ron for new schema. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
719 lines
24 KiB
Python
Executable File
719 lines
24 KiB
Python
Executable File
#!/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("<!--") and line.strip().endswith("-->"):
|
|
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",
|
|
"<!-- READ-ONLY — generated from systems.db -->",
|
|
"",
|
|
"| | |",
|
|
"|---|---|",
|
|
]
|
|
|
|
# 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<!-- READ-ONLY — regenerated from star-map.json edges -->\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"
|
|
"<!-- READ-ONLY — regenerated from star-map.json edges -->\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": "<!-- Required for manufacturing/transit_logistics base and narrative_notable systems -->",
|
|
"faction_notes": "<!-- Significant local factions not covered by standard columns -->",
|
|
"silence_topic": "<!-- Required for narrative_notable and Institutional Core / Diplomatic Periphery zones -->",
|
|
"narrative_hook": "<!-- Required if narrative_notable = true. One sentence. -->",
|
|
"calibration_note": "<!-- Unusual column combinations requiring justification -->",
|
|
}
|
|
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()
|