1"""Phase 2.1 — Load the read-only snapshot into typed parquet tables (data/interim).23Outputs4 stores_raw.parquet one row per store in the DB, merged with the discovery registry (stores.json)5 listings_raw.parquet one row per product row in the DB (all statuses), details JSON parsed into flat fields6 sync_log.parquet sync events7 ingest_summary.json row counts, checksum of the snapshot, parse statistics8"""9from __future__ import annotations10import ast, json, re11import numpy as np, pandas as pd12from src.common import ROOT, SETTINGS, P, get_logger, snapshot_conn, sha256_file, ledger, register_output, write_json1314log = get_logger("ingest")15SNAP = ROOT / SETTINGS["snapshot"]["path"]161718def _loads(s):19 if s is None or s == "":20 return None21 try:22 return json.loads(s)23 except Exception:24 try:25 return ast.literal_eval(s)26 except Exception:27 return None282930def parse_details(s: str | None) -> dict:31 d = _loads(s)32 out = {"n_variants": np.nan, "has_sku": False, "sku": None, "grams": np.nan, "n_options": np.nan,33 "variant_prices_min": np.nan, "variant_prices_max": np.nan, "n_variant_prices": np.nan, "on_sale_flag": np.nan,34 "merchant_categories": None, "published_at": None, "created_at": None}35 if not isinstance(d, dict):36 return out37 vs = d.get("variants")38 if isinstance(vs, list) and vs:39 out["n_variants"] = len(vs)40 prices = [v.get("price") for v in vs if isinstance(v, dict) and isinstance(v.get("price"), (int, float))]41 if prices:42 out["variant_prices_min"], out["variant_prices_max"], out["n_variant_prices"] = min(prices), max(prices), len(prices)43 skus = [str(v.get("sku")).strip() for v in vs if isinstance(v, dict) and v.get("sku") not in (None, "", "None")]44 if skus:45 out["has_sku"], out["sku"] = True, skus[0][:64]46 grams = [v.get("grams") for v in vs if isinstance(v, dict) and isinstance(v.get("grams"), (int, float)) and v.get("grams") > 0]47 if grams:48 out["grams"] = float(grams[0])49 elif isinstance(d.get("variations"), int):50 out["n_variants"] = d["variations"]51 if d.get("sku") not in (None, "", "None"):52 out["has_sku"], out["sku"] = True, str(d["sku"])[:64]53 if isinstance(d.get("weight"), str):54 try:55 w = float(d["weight"]);56 if w > 0: out["grams"] = w * 1000 if w < 50 else w # WooCommerce weight is usually kg57 except ValueError:58 pass59 opts = d.get("options") or d.get("attributes")60 if isinstance(opts, list):61 out["n_options"] = len(opts)62 if "on_sale" in d:63 out["on_sale_flag"] = bool(d["on_sale"])64 cats = d.get("categories")65 if isinstance(cats, list) and cats:66 out["merchant_categories"] = " | ".join(str(c) for c in cats[:6])[:300]67 out["published_at"], out["created_at"] = d.get("published_at"), d.get("created_at")68 return out697071def main():72 sha = sha256_file(SNAP)73 assert sha == SETTINGS["snapshot"]["sha256"], f"snapshot checksum mismatch: {sha}"74 con = snapshot_conn()7576 # ---- stores -------------------------------------------------------------77 st = pd.read_sql("select * from stores", con)78 reg = json.load(open(ROOT / SETTINGS["snapshot"]["stores_registry"]))["stores"]79 rg = pd.DataFrame(reg)80 rg["geo_d"] = rg["geo"].map(_loads)81 rg["reg_lat"] = rg["geo_d"].map(lambda g: g.get("lat") if isinstance(g, dict) else np.nan)82 rg["reg_lng"] = rg["geo_d"].map(lambda g: g.get("lng") if isinstance(g, dict) else np.nan)83 rg["reg_geo_city"] = rg["geo_d"].map(lambda g: g.get("city") if isinstance(g, dict) else None)84 rg["reg_postal_prefix"] = rg["postal_prefix"].replace({"None": None, "": None})85 rg["reg_region"] = rg["region"].replace({"None": None, "": None})86 rg["reg_status"] = rg["status"]87 rg["reg_verification_date"] = rg["verification_date"]88 rg["reg_phone"] = rg["phone"].replace({"None": None, "": None})89 rg["reg_discovery_sources"] = rg["discovery_sources"].map(lambda s: json.dumps(_loads(s) or []))90 rg["reg_store_kind"] = rg.get("store_kind")91 keep = ["id", "reg_lat", "reg_lng", "reg_geo_city", "reg_postal_prefix", "reg_region", "reg_status", "reg_verification_date",92 "reg_phone", "reg_discovery_sources", "reg_store_kind"]93 st = st.merge(rg[keep], on="id", how="left")94 for c in ("categories", "discovery_sources", "certifications", "socials"):95 st[c + "_list"] = st[c].map(lambda s: json.dumps(_loads(s) or []))96 st["n_certifications"] = st["certifications"].map(lambda s: len(_loads(s) or []) if _loads(s) else 0)97 st["cert_labels"] = st["certifications"].map(lambda s: " | ".join(sorted({(c.get("label") or "") for c in (_loads(s) or []) if isinstance(c, dict)})) if _loads(s) else "")98 st["last_sync_utc"] = pd.to_datetime(st["last_sync"], unit="s", utc=True)99 st = st.drop(columns=["categories", "discovery_sources", "certifications", "socials", "description_meta", "shipping_info", "email", "cover_url", "logo_url"])100 st.to_parquet(P["interim"] / "stores_raw.parquet", index=False)101 log.info("stores_raw %s rows (%s enabled with products)", len(st), int(((st.enabled == 1) & (st.product_count > 0)).sum()))102103 # ---- products -----------------------------------------------------------104 cols = ("uid, store_id, external_id, url, title, substr(description,1,600) as description_head, length(description) as description_len, "105 "price, price_max, compare_at_price, currency, images, category, product_type, tags, vendor, available, content_hash, "106 "first_seen, last_seen, active, miss_count, details, listing_status, price_on_request")107 pr = pd.read_sql(f"select {cols} from products", con)108 log.info("products %s rows", len(pr))109 det = pd.DataFrame([parse_details(s) for s in pr["details"].values])110 pr = pd.concat([pr.drop(columns=["details"]), det], axis=1)111 pr["n_images"] = pr["images"].map(lambda s: len(_loads(s) or []) if s else 0)112 pr = pr.drop(columns=["images"])113 pr["tags_list"] = pr["tags"].map(lambda s: json.dumps(_loads(s) or []) if s else "[]")114 pr["n_tags"] = pr["tags_list"].map(lambda s: len(json.loads(s)))115 pr = pr.drop(columns=["tags"])116 pr["first_seen_utc"] = pd.to_datetime(pr["first_seen"], unit="s", utc=True)117 pr["last_seen_utc"] = pd.to_datetime(pr["last_seen"], unit="s", utc=True)118 pr.to_parquet(P["interim"] / "listings_raw.parquet", index=False)119120 # ---- sync log -------------------------------------------------------------121 sl = pd.read_sql("select ts, store_id, found, added, updated, removed, status from sync_log", con)122 sl["ts_utc"] = pd.to_datetime(sl["ts"], unit="s", utc=True)123 sl.to_parquet(P["interim"] / "sync_log.parquet", index=False)124125 summ = {"snapshot_sha256": sha, "stores": len(st), "stores_live": int(((st.enabled == 1) & (st.product_count > 0)).sum()),126 "products": len(pr), "products_active_published": int(((pr.active == 1) & (pr.listing_status == "published")).sum()),127 "sync_log": len(sl), "details_with_variants": int(pr.n_variants.notna().sum()), "with_sku": int(pr.has_sku.sum()),128 "with_grams": int(pr.grams.notna().sum()), "registry_rows": len(rg),129 "registry_adds_geo_to_db_stores": int((st.lat.isna() & st.reg_lat.notna()).sum()),130 "registry_adds_postal_to_db_stores": int(((st.postal_prefix.isna() | (st.postal_prefix == "")) & st.reg_postal_prefix.notna()).sum())}131 write_json(summ, P["interim"] / "ingest_summary.json", "ingest")132 for f in ("stores_raw.parquet", "listings_raw.parquet", "sync_log.parquet"):133 register_output(P["interim"] / f, "ingest")134 ledger("ingest.load_snapshot", **summ)135 log.info(json.dumps(summ))136137138if __name__ == "__main__":139 main()140