1"""Phase 1.1 — Load the authorised Food-Ka SQLite snapshot (read-only) into Parquet tables.23Nothing is modified in the snapshot. Output: data/interim/raw_{products,price_log,sync_log,product_links,off_cache}.parquet4plus sources.parquet from the JSON registry.5"""6from __future__ import annotations7import json, sqlite38import pandas as pd9from src.common import ROOT, SETTINGS, path, verify_snapshot, snapshot_db, get_logger, ledger, write_manifest1011log = get_logger("ingest")1213def main() -> None:14 digest = verify_snapshot()15 log.info("snapshot verified sha256=%s", digest[:16])16 con = sqlite3.connect(f"file:{snapshot_db()}?mode=ro", uri=True)17 out = path("interim"); files = []18 for t in ["products", "price_log", "sync_log", "product_links", "off_cache"]:19 df = pd.read_sql_query(f"SELECT * FROM {t}", con)20 f = out / f"raw_{t}.parquet"; df.to_parquet(f, index=False); files.append(f)21 log.info("%s: %d rows x %d cols", t, *df.shape)22 reg = json.loads((ROOT / SETTINGS["snapshot"]["sources_registry"]).read_text())["sources"]23 src = pd.DataFrame(reg); f = out / "sources.parquet"; src.to_parquet(f, index=False); files.append(f)24 log.info("sources registry: %d entries (%s)", len(src), src["status"].value_counts().to_dict())25 write_manifest("ingest", files, {"snapshot_sha256": digest})26 ledger("1", "load_snapshot", f"{len(files)} parquet files written to data/interim")2728if __name__ == "__main__":29 main()30