"""Executive Dashboard service — read-only overview for CEO / CFO / SCM Director.

One endpoint (`/api/executive/dashboard`) calls every section. Each section
returns its own block and is wrapped in try/except so a single broken query
can't take down the whole page.

Numbers come from the same physical tables the operational modules use —
no separate datamart. Inventory uses `erp_stock_current`, revenue uses
`erp_transactions` × `lookup_channel_map`, demand uses `forecasts`, accuracy
uses `backtest_results`.
"""
from __future__ import annotations

from datetime import date, datetime, timedelta
from typing import Optional

import pandas as pd
from sqlalchemy import text
from sqlalchemy.orm import Session

from backend.services.coverage_classifier import classify_coverage


# Board target for total WH+store inventory value (purchase cost basis).
INVENTORY_TARGET_EUR = 5_000_000.0

# Suppliers flagged as strategic / private-label upstream — surfaced in the
# Supply Arrivals card with a distinct badge.
ABC_SUPPLIER_KEYS = ("abc nutritional", "abc nutrition", "abc")

STOCKOUT_TIERS = ("01 GOLD", "02 SILVER", "03 BRONZE")
TIER_DISPLAY = {"01 GOLD": "Gold", "02 SILVER": "Silver", "03 BRONZE": "Bronze"}
STOCKOUT_LIMIT = 10


# ─────────────────────────────────────────────────────────────────────────
# Helpers
# ─────────────────────────────────────────────────────────────────────────
def _iso_week_now(db: Session) -> tuple[int, int]:
    r = db.execute(text(
        "SELECT EXTRACT(ISOYEAR FROM now())::int AS y, "
        "       EXTRACT(WEEK    FROM now())::int AS w"
    )).mappings().first()
    return int(r["y"]), int(r["w"])


def _iso_week_first_monday(year: int, week: int) -> date:
    """Return the Monday of the given ISO year-week."""
    return datetime.strptime(f"{year}-W{week:02d}-1", "%G-W%V-%u").date()


def _last_completed_iso_week() -> tuple[int, int]:
    """ISO year/week most recently fully completed (i.e. last Sunday's week)."""
    today = date.today()
    # Last Sunday — if today IS Sunday, count today's week as completed.
    days_since_sun = (today.weekday() + 1) % 7
    last_sun = today - timedelta(days=days_since_sun)
    iso = last_sun.isocalendar()
    return int(iso[0]), int(iso[1])


def _last_iso_week_with_data(db: Session) -> Optional[tuple[int, int]]:
    """Most recent ISO (year, week) that has at least one row in
    erp_transactions. Used to anchor the dashboard's "last week" to data
    we actually have, not what the calendar says. Returns None on empty DB."""
    r = db.execute(text("""
        SELECT EXTRACT(ISOYEAR FROM max_dt)::int AS y,
               EXTRACT(WEEK    FROM max_dt)::int AS w
        FROM (SELECT MAX(transaction_date) AS max_dt FROM erp_transactions) x
        WHERE max_dt IS NOT NULL
    """)).first()
    return (int(r[0]), int(r[1])) if r and r[0] is not None else None


def _last_completed_week_for_revenue(db: Session) -> tuple[int, int]:
    """Pick the earlier of (calendar-last-completed-week, last-week-with-data).
    If data is fresh through this Sunday, returns the calendar week. If data
    only goes through last Wednesday (e.g. CW20 partial), we still pick CW20
    if it's the latest fully-loaded week.

    Empty DB → returns calendar week (lets the empty-state messaging fire)."""
    cal_y, cal_w = _last_completed_iso_week()
    data_yw = _last_iso_week_with_data(db)
    if data_yw is None:
        return cal_y, cal_w
    data_y, data_w = data_yw
    # Compare via year*100+week — works for normal ranges.
    if data_y * 100 + data_w < cal_y * 100 + cal_w:
        return data_y, data_w
    return cal_y, cal_w


def _shift_iso_week(year: int, week: int, delta: int) -> tuple[int, int]:
    mon = _iso_week_first_monday(year, week) + timedelta(weeks=delta)
    iso = mon.isocalendar()
    return int(iso[0]), int(iso[1])


def _yw_label(year: int, week: int) -> str:
    return f"{year}-W{week:02d}"


# ─────────────────────────────────────────────────────────────────────────
# Section 1 — Inventory snapshot
# ─────────────────────────────────────────────────────────────────────────
def get_inventory_snapshot(db: Session) -> dict:
    """Current WH + store inventory value at cost. WoW comparison if a
    historical snapshot table exists — otherwise prev_week stays None."""
    r = db.execute(text("""
        SELECT
            COALESCE(SUM(CASE WHEN ds.is_warehouse
                              THEN esc.stock_qty * COALESCE(ec.cost_price, 0)
                              ELSE 0 END), 0)::float AS wh_eur,
            COALESCE(SUM(CASE WHEN NOT ds.is_warehouse
                              THEN esc.stock_qty * COALESCE(ec.cost_price, 0)
                              ELSE 0 END), 0)::float AS store_eur,
            COUNT(DISTINCT esc.product_id) AS n_skus
        FROM erp_stock_current esc
        JOIN dim_stores  ds ON ds.id = esc.store_id
        LEFT JOIN erp_costs ec ON ec.product_id = esc.product_id
        WHERE esc.stock_qty > 0
    """)).mappings().first()

    wh = float(r["wh_eur"] or 0.0)
    store = float(r["store_eur"] or 0.0)
    total = wh + store

    # Historical snapshot — we don't yet keep weekly stock snapshots, so
    # prev_week stays None until that job exists. UI handles missing values.
    prev_week_total: Optional[float] = None
    wow_change_eur: Optional[float] = None
    wow_change_pct: Optional[float] = None

    vs_target_pct = ((total - INVENTORY_TARGET_EUR) / INVENTORY_TARGET_EUR
                     * 100.0) if INVENTORY_TARGET_EUR else 0.0

    return {
        "total_inventory_eur": total,
        "total_inventory_eur_prev_week": prev_week_total,
        "wow_change_eur": wow_change_eur,
        "wow_change_pct": wow_change_pct,
        "target_eur": INVENTORY_TARGET_EUR,
        "vs_target_pct": vs_target_pct,
        "wh_eur": wh,
        "store_eur": store,
        "n_skus_with_stock": int(r["n_skus"] or 0),
    }


# ─────────────────────────────────────────────────────────────────────────
# Section 2 — Revenue pulse
# ─────────────────────────────────────────────────────────────────────────
def get_revenue_pulse(db: Session) -> dict:
    """MTD revenue and WoW deltas per channel (retail / wholesale / webshop).

    `lookup_channel_map.channel` is the canonical channel value; we join
    on `erp_transactions.channel_map_id`. Revenue = NET of VAT (canonical):
    COALESCE(NULLIF(tax_base,0), total_value*0.80), already net of returns
    where the doc type is negative.
    """
    today = date.today()
    month_start = today.replace(day=1)

    # Anchor "last week" to the most recent ISO week that has data — falls
    # back to calendar last-completed-week when DB is fresh. Without this
    # we'd show €0 "last week revenue" any time the data import lags the
    # calendar (e.g. Sunday dashboards before Monday morning's sales pull).
    last_y, last_w = _last_completed_week_for_revenue(db)
    prev_y, prev_w = _shift_iso_week(last_y, last_w, -1)

    last_w_start = _iso_week_first_monday(last_y, last_w)
    last_w_end = last_w_start + timedelta(days=6)
    prev_w_start = _iso_week_first_monday(prev_y, prev_w)
    prev_w_end = prev_w_start + timedelta(days=6)

    # MTD by channel.
    mtd_rows = db.execute(text("""
        SELECT cm.channel,
               COALESCE(SUM(COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80)), 0)::float AS rev
        FROM erp_transactions et
        JOIN lookup_channel_map cm ON cm.id = et.channel_map_id
        WHERE et.transaction_date >= :ms
          AND et.transaction_date <= :today
        GROUP BY cm.channel
    """), {"ms": month_start, "today": today}).mappings().all()

    # Last completed week + week before, by channel.
    week_rows = db.execute(text("""
        SELECT cm.channel,
               SUM(CASE WHEN et.transaction_date BETWEEN :lws AND :lwe
                        THEN COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80) ELSE 0 END)::float AS last_w,
               SUM(CASE WHEN et.transaction_date BETWEEN :pws AND :pwe
                        THEN COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80) ELSE 0 END)::float AS prev_w
        FROM erp_transactions et
        JOIN lookup_channel_map cm ON cm.id = et.channel_map_id
        WHERE et.transaction_date BETWEEN :pws AND :lwe
        GROUP BY cm.channel
    """), {
        "lws": last_w_start, "lwe": last_w_end,
        "pws": prev_w_start, "pwe": prev_w_end,
    }).mappings().all()

    channels = ("retail", "wholesale", "webshop")
    mtd_map = {r["channel"]: float(r["rev"] or 0) for r in mtd_rows}
    week_map = {r["channel"]: (float(r["last_w"] or 0), float(r["prev_w"] or 0))
                for r in week_rows}

    by_channel = []
    mtd_total = last_w_total = prev_w_total = 0.0
    for ch in channels:
        mtd = mtd_map.get(ch, 0.0)
        last_w_val, prev_w_val = week_map.get(ch, (0.0, 0.0))
        wow_pct = ((last_w_val - prev_w_val) / prev_w_val * 100.0
                   if prev_w_val > 0 else None)
        by_channel.append({
            "channel": ch,
            "mtd_revenue": mtd,
            "last_week_revenue": last_w_val,
            "prev_week_revenue": prev_w_val,
            "wow_change_pct": wow_pct,
        })
        mtd_total += mtd
        last_w_total += last_w_val
        prev_w_total += prev_w_val

    wow_total_pct = ((last_w_total - prev_w_total) / prev_w_total * 100.0
                     if prev_w_total > 0 else None)

    return {
        "month_label": month_start.strftime("%B %Y"),
        "last_week_label": _yw_label(last_y, last_w),
        "prev_week_label": _yw_label(prev_y, prev_w),
        "mtd_revenue_total": mtd_total,
        "last_week_total": last_w_total,
        "prev_week_total": prev_w_total,
        "wow_change_pct": wow_total_pct,
        "by_channel": by_channel,
    }


def get_revenue_weekly(db: Session, weeks: int = 4) -> list[dict]:
    """Per-channel revenue stacked bars for the last N completed ISO weeks
    (anchored to last week with data — see _last_completed_week_for_revenue)."""
    last_y, last_w = _last_completed_week_for_revenue(db)
    earliest_y, earliest_w = _shift_iso_week(last_y, last_w, -(weeks - 1))
    start = _iso_week_first_monday(earliest_y, earliest_w)
    end = _iso_week_first_monday(last_y, last_w) + timedelta(days=6)

    rows = db.execute(text("""
        SELECT EXTRACT(ISOYEAR FROM et.transaction_date)::int AS y,
               EXTRACT(WEEK    FROM et.transaction_date)::int AS w,
               cm.channel,
               COALESCE(SUM(COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80)), 0)::float AS rev,
               COALESCE(SUM(et.ruc_eur),     0)::float AS ruc
        FROM erp_transactions et
        JOIN lookup_channel_map cm ON cm.id = et.channel_map_id
        WHERE et.transaction_date BETWEEN :s AND :e
        GROUP BY y, w, cm.channel
        ORDER BY y, w
    """), {"s": start, "e": end}).mappings().all()

    bucket: dict[tuple[int, int], dict[str, float]] = {}
    for r in rows:
        key = (int(r["y"]), int(r["w"]))
        bucket.setdefault(key, {
            "retail": 0, "wholesale": 0, "webshop": 0,
            "ruc_retail": 0, "ruc_wholesale": 0, "ruc_webshop": 0,
        })
        ch = r["channel"]
        bucket[key][ch]            = float(r["rev"] or 0)
        bucket[key][f"ruc_{ch}"]   = float(r["ruc"] or 0)

    out: list[dict] = []
    cur_y, cur_w = earliest_y, earliest_w
    empty = {
        "retail": 0, "wholesale": 0, "webshop": 0,
        "ruc_retail": 0, "ruc_wholesale": 0, "ruc_webshop": 0,
    }
    for _ in range(weeks):
        b = bucket.get((cur_y, cur_w), empty)
        out.append({
            "year_week": _yw_label(cur_y, cur_w),
            "retail":         b["retail"],
            "wholesale":      b["wholesale"],
            "webshop":        b["webshop"],
            "total":          b["retail"] + b["wholesale"] + b["webshop"],
            "ruc_retail":     b["ruc_retail"],
            "ruc_wholesale":  b["ruc_wholesale"],
            "ruc_webshop":    b["ruc_webshop"],
            "ruc_total":      b["ruc_retail"] + b["ruc_wholesale"] + b["ruc_webshop"],
        })
        cur_y, cur_w = _shift_iso_week(cur_y, cur_w, 1)
    return out


# ─────────────────────────────────────────────────────────────────────────
# Plan compare — monthly snapshot vs MTD actuals
# ─────────────────────────────────────────────────────────────────────────
def get_plan_compare(db: Session) -> dict:
    """Pull current-month snapshot (if locked) and compare to MTD actuals.

    Returns plan totals, MTD totals, and progress %s. When no snapshot is
    locked for the current month, plan_source='no_snapshot' and plan_*
    fields are None — frontend renders an "unlocked" state."""
    from datetime import date as _date
    from calendar import monthrange

    today = _date.today()
    cy, cm = today.year, today.month
    month_key = cy * 100 + cm
    month_label = today.strftime("%B %Y")
    days_in_month = monthrange(cy, cm)[1]
    days_observed = today.day

    snap = db.execute(text("""
        SELECT total_revenue_eur::float AS rev,
               total_ruc_eur::float     AS ruc
        FROM monthly_plan_snapshots
        WHERE month_key = :mk
    """), {"mk": month_key}).mappings().first()

    mtd = db.execute(text("""
        SELECT
          COALESCE(SUM(COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80)), 0)::float AS rev,
          COALESCE(SUM(et.ruc_eur),     0)::float AS ruc
        FROM erp_transactions et
        WHERE EXTRACT(YEAR  FROM et.transaction_date) = :y
          AND EXTRACT(MONTH FROM et.transaction_date) = :m
    """), {"y": cy, "m": cm}).mappings().first()

    plan_rev = float(snap["rev"]) if snap else None
    plan_ruc = float(snap["ruc"]) if snap else None
    mtd_rev  = float(mtd["rev"] or 0) if mtd else 0.0
    mtd_ruc  = float(mtd["ruc"] or 0) if mtd else 0.0

    def _pct(actual: float, plan: Optional[float]) -> Optional[float]:
        if plan is None or plan <= 0:
            return None
        return round(actual / plan * 100.0, 1)

    return {
        "month_key":            month_key,
        "month_label":          month_label,
        "plan_source":          "snapshot" if snap else "no_snapshot",
        "plan_revenue_eur":     plan_rev,
        "plan_ruc_eur":         plan_ruc,
        "mtd_revenue_eur":      mtd_rev,
        "mtd_ruc_eur":          mtd_ruc,
        "revenue_progress_pct": _pct(mtd_rev, plan_rev),
        "ruc_progress_pct":     _pct(mtd_ruc, plan_ruc),
        "days_in_month":        days_in_month,
        "days_observed":        days_observed,
    }


# ─────────────────────────────────────────────────────────────────────────
# Section 3 — Stockout risks
# ─────────────────────────────────────────────────────────────────────────
def get_stockout_risks(db: Session, limit: int = STOCKOUT_LIMIT) -> list[dict]:
    """Top Gold/Silver/Bronze SKUs with <3 weeks WH cover, sorted by weekly
    revenue at risk. Weekly demand uses **the next-13-week forecast average**
    for SKUs in `forecasts`, falling back to **13-week non-promo run-rate**
    for any planned SKU not yet in the live run. Aligns with Supply Coverage
    and Finance fallback so the same SKU has the same weekly_demand on every
    page."""
    cur_y, cur_w = _iso_week_now(db)
    horizon = [_shift_iso_week(cur_y, cur_w, i) for i in range(13)]
    horizon_yw = [y * 100 + w for (y, w) in horizon]

    # Realized per-unit price over the last 13 weeks from rekapitulacija
    # (`Vrijednost` = erp_transactions.total_value). One source of truth —
    # the same number Executive Revenue Pulse reports, so "revenue at risk"
    # reconciles with collected revenue.
    rows = pd.read_sql(text("""
        WITH
        wh AS (
            SELECT esc.product_id,
                   SUM(esc.stock_qty)::float AS wh_qty
            FROM erp_stock_current esc
            JOIN dim_stores ds ON ds.id = esc.store_id
            WHERE ds.is_warehouse
            GROUP BY esc.product_id
        ),
        fc AS (
            SELECT product_id,
                   AVG(COALESCE(total, 0))::float AS avg_demand
            FROM forecasts
            WHERE run_id = (SELECT MAX(run_id) FROM forecasts)
              AND year*100 + week = ANY(:yws)
            GROUP BY product_id
        ),
        run_rate AS (
            SELECT v.product_id,
                   COALESCE(
                       AVG(v.qty_total) FILTER (
                           WHERE NOT COALESCE(epw.is_erp_promo, FALSE)
                       ),
                       AVG(v.qty_total)
                   )::float AS avg_qty
            FROM v_sales_weekly_full v
            LEFT JOIN erp_promo_weeks epw
                ON epw.product_id = v.product_id
               AND epw.year       = v.year
               AND epw.week       = v.week
            WHERE v.year*100 + v.week >= :avg_cutoff_yw
            GROUP BY v.product_id
        ),
        realized_price AS (
            SELECT et.product_id,
                   SUM(COALESCE(NULLIF(et.tax_base,0), et.total_value*0.80))::float AS rev,
                   SUM(et.quantity)::float    AS qty
            FROM erp_transactions et
            WHERE et.transaction_date >= now() - INTERVAL '13 weeks'
            GROUP BY et.product_id
        ),
        incoming_in_lt AS (
            -- Sum incoming POs landing inside each SKU's lead-time window.
            -- We use the supply_master.lead_time_weeks per SKU and only count
            -- arrivals from current week up to (current_week + ceil(lt)).
            -- Filter NaN/NULL/<=0 lead times out — those SKUs get 0 via
            -- LEFT JOIN downstream and classify as MISSING_LT.
            SELECT inc.product_id, SUM(inc.quantity)::float AS qty
            FROM incoming_supply inc
            JOIN supply_master sm ON sm.product_id = inc.product_id
            WHERE inc.quantity > 0
              AND sm.lead_time_weeks IS NOT NULL
              AND sm.lead_time_weeks::text <> 'NaN'
              AND sm.lead_time_weeks > 0
              AND inc.year*100 + inc.week >= :cur_yw
              AND inc.year*100 + inc.week <  :cur_yw + CEIL(sm.lead_time_weeks)::int + 1
            GROUP BY inc.product_id
        ),
        nextpo AS (
            SELECT product_id, MIN(year*100 + week) AS next_yw
            FROM incoming_supply
            WHERE year*100 + week >= :cur_yw
              AND quantity > 0
            GROUP BY product_id
        )
        SELECT p.sku, p.name,
               sp.tier,
               COALESCE(c.name, '') AS category,
               COALESCE(wh.wh_qty, 0)        AS wh_stock,
               COALESCE(NULLIF(fc.avg_demand, 0), run_rate.avg_qty, 0) AS weekly_demand,
               CASE
                   WHEN COALESCE(rp.qty, 0) > 0 THEN rp.rev / rp.qty
                   ELSE 0
               END AS sell_price,
               sm.lead_time_weeks::float                  AS lead_time_weeks,
               COALESCE(inc_lt.qty, 0)::float             AS incoming_in_lt,
               nextpo.next_yw
        FROM dim_products p
        JOIN sku_planning sp        ON sp.product_id = p.id
        LEFT JOIN supply_master sm  ON sm.product_id = p.id
        LEFT JOIN dim_categories c  ON c.id = p.category_id
        LEFT JOIN wh                ON wh.product_id = p.id
        LEFT JOIN fc                ON fc.product_id = p.id
        LEFT JOIN run_rate          ON run_rate.product_id = p.id
        LEFT JOIN realized_price rp ON rp.product_id = p.id
        LEFT JOIN incoming_in_lt inc_lt ON inc_lt.product_id = p.id
        LEFT JOIN nextpo            ON nextpo.product_id = p.id
        WHERE sp.tier IN ('01 GOLD', '02 SILVER', '03 BRONZE')
          AND COALESCE(NULLIF(fc.avg_demand, 0), run_rate.avg_qty, 0) > 0
    """), db.bind, params={
        "yws": horizon_yw,
        "cur_yw": cur_y * 100 + cur_w,
        "avg_cutoff_yw": _shift_iso_week(cur_y, cur_w, -13)[0] * 100
                         + _shift_iso_week(cur_y, cur_w, -13)[1],
    })

    if rows.empty:
        return []

    # ── Trigger logic (Polleo spec, May 2026 refactor) ────────────────────
    # Alert when BOTH:
    #   1. bare cover (wh_stock / weekly_demand) < lead_time_weeks
    #   2. no incoming PO lands WITHIN the stock-consumption window
    #      (i.e. before stock physically runs out)
    #
    # Versus the previous classify_coverage rule which folded all incoming-
    # within-LT into the cover number and shaded results into 4 buckets.
    # The new rule is a binary "this WILL stockout" filter — only surface
    # SKUs where existing incoming POs definitively won't arrive in time.
    at_risk_rows = []
    for r in rows.itertuples():
        wh = float(r.wh_stock or 0)
        wd = float(r.weekly_demand or 0)
        lt = float(r.lead_time_weeks) if pd.notna(r.lead_time_weeks) else None
        if wd <= 0 or lt is None or lt <= 0:
            continue
        cover = wh / wd
        if cover >= lt:
            # Stock will outlast lead time — not at risk.
            continue

        # Weeks until next PO from current ISO week (approx via *52 — good
        # enough for short horizons that don't cross multiple year boundaries).
        next_yw = getattr(r, "next_yw", None)
        if next_yw is None or pd.isna(next_yw):
            po_saves = False
        else:
            ny, nw = int(next_yw) // 100, int(next_yw) % 100
            weeks_to_po = (ny - cur_y) * 52 + (nw - cur_w)
            # PO "saves" the stockout if it lands at or before the
            # consumption-end week. Past POs (weeks_to_po < 0) shouldn't
            # appear here since we only pull yw >= cur_yw, but defensive.
            po_saves = 0 <= weeks_to_po <= cover

        if po_saves:
            continue

        # Effective cover here equals bare cover — no in-time PO buffer.
        at_risk_rows.append({
            "sku": r.sku, "name": r.name, "tier": r.tier, "category": r.category,
            "wh_stock": wh,
            "weekly_demand": wd,
            "weeks_cover": cover,
            "effective_weeks_cover": cover,
            "status": "CRITICAL",
            "sell_price": float(r.sell_price or 0),
            "revenue_at_risk_weekly": wd * float(r.sell_price or 0),
            "incoming_in_lt": float(r.incoming_in_lt or 0),
            "lead_time_weeks": lt,
            "next_yw": next_yw,
        })

    if not at_risk_rows:
        return []

    at_risk = pd.DataFrame(at_risk_rows)
    at_risk["days_until_stockout"] = (at_risk["effective_weeks_cover"] * 7).astype(float)
    at_risk = at_risk.sort_values(
        "revenue_at_risk_weekly", ascending=False).head(limit)

    out: list[dict] = []
    for r in at_risk.itertuples():
        next_yw_label: Optional[str] = None
        if pd.notna(getattr(r, "next_yw", None)):
            yw = int(r.next_yw)
            next_yw_label = f"{yw // 100}-W{yw % 100:02d}"
        out.append({
            "sku": r.sku,
            "name": r.name,
            "tier": TIER_DISPLAY.get(r.tier, r.tier),
            "category": r.category,
            "wh_stock": round(float(r.wh_stock), 1),
            "weekly_demand": round(float(r.weekly_demand), 2),
            "weeks_cover": round(float(r.weeks_cover), 2),
            "effective_weeks_cover": round(float(r.effective_weeks_cover), 2),
            "lead_time_weeks": round(float(r.lead_time_weeks), 1),
            "incoming_in_lt": round(float(r.incoming_in_lt), 0),
            "status": r.status,
            "revenue_at_risk_weekly": round(float(r.revenue_at_risk_weekly), 2),
            "incoming_po_week": next_yw_label,
            "days_until_stockout": round(float(r.days_until_stockout), 1),
        })
    return out


# ─────────────────────────────────────────────────────────────────────────
# Section 4 — Supply arrivals (this + next week, by supplier)
# ─────────────────────────────────────────────────────────────────────────
def get_supply_arrivals(db: Session) -> dict:
    """PO arrivals grouped by supplier for the current ISO week and next.

    `incoming_supply.status` is currently 'NaN' for every row in this dataset
    so overdue detection can't be derived from a status field — overdue is
    inferred as PO weeks earlier than the current week.
    """
    cur_y, cur_w = _iso_week_now(db)
    next_y, next_w = _shift_iso_week(cur_y, cur_w, 1)
    cur_yw = cur_y * 100 + cur_w
    next_yw = next_y * 100 + next_w

    rows = db.execute(text("""
        SELECT inc.year, inc.week, inc.product_id,
               inc.quantity::float AS qty,
               COALESCE(ec.cost_price, 0)::float AS cost,
               COALESCE(ds.name, '(unknown)') AS supplier
        FROM incoming_supply inc
        LEFT JOIN supply_master sm ON sm.product_id = inc.product_id
        LEFT JOIN dim_suppliers ds ON ds.id = sm.supplier_id
        LEFT JOIN erp_costs     ec ON ec.product_id = inc.product_id
        WHERE inc.quantity > 0
          AND inc.year*100 + inc.week >= :cur_yw - 8
    """), {"cur_yw": cur_yw}).mappings().all()

    df = pd.DataFrame([dict(r) for r in rows])
    if df.empty:
        return {
            "this_week_label": _yw_label(cur_y, cur_w),
            "next_week_label": _yw_label(next_y, next_w),
            "this_week": [], "next_week": [],
            "this_week_total_eur": 0.0, "next_week_total_eur": 0.0,
            "overdue_pos": 0, "overdue_total_eur": 0.0,
        }

    df["yw"] = df["year"] * 100 + df["week"]
    df["eur"] = df["qty"] * df["cost"]

    def _supplier_block(slice_df: pd.DataFrame) -> list[dict]:
        if slice_df.empty:
            return []
        agg = (slice_df.groupby("supplier")
               .agg(po_count=("product_id", "count"),
                    total_qty=("qty", "sum"),
                    total_eur=("eur", "sum"))
               .reset_index()
               .sort_values("total_eur", ascending=False))
        out = []
        for r in agg.itertuples():
            name_lc = (r.supplier or "").lower()
            is_abc = any(k in name_lc for k in ABC_SUPPLIER_KEYS)
            out.append({
                "supplier_name": r.supplier,
                "po_count": int(r.po_count),
                "total_qty": round(float(r.total_qty), 0),
                "total_eur": round(float(r.total_eur), 2),
                "is_abc": bool(is_abc),
            })
        return out

    this_week_df = df[df["yw"] == cur_yw]
    next_week_df = df[df["yw"] == next_yw]
    overdue_df = df[df["yw"] < cur_yw]

    return {
        "this_week_label": _yw_label(cur_y, cur_w),
        "next_week_label": _yw_label(next_y, next_w),
        "this_week": _supplier_block(this_week_df),
        "next_week": _supplier_block(next_week_df),
        "this_week_total_eur": float(this_week_df["eur"].sum()),
        "next_week_total_eur": float(next_week_df["eur"].sum()),
        "overdue_pos": int(len(overdue_df)),
        "overdue_total_eur": float(overdue_df["eur"].sum()),
    }


# ─────────────────────────────────────────────────────────────────────────
# Section 5 — Forecast accuracy (most recent completed month, Gold+Silver+Bronze)
# ─────────────────────────────────────────────────────────────────────────
def get_forecast_accuracy_summary(db: Session) -> dict:
    """Monthly FA% for the most recent completed month, computed exactly as
    the Demand → Forecast Accuracy module computes its monthly rollup
    (`demand_service._monthly_rollup`):

        1. Per row, drop where actual <= 0 (matches `fa[fa["actual"] > 0]`).
        2. Snap each (year, week) to its Thursday's calendar month — the
           "Thursday rule" used everywhere else in the codebase.
        3. Per month, take totals: sumF = ΣF, sumA = ΣA across all rows.
        4. Drop months where sumA <= 0.
        5. fa   = max(0, 1 - |sumF - sumA| / sumA) × 100
           bias = (sumF - sumA) / sumA × 100

    Tier breakdown applies the same totals approach within each tier.
    """
    # Mirror demand_service: pull every backtest row (left-join sku_planning
    # so untiered SKUs still count toward the monthly total — that's the
    # canonical headline number). Tier breakdown below slices on the tier
    # column for the per-tier rows.
    df = pd.read_sql(text("""
        SELECT b.product_id, b.year, b.week,
               b.forecast::float AS forecast,
               b.actual::float   AS actual,
               COALESCE(sp.tier, 'N/A') AS tier
        FROM backtest_results b
        LEFT JOIN sku_planning sp ON sp.product_id = b.product_id
        WHERE b.actual IS NOT NULL
          AND b.forecast IS NOT NULL
          AND b.actual > 0
    """), db.bind)
    if df.empty:
        return {
            "fa_pct": None, "fa_pct_prev_month": None, "trend": "unknown",
            "bias_pct": None, "measurement_month": None, "tier_breakdown": [],
        }

    # Thursday-of-week → calendar month (matches `_week_to_month_name`).
    df["month"] = df.apply(
        lambda r: datetime.strptime(
            f"{int(r['year'])}-W{int(r['week']):02d}-4", "%G-W%V-%u"
        ).strftime("%Y-%m"),
        axis=1,
    )

    def _totals_fa(rows: pd.DataFrame) -> Optional[float]:
        """Sum-then-divide FA, matching demand_service._monthly_rollup."""
        if rows.empty:
            return None
        sum_a = float(rows["actual"].sum())
        if sum_a <= 0:
            return None
        sum_f = float(rows["forecast"].sum())
        return max(0.0, 1 - abs(sum_f - sum_a) / sum_a) * 100.0

    def _totals_bias(rows: pd.DataFrame) -> Optional[float]:
        if rows.empty:
            return None
        sum_a = float(rows["actual"].sum())
        if sum_a <= 0:
            return None
        sum_f = float(rows["forecast"].sum())
        return (sum_f - sum_a) / sum_a * 100.0

    months = sorted(df["month"].unique())
    if not months:
        return {
            "fa_pct": None, "fa_pct_prev_month": None, "trend": "unknown",
            "bias_pct": None, "measurement_month": None, "tier_breakdown": [],
        }

    latest_month = months[-1]
    prev_month = months[-2] if len(months) >= 2 else None

    cur = df[df["month"] == latest_month]
    prev = df[df["month"] == prev_month] if prev_month else pd.DataFrame()

    fa_now = _totals_fa(cur)
    fa_prev = _totals_fa(prev) if not prev.empty else None
    bias_now = _totals_bias(cur)

    if fa_now is None or fa_prev is None:
        trend = "unknown"
    elif fa_now > fa_prev + 1.0:
        trend = "up"
    elif fa_now < fa_prev - 1.0:
        trend = "down"
    else:
        trend = "stable"

    tier_breakdown = []
    for tier in STOCKOUT_TIERS:
        slc = cur[cur["tier"] == tier]
        tier_breakdown.append({
            "tier": TIER_DISPLAY.get(tier, tier),
            "fa_pct": (round(_totals_fa(slc), 1)
                       if _totals_fa(slc) is not None else None),
            "n_skus": int(slc["product_id"].nunique()),
        })

    return {
        "fa_pct": round(fa_now, 1) if fa_now is not None else None,
        "fa_pct_prev_month": (round(fa_prev, 1) if fa_prev is not None
                              else None),
        "trend": trend,
        "bias_pct": round(bias_now, 1) if bias_now is not None else None,
        "measurement_month": datetime.strptime(latest_month, "%Y-%m")
                                       .strftime("%B %Y"),
        "tier_breakdown": tier_breakdown,
    }


# ─────────────────────────────────────────────────────────────────────────
# Section 6 — Inventory trend (8-week chart)
# ─────────────────────────────────────────────────────────────────────────
def get_inventory_trend(db: Session, weeks: int = 8) -> dict:
    """Weekly inventory value time series. We don't currently retain
    historical stock snapshots, so this collapses to a single point
    representing today's inventory until that job exists."""
    snap = get_inventory_snapshot(db)
    cur_y, cur_w = _iso_week_now(db)
    return {
        "points": [{
            "year_week": _yw_label(cur_y, cur_w),
            "total_eur": snap["total_inventory_eur"],
        }],
        "has_history": False,
    }


# ─────────────────────────────────────────────────────────────────────────
# Orchestrator
# ─────────────────────────────────────────────────────────────────────────
def get_executive_dashboard(db: Session) -> dict:
    """Assemble every section. A failure in one section does not abort the
    others — it surfaces as `<section>_error` in the response."""
    cur_y, cur_w = _iso_week_now(db)
    out: dict = {
        "anchor_year_week": _yw_label(cur_y, cur_w),
        "refreshed_at": datetime.now().isoformat(timespec="seconds"),
        "inventory": None, "inventory_error": None,
        "revenue": None, "revenue_error": None,
        "stockout_risks": [], "stockout_error": None,
        "supply_arrivals": None, "supply_error": None,
        "forecast_accuracy": None, "forecast_accuracy_error": None,
        "inventory_trend": None, "inventory_trend_error": None,
        "revenue_weekly": [],
        "plan_compare": None, "plan_compare_error": None,
    }

    try:
        out["inventory"] = get_inventory_snapshot(db)
    except Exception as e:
        out["inventory_error"] = f"{type(e).__name__}: {e}"

    try:
        out["revenue"] = get_revenue_pulse(db)
    except Exception as e:
        out["revenue_error"] = f"{type(e).__name__}: {e}"

    try:
        out["stockout_risks"] = get_stockout_risks(db)
    except Exception as e:
        out["stockout_error"] = f"{type(e).__name__}: {e}"

    try:
        out["supply_arrivals"] = get_supply_arrivals(db)
    except Exception as e:
        out["supply_error"] = f"{type(e).__name__}: {e}"

    try:
        out["forecast_accuracy"] = get_forecast_accuracy_summary(db)
    except Exception as e:
        out["forecast_accuracy_error"] = f"{type(e).__name__}: {e}"

    try:
        out["inventory_trend"] = get_inventory_trend(db)
    except Exception as e:
        out["inventory_trend_error"] = f"{type(e).__name__}: {e}"

    try:
        out["revenue_weekly"] = get_revenue_weekly(db, weeks=4)
    except Exception:
        out["revenue_weekly"] = []

    try:
        out["plan_compare"] = get_plan_compare(db)
    except Exception as e:
        out["plan_compare_error"] = f"{type(e).__name__}: {e}"

    return out