SPB Git

spb/wp36_uqo

Public
3.3 KB · 86 lines python
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
Files · f9bf466deda3bcf4729bb4dc6e4078483843c761
Menu

Theme