"""Read-only extraction of the Auto-Ka SQLite snapshot into Parquet (data/interim).

Never modifies the snapshot. Restricted free-text/media columns are not extracted
(see reports/data_access_audit.md §9). Writes a JSON manifest with row counts,
time bounds and the SHA-256 of the source file.
"""
from __future__ import annotations

import hashlib
import json
import sqlite3
import time
from pathlib import Path

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq

from . import config as C

VEHICLE_COLUMNS = [
    "uid", "source", "external_id", "kind", "title", "make", "model", "trim", "year",
    "price", "price_label", "mileage_km", "mileage_label", "transmission", "fuel",
    "drivetrain", "body_type", "exterior_color", "interior_color", "engine", "doors",
    "seats", "vin", "stock_number", "dealer_name", "city", "region", "lat", "lng",
    "details", "content_hash", "first_seen", "last_seen", "updated_at", "miss_count",
    "active", "dup_of", "dup_sources",
]


def sha256_of(path: Path, chunk: int = 1 << 24) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        while True:
            b = f.read(chunk)
            if not b:
                break
            h.update(b)
    return h.hexdigest()


def _dump(con: sqlite3.Connection, sql: str, out: Path, chunksize: int = 100_000) -> int:
    n = 0
    writer = None
    for chunk in pd.read_sql_query(sql, con, chunksize=chunksize):
        table = pa.Table.from_pandas(chunk, preserve_index=False)
        if writer is None:
            writer = pq.ParquetWriter(out, table.schema, compression="zstd")
        else:
            table = table.cast(writer.schema)
        writer.write_table(table)
        n += len(chunk)
    if writer is not None:
        writer.close()
    return n


def run(snapshot: Path = C.SNAPSHOT, outdir: Path = C.INTERIM, with_hash: bool = True) -> dict:
    outdir.mkdir(parents=True, exist_ok=True)
    uri = f"file:{snapshot}?mode=ro"
    con = sqlite3.connect(uri, uri=True)
    t0 = time.time()
    manifest: dict = {"snapshot": str(snapshot), "created_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
    if with_hash:
        manifest["sha256"] = sha256_of(snapshot)
    cols = ", ".join(VEHICLE_COLUMNS)
    manifest["rows"] = {
        "vehicles": _dump(con, f"SELECT {cols} FROM vehicles", outdir / "vehicles.parquet"),
        "price_log": _dump(con, "SELECT uid, ts, price FROM price_log ORDER BY uid, ts", outdir / "price_log.parquet"),
        "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"),
        "dealers": _dump(con, "SELECT source, name, city, lat, lng, fetched_at FROM dealers", outdir / "dealers.parquet"),
        "vin_decode": _dump(con, "SELECT vin, data, fetched_at FROM vin_decode", outdir / "vin_decode.parquet"),
    }
    r = con.execute("SELECT MIN(ts), MAX(ts) FROM sync_log").fetchone()
    manifest["sync_ts_min"], manifest["sync_ts_max"] = r
    r = con.execute("SELECT MIN(first_seen), MAX(last_seen) FROM vehicles").fetchone()
    manifest["first_seen_min"], manifest["last_seen_max"] = r
    manifest["seconds"] = round(time.time() - t0, 1)
    con.close()
    (outdir / "manifest.json").write_text(json.dumps(manifest, indent=2))
    return manifest


if __name__ == "__main__":
    print(json.dumps(run(), indent=2))