jpmschweitzerandClaude Sonnet 4.6
e11a9c308a
New generate_brands binary reads 126 brand archetype templates (wiki/economics/archetypes/brand_templates.toml), assigns halo+volume pairs to all 48 hand-authored corps, outputs wiki/economics/corporations/generated_brands.toml. Result: 10,000 brand_product rows, 26,750 brand_inputs, all 48 corps covered. Brand structural validation V-B01..V-B06 passes. Generated file is gitignored (regenerated on each `make economy-db` run). make economy-db now runs generate-brands before import_economics.py. import_economics.py merges generated_brands.toml alongside hand-authored brands.toml. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1164 lines
44 KiB
Python
Executable File
1164 lines
44 KiB
Python
Executable File
#!/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 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"
|
||
|
||
|
||
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),
|
||
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);
|
||
"""
|
||
|
||
# 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"),
|
||
]
|
||
|
||
|
||
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"}
|
||
|
||
|
||
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"),
|
||
))
|
||
|
||
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
|
||
) 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),
|
||
]
|
||
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 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")
|
||
|
||
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
|
||
|
||
# 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_presence} corp_presence, "
|
||
f"{n_brands} brand_products, {n_brand_inputs} brand_inputs, "
|
||
f"{n_fiscal} system_fiscal\n")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|