728 lines
26 KiB
Python
728 lines
26 KiB
Python
"""Build a private, point-in-time inspection feature mart with DuckDB.
|
|
|
|
Inputs are either a contiguous set of bounded monthly CSV.gz exports or the
|
|
single complete-history development sample. Every input must have the manifest
|
|
written by the corresponding secure exporter.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import gzip
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
|
|
|
|
import duckdb
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
PRIVATE_DATA_ROOT = (PROJECT_ROOT / "data/private").resolve()
|
|
DEFAULT_INPUT = PRIVATE_DATA_ROOT / "inspection_batches"
|
|
DEFAULT_DATABASE = PRIVATE_DATA_ROOT / "warehouse/vehicle_health.duckdb"
|
|
DEFAULT_OUTPUT = PRIVATE_DATA_ROOT / "marts/inspection_feature_mart.parquet"
|
|
SQL_ROOT = PROJECT_ROOT / "sql/local"
|
|
|
|
EXPECTED_COLUMNS = (
|
|
"internal_event_id",
|
|
"vehicle_token",
|
|
"vehicle_bucket",
|
|
"event_ts",
|
|
"source_era",
|
|
"public_county",
|
|
"canonical_outcome",
|
|
"outcome_label_source",
|
|
"program_type",
|
|
"test_type",
|
|
"observed_make",
|
|
"observed_model",
|
|
"observed_model_year",
|
|
)
|
|
BOUNDED_KIND = "bounded_batch"
|
|
DEVELOPMENT_KIND = "vehicle_history_development_sample"
|
|
HEX_64 = re.compile(r"^[0-9a-f]{64}$")
|
|
FINGERPRINT = re.compile(r"^[0-9a-f]{16,64}$")
|
|
MEMORY_LIMIT = re.compile(r"^[1-9][0-9]*(?:MB|GB|TB)$", re.IGNORECASE)
|
|
|
|
SQL_FILES = (
|
|
"20_stage_events.sql",
|
|
"21_build_episodes.sql",
|
|
"22_build_features.sql",
|
|
"23_validate_mart.sql",
|
|
)
|
|
|
|
CREATE_STAGING_SQL = """
|
|
CREATE TABLE stg_events (
|
|
batch_file VARCHAR NOT NULL,
|
|
batch_kind VARCHAR NOT NULL,
|
|
batch_start TIMESTAMP NOT NULL,
|
|
batch_end TIMESTAMP NOT NULL,
|
|
internal_event_id VARCHAR,
|
|
vehicle_token VARCHAR,
|
|
vehicle_bucket VARCHAR,
|
|
event_ts VARCHAR,
|
|
source_era VARCHAR,
|
|
public_county VARCHAR,
|
|
canonical_outcome VARCHAR,
|
|
outcome_label_source VARCHAR,
|
|
program_type VARCHAR,
|
|
test_type VARCHAR,
|
|
observed_make VARCHAR,
|
|
observed_model VARCHAR,
|
|
observed_model_year VARCHAR
|
|
)
|
|
"""
|
|
|
|
# The dialect and every input column are explicit. In particular, do not let a
|
|
# sample without quoted values convince DuckDB that quote handling is disabled.
|
|
INGEST_SQL = """
|
|
INSERT INTO stg_events
|
|
SELECT
|
|
? AS batch_file,
|
|
? AS batch_kind,
|
|
?::TIMESTAMP AS batch_start,
|
|
?::TIMESTAMP AS batch_end,
|
|
source.internal_event_id,
|
|
source.vehicle_token,
|
|
source.vehicle_bucket,
|
|
source.event_ts,
|
|
source.source_era,
|
|
source.public_county,
|
|
source.canonical_outcome,
|
|
source.outcome_label_source,
|
|
source.program_type,
|
|
source.test_type,
|
|
source.observed_make,
|
|
source.observed_model,
|
|
source.observed_model_year
|
|
FROM read_csv(
|
|
?,
|
|
header = true,
|
|
auto_detect = false,
|
|
delim = ',',
|
|
quote = '"',
|
|
escape = '"',
|
|
nullstr = '',
|
|
strict_mode = true,
|
|
columns = {
|
|
'internal_event_id': 'VARCHAR',
|
|
'vehicle_token': 'VARCHAR',
|
|
'vehicle_bucket': 'VARCHAR',
|
|
'event_ts': 'VARCHAR',
|
|
'source_era': 'VARCHAR',
|
|
'public_county': 'VARCHAR',
|
|
'canonical_outcome': 'VARCHAR',
|
|
'outcome_label_source': 'VARCHAR',
|
|
'program_type': 'VARCHAR',
|
|
'test_type': 'VARCHAR',
|
|
'observed_make': 'VARCHAR',
|
|
'observed_model': 'VARCHAR',
|
|
'observed_model_year': 'VARCHAR'
|
|
}
|
|
) AS source
|
|
"""
|
|
|
|
|
|
class MartBuildError(RuntimeError):
|
|
"""Raised when private input or a mart invariant fails closed."""
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class InputBatch:
|
|
path: Path
|
|
manifest_path: Path
|
|
export_kind: str
|
|
start: datetime
|
|
end: datetime
|
|
rows: int
|
|
query_version: str
|
|
query_sha256: str
|
|
key_version: str
|
|
key_fingerprint: str
|
|
compressed_sha256: str
|
|
compressed_bytes: int
|
|
manifest: Mapping[str, Any]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BuildConfig:
|
|
input_path: Path = DEFAULT_INPUT
|
|
database_path: Path = DEFAULT_DATABASE
|
|
output_path: Path = DEFAULT_OUTPUT
|
|
episode_gap_days: int = 30
|
|
memory_limit: str = "4GB"
|
|
threads: int = max(1, min(4, os.cpu_count() or 1))
|
|
allow_gaps: bool = False
|
|
allow_external_output: bool = False
|
|
overwrite: bool = False
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class BuildSummary:
|
|
database_path: Path
|
|
output_path: Path
|
|
manifest_path: Path
|
|
input_rows: int
|
|
clean_events: int
|
|
episodes: int
|
|
mart_rows: int
|
|
eligible_rows: int
|
|
source_data_kind: str
|
|
population_estimate_allowed: bool
|
|
|
|
|
|
def file_sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def parse_manifest_timestamp(value: object, field: str) -> datetime:
|
|
if not isinstance(value, str):
|
|
raise MartBuildError(f"Manifest field {field!r} must be an ISO timestamp")
|
|
encoded = value[:-1] + "+00:00" if value.endswith("Z") else value
|
|
try:
|
|
parsed = datetime.fromisoformat(encoded)
|
|
except ValueError as exc:
|
|
raise MartBuildError(f"Manifest field {field!r} is not ISO-8601") from exc
|
|
if parsed.tzinfo is not None:
|
|
parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None)
|
|
return parsed
|
|
|
|
|
|
def _required_text(manifest: Mapping[str, Any], field: str) -> str:
|
|
value = manifest.get(field)
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise MartBuildError(f"Manifest field {field!r} must be non-empty text")
|
|
return value.strip()
|
|
|
|
|
|
def _required_nonnegative_int(manifest: Mapping[str, Any], field: str) -> int:
|
|
value = manifest.get(field)
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
|
|
raise MartBuildError(f"Manifest field {field!r} must be a nonnegative integer")
|
|
return value
|
|
|
|
|
|
def _read_header(path: Path) -> Tuple[str, ...]:
|
|
try:
|
|
with gzip.open(path, "rt", encoding="utf-8", newline="") as handle:
|
|
row = next(csv.reader(handle), None)
|
|
except (OSError, UnicodeError, csv.Error) as exc:
|
|
raise MartBuildError(f"Cannot read gzip CSV header: {path}") from exc
|
|
if row is None:
|
|
raise MartBuildError(f"Input gzip CSV is empty: {path}")
|
|
return tuple(row)
|
|
|
|
|
|
def _load_batch(path: Path) -> InputBatch:
|
|
manifest_path = path.with_name(path.name + ".manifest.json")
|
|
if not manifest_path.is_file():
|
|
raise MartBuildError(f"Missing paired manifest: {manifest_path}")
|
|
try:
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError) as exc:
|
|
raise MartBuildError(f"Cannot parse manifest: {manifest_path}") from exc
|
|
if not isinstance(manifest, dict):
|
|
raise MartBuildError(f"Manifest must contain a JSON object: {manifest_path}")
|
|
|
|
columns = manifest.get("columns")
|
|
if not isinstance(columns, (list, tuple)) or tuple(columns) != EXPECTED_COLUMNS:
|
|
raise MartBuildError(f"Unexpected manifest columns: {manifest_path}")
|
|
if _read_header(path) != EXPECTED_COLUMNS:
|
|
raise MartBuildError(f"CSV header does not match the private schema: {path}")
|
|
|
|
export_kind = _required_text(manifest, "export_kind")
|
|
if export_kind not in (BOUNDED_KIND, DEVELOPMENT_KIND):
|
|
raise MartBuildError(f"Unsupported export_kind {export_kind!r}: {manifest_path}")
|
|
start = parse_manifest_timestamp(
|
|
manifest.get("source_start_inclusive"), "source_start_inclusive"
|
|
)
|
|
end = parse_manifest_timestamp(
|
|
manifest.get("source_end_exclusive"), "source_end_exclusive"
|
|
)
|
|
if end <= start:
|
|
raise MartBuildError(f"Manifest source interval is empty or reversed: {path}")
|
|
|
|
rows = _required_nonnegative_int(manifest, "rows")
|
|
compressed_bytes = _required_nonnegative_int(manifest, "compressed_file_bytes")
|
|
actual_bytes = path.stat().st_size
|
|
if compressed_bytes != actual_bytes:
|
|
raise MartBuildError(
|
|
f"Compressed byte count mismatch for {path}: "
|
|
f"manifest={compressed_bytes}, actual={actual_bytes}"
|
|
)
|
|
compressed_sha256 = _required_text(manifest, "compressed_file_sha256").lower()
|
|
if not HEX_64.fullmatch(compressed_sha256):
|
|
raise MartBuildError(f"Invalid compressed SHA-256 in {manifest_path}")
|
|
actual_sha256 = file_sha256(path)
|
|
if compressed_sha256 != actual_sha256:
|
|
raise MartBuildError(f"Compressed SHA-256 mismatch for {path}")
|
|
|
|
query_sha256 = _required_text(manifest, "query_sha256").lower()
|
|
if not HEX_64.fullmatch(query_sha256):
|
|
raise MartBuildError(f"Invalid query SHA-256 in {manifest_path}")
|
|
key_fingerprint = _required_text(
|
|
manifest, "vehicle_token_key_fingerprint"
|
|
).lower()
|
|
if not FINGERPRINT.fullmatch(key_fingerprint):
|
|
raise MartBuildError(f"Invalid key fingerprint in {manifest_path}")
|
|
|
|
if export_kind == BOUNDED_KIND:
|
|
if manifest.get("classification") != (
|
|
"private_pseudonymized_analytical_staging"
|
|
):
|
|
raise MartBuildError(f"Unexpected bounded-batch classification: {path}")
|
|
else:
|
|
if manifest.get("classification") != (
|
|
"private_pseudonymized_development_only"
|
|
):
|
|
raise MartBuildError(f"Unexpected development classification: {path}")
|
|
if manifest.get("population_estimate_allowed") is not False:
|
|
raise MartBuildError(
|
|
"Development history manifest must prohibit population estimates"
|
|
)
|
|
|
|
return InputBatch(
|
|
path=path,
|
|
manifest_path=manifest_path,
|
|
export_kind=export_kind,
|
|
start=start,
|
|
end=end,
|
|
rows=rows,
|
|
query_version=_required_text(manifest, "query_version"),
|
|
query_sha256=query_sha256,
|
|
key_version=_required_text(manifest, "vehicle_token_key_version"),
|
|
key_fingerprint=key_fingerprint,
|
|
compressed_sha256=compressed_sha256,
|
|
compressed_bytes=compressed_bytes,
|
|
manifest=manifest,
|
|
)
|
|
|
|
|
|
def _discover_files(input_path: Path) -> List[Path]:
|
|
if input_path.is_file():
|
|
if not input_path.name.endswith(".csv.gz"):
|
|
raise MartBuildError("A feature-mart input file must end in .csv.gz")
|
|
return [input_path]
|
|
if not input_path.is_dir():
|
|
raise MartBuildError(f"Input path does not exist: {input_path}")
|
|
paths = sorted(path for path in input_path.glob("*.csv.gz") if path.is_file())
|
|
if not paths:
|
|
raise MartBuildError(f"No .csv.gz inputs found in {input_path}")
|
|
return paths
|
|
|
|
|
|
def discover_and_validate_batches(config: BuildConfig) -> List[InputBatch]:
|
|
input_path = config.input_path.resolve()
|
|
if not config.allow_external_output and not input_path.is_relative_to(
|
|
PRIVATE_DATA_ROOT
|
|
):
|
|
raise MartBuildError(f"Private inputs must stay under {PRIVATE_DATA_ROOT}")
|
|
|
|
batches = [_load_batch(path.resolve()) for path in _discover_files(input_path)]
|
|
kinds = {batch.export_kind for batch in batches}
|
|
if len(kinds) != 1:
|
|
raise MartBuildError("Do not mix bounded batches and development samples")
|
|
kind = next(iter(kinds))
|
|
|
|
key_versions = {batch.key_version for batch in batches}
|
|
fingerprints = {batch.key_fingerprint for batch in batches}
|
|
if len(key_versions) != 1 or len(fingerprints) != 1:
|
|
raise MartBuildError("All inputs must use one VIN-token key and version")
|
|
|
|
if kind == DEVELOPMENT_KIND:
|
|
if len(batches) != 1:
|
|
raise MartBuildError("A development mart accepts exactly one history sample")
|
|
return batches
|
|
|
|
query_versions = {batch.query_version for batch in batches}
|
|
query_hashes = {batch.query_sha256 for batch in batches}
|
|
if len(query_versions) != 1 or len(query_hashes) != 1:
|
|
raise MartBuildError("All bounded batches must use one extraction query version")
|
|
|
|
batches.sort(key=lambda batch: (batch.start, batch.end, str(batch.path)))
|
|
for previous, current in zip(batches, batches[1:]):
|
|
if current.start < previous.end:
|
|
raise MartBuildError(
|
|
f"Source intervals overlap: {previous.path.name} and "
|
|
f"{current.path.name}"
|
|
)
|
|
if current.start > previous.end and not config.allow_gaps:
|
|
raise MartBuildError(
|
|
f"Source interval gap between {previous.end.isoformat()} and "
|
|
f"{current.start.isoformat()}; use --allow-gaps only for an "
|
|
"explicitly incomplete analytical build"
|
|
)
|
|
return batches
|
|
|
|
|
|
def _validate_config(config: BuildConfig) -> BuildConfig:
|
|
if not 1 <= config.episode_gap_days <= 365:
|
|
raise MartBuildError("episode_gap_days must be between 1 and 365")
|
|
if not 1 <= config.threads <= 64:
|
|
raise MartBuildError("threads must be between 1 and 64")
|
|
if not MEMORY_LIMIT.fullmatch(config.memory_limit):
|
|
raise MartBuildError("memory_limit must look like 4096MB or 4GB")
|
|
|
|
database_path = config.database_path.resolve()
|
|
output_path = config.output_path.resolve()
|
|
if output_path.suffix.lower() != ".parquet":
|
|
raise MartBuildError("Feature-mart output must use a .parquet suffix")
|
|
if not config.allow_external_output:
|
|
for path in (database_path, output_path):
|
|
if not path.is_relative_to(PRIVATE_DATA_ROOT):
|
|
raise MartBuildError(f"Private outputs must stay under {PRIVATE_DATA_ROOT}")
|
|
if database_path == output_path:
|
|
raise MartBuildError("Database and Parquet output paths must differ")
|
|
|
|
return BuildConfig(
|
|
input_path=config.input_path.resolve(),
|
|
database_path=database_path,
|
|
output_path=output_path,
|
|
episode_gap_days=config.episode_gap_days,
|
|
memory_limit=config.memory_limit.upper(),
|
|
threads=config.threads,
|
|
allow_gaps=config.allow_gaps,
|
|
allow_external_output=config.allow_external_output,
|
|
overwrite=config.overwrite,
|
|
)
|
|
|
|
|
|
def _load_sql(name: str) -> str:
|
|
path = SQL_ROOT / name
|
|
try:
|
|
return path.read_text(encoding="utf-8")
|
|
except OSError as exc:
|
|
raise MartBuildError(f"Cannot read local transform SQL: {path}") from exc
|
|
|
|
|
|
def _remove_partial(paths: Iterable[Path]) -> None:
|
|
for path in paths:
|
|
path.unlink(missing_ok=True)
|
|
|
|
|
|
def _output_columns(connection: duckdb.DuckDBPyConnection) -> List[Dict[str, str]]:
|
|
rows = connection.execute("DESCRIBE SELECT * FROM feature_mart").fetchall()
|
|
return [
|
|
{"name": str(row[0]), "duckdb_type": str(row[1])}
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def build_feature_mart(config: BuildConfig) -> BuildSummary:
|
|
config = _validate_config(config)
|
|
batches = discover_and_validate_batches(config)
|
|
source_data_kind = batches[0].export_kind
|
|
population_estimate_allowed = (
|
|
source_data_kind == BOUNDED_KIND and not config.allow_gaps
|
|
)
|
|
|
|
database_path = config.database_path
|
|
output_path = config.output_path
|
|
manifest_path = output_path.with_name(output_path.name + ".manifest.json")
|
|
partial_database = database_path.with_name(database_path.name + ".partial")
|
|
partial_output = output_path.with_name(output_path.name + ".partial")
|
|
partial_manifest = manifest_path.with_name(manifest_path.name + ".partial")
|
|
partial_wal = partial_database.with_name(partial_database.name + ".wal")
|
|
partial_paths = (partial_database, partial_output, partial_manifest, partial_wal)
|
|
|
|
finals = (database_path, output_path, manifest_path)
|
|
existing = [path for path in finals if path.exists()]
|
|
if existing and not config.overwrite:
|
|
raise MartBuildError(
|
|
"Refusing to overwrite existing outputs: "
|
|
+ ", ".join(str(path) for path in existing)
|
|
)
|
|
|
|
os.umask(0o077)
|
|
database_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
temp_directory = database_path.parent / ".duckdb_tmp"
|
|
temp_directory.mkdir(parents=True, exist_ok=True, mode=0o700)
|
|
_remove_partial(partial_paths)
|
|
|
|
connection: Optional[duckdb.DuckDBPyConnection] = None
|
|
try:
|
|
connection = duckdb.connect(str(partial_database))
|
|
connection.execute("SET threads = ?", [config.threads])
|
|
connection.execute("SET memory_limit = ?", [config.memory_limit])
|
|
connection.execute("SET temp_directory = ?", [str(temp_directory)])
|
|
connection.execute("SET preserve_insertion_order = false")
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE build_config AS
|
|
SELECT
|
|
?::INTEGER AS episode_gap_days,
|
|
?::VARCHAR AS source_data_kind,
|
|
?::BOOLEAN AS population_estimate_allowed
|
|
""",
|
|
[
|
|
config.episode_gap_days,
|
|
source_data_kind,
|
|
population_estimate_allowed,
|
|
],
|
|
)
|
|
connection.execute(CREATE_STAGING_SQL)
|
|
|
|
for batch in batches:
|
|
connection.execute(
|
|
INGEST_SQL,
|
|
[
|
|
str(batch.path),
|
|
batch.export_kind,
|
|
batch.start.isoformat(sep=" "),
|
|
batch.end.isoformat(sep=" "),
|
|
str(batch.path),
|
|
],
|
|
)
|
|
|
|
actual_by_file = dict(
|
|
connection.execute(
|
|
"SELECT batch_file, count(*) FROM stg_events GROUP BY batch_file"
|
|
).fetchall()
|
|
)
|
|
for batch in batches:
|
|
actual = int(actual_by_file.get(str(batch.path), 0))
|
|
if actual != batch.rows:
|
|
raise MartBuildError(
|
|
f"CSV row count mismatch for {batch.path}: "
|
|
f"manifest={batch.rows}, actual={actual}"
|
|
)
|
|
|
|
duplicate_ids = connection.execute(
|
|
"""
|
|
SELECT count(*)
|
|
FROM (
|
|
SELECT internal_event_id
|
|
FROM stg_events
|
|
GROUP BY internal_event_id
|
|
HAVING count(*) > 1
|
|
)
|
|
"""
|
|
).fetchone()[0]
|
|
if duplicate_ids:
|
|
raise MartBuildError(
|
|
f"Input contains {duplicate_ids} duplicated internal event IDs"
|
|
)
|
|
|
|
inconsistent_buckets = connection.execute(
|
|
"""
|
|
SELECT count(*)
|
|
FROM (
|
|
SELECT lower(vehicle_token) AS vehicle_token
|
|
FROM stg_events
|
|
WHERE regexp_full_match(lower(vehicle_token), '^[0-9a-f]{64}$')
|
|
AND try_cast(vehicle_bucket AS INTEGER) BETWEEN 0 AND 99
|
|
GROUP BY lower(vehicle_token)
|
|
HAVING count(DISTINCT try_cast(vehicle_bucket AS INTEGER)) > 1
|
|
)
|
|
"""
|
|
).fetchone()[0]
|
|
if inconsistent_buckets:
|
|
raise MartBuildError(
|
|
f"Input contains {inconsistent_buckets} vehicle tokens with "
|
|
"inconsistent audit buckets"
|
|
)
|
|
|
|
sql_hashes: Dict[str, str] = {}
|
|
for name in SQL_FILES:
|
|
sql = _load_sql(name)
|
|
sql_hashes[name] = hashlib.sha256(sql.encode("utf-8")).hexdigest()
|
|
connection.execute(sql)
|
|
|
|
violations = connection.execute(
|
|
"""
|
|
SELECT check_name, violation_count
|
|
FROM mart_validation
|
|
WHERE violation_count <> 0
|
|
ORDER BY check_name
|
|
"""
|
|
).fetchall()
|
|
if violations:
|
|
details = ", ".join(f"{name}={count}" for name, count in violations)
|
|
raise MartBuildError(f"Mart invariant failure: {details}")
|
|
|
|
counts = connection.execute(
|
|
"""
|
|
SELECT
|
|
(SELECT count(*) FROM stg_events),
|
|
(SELECT count(*) FROM clean_events),
|
|
(SELECT count(*) FROM inspection_episodes),
|
|
(SELECT count(*) FROM feature_mart),
|
|
(SELECT count(*) FROM feature_mart
|
|
WHERE eligible_returning_target)
|
|
"""
|
|
).fetchone()
|
|
input_rows, clean_events, episodes, mart_rows, eligible_rows = map(
|
|
int, counts
|
|
)
|
|
|
|
columns = _output_columns(connection)
|
|
connection.execute("ANALYZE")
|
|
connection.execute(
|
|
"""
|
|
COPY (
|
|
SELECT *
|
|
FROM feature_mart
|
|
ORDER BY vehicle_token, episode_number
|
|
) TO ? (FORMAT PARQUET, COMPRESSION ZSTD)
|
|
""",
|
|
[str(partial_output)],
|
|
)
|
|
|
|
build_id = hashlib.sha256(
|
|
json.dumps(
|
|
{
|
|
"inputs": [batch.compressed_sha256 for batch in batches],
|
|
"sql": sql_hashes,
|
|
"gap": config.episode_gap_days,
|
|
},
|
|
sort_keys=True,
|
|
).encode("utf-8")
|
|
).hexdigest()[:16]
|
|
output_sha256 = file_sha256(partial_output)
|
|
build_manifest: Dict[str, Any] = {
|
|
"build_kind": "private_leakage_safe_inspection_feature_mart",
|
|
"build_id": build_id,
|
|
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
|
|
"classification": "private_pseudonymized_analytical_mart",
|
|
"source_data_kind": source_data_kind,
|
|
"population_estimate_allowed": population_estimate_allowed,
|
|
"episode_gap_days": config.episode_gap_days,
|
|
"vehicle_quality_contract": (
|
|
"eligibility uses only prior events: at most 50 prior events "
|
|
"and at most 4 prior events on any day"
|
|
),
|
|
"outcome_label_source_contract": {
|
|
"allowed_values": ["overall_result", "utah_obd_proxy"],
|
|
"null_allowed_only_for_unlabeled_outcomes": True,
|
|
"utah_obd_proxy_source_era": "utah",
|
|
"feature_role": "audit_only_not_a_predictor",
|
|
},
|
|
"vin_audit_contract": "vehicle_bucket < 10",
|
|
"inputs": [
|
|
{
|
|
"file": str(batch.path),
|
|
"manifest": str(batch.manifest_path),
|
|
"export_kind": batch.export_kind,
|
|
"query_version": batch.query_version,
|
|
"query_sha256": batch.query_sha256,
|
|
"source_start_inclusive": batch.start.isoformat(),
|
|
"source_end_exclusive": batch.end.isoformat(),
|
|
"rows": batch.rows,
|
|
"compressed_file_sha256": batch.compressed_sha256,
|
|
}
|
|
for batch in batches
|
|
],
|
|
"vehicle_token_key_version": batches[0].key_version,
|
|
"vehicle_token_key_fingerprint": batches[0].key_fingerprint,
|
|
"transform_sql_sha256": sql_hashes,
|
|
"row_counts": {
|
|
"input": input_rows,
|
|
"clean_events": clean_events,
|
|
"episodes": episodes,
|
|
"feature_mart": mart_rows,
|
|
"eligible_returning_targets": eligible_rows,
|
|
},
|
|
"columns": columns,
|
|
"parquet_sha256": output_sha256,
|
|
"parquet_bytes": partial_output.stat().st_size,
|
|
}
|
|
partial_manifest.write_text(
|
|
json.dumps(build_manifest, indent=2) + "\n", encoding="utf-8"
|
|
)
|
|
|
|
connection.execute("CHECKPOINT")
|
|
connection.close()
|
|
connection = None
|
|
|
|
os.replace(partial_database, database_path)
|
|
os.replace(partial_output, output_path)
|
|
os.replace(partial_manifest, manifest_path)
|
|
|
|
return BuildSummary(
|
|
database_path=database_path,
|
|
output_path=output_path,
|
|
manifest_path=manifest_path,
|
|
input_rows=input_rows,
|
|
clean_events=clean_events,
|
|
episodes=episodes,
|
|
mart_rows=mart_rows,
|
|
eligible_rows=eligible_rows,
|
|
source_data_kind=source_data_kind,
|
|
population_estimate_allowed=population_estimate_allowed,
|
|
)
|
|
except Exception:
|
|
if connection is not None:
|
|
connection.close()
|
|
_remove_partial(partial_paths)
|
|
raise
|
|
|
|
|
|
def parse_args(argv: Optional[Sequence[str]] = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--input",
|
|
"--batch-dir",
|
|
dest="input_path",
|
|
type=Path,
|
|
default=DEFAULT_INPUT,
|
|
help="A bounded-batch directory or one development history CSV.gz.",
|
|
)
|
|
parser.add_argument("--database", type=Path, default=DEFAULT_DATABASE)
|
|
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
|
parser.add_argument("--episode-gap-days", type=int, default=30)
|
|
parser.add_argument("--memory-limit", default="4GB")
|
|
parser.add_argument(
|
|
"--threads", type=int, default=max(1, min(4, os.cpu_count() or 1))
|
|
)
|
|
parser.add_argument("--allow-gaps", action="store_true")
|
|
parser.add_argument("--allow-external-output", action="store_true")
|
|
parser.add_argument("--overwrite", action="store_true")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: Optional[Sequence[str]] = None) -> int:
|
|
args = parse_args(argv)
|
|
config = BuildConfig(
|
|
input_path=args.input_path,
|
|
database_path=args.database,
|
|
output_path=args.output,
|
|
episode_gap_days=args.episode_gap_days,
|
|
memory_limit=args.memory_limit,
|
|
threads=args.threads,
|
|
allow_gaps=args.allow_gaps,
|
|
allow_external_output=args.allow_external_output,
|
|
overwrite=args.overwrite,
|
|
)
|
|
try:
|
|
summary = build_feature_mart(config)
|
|
except (MartBuildError, duckdb.Error, OSError) as exc:
|
|
print(f"Feature-mart build failed: {exc}", file=sys.stderr)
|
|
return 1
|
|
|
|
print(
|
|
f"Built {summary.mart_rows:,} episode rows "
|
|
f"({summary.eligible_rows:,} eligible returning targets)"
|
|
)
|
|
print(f"Private DuckDB warehouse: {summary.database_path}")
|
|
print(f"Private Parquet mart: {summary.output_path}")
|
|
print(f"Build manifest: {summary.manifest_path}")
|
|
if not summary.population_estimate_allowed:
|
|
print("Development sample: population estimates are prohibited.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|