#!/usr/bin/env python3
"""Build reviewed atlas snapshots with Python stdlib + SQLite; never use the network.

python3 scripts/build-intelligence.py [--check] [--database PATH]
The SQLite file is an internal workspace; the full catalog is a server-side snapshot.
"""

import argparse
import csv
from datetime import date
import hashlib
import io
import json
from pathlib import Path
import re
import sqlite3
from urllib.parse import urlsplit

ROOT = Path(__file__).resolve().parents[1]
ASSETS = ROOT / "scripts/assets"
PUBLIC = ROOT / "public/data/intelligence"
RETIRED_PUBLIC = (PUBLIC / "facilities.json", PUBLIC / "facilities.csv")
REVIEWED = "2026-09-27"
STATUSES = ("operational", "proposed", "under_construction", "permitted", "cancelled")
SETTINGS = ("above_ground", "underground", "mixed", "unclassified")
SETTING_REGISTER = ASSETS / "intelligence-setting-evidence.json"
FACILITY_SHA = "deca4eea8d71e904681f4a1d9ae2c64103157cdd69ee1462c4912e5ea74f3b7f"
BUILDING_SHA = "1c0d8c206eb2070785e594784fda90f615e6ed7fd9646d67e1a9de237b8cc9f4"
COMMIT = "c8ead6584b0553ad09aa1ac769f12456e29b4f5b"
FACILITY_SOURCE = {
    "name": "Compute Atlas", "author": "Edward Kubiak", "version": "v1.34.0",
    "url": f"https://github.com/ek33450505/compute-atlas/tree/{COMMIT}",
    "license": "CC BY 4.0", "licenseUrl": "https://creativecommons.org/licenses/by/4.0/",
    "snapshotDate": "2026-09-21", "commit": COMMIT, "sha256": FACILITY_SHA,
}
BUILDING_SOURCE = {
    "name": "PNNL-led IM3 Open Source Data Center Atlas / OpenStreetMap",
    "url": "https://doi.org/10.57931/3017294", "date": "2026-02-09",
    "license": "ODbL 1.0", "licenseUrl": "https://opendatacommons.org/licenses/odbl/1-0/",
    "sha256": BUILDING_SHA,
}
TERRITORIES = [("60", "AS", "American Samoa"), ("66", "GU", "Guam"),
               ("69", "MP", "Northern Mariana Islands"), ("72", "PR", "Puerto Rico"),
               ("78", "VI", "U.S. Virgin Islands")]


METHODOLOGY = """# Eco TerraVault intelligence atlas: sources and limits

Reviewed September 27, 2026. Two independent source layers describe different things. **Their counts must not be added together.** This atlas provides research context, not a complete census, live operating-status service or verified inventory of Eco TerraVault projects.

## Source-reported facility records

Source: **Compute Atlas by Edward Kubiak**, © 2026 Edward Kubiak, version `v1.34.0`, published September 21, 2026. The source is provided as is, without source-author endorsement of Eco TerraVault; its [original copyright, license and warranty notices](/data/intelligence/compute-atlas-license.txt) are retained. [Source project at the reviewed revision](https://github.com/ek33450505/compute-atlas/tree/c8ead6584b0553ad09aa1ac769f12456e29b4f5b) · [source methodology](https://github.com/ek33450505/compute-atlas/blob/c8ead6584b0553ad09aa1ac769f12456e29b4f5b/docs/methodology.md) · [CC BY 4.0](https://creativecommons.org/licenses/by/4.0/).

Eco TerraVault filters the 2,025-record release to its 1,605 records marked `data_center`, excluding 188 crypto-mining records and 232 power-generation records. It preserves source IDs, names, operators, city/jurisdiction, reported status, confidence, update dates and citations. A record may refer to a campus, an individual site, or grouped locations, not one building. Campus-level and individual-site records can overlap; these are source records, not a deduplicated count of unique physical facilities. The current status labels are source-reported: operational 747, proposed 424, under construction 238, permitted 108 and cancelled 88. They have not all been independently checked by Eco TerraVault. Eighteen records have the source confidence `rumored`; confidence is displayed rather than silently upgraded. Unknown facts remain unknown.

A cancelled proposal is not a decommissioned operating facility. This source does not supply dormant or decommissioned categories. Status histories are incomplete and can concern expansions; this release does not derive national trends or original opening dates from them. Neither `lastUpdated` nor a source retrieval date is an event date. Record counts are not power capacity, water consumption or environmental scores. Optional technical quantities are excluded from this public derivative.

The catalog has records in 55 of the 56 displayed jurisdictions. American Samoa has none in this snapshot; this does not prove there are no data centers there. Source coverage is partial and favors publicly documented facilities. Current live-source counts can differ because this atlas uses a reproducible frozen release.

Source SHA-256: `deca4eea8d71e904681f4a1d9ae2c64103157cdd69ee1462c4912e5ea74f3b7f`.

[[SETTING_METHODOLOGY]]

## Mapped building records

Source: **PNNL-led IM3 Open Source Data Center Atlas**, February 9, 2026 release. [Publication and DOI](https://doi.org/10.57931/3017294). The retained GeoPackage is from authors' repository commit `74ab37d5b9d200400a01639f9ffc3c3a8b716314`.

The building layer has 1,239 rows. Two county-boundary duplicate rows are removed by exact building ID, yielding **1,237 mapped building records** across the 56-jurisdiction display area. This comprises the existing 1,236 records in the 50 states and D.C. plus one Puerto Rico record (`00365941471`). Other territory zeros mean no records in this layer. Campus and point layers are excluded. We do not infer operational status, ownership, capacity, event history or site suitability from a mapped building.

Derived building data is available under the [Open Database License 1.0](https://opendatacommons.org/licenses/odbl/1-0/). Contains information from the IM3 atlas derived from © OpenStreetMap contributors. The complete derived building-ID table, source notices, license and aggregation code accompany this release. The facility compilation's CC BY license does not replace this layer's ODbL terms. Jurisdiction downloads identify both layers and licenses separately.

Source SHA-256: `1c0d8c206eb2070785e594784fda90f615e6ed7fd9646d67e1a9de237b8cc9f4`.

## Downloads and reproducibility

- The full normalized facility catalog is held in `lib/server/intelligence-facilities.json`, outside the public download directory. The website retrieves bounded pages of at most 24 records through `/api/intelligence/catalog`, using jurisdiction, reported lifecycle, reviewed physical setting, text search and offset filters. Every result retains source citations and snapshot/license metadata. No street coordinates, technical capacity or raw narrative notes are included.
- The former full-catalog JSON and CSV downloads are retired. Bounded browsing is a presentation and delivery choice, not authentication, a paid-access entitlement or a restriction on the original source's CC BY 4.0 reuse rights. The attributed upstream source remains linked above. Eco TerraVault's professional services concern scoped analysis and advice, not exclusive ownership of this public-source dataset.
- `summary.json` and `jurisdictions.csv`: all 56 jurisdictions, with separate facility/building totals, facility lifecycle counts and four physical-setting categories. Summary JSON also includes physical-setting-by-lifecycle cross-tabulations.
- `building-records.csv`: all 1,237 building IDs and their source jurisdiction.
- `compute-atlas-license.txt`, `ODbL-1.0.txt`, `im3-source-notices.txt`: source-specific notices.
- `build-intelligence.py`: maintained normalization and export code. Reproduction also requires the pinned source files retained in the project repository.

CSV text beginning with spreadsheet formula characters is prefixed with an apostrophe; this is an export safety transformation, not source content. JSON preserves the original text. Year/month/day precision is retained where supplied; no missing dates are filled in.

The Python standard-library pipeline verifies source checksums, uses SQLite to normalize facilities, citations, jurisdictions and building IDs, then emits a deterministic server-side catalog and separate public summary, building and license files. Its local SQLite workspace is not a public cloud database. `--check` compares rebuilt outputs without writing files and fails if the retired public full-catalog files exist. Human review precedes a new source release. A previous snapshot remains recoverable through version control.

Reported events, where separately shown, form a curated source register. They are not generated from lifecycle transitions, nearby natural hazards or missing facility records. Neither no recorded event nor no record establishes absence.
"""


def verified_bytes(path, expected):
    raw = path.read_bytes()
    if hashlib.sha256(raw).hexdigest() != expected:
        raise ValueError(f"Source checksum changed; review required: {path.name}")
    return raw


def http_url(value):
    if not isinstance(value, str) or any(char.isspace() for char in value):
        raise ValueError("Source URL must be a whitespace-free HTTP(S) URL")
    parsed = urlsplit(value)
    if parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username or parsed.password:
        raise ValueError(f"Unsupported source URL: {value}")
    return value


def source_date(value):
    if not isinstance(value, str) or not re.fullmatch(r"\d{4}(?:-\d{2}(?:-\d{2})?)?", value):
        raise ValueError(f"Unsupported date precision: {value!r}")
    date.fromisoformat(value + ("-01-01" if len(value) == 4 else "-01" if len(value) == 7 else ""))
    return value  # Preserve year/month/day precision; never fabricate a day.


def csv_cell(value):
    if isinstance(value, str) and (value.startswith(("\t", "\r", "\n")) or value.lstrip().startswith(("=", "+", "-", "@"))):
        return "'" + value
    return value


def csv_text(rows):
    stream = io.StringIO(newline="")
    csv.writer(stream, lineterminator="\n").writerows([[csv_cell(v) for v in row] for row in rows])
    return stream.getvalue()


def json_text(value, compact=False):
    return json.dumps(value, ensure_ascii=False, **({"separators": (",", ":")} if compact else {"indent": 2})) + "\n"


def reviewed_day(value):
    if not isinstance(value, str) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}", value):
        raise ValueError("Setting review dates require YYYY-MM-DD precision")
    return source_date(value)


def validated_settings(register, facilities):
    """Apply exact, human-reviewed evidence joins; never classify from keywords."""
    if not isinstance(register, dict) or set(register) != {"version", "catalogSourceSha256", "reviewedOn", "entries"} or register["version"] != 1:
        raise ValueError("Unsupported physical-setting register schema")
    if register["catalogSourceSha256"] != FACILITY_SHA:
        raise ValueError("Physical-setting review belongs to a different source snapshot; reconcile identities before reuse")
    register_date = reviewed_day(register["reviewedOn"])
    if register_date > REVIEWED or not isinstance(register["entries"], list):
        raise ValueError("Invalid physical-setting review coverage date or entries")
    catalog_by_id = {item["id"]: item for item in facilities}
    reviewed = {}
    required = {"catalogId", "jurisdiction", "category", "reviewedOn", "evidenceSummary", "limitations", "evidenceScope", "sources"}
    for entry in register["entries"]:
        if not isinstance(entry, dict) or set(entry) != required:
            raise ValueError("Physical-setting entry fields do not match the reviewed contract")
        identifier = entry["catalogId"]
        if not isinstance(identifier, str) or identifier not in catalog_by_id or identifier in reviewed:
            raise ValueError(f"Unknown or duplicate physical-setting catalog ID: {identifier}")
        if entry["jurisdiction"] != catalog_by_id[identifier]["jurisdiction"]:
            raise ValueError(f"Physical-setting jurisdiction mismatch: {identifier}")
        if entry["category"] not in SETTINGS:
            raise ValueError(f"Unsupported physical-setting category: {identifier}")
        if reviewed_day(entry["reviewedOn"]) > register_date:
            raise ValueError(f"Entry review occurs after the physical-setting register date: {identifier}")
        if any(not isinstance(entry[key], str) or not entry[key].strip() for key in ("evidenceSummary", "limitations", "evidenceScope")):
            raise ValueError(f"Physical-setting summary, limits and scope are required: {identifier}")
        if not isinstance(entry["sources"], list) or not entry["sources"]:
            raise ValueError(f"Physical-setting evidence requires a primary citation: {identifier}")
        citations = []
        seen_urls = set()
        for source in entry["sources"]:
            if not isinstance(source, dict) or not {"url", "title", "publisher"}.issubset(source) or set(source) - {"url", "title", "publisher", "retrievedAt"}:
                raise ValueError(f"Invalid physical-setting citation schema: {identifier}")
            url = http_url(source["url"])
            if url in seen_urls or any(not isinstance(source[key], str) or not source[key].strip() for key in ("title", "publisher")):
                raise ValueError(f"Physical-setting citation is empty or duplicated: {identifier}")
            seen_urls.add(url)
            citation = {"url": url, "title": source["title"], "publisher": source["publisher"]}
            if "retrievedAt" in source:
                citation["retrievedAt"] = source_date(source["retrievedAt"])
                if citation["retrievedAt"] > entry["reviewedOn"]:
                    raise ValueError(f"Physical-setting citation was retrieved after its review: {identifier}")
            citations.append(citation)
        reviewed[identifier] = {key: entry[key] for key in ("category", "reviewedOn", "evidenceSummary", "limitations", "evidenceScope")}
        reviewed[identifier]["sources"] = citations
    return {item["id"]: reviewed.get(item["id"], {
        "category": "unclassified", "reviewedOn": None,
        "evidenceSummary": "No reviewed physical-setting evidence is attached to this catalog record.",
        "limitations": "Unclassified is a research coverage gap, not evidence that a facility is above ground or underground.",
        "evidenceScope": "Physical data-hall setting has not been established for this source record.",
        "sources": [],
    }) for item in facilities}


def setting_methodology(review, counts):
    return f"""## Physical setting: reviewed evidence

Eco TerraVault review date: **{review['reviewedOn']}**. The physical-setting evidence register addresses **{review['reviewedRecords']} selected catalog records**, of which **{review['classifiedRecords']} are classified**. The complete 1,605-record catalog contains {counts['above_ground']} above-ground, {counts['underground']} underground, {counts['mixed']} mixed and {counts['unclassified']} unclassified records. This small, purposefully selected review is not representative coverage and does not establish national setting totals or market shares.

- **Above ground**: a reviewed primary source supports above-ground data halls for the identified record and stated scope. A multistory building may still be above ground.
- **Underground**: a reviewed primary source supports data halls below ground or inside a subterranean facility for the identified record and stated scope.
- **Mixed**: primary evidence explicitly establishes both above-ground and underground data halls within the same catalog record's scope. Conflicting or ambiguous descriptions do not establish mixed use.
- **Unclassified**: available reviewed evidence does not establish a physical setting, or the record has not received a setting review. Missing evidence never defaults to above ground.

The register `scripts/assets/intelligence-setting-evidence.json` is bound to the pinned catalog's SHA-256 and joins by exact catalog ID and jurisdiction. A new source snapshot requires explicit review of record identities and scope before the evidence register can be reused. The offline build rejects duplicate or unknown IDs, jurisdiction mismatches, unsupported categories, incomplete review metadata and missing or malformed citations. It copies only the explicitly reviewed category; it does not scan facility names or source text for keywords. References to underground utility feeds, fiber, water pipes, cooling loops or power generation do not establish underground data halls. Each classified record carries its review date, summary, limitations, evidence scope and primary-source links. Unreviewed records carry a null setting-review date and no setting citations.

Setting is independent of the inherited Compute Atlas lifecycle. A source may describe a proposed building or one part of a campus; the evidence scope and limitations must remain visible. The review does not update lifecycle status, prove current operation, certify a design, establish a national underground inventory, or infer environmental performance. Records and campus-level overlaps remain counted as source records. National and jurisdiction setting totals and setting-by-lifecycle tables reconcile exactly to their existing catalog totals; building and event layers are unchanged.

Evidence register SHA-256: `{review['registerSha256']}`. Changes to this editorial register require a new evidence review and deterministic regeneration alongside the original source snapshots."""


def build():
    raw = json.loads(verified_bytes(ASSETS / "compute-atlas-v1.34.0-facilities.json", FACILITY_SHA))
    building_file = ASSETS / "im3-data-centers-2026-02-09.gpkg"
    verified_bytes(building_file, BUILDING_SHA)
    state_geometry = json.loads((ROOT / "lib/us-map-paths.json").read_text())
    jurisdictions = [{"abbr": s["abbr"], "fips": s["fips"], "name": s["name"],
                      "kind": "district" if s["abbr"] == "DC" else "state"} for s in state_geometry["states"]]
    jurisdictions += [{"abbr": a, "fips": f, "name": n, "kind": "territory"} for f, a, n in TERRITORIES]
    jurisdictions.sort(key=lambda item: item["name"])
    lookup = {j["abbr"]: j for j in jurisdictions}
    if len(lookup) != 56 or len({j["fips"] for j in jurisdictions}) != 56:
        raise ValueError("Expected 50 states, D.C. and five territories")
    db = sqlite3.connect(":memory:")
    db.execute("PRAGMA foreign_keys = ON")
    db.executescript("""
      CREATE TABLE jurisdictions(abbr TEXT PRIMARY KEY, fips TEXT UNIQUE NOT NULL, name TEXT NOT NULL, kind TEXT NOT NULL);
      CREATE TABLE facilities(id TEXT PRIMARY KEY, name TEXT NOT NULL, operator TEXT NOT NULL,
        jurisdiction TEXT NOT NULL REFERENCES jurisdictions(abbr), city TEXT NOT NULL, status TEXT NOT NULL,
        confidence TEXT NOT NULL, last_updated TEXT NOT NULL);
      CREATE TABLE sources(facility_id TEXT NOT NULL REFERENCES facilities(id), position INTEGER NOT NULL,
        url TEXT NOT NULL, title TEXT NOT NULL, publisher TEXT NOT NULL, retrieved_at TEXT NOT NULL,
        PRIMARY KEY(facility_id, position));
      CREATE TABLE facility_settings(facility_id TEXT PRIMARY KEY REFERENCES facilities(id),
        category TEXT NOT NULL, reviewed_on TEXT, evidence_summary TEXT NOT NULL,
        limitations TEXT NOT NULL, evidence_scope TEXT NOT NULL);
      CREATE TABLE setting_sources(facility_id TEXT NOT NULL REFERENCES facility_settings(facility_id),
        position INTEGER NOT NULL, url TEXT NOT NULL, title TEXT NOT NULL, publisher TEXT NOT NULL,
        retrieved_at TEXT, PRIMARY KEY(facility_id, position));
      CREATE TABLE building_records(id TEXT PRIMARY KEY, jurisdiction TEXT NOT NULL REFERENCES jurisdictions(abbr));
      CREATE TABLE metadata(key TEXT PRIMARY KEY, value TEXT NOT NULL);
      CREATE INDEX facilities_jurisdiction_status ON facilities(jurisdiction, status);
    """)
    db.executemany("INSERT INTO jurisdictions VALUES(?,?,?,?)", [(j["abbr"], j["fips"], j["name"], j["kind"]) for j in jurisdictions])
    all_ids = set()
    for item in raw:
        identifier = item["id"]
        if not re.fullmatch(r"[a-z0-9-]+", identifier) or identifier in all_ids:
            raise ValueError(f"Invalid or duplicate source ID: {identifier}")
        all_ids.add(identifier)
        if item["facilityType"] not in ("data_center", "crypto_mining", "power_generation"):
            raise ValueError("Unknown source facility type")
        if item["facilityType"] != "data_center":
            continue
        if item["status"] not in STATUSES or item["confidence"] not in ("confirmed", "reported", "rumored"):
            raise ValueError(f"Unsupported lifecycle/confidence: {identifier}")
        if item["location"]["state"] not in lookup or not item["sources"]:
            raise ValueError(f"Missing jurisdiction or sources: {identifier}")
        values = (identifier, item["name"], item["operator"], item["location"]["state"],
                  item["location"].get("city", ""), item["status"], item["confidence"], source_date(item["lastUpdated"]))
        if any(not isinstance(v, str) for v in values) or not values[1] or not values[2]:
            raise ValueError(f"Invalid public identity fields: {identifier}")
        db.execute("INSERT INTO facilities VALUES(?,?,?,?,?,?,?,?)", values)
        for position, source in enumerate(item["sources"]):
            if not isinstance(source["label"], str) or not source["label"]:
                raise ValueError(f"Source title missing: {identifier}")
            db.execute("INSERT INTO sources VALUES(?,?,?,?,?,?)", (identifier, position, http_url(source["url"]),
                       source["label"], source.get("publisher", ""), source_date(source["retrievedAt"])))
    with sqlite3.connect(f"file:{building_file}?mode=ro", uri=True) as source_db:
        rows = source_db.execute("SELECT id, state_abb, state_id, type FROM building").fetchall()
    if len(rows) != 1239:
        raise ValueError("Unexpected IM3 building-layer row count")
    building_ids = {}
    for identifier, abbr, fips, kind in rows:
        if not identifier or kind != "building" or abbr not in lookup or str(fips).zfill(2) != lookup[abbr]["fips"]:
            raise ValueError("Invalid IM3 building identity")
        if identifier in building_ids and building_ids[identifier] != abbr:
            raise ValueError("Building assigned to multiple jurisdictions")
        building_ids[identifier] = abbr
    db.executemany("INSERT INTO building_records VALUES(?,?)", sorted(building_ids.items()))
    facilities = []
    for row in db.execute("SELECT * FROM facilities ORDER BY id"):
        record = dict(zip(("id", "name", "operator", "jurisdiction", "city", "status", "confidence", "lastUpdated"), row))
        record["sources"] = [dict(zip(("url", "title", "publisher", "retrievedAt"), s)) for s in db.execute(
            "SELECT url,title,publisher,retrieved_at FROM sources WHERE facility_id=? ORDER BY position", (record["id"],))]
        facilities.append(record)
    setting_bytes = SETTING_REGISTER.read_bytes()
    setting_register = json.loads(setting_bytes)
    settings = validated_settings(setting_register, facilities)
    for record in facilities:
        setting = settings[record["id"]]
        record["setting"] = setting
        db.execute("INSERT INTO facility_settings VALUES(?,?,?,?,?,?)", (record["id"], setting["category"],
                   setting["reviewedOn"], setting["evidenceSummary"], setting["limitations"], setting["evidenceScope"]))
        for position, source in enumerate(setting["sources"]):
            db.execute("INSERT INTO setting_sources VALUES(?,?,?,?,?,?)", (record["id"], position,
                       source["url"], source["title"], source["publisher"], source.get("retrievedAt")))
    for j in jurisdictions:
        j["facilities"] = db.execute("SELECT count(*) FROM facilities WHERE jurisdiction=?", (j["abbr"],)).fetchone()[0]
        j["buildings"] = db.execute("SELECT count(*) FROM building_records WHERE jurisdiction=?", (j["abbr"],)).fetchone()[0]
        j["statuses"] = {s: db.execute("SELECT count(*) FROM facilities WHERE jurisdiction=? AND status=?", (j["abbr"], s)).fetchone()[0] for s in STATUSES}
        local = [record for record in facilities if record["jurisdiction"] == j["abbr"]]
        j["settings"] = {setting: sum(record["setting"]["category"] == setting for record in local) for setting in SETTINGS}
        j["settingStatuses"] = {setting: {status: sum(record["setting"]["category"] == setting and record["status"] == status for record in local)
                                           for status in STATUSES} for setting in SETTINGS}
    status_totals = {s: sum(j["statuses"][s] for j in jurisdictions) for s in STATUSES}
    setting_totals = {setting: sum(j["settings"][setting] for j in jurisdictions) for setting in SETTINGS}
    setting_status_totals = {setting: {status: sum(j["settingStatuses"][setting][status] for j in jurisdictions)
                                      for status in STATUSES} for setting in SETTINGS}
    setting_review = {"reviewedOn": setting_register["reviewedOn"], "reviewedRecords": len(setting_register["entries"]),
                      "classifiedRecords": sum(setting_totals[setting] for setting in SETTINGS if setting != "unclassified"),
                      "registerSha256": hashlib.sha256(setting_bytes).hexdigest()}
    if sum(setting_totals.values()) != len(facilities) or any(sum(setting_status_totals[setting][status] for setting in SETTINGS) != status_totals[status] for status in STATUSES):
        raise ValueError("Physical-setting counts do not reconcile to the source catalog")
    if len(raw) != 2025 or len(facilities) != 1605 or len(building_ids) != 1237:
        raise ValueError("Reviewed source totals changed")
    if status_totals != dict(zip(STATUSES, (747, 424, 238, 108, 88))):
        raise ValueError("Reviewed lifecycle totals changed")
    legacy = json.loads((ROOT / "lib/data-center-counts.json").read_text())
    if any(j["buildings"] != legacy["counts"][j["fips"]] for j in jurisdictions if j["kind"] != "territory"):
        raise ValueError("Expanded building layer differs from the existing 51-jurisdiction layer")
    if {k: v for k, v in building_ids.items() if lookup[v]["kind"] == "territory"} != {"00365941471": "PR"}:
        raise ValueError("Unexpected territorial building records")
    summary = {"version": 1, "reviewedOn": REVIEWED, "facilitySource": FACILITY_SOURCE,
               "buildingSource": BUILDING_SOURCE, "facilityTotal": len(facilities), "buildingTotal": len(building_ids),
               "statusTotals": status_totals, "settingTotals": setting_totals, "settingStatusTotals": setting_status_totals,
               "settingReview": setting_review, "jurisdictions": jurisdictions}
    catalog = {"version": 1, "reviewedOn": REVIEWED, "facilitySource": FACILITY_SOURCE,
               "unit": "source-reported data-center facility records", "total": len(facilities), "records": facilities}
    jurisdiction_csv = [["abbr", "fips", "jurisdiction", "kind", "source_reported_facilities", "mapped_buildings", *STATUSES, *[f"setting_{setting}" for setting in SETTINGS], "facility_snapshot", "building_snapshot", "coverage_note", "facility_license", "building_license"]]
    jurisdiction_csv += [[j["abbr"], j["fips"], j["name"], j["kind"], j["facilities"], j["buildings"], *[j["statuses"][s] for s in STATUSES], *[j["settings"][setting] for setting in SETTINGS], FACILITY_SOURCE["snapshotDate"], BUILDING_SOURCE["date"], "Partial source and setting review coverage; none recorded is not verified absence", FACILITY_SOURCE["licenseUrl"], BUILDING_SOURCE["licenseUrl"]] for j in jurisdictions]
    building_csv = [["building_id", "jurisdiction", "source_date", "license_url"]] + [[i, a, BUILDING_SOURCE["date"], BUILDING_SOURCE["licenseUrl"]] for i, a in sorted(building_ids.items())]
    db.executemany("INSERT INTO metadata VALUES(?,?)", [("facility_source", json_text(FACILITY_SOURCE)), ("building_source", json_text(BUILDING_SOURCE)),
                   ("setting_review", json_text(setting_review)), ("reviewed_on", REVIEWED), ("schema_version", "2")])
    db.commit()
    if db.execute("PRAGMA integrity_check").fetchone()[0] != "ok" or db.execute("PRAGMA foreign_key_check").fetchall():
        raise ValueError("Normalized catalog database failed integrity validation")
    outputs = {ROOT / "lib/intelligence-summary.json": json_text(summary), PUBLIC / "summary.json": json_text(summary),
               ROOT / "lib/server/intelligence-facilities.json": json_text(catalog, compact=True),
               PUBLIC / "jurisdictions.csv": csv_text(jurisdiction_csv), PUBLIC / "building-records.csv": csv_text(building_csv),
               PUBLIC / "compute-atlas-license.txt": (ASSETS / "compute-atlas-v1.34.0-LICENSE-DATA.txt").read_text(),
               PUBLIC / "ODbL-1.0.txt": (ROOT / "public/data/ODbL-1.0.txt").read_text(),
               PUBLIC / "im3-source-notices.txt": (ROOT / "public/data/im3-source-notices.txt").read_text(),
               PUBLIC / "build-intelligence.py": Path(__file__).read_text(),
               PUBLIC / "methodology.md": METHODOLOGY.replace("[[SETTING_METHODOLOGY]]", setting_methodology(setting_review, setting_totals))}
    return db, outputs


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--check", action="store_true", help="Compare server/public outputs and reject retired downloads without writing files")
    parser.add_argument("--database", type=Path, default=ROOT / ".backups/intelligence.sqlite", help="Local reproducible SQLite output; never a public server database")
    args = parser.parse_args()
    db, outputs = build()
    try:
        for path in RETIRED_PUBLIC:
            if args.check and path.exists():
                raise SystemExit(f"Retired public full-catalog file must be removed: {path.relative_to(ROOT)}")
        for path, value in outputs.items():
            if args.check:
                if not path.is_file() or path.read_text(encoding="utf-8") != value:
                    raise SystemExit(f"Snapshot differs: {path.relative_to(ROOT)}")
            else:
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_text(value, encoding="utf-8")
        if not args.check:
            for path in RETIRED_PUBLIC:
                path.unlink(missing_ok=True)
            args.database.parent.mkdir(parents=True, exist_ok=True)
            with sqlite3.connect(args.database) as target:
                db.backup(target)
        print(f"{'Verified' if args.check else 'Generated'} 1,605 facility records and 1,237 building records in separate layers across 56 jurisdictions.")
    finally:
        db.close()


if __name__ == "__main__":
    main()