"""Wholesale Review > Control Module.

First control: KAM commit-vs-pull reconciliation. For the last N closed
ISO weeks, every on-top wholesale commit is classified as either a
"first-buy" (the partner had no baseline buying pattern for that SKU)
or "incremental" (commit is supposed to LIFT the partner's normal rate)
and compared against what actually shipped in `erp_transactions`.

This page is wholesale-channel-specific — it answers "did the lift my
KAM promised on $partner show up in their order?". Demand Review (the
consensus sign-off page) lives separately and is cross-channel.

Other wholesale-side controls will land here over time (buyer-level
baseline drift, promo-policy compliance, etc.) — each as a standalone
function under the same namespace so the page can mount them as cards.
"""
from __future__ import annotations

from datetime import date, timedelta
from typing import Optional

from sqlalchemy import text
from sqlalchemy.orm import Session


# Hit-threshold for incremental commits — KAM gets credit when actual
# exceeded baseline by AT LEAST this fraction of the committed lift.
# 70% matches the 30% tolerance band used by the rest of the FA pages.
INCREMENTAL_HIT_RATIO = 0.70

# Trailing window for baseline rate calculation per (partner, SKU). 13
# weeks = quarter, matches the FA / Slow Movers / Locked Cash convention.
BASELINE_WEEKS = 13

# Wholesale channel_map_ids — RAC(6), TRC(7), VPT(8), VPB(9), RIZ(10) + RPE(11).
# RPE added 2026-05 (was an unmapped doc type carrying ~82k wholesale
# units). RAC(6) added 2026-06 — KAMs started booking wholesale orders
# through the RAC doc type (gyms/resellers via the webshop UI, priced &
# serviced as wholesale), so it now counts toward the KAM·CM FA scope.
# This is the full 'wholesale' channel: keep in sync with lookup_channel_map
# — SELECT id FROM lookup_channel_map WHERE channel = 'wholesale'.
WS_CHANNEL_MAP_IDS = (6, 7, 8, 9, 10, 11)


def _last_n_closed_iso_weeks(n: int = 4, today: Optional[date] = None
                              ) -> list[tuple[int, int]]:
    """Return [(iso_year, iso_week), ...] for the n most recently CLOSED
    ISO weeks ending strictly before `today`'s week. Current week is
    excluded — partial-week actuals always read as "actuals miss commits"
    and would false-fire the control.

    Result is ordered oldest → newest.
    """
    today = today or date.today()
    cur_iso = today.isocalendar()
    cur_yw = int(cur_iso[0]) * 100 + int(cur_iso[1])
    out: list[tuple[int, int]] = []
    # Walk backwards a week at a time from yesterday-of-prev-week
    # — go via last Sunday so we land in the previous ISO week reliably.
    last_sun = today - timedelta(days=today.weekday() + 1)
    for _ in range(n):
        iso = last_sun.isocalendar()
        yw = int(iso[0]) * 100 + int(iso[1])
        if yw < cur_yw:
            out.append((int(iso[0]), int(iso[1])))
        last_sun -= timedelta(days=7)
    return list(reversed(out))


def report_kam_commit_control(db: Session, n_weeks: int = 4) -> dict:
    """KAM commit-vs-pull control over the last `n_weeks` closed weeks.

    Output structure:
      window:    metadata about the window and matched/unmatched buyers
      coverage:  buyer→partner reconciliation
      summary:   counts + hit rates split by first-buy vs incremental
      by_kam:    per-person rollup
      rows:      per (person, buyer, sku, week) detail rows with
                 commit_qty / baseline_qty / actual_qty / classification
    """
    weeks = _last_n_closed_iso_weeks(n_weeks)
    if not weeks:
        return _empty_response(weeks)
    yw_keys = [y * 100 + w for (y, w) in weeks]

    # ── Baseline window: BASELINE_WEEKS weeks ending strictly before the
    # earliest commit week. We use one shared baseline window for all
    # weeks in the report (not a per-week sliding window) — the 13w avg
    # is robust to single-week noise, and a single window keeps the
    # "what is regular" definition consistent across the 4-week report.
    earliest_y, earliest_w = weeks[0]
    earliest_yw = earliest_y * 100 + earliest_w
    # Compute (year*100+week) of the week BASELINE_WEEKS before earliest.
    # Use date math to handle year boundaries cleanly.
    from datetime import date as _date
    try:
        earliest_monday = _date.fromisocalendar(earliest_y, earliest_w, 1)
    except (ValueError, AttributeError):
        # Python < 3.8 fallback — won't happen here, but defensive.
        return _empty_response(weeks)
    baseline_end_sun = earliest_monday - timedelta(days=1)
    baseline_start_mon = baseline_end_sun - timedelta(weeks=BASELINE_WEEKS - 1, days=6)

    # ── 1) Coverage: distinct KAM buyers in the window → ERP partner match
    # Mirrors the existing KAM·CM FA matching logic in demand_repo so the
    # two surfaces agree on who's matched. ILIKE substring + DISTINCT ON
    # highest-volume partner wins.
    cov_rows = db.execute(text("""
        WITH window_buyers AS (
            SELECT DISTINCT LOWER(TRIM(buyer)) AS buyer_lc,
                            MIN(buyer)       AS buyer
            FROM on_top_inputs
            WHERE channel = 'wholesale'
              AND buyer IS NOT NULL
              AND TRIM(buyer) <> ''
              AND year_week = ANY(:yws)
            GROUP BY LOWER(TRIM(buyer))
        ),
        candidates AS (
            SELECT wb.buyer_lc, wb.buyer,
                   dp.id AS partner_id, dp.name AS partner_name,
                   COUNT(et.partner_id) AS n_txn
            FROM window_buyers wb
            JOIN dim_partners dp
              ON LOWER(dp.name) ILIKE '%' || wb.buyer_lc || '%'
            JOIN erp_transactions et
              ON et.partner_id = dp.id
             AND et.channel_map_id = ANY(:ws_ids)
            GROUP BY wb.buyer_lc, wb.buyer, dp.id, dp.name
        ),
        matched AS (
            SELECT DISTINCT ON (buyer_lc)
                   buyer_lc, buyer, partner_id, partner_name
            FROM candidates
            ORDER BY buyer_lc, n_txn DESC
        ),
        commits_per_buyer AS (
            SELECT LOWER(TRIM(buyer)) AS buyer_lc,
                   COUNT(*)::int        AS n_commits,
                   SUM(quantity)::float AS total_commit_qty
            FROM on_top_inputs
            WHERE channel = 'wholesale'
              AND year_week = ANY(:yws)
            GROUP BY LOWER(TRIM(buyer))
        )
        SELECT wb.buyer,
               m.partner_id,
               m.partner_name,
               c.n_commits,
               c.total_commit_qty
        FROM window_buyers wb
        LEFT JOIN matched m ON m.buyer_lc = wb.buyer_lc
        LEFT JOIN commits_per_buyer c ON c.buyer_lc = wb.buyer_lc
        ORDER BY c.total_commit_qty DESC NULLS LAST
    """), {"yws": yw_keys, "ws_ids": list(WS_CHANNEL_MAP_IDS)}).mappings().all()

    coverage_rows = []
    unmatched = []
    matched = []
    for r in cov_rows:
        d = dict(r)
        d["matched"] = d.get("partner_id") is not None
        d["n_commits"] = int(d.get("n_commits") or 0)
        d["total_commit_qty"] = float(d.get("total_commit_qty") or 0)
        coverage_rows.append(d)
        if d["matched"]:
            matched.append(d["buyer"])
        else:
            unmatched.append(d["buyer"])

    # ── 2) Per (person, buyer, sku, week) detail with baseline + actual
    # Three joins:
    #   - on_top_inputs aggregated by submitted_by × buyer × product × week
    #   - per-(partner, product, week) actuals from erp_transactions in
    #     both the window AND the baseline period
    #   - baseline_avg per (partner, product) across the baseline window
    detail = db.execute(text("""
        WITH window_yws AS (SELECT UNNEST(CAST(:yws AS INTEGER[])) AS yw),
        buyer_match AS (
            -- Same logic as coverage CTE above, isolated so we can join below
            WITH wb AS (
                SELECT DISTINCT LOWER(TRIM(buyer)) AS buyer_lc
                FROM on_top_inputs
                WHERE channel = 'wholesale'
                  AND buyer IS NOT NULL AND TRIM(buyer) <> ''
                  AND year_week = ANY(:yws)
            ),
            cand AS (
                SELECT wb.buyer_lc, dp.id AS partner_id, dp.name AS partner_name,
                       COUNT(et.partner_id) AS n_txn
                FROM wb
                JOIN dim_partners dp
                  ON LOWER(dp.name) ILIKE '%' || wb.buyer_lc || '%'
                JOIN erp_transactions et
                  ON et.partner_id = dp.id
                 AND et.channel_map_id = ANY(:ws_ids)
                GROUP BY wb.buyer_lc, dp.id, dp.name
            )
            SELECT DISTINCT ON (buyer_lc) buyer_lc, partner_id, partner_name
            FROM cand
            ORDER BY buyer_lc, n_txn DESC
        ),
        ot AS (
            SELECT oti.submitted_by_id,
                   LOWER(TRIM(oti.buyer)) AS buyer_lc,
                   MIN(oti.buyer)         AS buyer,
                   oti.product_id,
                   oti.year_week,
                   SUM(oti.quantity)::float AS commit_qty
            FROM on_top_inputs oti
            WHERE oti.channel = 'wholesale'
              AND oti.buyer IS NOT NULL
              AND TRIM(oti.buyer) <> ''
              AND oti.year_week = ANY(:yws)
            GROUP BY oti.submitted_by_id, LOWER(TRIM(oti.buyer)),
                     oti.product_id, oti.year_week
        ),
        actuals_window AS (
            SELECT et.partner_id, et.product_id,
                   EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100
                 + EXTRACT(WEEK    FROM et.transaction_date)::int AS year_week,
                   SUM(et.quantity)::float AS qty
            FROM erp_transactions et
            WHERE et.channel_map_id = ANY(:ws_ids)
            GROUP BY et.partner_id, et.product_id,
                     EXTRACT(ISOYEAR FROM et.transaction_date)::int,
                     EXTRACT(WEEK    FROM et.transaction_date)::int
        ),
        baseline_agg AS (
            -- Average weekly quantity per (partner, product) across the
            -- 13-week pre-window period. Divided by BASELINE_WEEKS so a
            -- partner that bought 130 units once and zero otherwise has a
            -- baseline of 10 (not 130) — captures the "normal weekly rate".
            SELECT et.partner_id, et.product_id,
                   SUM(et.quantity)::float / :baseline_weeks AS baseline_qty
            FROM erp_transactions et
            WHERE et.channel_map_id = ANY(:ws_ids)
              AND et.transaction_date BETWEEN DATE :base_start AND DATE :base_end
            GROUP BY et.partner_id, et.product_id
        )
        SELECT
            COALESCE(NULLIF(u.display_name, ''), u.username, '(unassigned)') AS person,
            ot.buyer,
            bm.partner_id,
            bm.partner_name,
            p.sku,
            p.name AS sku_name,
            sp.tier,
            (ot.year_week / 100) AS year,
            (ot.year_week % 100) AS week,
            ot.commit_qty,
            COALESCE(ba.baseline_qty, 0)::float AS baseline_qty,
            COALESCE(aw.qty, 0)::float          AS actual_qty
        FROM ot
        JOIN dim_products p ON p.id = ot.product_id
        LEFT JOIN sku_planning sp ON sp.product_id = ot.product_id
        LEFT JOIN users u ON u.id = ot.submitted_by_id
        LEFT JOIN buyer_match bm ON bm.buyer_lc = ot.buyer_lc
        LEFT JOIN baseline_agg ba
               ON ba.partner_id = bm.partner_id
              AND ba.product_id = ot.product_id
        LEFT JOIN actuals_window aw
               ON aw.partner_id = bm.partner_id
              AND aw.product_id = ot.product_id
              AND aw.year_week  = ot.year_week
        ORDER BY ot.year_week, person, ot.buyer, p.sku
    """), {
        "yws": yw_keys,
        "ws_ids": list(WS_CHANNEL_MAP_IDS),
        "baseline_weeks": BASELINE_WEEKS,
        "base_start": baseline_start_mon.isoformat(),
        "base_end":   baseline_end_sun.isoformat(),
    }).mappings().all()

    rows = []
    for r in detail:
        d = dict(r)
        d["commit_qty"]   = float(d.get("commit_qty") or 0)
        d["baseline_qty"] = float(d.get("baseline_qty") or 0)
        d["actual_qty"]   = float(d.get("actual_qty") or 0)
        d["is_first_buy"] = d["baseline_qty"] <= 0.0
        d["is_unmatched"] = d.get("partner_id") is None
        d["incremental_qty"] = d["actual_qty"] - d["baseline_qty"]

        if d["is_unmatched"]:
            # Can't evaluate — surface separately, no actual to compare against
            d["classification"] = "UNMATCHED"
        elif d["is_first_buy"]:
            d["classification"] = "FIRST_BUY_HIT" if d["actual_qty"] > 0 else "FIRST_BUY_MISS"
        else:
            target_lift = d["commit_qty"] * INCREMENTAL_HIT_RATIO
            if d["incremental_qty"] >= target_lift:
                d["classification"] = "INCREMENTAL_HIT"
            elif d["incremental_qty"] > 0:
                d["classification"] = "INCREMENTAL_PARTIAL"
            else:
                d["classification"] = "INCREMENTAL_MISS"
        rows.append(d)

    # ── 3) Summary + by_kam rollup
    summary = {
        "n_commits":          len(rows),
        "n_first_buy":        sum(1 for r in rows if r["is_first_buy"] and not r["is_unmatched"]),
        "n_incremental":      sum(1 for r in rows if (not r["is_first_buy"]) and not r["is_unmatched"]),
        "n_unmatched":        sum(1 for r in rows if r["is_unmatched"]),
        "n_first_buy_hits":   sum(1 for r in rows if r["classification"] == "FIRST_BUY_HIT"),
        "n_incremental_hits": sum(1 for r in rows if r["classification"] == "INCREMENTAL_HIT"),
        "n_incremental_partial": sum(1 for r in rows if r["classification"] == "INCREMENTAL_PARTIAL"),
        "n_incremental_miss": sum(1 for r in rows if r["classification"] == "INCREMENTAL_MISS"),
        "n_first_buy_miss":   sum(1 for r in rows if r["classification"] == "FIRST_BUY_MISS"),
        "total_commit_qty":   sum(r["commit_qty"]   for r in rows),
        "total_actual_qty":   sum(r["actual_qty"]   for r in rows if not r["is_unmatched"]),
        "total_baseline_qty": sum(r["baseline_qty"] for r in rows if not r["is_unmatched"]),
        "incremental_realized_qty":
            sum(max(0.0, r["incremental_qty"]) for r in rows
                if not r["is_first_buy"] and not r["is_unmatched"]),
        "incremental_committed_qty":
            sum(r["commit_qty"] for r in rows
                if not r["is_first_buy"] and not r["is_unmatched"]),
    }
    summary["first_buy_hit_rate"] = (
        summary["n_first_buy_hits"] / summary["n_first_buy"]
        if summary["n_first_buy"] else 0.0
    )
    summary["incremental_hit_rate"] = (
        summary["n_incremental_hits"] / summary["n_incremental"]
        if summary["n_incremental"] else 0.0
    )
    summary["incremental_lift_realized_pct"] = (
        summary["incremental_realized_qty"] / summary["incremental_committed_qty"]
        if summary["incremental_committed_qty"] else 0.0
    )

    by_kam: dict[str, dict] = {}
    for r in rows:
        k = r["person"]
        slot = by_kam.setdefault(k, {
            "person": k,
            "n_commits": 0, "n_hits": 0, "n_partial": 0, "n_miss": 0, "n_unmatched": 0,
            "commit_qty": 0.0, "actual_qty": 0.0, "baseline_qty": 0.0,
        })
        slot["n_commits"]   += 1
        slot["commit_qty"]  += r["commit_qty"]
        if r["is_unmatched"]:
            slot["n_unmatched"] += 1
            continue
        slot["actual_qty"]   += r["actual_qty"]
        slot["baseline_qty"] += r["baseline_qty"]
        cls = r["classification"]
        if cls in ("FIRST_BUY_HIT", "INCREMENTAL_HIT"):
            slot["n_hits"] += 1
        elif cls == "INCREMENTAL_PARTIAL":
            slot["n_partial"] += 1
        elif cls in ("FIRST_BUY_MISS", "INCREMENTAL_MISS"):
            slot["n_miss"] += 1
    by_kam_list = sorted(by_kam.values(),
                          key=lambda x: x["commit_qty"], reverse=True)
    for slot in by_kam_list:
        evaluable = slot["n_commits"] - slot["n_unmatched"]
        slot["hit_rate"] = (slot["n_hits"] / evaluable) if evaluable else 0.0

    return {
        "window": {
            "n_weeks":      n_weeks,
            "weeks":        [f"CW{w:02d}" for (_, w) in weeks],
            "year_week_keys": yw_keys,
            "baseline_window": {
                "weeks":       BASELINE_WEEKS,
                "start":       baseline_start_mon.isoformat(),
                "end":         baseline_end_sun.isoformat(),
            },
            "hit_threshold_ratio": INCREMENTAL_HIT_RATIO,
        },
        "coverage": {
            "n_total":     len(coverage_rows),
            "n_matched":   len(matched),
            "n_unmatched": len(unmatched),
            "matched":     matched,
            "unmatched":   unmatched,
            "rows":        coverage_rows,
        },
        "summary": summary,
        "by_kam":  by_kam_list,
        "rows":    rows,
    }


def _empty_response(weeks: list[tuple[int, int]]) -> dict:
    return {
        "window": {
            "n_weeks": 0,
            "weeks": [f"CW{w:02d}" for (_, w) in weeks],
            "year_week_keys": [],
            "baseline_window": {"weeks": BASELINE_WEEKS, "start": None, "end": None},
            "hit_threshold_ratio": INCREMENTAL_HIT_RATIO,
        },
        "coverage": {"n_total": 0, "n_matched": 0, "n_unmatched": 0,
                     "matched": [], "unmatched": [], "rows": []},
        "summary":  {},
        "by_kam":   [],
        "rows":     [],
    }


# ─────────────────────────────────────────────────────────────────────
# Buyer Planner — KAM → buyer → week grid with OOS-risk flags
# ─────────────────────────────────────────────────────────────────────
# Lets a KAM drill into one buyer and see, week by week, the SKUs that
# buyer has on order — with a red flag on any (SKU, week) where the SKU
# is projected out of stock. Stock is shared, so per the agreed rule we
# flag EVERY buyer who ordered an OOS SKU that week (no allocation).
#
# Buyer→partner resolution goes through buyer_partner_aliases so the
# unmatched labels (MCI, NTL, Plodine, Tisak, Ostalo) can be mapped from
# the UI; until mapped they show "needs mapping".
def _buyer_partner_map(db: Session) -> dict[str, dict]:
    """buyer_lc -> {partner_ids: [...], partner_name, matched}.

    The mapping is one-to-MANY: a single KAM buyer label can fan out to
    several ERP partners (e.g. Selma's 'Ostalo' = Tifon + KTC + Vrutak +
    I Novine). Consumers aggregate ERP pull across all of partner_ids.
    `partner_id` (first id) is kept for back-compat with the single-partner
    nav/selected shapes. Unmapped buyers won't be in the dict."""
    rows = db.execute(text("""
        SELECT a.buyer_lc, a.partner_id, dp.name AS partner_name
        FROM buyer_partner_aliases a
        JOIN dim_partners dp ON dp.id = a.partner_id
        ORDER BY a.buyer_lc, dp.name
    """)).mappings().all()
    out: dict[str, dict] = {}
    for r in rows:
        slot = out.setdefault(r["buyer_lc"], {"partner_ids": [], "_names": []})
        slot["partner_ids"].append(int(r["partner_id"]))
        if r["partner_name"]:
            slot["_names"].append(r["partner_name"])
    for slot in out.values():
        names = slot.pop("_names")
        slot["partner_name"] = (
            " + ".join(names) if len(names) <= 2 else f"{len(names)} partners"
        )
        slot["partner_id"] = slot["partner_ids"][0] if slot["partner_ids"] else None
        slot["matched"] = len(slot["partner_ids"]) > 0
    return out


def _kam_buyer_nav(db: Session) -> list[dict]:
    """KAM → buyers navigation tree. Each buyer carries match status (via
    alias table) + how many weeks / total qty they have on order."""
    rows = db.execute(text("""
        SELECT oti.submitted_by_id,
               COALESCE(NULLIF(u.display_name, ''), u.username, '(unassigned)') AS person,
               oti.buyer,
               LOWER(TRIM(oti.buyer)) AS buyer_lc,
               COUNT(DISTINCT oti.year_week) AS n_weeks,
               SUM(oti.quantity)::float      AS total_qty
        FROM on_top_inputs oti
        LEFT JOIN users u ON u.id = oti.submitted_by_id
        WHERE oti.channel = 'wholesale'
          AND oti.buyer IS NOT NULL AND TRIM(oti.buyer) <> ''
        GROUP BY oti.submitted_by_id, person, oti.buyer, LOWER(TRIM(oti.buyer))
        ORDER BY person, oti.buyer
    """)).mappings().all()
    amap = _buyer_partner_map(db)
    kams: dict[str, dict] = {}
    for r in rows:
        person = r["person"]
        slot = kams.setdefault(person, {"person": person,
                                         "submitted_by_id": r["submitted_by_id"],
                                         "buyers": []})
        alias = amap.get(r["buyer_lc"], {})
        slot["buyers"].append({
            "buyer":        r["buyer"],
            "buyer_lc":     r["buyer_lc"],
            "partner_id":   alias.get("partner_id"),
            "partner_name": alias.get("partner_name"),
            "matched":      alias.get("partner_id") is not None,
            "n_weeks":      int(r["n_weeks"]),
            "total_qty":    float(r["total_qty"]),
        })
    return list(kams.values())


def _oos_weeks_for_products(db: Session, product_ids: list[int],
                             order_yws: list[int]) -> set[tuple[int, int]]:
    """Return the set of (product_id, year_week) projected out of stock.

    Walk-forward per SKU from the CURRENT ISO week across the horizon
    spanned by the buyer's order weeks: opening = current stock; each
    week subtract forecast demand (latest run total, fallback trailing
    13w avg) and add incoming_supply. When (opening − demand + incoming)
    goes negative that week is flagged OOS. Only future weeks are walked
    — a past order week can't be "at risk" any more.
    """
    if not product_ids or not order_yws:
        return set()
    today = date.today()
    cur_iso = today.isocalendar()
    cur_yw = int(cur_iso[0]) * 100 + int(cur_iso[1])
    max_order_yw = max(order_yws)
    if max_order_yw < cur_yw:
        return set()

    horizon: list[tuple[int, int]] = []
    cur = today - timedelta(days=today.weekday())
    for _ in range(26):
        iso = cur.isocalendar()
        yw = int(iso[0]) * 100 + int(iso[1])
        horizon.append((int(iso[0]), int(iso[1])))
        if yw >= max_order_yw:
            break
        cur += timedelta(days=7)

    stock_rows = db.execute(text("""
        SELECT esc.product_id, SUM(esc.stock_qty)::float AS qty
        FROM erp_stock_current esc
        JOIN dim_stores ds ON ds.id = esc.store_id
        WHERE ds.is_warehouse = TRUE AND esc.product_id = ANY(:pids)
        GROUP BY esc.product_id
    """), {"pids": product_ids}).mappings().all()
    stock_map = {int(r["product_id"]): float(r["qty"] or 0) for r in stock_rows}

    fc_rows = db.execute(text("""
        SELECT DISTINCT ON (product_id, year, week)
               product_id, year, week, COALESCE(total, 0)::float AS demand
        FROM forecasts
        WHERE product_id = ANY(:pids)
        ORDER BY product_id, year, week, run_id DESC
    """), {"pids": product_ids}).mappings().all()
    fc_map = {(int(r["product_id"]), int(r["year"]) * 100 + int(r["week"])):
              float(r["demand"]) for r in fc_rows}

    avg_rows = db.execute(text("""
        WITH last13 AS (
            SELECT year, week FROM v_sales_weekly_full
            GROUP BY year, week ORDER BY year DESC, week DESC LIMIT 13
        )
        SELECT v.product_id, AVG(COALESCE(v.qty_total, 0))::float AS avg_qty
        FROM v_sales_weekly_full v JOIN last13 l USING (year, week)
        WHERE v.product_id = ANY(:pids)
        GROUP BY v.product_id
    """), {"pids": product_ids}).mappings().all()
    avg_map = {int(r["product_id"]): float(r["avg_qty"] or 0) for r in avg_rows}

    inc_rows = db.execute(text("""
        SELECT product_id, year, week, SUM(quantity)::float AS qty
        FROM incoming_supply
        WHERE product_id = ANY(:pids)
        GROUP BY product_id, year, week
    """), {"pids": product_ids}).mappings().all()
    inc_map = {(int(r["product_id"]), int(r["year"]) * 100 + int(r["week"])):
               float(r["qty"]) for r in inc_rows}

    flagged: set[tuple[int, int]] = set()
    for pid in product_ids:
        opening = stock_map.get(pid, 0.0)
        for (y, w) in horizon:
            yw = y * 100 + w
            demand = fc_map.get((pid, yw))
            if demand is None:
                demand = avg_map.get(pid, 0.0)
            incoming = inc_map.get((pid, yw), 0.0)
            closing = opening - demand + incoming
            if closing < 0:
                flagged.add((pid, yw))
                opening = 0.0
            else:
                opening = closing
    return flagged


def get_buyer_planner(db: Session, *, kam: Optional[str] = None,
                      buyer: Optional[str] = None) -> dict:
    """KAM → buyer → SKU×week grid with OOS-risk flags.

    Without `buyer` the response is just the navigation tree. With
    `buyer` selected, returns the per-SKU grid for that buyer: rows =
    SKUs ordered, columns = weeks with orders, cells = on-top qty +
    oos_risk flag.
    """
    nav = _kam_buyer_nav(db)
    result: dict = {"kams": nav, "selected": None, "grid": None}
    if not buyer:
        return result

    buyer_lc = buyer.strip().lower()
    params: dict = {"buyer_lc": buyer_lc}
    kam_clause = ""
    if kam:
        kam_clause = " AND COALESCE(NULLIF(u.display_name,''), u.username, '(unassigned)') = :kam"
        params["kam"] = kam
    orders = db.execute(text(f"""
        SELECT oti.product_id, p.sku, p.name AS sku_name, sp.tier,
               oti.year_week, SUM(oti.quantity)::float AS qty
        FROM on_top_inputs oti
        JOIN dim_products p ON p.id = oti.product_id
        LEFT JOIN sku_planning sp ON sp.product_id = oti.product_id
        LEFT JOIN users u ON u.id = oti.submitted_by_id
        WHERE oti.channel = 'wholesale'
          AND LOWER(TRIM(oti.buyer)) = :buyer_lc
          {kam_clause}
        GROUP BY oti.product_id, p.sku, p.name, sp.tier, oti.year_week
        ORDER BY p.sku, oti.year_week
    """), params).mappings().all()

    if not orders:
        result["selected"] = {"kam": kam, "buyer": buyer}
        result["grid"] = {"weeks": [], "year_weeks": [], "rows": [], "n_oos_cells": 0}
        return result

    product_ids = sorted({int(r["product_id"]) for r in orders})
    order_yws   = sorted({int(r["year_week"]) for r in orders})
    oos = _oos_weeks_for_products(db, product_ids, order_yws)

    week_labels = [f"CW{yw % 100:02d}" for yw in order_yws]
    today_iso = date.today().isocalendar()
    cur_yw = int(today_iso[0]) * 100 + int(today_iso[1])

    by_sku: dict[int, dict] = {}
    for r in orders:
        pid = int(r["product_id"])
        slot = by_sku.setdefault(pid, {
            "product_id": pid, "sku": r["sku"], "name": r["sku_name"],
            "tier": r["tier"], "cells": {}, "total_qty": 0.0, "any_oos": False,
        })
        yw = int(r["year_week"])
        lbl = f"CW{yw % 100:02d}"
        is_oos = (pid, yw) in oos
        slot["cells"][lbl] = {
            "year_week": yw, "qty": float(r["qty"]),
            "oos_risk": is_oos, "is_future": yw >= cur_yw,
        }
        slot["total_qty"] += float(r["qty"])
        if is_oos:
            slot["any_oos"] = True

    rows = sorted(by_sku.values(), key=lambda x: (not x["any_oos"], x["sku"]))
    result["selected"] = {"kam": kam, "buyer": buyer}
    result["grid"] = {
        "weeks":       week_labels,
        "year_weeks":  order_yws,
        "rows":        rows,
        "n_oos_cells": len(oos & {(p, yw) for p in product_ids for yw in order_yws}),
    }
    return result


# ─────────────────────────────────────────────────────────────────────
# Inline editing — patch on_top_inputs + rewrite forecasts.total live
# ─────────────────────────────────────────────────────────────────────
# Per the agreed design: editing a cell writes the new quantity to
# on_top_inputs AND immediately re-aggregates that (SKU, week)'s
# wholesale on-tops into forecasts.on_top_wholesale + recomputes
# forecasts.total. The change shows instantly on Revenue Forecast,
# Decomposition and Stock Projection. Edits live in the current S&OP
# cycle's rows, so prior cycles' history is untouched. The on-top
# component survives the next 'Run forecast' because it's persisted in
# on_top_inputs (the engine re-reads it).

def _current_cycle_id(db: Session) -> Optional[int]:
    """Latest sop_cycles id — edits write into this cycle's rows so prior
    cycles' history stays immutable."""
    row = db.execute(text(
        "SELECT id FROM sop_cycles ORDER BY id DESC LIMIT 1"
    )).first()
    return int(row[0]) if row else None


def _recompute_ws_forecast(db: Session, product_id: int, year_week: int) -> None:
    """Re-aggregate wholesale on-tops for (product, week) and rewrite the
    latest forecast run's on_top_wholesale + total. Inserts a minimal
    forecast row when the week wasn't forecast for this SKU but now has
    an on-top (e.g. a date-move into a fresh week)."""
    year, week = year_week // 100, year_week % 100
    new_ws = float(db.execute(text("""
        SELECT COALESCE(SUM(quantity), 0)::float
        FROM on_top_inputs
        WHERE product_id = :p AND year_week = :yw AND channel = 'wholesale'
    """), {"p": product_id, "yw": year_week}).scalar() or 0.0)

    row = db.execute(text("""
        SELECT id, COALESCE(baseline,0)::float AS baseline,
               COALESCE(planner_factor,1)::float AS pf,
               COALESCE(on_top_retail,0)::float AS mp,
               COALESCE(promo_uplift,0)::float AS promo
        FROM forecasts
        WHERE product_id = :p AND year = :y AND week = :w
        ORDER BY run_id DESC LIMIT 1
    """), {"p": product_id, "y": year, "w": week}).mappings().first()

    if row:
        new_total = row["baseline"] * row["pf"] + new_ws + row["mp"] + row["promo"]
        db.execute(text("""
            UPDATE forecasts
            SET on_top_wholesale = :ws, total = :total
            WHERE id = :id
        """), {"ws": new_ws, "total": new_total, "id": row["id"]})
    elif new_ws > 0:
        run_id = db.execute(text("SELECT MAX(run_id) FROM forecasts")).scalar()
        db.execute(text("""
            INSERT INTO forecasts
                (run_id, product_id, year, week, baseline, planner_factor,
                 on_top_wholesale, on_top_retail, promo_uplift, total)
            VALUES (:run, :p, :y, :w, 0, 1, :ws, 0, 0, :ws)
        """), {"run": run_id, "p": product_id, "y": year, "w": week, "ws": new_ws})


def _recompute_retail_forecast(db: Session, product_id: int, year_week: int) -> None:
    """Retail twin of `_recompute_ws_forecast`: re-aggregate retail on-tops for
    (product, week) into the latest forecast run's on_top_retail + total."""
    year, week = year_week // 100, year_week % 100
    new_mp = float(db.execute(text("""
        SELECT COALESCE(SUM(quantity), 0)::float
        FROM on_top_inputs
        WHERE product_id = :p AND year_week = :yw AND channel = 'retail'
    """), {"p": product_id, "yw": year_week}).scalar() or 0.0)

    row = db.execute(text("""
        SELECT id, COALESCE(baseline,0)::float AS baseline,
               COALESCE(planner_factor,1)::float AS pf,
               COALESCE(on_top_wholesale,0)::float AS ws,
               COALESCE(promo_uplift,0)::float AS promo
        FROM forecasts
        WHERE product_id = :p AND year = :y AND week = :w
        ORDER BY run_id DESC LIMIT 1
    """), {"p": product_id, "y": year, "w": week}).mappings().first()

    if row:
        new_total = row["baseline"] * row["pf"] + row["ws"] + new_mp + row["promo"]
        db.execute(text("""
            UPDATE forecasts SET on_top_retail = :mp, total = :total WHERE id = :id
        """), {"mp": new_mp, "total": new_total, "id": row["id"]})
    elif new_mp > 0:
        run_id = db.execute(text("SELECT MAX(run_id) FROM forecasts")).scalar()
        db.execute(text("""
            INSERT INTO forecasts
                (run_id, product_id, year, week, baseline, planner_factor,
                 on_top_wholesale, on_top_retail, promo_uplift, total)
            VALUES (:run, :p, :y, :w, 0, 1, 0, :mp, 0, :mp)
        """), {"run": run_id, "p": product_id, "y": year, "w": week, "mp": new_mp})


def _apply_cell_edit(db: Session, *, product_id: int, buyer: str,
                     year_week: int, new_quantity: Optional[float],
                     new_year_week: Optional[int]) -> set[tuple[int, int]]:
    """Mutate on_top_inputs for one cell. Does NOT recompute forecasts or
    commit — the caller batches those so a multi-cell save recomputes each
    affected (product, week) only once. Returns the set of affected
    (product_id, year_week) pairs.

    new_quantity=None → keep current total (pure date move).
    new_quantity=0    → remove the commit.
    new_year_week=None → keep the same week (pure qty edit).
    """
    buyer_lc = buyer.strip().lower()

    # Source rows for this cell — across cycles, matching what the grid
    # aggregates. Preserve the first row's cycle/submitter on re-insert.
    src = db.execute(text("""
        SELECT id, cycle_id, submitted_by_id, quantity, buyer
        FROM on_top_inputs
        WHERE product_id = :p AND channel = 'wholesale'
          AND LOWER(TRIM(buyer)) = :blc AND year_week = :yw
        ORDER BY id
    """), {"p": product_id, "blc": buyer_lc, "yw": year_week}).mappings().all()
    if not src:
        raise ValueError(f"no on-top commit at cell product={product_id} CW{year_week % 100}")

    cur_qty = sum(float(r["quantity"]) for r in src)
    orig_cycle = src[0]["cycle_id"]
    orig_submitter = src[0]["submitted_by_id"]
    orig_buyer_str = src[0]["buyer"]

    eff_qty = float(new_quantity) if new_quantity is not None else cur_qty
    target_week = int(new_year_week) if new_year_week is not None else year_week
    affected = {year_week, target_week}

    db.execute(text("""
        DELETE FROM on_top_inputs
        WHERE product_id = :p AND channel = 'wholesale'
          AND LOWER(TRIM(buyer)) = :blc AND year_week = :yw
    """), {"p": product_id, "blc": buyer_lc, "yw": year_week})

    if eff_qty > 0:
        existing = db.execute(text("""
            SELECT id, quantity FROM on_top_inputs
            WHERE cycle_id = :cyc AND submitted_by_id = :sub
              AND channel = 'wholesale' AND LOWER(TRIM(buyer)) = :blc
              AND product_id = :p AND year_week = :tw
        """), {"cyc": orig_cycle, "sub": orig_submitter, "blc": buyer_lc,
                "p": product_id, "tw": target_week}).mappings().first()
        if existing:
            db.execute(text("UPDATE on_top_inputs SET quantity = :q WHERE id = :id"),
                       {"q": float(existing["quantity"]) + eff_qty if target_week != year_week else eff_qty,
                        "id": existing["id"]})
        else:
            db.execute(text("""
                INSERT INTO on_top_inputs
                    (cycle_id, product_id, year_week, quantity, channel,
                     buyer, submitted_by_id)
                VALUES (:cyc, :p, :tw, :q, 'wholesale', :buyer, :sub)
            """), {"cyc": orig_cycle, "p": product_id, "tw": target_week,
                    "q": eff_qty, "buyer": orig_buyer_str, "sub": orig_submitter})
    # Flush so a later edit in the same batch sees this mutation.
    db.flush()
    return {(product_id, yw) for yw in affected}


def edit_buyer_cell(db: Session, *, product_id: int, buyer: str,
                    year_week: int, new_quantity: Optional[float] = None,
                    new_year_week: Optional[int] = None,
                    editor_user_id: Optional[int] = None) -> dict:
    """Single-cell edit — thin wrapper over the batch path."""
    affected = _apply_cell_edit(db, product_id=product_id, buyer=buyer,
                                year_week=year_week, new_quantity=new_quantity,
                                new_year_week=new_year_week)
    for (pid, yw) in affected:
        _recompute_ws_forecast(db, pid, yw)
    db.commit()
    return {"product_id": product_id, "buyer": buyer,
            "from_year_week": year_week,
            "to_year_week": new_year_week or year_week,
            "new_quantity": new_quantity if new_quantity is not None else 0.0,
            "affected_weeks": sorted({yw for (_, yw) in affected})}


def edit_buyer_cells_batch(db: Session, edits: list[dict],
                           editor_user_id: Optional[int] = None) -> dict:
    """Apply a list of cell edits in ONE transaction, then recompute each
    affected (product, week) forecast exactly once. All-or-nothing: any
    bad edit raises and the whole batch rolls back."""
    affected: set[tuple[int, int]] = set()
    applied = 0
    try:
        for e in edits:
            aff = _apply_cell_edit(
                db,
                product_id=int(e["product_id"]),
                buyer=str(e["buyer"]),
                year_week=int(e["year_week"]),
                new_quantity=e.get("new_quantity"),
                new_year_week=e.get("new_year_week"),
            )
            affected |= aff
            applied += 1
        for (pid, yw) in affected:
            _recompute_ws_forecast(db, pid, yw)
        db.commit()
    except Exception:
        db.rollback()
        raise
    return {"n_applied": applied,
            "affected_weeks": sorted({yw for (_, yw) in affected})}


# ─────────────────────────────────────────────────────────────────────
# Buyer → ERP partner mapping (resolve the unmatched labels)
# ─────────────────────────────────────────────────────────────────────
def search_partners(db: Session, q: str, limit: int = 20) -> list[dict]:
    """Type-ahead search over dim_partners for the mapping dropdown.
    Ranks partners that actually have wholesale ERP transactions first."""
    like = f"%{q.strip().lower()}%"
    rows = db.execute(text("""
        SELECT dp.id, dp.name,
               COUNT(et.id) AS n_txn
        FROM dim_partners dp
        LEFT JOIN erp_transactions et
               ON et.partner_id = dp.id AND et.channel_map_id = ANY(:ws)
        WHERE LOWER(dp.name) LIKE :like
        GROUP BY dp.id, dp.name
        ORDER BY n_txn DESC, dp.name
        LIMIT :lim
    """), {"like": like, "ws": list(WS_CHANNEL_MAP_IDS), "lim": limit}).mappings().all()
    return [{"id": int(r["id"]), "name": r["name"], "n_txn": int(r["n_txn"])}
            for r in rows]


def upsert_buyer_alias(db: Session, *, buyer_lc: str,
                       partner_id: Optional[int],
                       editor_user_id: Optional[int] = None) -> dict:
    """Map (or unmap) a KAM buyer label to a dim_partners row.

    The mapping is now one-to-many (a buyer can point at several
    partners), so the UI 'set to this partner' semantic = replace all
    members with the single chosen partner. partner_id=None unmaps
    (removes all members). Multi-partner buckets like 'Ostalo' are seeded
    directly in the DB; the single-set UI would collapse them to one, so
    don't re-map those from the UI unless you mean to."""
    buyer_lc = buyer_lc.strip().lower()
    db.execute(text("DELETE FROM buyer_partner_aliases WHERE buyer_lc = :blc"),
               {"blc": buyer_lc})
    if partner_id is not None:
        db.execute(text("""
            INSERT INTO buyer_partner_aliases (buyer_lc, partner_id, updated_by, updated_at)
            VALUES (:blc, :pid, :uid, now())
        """), {"blc": buyer_lc, "pid": partner_id, "uid": editor_user_id})
    db.commit()
    name = None
    if partner_id is not None:
        name = db.execute(text("SELECT name FROM dim_partners WHERE id = :id"),
                          {"id": partner_id}).scalar()
    return {"buyer_lc": buyer_lc, "partner_id": partner_id, "partner_name": name}


# ─────────────────────────────────────────────────────────────────────
# Buyer commit accuracy — committed lift vs actual increment over baseline
# ─────────────────────────────────────────────────────────────────────
# Tracks how well a KAM's committed increment for a buyer materialized.
# The committed number is a LIFT ABOVE BASELINE (what the buyer already
# pulls), so we score it as:
#
#   increment = actual_pull − baseline_rate     (per closed week)
#   FA %      = increment / committed × 100      (100% = lift delivered)
#
# baseline_rate = that (partner, SKU) pair's trailing-13w average weekly
# pull, i.e. "what they were already buying" before the committed push.
# FA is only computable for CLOSED weeks (actual exists); future commit
# weeks are returned with actual/FA = null and is_future=True.

def report_buyer_fa(db: Session, *, kam: Optional[str] = None,
                    buyer: Optional[str] = None) -> dict:
    nav = _kam_buyer_nav(db)
    result: dict = {"kams": nav, "selected": None, "summary": None,
                    "articles": [], "baseline_weeks": BASELINE_WEEKS}
    if not buyer:
        return result

    buyer_lc = buyer.strip().lower()
    pmap = _buyer_partner_map(db)
    partner = pmap.get(buyer_lc, {})
    partner_ids = partner.get("partner_ids", [])   # one buyer can fan out to many
    result["selected"] = {"kam": kam, "buyer": buyer,
                          "partner_id": partner.get("partner_id"),
                          "partner_name": partner.get("partner_name"),
                          "matched": len(partner_ids) > 0}

    # Committed on-tops per (product, week) for this buyer
    params: dict = {"blc": buyer_lc}
    kam_clause = ""
    if kam:
        kam_clause = " AND COALESCE(NULLIF(u.display_name,''), u.username, '(unassigned)') = :kam"
        params["kam"] = kam
    commits = db.execute(text(f"""
        SELECT oti.product_id, p.sku, p.name AS sku_name, sp.tier,
               oti.year_week, SUM(oti.quantity)::float AS committed
        FROM on_top_inputs oti
        JOIN dim_products p ON p.id = oti.product_id
        LEFT JOIN sku_planning sp ON sp.product_id = oti.product_id
        LEFT JOIN users u ON u.id = oti.submitted_by_id
        WHERE oti.channel = 'wholesale'
          AND LOWER(TRIM(oti.buyer)) = :blc
          {kam_clause}
        GROUP BY oti.product_id, p.sku, p.name, sp.tier, oti.year_week
        ORDER BY p.sku, oti.year_week
    """), params).mappings().all()
    if not commits:
        result["summary"] = {"committed": 0, "baseline": 0, "actual": 0,
                             "increment": 0, "fa_pct": None, "n_articles": 0,
                             "n_weeks_scored": 0}
        return result

    product_ids = sorted({int(r["product_id"]) for r in commits})
    commit_yws  = sorted({int(r["year_week"]) for r in commits})

    today_iso = date.today().isocalendar()
    cur_yw = int(today_iso[0]) * 100 + int(today_iso[1])

    # Baseline: trailing-13w average weekly pull per (partner, product),
    # anchored to end BEFORE the earliest scored (closed) commit week —
    # i.e. "what they were buying BEFORE the KAM's push". Anchoring to
    # last-completed-week would let the committed lift itself leak into
    # the baseline and understate the measured increment.
    from datetime import date as _date, timedelta as _td
    base_window: dict = {"start": None, "end": None}
    baseline_map: dict[int, float] = {}
    actual_map: dict[tuple[int, int], float] = {}
    if partner_ids:
        closed_commit_yws = sorted(yw for yw in commit_yws if yw < cur_yw)
        if closed_commit_yws:
            ey, ew = closed_commit_yws[0] // 100, closed_commit_yws[0] % 100
            earliest_mon = _date.fromisocalendar(ey, ew, 1)
            base_end_d = earliest_mon - _td(days=1)        # Sunday before the earliest scored week
        else:
            # No closed commits to score → baseline anchor is moot; fall
            # back to last completed week so the field is still populated.
            base_end_d = date.today() - _td(days=date.today().weekday() + 1)
        base_start_d = base_end_d - _td(weeks=BASELINE_WEEKS - 1, days=6)
        base_window = {"start": base_start_d.isoformat(), "end": base_end_d.isoformat()}
        # Aggregate ERP pull across ALL of the buyer's partners (one buyer
        # label can fan out to several — e.g. Ostalo = 4 partners).
        b_rows = db.execute(text("""
            SELECT et.product_id,
                   SUM(et.quantity)::float / :bw AS baseline_rate
            FROM erp_transactions et
            WHERE et.partner_id = ANY(:partner_ids)
              AND et.channel_map_id = ANY(:ws)
              AND et.transaction_date BETWEEN DATE :bstart AND DATE :bend
              AND et.product_id = ANY(:pids)
            GROUP BY et.product_id
        """), {"partner_ids": partner_ids, "ws": list(WS_CHANNEL_MAP_IDS), "bw": BASELINE_WEEKS,
                "bstart": base_start_d.isoformat(), "bend": base_end_d.isoformat(),
                "pids": product_ids}).mappings().all()
        baseline_map = {int(r["product_id"]): float(r["baseline_rate"] or 0) for r in b_rows}

        a_rows = db.execute(text("""
            SELECT et.product_id,
                   EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100
                 + EXTRACT(WEEK    FROM et.transaction_date)::int AS yw,
                   SUM(et.quantity)::float AS qty
            FROM erp_transactions et
            WHERE et.partner_id = ANY(:partner_ids)
              AND et.channel_map_id = ANY(:ws)
              AND et.product_id = ANY(:pids)
              AND (EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100
                 + EXTRACT(WEEK FROM et.transaction_date)::int) = ANY(:yws)
            GROUP BY et.product_id,
                     EXTRACT(ISOYEAR FROM et.transaction_date)::int,
                     EXTRACT(WEEK FROM et.transaction_date)::int
        """), {"partner_ids": partner_ids, "ws": list(WS_CHANNEL_MAP_IDS),
                "pids": product_ids, "yws": commit_yws}).mappings().all()
        actual_map = {(int(r["product_id"]), int(r["yw"])): float(r["qty"]) for r in a_rows}

        # Cadence in the same trailing-13w pre-push window: distinct ISO weeks
        # this buyer pulled anything (across ALL SKUs) + their typical order
        # size on the weeks they did pull. Lets the planner read the baseline
        # rate correctly — a 100/wk baseline on a biweekly buyer means
        # ~200/order, and single-week scoring of a cadence buyer is noisy.
        cad = db.execute(text("""
            SELECT COUNT(DISTINCT EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100
                                + EXTRACT(WEEK FROM et.transaction_date)::int) AS wks,
                   SUM(et.quantity)::float AS total_qty
            FROM erp_transactions et
            WHERE et.partner_id = ANY(:partner_ids)
              AND et.channel_map_id = ANY(:ws)
              AND et.transaction_date BETWEEN DATE :bstart AND DATE :bend
              AND et.quantity > 0
        """), {"partner_ids": partner_ids, "ws": list(WS_CHANNEL_MAP_IDS),
                "bstart": base_start_d.isoformat(),
                "bend":   base_end_d.isoformat()}).first()
        weeks_pulled = int(cad[0] or 0)
        total_pulled = float(cad[1] or 0.0)
        typical_order = round(total_pulled / weeks_pulled, 1) if weeks_pulled > 0 else 0.0
        # Label is for human reading — based on how many of the 13 weeks had a
        # pull (cycle ≈ 13/k weeks).  ≥10 → ~weekly (rare misses), 5-9 → biweekly
        # (cycle 1.4-2.6 wks), 3-4 → monthly-ish (3-4 wk cycle), 1-2 → sporadic.
        if   weeks_pulled >= 10: cadence_label = "weekly"
        elif weeks_pulled >=  5: cadence_label = "biweekly"
        elif weeks_pulled >=  3: cadence_label = "monthly"
        elif weeks_pulled >=  1: cadence_label = "sporadic"
        else:                    cadence_label = "no history"
        result["selected"]["weeks_pulled"]      = weeks_pulled
        result["selected"]["typical_order_qty"] = typical_order
        result["selected"]["cadence_label"]     = cadence_label
        result["selected"]["baseline_weeks"]    = BASELINE_WEEKS

    def _fa(increment: float, committed: float) -> Optional[float]:
        if committed <= 0:
            return None
        return round(increment / committed * 100.0, 1)

    # Assemble per article → weeks
    by_sku: dict[int, dict] = {}
    for r in commits:
        pid = int(r["product_id"])
        yw = int(r["year_week"])
        committed = float(r["committed"])
        is_future = yw >= cur_yw
        baseline_rate = baseline_map.get(pid, 0.0)
        actual = actual_map.get((pid, yw))
        # increment + FA only for closed weeks with a matched partner
        if (not is_future) and partner_ids:
            actual_v = float(actual or 0.0)
            increment = actual_v - baseline_rate
            fa = _fa(increment, committed)
            scored = True
        else:
            actual_v = None
            increment = None
            fa = None
            scored = False

        slot = by_sku.setdefault(pid, {
            "product_id": pid, "sku": r["sku"], "name": r["sku_name"],
            "tier": r["tier"], "baseline_rate": round(baseline_rate, 1),
            "weeks": [], "committed": 0.0, "actual": 0.0,
            "baseline": 0.0, "increment": 0.0,
            "_scored_committed": 0.0, "_scored_increment": 0.0,
            "n_weeks_scored": 0,
        })
        slot["weeks"].append({
            "year_week": yw, "cw_label": f"CW{yw % 100:02d}",
            "committed": round(committed, 1),
            "baseline": round(baseline_rate, 1) if scored else None,
            "actual": round(actual_v, 1) if actual_v is not None else None,
            "increment": round(increment, 1) if increment is not None else None,
            "fa_pct": fa, "is_future": is_future, "scored": scored,
        })
        slot["committed"] += committed
        if scored:
            slot["actual"]    += actual_v
            slot["baseline"]  += baseline_rate
            slot["increment"] += increment
            slot["_scored_committed"] += committed
            slot["_scored_increment"] += increment
            slot["n_weeks_scored"] += 1

    articles = []
    for slot in by_sku.values():
        slot["fa_pct"] = _fa(slot["_scored_increment"], slot["_scored_committed"])
        for k in ("committed", "actual", "baseline", "increment"):
            slot[k] = round(slot[k], 1)
        slot.pop("_scored_committed"); slot.pop("_scored_increment")
        articles.append(slot)
    # Worst-first: lowest FA among scored articles surfaces on top
    articles.sort(key=lambda a: (a["fa_pct"] is None, a["fa_pct"] if a["fa_pct"] is not None else 1e9))

    sum_committed = sum(a["committed"] for a in articles)
    sum_scored_committed = sum(
        wk["committed"] for a in articles for wk in a["weeks"] if wk["scored"]
    )
    sum_increment = sum(a["increment"] for a in articles)
    sum_actual    = sum(a["actual"] for a in articles)
    sum_baseline  = sum(a["baseline"] for a in articles)
    n_weeks_scored = sum(a["n_weeks_scored"] for a in articles)

    result["summary"] = {
        "committed":      round(sum_committed, 1),
        "scored_committed": round(sum_scored_committed, 1),
        "baseline":       round(sum_baseline, 1),
        "actual":         round(sum_actual, 1),
        "increment":      round(sum_increment, 1),
        "fa_pct":         _fa(sum_increment, sum_scored_committed),
        "n_articles":     len(articles),
        "n_weeks_scored": n_weeks_scored,
    }
    result["articles"] = articles
    result["baseline_window"] = base_window   # {start, end} of the pre-push window
    return result


# ─────────────────────────────────────────────────────────────────────
# Availability (wholesale) — available-to-promise per buyer, per week
# ─────────────────────────────────────────────────────────────────────
# For a buyer's LISTED assortment (wholesale_listings), project how much
# of each SKU is free to promise THIS buyer each week:
#
#   running  = warehouse stock (wholesale ships from WH, not stores)
#   each week: running += incoming POs
#              demand_others = total forecast demand − this buyer's on-top
#              ATP = running − demand_others   (what's left after everyone else)
#              then running -= total demand    (carry forward)
#
# A week is "can't supply" when ATP ≤ 0 (nothing free) or, when the buyer
# has an on-top projection that week, when ATP < that projection (can't
# cover what they planned to order). Reuses the latest forecast run +
# trailing-13w fallback, same as the rest of the module.
AVAIL_HORIZON_WEEKS = 13


def _avail_horizon(today: Optional[date] = None) -> list[tuple[int, int]]:
    today = today or date.today()
    out: list[tuple[int, int]] = []
    cur = today - timedelta(days=today.weekday())   # Monday this week
    for _ in range(AVAIL_HORIZON_WEEKS):
        iso = cur.isocalendar()
        out.append((int(iso[0]), int(iso[1])))
        cur += timedelta(days=7)
    return out


def _listings_nav(db: Session) -> list[dict]:
    """KAM → buyers tree from wholesale_listings (with SKU counts)."""
    rows = db.execute(text("""
        SELECT kam, buyer, buyer_lc,
               COUNT(*)::int AS n_skus,
               COUNT(*) FILTER (WHERE product_id IS NULL)::int AS n_unmatched
        FROM wholesale_listings
        GROUP BY kam, buyer, buyer_lc
        ORDER BY kam, buyer
    """)).mappings().all()
    kams: dict[str, dict] = {}
    for r in rows:
        slot = kams.setdefault(r["kam"], {"kam": r["kam"], "buyers": []})
        slot["buyers"].append({
            "buyer": r["buyer"], "buyer_lc": r["buyer_lc"],
            "n_skus": int(r["n_skus"]), "n_unmatched": int(r["n_unmatched"]),
        })
    return list(kams.values())


def report_wholesale_availability(db: Session, *, kam: Optional[str] = None,
                                  buyer: Optional[str] = None,
                                  exclude_on_top: bool = False,
                                  commit_filter: str = "all") -> dict:
    """`commit_filter` restricts the listed assortment by what the buyer
    committed somewhere in the horizon:
        "all"    — every listed SKU (default)
        "on_top" — only SKUs with an on-top portion (lift above regular)
        "reg"    — only SKUs with a regular-increase portion
        "any"    — SKUs carrying EITHER (union: on-top OR regular-increase)
    """
    nav = _listings_nav(db)
    horizon = _avail_horizon()
    week_labels = [f"CW{w:02d}" for (_, w) in horizon]
    result: dict = {"kams": nav, "selected": None,
                    "weeks": week_labels,
                    "year_weeks": [y * 100 + w for (y, w) in horizon],
                    "rows": [], "n_gap_cells": 0}
    if not buyer:
        return result

    buyer_lc = buyer.strip().lower()
    # __ALL__ sentinel ⇒ aggregate across every buyer under this KAM. The math
    # is the same per-cell (ATP vs projection); we just expand the set of
    # buyers we treat as "this" — and dedupe the listed assortment so a SKU
    # carried by several buyers shows once with its summed projection.
    all_mode = buyer_lc in ('__all__', 'all')
    if all_mode and not kam:
        return result   # need a KAM to scope "all buyers"
    if all_mode:
        rows_b = db.execute(text(
            "SELECT DISTINCT buyer_lc FROM wholesale_listings WHERE kam = :kam"
        ), {"kam": kam}).all()
        buyer_lcs = [r[0] for r in rows_b]
        if not buyer_lcs:
            return result
    else:
        buyer_lcs = [buyer_lc]

    listed = db.execute(text("""
        SELECT DISTINCT ON (wl.product_id) wl.product_id, wl.sku, wl.rank,
               COALESCE(p.name, wl.sku) AS name, sp.tier
        FROM wholesale_listings wl
        LEFT JOIN dim_products p ON p.id = wl.product_id
        LEFT JOIN sku_planning sp ON sp.product_id = wl.product_id
        WHERE wl.kam = :kam AND wl.buyer_lc = ANY(:blcs)
        ORDER BY wl.product_id, wl.sku
    """), {"kam": kam, "blcs": buyer_lcs}).mappings().all()
    result["selected"] = {
        "kam": kam,
        "buyer": "All buyers" if all_mode else buyer,
        "n_listed": len(listed),
        "n_buyers": len(buyer_lcs) if all_mode else 1,
        "all_mode": all_mode,
    }
    if not listed:
        return result

    product_ids = sorted({int(r["product_id"]) for r in listed if r["product_id"]})
    yw_keys = [y * 100 + w for (y, w) in horizon]

    # Warehouse stock (wholesale supplies from WH only)
    wh = db.execute(text("""
        SELECT esc.product_id, SUM(esc.stock_qty)::float AS qty
        FROM erp_stock_current esc
        JOIN dim_stores ds ON ds.id = esc.store_id
        WHERE ds.is_warehouse = TRUE AND esc.product_id = ANY(:pids)
        GROUP BY esc.product_id
    """), {"pids": product_ids}).mappings().all() if product_ids else []
    wh_map = {int(r["product_id"]): float(r["qty"] or 0) for r in wh}

    # Forecast total per (pid, yw) — latest run
    fc = db.execute(text("""
        SELECT DISTINCT ON (product_id, year, week)
               product_id, year, week, COALESCE(total, 0)::float AS demand
        FROM forecasts
        WHERE product_id = ANY(:pids) AND (year * 100 + week) = ANY(:yws)
        ORDER BY product_id, year, week, run_id DESC
    """), {"pids": product_ids, "yws": yw_keys}).mappings().all() if product_ids else []
    fc_map = {(int(r["product_id"]), int(r["year"]) * 100 + int(r["week"])):
              float(r["demand"]) for r in fc}

    # Trailing-13w avg fallback per pid
    avg = db.execute(text("""
        WITH last13 AS (
            SELECT year, week FROM v_sales_weekly_full
            GROUP BY year, week ORDER BY year DESC, week DESC LIMIT 13
        )
        SELECT v.product_id, AVG(COALESCE(v.qty_total, 0))::float AS a
        FROM v_sales_weekly_full v JOIN last13 l USING (year, week)
        WHERE v.product_id = ANY(:pids)
        GROUP BY v.product_id
    """), {"pids": product_ids}).mappings().all() if product_ids else []
    avg_map = {int(r["product_id"]): float(r["a"] or 0) for r in avg}

    # Incoming per (pid, yw)
    inc = db.execute(text("""
        SELECT product_id, year, week, SUM(quantity)::float AS q
        FROM incoming_supply
        WHERE product_id = ANY(:pids) AND (year * 100 + week) = ANY(:yws)
        GROUP BY product_id, year, week
    """), {"pids": product_ids, "yws": yw_keys}).mappings().all() if product_ids else []
    inc_map = {(int(r["product_id"]), int(r["year"]) * 100 + int(r["week"])):
               float(r["q"]) for r in inc}

    # Buyer projection per (pid, yw), split into TOTAL and just the regular-
    # increase portion. The on-top portion = total - reg_inc. Lets the UI
    # toggle on-top in/out via `exclude_on_top` and filter to SKUs that carry
    # both portions via `commit_filter`.
    #
    # Scoped to the SELECTED KAM as well as the buyer: a buyer managed by two
    # KAMs (e.g. Spar = Patrik + Selma) must only show THIS KAM's commits, so
    # the on-top/regular-increase flags and the projection match the per-KAM
    # listing the grid is built from. Without a KAM we fall back to buyer-only.
    ot_params: dict = {"blcs": buyer_lcs, "yws": yw_keys}
    ot_kam_clause = ""
    if kam:
        ot_kam_clause = " AND COALESCE(NULLIF(u.display_name,''), u.username, '(unassigned)') = :kam"
        ot_params["kam"] = kam
    ot = db.execute(text(f"""
        SELECT oti.product_id, oti.year_week,
               SUM(oti.quantity)::float             AS q_total,
               SUM(oti.regular_increase_qty)::float AS q_reg
        FROM on_top_inputs oti
        LEFT JOIN users u ON u.id = oti.submitted_by_id
        WHERE oti.channel = 'wholesale' AND LOWER(TRIM(oti.buyer)) = ANY(:blcs)
          AND oti.year_week = ANY(:yws)
          {ot_kam_clause}
        GROUP BY oti.product_id, oti.year_week
    """), ot_params).mappings().all()
    ot_total_map = {(int(r["product_id"]), int(r["year_week"])): float(r["q_total"]) for r in ot}
    ot_reg_map   = {(int(r["product_id"]), int(r["year_week"])): float(r["q_reg"])   for r in ot}

    # Per-SKU set of buyers who committed (on-top or regular) in the horizon —
    # for the summary "affected buyers" column (matters in __ALL__ mode; in a
    # single-buyer view it's just that buyer).
    buyers_by_pid: dict[int, list[str]] = {}
    bp = db.execute(text(f"""
        SELECT oti.product_id, oti.buyer, SUM(oti.quantity)::float AS q
        FROM on_top_inputs oti
        LEFT JOIN users u ON u.id = oti.submitted_by_id
        WHERE oti.channel = 'wholesale' AND LOWER(TRIM(oti.buyer)) = ANY(:blcs)
          AND oti.year_week = ANY(:yws){ot_kam_clause}
        GROUP BY oti.product_id, oti.buyer
        ORDER BY q DESC
    """), ot_params).mappings().all()
    for r in bp:
        if r["product_id"] is not None:
            buyers_by_pid.setdefault(int(r["product_id"]), []).append(r["buyer"])

    all_rows = []
    for L in listed:
        pid = L["product_id"]
        running = wh_map.get(int(pid), 0.0) if pid else 0.0
        cells: dict[str, dict] = {}
        any_gap = False
        gap_count = 0
        has_on_top = has_reg_inc = False
        first_gap_yw: Optional[int] = None
        first_gap_cw: Optional[str] = None
        recovery_yw: Optional[int] = None       # week we can cover the buyer again
        recovery_cw: Optional[str] = None
        next_inc_yw: Optional[int] = None        # next delivery (PO) after the gap
        next_inc_cw: Optional[str] = None
        worst_shortfall = 0.0
        on_top_total = 0.0
        reg_total = 0.0
        for (y, w) in horizon:
            yw = y * 100 + w
            lbl = f"CW{w:02d}"
            demand_total = fc_map.get((pid, yw))
            if demand_total is None:
                demand_total = avg_map.get(int(pid), 0.0) if pid else 0.0
            incoming = inc_map.get((pid, yw), 0.0)
            proj_total = ot_total_map.get((pid, yw), 0.0)
            proj_reg   = ot_reg_map.get((pid, yw), 0.0)
            proj_on    = max(0.0, proj_total - proj_reg)
            if proj_on  > 0: has_on_top = True
            if proj_reg > 0: has_reg_inc = True
            # When the user toggles on-top OFF, only the regular-increase
            # portion of the buyer's projection counts. Demand on the rest
            # of the world (and therefore ATP) recomputes against that.
            proj = proj_reg if exclude_on_top else proj_total
            demand_others = max(0.0, demand_total - proj)
            running += incoming
            atp = running - demand_others                  # free for this buyer
            # "can't supply" = the buyer projected an on-top this week and we
            # don't have enough free to cover it. Weeks with no projection
            # aren't flagged (the buyer isn't asking for anything) — the ATP
            # number still shows the headroom.
            on_top_total += proj_on
            reg_total += proj_reg
            gap = proj > 0 and atp < proj
            if gap:
                any_gap = True
                gap_count += 1
                if first_gap_yw is None:
                    first_gap_yw, first_gap_cw = yw, lbl
                worst_shortfall = max(worst_shortfall, proj - max(0.0, atp))
            elif first_gap_yw is not None and recovery_yw is None and proj > 0:
                # first week after a gap where we can cover the buyer's order again
                recovery_yw, recovery_cw = yw, lbl
            if (first_gap_yw is not None and yw > first_gap_yw
                    and next_inc_yw is None and incoming > 0):
                next_inc_yw, next_inc_cw = yw, lbl    # next delivery after the gap
            cells[lbl] = {
                "year_week": yw,
                "atp": round(max(0.0, atp), 1),
                "projection":  round(proj, 1),
                "proj_on_top": round(proj_on, 1),
                "proj_reg":    round(proj_reg, 1),
                "gap": gap,
            }
            running = max(0.0, running - demand_total)     # carry forward
        all_rows.append({
            "product_id": int(pid) if pid else None,
            "sku": L["sku"], "name": L["name"], "tier": L["tier"],
            "rank": L["rank"], "wh_stock": round(wh_map.get(int(pid), 0.0) if pid else 0.0, 0),
            "any_gap": any_gap, "has_on_top": has_on_top, "has_reg_inc": has_reg_inc,
            "cells": cells, "_gap_count": gap_count,
            # summary fields (one-row-per-SKU view + Excel)
            "n_gap_weeks":    gap_count,
            "first_gap_cw":   first_gap_cw,
            "first_gap_yw":   first_gap_yw,
            # "available again": cover-buyer-again week, else next delivery (PO)
            "recovery_cw":    recovery_cw or next_inc_cw,
            "recovery_yw":    recovery_yw or next_inc_yw,
            "shortfall_units": round(worst_shortfall, 0),
            "on_top_units":   round(on_top_total, 0),
            "reg_units":      round(reg_total, 0),
            "buyers":         buyers_by_pid.get(int(pid), []) if pid else [],
        })

    # Filter the listed assortment by what the buyer committed.
    def _keep(r: dict) -> bool:
        if commit_filter == "on_top": return r["has_on_top"]
        if commit_filter == "reg":    return r["has_reg_inc"]
        if commit_filter == "any":    return r["has_on_top"] or r["has_reg_inc"]
        return True
    rows = [r for r in all_rows if _keep(r)]
    # n_gap counts only the SKU·week cells we still show, so it agrees with
    # the filtered grid.
    n_gap = sum(r.pop("_gap_count") for r in rows)
    # Gaps first, then by SKU
    rows.sort(key=lambda r: (not r["any_gap"], r["sku"]))
    result["rows"] = rows
    result["n_gap_cells"] = n_gap
    return result


def export_wholesale_availability_xlsx(
    db: Session, *, kam: Optional[str] = None, buyer: Optional[str] = None,
    exclude_on_top: bool = False, commit_filter: str = "all",
    only_gaps: bool = False,
) -> bytes:
    """Excel of the wholesale availability grid, mirroring the on-screen table:
    one row per listed SKU, one column per ISO week. Each week cell shows
    available-to-promise — and, where the buyer projected an order that week,
    "ATP / projection". Cells are colour-coded RED when we can't cover the
    buyer's projection (supply gap / OOS) and GREEN where a projection is
    fully covered. Per-SKU columns flag the commit type and the horizon-total
    on-top vs regular-increase units.

    Honours the same filters as the page: `only_gaps` keeps only SKUs with at
    least one short week, and `commit_filter` (all/on_top/reg/any) restricts
    by commit type."""
    import io
    from openpyxl import Workbook
    from openpyxl.styles import Font, PatternFill, Alignment

    rep = report_wholesale_availability(
        db, kam=kam, buyer=buyer,
        exclude_on_top=exclude_on_top, commit_filter=commit_filter,
    )
    weeks: list[str] = rep.get("weeks", [])
    rows: list[dict] = rep.get("rows", [])
    if only_gaps:
        rows = [r for r in rows if r.get("any_gap")]

    # Excel's standard good/bad conditional-format palette.
    RED_FILL   = PatternFill("solid", fgColor="FFC7CE")
    RED_FONT   = Font(color="9C0006")
    GREEN_FILL = PatternFill("solid", fgColor="C6EFCE")
    GREEN_FONT = Font(color="006100")
    HEADER_FONT = Font(bold=True)

    wb = Workbook()
    ws = wb.active
    ws.title = "Availability WS"
    LEAD = ["SKU", "Name", "Rank", "Commit type", "On-top (u)", "Reg.incr (u)", "WH stock"]
    ws.append([*LEAD, *weeks])

    for r in rows:
        cells = r.get("cells", {}) or {}
        has_ot = bool(r.get("has_on_top"))
        has_rg = bool(r.get("has_reg_inc"))
        commit_type = ("On-top + Reg" if has_ot and has_rg
                       else "On-top" if has_ot
                       else "Reg increase" if has_rg
                       else "—")
        on_top_u = round(sum((c.get("proj_on_top") or 0.0) for c in cells.values()), 1)
        reg_u    = round(sum((c.get("proj_reg")    or 0.0) for c in cells.values()), 1)
        line: list = [r.get("sku"), r.get("name"), r.get("rank"), commit_type,
                      on_top_u, reg_u, r.get("wh_stock")]
        for w in weeks:
            c = cells.get(w)
            if not c:
                line.append("")
                continue
            atp = round(c.get("atp") or 0.0, 1)
            proj = round(c.get("projection") or 0.0, 1)
            # Mirror the table: show "ATP / projection" when the buyer ordered
            # that week, otherwise just the ATP headroom.
            line.append(f"{atp:g} / {proj:g}" if proj > 0 else atp)
        row_idx = ws.max_row + 1
        ws.append(line)
        # Colour the week cells (cols after the lead block).
        for j, w in enumerate(weeks):
            c = cells.get(w)
            if not c:
                continue
            cell = ws.cell(row=row_idx, column=len(LEAD) + 1 + j)
            cell.alignment = Alignment(horizontal="right")
            if c.get("gap"):
                cell.fill, cell.font = RED_FILL, RED_FONT          # can't supply
            elif (c.get("projection") or 0.0) > 0:
                cell.fill, cell.font = GREEN_FILL, GREEN_FONT      # ordered & covered

    for cell in ws[1]:
        cell.font = HEADER_FONT
    # Sensible widths + freeze the lead block and header.
    widths = {"A": 12, "B": 34, "C": 8, "D": 13, "E": 11, "F": 12, "G": 10}
    for col, wdt in widths.items():
        ws.column_dimensions[col].width = wdt
    ws.freeze_panes = "H2"

    bio = io.BytesIO()
    wb.save(bio)
    return bio.getvalue()


def export_wholesale_availability_summary_xlsx(
    db: Session, *, kam: Optional[str] = None, buyer: Optional[str] = None,
    exclude_on_top: bool = False, commit_filter: str = "all",
    only_gaps: bool = False,
) -> bytes:
    """Simple one-row-per-SKU summary Excel for a KAM: which listed articles
    won't be available for their buyer(s) — earliest short week, units short,
    affected buyers, and the committed on-top (big buys) highlighted. At-risk
    rows first and tinted red."""
    import io
    from openpyxl import Workbook
    from openpyxl.styles import Font, PatternFill, Alignment

    rep = report_wholesale_availability(
        db, kam=kam, buyer=buyer,
        exclude_on_top=exclude_on_top, commit_filter=commit_filter,
    )
    rows: list[dict] = rep.get("rows", [])
    if only_gaps:
        rows = [r for r in rows if r.get("any_gap")]

    RED_FILL   = PatternFill("solid", fgColor="FFC7CE")
    RED_FONT   = Font(color="9C0006")
    OT_FILL    = PatternFill("solid", fgColor="FFF2CC")   # highlight on-top (big buys)
    HEADER_FONT = Font(bold=True)

    wb = Workbook()
    ws = wb.active
    ws.title = "KAM availability"
    sel = rep.get("selected") or {}
    ws.append([f"KAM: {sel.get('kam') or '—'}   Kupac: {sel.get('buyer') or '—'}"])
    ws.append([])
    HEAD = ["SKU", "Naziv", "Rang", "Status", "Bez zalihe od", "Ponovno dostupno od",
            "Manjak (kom)", "On-top (kom)", "Reg. povećanje (kom)",
            "Tjedana u riziku", "WH zaliha", "Kupci"]
    ws.append(HEAD)

    for r in rows:
        at_risk = bool(r.get("any_gap"))
        on_top_u = float(r.get("on_top_units") or 0)
        line = [
            r.get("sku"), r.get("name"), r.get("rank"),
            "RIZIK" if at_risk else "OK",
            r.get("first_gap_cw") or "—",
            r.get("recovery_cw") or "—",
            r.get("shortfall_units") or 0,
            on_top_u, r.get("reg_units") or 0,
            r.get("n_gap_weeks") or 0,
            r.get("wh_stock") or 0,
            ", ".join(r.get("buyers") or []),
        ]
        ridx = ws.max_row + 1
        ws.append(line)
        if at_risk:
            for col in range(1, len(HEAD) + 1):
                ws.cell(row=ridx, column=col).fill = RED_FILL
                ws.cell(row=ridx, column=col).font = RED_FONT
        if on_top_u > 0:                       # flag the big-buy commits (On-top col = 8)
            ws.cell(row=ridx, column=8).fill = OT_FILL
        for col in (7, 8, 9, 10, 11):
            ws.cell(row=ridx, column=col).alignment = Alignment(horizontal="right")

    for cell in ws[3]:
        cell.font = HEADER_FONT
    widths = {"A": 12, "B": 36, "C": 7, "D": 8, "E": 14, "F": 18, "G": 12,
              "H": 12, "I": 17, "J": 16, "K": 11, "L": 30}
    for col, wdt in widths.items():
        ws.column_dimensions[col].width = wdt
    ws.freeze_panes = "A4"

    bio = io.BytesIO()
    wb.save(bio)
    return bio.getvalue()