1"""Read-only extraction of the Auto-Ka SQLite snapshot into Parquet (data/interim).23Never modifies the snapshot. Restricted free-text/media columns are not extracted4(see reports/data_access_audit.md §9). Writes a JSON manifest with row counts,5time bounds and the SHA-256 of the source file.6"""7from __future__ import annotations89import hashlib10import json11import sqlite312import time13from pathlib import Path1415import pandas as pd16import pyarrow as pa17import pyarrow.parquet as pq1819from . import config as C2021VEHICLE_COLUMNS = [22 "uid", "source", "external_id", "kind", "title", "make", "model", "trim", "year",23 "price", "price_label", "mileage_km", "mileage_label", "transmission", "fuel",24 "drivetrain", "body_type", "exterior_color", "interior_color", "engine", "doors",25 "seats", "vin", "stock_number", "dealer_name", "city", "region", "lat", "lng",26 "details", "content_hash", "first_seen", "last_seen", "updated_at", "miss_count",27 "active", "dup_of", "dup_sources",28]293031def sha256_of(path: Path, chunk: int = 1 << 24) -> str:32 h = hashlib.sha256()33 with open(path, "rb") as f:34 while True:35 b = f.read(chunk)36 if not b:37 break38 h.update(b)39 return h.hexdigest()404142def _dump(con: sqlite3.Connection, sql: str, out: Path, chunksize: int = 100_000) -> int:43 n = 044 writer = None45 for chunk in pd.read_sql_query(sql, con, chunksize=chunksize):46 table = pa.Table.from_pandas(chunk, preserve_index=False)47 if writer is None:48 writer = pq.ParquetWriter(out, table.schema, compression="zstd")49 else:50 table = table.cast(writer.schema)51 writer.write_table(table)52 n += len(chunk)53 if writer is not None:54 writer.close()55 return n565758def run(snapshot: Path = C.SNAPSHOT, outdir: Path = C.INTERIM, with_hash: bool = True) -> dict:59 outdir.mkdir(parents=True, exist_ok=True)60 uri = f"file:{snapshot}?mode=ro"61 con = sqlite3.connect(uri, uri=True)62 t0 = time.time()63 manifest: dict = {"snapshot": str(snapshot), "created_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}64 if with_hash:65 manifest["sha256"] = sha256_of(snapshot)66 cols = ", ".join(VEHICLE_COLUMNS)67 manifest["rows"] = {68 "vehicles": _dump(con, f"SELECT {cols} FROM vehicles", outdir / "vehicles.parquet"),69 "price_log": _dump(con, "SELECT uid, ts, price FROM price_log ORDER BY uid, ts", outdir / "price_log.parquet"),70 "sync_log": _dump(con, "SELECT id, source, ts, found, added, updated, removed, ok, message, stats FROM sync_log ORDER BY ts", outdir / "sync_log.parquet"),71 "dealers": _dump(con, "SELECT source, name, city, lat, lng, fetched_at FROM dealers", outdir / "dealers.parquet"),72 "vin_decode": _dump(con, "SELECT vin, data, fetched_at FROM vin_decode", outdir / "vin_decode.parquet"),73 }74 r = con.execute("SELECT MIN(ts), MAX(ts) FROM sync_log").fetchone()75 manifest["sync_ts_min"], manifest["sync_ts_max"] = r76 r = con.execute("SELECT MIN(first_seen), MAX(last_seen) FROM vehicles").fetchone()77 manifest["first_seen_min"], manifest["last_seen_max"] = r78 manifest["seconds"] = round(time.time() - t0, 1)79 con.close()80 (outdir / "manifest.json").write_text(json.dumps(manifest, indent=2))81 return manifest828384if __name__ == "__main__":85 print(json.dumps(run(), indent=2))86