"""Phase 1.3 — reconstruct the product × day (and × week) price panel from the event log.

Data-generating process (Food-Ka db.py, read-only copy): price_log holds one row at a product's first observation and
one row at every change of the displayed price (NULL = no displayed price). Between two rows the product kept the
recorded price and was re-observed at every connector pass (median 7.2 h). A product's observation span is
[first_seen, last_seen]; after two consecutive misses it is deactivated (active=0) — gaps inside the span are not
logged (limitation documented in reports/phase1_data_audit.md).

Outputs (data/processed):
  offers_daily.parquet  — uid × local date: duration-weighted price statistics
  offers_weekly.parquet — uid × week (Thu→Wed 'flyer week'): same statistics
  products.parquet      — canonical product table with re-parsed sizes, unit prices, flags
  snapshots.parquet     — connector passes (from sync_log)
"""
from __future__ import annotations
import json
import duckdb, pandas as pd, numpy as np
from src.common import path, get_logger, ledger, write_manifest, TZ, SETTINGS
from src.cleaning.units import parse_size, parse_price_label, unit_price

log = get_logger("panel")

def canonical_products(prod: pd.DataFrame) -> pd.DataFrame:
    out = prod.copy()
    sizes = [parse_size(sl, nm) for sl, nm in zip(out["size_label"], out["name"])]
    labels = [parse_price_label(pl) for pl in out["price_label"]]
    out["size_qty"] = [s.qty if s else np.nan for s in sizes]
    out["size_base"] = [s.base if s else None for s in sizes]
    out["size_approx"] = [bool(s.approx) if s else False for s in sizes]
    out["size_source"] = [s.source if s else None for s in sizes]
    out["size_pack"] = [s.pack if s else np.nan for s in sizes]
    out["per_weight_label"] = [l.per_weight_unit for l in labels]
    out["multibuy_n"] = [l.multibuy_n for l in labels]
    up = [unit_price(p, s, l) for p, s, l in zip(out["price"], sizes, labels)]
    out["unit_price_rp"] = [u[0] if u else np.nan for u in up]
    out["unit_base_rp"] = [u[1] if u else None for u in up]
    det = out["details"].apply(lambda s: json.loads(s) if isinstance(s, str) and s else {})
    out["flyer_id"] = det.apply(lambda d: d.get("flyer_id"))
    out["valid_from"] = det.apply(lambda d: d.get("valid_from"))
    out["valid_to"] = det.apply(lambda d: d.get("valid_to"))
    out["fine_print"] = det.apply(lambda d: d.get("fine_print"))
    out["off_barcode"] = det.apply(lambda d: (d.get("off") or {}).get("code") if isinstance(d.get("off"), dict) else None)
    fp = out["fine_print"].fillna("").str.lower()
    out["flag_weighted"] = out["per_weight_label"].notna() | out["size_approx"]
    out["flag_multibuy"] = out["multibuy_n"].notna()
    out["flag_loyalty"] = fp.str.contains("pc optimum|moi|scène|scene|points|membre|member", regex=True)
    out["flag_min_qty"] = fp.str.contains("limite|limit|minimum|achat de|à l'achat", regex=True)
    out["flag_deposit_tax"] = fp.str.contains(r"\+\s*tx|taxes|consigne|dépôt|deposit", regex=True)
    out["flag_online_only"] = out["source"].isin(["epipresto", "bocoboco", "aliments_merci", "maturin", "costco"])
    out["flag_price_missing"] = out["price"].isna()
    out["flag_unit_missing"] = out["unit_price_rp"].isna()
    out["flag_reg_inconsistent"] = out["regular_price"].notna() & (out["regular_price"] <= out["price"])
    out["flag_unrealistic"] = out["price"].notna() & ((out["price"] < 0.05) | (out["price"] > 1500) |
                              (out["unit_price_rp"].notna() & (out["unit_base_rp"] == "100 g") & (out["unit_price_rp"] > 50)))
    out["gtin_valid"] = out["external_id"].apply(_gs1_ok)
    out["first_seen_local"] = pd.to_datetime(out["first_seen"], unit="s", utc=True).dt.tz_convert(TZ)
    out["last_seen_local"] = pd.to_datetime(out["last_seen"], unit="s", utc=True).dt.tz_convert(TZ)
    out["updated_local"] = pd.to_datetime(out["updated_at"], unit="s", utc=True).dt.tz_convert(TZ)
    out["span_days"] = (out["last_seen"] - out["first_seen"]) / 86400.0
    return out.drop(columns=["details", "images", "description"])

def _gs1_ok(code: str) -> bool:
    if not isinstance(code, str) or not code.isdigit() or len(code) not in (8, 12, 13, 14):
        return False
    d = [int(c) for c in code]; body, chk = d[:-1], d[-1]
    s = sum(x * (3 if (len(body) - i) % 2 == 1 else 1) for i, x in enumerate(body))
    return (10 - s % 10) % 10 == chk

def build(con: duckdb.DuckDBPyConnection) -> None:
    con.execute(f"SET TimeZone='{TZ}'")
    con.execute("""
    CREATE OR REPLACE TABLE ev AS
    SELECT pl.uid, p.source, to_timestamp(pl.ts) AS ts, pl.price,
           to_timestamp(p.first_seen) AS first_seen, to_timestamp(p.last_seen) AS last_seen
    FROM raw_price_log pl JOIN raw_products p USING (uid)""")
    # segments: a price holds from its event until the next event, or until last_seen
    con.execute("""
    CREATE OR REPLACE TABLE seg AS
    SELECT uid, source, price, ts AS seg_start,
           COALESCE(LEAD(ts) OVER (PARTITION BY uid ORDER BY ts), last_seen) AS seg_end
    FROM ev""")
    con.execute("DELETE FROM seg WHERE seg_end <= seg_start")
    # expand to local days
    con.execute("""
    CREATE OR REPLACE TABLE segday AS
    WITH s AS (
      SELECT *, CAST(seg_start AT TIME ZONE 'UTC' AS TIMESTAMP) AS a, CAST(seg_end AT TIME ZONE 'UTC' AS TIMESTAMP) AS b FROM seg
    ), s2 AS (
      SELECT uid, source, price, seg_start, seg_end,
             (seg_start AT TIME ZONE '%(tz)s')::DATE AS d0, (seg_end AT TIME ZONE '%(tz)s')::DATE AS d1 FROM seg
    ), g AS (
      SELECT uid, source, price, seg_start, seg_end, UNNEST(generate_series(d0, d1, INTERVAL 1 DAY))::DATE AS day FROM s2
    )
    SELECT uid, source, price, day,
           GREATEST(seg_start, (day::TIMESTAMP) AT TIME ZONE '%(tz)s') AS a,
           LEAST(seg_end, ((day + INTERVAL 1 DAY)::TIMESTAMP) AT TIME ZONE '%(tz)s') AS b
    FROM g""" % {"tz": TZ})
    con.execute("DELETE FROM segday WHERE b <= a")
    con.execute("ALTER TABLE segday ADD COLUMN hours DOUBLE")
    con.execute("UPDATE segday SET hours = epoch(b - a) / 3600.0")
    _aggregate(con, "segday", "day", "offers_daily")
    # flyer week: Thursday -> Wednesday, keyed by the Thursday
    con.execute("""CREATE OR REPLACE TABLE segweek AS
        SELECT *, day - CAST((dayofweek(day) + 3) % 7 AS INTEGER) AS week FROM segday""")  # dayofweek: Sun=0 ... Thu=4 -> offset 0
    _aggregate(con, "segweek", "week", "offers_weekly")

def _aggregate(con, table: str, key: str, out: str) -> None:
    con.execute(f"""
    CREATE OR REPLACE TABLE {out} AS
    WITH byprice AS (
      SELECT uid, source, {key}, price, SUM(hours) AS h, MIN(a) AS first_a
      FROM {table} GROUP BY uid, source, {key}, price
    ), tot AS (
      SELECT uid, {key}, SUM(h) AS hours_obs, SUM(CASE WHEN price IS NULL THEN h ELSE 0 END) AS hours_null,
             COUNT(*) FILTER (WHERE price IS NOT NULL) AS n_distinct
      FROM byprice GROUP BY uid, {key}
    ), ranked AS (
      SELECT *, ROW_NUMBER() OVER (PARTITION BY uid, {key} ORDER BY h DESC, price ASC) AS rk
      FROM byprice WHERE price IS NOT NULL
    ), modep AS (SELECT uid, {key}, price AS p_mode, h AS mode_hours FROM ranked WHERE rk = 1),
    stats AS (
      SELECT uid, source, {key}, MIN(price) AS p_min, MAX(price) AS p_max,
             SUM(price * hours) / NULLIF(SUM(hours), 0) AS p_twmean,
             ARG_MAX(price, b) AS p_close, ARG_MIN(price, a) AS p_open,
             COUNT(*) - 1 AS n_changes
      FROM {table} WHERE price IS NOT NULL GROUP BY uid, source, {key}
    ), med AS (
      SELECT uid, {key}, QUANTILE_CONT(price, 0.5) AS p_median_events FROM {table} WHERE price IS NOT NULL GROUP BY uid, {key}
    )
    SELECT s.uid, s.source, s.{key}, s.p_open, s.p_close, s.p_min, s.p_max, s.p_twmean, m.p_mode, m.mode_hours,
           md.p_median_events, t.n_distinct, s.n_changes, t.hours_obs, t.hours_null,
           m.mode_hours / NULLIF(t.hours_obs - t.hours_null, 0) AS mode_share,
           (s.p_max - s.p_min) / NULLIF(m.p_mode, 0) AS rel_range
    FROM stats s JOIN modep m USING (uid, {key}) JOIN tot t USING (uid, {key}) JOIN med md USING (uid, {key})""")
    con.execute(f"COPY {out} TO '{path('processed') / (out + '.parquet')}' (FORMAT PARQUET)")
    n, = con.execute(f"SELECT COUNT(*) FROM {out}").fetchone()
    log.info("%s: %d rows", out, n)

def main() -> None:
    interim = path("interim")
    con = duckdb.connect()
    for t in ["products", "price_log", "sync_log"]:
        con.execute(f"CREATE VIEW raw_{t} AS SELECT * FROM read_parquet('{interim / ('raw_' + t + '.parquet')}')")
    build(con)
    prod = canonical_products(pd.read_parquet(interim / "raw_products.parquet"))
    prod.to_parquet(path("processed") / "products.parquet", index=False)
    log.info("products canonical: %d rows; unit price available for %.1f%% (Food-Ka: %.1f%%)", len(prod),
             100 * prod["unit_price_rp"].notna().mean(), 100 * prod["unit_price"].notna().mean())
    sl = pd.read_parquet(interim / "raw_sync_log.parquet")
    sl["ts_utc"] = pd.to_datetime(sl["ts"], unit="s", utc=True)
    st = sl["stats"].apply(lambda s: json.loads(s) if isinstance(s, str) and s else {})
    sl["null_price_rate"] = st.apply(lambda d: d.get("null_price_rate")); sl["missed"] = st.apply(lambda d: d.get("missed"))
    sl["pipeline_commit"] = SETTINGS["snapshot"]["foodka_commit"][:7]
    sl.drop(columns=["stats"]).to_parquet(path("processed") / "snapshots.parquet", index=False)
    files = [path("processed") / f for f in ["offers_daily.parquet", "offers_weekly.parquet", "products.parquet", "snapshots.parquet"]]
    write_manifest("panel", files)
    ledger("1", "build_panel", "offers_daily/weekly, products, snapshots written")

if __name__ == "__main__":
    main()