#!/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_specialization/hq_placement/headquarters_body/headquarters_city_id on corporations (D-242: HQ-placement key + baked CityTenant link, from corp_hq_placement.toml) - corp_presence from wiki/corporations/*.md (headquarters location data) - atlas_city_names Standalone-HQ settlement rows (D-242, T-1074) 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) This file is the CLI entrypoint and single-transaction orchestrator; the import steps live in the economy_import package (T-1067). The stamped source set is defined in tooling/generator_sources.py. 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 sys from pathlib import Path # tooling/economy-db is not a package (hyphenated dir) — put it on sys.path so # the economy_import package resolves; its __init__ adds tooling/ for # generator_sources and schema_version. sys.path.insert(0, str(Path(__file__).resolve().parent)) from economy_import import ( # noqa: E402 atlas, bodies, brands, corporations, db, economy, migration, specialization, stamp, traits, validators, ) from economy_import.errors import ImportAborted # noqa: E402 from economy_import.paths import DB_PATH # noqa: E402 from generator_sources import IMPORT_ECONOMICS_SOURCES # noqa: E402 def main() -> None: 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") parser.add_argument( "--strict-specialization", action="store_true", help="Treat D-237 completeness gates (V-SES-02, V-FAC-01) as hard errors. " "Off by default until the #1014 fallback and #1016 content pass land.", ) 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("\n Economics Import Pipeline") print(f" DB: {db_path}") if args.dry_run: print(" Mode: DRY RUN") print() # Load wiki corps before opening DB — allows early exit on parse failures print(" Loading wiki corporations...") wiki_corps = corporations.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: brands.regenerate_brands() except ImportAborted: sys.exit(1) conn = db.connect(db_path) # 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...") migration.apply_schema_migrations(conn) # Atlas index tables: apply canonical DDL + empty geometry (D-223, #951) atlas.ensure_atlas_index_schema(conn, args.dry_run) print(" tables and columns ready") # Clear economics tables in FK-safe order (children before parents) if not args.dry_run: db.clear_economics_tables(conn) # 2. Gate links print(" [2/10] Importing gate links...") n_links = economy.import_gate_links(conn, args.dry_run) print(f" {n_links} rows (bidirectional)") # 3. Commodities print(" [3/10] Importing commodities...") n_commodities = economy.import_commodities(conn, args.dry_run) print(f" {n_commodities} commodities") # 4. Production chains print(" [4/10] Importing production chains...") n_chains, n_inputs = economy.import_chains(conn, args.dry_run) print(f" {n_chains} chains, {n_inputs} inputs") # 4b. System specialization (D-237 authored layer) — after commodities # (FK) and before currency zones. UPSERTs onto pre-existing # system_economy / system_factions rows; unauthored systems stay NULL. print(" [4b/10] Importing system specialization (D-237)...") spec = specialization.import_system_specialization( conn, args.dry_run, strict=args.strict_specialization ) print( f" vocab {spec['vocab']} | economic {spec['economic']} | " f"cultural {spec['cultural']} | faction {spec['faction']}" ) if spec["missing_economy_row"]: print( f" WARNING: {len(spec['missing_economy_row'])} authored " f"system(s) lack a system_economy row (values dropped): " f"{spec['missing_economy_row']}" ) if spec["missing_faction_row"]: print( f" WARNING: {len(spec['missing_faction_row'])} authored " f"system(s) lack a system_factions row (faction dropped): " f"{spec['missing_faction_row']}" ) # 5. Currency zones print(" [5/10] Setting currency zones...") zones = economy.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 = economy.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 = corporations.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)") # 7b. Corp specialization + HQ placement (D-242, T-1074) — Phase A. # Reset + derive on every run (PR #177 H1/T2): the importer-owned # corporations columns are NULLed and re-derived from source, never # kept from a prior run. Must run after corp sync (rows must exist), # before corp_presence (which reads headquarters_body back off # corporations), and before step 12 (the H2 city-presence tiebreak # reads atlas_city_names BEFORE this run's clear/rebuild — the # previous run's settled state, by design). print(" [7b/10] Importing corp specialization + HQ placement (D-242)...") corp_spec = corporations.populate_corp_specialization(conn, wiki_corps, args.dry_run) print( f" {corp_spec['specialized']}/{corp_spec['total']} corps specialized, " f"{corp_spec['hq_resolved']} headquarters_body resolved (recomputed every run; " f"{corp_spec['hq_overridden']} authored overrides)" ) # 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 = corporations.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 = brands.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/13] Populating system_fiscal...") n_fiscal = economy.import_system_fiscal(conn, args.dry_run) print(f" {n_fiscal} system_fiscal rows") # 11. body_radius_km fallback from planet_class (D-204, #910) print(" [11/13] Populating body_radius_km fallback...") n_radius = bodies.populate_body_radius_km(conn, args.dry_run) print(f" {n_radius} bodies updated") # 12. atlas_city_names from wiki markers.json (D-207, #908) print(" [12/13] Populating atlas_city_names from wiki content...") n_cities = atlas.populate_atlas_city_names(conn, args.dry_run) print(f" {n_cities} city name rows") # 12b. atlas_feature_names (rivers/mountains) from wiki markers.json # (D-223, T-1169) — mirrors step 12's pool-load shape exactly, over a # different names-only markers.json key set. Independent of the # settlement pool, so order relative to step 13 doesn't matter; placed # here to stay adjacent to its sibling pool-load step. print(" [12b/13] Populating atlas_feature_names from wiki content...") n_features = atlas.populate_atlas_feature_names(conn, args.dry_run) print(f" {n_features} feature name rows") # 13. Standalone-HQ settlements + CityTenant city links (D-242, T-1074) — Phase B. # SUPERSEDES the retired corp-HQ cross-reference (D-207, #909) — that # step inserted one atlas_city_names row per corp HQ with no # UNIQUE(body_id, name), producing duplicate co-named "cities" (10 # Groombridge rows on GJ380c). Must run after populate_atlas_city_names # (the city pool Standalone HQs join and CityTenant HQs tenant) and # after step 7b (corp_specialization/hq_placement/headquarters_body). print(" [13/13] Emitting Standalone-HQ settlements + CityTenant links (D-242)...") hq_settlements = corporations.populate_standalone_hq_settlements(conn, args.dry_run) print( f" {hq_settlements['standalone_inserted']} Standalone-HQ settlements, " f"{hq_settlements['tenant_linked']} CityTenant links " f"({hq_settlements['tenant_unmatched']} unmatched)" ) # 13b. Per-settlement population + settlement_class bake (D-242, T-1075). # Runs over the CORRECTED pool — after both T-1074 steps, so # Standalone-HQ settlements are included in the rank-size spread, not # bolted on after. print(" [13b/13] Baking settlement population + settlement_class (D-242)...") pop_bake = atlas.populate_settlement_population_class(conn, args.dry_run) print( f" {pop_bake['cities_populated']} cities populated across " f"{pop_bake['bodies_spread']} bodies, {pop_bake['name_locked_applied']} " f"NameLocked pins applied" ) # 14. Architecture-flavor trait templates (D-232, #993). Catalog first, # then sparse per-body hero bias (FK -> trait_templates + bodies). print(" [14/15] Baking trait_templates catalog (D-232)...") n_templates = traits.populate_trait_templates(conn, args.dry_run) print(f" {n_templates} trait templates") print(" [15/15] Baking atlas_body_trait_bias hero pins (D-232)...") n_bias = traits.populate_atlas_body_trait_bias(conn, args.dry_run) print(f" {n_bias} body trait-bias rows") # 16. axial_tilt_deg from body-def frontmatter (T-1024, D-239 §2) print(" [16/17] Populating axial_tilt_deg from body-def frontmatter...") n_tilt = bodies.populate_axial_tilt_deg(conn, args.dry_run) print(f" {n_tilt} bodies updated with axial_tilt_deg") # 17. biosphere_class from body frontmatter override + two-gate default (D-247, T-1085) print(" [17/19] Populating biosphere_class (D-247)...") n_bio = bodies.populate_biosphere_class(conn, args.dry_run) print(f" {n_bio} bodies updated with biosphere_class") # 18. Architecture zone-type bias table (T-988, D-235 step 2) — must # follow trait_templates (V-TT-06 validates against its visual_bundle). print(" [18/19] Baking architecture_zone_bias table (D-235)...") n_zone_bias = traits.populate_architecture_zone_bias(conn, args.dry_run) print(f" {n_zone_bias} zone-bias rows") # 19. Color register bands (T-988, D-235) — numeric HSV sampling bands # per trait-template color_register; also follows trait_templates (V-TT-07). print(" [19/19] Baking color_register_bands table (D-235)...") n_color_bands = traits.populate_color_register_bands(conn, args.dry_run) print(f" {n_color_bands} color register bands") # Validate structural integrity (FK, chain refs, chain completeness). # These errors indicate broken imported data — do NOT commit. print("\n Validating structural integrity...") struct_errors = validators.validate(conn) struct_errors.extend(brands.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: stamp.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( validators.validate_commodity_coverage(conn, wiki_corps, commodity_ids_for_coverage) ) coverage_errors.extend(validators.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()