chore(tooling): update tooling for GJ ID migration and wiki sync
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>
This commit is contained in:
Claude Opus 4.6
+550
-214
@@ -1,16 +1,20 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Settled Reach Wiki Sync — parse wiki frontmatter into SQLite.
|
||||
Settled Reach Wiki Sync — wiki frontmatter ↔ systems.db
|
||||
|
||||
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.
|
||||
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 the sync logic. Called by tooling/process-wiki-changes.
|
||||
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
|
||||
|
||||
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
|
||||
Database: server/data/systems.db
|
||||
Schema: server/data/systems-schema.sql
|
||||
"""
|
||||
|
||||
import json
|
||||
@@ -24,71 +28,92 @@ from pathlib import Path
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
CONFIG_PATH = SCRIPT_DIR / "config.json"
|
||||
WORKTREE_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
|
||||
SCHEMA_PATH = WORKTREE_ROOT / "db" / "schema.sql"
|
||||
SCHEMA_PATH = WORKTREE_ROOT / "server" / "data" / "systems-schema.sql"
|
||||
WIKI_DIR = WORKTREE_ROOT / "wiki" / "star-systems"
|
||||
DB_PATH = (WORKTREE_ROOT / ".." / "settledreach.db").resolve()
|
||||
DB_PATH = WORKTREE_ROOT / "server" / "data" / "systems.db"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config / DB
|
||||
# DB helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
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"])
|
||||
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)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Column definitions (must match db/schema.sql star_systems table)
|
||||
# Table → column mapping
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
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",
|
||||
]
|
||||
# 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",
|
||||
"horizon_station", "aperture_count", "gate_connections",
|
||||
"historical_event_age_years", "religious_generation_count",
|
||||
"asteroid_belt", "gas_giant", "horizon_station",
|
||||
"aperture_count", "gate_connections",
|
||||
"religious_generation_count", "economic_tier", "population",
|
||||
"stability_index", "narrative_notable",
|
||||
}
|
||||
|
||||
BODY_SECTION_MAP = {
|
||||
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",
|
||||
@@ -98,19 +123,22 @@ BODY_SECTION_MAP = {
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Parsing
|
||||
# Frontmatter parsing (for migration/bootstrap from existing wiki)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def parse_frontmatter(text: str) -> dict[str, str]:
|
||||
"""Extract YAML frontmatter between --- delimiters, ignoring comments."""
|
||||
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 = {}
|
||||
for line in match.group(1).split("\n"):
|
||||
line = line.strip()
|
||||
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:
|
||||
@@ -120,13 +148,64 @@ def parse_frontmatter(text: str) -> dict[str, str]:
|
||||
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 = ""
|
||||
value = None
|
||||
else:
|
||||
result[key] = value
|
||||
continue
|
||||
|
||||
result[key] = value
|
||||
|
||||
@@ -134,7 +213,7 @@ def parse_frontmatter(text: str) -> dict[str, str]:
|
||||
|
||||
|
||||
def parse_body_sections(text: str) -> dict[str, str]:
|
||||
"""Extract content from ## sections in the markdown body."""
|
||||
"""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
|
||||
|
||||
@@ -147,7 +226,7 @@ def parse_body_sections(text: str) -> dict[str, str]:
|
||||
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_key = PROSE_SECTIONS.get(header)
|
||||
current_lines = []
|
||||
elif current_key is not None:
|
||||
if line.strip().startswith("<!--") and line.strip().endswith("-->"):
|
||||
@@ -160,65 +239,16 @@ def parse_body_sections(text: str) -> dict[str, str]:
|
||||
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":
|
||||
if col in BOOLEAN_COLUMNS:
|
||||
lower = val_str.lower()
|
||||
if lower in ("true", "yes", "1"):
|
||||
return 1
|
||||
@@ -232,103 +262,416 @@ def coerce_value(col, val):
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
if col in REAL_COLUMNS:
|
||||
try:
|
||||
return float(val_str)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
return val_str
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Sync
|
||||
# Import from wiki (migration/bootstrap)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def sync(cfg):
|
||||
"""Parse wiki and upsert into star_systems table."""
|
||||
wiki_systems, warnings, body_count = load_wiki_pages()
|
||||
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}"}
|
||||
|
||||
if not wiki_systems:
|
||||
return {"ok": False, "error": "No wiki pages found", "warnings": warnings}
|
||||
conn = get_connection()
|
||||
ensure_schema(conn)
|
||||
|
||||
conn = get_connection(cfg)
|
||||
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:
|
||||
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()
|
||||
return f"{int(pop):,}"
|
||||
except (ValueError, TypeError):
|
||||
return str(pop)
|
||||
|
||||
|
||||
def stats(cfg):
|
||||
"""Show per-column fill rates."""
|
||||
conn = get_connection(cfg)
|
||||
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:
|
||||
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("No systems in database. Run import first.")
|
||||
return {"ok": True, "total": 0}
|
||||
|
||||
print(f"Star systems: {total}")
|
||||
print(f"Completion: {filled}/{total_cells} ({pct:.1f}%)")
|
||||
print(f"Database: {DB_PATH}")
|
||||
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)}
|
||||
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:
|
||||
return {"ok": False, "error": f"Table not found — run sync first: {exc}"}
|
||||
print(f"Table not found — run import first: {exc}")
|
||||
return {"ok": False, "error": str(exc)}
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
@@ -337,45 +680,38 @@ def stats(cfg):
|
||||
# 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():
|
||||
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)
|
||||
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)
|
||||
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":
|
||||
result = stats(cfg)
|
||||
if not result.get("ok"):
|
||||
print(json.dumps(result, indent=2))
|
||||
stats()
|
||||
else:
|
||||
result = {"ok": False, "error": f"Unknown: {cmd}"}
|
||||
print(json.dumps(result, indent=2))
|
||||
|
||||
sys.exit(0 if result.get("ok") else 1)
|
||||
print(json.dumps({"ok": False, "error": f"Unknown: {cmd}"}, indent=2))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
Reference in New Issue
Block a user