diff --git a/server/data/systems.db b/server/data/systems.db index bb667d19d..074b0692b 100644 Binary files a/server/data/systems.db and b/server/data/systems.db differ diff --git a/tooling/economy-db/import_economics.py b/tooling/economy-db/import_economics.py index 7ff6fd144..27dbf434d 100755 --- a/tooling/economy-db/import_economics.py +++ b/tooling/economy-db/import_economics.py @@ -3,12 +3,19 @@ Import economics data into systems.db. Reads TOML/JSON source files and populates the economics tables: - - gate_links from docs/design/star-map.json (335 edges, bidirectional) - - commodities from wiki/economics/commodities.toml (36 types) + - gate_links from docs/design/star-map.json (335 edges, bidirectional) + - commodities from wiki/economics/commodities.toml (36 types) - production_chains + chain_inputs from wiki/economics/production_chains.toml - - currency_zone on star_systems (default TRACTUS_PRIMARY) + - currency_zone on star_systems (default TRACTUS_PRIMARY) + - gate_energy_connected on star_systems (D-186: false for MARK_PRIMARY zones) + - corporations from wiki/corporations/*.md (sync + insert new records) + - corp_presence from wiki/corporations/*.md (headquarters location data) -Does NOT populate corp_presence — that's a future pipeline step. +Validation (hard errors, non-zero exit on any failure): + - Wiki corporation names must match DB proper_name records (D-182 sync constraint) + - Chain completeness: every intermediate commodity has at least one production chain + - Commodity coverage: 3+ corporations per major commodity type (D-175) + - System coverage: 1+ corporation per inhabited system with population > 100K (D-175) Usage: python3 tooling/economy-db/import_economics.py @@ -18,6 +25,7 @@ Usage: import argparse import json +import re import sqlite3 import sys import tomllib @@ -30,6 +38,7 @@ STAR_MAP = REPO_ROOT / "docs" / "design" / "star-map.json" COMMODITIES_TOML = REPO_ROOT / "wiki" / "economics" / "commodities.toml" CHAINS_TOML = REPO_ROOT / "wiki" / "economics" / "production_chains.toml" SCHEMA_SQL = REPO_ROOT / "server" / "data" / "systems-schema.sql" +CORPORATIONS_DIR = REPO_ROOT / "wiki" / "corporations" # --------------------------------------------------------------------------- @@ -102,6 +111,7 @@ CREATE INDEX IF NOT EXISTS idx_corp_presence_location ON corp_presence(location_ # Columns to add to existing tables (ALTER TABLE is idempotent via try/except) COLUMN_MIGRATIONS = [ ("star_systems", "currency_zone", "TEXT DEFAULT 'TRACTUS_PRIMARY'"), + ("star_systems", "gate_energy_connected", "INTEGER DEFAULT 1"), ("corporations", "behavioral_archetype", "TEXT"), ("corporations", "supply_chain_role", "TEXT"), ("corporations", "shadow_economy_access", "INTEGER DEFAULT 0"), @@ -263,11 +273,211 @@ def set_currency_zones(conn: sqlite3.Connection, dry_run: bool) -> dict: return counts +# --------------------------------------------------------------------------- +# Gate energy connectivity (D-186) +# --------------------------------------------------------------------------- + +def set_gate_energy(conn: sqlite3.Connection, dry_run: bool) -> dict: + """Set gate_energy_connected on star_systems based on currency_zone. + + MARK_PRIMARY zones default to false (Compact refused Gate Corp dependency). + All other zones default to true. + """ + if dry_run: + return {"on_grid": "non-MARK_PRIMARY", "off_grid": "MARK_PRIMARY"} + + # Default: all systems on-grid + conn.execute("UPDATE star_systems SET gate_energy_connected = 1 WHERE gate_energy_connected IS NULL") + + # MARK_PRIMARY zones are off-grid (Compact energy sovereignty) + conn.execute("UPDATE star_systems SET gate_energy_connected = 0 WHERE currency_zone = 'MARK_PRIMARY'") + + counts = {} + for row in conn.execute( + "SELECT gate_energy_connected, COUNT(*) FROM star_systems GROUP BY gate_energy_connected" + ): + label = "on_grid" if row[0] == 1 else "off_grid" + counts[label] = row[1] + + return counts + + +# --------------------------------------------------------------------------- +# Corporation wiki parsing +# --------------------------------------------------------------------------- + +def _parse_corp_frontmatter(path: Path) -> dict | None: + """Parse YAML frontmatter from a wiki corporation markdown file.""" + text = path.read_text() + lines = text.split("\n") + if not lines or lines[0].strip() != "---": + return None + end_idx = None + for i, line in enumerate(lines[1:], 1): + if line.strip() == "---": + end_idx = i + break + if end_idx is None: + return None + fm: dict = {} + for line in lines[1:end_idx]: + if ":" not in line: + continue + key, _, val = line.partition(":") + key = key.strip() + val = val.strip() + if val.startswith("[") and val.endswith("]"): + items = [x.strip().strip('"').strip("'") for x in val[1:-1].split(",")] + fm[key] = [item for item in items if item] + else: + fm[key] = val.strip('"').strip("'") + return fm + + +def load_wiki_corps() -> list[dict]: + """Load all wiki corporation files. Returns list of parsed corp records.""" + corps = [] + for md_file in sorted(CORPORATIONS_DIR.glob("*.md")): + if md_file.name == "index.md": + continue + fm = _parse_corp_frontmatter(md_file) + if not fm or not fm.get("slug") or not fm.get("title"): + continue + hq = fm.get("headquarters", "") + m = re.search(r"\(([^)]+)\)", hq) + system_id = m.group(1) if m else None + corps.append({ + "corp_id": fm["slug"], + "proper_name": fm["title"], + "system_id": system_id, + "tags": fm.get("tags", []), + "scope": fm.get("scope", ""), + }) + return corps + + +# --------------------------------------------------------------------------- +# Corporation sync (D-182: wiki is source of truth) +# --------------------------------------------------------------------------- + +def sync_corporations( + conn: sqlite3.Connection, wiki_corps: list[dict], dry_run: bool +) -> list[str]: + """Sync wiki corps to DB. Hard error on proper_name divergence (D-182). + + Returns list of error strings. Inserts corps that exist in wiki but not DB. + Corps that exist only in DB (legacy records) are left untouched. + headquarters_system is only written if the system_id exists in star_systems + (to avoid FK violations when atlas hasn't yet registered the system). + """ + errors: list[str] = [] + existing = { + r[0]: r[1] + for r in conn.execute("SELECT corp_id, proper_name FROM corporations").fetchall() + } + valid_systems = { + r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall() + } + + to_insert = [] + for corp in wiki_corps: + corp_id = corp["corp_id"] + proper_name = corp["proper_name"] + if corp_id in existing: + if existing[corp_id] != proper_name: + errors.append( + f"name divergence: corp_id='{corp_id}' " + f"wiki='{proper_name}' db='{existing[corp_id]}'" + ) + else: + system_id = corp.get("system_id") + hq_system = system_id if system_id and system_id in valid_systems else None + if system_id and system_id not in valid_systems: + print(f" warning: {corp_id} HQ system '{system_id}' not in DB, " + f"headquarters_system set to NULL") + to_insert.append(( + corp_id, + proper_name, + "corporation", + corp.get("scope") or None, + hq_system, + )) + + if not dry_run and not errors: + conn.executemany( + """INSERT OR IGNORE INTO corporations + (corp_id, proper_name, corp_type, scope, headquarters_system) + VALUES (?, ?, ?, ?, ?)""", + to_insert, + ) + + return errors + + +# --------------------------------------------------------------------------- +# Corp presence population +# --------------------------------------------------------------------------- + +def import_corp_presence( + conn: sqlite3.Connection, + wiki_corps: list[dict], + commodity_ids: set[str], + dry_run: bool, +) -> int: + """Populate corp_presence from wiki headquarters data. + + Each corporation gets one presence row at its headquarters system. + primary_operation is set to the first commodity tag that matches a known + commodity ID, or None if no commodity tags are present. + """ + valid_systems = { + r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall() + } + + rows = [] + skipped = [] + for corp in wiki_corps: + system_id = corp.get("system_id") + if not system_id: + skipped.append(f"{corp['corp_id']} (no headquarters system parsed)") + continue + if system_id not in valid_systems: + skipped.append(f"{corp['corp_id']} (system '{system_id}' not in DB)") + continue + primary_op = next( + (tag for tag in corp.get("tags", []) if tag in commodity_ids), None + ) + rows.append((corp["corp_id"], system_id, "system", primary_op)) + + if skipped: + for s in skipped: + print(f" warning: skipped corp_presence for {s}") + + if not dry_run: + conn.execute("DELETE FROM corp_presence") + conn.executemany( + """INSERT OR IGNORE INTO corp_presence + (corp_id, location_id, location_type, primary_operation) + VALUES (?, ?, ?, ?)""", + rows, + ) + + return len(rows) + + # --------------------------------------------------------------------------- # Validation # --------------------------------------------------------------------------- def validate(conn: sqlite3.Connection) -> list[str]: + """Validate structural integrity of imported data. + + Checks FK integrity, chain commodity references, and chain completeness. + These are hard blockers — broken data must not be committed. + + Coverage validation (commodity/system thresholds) is separate and runs + after commit via _validate_commodity_coverage() and _validate_system_coverage(). + """ errors = [] # FK integrity @@ -297,9 +507,75 @@ def validate(conn: sqlite3.Connection) -> list[str]: for chain_id, cid in orphan_outputs: errors.append(f"production_chains: chain '{chain_id}' outputs unknown commodity '{cid}'") + # Chain completeness: every intermediate commodity must have at least one producer + missing_chains = conn.execute(""" + SELECT c.commodity_id, c.name + FROM commodities c + WHERE c.tier = 'intermediate' + AND c.commodity_id NOT IN (SELECT output_commodity_id FROM production_chains) + ORDER BY c.commodity_id + """).fetchall() + for cid, name in missing_chains: + errors.append(f"chain completeness: no production chain produces intermediate '{cid}' ({name})") + return errors +def _validate_commodity_coverage( + conn: sqlite3.Connection, wiki_corps: list[dict], commodity_ids: set[str] +) -> list[str]: + """3+ corporations per major commodity type (raw + intermediate). D-175.""" + errors: list[str] = [] + major = [ + r[0] + for r in conn.execute( + "SELECT commodity_id FROM commodities " + "WHERE tier IN ('raw', 'intermediate') ORDER BY commodity_id" + ).fetchall() + ] + + # Build commodity → corp set from wiki tags filtered to known commodity IDs + coverage: dict[str, set[str]] = {cid: set() for cid in major} + for corp in wiki_corps: + for tag in corp.get("tags", []): + if tag in coverage: + coverage[tag].add(corp["corp_id"]) + + for cid in major: + n = len(coverage[cid]) + if n < 3: + corp_list = sorted(coverage[cid]) if coverage[cid] else ["none"] + errors.append( + f"commodity coverage: '{cid}' has {n}/3 corp(s) — {corp_list}" + ) + + return errors + + +def _validate_system_coverage( + conn: sqlite3.Connection, wiki_corps: list[dict] +) -> list[str]: + """1+ corporation per inhabited system with population > 100K. D-175. + + Uses wiki_corps headquarters data (not DB corp_presence) so this check + is accurate in both dry-run and real-run modes. + """ + covered = {c["system_id"] for c in wiki_corps if c.get("system_id")} + populated = conn.execute(""" + SELECT se.system_id, ss.proper_name, se.population + FROM system_economy se + JOIN star_systems ss ON se.system_id = ss.system_id + WHERE se.population > 100000 + ORDER BY se.system_id + """).fetchall() + + return [ + f"system coverage: no corp presence in '{sid}' ({name}, pop={pop:,})" + for sid, name, pop in populated + if sid not in covered + ] + + # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- @@ -321,66 +597,126 @@ def main(): print(f" Mode: DRY RUN") print() + # Load wiki corps before opening DB — allows early exit on parse failures + print(" Loading wiki corporations...") + wiki_corps = load_wiki_corps() + print(f" {len(wiki_corps)} corporation files parsed") + conn = sqlite3.connect(str(db_path)) conn.execute("PRAGMA foreign_keys=ON") # 1. Migrate schema - print(" [1/5] Schema migration...") + print(" [1/8] Schema migration...") for table, col, col_type in COLUMN_MIGRATIONS: _add_column(conn, table, col, col_type) conn.executescript(MIGRATION_SQL) print(" tables and columns ready") # Clear economics tables in FK-safe order (children before parents) + # corp_presence cleared here; corporations table is append-only (never cleared) if not args.dry_run: + conn.execute("DELETE FROM corp_presence") conn.execute("DELETE FROM chain_inputs") conn.execute("DELETE FROM production_chains") conn.execute("DELETE FROM commodities") conn.execute("DELETE FROM gate_links") # 2. Gate links - print(" [2/5] Importing gate links...") + print(" [2/8] Importing gate links...") n_links = import_gate_links(conn, args.dry_run) print(f" {n_links} rows (bidirectional)") # 3. Commodities - print(" [3/5] Importing commodities...") + print(" [3/8] Importing commodities...") n_commodities = import_commodities(conn, args.dry_run) print(f" {n_commodities} commodities") # 4. Production chains - print(" [4/5] Importing production chains...") + print(" [4/8] Importing production chains...") n_chains, n_inputs = import_chains(conn, args.dry_run) print(f" {n_chains} chains, {n_inputs} inputs") # 5. Currency zones - print(" [5/5] Setting currency zones...") + print(" [5/8] Setting currency zones...") zones = set_currency_zones(conn, args.dry_run) for zone, count in sorted(zones.items()): print(f" {zone}: {count}") - # Validate - print("\n Validating...") - errors = validate(conn) - if errors: - print(f" ERRORS ({len(errors)}):") - for e in errors: + # 6. Gate energy connectivity (D-186) — must run after currency zones + print(" [6/8] Setting gate energy connectivity...") + energy = set_gate_energy(conn, args.dry_run) + for label, count in sorted(energy.items()): + print(f" {label}: {count}") + + # 7. Sync corporations from wiki (D-182: hard error on name divergence) + print(" [7/8] Syncing corporations...") + corp_errors = sync_corporations(conn, wiki_corps, args.dry_run) + if corp_errors: + print(f" ERRORS ({len(corp_errors)}) — name divergence detected (D-182):") + for e in corp_errors: + print(f" - {e}") + print(" Fix: update wiki title or DB proper_name to match, then re-run.") + conn.close() + sys.exit(1) + n_db_corps = conn.execute("SELECT COUNT(*) FROM corporations").fetchone()[0] + print(f" {n_db_corps} corporations in DB ({len(wiki_corps)} from wiki)") + + # 8. Corp presence from wiki headquarters data + print(" [8/8] Importing corp presence...") + commodity_ids = { + r[0] for r in conn.execute("SELECT commodity_id FROM commodities").fetchall() + } + n_presence = import_corp_presence(conn, wiki_corps, commodity_ids, args.dry_run) + print(f" {n_presence} corp_presence rows") + + # Validate structural integrity (FK, chain refs, chain completeness). + # These errors indicate broken imported data — do NOT commit. + print("\n Validating structural integrity...") + struct_errors = validate(conn) + if struct_errors: + print(f" STRUCTURAL ERRORS ({len(struct_errors)}) — rolling back:") + for e in struct_errors: print(f" - {e}") conn.close() sys.exit(1) else: - print(" FK integrity OK") + print(" FK integrity and chain completeness OK") + # Commit all imported data (corps, presence, etc.) before coverage check. + # Coverage validation is a Phase 2 gate (D-175) — data should be persisted + # so tools can query it and report gaps clearly. if not args.dry_run: conn.commit() - print("\n Committed.") + print(" Data committed.") else: - print("\n Dry run — no changes written.") + print(" Dry run — no changes written.") + + # Validate coverage (hard errors per D-175, but after commit so data is usable). + print("\n Validating coverage (D-175 Phase 2 gate)...") + coverage_errors: list[str] = [] + commodity_ids_for_coverage = { + r[0] for r in conn.execute("SELECT commodity_id FROM commodities").fetchall() + } + coverage_errors.extend( + _validate_commodity_coverage(conn, wiki_corps, commodity_ids_for_coverage) + ) + coverage_errors.extend(_validate_system_coverage(conn, wiki_corps)) + + if coverage_errors: + print(f" COVERAGE ERRORS ({len(coverage_errors)}) — Phase 2 gate not met:") + for e in coverage_errors: + print(f" - {e}") + print("\n Data committed but Phase 2 gate is NOT met. " + "Add corporations to meet coverage thresholds and re-run.") + conn.close() + sys.exit(1) + else: + print(" All coverage thresholds met — Phase 2 gate PASSED.") conn.close() print(f"\n Done: {n_links} gate_links, {n_commodities} commodities, " - f"{n_chains} chains, {n_inputs} inputs\n") + f"{n_chains} chains, {n_inputs} inputs, {n_presence} corp_presence\n") if __name__ == "__main__":