feat(db): add wiki sync pipeline and process-wiki-system-changes
New data flow: wiki → star-map.json + SQLite. - tooling/process-wiki-system-changes: unified script that reads wiki, updates star-map.json topology, syncs to SQLite - tooling/db/wiki_sync.py: wiki frontmatter parser + SQLite upsert - generate-stubs.py: updated for full-frontmatter wiki format - Removed extract-csv.py (replaced by SQLite queries) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 4.6
Executable
+382
@@ -0,0 +1,382 @@
|
||||
#!/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("<!--") 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
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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()
|
||||
Executable
+201
@@ -0,0 +1,201 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
process-wiki-system-changes
|
||||
Propagates wiki changes to star-map.json and SQLite.
|
||||
|
||||
Wiki pages (wiki/star-systems/*/index.md) are the single source of truth.
|
||||
This script:
|
||||
1. Reads all wiki frontmatter
|
||||
2. Updates star-map.json nodes (preserving edges/topology)
|
||||
3. Syncs all data into SQLite star_systems table
|
||||
4. Reports completion stats
|
||||
|
||||
Usage:
|
||||
tooling/process-wiki-system-changes Run full sync
|
||||
tooling/process-wiki-system-changes --dry-run Show what would change
|
||||
tooling/process-wiki-system-changes --stats Just show fill rates
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
WORKTREE_ROOT = (SCRIPT_DIR / "..").resolve()
|
||||
STAR_MAP_PATH = WORKTREE_ROOT / "docs" / "design" / "star-map.json"
|
||||
|
||||
# Import wiki_sync for DB operations
|
||||
sys.path.insert(0, str(SCRIPT_DIR / "db"))
|
||||
import wiki_sync
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Star-map.json update
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Fields that star-map.json nodes carry (topology + join keys)
|
||||
STAR_MAP_NODE_FIELDS = [
|
||||
"system_id", "astronomical_id",
|
||||
"gate_topology", "aperture_count", "gate_connections",
|
||||
"hop_distance_from_gateway",
|
||||
]
|
||||
|
||||
|
||||
def update_star_map(wiki_systems: dict) -> dict:
|
||||
"""Update star-map.json node fields from wiki data. Returns summary."""
|
||||
with open(STAR_MAP_PATH, "r") as f:
|
||||
data = json.load(f)
|
||||
|
||||
nodes = data["nodes"]
|
||||
updated = 0
|
||||
added = 0
|
||||
wiki_sids = set(wiki_systems.keys())
|
||||
map_sids = {n["system_id"] for n in nodes}
|
||||
|
||||
# Update existing nodes
|
||||
for node in nodes:
|
||||
sid = node["system_id"]
|
||||
if sid not in wiki_systems:
|
||||
continue
|
||||
|
||||
wiki = wiki_systems[sid]
|
||||
changed = False
|
||||
for field in STAR_MAP_NODE_FIELDS:
|
||||
if field in wiki and wiki[field] != "":
|
||||
val = wiki[field]
|
||||
# Coerce integers
|
||||
if field in ("aperture_count", "gate_connections", "hop_distance_from_gateway"):
|
||||
try:
|
||||
val = int(val)
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
if node.get(field) != val:
|
||||
node[field] = val
|
||||
changed = True
|
||||
|
||||
# Preserve is_gateway flag
|
||||
if changed:
|
||||
updated += 1
|
||||
|
||||
# Add nodes for wiki pages that aren't in star-map yet
|
||||
for sid in wiki_sids - map_sids:
|
||||
wiki = wiki_systems[sid]
|
||||
new_node = {"system_id": sid}
|
||||
for field in STAR_MAP_NODE_FIELDS:
|
||||
if field in wiki and wiki[field] != "":
|
||||
val = wiki[field]
|
||||
if field in ("aperture_count", "gate_connections", "hop_distance_from_gateway"):
|
||||
try:
|
||||
val = int(val)
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
new_node[field] = val
|
||||
nodes.append(new_node)
|
||||
added += 1
|
||||
|
||||
with open(STAR_MAP_PATH, "w") as f:
|
||||
json.dump(data, f, indent=2)
|
||||
f.write("\n")
|
||||
|
||||
return {
|
||||
"nodes": len(nodes),
|
||||
"edges": len(data["edges"]),
|
||||
"updated": updated,
|
||||
"added": added,
|
||||
}
|
||||
|
||||
|
||||
def dry_run(wiki_systems: dict) -> dict:
|
||||
"""Show what would change in star-map.json without writing."""
|
||||
with open(STAR_MAP_PATH, "r") as f:
|
||||
data = json.load(f)
|
||||
|
||||
changes = []
|
||||
for node in data["nodes"]:
|
||||
sid = node["system_id"]
|
||||
if sid not in wiki_systems:
|
||||
continue
|
||||
wiki = wiki_systems[sid]
|
||||
for field in STAR_MAP_NODE_FIELDS:
|
||||
if field in wiki and wiki[field] != "":
|
||||
val = wiki[field]
|
||||
if field in ("aperture_count", "gate_connections", "hop_distance_from_gateway"):
|
||||
try:
|
||||
val = int(val)
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
if node.get(field) != val:
|
||||
changes.append(f" {sid}.{field}: {node.get(field)} -> {val}")
|
||||
|
||||
return {"changes": changes}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Propagate wiki changes to star-map.json and SQLite"
|
||||
)
|
||||
parser.add_argument("--dry-run", action="store_true", help="Show changes without writing")
|
||||
parser.add_argument("--stats", action="store_true", help="Just show fill rates")
|
||||
args = parser.parse_args()
|
||||
|
||||
cfg = wiki_sync.load_config()
|
||||
|
||||
if args.stats:
|
||||
wiki_sync.stats(cfg)
|
||||
return
|
||||
|
||||
# Load wiki
|
||||
wiki_systems, warnings, body_count = wiki_sync.load_wiki_pages()
|
||||
if not wiki_systems:
|
||||
print("ERROR: No wiki pages found")
|
||||
for w in warnings:
|
||||
print(f" {w}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f"Loaded {len(wiki_systems)} wiki pages ({body_count} body sections filled)")
|
||||
if warnings:
|
||||
print(f"Warnings: {len(warnings)}")
|
||||
for w in warnings[:5]:
|
||||
print(f" {w}")
|
||||
if len(warnings) > 5:
|
||||
print(f" ... and {len(warnings) - 5} more")
|
||||
|
||||
if args.dry_run:
|
||||
result = dry_run(wiki_systems)
|
||||
if result["changes"]:
|
||||
print(f"\nstar-map.json changes ({len(result['changes'])}):")
|
||||
for c in result["changes"]:
|
||||
print(c)
|
||||
else:
|
||||
print("\nstar-map.json: no changes")
|
||||
print("\n(dry run — nothing written)")
|
||||
return
|
||||
|
||||
# Step 1: Update star-map.json
|
||||
print("\n--- Updating star-map.json ---")
|
||||
map_result = update_star_map(wiki_systems)
|
||||
print(f"Nodes: {map_result['nodes']}, Edges: {map_result['edges']}")
|
||||
print(f"Updated: {map_result['updated']}, Added: {map_result['added']}")
|
||||
|
||||
# Step 2: Sync to SQLite
|
||||
print("\n--- Syncing to SQLite ---")
|
||||
db_result = wiki_sync.sync(cfg)
|
||||
if db_result["ok"]:
|
||||
print(db_result["summary"])
|
||||
else:
|
||||
print(f"ERROR: {db_result.get('error')}")
|
||||
sys.exit(1)
|
||||
|
||||
# Step 3: Quick stats
|
||||
print("\n--- Completion ---")
|
||||
wiki_sync.stats(cfg)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,213 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
extract-csv.py
|
||||
Extracts all wiki system data into a single CSV file.
|
||||
|
||||
Reads: docs/wiki/systems/*/index.md (YAML frontmatter + markdown body sections)
|
||||
Writes: docs/design/star-systems.csv
|
||||
|
||||
Body sections (## headers) are mapped to columns:
|
||||
Supply Dependency -> supply_dependency
|
||||
Faction Notes -> faction_notes
|
||||
Silence Topic -> silence_topic
|
||||
Narrative Hook -> narrative_hook
|
||||
Calibration Note -> calibration_note
|
||||
"""
|
||||
|
||||
import csv
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
WIKI_DIR = Path(__file__).parent.parent.parent / "docs/wiki/systems"
|
||||
OUTPUT_CSV = Path(__file__).parent.parent.parent / "docs/design/star-systems.csv"
|
||||
|
||||
# CSV column order — matches systems-framework.md section order
|
||||
COLUMNS = [
|
||||
# I. Identity and Location
|
||||
"system_id", "astronomical_id", "proper_name", "system_name", "star_type",
|
||||
"geographic_sector", "geographic_band", "political_zone",
|
||||
# II. Physical Character
|
||||
"habitable_planet_count", "inhabited_planet_count", "asteroid_belt",
|
||||
"gas_giant", "habitability_profile",
|
||||
# III. Gate Infrastructure
|
||||
"horizon_station", "aperture_count", "gate_connections", "gate_topology",
|
||||
"span_gate_network",
|
||||
# IV. Settlement History
|
||||
"settlement_wave", "founding_motivation", "founding_culture_primary",
|
||||
"founding_culture_secondary", "cultural_persistence", "historical_event",
|
||||
"historical_event_age_years", "religious_status", "religious_generation_count",
|
||||
# V. Economic Life
|
||||
"economic_tier", "population", "economic_base_primary",
|
||||
"economic_base_secondary", "distribution_index", "imprint_access",
|
||||
"supply_dependency",
|
||||
# VI. Governance
|
||||
"governance_type", "dominant_faction", "primary_fault_line",
|
||||
"secondary_fault_line",
|
||||
# VII. Faction Presence
|
||||
"assembly_presence", "commission_presence", "syndic_presence", "syndic_type",
|
||||
"separatist_presence", "separatist_type", "institute_presence",
|
||||
"guardians_presence", "faction_notes",
|
||||
# VIII. Cultural Voice
|
||||
"cultural_register", "cultural_register_secondary", "ambient_anxiety",
|
||||
"local_pride", "atmospheric_tone", "atmospheric_tone_secondary",
|
||||
"active_situation", "silence_threshold", "silence_topic",
|
||||
# IX. Political Character
|
||||
"earth_alignment", "earth_proximity", "earth_tension",
|
||||
# X. Narrative Profile
|
||||
"primary_archetype", "secondary_archetype", "narrative_notable",
|
||||
"narrative_hook", "generation_priority",
|
||||
# XI. Content Notes
|
||||
"stability_index", "system_volatility", "cultural_corridor",
|
||||
"calibration_note",
|
||||
]
|
||||
|
||||
# Map markdown section headers to column names
|
||||
BODY_SECTION_MAP = {
|
||||
"supply dependency": "supply_dependency",
|
||||
"faction notes": "faction_notes",
|
||||
"silence topic": "silence_topic",
|
||||
"narrative hook": "narrative_hook",
|
||||
"calibration note": "calibration_note",
|
||||
}
|
||||
|
||||
|
||||
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 {}
|
||||
|
||||
fm_text = match.group(1)
|
||||
result = {}
|
||||
|
||||
for line in fm_text.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()
|
||||
|
||||
# Strip quotes
|
||||
if value.startswith('"') and value.endswith('"'):
|
||||
value = value[1:-1]
|
||||
elif value.startswith("'") and value.endswith("'"):
|
||||
value = value[1:-1]
|
||||
|
||||
# Convert YAML null to empty string
|
||||
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."""
|
||||
# Find where frontmatter ends
|
||||
match = re.match(r"^---\n.*?\n---\n", text, re.DOTALL)
|
||||
if not match:
|
||||
body = text
|
||||
else:
|
||||
body = text[match.end():]
|
||||
|
||||
sections: dict[str, str] = {}
|
||||
current_key: str | None = None
|
||||
current_lines: list[str] = []
|
||||
|
||||
for line in body.split("\n"):
|
||||
if line.startswith("## "):
|
||||
# Save previous section
|
||||
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:
|
||||
# Skip HTML comments and the Derived Fields section
|
||||
if line.strip().startswith("<!--") and line.strip().endswith("-->"):
|
||||
continue
|
||||
current_lines.append(line)
|
||||
|
||||
# Save last section
|
||||
if current_key is not None:
|
||||
sections[current_key] = "\n".join(current_lines).strip()
|
||||
|
||||
return sections
|
||||
|
||||
|
||||
def main():
|
||||
wiki_dirs = sorted(WIKI_DIR.iterdir())
|
||||
rows: list[dict[str, str]] = []
|
||||
errors: list[str] = []
|
||||
|
||||
for d in wiki_dirs:
|
||||
if not d.is_dir():
|
||||
continue
|
||||
index_file = d / "index.md"
|
||||
if not index_file.exists():
|
||||
errors.append(f"Missing index.md in {d.name}")
|
||||
continue
|
||||
|
||||
text = index_file.read_text()
|
||||
|
||||
# Parse frontmatter
|
||||
fm = parse_frontmatter(text)
|
||||
|
||||
# Parse body sections
|
||||
body = parse_body_sections(text)
|
||||
|
||||
# Merge into row
|
||||
row: dict[str, str] = {}
|
||||
for col in COLUMNS:
|
||||
if col in fm:
|
||||
row[col] = fm[col]
|
||||
elif col in body:
|
||||
row[col] = body[col]
|
||||
else:
|
||||
row[col] = ""
|
||||
|
||||
rows.append(row)
|
||||
|
||||
# Sort by system_id
|
||||
rows.sort(key=lambda r: r.get("system_id", ""))
|
||||
|
||||
# Write CSV
|
||||
OUTPUT_CSV.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(OUTPUT_CSV, "w", newline="") as f:
|
||||
writer = csv.DictWriter(f, fieldnames=COLUMNS)
|
||||
writer.writeheader()
|
||||
writer.writerows(rows)
|
||||
|
||||
# Summary
|
||||
non_empty_counts = {col: 0 for col in COLUMNS}
|
||||
for row in rows:
|
||||
for col in COLUMNS:
|
||||
if row.get(col, ""):
|
||||
non_empty_counts[col] += 1
|
||||
|
||||
print(f"Extracted {len(rows)} systems to {OUTPUT_CSV}")
|
||||
if errors:
|
||||
print(f"\nErrors ({len(errors)}):")
|
||||
for e in errors:
|
||||
print(f" {e}")
|
||||
|
||||
# Completion stats
|
||||
total_cells = len(rows) * len(COLUMNS)
|
||||
filled_cells = sum(non_empty_counts.values())
|
||||
print(f"\nCompletion: {filled_cells}/{total_cells} cells filled ({filled_cells/total_cells*100:.1f}%)")
|
||||
print(f"\nColumn fill rates:")
|
||||
for col in COLUMNS:
|
||||
count = non_empty_counts[col]
|
||||
pct = count / len(rows) * 100 if rows else 0
|
||||
bar = "█" * int(pct / 5) + "░" * (20 - int(pct / 5))
|
||||
status = "✓" if pct == 100 else " "
|
||||
print(f" {status} {col:35s} {count:3d}/{len(rows)} {bar} {pct:.0f}%")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,21 +1,24 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
generate-stubs.py
|
||||
Generates wiki stub markdown files for all 300 star systems from star-map.json.
|
||||
Generates wiki stub markdown files for all star systems from star-map.json.
|
||||
|
||||
Each system gets: docs/wiki/systems/{gj_dir}/index.md
|
||||
Each system gets: wiki/star-systems/{gj_dir}/index.md
|
||||
where {gj_dir} is the astronomical_id with spaces replaced by hyphens.
|
||||
|
||||
Pre-fills all fields available from star-map.json; marks the rest as null (~).
|
||||
Safe to re-run: skips files that already exist.
|
||||
Wiki pages are the SINGLE SOURCE OF TRUTH for all per-system data.
|
||||
star-map.json provides initial values for topology/identity fields;
|
||||
once written to wiki, the wiki page is authoritative.
|
||||
|
||||
Safe to re-run: skips files that already exist (use --force to overwrite).
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
STAR_MAP_PATH = Path(__file__).parent.parent.parent / "docs/design/star-map.json"
|
||||
WIKI_DIR = Path(__file__).parent.parent.parent / "docs/wiki/systems"
|
||||
WIKI_DIR = Path(__file__).parent.parent.parent / "wiki/star-systems"
|
||||
|
||||
|
||||
def gj_to_dirname(astronomical_id: str) -> str:
|
||||
@@ -24,11 +27,12 @@ def gj_to_dirname(astronomical_id: str) -> str:
|
||||
|
||||
|
||||
def render_stub(node: dict, adjacent_ids: list[str]) -> str:
|
||||
"""Render a wiki stub markdown file for a single system."""
|
||||
"""Render a wiki stub with ALL fields in frontmatter."""
|
||||
sid = node["system_id"]
|
||||
astro_id = node["astronomical_id"]
|
||||
proper_name = node.get("proper_name")
|
||||
system_name = proper_name if proper_name else node.get("system_name", f"System {sid}")
|
||||
|
||||
star_type = node.get("star_type", "~")
|
||||
sector = node.get("geographic_sector", "~")
|
||||
band = node.get("geographic_band", "~")
|
||||
@@ -40,35 +44,20 @@ def render_stub(node: dict, adjacent_ids: list[str]) -> str:
|
||||
topology = node.get("gate_topology", "~")
|
||||
earth_prox = node.get("earth_proximity", "~")
|
||||
hop = node.get("hop_distance_from_gateway", "~")
|
||||
dist_ly = node.get("dist_ly", 0)
|
||||
spectral = node.get("spectral_class", "~")
|
||||
|
||||
# Format proper_name for YAML (null or quoted string)
|
||||
# Format proper_name for YAML
|
||||
proper_yaml = f'"{proper_name}"' if proper_name else "~"
|
||||
|
||||
# Adjacent systems list for derived section
|
||||
# Adjacent systems list
|
||||
adjacent_str = ", ".join(adjacent_ids) if adjacent_ids else "none"
|
||||
|
||||
# Approximate settlement age from wave
|
||||
wave_ages = {
|
||||
"wave_1": "~550 years",
|
||||
"wave_2": "~400 years",
|
||||
"wave_3": "~200 years",
|
||||
"wave_4": "~70 years",
|
||||
"wave_5": "~20 years",
|
||||
"unsettled": "unsettled",
|
||||
}
|
||||
age_str = wave_ages.get(wave, "unknown")
|
||||
|
||||
# Compound zone display
|
||||
compound = f"{sector} / {band} / {zone}"
|
||||
|
||||
return f"""---
|
||||
# ============================================================
|
||||
# {astro_id} — {system_name}
|
||||
# Star Systems Wiki — The Settled Reach
|
||||
# ============================================================
|
||||
# Auto-generated stub. Pre-filled fields from star-map.json.
|
||||
# Fields marked ~ require authoring.
|
||||
# ============================================================
|
||||
# Source of truth for all per-system data.
|
||||
# Topology graph (edges) lives in star-map.json.
|
||||
|
||||
# I. Identity and Location
|
||||
system_id: {sid}
|
||||
@@ -76,6 +65,8 @@ astronomical_id: "{astro_id}"
|
||||
proper_name: {proper_yaml}
|
||||
system_name: "{system_name}"
|
||||
star_type: {star_type}
|
||||
spectral_class: "{spectral}"
|
||||
dist_ly: {dist_ly}
|
||||
geographic_sector: {sector}
|
||||
geographic_band: {band}
|
||||
political_zone: {zone}
|
||||
@@ -174,16 +165,24 @@ cultural_corridor: ~
|
||||
## Calibration Note
|
||||
<!-- Unusual column combinations requiring justification -->
|
||||
|
||||
## Derived Fields
|
||||
<!-- Auto-generated from star-map.json, do not edit manually -->
|
||||
- **Compound Zone:** {compound}
|
||||
- **Settlement Age:** {age_str}
|
||||
## Topology
|
||||
<!-- READ-ONLY — regenerated from star-map.json edges -->
|
||||
- **Hop Distance from Gateway:** {hop}
|
||||
- **Adjacent Systems:** {adjacent_str}
|
||||
"""
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Generate wiki stub files from star-map.json"
|
||||
)
|
||||
parser.add_argument(
|
||||
"--force",
|
||||
action="store_true",
|
||||
help="Overwrite existing index.md files (never touches subdirectories)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
with open(STAR_MAP_PATH, "r") as f:
|
||||
data = json.load(f)
|
||||
|
||||
@@ -201,6 +200,7 @@ def main():
|
||||
adj[sid].sort()
|
||||
|
||||
created = 0
|
||||
overwritten = 0
|
||||
skipped = 0
|
||||
|
||||
for node in sorted(nodes, key=lambda n: n["system_id"]):
|
||||
@@ -210,19 +210,23 @@ def main():
|
||||
outdir = WIKI_DIR / dirname
|
||||
outfile = outdir / "index.md"
|
||||
|
||||
if outfile.exists():
|
||||
if outfile.exists() and not args.force:
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
existed = outfile.exists()
|
||||
outdir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
adjacent = adj.get(sid, [])
|
||||
content = render_stub(node, adjacent)
|
||||
|
||||
outfile.write_text(content)
|
||||
created += 1
|
||||
if existed:
|
||||
overwritten += 1
|
||||
else:
|
||||
created += 1
|
||||
|
||||
print(f"Wiki stubs generated: {created} created, {skipped} skipped (already exist)")
|
||||
print(f"Wiki stubs: {created} created, {overwritten} overwritten, {skipped} skipped")
|
||||
print(f"Output directory: {WIKI_DIR}")
|
||||
|
||||
# Print a few examples
|
||||
|
||||
Reference in New Issue
Block a user