#!/usr/bin/env python3 """ 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) - production_chains + chain_inputs from wiki/economics/production_chains.toml - 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) 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 python3 tooling/economy-db/import_economics.py --dry-run python3 tooling/economy-db/import_economics.py --db path/to/systems.db """ import argparse import hashlib import json import re import sqlite3 import sys import tomllib from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent.parent DB_PATH = REPO_ROOT / "server" / "data" / "systems.db" 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" BRANDS_TOML = REPO_ROOT / "wiki" / "economics" / "corporations" / "brands.toml" GENERATED_BRANDS_TOML = REPO_ROOT / "wiki" / "economics" / "corporations" / "generated_brands.toml" # Rust sources for the generate_brands subroutine. import_economics shells out to # tooling/generate-brands as part of its normal flow (see regenerate_brands()), so # both Rust files contribute to this script's effective source SHA: any change to # either must invalidate the meta stamp even though Python hasn't changed. GENERATE_BRANDS_RS = REPO_ROOT / "server" / "src" / "bin" / "generate_brands" / "main.rs" GENERATE_BRANDS_NAMES_RS = REPO_ROOT / "server" / "src" / "bin" / "generate_brands" / "names.rs" GENERATE_BRANDS_WRAPPER = REPO_ROOT / "tooling" / "generate-brands" def _file_sha1(*paths: Path) -> str: """Return SHA-1 hex of the concatenated content of one or more files. Files are sorted by path for determinism. Missing files raise FileNotFoundError rather than silently contributing an empty-string hash — a ghost SHA masks real breakage (review comment H2: da39a3ee… convergence could produce vacuous passes). """ h = hashlib.sha1() for p in sorted(paths): if not p.exists(): raise FileNotFoundError(f"generator source not found: {p}") h.update(p.read_bytes()) return h.hexdigest() # Canonical source set for import_economics' meta stamp. Covers its own .py file # plus the Rust binary it invokes (generate_brands main.rs + names.rs + wrapper # script) so any change to the brand generation pipeline flips the stamp. Keep # this list in sync with GENERATOR_SOURCES["import_economics"] in # tooling/check-systems-db-stamp. IMPORT_ECONOMICS_SOURCES: tuple[Path, ...] = ( Path(__file__), GENERATE_BRANDS_RS, GENERATE_BRANDS_NAMES_RS, GENERATE_BRANDS_WRAPPER, ) def _write_stamp(conn: sqlite3.Connection, generator_name: str, *source_files: Path) -> None: """Upsert a row in the meta table recording this generator's current source SHA. Called after every successful non-dry-run commit. Idempotent: running twice on the same sources writes the same sha with an updated timestamp. Only one stamp is written by this module: ``import_economics``, whose source set includes the Rust binary it invokes (see IMPORT_ECONOMICS_SOURCES). generate_atlas writes its own stamp. generate_brands does NOT write a stamp of its own — it's a subroutine of import_economics, not an independent DB writer (PR #136 review T2/H3). The meta table is created by the MIGRATION_SQL block above; this function assumes it exists (caller must run migrations first). """ schema_sha = _file_sha1(SCHEMA_SQL) generator_sha = _file_sha1(*source_files) conn.execute( """INSERT OR REPLACE INTO meta (generator_name, schema_version, generator_sha, generated_at) VALUES (?, ?, ?, datetime('now'))""", (generator_name, schema_sha, generator_sha), ) def regenerate_brands() -> None: """Run the Rust generate_brands binary to refresh generated_brands.toml. Invoked as the first step of import_economics' main flow so the TOML on disk always matches the current Rust source before the Python import reads it. This replaces the former split (tooling/generate-brands run separately by make regen-db) with a single, coherent brand pipeline owned by one stamp. The wrapper script builds the binary on demand and runs it with the default canonical seed=1; callers that need non-canonical seeds must still invoke the wrapper directly (experimentation only — committed output must be seed=1). """ import subprocess if not GENERATE_BRANDS_WRAPPER.exists(): raise FileNotFoundError( f"generate_brands wrapper not found at {GENERATE_BRANDS_WRAPPER}" ) print(" [pre/10] Running generate_brands (Rust) to refresh generated_brands.toml...") result = subprocess.run( [str(GENERATE_BRANDS_WRAPPER)], cwd=str(REPO_ROOT), capture_output=True, text=True, ) if result.returncode != 0: print(result.stdout, file=sys.stderr) print(result.stderr, file=sys.stderr) raise _ImportAborted() # Print the Rust binary's own summary lines (brands generated, coverage). # Indent so they fold under the pre-step heading. for line in result.stdout.splitlines(): if line.strip(): print(f" {line}") class _ImportAborted(Exception): """Raised internally by main() when a validation step wants a clean rollback + exit 1. Caught only by main(); error messages are printed before raising so the user sees them.""" # --------------------------------------------------------------------------- # Schema migration — add new tables and columns to existing DB # --------------------------------------------------------------------------- MIGRATION_SQL = """ -- Economics tables (idempotent — safe to re-run) -- Brand layer tables (D-189, #827) CREATE TABLE IF NOT EXISTS brand_products ( brand_product_id TEXT PRIMARY KEY, corp_id TEXT NOT NULL REFERENCES corporations(corp_id), product_name TEXT NOT NULL, brand_category TEXT NOT NULL, value_trajectory TEXT NOT NULL, scarcity_class TEXT NOT NULL, product_subcategory TEXT, base_premium_multiplier REAL NOT NULL DEFAULT 1.0, premium_floor REAL NOT NULL DEFAULT 0.0, origin_system TEXT REFERENCES star_systems(system_id), terroir_locked INTEGER NOT NULL DEFAULT 0, currency_denomination TEXT NOT NULL DEFAULT 'tractus', shadow_viable INTEGER NOT NULL DEFAULT 0, brand_tier TEXT NOT NULL, halo_brand_id TEXT REFERENCES brand_products(brand_product_id), price_tier TEXT, updated_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS brand_inputs ( brand_product_id TEXT NOT NULL REFERENCES brand_products(brand_product_id), commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id), quantity REAL NOT NULL, PRIMARY KEY (brand_product_id, commodity_id) ); CREATE TABLE IF NOT EXISTS system_fiscal ( system_id TEXT PRIMARY KEY REFERENCES star_systems(system_id), corp_tax_rate REAL NOT NULL DEFAULT 0.22, collection_efficiency REAL NOT NULL DEFAULT 1.0, updated_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS corp_financial_state ( corp_id TEXT PRIMARY KEY REFERENCES corporations(corp_id), health_metric REAL NOT NULL DEFAULT 1.0, updated_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS corp_lifecycle_events ( event_id INTEGER PRIMARY KEY AUTOINCREMENT, corp_id TEXT NOT NULL REFERENCES corporations(corp_id), event_type TEXT NOT NULL, event_tick INTEGER NOT NULL DEFAULT 0, event_data TEXT, created_at TEXT DEFAULT (datetime('now')) ); CREATE INDEX IF NOT EXISTS idx_brand_products_corp_category ON brand_products(corp_id, brand_category); CREATE INDEX IF NOT EXISTS idx_brand_products_origin ON brand_products(origin_system); CREATE INDEX IF NOT EXISTS idx_brand_products_tier ON brand_products(brand_tier); CREATE INDEX IF NOT EXISTS idx_brand_inputs_commodity ON brand_inputs(commodity_id); CREATE INDEX IF NOT EXISTS idx_corp_lifecycle_events_corp ON corp_lifecycle_events(corp_id); CREATE TABLE IF NOT EXISTS gate_links ( from_system_id TEXT NOT NULL REFERENCES star_systems(system_id), to_system_id TEXT NOT NULL REFERENCES star_systems(system_id), PRIMARY KEY (from_system_id, to_system_id) ); CREATE TABLE IF NOT EXISTS commodities ( commodity_id TEXT PRIMARY KEY, name TEXT NOT NULL, tier TEXT NOT NULL, elasticity TEXT NOT NULL, base_price REAL NOT NULL, bulk_class TEXT, unit TEXT, production_ubiquity TEXT, demand_model TEXT, commission_certifiable INTEGER DEFAULT 0, compact_contested INTEGER DEFAULT 0, shadow_viable INTEGER DEFAULT 0, panic_threshold_weeks INTEGER DEFAULT 0, description TEXT, updated_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS production_chains ( chain_id TEXT PRIMARY KEY, output_commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id), output_quantity REAL NOT NULL DEFAULT 1.0, location_bound INTEGER DEFAULT 0, description TEXT, updated_at TEXT DEFAULT (datetime('now')) ); CREATE TABLE IF NOT EXISTS chain_inputs ( chain_id TEXT NOT NULL REFERENCES production_chains(chain_id), input_commodity_id TEXT NOT NULL REFERENCES commodities(commodity_id), quantity REAL NOT NULL, PRIMARY KEY (chain_id, input_commodity_id) ); CREATE TABLE IF NOT EXISTS corp_presence ( corp_id TEXT NOT NULL REFERENCES corporations(corp_id), location_id TEXT NOT NULL, location_type TEXT NOT NULL, primary_operation TEXT, updated_at TEXT DEFAULT (datetime('now')), PRIMARY KEY (corp_id, location_id) ); -- New indexes CREATE INDEX IF NOT EXISTS idx_star_systems_currency ON star_systems(currency_zone); CREATE INDEX IF NOT EXISTS idx_gate_links_from ON gate_links(from_system_id); CREATE INDEX IF NOT EXISTS idx_gate_links_to ON gate_links(to_system_id); CREATE INDEX IF NOT EXISTS idx_commodities_tier ON commodities(tier); CREATE INDEX IF NOT EXISTS idx_production_chains_output ON production_chains(output_commodity_id); CREATE INDEX IF NOT EXISTS idx_chain_inputs_commodity ON chain_inputs(input_commodity_id); CREATE INDEX IF NOT EXISTS idx_corp_presence_corp ON corp_presence(corp_id); CREATE INDEX IF NOT EXISTS idx_corp_presence_location ON corp_presence(location_id); -- Generator metadata stamp (#855, #856) CREATE TABLE IF NOT EXISTS meta ( generator_name TEXT PRIMARY KEY, schema_version TEXT NOT NULL, generator_sha TEXT NOT NULL, generated_at TEXT NOT NULL DEFAULT (datetime('now')) ); -- Drop the pre-merge 'generate_brands' stamp row if it exists (PR #136 review T2/H3). -- The Rust brand binary is now a subroutine of import_economics — its source -- SHA contributes to the 'import_economics' stamp — so it no longer merits its -- own meta row. This DELETE makes the check-systems-db-stamp "unknown generator" -- path (fail-closed per T6) compatible with older DBs that still have the row. DELETE FROM meta WHERE generator_name = 'generate_brands'; """ # 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"), ("brand_products", "price_tier", "TEXT"), ] def _add_column(conn: sqlite3.Connection, table: str, col: str, col_type: str): """Add a column if it doesn't exist. SQLite has no IF NOT EXISTS for ALTER.""" try: conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} {col_type}") except sqlite3.OperationalError as e: if "duplicate column" in str(e).lower(): pass # already exists else: raise # --------------------------------------------------------------------------- # Gate links # --------------------------------------------------------------------------- def import_gate_links(conn: sqlite3.Connection, dry_run: bool) -> int: with open(STAR_MAP) as f: data = json.load(f) edges = data["edges"] system_ids = {r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall()} rows = [] skipped = [] for a, b in edges: if a not in system_ids: skipped.append(a) continue if b not in system_ids: skipped.append(b) continue rows.append((a, b)) rows.append((b, a)) if skipped: unique_skipped = sorted(set(skipped)) print(f" warning: {len(unique_skipped)} system(s) in star-map.json not in DB: {unique_skipped[:5]}...") if not dry_run: conn.executemany( "INSERT OR IGNORE INTO gate_links (from_system_id, to_system_id) VALUES (?, ?)", rows, ) return len(rows) # --------------------------------------------------------------------------- # Commodities # --------------------------------------------------------------------------- def import_commodities(conn: sqlite3.Connection, dry_run: bool) -> int: with open(COMMODITIES_TOML, "rb") as f: data = tomllib.load(f) rows = [] for cid, c in data.items(): rows.append(( cid, c["name"], c["tier"], c["elasticity"], c["base_price"], c.get("bulk_class"), c.get("unit"), c.get("production_ubiquity"), c.get("demand_model"), int(c.get("commission_certifiable", False)), int(c.get("compact_contested", False)), int(c.get("shadow_viable", False)), c.get("panic_threshold_weeks", 0), c.get("description"), )) if not dry_run: conn.executemany( """INSERT INTO commodities ( commodity_id, name, tier, elasticity, base_price, bulk_class, unit, production_ubiquity, demand_model, commission_certifiable, compact_contested, shadow_viable, panic_threshold_weeks, description ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", rows, ) return len(rows) # --------------------------------------------------------------------------- # Production chains + inputs # --------------------------------------------------------------------------- def import_chains(conn: sqlite3.Connection, dry_run: bool) -> tuple[int, int]: with open(CHAINS_TOML, "rb") as f: data = tomllib.load(f) chain_rows = [] input_rows = [] for chain_id, c in data.items(): chain_rows.append(( chain_id, c["output"], c.get("output_quantity", 1.0), int(c.get("location_bound", False)), c.get("description"), )) for inp in c.get("inputs", []): input_rows.append(( chain_id, inp["commodity"], inp["quantity"], )) if not dry_run: conn.executemany( """INSERT INTO production_chains ( chain_id, output_commodity_id, output_quantity, location_bound, description ) VALUES (?, ?, ?, ?, ?)""", chain_rows, ) conn.executemany( """INSERT INTO chain_inputs ( chain_id, input_commodity_id, quantity ) VALUES (?, ?, ?)""", input_rows, ) return len(chain_rows), len(input_rows) # --------------------------------------------------------------------------- # Currency zones # --------------------------------------------------------------------------- def set_currency_zones(conn: sqlite3.Connection, dry_run: bool) -> dict: """Set currency_zone on star_systems from wiki/economics/currency_zones.toml. Default: TRACTUS_PRIMARY. Sol (GJ 0): MIXED (set before file is read). MARK_PRIMARY and MIXED assignments come from the TOML file (D-172). """ if dry_run: return {"TRACTUS_PRIMARY": "all", "MIXED": "GJ 0 + toml"} # Default everything to TRACTUS_PRIMARY conn.execute("UPDATE star_systems SET currency_zone = 'TRACTUS_PRIMARY'") # Sol system is MIXED (Earth legacy currency presence — set before TOML load) conn.execute("UPDATE star_systems SET currency_zone = 'MIXED' WHERE system_id = 'GJ 0'") # Load MARK_PRIMARY and MIXED assignments from authored TOML (D-172) zones_path = REPO_ROOT / "wiki" / "economics" / "currency_zones.toml" if zones_path.exists(): import tomllib # Python 3.11+ with open(zones_path, "rb") as f: zones = tomllib.load(f) mark_ids = [entry["system_id"] for entry in zones.get("mark_primary", [])] mixed_ids = [entry["system_id"] for entry in zones.get("mixed", [])] for sid in mark_ids: conn.execute( "UPDATE star_systems SET currency_zone = 'MARK_PRIMARY' WHERE system_id = ?", (sid,), ) for sid in mixed_ids: conn.execute( "UPDATE star_systems SET currency_zone = 'MIXED' WHERE system_id = ?", (sid,), ) else: print(" warning: wiki/economics/currency_zones.toml not found — " "all systems default to TRACTUS_PRIMARY / Sol to MIXED") counts = {} for row in conn.execute("SELECT currency_zone, COUNT(*) FROM star_systems GROUP BY currency_zone"): counts[row[0]] = row[1] 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 _resolve_hq_location( conn: sqlite3.Connection, system_id: str, headquarters_body: str | None, ) -> tuple[str, str] | None: """Resolve a corp's HQ to a (location_id, location_type) pair. Resolution order: 1. Use headquarters_body from corporations table if set (body or station). 2. Most-populated body in the system. 3. Any body in the system. 4. Any station in the system. Returns None if no body or station found. """ if headquarters_body: # Determine whether it's a body or station body = conn.execute( "SELECT body_id FROM bodies WHERE body_id = ?", (headquarters_body,) ).fetchone() if body: return (headquarters_body, "body") station = conn.execute( "SELECT station_id FROM stations WHERE station_id = ?", (headquarters_body,), ).fetchone() if station: return (headquarters_body, "station") # Most-populated body body = conn.execute( """SELECT body_id FROM bodies WHERE system_id = ? ORDER BY population DESC LIMIT 1""", (system_id,), ).fetchone() if body: return (body[0], "body") # Any station station = conn.execute( "SELECT station_id FROM stations WHERE system_id = ? LIMIT 1", (system_id,), ).fetchone() if station: return (station[0], "station") return None 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 body or station. location_type is 'body' or 'station' per schema (D-182). primary_operation is set to the first commodity tag matching a known commodity ID. """ valid_systems = { r[0] for r in conn.execute("SELECT system_id FROM star_systems").fetchall() } # Load headquarters_body from corporations table (set during import) hq_body_map: dict[str, str | None] = { r[0]: r[1] for r in conn.execute( "SELECT corp_id, headquarters_body FROM corporations" ).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 hq_body = hq_body_map.get(corp["corp_id"]) location = _resolve_hq_location(conn, system_id, hq_body) if not location: skipped.append( f"{corp['corp_id']} (no body/station found in system '{system_id}')" ) continue location_id, location_type = location primary_op = next( (tag for tag in corp.get("tags", []) if tag in commodity_ids), None ) rows.append((corp["corp_id"], location_id, location_type, 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 fk_issues = conn.execute("PRAGMA foreign_key_check").fetchall() if fk_issues: for issue in fk_issues[:10]: errors.append(f"FK violation: table={issue[0]} rowid={issue[1]} " f"parent={issue[2]} fkid={issue[3]}") # Chain inputs reference valid commodities orphan_inputs = conn.execute(""" SELECT ci.chain_id, ci.input_commodity_id FROM chain_inputs ci LEFT JOIN commodities c ON ci.input_commodity_id = c.commodity_id WHERE c.commodity_id IS NULL """).fetchall() for chain_id, cid in orphan_inputs: errors.append(f"chain_inputs: chain '{chain_id}' references unknown commodity '{cid}'") # Chain outputs reference valid commodities orphan_outputs = conn.execute(""" SELECT pc.chain_id, pc.output_commodity_id FROM production_chains pc LEFT JOIN commodities c ON pc.output_commodity_id = c.commodity_id WHERE c.commodity_id IS NULL """).fetchall() 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 ] # --------------------------------------------------------------------------- # Brand layer import (D-189, #827) # --------------------------------------------------------------------------- VALID_BRAND_CATEGORIES = { "terroir", "heritage_craft", "tech_premium", "cultural", "service_premium", "commodity_branded", "design_heritage", "platform_catalogue", } VALID_VALUE_TRAJECTORIES = {"appreciating", "depreciating", "timeless"} VALID_SCARCITY_CLASSES = {"capped", "constrained", "scalable", "unlimited"} VALID_BRAND_TIERS = {"halo", "volume"} VALID_CURRENCY_DENOMINATIONS = {"tractus", "mark", "mixed", "sol_adjacent"} VALID_PRICE_TIERS = {"mass", "premium", "luxury", "flagship", "institutional"} def _load_brand_file(path) -> tuple[list, list]: """Load brand_products and brand_inputs from a TOML file. Returns empty lists if missing.""" if not path.exists(): return [], [] with open(path, "rb") as f: data = tomllib.load(f) return data.get("brand_products", []), data.get("brand_inputs", []) def import_brands( conn: sqlite3.Connection, dry_run: bool ) -> tuple[int, int]: """Import brand_products and brand_inputs from brands.toml and generated_brands.toml. Hand-authored brands (brands.toml) are imported first; generated brands (generated_brands.toml, produced by `tooling/generate-brands`) are merged in. Returns (n_products, n_inputs). """ if not BRANDS_TOML.exists(): print(" warning: brands.toml not found — brand layer skipped") return 0, 0 products_authored, inputs_authored = _load_brand_file(BRANDS_TOML) products_generated, inputs_generated = _load_brand_file(GENERATED_BRANDS_TOML) if products_generated: print(f" merging {len(products_generated)} generated brand_products from generated_brands.toml") products = products_authored + products_generated inputs = inputs_authored + inputs_generated product_rows = [] for p in products: product_rows.append(( p["brand_product_id"], p["corp_id"], p["product_name"], p["brand_category"], p["value_trajectory"], p["scarcity_class"], p.get("product_subcategory"), p.get("base_premium_multiplier", 1.0), p.get("premium_floor", 0.0), p.get("origin_system"), int(p.get("terroir_locked", False)), p.get("currency_denomination", "tractus"), int(p.get("shadow_viable", False)), p["brand_tier"], p.get("halo_brand_id"), p.get("price_tier"), )) input_rows = [] for inp in inputs: input_rows.append(( inp["brand_product_id"], inp["commodity_id"], inp["quantity"], )) if not dry_run: conn.executemany( """INSERT OR REPLACE INTO brand_products ( brand_product_id, corp_id, product_name, brand_category, value_trajectory, scarcity_class, product_subcategory, base_premium_multiplier, premium_floor, origin_system, terroir_locked, currency_denomination, shadow_viable, brand_tier, halo_brand_id, price_tier ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", product_rows, ) conn.executemany( """INSERT OR REPLACE INTO brand_inputs (brand_product_id, commodity_id, quantity) VALUES (?, ?, ?)""", input_rows, ) return len(product_rows), len(input_rows) def import_system_fiscal(conn: sqlite3.Connection, dry_run: bool) -> int: """Populate system_fiscal with hardcoded Phase 2 values. Phase 2 values (NOT derived from D-189 §6 yet): - corp_tax_rate = 0.22 (flat default) - collection_efficiency = 0.85 (mid-reach average placeholder) The D-189 §6 formula `collection_efficiency = 1.0 - shadow_economy_intensity × 0.6` is deliberately NOT implemented here — `shadow_economy_intensity` is not yet per-system in the DB (pending the shadow_economy.toml pipeline). When that pipeline lands, replace the hardcoded 0.85 with the derivation and wire `shadow_economy_intensity` through the SELECT. Tracked as a Phase 3 follow-up. """ inhabited = conn.execute(""" SELECT ss.system_id, COALESCE(se.population, 0) FROM star_systems ss LEFT JOIN system_economy se ON ss.system_id = se.system_id WHERE ss.inhabited_planet_count > 0 OR se.population > 0 ORDER BY ss.system_id """).fetchall() PHASE2_CORP_TAX_RATE = 0.22 PHASE2_COLLECTION_EFFICIENCY = 0.85 rows = [ (system_id, PHASE2_CORP_TAX_RATE, PHASE2_COLLECTION_EFFICIENCY) for system_id, _pop in inhabited ] if not dry_run: conn.executemany( """INSERT OR IGNORE INTO system_fiscal (system_id, corp_tax_rate, collection_efficiency) VALUES (?, ?, ?)""", rows, ) return len(rows) def validate_brands(conn: sqlite3.Connection) -> list[str]: """Brand layer structural validation rules V-B01 through V-B06. V-B01: Every brand_products row has a valid corp_id (FK to corporations). V-B02: Every brand_inputs row has valid brand_product_id and commodity_id FKs. V-B03: Every halo brand has at least one brand_inputs entry (demand stub must consume). V-B04: Every volume tier must reference an existing halo brand_product_id. V-B05: No brand_product_id is used as halo_brand_id by a non-volume-tier product. V-B06: Every enum column (brand_category, value_trajectory, scarcity_class, brand_tier, currency_denomination) is a member of its VALID_* set. """ errors: list[str] = [] # V-B01: brand_products → corporations FK orphan_corps = conn.execute(""" SELECT bp.brand_product_id, bp.corp_id FROM brand_products bp LEFT JOIN corporations c ON bp.corp_id = c.corp_id WHERE c.corp_id IS NULL """).fetchall() for pid, corp_id in orphan_corps: errors.append( f"V-B01: brand_product '{pid}' references unknown corp_id '{corp_id}'" ) # V-B02: brand_inputs → brand_products and brand_inputs → commodities FKs orphan_inputs_bp = conn.execute(""" SELECT bi.brand_product_id, bi.commodity_id FROM brand_inputs bi LEFT JOIN brand_products bp ON bi.brand_product_id = bp.brand_product_id WHERE bp.brand_product_id IS NULL """).fetchall() for pid, cid in orphan_inputs_bp: errors.append( f"V-B02: brand_inputs row ({pid}, {cid}) references unknown brand_product_id" ) orphan_inputs_comm = conn.execute(""" SELECT bi.brand_product_id, bi.commodity_id FROM brand_inputs bi LEFT JOIN commodities c ON bi.commodity_id = c.commodity_id WHERE c.commodity_id IS NULL """).fetchall() for pid, cid in orphan_inputs_comm: errors.append( f"V-B02: brand_inputs row ({pid}, {cid}) references unknown commodity_id '{cid}'" ) # V-B03: every halo brand has at least one brand_inputs entry halo_no_inputs = conn.execute(""" SELECT bp.brand_product_id FROM brand_products bp WHERE bp.brand_tier = 'halo' AND bp.brand_product_id NOT IN (SELECT brand_product_id FROM brand_inputs) """).fetchall() for (pid,) in halo_no_inputs: errors.append( f"V-B03: halo brand '{pid}' has no brand_inputs entries " f"(must consume at least one commodity as a demand node)" ) # V-B04: volume tiers reference valid halo_brand_id volume_bad_halo = conn.execute(""" SELECT bp.brand_product_id, bp.halo_brand_id FROM brand_products bp WHERE bp.brand_tier = 'volume' AND (bp.halo_brand_id IS NULL OR bp.halo_brand_id NOT IN (SELECT brand_product_id FROM brand_products)) """).fetchall() for pid, halo_id in volume_bad_halo: errors.append( f"V-B04: volume brand '{pid}' has invalid halo_brand_id '{halo_id}'" ) # V-B05: halo_brand_id must only point to halo-tier products halo_points_to_non_halo = conn.execute(""" SELECT child.brand_product_id, child.halo_brand_id, parent.brand_tier FROM brand_products child JOIN brand_products parent ON child.halo_brand_id = parent.brand_product_id WHERE child.brand_tier = 'volume' AND parent.brand_tier != 'halo' """).fetchall() for child_id, halo_id, parent_tier in halo_points_to_non_halo: errors.append( f"V-B05: volume brand '{child_id}' points to '{halo_id}' " f"which has brand_tier='{parent_tier}', not 'halo'" ) # V-B06: every enum column is in its VALID_* set. The SQL columns are # plain TEXT without CHECK constraints, so a typo like `terrior` would # otherwise silently import. enum_checks = [ ("brand_category", VALID_BRAND_CATEGORIES), ("value_trajectory", VALID_VALUE_TRAJECTORIES), ("scarcity_class", VALID_SCARCITY_CLASSES), ("brand_tier", VALID_BRAND_TIERS), ("currency_denomination", VALID_CURRENCY_DENOMINATIONS), ("price_tier", VALID_PRICE_TIERS), ] for column, valid_set in enum_checks: bad = conn.execute( f"SELECT brand_product_id, {column} FROM brand_products" ).fetchall() for pid, value in bad: if value is None: continue # nullable columns (e.g. price_tier) may be unset if value not in valid_set: errors.append( f"V-B06: brand_product '{pid}' has {column}='{value}' — " f"must be one of {sorted(valid_set)}" ) return errors # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description="Import economics data into systems.db") parser.add_argument("--db", default=str(DB_PATH), help="Path to systems.db") parser.add_argument("--dry-run", action="store_true", help="Validate 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) print(f"\n Economics Import Pipeline") print(f" DB: {db_path}") if args.dry_run: 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") # Regenerate generated_brands.toml via the Rust binary before the Python # import reads it. Single pipeline, single stamp — resolves review T2/H3 # ("on-behalf stamping" coupling) by folding brand generation into # import_economics' flow rather than having the caller (Makefile / user) # remember to run it first. Skipped on --dry-run to avoid a disk # side-effect during validation. if not args.dry_run: try: regenerate_brands() except _ImportAborted: sys.exit(1) conn = sqlite3.connect(str(db_path)) conn.execute("PRAGMA foreign_keys=ON") # The clear-then-reimport cycle below runs as a single explicit # transaction. Any crash, validation error, or KeyboardInterrupt # between the first DELETE and the final commit rolls everything # back — the DB never ends up half-cleared with stale rows in some # tables and empty rows in others. On success we commit exactly # once, immediately after structural validation passes. conn.execute("BEGIN") try: # 1. Migrate schema (idempotent, inside the tx so a crash here # leaves no half-applied ALTER TABLE.) print(" [1/10] 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 brand_inputs") conn.execute("DELETE FROM brand_products") conn.execute("DELETE FROM system_fiscal") conn.execute("DELETE FROM corp_financial_state") conn.execute("DELETE FROM corp_lifecycle_events") 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/10] Importing gate links...") n_links = import_gate_links(conn, args.dry_run) print(f" {n_links} rows (bidirectional)") # 3. Commodities print(" [3/10] Importing commodities...") n_commodities = import_commodities(conn, args.dry_run) print(f" {n_commodities} commodities") # 4. Production chains print(" [4/10] 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/10] Setting currency zones...") zones = set_currency_zones(conn, args.dry_run) for zone, count in sorted(zones.items()): print(f" {zone}: {count}") # 6. Gate energy connectivity (D-186) — must run after currency zones print(" [6/10] 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/10] 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.") raise _ImportAborted() 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/10] 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") # 9. Brand products and inputs (D-189, #827) print(" [9/10] Importing brand products and inputs...") n_brands, n_brand_inputs = import_brands(conn, args.dry_run) print(f" {n_brands} brand_products, {n_brand_inputs} brand_inputs") # 10. System fiscal parameters (D-189 section 6) print(" [10/10] Populating system_fiscal...") n_fiscal = import_system_fiscal(conn, args.dry_run) print(f" {n_fiscal} system_fiscal 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) struct_errors.extend(validate_brands(conn)) if struct_errors: print(f" STRUCTURAL ERRORS ({len(struct_errors)}) — rolling back:") for e in struct_errors: print(f" - {e}") raise _ImportAborted() print(" FK integrity, chain completeness, and brand layer (V-B01–V-B06) 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(" Data committed.") else: # Dry-run: leave the transaction open so the coverage check below # can still SELECT against the in-memory imported data. The # transaction is discarded when conn.close() runs on exit. print(" Dry run — no changes written.") except _ImportAborted: conn.rollback() conn.close() sys.exit(1) except BaseException: # Any other exception (KeyboardInterrupt, MemoryError, DB error, # programmer error) triggers a rollback so the DB is never left in # a half-imported state. Re-raise so the user sees the traceback. conn.rollback() conn.close() raise # Stamp generator metadata (#855, #856): record source SHAs so the # pre-push hook can detect stale DB snapshots. Written BEFORE the # coverage gate — the stamp records generator execution (code version), # not data completeness. Coverage gaps (#860) are pre-existing data # issues and must not prevent the stamp from landing. if not args.dry_run: try: _write_stamp(conn, "import_economics", *IMPORT_ECONOMICS_SOURCES) conn.commit() print(" Stamped: import_economics (covers brand pipeline Rust sources)") except Exception as exc: # noqa: BLE001 print(f" WARNING: failed to write generator stamp: {exc}", file=sys.stderr) # 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(2) # exit 2 = coverage warning (data+stamp committed); exit 1 = real error 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_presence} corp_presence, " f"{n_brands} brand_products, {n_brand_inputs} brand_inputs, " f"{n_fiscal} system_fiscal\n") if __name__ == "__main__": main()