"""SQL-only data access for the promo module.

Data sources:
    erp_promo_weeks       — per (product × ISO-year × ISO-week) promo flag
    promo_calendar_entries — unified promo calendar (currently empty)
    promo_proposals       — planner submissions
    v_sales_weekly_full        — qty_total used for before/during/after math
    dim_products, sku_planning, dim_categories — metadata joins

Contiguous-run detection runs in SQL via the standard "gaps-and-islands"
window-function trick. We compute an absolute week index `wk_abs` per row
(year × 52 + week) and group rows where `wk_abs - ROW_NUMBER()` stays
constant — that's a contiguous run.

The 4-week BEFORE/AFTER windows for performance math are also computed
in SQL by joining to v_sales_weekly_full with a calculated week range. SKUs
whose own promo weeks overlap the BEFORE/AFTER windows are excluded
from the average (matches build_promo_performance.py).
"""
from __future__ import annotations

from typing import Optional

from sqlalchemy import text

from backend.repositories.base import BaseRepository


class PromoRepository(BaseRepository):

    # ------------------------------------------------------------------
    # Promo history — contiguous runs of erp_promo_weeks per SKU
    # ------------------------------------------------------------------

    def get_promo_events(
        self,
        *,
        category: Optional[list[str]] = None,
        tier: Optional[list[str]] = None,
        min_start_yw: Optional[int] = None,
        max_start_yw: Optional[int] = None,
        max_events: int = 5000,
    ) -> list[dict]:
        """Detect contiguous promo periods per SKU using gaps-and-islands.

        Each "island" is a sequence of consecutive ISO weeks all flagged
        is_erp_promo=true. promo_types are concatenated (DISTINCT) across
        the run. ORDER is most-recent first so the History UI shows
        upcoming and recent campaigns at the top.
        """
        where_clauses: list[str] = ["epw.is_erp_promo = true"]
        params: dict = {"limit": max_events}
        if category:
            where_clauses.append("dc.name = ANY(:cats)")
            params["cats"] = category
        if tier:
            where_clauses.append("UPPER(sp.tier) LIKE ANY(:tier_pats)")
            params["tier_pats"] = [f"%{t.upper()}%" for t in tier]
        if min_start_yw is not None:
            params["min_yw"] = min_start_yw
        if max_start_yw is not None:
            params["max_yw"] = max_start_yw

        where_sql = " AND ".join(where_clauses)
        start_yw_filter = ""
        if min_start_yw is not None:
            start_yw_filter += " AND start_yw >= :min_yw"
        if max_start_yw is not None:
            start_yw_filter += " AND start_yw <= :max_yw"

        sql = f"""
            WITH weeks AS (
                SELECT
                    epw.product_id,
                    epw.year, epw.week,
                    (epw.year * 52 + epw.week) AS wk_abs,
                    COALESCE(epw.promo_types, '') AS promo_types,
                    ROW_NUMBER() OVER (PARTITION BY epw.product_id
                                       ORDER BY epw.year, epw.week) AS rn
                FROM erp_promo_weeks epw
                JOIN dim_products p ON p.id = epw.product_id
                LEFT JOIN dim_categories dc ON dc.id = p.category_id
                LEFT JOIN sku_planning sp ON sp.product_id = p.id
                WHERE {where_sql}
            ),
            runs AS (
                SELECT
                    product_id,
                    (wk_abs - rn) AS island_id,
                    MIN(year * 100 + week)              AS start_yw,
                    MAX(year * 100 + week)              AS end_yw,
                    MIN(year) AS start_year, MIN(week) AS start_week,
                    MAX(year) AS end_year,   MAX(week) AS end_week,
                    COUNT(*)                            AS n_weeks,
                    STRING_AGG(DISTINCT NULLIF(promo_types, ''), '; ') AS promo_types
                FROM weeks
                GROUP BY product_id, (wk_abs - rn)
            )
            SELECT
                p.sku,
                p.name,
                dc.name      AS category,
                sp.tier,
                r.start_year, r.start_week,
                r.end_year,   r.end_week,
                r.start_yw, r.end_yw,
                r.n_weeks,
                r.promo_types
            FROM runs r
            JOIN dim_products p ON p.id = r.product_id
            LEFT JOIN dim_categories dc ON dc.id = p.category_id
            LEFT JOIN sku_planning sp   ON sp.product_id = p.id
            WHERE 1 = 1 {start_yw_filter}
            ORDER BY r.start_yw DESC, p.sku
            LIMIT :limit
        """
        rows = self.db.execute(text(sql), params).mappings().all()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # Promo performance — before/during/after math joined to weekly sales
    # ------------------------------------------------------------------

    def get_promo_performance(
        self,
        *,
        category: Optional[list[str]] = None,
        tier: Optional[list[str]] = None,
        max_weeks: int = 13,
        min_start_yw: Optional[int] = None,
        max_events: int = 5000,
    ) -> list[dict]:
        """For each contiguous promo run, compute the same uplift /
        cannibalization metrics as build_promo_performance.py.

        BEFORE window = the 4 ISO weeks immediately before start_week
                         (excluding any week that is also is_erp_promo
                          for the same SKU — separate campaigns nearby
                          would contaminate the baseline).
        AFTER  window = same shape, immediately after end_week.

        n_weeks <= max_weeks filters out year-long evergreen flags that
        aren't really "campaigns" in the strategic sense.
        """
        where_clauses: list[str] = ["epw.is_erp_promo = true"]
        params: dict = {
            "max_weeks": max_weeks,
            "limit":     max_events,
            "window":    4,
        }
        if category:
            where_clauses.append("dc.name = ANY(:cats)")
            params["cats"] = category
        if tier:
            where_clauses.append("UPPER(sp.tier) LIKE ANY(:tier_pats)")
            params["tier_pats"] = [f"%{t.upper()}%" for t in tier]
        if min_start_yw is not None:
            params["min_yw"] = min_start_yw

        where_sql = " AND ".join(where_clauses)

        # Build the runs CTE once, then compute averages by joining
        # v_sales_weekly_full per row with a calculated week range. The
        # promo-overlap exclusion is done with NOT EXISTS against
        # erp_promo_weeks for the same product.
        start_filter = ""
        if min_start_yw is not None:
            start_filter = "AND (start_year * 100 + start_week) >= :min_yw"

        sql = f"""
            WITH weeks AS (
                SELECT
                    epw.product_id,
                    epw.year, epw.week,
                    (epw.year * 52 + epw.week) AS wk_abs,
                    COALESCE(epw.promo_types, '') AS promo_types,
                    ROW_NUMBER() OVER (PARTITION BY epw.product_id
                                       ORDER BY epw.year, epw.week) AS rn
                FROM erp_promo_weeks epw
                JOIN dim_products p ON p.id = epw.product_id
                LEFT JOIN dim_categories dc ON dc.id = p.category_id
                LEFT JOIN sku_planning sp ON sp.product_id = p.id
                WHERE {where_sql}
            ),
            runs AS (
                SELECT
                    product_id,
                    MIN(year * 52 + week) AS start_abs,
                    MAX(year * 52 + week) AS end_abs,
                    MIN(year) AS start_year, MIN(week) AS start_week,
                    MAX(year) AS end_year,   MAX(week) AS end_week,
                    COUNT(*)              AS n_weeks,
                    STRING_AGG(DISTINCT NULLIF(promo_types, ''), '; ') AS promo_types
                FROM weeks
                GROUP BY product_id, (wk_abs - rn)
                HAVING COUNT(*) <= :max_weeks
            ),
            during AS (
                SELECT
                    r.product_id, r.start_abs, r.end_abs,
                    AVG(vsw.qty_total)::float AS qty_during
                FROM runs r
                LEFT JOIN v_sales_weekly_full vsw
                    ON vsw.product_id = r.product_id
                    AND (vsw.year * 52 + vsw.week) BETWEEN r.start_abs AND r.end_abs
                GROUP BY r.product_id, r.start_abs, r.end_abs
            ),
            before_w AS (
                SELECT
                    r.product_id, r.start_abs, r.end_abs,
                    AVG(vsw.qty_total)::float AS qty_before,
                    COUNT(vsw.qty_total)      AS n_before
                FROM runs r
                LEFT JOIN v_sales_weekly_full vsw
                    ON vsw.product_id = r.product_id
                    AND (vsw.year * 52 + vsw.week) BETWEEN r.start_abs - :window AND r.start_abs - 1
                    AND NOT EXISTS (
                        SELECT 1 FROM erp_promo_weeks o
                        WHERE o.product_id = r.product_id
                          AND o.year = vsw.year AND o.week = vsw.week
                          AND o.is_erp_promo = true
                    )
                GROUP BY r.product_id, r.start_abs, r.end_abs
            ),
            after_w AS (
                SELECT
                    r.product_id, r.start_abs, r.end_abs,
                    AVG(vsw.qty_total)::float AS qty_after,
                    COUNT(vsw.qty_total)      AS n_after
                FROM runs r
                LEFT JOIN v_sales_weekly_full vsw
                    ON vsw.product_id = r.product_id
                    AND (vsw.year * 52 + vsw.week) BETWEEN r.end_abs + 1 AND r.end_abs + :window
                    AND NOT EXISTS (
                        SELECT 1 FROM erp_promo_weeks o
                        WHERE o.product_id = r.product_id
                          AND o.year = vsw.year AND o.week = vsw.week
                          AND o.is_erp_promo = true
                    )
                GROUP BY r.product_id, r.start_abs, r.end_abs
            )
            SELECT
                p.sku,
                p.name,
                dc.name AS category,
                sp.tier,
                r.start_year, r.start_week,
                r.end_year,   r.end_week,
                r.n_weeks,
                r.promo_types,
                b.qty_before,
                d.qty_during,
                a.qty_after,
                CASE WHEN b.qty_before > 0 AND d.qty_during IS NOT NULL
                     THEN d.qty_during / b.qty_before ELSE NULL END AS actual_uplift,
                CASE WHEN b.qty_before > 0 AND a.qty_after  IS NOT NULL
                     THEN a.qty_after  / b.qty_before ELSE NULL END AS cannibalization,
                CASE WHEN b.qty_before > 0
                     THEN ( COALESCE(d.qty_during, 0) * r.n_weeks
                          + COALESCE(a.qty_after, 0) * COALESCE(a.n_after, 0) )
                         / ( b.qty_before * (r.n_weeks + COALESCE(a.n_after, :window)) )
                     ELSE NULL END AS net_effect
            FROM runs r
            JOIN dim_products p ON p.id = r.product_id
            LEFT JOIN dim_categories dc ON dc.id = p.category_id
            LEFT JOIN sku_planning sp   ON sp.product_id = p.id
            LEFT JOIN during   d ON d.product_id = r.product_id
                               AND d.start_abs = r.start_abs AND d.end_abs = r.end_abs
            LEFT JOIN before_w b ON b.product_id = r.product_id
                               AND b.start_abs = r.start_abs AND b.end_abs = r.end_abs
            LEFT JOIN after_w  a ON a.product_id = r.product_id
                               AND a.start_abs = r.start_abs AND a.end_abs = r.end_abs
            WHERE 1=1 {start_filter}
            ORDER BY r.start_abs DESC, p.sku
            LIMIT :limit
        """
        rows = self.db.execute(text(sql), params).mappings().all()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # Proposals — list + create
    # ------------------------------------------------------------------

    def get_proposals(
        self,
        *,
        status: Optional[str] = None,
        proposed_by_id: Optional[int] = None,
    ) -> list[dict]:
        """List promo_proposals. JOIN users for display name. JSONB
        `skus` is returned as-is for the service to parse."""
        where_clauses: list[str] = []
        params: dict = {}
        if status:
            where_clauses.append("pp.status = :status")
            params["status"] = status
        if proposed_by_id is not None:
            where_clauses.append("pp.proposed_by_id = :pid")
            params["pid"] = proposed_by_id
        where_sql = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else ""
        sql = f"""
            SELECT
                pp.id, pp.name, pp.source, pp.mechanic, pp.discount_pct::float,
                pp.start_year, pp.start_week, pp.end_year, pp.end_week,
                pp.skus, pp.status, pp.proposed_by_id, pp.channels, pp.log,
                pp.planner_state, pp.created_at,
                COALESCE(NULLIF(u.display_name, ''), u.username) AS proposed_by
            FROM promo_proposals pp
            LEFT JOIN users u ON u.id = pp.proposed_by_id
            {where_sql}
            ORDER BY pp.created_at DESC NULLS LAST, pp.id DESC
        """
        rows = self.db.execute(text(sql), params).mappings().all()
        return [dict(r) for r in rows]

    def create_proposal(self, data: dict) -> dict:
        """Insert a promo_proposals row. `skus` is JSONB — pass via
        psycopg2's automatic JSON adaptation by serialising in Python."""
        import json
        sql = """
            INSERT INTO promo_proposals
                (name, source, mechanic, discount_pct,
                 start_year, start_week, end_year, end_week,
                 skus, status, objective, proposed_by_id, channels, log, planner_state)
            VALUES
                (:name, :source, :mechanic, :discount_pct,
                 :start_year, :start_week, :end_year, :end_week,
                 CAST(:skus AS JSONB), :status, :objective, :proposed_by_id, :channels, :log,
                 CAST(:planner_state AS JSONB))
            RETURNING id, created_at
        """
        ps = data.get("planner_state")
        params = {
            "name":           data.get("name"),
            "source":         data.get("source", "planner"),
            "mechanic":       data.get("mechanic"),
            "discount_pct":   data.get("discount_pct"),
            "start_year":     data.get("start_year"),
            "start_week":     data.get("start_week"),
            "end_year":       data.get("end_year"),
            "end_week":       data.get("end_week"),
            "skus":           json.dumps(data.get("skus") or []),
            "status":         data.get("status", "draft"),
            "objective":      data.get("objective"),
            "proposed_by_id": data.get("proposed_by_id"),
            "channels":       data.get("channels", "retail"),
            "log":            data.get("log"),
            "planner_state":  json.dumps(ps) if ps is not None else None,
        }
        row = self.db.execute(text(sql), params).mappings().first()
        self.db.commit()
        return dict(row) if row else {}

    def update_proposal(self, proposal_id: int, data: dict) -> Optional[dict]:
        """Overwrite an existing proposal's editable content (used when a
        planner reopens a draft / revision and re-saves). Updates the flat
        columns, the skus JSONB and the planner_state working copy, and
        optionally the status. Appends a log line rather than replacing the
        audit trail. Returns {id, created_at} or None if the row is gone."""
        import json
        # Build SET clauses only for keys actually supplied, so a partial
        # update (e.g. status-only re-save) doesn't blank out other fields.
        cols = {
            "name": "name", "source": "source", "mechanic": "mechanic",
            "discount_pct": "discount_pct", "start_year": "start_year",
            "start_week": "start_week", "end_year": "end_year",
            "end_week": "end_week", "status": "status", "channels": "channels",
            "objective": "objective",
        }
        sets: list[str] = []
        params: dict = {"pid": proposal_id}
        for key, col in cols.items():
            if key in data and data[key] is not None:
                sets.append(f"{col} = :{key}")
                params[key] = data[key]
        if "skus" in data and data["skus"] is not None:
            sets.append("skus = CAST(:skus AS JSONB)")
            params["skus"] = json.dumps(data["skus"])
        if "planner_state" in data and data["planner_state"] is not None:
            sets.append("planner_state = CAST(:planner_state AS JSONB)")
            params["planner_state"] = json.dumps(data["planner_state"])
        # Append a log line (don't clobber the existing audit trail).
        log_line = data.get("log_line")
        if log_line:
            sets.append("log = CONCAT_WS(E'\\n', NULLIF(log, ''), :log_line)")
            params["log_line"] = log_line
        if not sets:
            # Nothing to update — just confirm existence.
            row = self.db.execute(
                text("SELECT id, created_at FROM promo_proposals WHERE id = :pid"),
                {"pid": proposal_id},
            ).mappings().first()
            return dict(row) if row else None
        sql = f"""
            UPDATE promo_proposals
               SET {", ".join(sets)}
             WHERE id = :pid
         RETURNING id, created_at
        """
        row = self.db.execute(text(sql), params).mappings().first()
        self.db.commit()
        return dict(row) if row else None

    # ------------------------------------------------------------------
    # Analog forecast — find SKU's own historical promos
    # ------------------------------------------------------------------

    def get_sku_promo_history(self, *, sku: str, max_weeks: int = 13) -> list[dict]:
        """Per-SKU contiguous promo runs with computed uplift, used as
        seed data for the Planner's analog forecast. Calls the same
        gaps-and-islands logic but scoped to one product."""
        sql = """
            WITH p AS (
                SELECT id, sku, name FROM dim_products WHERE sku = :sku
            ),
            weeks AS (
                SELECT
                    epw.year, epw.week,
                    (epw.year * 52 + epw.week) AS wk_abs,
                    COALESCE(epw.promo_types, '') AS promo_types,
                    ROW_NUMBER() OVER (ORDER BY epw.year, epw.week) AS rn
                FROM erp_promo_weeks epw
                JOIN p ON p.id = epw.product_id
                WHERE epw.is_erp_promo = true
            ),
            runs AS (
                SELECT
                    MIN(year * 52 + week) AS start_abs,
                    MAX(year * 52 + week) AS end_abs,
                    MIN(year) AS start_year, MIN(week) AS start_week,
                    MAX(year) AS end_year,   MAX(week) AS end_week,
                    COUNT(*) AS n_weeks,
                    STRING_AGG(DISTINCT NULLIF(promo_types, ''), '; ') AS promo_types
                FROM weeks
                GROUP BY (wk_abs - rn)
                HAVING COUNT(*) <= :max_weeks
            )
            SELECT
                r.start_year, r.start_week, r.end_year, r.end_week,
                r.n_weeks, r.promo_types,
                ( SELECT AVG(vsw.qty_total)::float FROM v_sales_weekly_full vsw, p
                  WHERE vsw.product_id = p.id
                    AND (vsw.year * 52 + vsw.week) BETWEEN r.start_abs AND r.end_abs ) AS qty_during,
                ( SELECT AVG(vsw.qty_total)::float FROM v_sales_weekly_full vsw, p
                  WHERE vsw.product_id = p.id
                    AND (vsw.year * 52 + vsw.week) BETWEEN r.start_abs - 4 AND r.start_abs - 1 ) AS qty_before
            FROM runs r
            ORDER BY r.start_abs DESC
        """
        rows = self.db.execute(text(sql), {"sku": sku, "max_weeks": max_weeks}).mappings().all()
        return [dict(r) for r in rows]

    def get_sku_baseline(self, *, sku: str, weeks: int = 13) -> Optional[float]:
        """Trailing N-week non-promo avg weekly qty for a SKU. Used as
        the 'baseline' value the Planner shows alongside expected uplift."""
        sql = """
            WITH p AS (SELECT id FROM dim_products WHERE sku = :sku),
            max_yw AS (SELECT MAX(year * 100 + week) AS m FROM v_sales_weekly_full)
            SELECT AVG(vsw.qty_total)::float AS baseline
            FROM v_sales_weekly_full vsw
            CROSS JOIN max_yw mx
            JOIN p ON p.id = vsw.product_id
            LEFT JOIN erp_promo_weeks epw
                ON epw.product_id = vsw.product_id
                AND epw.year = vsw.year AND epw.week = vsw.week
            WHERE vsw.year * 100 + vsw.week > mx.m - :weeks
              AND NOT COALESCE(epw.is_erp_promo, false)
        """
        row = self.db.execute(text(sql), {"sku": sku, "weeks": weeks}).mappings().first()
        return float(row["baseline"]) if row and row.get("baseline") is not None else None

    def get_group_baseline(self, *, skus: list[str], weeks: int = 13) -> Optional[dict]:
        """Average of per-SKU trailing non-promo weekly baselines across a set of
        sibling SKUs (a SKU's parent-group). Seeds a baseline for a NEW SKU that
        has no sales of its own = the typical sibling's weekly volume. Returns
        {baseline, n} over siblings that actually sold, or None."""
        if not skus:
            return None
        sql = """
            WITH ids AS (SELECT id FROM dim_products WHERE sku = ANY(:skus)),
            max_yw AS (SELECT MAX(year * 100 + week) AS m FROM v_sales_weekly_full),
            per_sku AS (
                SELECT vsw.product_id, AVG(vsw.qty_total)::float AS bl
                FROM v_sales_weekly_full vsw
                CROSS JOIN max_yw mx
                JOIN ids ON ids.id = vsw.product_id
                LEFT JOIN erp_promo_weeks epw
                    ON epw.product_id = vsw.product_id
                   AND epw.year = vsw.year AND epw.week = vsw.week
                WHERE vsw.year * 100 + vsw.week > mx.m - :weeks
                  AND NOT COALESCE(epw.is_erp_promo, false)
                GROUP BY vsw.product_id
            )
            SELECT AVG(bl)::float AS baseline, COUNT(*) AS n
            FROM per_sku WHERE bl > 0
        """
        row = self.db.execute(text(sql), {"skus": skus, "weeks": weeks}).mappings().first()
        if not row or row.get("baseline") is None or not row.get("n"):
            return None
        return {"baseline": float(row["baseline"]), "n": int(row["n"])}

    def get_group_price_cost(self, *, skus: list[str]) -> Optional[dict]:
        """Average sell price + cost across a set of sibling SKUs — used to seed
        price/cost for a NEW SKU that has none. Zeros are ignored."""
        if not skus:
            return None
        sql = """
            SELECT AVG(NULLIF(ep.avg_sell_price, 0))::float AS price,
                   AVG(NULLIF(ec.cost_price, 0))::float     AS cost
            FROM dim_products p
            LEFT JOIN erp_prices ep ON ep.product_id = p.id
            LEFT JOIN erp_costs  ec ON ec.product_id = p.id
            WHERE p.sku = ANY(:skus)
        """
        row = self.db.execute(text(sql), {"skus": skus}).mappings().first()
        if not row:
            return None
        return {"price": row.get("price"), "cost": row.get("cost")}

    def get_sku_info(self, *, sku: str) -> Optional[dict]:
        """SKU metadata (name + category) for the analog forecast response."""
        sql = """
            SELECT p.sku, p.name, dc.name AS category, sp.tier
            FROM dim_products p
            LEFT JOIN dim_categories dc ON dc.id = p.category_id
            LEFT JOIN sku_planning sp   ON sp.product_id = p.id
            WHERE p.sku = :sku
        """
        row = self.db.execute(text(sql), {"sku": sku}).mappings().first()
        return dict(row) if row else None

    # ------------------------------------------------------------------
    # NC30 (najniža cijena u zadnjih 30 dana) — Croatian price-floor law
    # ------------------------------------------------------------------

    def get_nc30(self, *, sku: str) -> Optional[float]:
        """Lowest price in the past 30 days for a SKU. Promo prices must
        be ≤ NC30 to be legally publishable. Returns None if not set."""
        sql = """
            SELECT n.nc30_price::float AS nc30
            FROM erp_nc30 n
            JOIN dim_products p ON p.id = n.product_id
            WHERE p.sku = :sku AND n.nc30_price > 0
            ORDER BY n.snapshot_date DESC NULLS LAST
            LIMIT 1
        """
        row = self.db.execute(text(sql), {"sku": sku}).mappings().first()
        return float(row["nc30"]) if row and row.get("nc30") is not None else None

    def get_nc30_batch(self, *, skus: list[str]) -> dict[str, float]:
        """Bulk variant of get_nc30 — used by the planner when many SKUs
        are selected. Returns {sku: nc30_price}."""
        if not skus:
            return {}
        sql = """
            SELECT p.sku, n.nc30_price::float AS nc30
            FROM erp_nc30 n
            JOIN dim_products p ON p.id = n.product_id
            WHERE p.sku = ANY(:skus) AND n.nc30_price > 0
        """
        rows = self.db.execute(text(sql), {"skus": skus}).mappings().all()
        return {r["sku"]: float(r["nc30"]) for r in rows}

    # ------------------------------------------------------------------
    # SKU snapshot — comprehensive per-SKU data for the planner
    # ------------------------------------------------------------------

    def get_sku_snapshot(self, *, sku: str) -> Optional[dict]:
        """One-shot fetch of price, cost, RUC, NC30, baseline run-rate,
        on-hand stock, and category metadata. Drives the Planner's
        per-SKU tab. Mirrors the data Streamlit pulls from sku_prices.csv +
        sku_costs.csv + nc30.csv + stock.csv."""
        pid_row = self.db.execute(
            text("SELECT id FROM dim_products WHERE sku = :sku"), {"sku": sku},
        ).mappings().first()
        if not pid_row:
            return None
        pid = int(pid_row["id"])

        # ruc_unit + cost_price use the same realized-first cascade as the
        # demand SKU detail card (see demand_repo.get_sku_pricing). Without
        # this, brand-new NPD SKUs that the ERP hasn't cost-mastered yet
        # show "—" in the snapshot even though we have realized values
        # in erp_transactions and a planned cost in npd_products.
        meta_sql = """
            WITH txn AS (
                SELECT SUM(et.quantity)::float      AS units,
                       SUM(et.ruc_eur)::float       AS ruc_total,
                       SUM(et.purchase_value)::float AS pv_total
                FROM erp_transactions et
                WHERE et.product_id = :pid
                  AND et.transaction_date >= (CURRENT_DATE - INTERVAL '13 weeks')
            )
            SELECT
                p.sku, p.name,
                dc.name                            AS category,
                sp.tier, sp.total_xyz              AS xyz,
                ep.normal_retail_ppp::float        AS price_retail,
                ep.normal_webshop_ppp::float       AS price_webshop,
                ep.avg_sell_price::float           AS avg_sell_price,
                COALESCE(
                    ec.cost_price::float,
                    (txn.pv_total / NULLIF(txn.units, 0))::float,
                    np.cost_price::float
                )                                  AS cost_price,
                COALESCE(
                    (txn.ruc_total / NULLIF(txn.units, 0))::float,
                    ec.ruc::float
                )                                  AS ruc_unit,
                n.nc30_price::float                AS nc30,
                sm.lead_time_weeks::float          AS lead_time_weeks,
                sm.moq::float                      AS moq
            FROM dim_products p
            LEFT JOIN dim_categories dc ON dc.id = p.category_id
            LEFT JOIN sku_planning   sp ON sp.product_id = p.id
            LEFT JOIN erp_prices     ep ON ep.product_id = p.id
            LEFT JOIN erp_costs      ec ON ec.product_id = p.id
            LEFT JOIN erp_nc30        n ON n.product_id  = p.id
            LEFT JOIN supply_master  sm ON sm.product_id = p.id
            LEFT JOIN npd_products   np ON np.sku        = p.sku
            CROSS JOIN txn
            WHERE p.id = :pid
        """
        meta = self.db.execute(text(meta_sql), {"pid": pid}).mappings().first()
        if not meta:
            return None

        baseline_sql = """
            WITH max_yw AS (SELECT MAX(year * 100 + week) AS m FROM v_sales_weekly_full)
            SELECT
                AVG(vsw.qty_total)::float AS avg_weekly,
                COUNT(*)                  AS n_weeks
            FROM v_sales_weekly_full vsw
            CROSS JOIN max_yw mx
            LEFT JOIN erp_promo_weeks epw
                ON epw.product_id = vsw.product_id
                AND epw.year = vsw.year AND epw.week = vsw.week
            WHERE vsw.product_id = :pid
              AND vsw.year * 100 + vsw.week > mx.m - 13
              AND NOT COALESCE(epw.is_erp_promo, false)
              AND vsw.qty_total > 0
        """
        b = self.db.execute(text(baseline_sql), {"pid": pid}).mappings().first() or {}

        stock_row = self.db.execute(
            text("SELECT COALESCE(SUM(stock_qty), 0)::float AS on_hand "
                 "FROM erp_stock_current WHERE product_id = :pid"),
            {"pid": pid},
        ).mappings().first() or {}

        out = dict(meta)
        out["baseline_avg_weekly"] = float(b.get("avg_weekly") or 0)
        out["baseline_n_weeks"]    = int(b.get("n_weeks") or 0)
        out["on_hand"]             = float(stock_row.get("on_hand") or 0)
        return out

    # ------------------------------------------------------------------
    # Parent groups — product-family detection via name stem
    # ------------------------------------------------------------------

    def get_dim_products_for_family_grouping(self) -> list[dict]:
        """Return all active products with name + tier + category. The
        service applies the family_key() heuristic on the result to build
        parent groups. We do family detection in Python (not SQL) because
        the token-stripping rules are complex."""
        sql = """
            SELECT
                p.sku, p.name,
                dc.name AS category,
                sp.tier
            FROM dim_products p
            LEFT JOIN dim_categories dc ON dc.id = p.category_id
            LEFT JOIN sku_planning   sp ON sp.product_id = p.id
            WHERE p.active = true
              AND p.name IS NOT NULL
            ORDER BY p.name
        """
        rows = self.db.execute(text(sql)).mappings().all()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # Marketing history — Magento coupon-driven webshop campaigns
    # ------------------------------------------------------------------

    def get_marketing_campaigns(self) -> list[dict]:
        """Aggregated coupon-orders per campaign. Returns campaign-level
        summary + nested per-SKU and per-coupon rows. Webshop-only;
        cancelled orders excluded by the loader at import time."""
        rows = self.db.execute(text("""
            SELECT 1 FROM webshop_coupon_orders LIMIT 1
        """)).fetchall()
        if not rows:
            return []

        sql = """
            SELECT
                wc.id           AS campaign_id,
                wc.name         AS campaign_name,
                wc.label        AS campaign_label,
                wc.start_date,
                wc.end_date,
                MIN(wco.order_date) AS first_date,
                MAX(wco.order_date) AS last_date,
                COUNT(*)                                          AS n_orders,
                COUNT(DISTINCT wco.coupon_code)                   AS n_coupons,
                COUNT(DISTINCT wco.product_id)                    AS n_skus,
                SUM(wco.quantity)::float                          AS total_units,
                SUM(wco.original_price)::float                    AS total_rev_before,
                SUM(wco.final_price)::float                       AS total_rev_after,
                SUM(wco.discount_amount)::float                   AS total_discount
            FROM webshop_coupon_orders wco
            JOIN webshop_campaigns wc ON wc.id = wco.campaign_id
            GROUP BY wc.id, wc.name, wc.label, wc.start_date, wc.end_date
            ORDER BY MIN(wco.order_date) DESC
        """
        return [dict(r) for r in self.db.execute(text(sql)).mappings().all()]

    def get_marketing_campaign_skus(self, *, campaign_id: int) -> list[dict]:
        sql = """
            SELECT
                p.sku, p.name,
                dc.name AS category,
                sp.tier,
                COUNT(DISTINCT wco.order_id)               AS n_orders,
                SUM(wco.quantity)::float                   AS units,
                SUM(wco.original_price)::float             AS rev_before,
                SUM(wco.final_price)::float                AS rev_after,
                SUM(wco.discount_amount)::float            AS discount,
                CASE WHEN SUM(wco.original_price) > 0
                     THEN (SUM(wco.discount_amount) / SUM(wco.original_price) * 100)::float
                     ELSE 0 END                            AS disc_pct
            FROM webshop_coupon_orders wco
            JOIN dim_products p          ON p.id = wco.product_id
            LEFT JOIN dim_categories dc  ON dc.id = p.category_id
            LEFT JOIN sku_planning  sp   ON sp.product_id = p.id
            WHERE wco.campaign_id = :cid
            GROUP BY p.sku, p.name, dc.name, sp.tier
            ORDER BY units DESC
        """
        rows = self.db.execute(text(sql), {"cid": campaign_id}).mappings().all()
        return [dict(r) for r in rows]

    def get_marketing_campaign_coupons(self, *, campaign_id: int) -> list[dict]:
        sql = """
            SELECT
                wco.coupon_code AS code,
                MIN(wco.order_date)                       AS first_date,
                MAX(wco.order_date)                       AS last_date,
                COUNT(DISTINCT wco.order_id)              AS n_orders,
                SUM(wco.quantity)::float                  AS units,
                SUM(wco.discount_amount)::float           AS discount,
                CASE WHEN SUM(wco.original_price) > 0
                     THEN (SUM(wco.discount_amount) / SUM(wco.original_price) * 100)::float
                     ELSE 0 END                           AS avg_disc_pct
            FROM webshop_coupon_orders wco
            WHERE wco.campaign_id = :cid
            GROUP BY wco.coupon_code
            ORDER BY units DESC
        """
        rows = self.db.execute(text(sql), {"cid": campaign_id}).mappings().all()
        return [dict(r) for r in rows]

    def get_marketing_daily(self, *, campaign_id: int) -> list[dict]:
        """Per-(date, coupon_code) units roll-up for the stacked-bar chart."""
        sql = """
            SELECT
                wco.order_date          AS day,
                wco.coupon_code         AS code,
                SUM(wco.quantity)::float AS units
            FROM webshop_coupon_orders wco
            WHERE wco.campaign_id = :cid
            GROUP BY wco.order_date, wco.coupon_code
            ORDER BY day, code
        """
        rows = self.db.execute(text(sql), {"cid": campaign_id}).mappings().all()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # Forecaster — past-promo stats for a SKU + sibling fallback
    # ------------------------------------------------------------------

    def get_subcategory_siblings(self, *, sku: str, limit: int = 50) -> list[str]:
        """Return up to `limit` sibling SKUs in the same category. Used as
        a fallback when the focal SKU has no past promos."""
        sql = """
            WITH focal AS (
                SELECT category_id FROM dim_products WHERE sku = :sku
            )
            SELECT p.sku
            FROM dim_products p, focal
            WHERE p.category_id = focal.category_id
              AND p.sku <> :sku
              AND p.active = true
            ORDER BY p.sku
            LIMIT :limit
        """
        rows = self.db.execute(text(sql), {"sku": sku, "limit": limit}).fetchall()
        return [r[0] for r in rows]

    def get_category_price_stats(self, *, category: str) -> Optional[dict]:
        """p10/p25/median/p75 of avg_sell_price across SKUs in a category.
        Used for the price-disruptor multiplier (anchored to p10)."""
        sql = """
            WITH prices AS (
                SELECT ep.avg_sell_price::float AS price
                FROM erp_prices ep
                JOIN dim_products p ON p.id = ep.product_id
                LEFT JOIN dim_categories dc ON dc.id = p.category_id
                WHERE dc.name = :cat AND ep.avg_sell_price > 0
            )
            SELECT
                COUNT(*)                                          AS n_skus,
                PERCENTILE_CONT(0.10) WITHIN GROUP (ORDER BY price)::float AS p10,
                PERCENTILE_CONT(0.25) WITHIN GROUP (ORDER BY price)::float AS p25,
                PERCENTILE_CONT(0.50) WITHIN GROUP (ORDER BY price)::float AS p50,
                PERCENTILE_CONT(0.75) WITHIN GROUP (ORDER BY price)::float AS p75
            FROM prices
        """
        row = self.db.execute(text(sql), {"cat": category}).mappings().first()
        if not row or not row.get("n_skus"):
            return None
        return dict(row)

    # ------------------------------------------------------------------
    # Calendar — list promo_proposals (these are the calendar entries)
    # ------------------------------------------------------------------

    def list_calendar_entries(
        self,
        *,
        source: Optional[list[str]] = None,
        status: Optional[list[str]] = None,
        min_yw: Optional[int] = None,
        max_yw: Optional[int] = None,
    ) -> list[dict]:
        """List promo_proposals filtered for the Calendar view. The
        promo_proposals table is our 'calendar store' — each row is a
        proposed/approved/live campaign. Matches the shape of
        PromoCalendar/data/promo_calendar.csv."""
        where_clauses: list[str] = []
        params: dict = {}
        if source:
            where_clauses.append("pp.source = ANY(:sources)")
            params["sources"] = source
        if status:
            where_clauses.append("pp.status = ANY(:statuses)")
            params["statuses"] = status
        if min_yw is not None:
            where_clauses.append("(pp.end_year * 100 + pp.end_week) >= :min_yw")
            params["min_yw"] = min_yw
        if max_yw is not None:
            where_clauses.append("(pp.start_year * 100 + pp.start_week) <= :max_yw")
            params["max_yw"] = max_yw
        where_sql = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else ""

        sql = f"""
            SELECT
                pp.id, pp.name, pp.source, pp.mechanic, pp.discount_pct::float,
                pp.start_year, pp.start_week, pp.end_year, pp.end_week,
                pp.skus, pp.status, pp.objective, pp.proposed_by_id, pp.channels, pp.log,
                pp.created_at,
                COALESCE(NULLIF(u.display_name, ''), u.username) AS proposed_by
            FROM promo_proposals pp
            LEFT JOIN users u ON u.id = pp.proposed_by_id
            {where_sql}
            ORDER BY pp.start_year ASC NULLS LAST, pp.start_week ASC NULLS LAST, pp.id DESC
        """
        rows = self.db.execute(text(sql), params).mappings().all()
        return [dict(r) for r in rows]

    def get_proposal(self, *, proposal_id: int) -> Optional[dict]:
        sql = """
            SELECT
                pp.id, pp.name, pp.source, pp.mechanic, pp.discount_pct::float,
                pp.start_year, pp.start_week, pp.end_year, pp.end_week,
                pp.skus, pp.status, pp.objective, pp.proposed_by_id, pp.channels, pp.log,
                pp.created_at,
                COALESCE(NULLIF(u.display_name, ''), u.username) AS proposed_by
            FROM promo_proposals pp
            LEFT JOIN users u ON u.id = pp.proposed_by_id
            WHERE pp.id = :pid
        """
        row = self.db.execute(text(sql), {"pid": proposal_id}).mappings().first()
        return dict(row) if row else None

    def update_proposal_status(
        self, *, proposal_id: int, new_status: str, log_line: Optional[str] = None,
    ) -> bool:
        """Update status + optionally append a log line. Used by re-submit,
        withdraw, approve, reject, request-revision."""
        if log_line:
            sql = """
                UPDATE promo_proposals
                SET status = :status,
                    log = COALESCE(NULLIF(log, ''), '') ||
                          CASE WHEN COALESCE(log, '') = '' THEN '' ELSE E'\\n' END ||
                          :line
                WHERE id = :pid
            """
            res = self.db.execute(text(sql), {
                "status": new_status, "line": log_line, "pid": proposal_id,
            })
        else:
            sql = "UPDATE promo_proposals SET status = :status WHERE id = :pid"
            res = self.db.execute(text(sql), {"status": new_status, "pid": proposal_id})
        if (res.rowcount or 0) > 0:
            self.db.commit()
            return True
        return False

    def delete_proposal(self, *, proposal_id: int) -> bool:
        """Hard delete. promo_approvals references with ON DELETE no-action,
        so we delete the approvals first."""
        self.db.execute(text("DELETE FROM promo_approvals WHERE proposal_id = :pid"),
                         {"pid": proposal_id})
        res = self.db.execute(text("DELETE FROM promo_proposals WHERE id = :pid"),
                              {"pid": proposal_id})
        if (res.rowcount or 0) > 0:
            self.db.commit()
            return True
        self.db.rollback()
        return False

    def add_approval(
        self, *, proposal_id: int, reviewed_by_id: Optional[int],
        decision: str, feedback: Optional[str] = None,
    ) -> int:
        """Insert into promo_approvals. Returns new id."""
        sql = """
            INSERT INTO promo_approvals (proposal_id, reviewed_by_id, decision, feedback)
            VALUES (:pid, :rid, :decision, :feedback)
            RETURNING id
        """
        row = self.db.execute(text(sql), {
            "pid": proposal_id, "rid": reviewed_by_id,
            "decision": decision, "feedback": feedback,
        }).mappings().first()
        self.db.commit()
        return int(row["id"]) if row else 0

    def list_approvals(self, *, proposal_id: Optional[int] = None) -> list[dict]:
        sql_where = "WHERE pa.proposal_id = :pid" if proposal_id is not None else ""
        sql = f"""
            SELECT
                pa.id, pa.proposal_id, pa.decision, pa.feedback, pa.decided_at,
                pa.reviewed_by_id,
                COALESCE(NULLIF(u.display_name, ''), u.username) AS reviewed_by
            FROM promo_approvals pa
            LEFT JOIN users u ON u.id = pa.reviewed_by_id
            {sql_where}
            ORDER BY pa.decided_at DESC NULLS LAST, pa.id DESC
        """
        params = {"pid": proposal_id} if proposal_id is not None else {}
        rows = self.db.execute(text(sql), params).mappings().all()
        return [dict(r) for r in rows]

    # ------------------------------------------------------------------
    # Calendar stats — KPI strip
    # ------------------------------------------------------------------

    def get_calendar_stats(
        self,
        *,
        source: Optional[list[str]] = None,
        status: Optional[list[str]] = None,
        min_yw: Optional[int] = None,
        max_yw: Optional[int] = None,
    ) -> dict:
        """5 KPIs matching the Calendar dashboard: total / pending / SKUs / conflicts / units.
        Conflicts count comes from list_calendar_entries — caller computes."""
        where_clauses: list[str] = []
        params: dict = {}
        if source:
            where_clauses.append("source = ANY(:sources)")
            params["sources"] = source
        if status:
            where_clauses.append("status = ANY(:statuses)")
            params["statuses"] = status
        if min_yw is not None:
            where_clauses.append("(end_year * 100 + end_week) >= :min_yw")
            params["min_yw"] = min_yw
        if max_yw is not None:
            where_clauses.append("(start_year * 100 + start_week) <= :max_yw")
            params["max_yw"] = max_yw
        where_sql = ("WHERE " + " AND ".join(where_clauses)) if where_clauses else ""

        sql = f"""
            SELECT
                COUNT(*)                                                 AS total,
                COUNT(*) FILTER (WHERE status IN ('draft','submitted'))  AS pending,
                COUNT(*) FILTER (WHERE status IN ('approved'))           AS approved,
                jsonb_path_query_array(
                    jsonb_agg(COALESCE(skus, '[]'::jsonb)),
                    '$[*][*].sku'
                ) AS sku_list
            FROM promo_proposals
            {where_sql}
        """
        row = self.db.execute(text(sql), params).mappings().first() or {}
        sku_list = row.get("sku_list") or []
        unique_skus = len({s for s in sku_list if s})
        return {
            "total":    int(row.get("total") or 0),
            "pending":  int(row.get("pending") or 0),
            "approved": int(row.get("approved") or 0),
            "n_skus":   unique_skus,
        }

    # ------------------------------------------------------------------
    # Category → SKU map (for category-vs-SKU conflict detection)
    # ------------------------------------------------------------------

    def get_category_sku_map(self) -> dict[str, set[str]]:
        """Return {category_name: set(sku)} for the categories that appear
        in any proposal (we don't expand all categories — only the ones
        the conflict detector needs)."""
        sql = """
            SELECT dc.name AS category, p.sku
            FROM dim_products p
            JOIN dim_categories dc ON dc.id = p.category_id
            WHERE p.active = true
        """
        rows = self.db.execute(text(sql)).mappings().all()
        out: dict[str, set[str]] = {}
        for r in rows:
            cat = r.get("category")
            sku = r.get("sku")
            if cat and sku:
                out.setdefault(str(cat), set()).add(str(sku))
        return out

    def get_discount_ranges_by_name(self, names: list[str]) -> dict[str, tuple]:
        """{campaign_name: (min_pct, max_pct)} from erp_promo_items, matched by
        erp_promo_campaigns.type = proposal name. Used to show the Rabat range
        on calendar entries that originated from an ERP campaign. Empty for
        manually-created proposals whose name doesn't match a campaign."""
        names = [n for n in (names or []) if n]
        if not names:
            return {}
        sql = """
            SELECT c.type AS name,
                   MIN(i.discount_pct)::float AS dmin,
                   MAX(i.discount_pct)::float AS dmax
            FROM erp_promo_campaigns c
            JOIN erp_promo_items i ON i.campaign_id = c.id
            WHERE c.type = ANY(:names) AND i.discount_pct IS NOT NULL
            GROUP BY c.type
        """
        rows = self.db.execute(text(sql), {"names": names}).mappings().all()
        return {str(r["name"]): (r["dmin"], r["dmax"]) for r in rows}

    def get_sku_overlaps(self, *, proposal_ids: list[int]) -> list[dict]:
        """SKUs that appear in 2+ of the given proposals (double-promoted).
        Returns one dict per overlapping SKU with its name, category, and the
        list of promos it's in — each with that promo's discount %. Empty if
        fewer than 2 proposals or no overlaps."""
        ids = [int(i) for i in (proposal_ids or [])]
        if len(ids) < 2:
            return []
        sql = """
            WITH promo_skus AS (
                SELECT pp.id   AS proposal_id,
                       pp.name AS proposal_name,
                       (elem->>'sku') AS sku
                FROM promo_proposals pp,
                     LATERAL jsonb_array_elements(pp.skus) AS elem
                WHERE pp.id = ANY(:ids) AND pp.skus IS NOT NULL
            ),
            overlap AS (
                SELECT sku FROM promo_skus
                WHERE sku IS NOT NULL AND sku <> ''
                GROUP BY sku HAVING COUNT(DISTINCT proposal_id) >= 2
            )
            SELECT ps.sku,
                   p.name  AS product_name,
                   dc.name AS category,
                   ps.proposal_id,
                   ps.proposal_name,
                   (SELECT MIN(i.discount_pct)
                      FROM erp_promo_items i
                      JOIN erp_promo_campaigns c ON c.id = i.campaign_id
                     WHERE c.type = ps.proposal_name AND i.product_id = p.id)::float AS discount_pct
            FROM promo_skus ps
            JOIN overlap o ON o.sku = ps.sku
            LEFT JOIN dim_products p   ON p.sku = ps.sku
            LEFT JOIN dim_categories dc ON dc.id = p.category_id
            ORDER BY ps.sku, ps.proposal_id
        """
        rows = self.db.execute(text(sql), {"ids": ids}).mappings().all()
        out: dict[str, dict] = {}
        for r in rows:
            sku = str(r["sku"])
            entry = out.setdefault(sku, {
                "sku": sku, "name": r.get("product_name"),
                "category": r.get("category"), "promos": [],
            })
            entry["promos"].append({
                "proposal_id": int(r["proposal_id"]),
                "proposal_name": r.get("proposal_name"),
                "discount_pct": r.get("discount_pct"),
            })
        return list(out.values())

    def get_proposal_sku_detail(self, *, proposal_id: int) -> list[dict]:
        """Per-SKU detail for one proposal: sku, name, category, regular price,
        promo discount % (from the matching ERP campaign), and the implied promo
        price. Used by the calendar drill-down + detail panel. Returns rows in
        the proposal's own SKU order; empty if the proposal has no SKUs."""
        row = self.db.execute(
            text("SELECT name, skus FROM promo_proposals WHERE id = :pid"),
            {"pid": proposal_id},
        ).mappings().first()
        if not row:
            return []
        cname = row.get("name")
        skus_raw = row.get("skus") or []
        order: list[str] = []
        for s in skus_raw if isinstance(skus_raw, list) else []:
            v = str(s.get("sku") if isinstance(s, dict) else s).strip()
            if v:
                order.append(v)
        if not order:
            return []

        sql = """
            SELECT p.sku,
                   p.name,
                   dc.name AS category,
                   COALESCE(NULLIF(pr.normal_retail_ppp, 0),
                            pr.avg_sell_price,
                            NULLIF(pr.normal_webshop_ppp, 0))::float AS price,
                   (SELECT MIN(i.discount_pct)
                      FROM erp_promo_items i
                      JOIN erp_promo_campaigns c ON c.id = i.campaign_id
                     WHERE c.type = :cname AND i.product_id = p.id)::float AS discount_pct
            FROM dim_products p
            LEFT JOIN dim_categories dc ON dc.id = p.category_id
            LEFT JOIN erp_prices pr ON pr.product_id = p.id
            WHERE p.sku = ANY(:skus)
        """
        rows = self.db.execute(text(sql), {"cname": cname, "skus": order}).mappings().all()
        by_sku = {str(r["sku"]): dict(r) for r in rows}
        out: list[dict] = []
        for sku in order:
            r = by_sku.get(sku)
            if not r:
                out.append({"sku": sku, "name": None, "category": None,
                            "price": None, "discount_pct": None, "promo_price": None})
                continue
            price = r.get("price")
            disc = r.get("discount_pct")
            promo_price = (round(price * (1 - disc / 100.0), 2)
                           if price is not None and disc is not None else None)
            out.append({
                "sku": sku, "name": r.get("name"), "category": r.get("category"),
                "price": price, "discount_pct": disc, "promo_price": promo_price,
            })
        return out