refactor(tooling): T-1290 — the wiki domain, and the renderer that must not run
`reach wiki stats` and `reach wiki gttr-hook` replace tooling/db/wiki_sync.py and populate_gttr_hook.py. Both are output-identical to the originals: `stats` byte-for-byte, and all 301 extracted GTTR hooks line-for-line. wiki_sync.py moved whole, but generate_wiki() and import_from_wiki() are NOT verbs. Before porting, the old `--generate` was run against a clean tree to get a parity baseline. It changed all 301 system pages, +940 / -10,761, and was reverted at once. It deletes the Celestial Bodies / Stations blocks (owned by the Rust atlas sync, which it does not know about), deletes the Industries / Exports / Imports rows (nothing writes those any more), and rewrites star types where systems.db and the pages disagree. D-262, CLAUDE.md and the wiki skill all described it as the routine, prose-preserving render. CLAUDE.md and the skill now say not to run it; D-262 needs amending — T-1292. Provenance moves to tooling/archive/, with a README naming what each script did and why it is not run: - pql-migrate/ (the T-1271 ruling) - wiki-bootstrap/: assign-astro-ids + its catalog, migrate-s-to-gj, patch-core-sector (hardcodes a dead path), fill-missing-globes, generate-stubs and find-stubs (finds 0 stubs — Phase 1 is done), backfill_cultural_corridor (a raw systems.db patch script, outside D-262), and process-wiki-system-changes, whose last step is the destructive render Also: - stats() printed "run import first" and exited 0 when a table was missing; it now fails with a remedy. generate_wiki() counted created pages after writing them, so `created` was always 0. - tooling/godot-cold-parse and godot-parse-sweep were never retired after T-1283, and the pr-process skill still told agents to run them. Removed; the skill and parse_sweep.gd now name the reach verbs. - systems.db re-stamped: schema comments changed, and the stamp records the schema file's SHA for tamper detection. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
@@ -1,162 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Backfill star_systems.cultural_corridor and bodies.cultural_corridor.
|
||||
|
||||
The schema has a `cultural_corridor` column on both tables, but
|
||||
`wiki_sync.py` never populated it from the wiki index.md files — the
|
||||
cultural/geographic identity of each system lives in
|
||||
`star_systems.geographic_sector` instead (values: core, north_reach,
|
||||
south_reach, east_reach, west_reach, deep_frontier). These two fields
|
||||
refer to the same concept: which arc of the reach the system belongs
|
||||
to. Leaving `cultural_corridor` NULL on 99%+ of rows defeats every
|
||||
downstream consumer that actually wants to filter by corridor
|
||||
(the atlas UI, narrative tools, and the server generation cascade).
|
||||
|
||||
This script treats `geographic_sector` as the source of truth and
|
||||
copies it into `cultural_corridor`:
|
||||
|
||||
star_systems.cultural_corridor := star_systems.geographic_sector
|
||||
WHERE cultural_corridor IS NULL
|
||||
|
||||
bodies.cultural_corridor := parent star_systems.cultural_corridor
|
||||
WHERE bodies.cultural_corridor IS NULL
|
||||
|
||||
It is safe to re-run — idempotent, NULL-only updates, explicit
|
||||
transaction wrapper so a crash never leaves a half-populated state.
|
||||
Run it after any `wiki_sync.py` pass that creates fresh systems.db
|
||||
rows.
|
||||
|
||||
Usage:
|
||||
tooling/db/backfill_cultural_corridor.py
|
||||
tooling/db/backfill_cultural_corridor.py --db path/to/systems.db
|
||||
tooling/db/backfill_cultural_corridor.py --dry-run
|
||||
|
||||
Decisions: D-191 (atlas pipeline — downstream consumer)
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
REPO_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
|
||||
DB_PATH = REPO_ROOT / "server" / "data" / "systems.db"
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Backfill cultural_corridor on star_systems and bodies"
|
||||
)
|
||||
parser.add_argument("--db", default=str(DB_PATH), help="Path to systems.db")
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Report counts without writing",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
db_path = Path(args.db)
|
||||
if not db_path.exists():
|
||||
print(f"error: {db_path} not found", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
conn = sqlite3.connect(str(db_path))
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
|
||||
# Counts before.
|
||||
before_systems_null = conn.execute(
|
||||
"SELECT COUNT(*) FROM star_systems WHERE cultural_corridor IS NULL"
|
||||
).fetchone()[0]
|
||||
before_bodies_null = conn.execute(
|
||||
"SELECT COUNT(*) FROM bodies WHERE cultural_corridor IS NULL"
|
||||
).fetchone()[0]
|
||||
|
||||
print("\n cultural_corridor backfill")
|
||||
print(f" DB: {db_path}")
|
||||
if args.dry_run:
|
||||
print(" Mode: DRY RUN")
|
||||
print()
|
||||
print(" Before:")
|
||||
print(f" star_systems.cultural_corridor NULL: {before_systems_null}")
|
||||
print(f" bodies.cultural_corridor NULL: {before_bodies_null}")
|
||||
|
||||
conn.execute("BEGIN")
|
||||
try:
|
||||
# 1. Star systems — copy geographic_sector into cultural_corridor
|
||||
# where the latter is still NULL. If geographic_sector is also
|
||||
# NULL, leave cultural_corridor NULL — there is nothing to
|
||||
# copy and a bogus placeholder is worse than honest NULL.
|
||||
sys_rows_updated = conn.execute(
|
||||
"""
|
||||
UPDATE star_systems
|
||||
SET cultural_corridor = geographic_sector
|
||||
WHERE cultural_corridor IS NULL
|
||||
AND geographic_sector IS NOT NULL
|
||||
"""
|
||||
).rowcount
|
||||
|
||||
# 2. Bodies — inherit from the parent star_systems row.
|
||||
body_rows_updated = conn.execute(
|
||||
"""
|
||||
UPDATE bodies
|
||||
SET cultural_corridor = (
|
||||
SELECT s.cultural_corridor
|
||||
FROM star_systems s
|
||||
WHERE s.system_id = bodies.system_id
|
||||
)
|
||||
WHERE cultural_corridor IS NULL
|
||||
AND EXISTS (
|
||||
SELECT 1
|
||||
FROM star_systems s
|
||||
WHERE s.system_id = bodies.system_id
|
||||
AND s.cultural_corridor IS NOT NULL
|
||||
)
|
||||
"""
|
||||
).rowcount
|
||||
|
||||
if args.dry_run:
|
||||
conn.rollback()
|
||||
print()
|
||||
print(" Would update:")
|
||||
print(f" star_systems: {sys_rows_updated}")
|
||||
print(f" bodies: {body_rows_updated}")
|
||||
print("\n Dry run — no changes written.")
|
||||
else:
|
||||
conn.commit()
|
||||
print()
|
||||
print(" Updated:")
|
||||
print(f" star_systems: {sys_rows_updated}")
|
||||
print(f" bodies: {body_rows_updated}")
|
||||
|
||||
# Counts after.
|
||||
after_systems_null = conn.execute(
|
||||
"SELECT COUNT(*) FROM star_systems WHERE cultural_corridor IS NULL"
|
||||
).fetchone()[0]
|
||||
after_bodies_null = conn.execute(
|
||||
"SELECT COUNT(*) FROM bodies WHERE cultural_corridor IS NULL"
|
||||
).fetchone()[0]
|
||||
print()
|
||||
print(" After:")
|
||||
print(f" star_systems.cultural_corridor NULL: {after_systems_null}")
|
||||
print(f" bodies.cultural_corridor NULL: {after_bodies_null}")
|
||||
|
||||
# Show the distribution so the outcome is visible.
|
||||
print()
|
||||
print(" star_systems.cultural_corridor distribution:")
|
||||
for corridor, count in conn.execute(
|
||||
"SELECT cultural_corridor, COUNT(*) FROM star_systems "
|
||||
"GROUP BY cultural_corridor ORDER BY COUNT(*) DESC"
|
||||
).fetchall():
|
||||
print(f" {corridor!r}: {count}")
|
||||
except BaseException:
|
||||
conn.rollback()
|
||||
conn.close()
|
||||
raise
|
||||
|
||||
conn.close()
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,215 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Extract a short narrative hook from every wiki/star-systems/<slug>/gttr.md
|
||||
and store it on `star_systems.gttr_hook`.
|
||||
|
||||
The GTTR ("Drifter's Guide to the Reach") files follow a consistent
|
||||
format: an H1 title, then a paragraph that opens with `**NAME**` followed
|
||||
by a 1-3 sentence characterisation. That characterisation is the hook —
|
||||
per D-191 §4 and the #833 naming pipeline it is the single strongest
|
||||
cultural signal available for each system, capturing things like:
|
||||
|
||||
- Gateway: "the most connected system in the Reach, the site of its
|
||||
oldest research institution, and the location of a gate that goes
|
||||
to Earth and that nobody uses"
|
||||
- Ran: "a very nice place to live, provided you are the sort of
|
||||
person whose grandparents owned the land"
|
||||
- ACB: "where the Lattice Commission lives, which means it is where
|
||||
the rules live"
|
||||
- Posto Avançado: "forward post — the place beyond the established
|
||||
line"
|
||||
|
||||
This script parses each gttr.md, regex-extracts the first `**NAME**`
|
||||
paragraph, normalises whitespace, and truncates at a soft word cap so
|
||||
the hook stays cheap to inject into prompts. Empty or unmatched files
|
||||
leave the column NULL.
|
||||
|
||||
Idempotent, safe to re-run after any wiki update. Explicit transaction
|
||||
wrapper with rollback on exception.
|
||||
|
||||
Usage:
|
||||
tooling/db/populate_gttr_hook.py
|
||||
tooling/db/populate_gttr_hook.py --max-words 45
|
||||
tooling/db/populate_gttr_hook.py --dry-run
|
||||
tooling/db/populate_gttr_hook.py --system "GJ 71"
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
SCRIPT_DIR = Path(__file__).resolve().parent
|
||||
REPO_ROOT = (SCRIPT_DIR / ".." / "..").resolve()
|
||||
DB_PATH = REPO_ROOT / "server" / "data" / "systems.db"
|
||||
WIKI_SYSTEMS = REPO_ROOT / "wiki" / "star-systems"
|
||||
|
||||
# Matches the first paragraph that opens with `**NAME**` at the start
|
||||
# of a line. Captures everything up to a blank line or the next H1/H2.
|
||||
# Non-greedy on the content.
|
||||
# Match a line that opens with `**NAME**` (any non-asterisk chars, since
|
||||
# names contain accents from many alphabets — Á, Ž, Ç, Ñ, etc) followed
|
||||
# by the paragraph body up to a blank line or the next markdown heading.
|
||||
_HOOK_RE = re.compile(
|
||||
r"^\*\*([^\n*]+?)\*\*(.+?)(?:\n\n|\n#)",
|
||||
re.DOTALL | re.MULTILINE,
|
||||
)
|
||||
|
||||
|
||||
def _system_slug_to_system_id(slug: str) -> str:
|
||||
"""Convert 'GJ-244A' → 'GJ 244A'. Mirrors populate_terrain_reference.py."""
|
||||
if slug.startswith("GJ-"):
|
||||
return "GJ " + slug[3:]
|
||||
return slug
|
||||
|
||||
|
||||
def extract_hook(gttr_path: Path, max_words: int) -> str | None:
|
||||
"""Return a compact one-paragraph hook for this gttr.md, or None if
|
||||
the file is missing or the opening paragraph can't be located."""
|
||||
if not gttr_path.exists():
|
||||
return None
|
||||
text = gttr_path.read_text()
|
||||
|
||||
match = _HOOK_RE.search(text)
|
||||
if not match:
|
||||
return None
|
||||
|
||||
name_token = match.group(1).strip()
|
||||
body = match.group(2).strip()
|
||||
|
||||
# Reconstruct "NAME is …" without the markdown asterisks. Add a
|
||||
# single space between the name and the body since `.strip()` above
|
||||
# removed the leading space from the captured body.
|
||||
hook = f"{name_token} {body}"
|
||||
|
||||
# Collapse whitespace so multi-line paragraphs become one clean line.
|
||||
hook = re.sub(r"\s+", " ", hook).strip()
|
||||
|
||||
# Strip a leading orphan "is" that comes from `**NAME**` + " is …":
|
||||
# the regex captures the word "is" on its own because the opener is
|
||||
# typically `**GATEWAY** (known as Tau Ceti...) is the most connected…`.
|
||||
# Normal reading already works — this is just hygiene.
|
||||
hook = re.sub(r"\s*\(\s*\)\s*", " ", hook)
|
||||
|
||||
# Hard cap at max_words. Truncate at the last word boundary before
|
||||
# the cap and append an ellipsis so the reader knows it continues.
|
||||
words = hook.split()
|
||||
if len(words) > max_words:
|
||||
hook = " ".join(words[:max_words]).rstrip(",;:") + "…"
|
||||
|
||||
return hook
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Populate star_systems.gttr_hook from wiki gttr.md files"
|
||||
)
|
||||
parser.add_argument("--db", default=str(DB_PATH), help="Path to systems.db")
|
||||
parser.add_argument(
|
||||
"--system",
|
||||
help="Process only this system_id (e.g. 'GJ 71')",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--max-words",
|
||||
type=int,
|
||||
default=45,
|
||||
help="Soft cap on hook length (default: 45 words)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--dry-run",
|
||||
action="store_true",
|
||||
help="Extract and print but do not write to the DB",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--verbose",
|
||||
action="store_true",
|
||||
help="Print every extracted hook",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
|
||||
db_path = Path(args.db)
|
||||
if not db_path.exists():
|
||||
print(f"error: {db_path} not found", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
conn = sqlite3.connect(str(db_path), timeout=30.0)
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA busy_timeout=15000")
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
|
||||
# Ensure the column exists — idempotent ADD COLUMN for old DBs.
|
||||
try:
|
||||
conn.execute("ALTER TABLE star_systems ADD COLUMN gttr_hook TEXT")
|
||||
except sqlite3.OperationalError:
|
||||
pass # column already exists
|
||||
|
||||
print("\n populate_gttr_hook.py")
|
||||
print(f" DB: {db_path}")
|
||||
print(f" max words: {args.max_words}")
|
||||
if args.dry_run:
|
||||
print(" Mode: DRY RUN")
|
||||
print()
|
||||
|
||||
rows = conn.execute(
|
||||
"SELECT system_id, proper_name FROM star_systems ORDER BY system_id"
|
||||
).fetchall()
|
||||
if args.system:
|
||||
rows = [r for r in rows if r[0] == args.system]
|
||||
|
||||
updated = 0
|
||||
missing = 0
|
||||
unmatched = 0
|
||||
|
||||
conn.execute("BEGIN")
|
||||
try:
|
||||
for system_id, proper_name in rows:
|
||||
slug = system_id.replace(" ", "-", 1)
|
||||
gttr_path = WIKI_SYSTEMS / slug / "gttr.md"
|
||||
|
||||
hook = extract_hook(gttr_path, args.max_words)
|
||||
if hook is None:
|
||||
if not gttr_path.exists():
|
||||
missing += 1
|
||||
if args.verbose:
|
||||
print(f" MISSING {system_id:10s} {gttr_path}")
|
||||
else:
|
||||
unmatched += 1
|
||||
if args.verbose:
|
||||
print(f" UNMATCHED {system_id:10s} {gttr_path}")
|
||||
continue
|
||||
|
||||
if args.verbose:
|
||||
display = f"{system_id} ({proper_name})" if proper_name else system_id
|
||||
print(f" {display}")
|
||||
print(f" → {hook}")
|
||||
|
||||
if not args.dry_run:
|
||||
conn.execute(
|
||||
"UPDATE star_systems SET gttr_hook = ? WHERE system_id = ?",
|
||||
(hook, system_id),
|
||||
)
|
||||
updated += 1
|
||||
|
||||
if args.dry_run:
|
||||
conn.rollback()
|
||||
else:
|
||||
conn.commit()
|
||||
except BaseException:
|
||||
conn.rollback()
|
||||
conn.close()
|
||||
raise
|
||||
|
||||
conn.close()
|
||||
|
||||
print("\n Done:")
|
||||
print(f" updated: {updated}")
|
||||
print(f" missing: {missing}")
|
||||
print(f" unmatched: {unmatched}")
|
||||
if args.dry_run:
|
||||
print("\n Dry run — no DB writes.")
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,728 +0,0 @@
|
||||
#!/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} |")
|
||||
|
||||
# Specialization (D-237 authored layer) — surface the generator-input enums
|
||||
# on the wiki face. NULL = unauthored (heuristic fallback / corridor default).
|
||||
if e:
|
||||
es = e["economic_specialization"] if "economic_specialization" in e.keys() else None
|
||||
cs = e["cultural_specialization"] if "cultural_specialization" in e.keys() else None
|
||||
if es:
|
||||
lines.append(f"| **Specialization** | {es} |")
|
||||
if cs:
|
||||
lines.append(f"| **Cultural Register** | {cs} |")
|
||||
|
||||
# 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()
|
||||
Reference in New Issue
Block a user