"""Phase 2.1 — Load the read-only snapshot into typed parquet tables (data/interim).

Outputs
  stores_raw.parquet    one row per store in the DB, merged with the discovery registry (stores.json)
  listings_raw.parquet  one row per product row in the DB (all statuses), details JSON parsed into flat fields
  sync_log.parquet      sync events
  ingest_summary.json   row counts, checksum of the snapshot, parse statistics
"""
from __future__ import annotations
import ast, json, re
import numpy as np, pandas as pd
from src.common import ROOT, SETTINGS, P, get_logger, snapshot_conn, sha256_file, ledger, register_output, write_json

log = get_logger("ingest")
SNAP = ROOT / SETTINGS["snapshot"]["path"]


def _loads(s):
    if s is None or s == "":
        return None
    try:
        return json.loads(s)
    except Exception:
        try:
            return ast.literal_eval(s)
        except Exception:
            return None


def parse_details(s: str | None) -> dict:
    d = _loads(s)
    out = {"n_variants": np.nan, "has_sku": False, "sku": None, "grams": np.nan, "n_options": np.nan,
           "variant_prices_min": np.nan, "variant_prices_max": np.nan, "n_variant_prices": np.nan, "on_sale_flag": np.nan,
           "merchant_categories": None, "published_at": None, "created_at": None}
    if not isinstance(d, dict):
        return out
    vs = d.get("variants")
    if isinstance(vs, list) and vs:
        out["n_variants"] = len(vs)
        prices = [v.get("price") for v in vs if isinstance(v, dict) and isinstance(v.get("price"), (int, float))]
        if prices:
            out["variant_prices_min"], out["variant_prices_max"], out["n_variant_prices"] = min(prices), max(prices), len(prices)
        skus = [str(v.get("sku")).strip() for v in vs if isinstance(v, dict) and v.get("sku") not in (None, "", "None")]
        if skus:
            out["has_sku"], out["sku"] = True, skus[0][:64]
        grams = [v.get("grams") for v in vs if isinstance(v, dict) and isinstance(v.get("grams"), (int, float)) and v.get("grams") > 0]
        if grams:
            out["grams"] = float(grams[0])
    elif isinstance(d.get("variations"), int):
        out["n_variants"] = d["variations"]
    if d.get("sku") not in (None, "", "None"):
        out["has_sku"], out["sku"] = True, str(d["sku"])[:64]
    if isinstance(d.get("weight"), str):
        try:
            w = float(d["weight"]);
            if w > 0: out["grams"] = w * 1000 if w < 50 else w   # WooCommerce weight is usually kg
        except ValueError:
            pass
    opts = d.get("options") or d.get("attributes")
    if isinstance(opts, list):
        out["n_options"] = len(opts)
    if "on_sale" in d:
        out["on_sale_flag"] = bool(d["on_sale"])
    cats = d.get("categories")
    if isinstance(cats, list) and cats:
        out["merchant_categories"] = " | ".join(str(c) for c in cats[:6])[:300]
    out["published_at"], out["created_at"] = d.get("published_at"), d.get("created_at")
    return out


def main():
    sha = sha256_file(SNAP)
    assert sha == SETTINGS["snapshot"]["sha256"], f"snapshot checksum mismatch: {sha}"
    con = snapshot_conn()

    # ---- stores -------------------------------------------------------------
    st = pd.read_sql("select * from stores", con)
    reg = json.load(open(ROOT / SETTINGS["snapshot"]["stores_registry"]))["stores"]
    rg = pd.DataFrame(reg)
    rg["geo_d"] = rg["geo"].map(_loads)
    rg["reg_lat"] = rg["geo_d"].map(lambda g: g.get("lat") if isinstance(g, dict) else np.nan)
    rg["reg_lng"] = rg["geo_d"].map(lambda g: g.get("lng") if isinstance(g, dict) else np.nan)
    rg["reg_geo_city"] = rg["geo_d"].map(lambda g: g.get("city") if isinstance(g, dict) else None)
    rg["reg_postal_prefix"] = rg["postal_prefix"].replace({"None": None, "": None})
    rg["reg_region"] = rg["region"].replace({"None": None, "": None})
    rg["reg_status"] = rg["status"]
    rg["reg_verification_date"] = rg["verification_date"]
    rg["reg_phone"] = rg["phone"].replace({"None": None, "": None})
    rg["reg_discovery_sources"] = rg["discovery_sources"].map(lambda s: json.dumps(_loads(s) or []))
    rg["reg_store_kind"] = rg.get("store_kind")
    keep = ["id", "reg_lat", "reg_lng", "reg_geo_city", "reg_postal_prefix", "reg_region", "reg_status", "reg_verification_date",
            "reg_phone", "reg_discovery_sources", "reg_store_kind"]
    st = st.merge(rg[keep], on="id", how="left")
    for c in ("categories", "discovery_sources", "certifications", "socials"):
        st[c + "_list"] = st[c].map(lambda s: json.dumps(_loads(s) or []))
    st["n_certifications"] = st["certifications"].map(lambda s: len(_loads(s) or []) if _loads(s) else 0)
    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 "")
    st["last_sync_utc"] = pd.to_datetime(st["last_sync"], unit="s", utc=True)
    st = st.drop(columns=["categories", "discovery_sources", "certifications", "socials", "description_meta", "shipping_info", "email", "cover_url", "logo_url"])
    st.to_parquet(P["interim"] / "stores_raw.parquet", index=False)
    log.info("stores_raw %s rows (%s enabled with products)", len(st), int(((st.enabled == 1) & (st.product_count > 0)).sum()))

    # ---- products -----------------------------------------------------------
    cols = ("uid, store_id, external_id, url, title, substr(description,1,600) as description_head, length(description) as description_len, "
            "price, price_max, compare_at_price, currency, images, category, product_type, tags, vendor, available, content_hash, "
            "first_seen, last_seen, active, miss_count, details, listing_status, price_on_request")
    pr = pd.read_sql(f"select {cols} from products", con)
    log.info("products %s rows", len(pr))
    det = pd.DataFrame([parse_details(s) for s in pr["details"].values])
    pr = pd.concat([pr.drop(columns=["details"]), det], axis=1)
    pr["n_images"] = pr["images"].map(lambda s: len(_loads(s) or []) if s else 0)
    pr = pr.drop(columns=["images"])
    pr["tags_list"] = pr["tags"].map(lambda s: json.dumps(_loads(s) or []) if s else "[]")
    pr["n_tags"] = pr["tags_list"].map(lambda s: len(json.loads(s)))
    pr = pr.drop(columns=["tags"])
    pr["first_seen_utc"] = pd.to_datetime(pr["first_seen"], unit="s", utc=True)
    pr["last_seen_utc"] = pd.to_datetime(pr["last_seen"], unit="s", utc=True)
    pr.to_parquet(P["interim"] / "listings_raw.parquet", index=False)

    # ---- sync log -------------------------------------------------------------
    sl = pd.read_sql("select ts, store_id, found, added, updated, removed, status from sync_log", con)
    sl["ts_utc"] = pd.to_datetime(sl["ts"], unit="s", utc=True)
    sl.to_parquet(P["interim"] / "sync_log.parquet", index=False)

    summ = {"snapshot_sha256": sha, "stores": len(st), "stores_live": int(((st.enabled == 1) & (st.product_count > 0)).sum()),
            "products": len(pr), "products_active_published": int(((pr.active == 1) & (pr.listing_status == "published")).sum()),
            "sync_log": len(sl), "details_with_variants": int(pr.n_variants.notna().sum()), "with_sku": int(pr.has_sku.sum()),
            "with_grams": int(pr.grams.notna().sum()), "registry_rows": len(rg),
            "registry_adds_geo_to_db_stores": int((st.lat.isna() & st.reg_lat.notna()).sum()),
            "registry_adds_postal_to_db_stores": int(((st.postal_prefix.isna() | (st.postal_prefix == "")) & st.reg_postal_prefix.notna()).sum())}
    write_json(summ, P["interim"] / "ingest_summary.json", "ingest")
    for f in ("stores_raw.parquet", "listings_raw.parquet", "sync_log.parquet"):
        register_output(P["interim"] / f, "ingest")
    ledger("ingest.load_snapshot", **summ)
    log.info(json.dumps(summ))


if __name__ == "__main__":
    main()