platform/modules/customers.py

"""Customers module — the "why" behind the brand divergence (Fisch −5.5% vs Royal +40%).

Centerpiece is the **customer revenue bridge**: it decomposes the YoY revenue change into
New (+), Expansion (+), Contraction (−) and Lost (−). That decomposition both answers the
business question and self-validates — the four components must reconcile last-period total
to this-period total exactly (built into validate()).

Plus a MECE value-tier segmentation (LTM), an at-risk / win-back list ranked by dollars at
stake, and new/lost customer lists. All order-level (sale.order), RI+FFS scope, excluded
accounts removed — reusing the Sales module's order_domain so scope is identical everywhere.
"""
import sys
import datetime as dt
import statistics
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import core.odoo as O
import core.periods as P
import modules.sales as sales_mod


# ---- per-customer grouped reads: STORE-backed (OM-2 retrofit 2026-07-12) with LIVE fallback --
# _cust_rev/_last_order_dates are the choke points behind the Customers bundle, the Map's slow
# wave-1 and the Customer List queues. SQL mirrors sales.order_domain incl. the agent filter
# (partner_ids; an EMPTY set matches no orders, same as the live domain). validate() stays live.
USE_STORE = True


def _cust_group_store(date_from, date_to, team_id, agent_pids, select):
    import harness.datastore as DS
    params, w = [], ["state IN ('sale','done')"]
    teams = sales_mod.TEAMS(team_id)
    w.append("team_id IN (" + ",".join("?" * len(teams)) + ")")
    params += list(teams)
    if date_from:
        w.append("CAST(date_order AS TIMESTAMP) >= CAST(? AS TIMESTAMP)")
        params.append(f"{date_from} 00:00:00")
    if date_to:
        w.append("CAST(date_order AS TIMESTAMP) <= CAST(? AS TIMESTAMP)")
        params.append(f"{date_to} 23:59:59")
    ex = O.excluded_partner_ids()
    if ex:
        w.append("partner_id NOT IN (" + ",".join("?" * len(ex)) + ")")
        params += list(ex)
    if agent_pids is not None:
        if not agent_pids:
            return []                                    # empty set matches no orders
        w.append("partner_id IN (" + ",".join("?" * len(agent_pids)) + ")")
        params += list(agent_pids)
    if O.doc_mode() == 'invoice':
        w.append("invoice_status = 'invoiced'")
    return DS.ro_con().execute(
        f"SELECT o.partner_id, coalesce(p.name, '#' || o.partner_id), {select} "
        "FROM sale_order o LEFT JOIN res_partner p ON p.id = o.partner_id "
        "WHERE " + " AND ".join(w) + " GROUP BY 1, 2", params).fetchall()


def _cust_rev(date_from, date_to, team_id=None, agent_pids=None):
    """{partner_id: {'name', 'rev', 'orders'}} over a window (RI+FFS, excluded accounts removed).
    agent_pids (frozenset|None) restricts to one Agent's customers — the module-wide Agent filter."""
    if USE_STORE:
        try:
            rows = _cust_group_store(date_from, date_to, team_id, agent_pids,
                                     'sum(amount_untaxed), count(*)')
            return {r[0]: {'name': r[1], 'rev': r[2] or 0.0, 'orders': r[3]}
                    for r in rows if r[0]}
        except Exception:
            pass
    g = O.read_group('sale.order', sales_mod.order_domain(date_from, date_to, team_id, partner_ids=agent_pids),
                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
    out = {}
    for r in g:
        pid = O.m2o_id(r.get('partner_id'))
        if not pid:
            continue
        out[pid] = {'name': O.m2o_name(r.get('partner_id')),
                    'rev': r.get('amount_untaxed') or 0.0,
                    'orders': r.get('__count') or 0}
    return out


# ----------------------------------------------------------- agent filter (module-wide)
# Agent = res.partner.agent_ids (the assigned sales agent on a customer; the same attribute the
# rollups slice by). The Customer module can be filtered to one agent's whole book: we resolve the
# agent name → the set of partner ids assigned to them, then pass it as agent_pids everywhere.
def agent_options(t=None, team_id=None):
    """Agent names that have customers with YTD activity in scope — for the module filter dropdown.
    Reuses the agent rollup (cheap, cached at the app layer). '(none)' = unassigned customers."""
    return [r['group'] for r in by_dimension('agent', t, team_id=team_id)]


def agent_partner_ids(agent_name, customers_only=False):
    """frozenset of partner ids assigned to `agent_name` via res.partner.agent_ids. None when no
    agent is selected ('All agents'); an EMPTY frozenset (matches nobody) when the agent has no
    customers. '(none)' resolves to customers with no agent assigned.

    ⭐ WAVE 20 (R3, DEBT D-30) — `customers_only` IS THE EXPLICIT POLICY THIS FUNCTION WAS
    MISSING, and the two callers genuinely want different answers:

      * **Permission scope (default, `False`)** — an agent-LOGIN user's whole app is bounded by
        this set, so it must be GENEROUS: archived accounts and Odoo address records included.
        Narrowing it would hide an agent's own data from them, and the fails-closed trap in the
        note below is what that costs.
      * **Display/reporting (`True`)** — "how many accounts does Martin have" must answer what
        Odoo answers. MEASURED 2026-08-05: the generous set is 503 for Martin and the honest one
        is **494**, the owner's number; the 9-row gap is entirely `type in (delivery, other)`
        ADDRESS records that ride the pool because they appear on orders, two of them with no
        name at all.

    Both are the SAME m2m-contains resolver — which is D-30's actual requirement. The bug was
    never the generosity; it was that the Agent COLUMN used a THIRD rule (`agent_ids[0]`, first
    agent only) that agreed with neither, so an admin filtering `Agent = X` and X's own login saw
    different books. The column now lists every agent on the partner and this states its policy
    out loud, so the two can be reconciled by reading them instead of by measuring them.

    INCLUDES INACTIVE/archived partners (active in [True,False]) — the agent's book is the whole
    book (dormant + archived accounts too), which is what the Agents page promises AND what an
    AGENT-LOGIN user must see as their complete, isolated book. Active-only used to drop archived
    customers entirely (e.g. an agent whose whole small book is archived resolved to an EMPTY book —
    the fails-closed trap). Resolves the agent name to EVERY matching partner id (not just the
    first) so a same-name collision can't silently mis-scope the book (owner 2026-07-21)."""
    if not agent_name or agent_name in ('All agents', 'All'):
        return None
    _all = [('active', 'in', [True, False])]
    # The display policy: a real customer record, still active. See the docstring for the
    # measured 503-vs-494 this closes.
    _qual = [('customer_rank', '>', 0), ('active', '=', True)] if customers_only else _all
    if agent_name == '(none)':
        rows = O.search_read('res.partner', [('customer_rank', '>', 0),
                                             ('agent_ids', '=', False)] + _all, ['id'], limit=100000)
        return frozenset(r['id'] for r in rows)
    ag = O.search_read('res.partner', [('name', '=', agent_name)] + _all, ['id'], limit=10)
    if not ag:
        return frozenset()
    rows = O.search_read('res.partner', [('agent_ids', 'in', [a['id'] for a in ag])] + _qual,
                         ['id'], limit=100000)
    return frozenset(r['id'] for r in rows)


# ----------------------------------------------------------- revenue bridge
def revenue_bridge(t=None, team_id=None, agent_pids=None):
    """Decompose YTD-vs-same-period-LY revenue change into New/Expansion/Contraction/Lost."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    tset, lset = set(this), set(last)

    new_ids = tset - lset
    lost_ids = lset - tset
    both = tset & lset

    new_rev = sum(this[p]['rev'] for p in new_ids)
    lost_rev = sum(last[p]['rev'] for p in lost_ids)
    expansion = sum(this[p]['rev'] - last[p]['rev'] for p in both if this[p]['rev'] > last[p]['rev'])
    contraction = sum(this[p]['rev'] - last[p]['rev'] for p in both if this[p]['rev'] < last[p]['rev'])

    this_total = sum(v['rev'] for v in this.values())
    last_total = sum(v['rev'] for v in last.values())
    return {
        'last_total': last_total,
        'this_total': this_total,
        'change': this_total - last_total,
        'new': {'rev': new_rev, 'n': len(new_ids)},
        'expansion': {'rev': expansion, 'n': sum(1 for p in both if this[p]['rev'] > last[p]['rev'])},
        'contraction': {'rev': contraction, 'n': sum(1 for p in both if this[p]['rev'] < last[p]['rev'])},
        'lost': {'rev': -lost_rev, 'n': len(lost_ids)},
        'retained_n': len(both),
        'active_this': len(tset),
        'active_last': len(lset),
    }


def bridge_by_brand(t=None, agent_pids=None):
    rows = []
    for tid in O.TEAM_IDS:
        b = revenue_bridge(t, team_id=tid, agent_pids=agent_pids)
        rows.append({'brand': O.TEAM_NAMES[tid], **{
            'last': b['last_total'], 'this': b['this_total'], 'change': b['change'],
            'new': b['new']['rev'], 'expansion': b['expansion']['rev'],
            'contraction': b['contraction']['rev'], 'lost': b['lost']['rev'],
            'new_n': b['new']['n'], 'lost_n': b['lost']['n']}})
    return rows


def bridge_component_customers(component, t=None, team_id=None, agent_pids=None):
    """The customers contributing to one revenue-bridge component (New / Expansion / Contraction /
    Lost), each with LY, YTD, the bridge delta and share of that component — for the click-through
    drawer off the Revenue-bridge chart."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    tset, lset = set(this), set(last)
    comp = (component or '').strip().lower()
    rows = []
    if comp == 'new':
        rows = [{'pid': p, 'customer': this[p]['name'], 'rev_ly': 0.0, 'rev_ytd': this[p]['rev'],
                 'change': this[p]['rev']} for p in (tset - lset)]
    elif comp == 'lost':
        rows = [{'pid': p, 'customer': last[p]['name'], 'rev_ly': last[p]['rev'], 'rev_ytd': 0.0,
                 'change': -last[p]['rev']} for p in (lset - tset)]
    elif comp in ('expansion', 'contraction'):
        up = comp == 'expansion'
        for p in (tset & lset):
            d = this[p]['rev'] - last[p]['rev']
            if (d > 0) == up and d != 0:
                rows.append({'pid': p, 'customer': this[p]['name'], 'rev_ly': last[p]['rev'],
                             'rev_ytd': this[p]['rev'], 'change': d})
    denom = sum(abs(r['change']) for r in rows) or 1.0
    for r in rows:
        r['at_risk'] = max(0.0, r['rev_ly'] - r['rev_ytd'])
        r['pct_of_component'] = abs(r['change']) / denom * 100.0
    rows.sort(key=lambda r: -abs(r['change']))
    _attach_attrs(rows)
    return {'component': component, 'rows': rows, 'n': len(rows),
            'total': sum(r['change'] for r in rows)}


def period_customers(ym, t=None, team_id=None, agent_pids=None):
    """Customers who purchased in calendar month `ym` (YYYY-MM) this year — that month's revenue,
    the same month last year, order count, and each customer's YTD-vs-LY at-risk for context.
    Powers the click-through drawer off the Monthly (this-year vs last-year) bar chart."""
    t = t or P.today()
    y, m = int(ym[:4]), int(ym[5:7])

    def _bounds(yr):
        s = dt.date(yr, m, 1)
        e = dt.date(yr + (m // 12), (m % 12) + 1, 1) - dt.timedelta(days=1)
        return s.isoformat(), e.isoformat()
    ts, te = _bounds(y)
    ls, le = _bounds(y - 1)
    this = _cust_rev(ts, te, team_id, agent_pids)
    last = _cust_rev(ls, le, team_id, agent_pids)
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    ytd = _cust_rev(yf, yt, team_id, agent_pids)
    lytd = _cust_rev(lf, lt, team_id, agent_pids)
    rows = [{'pid': p, 'customer': v['name'], 'rev_this': v['rev'], 'orders': v['orders'],
             'rev_ly': last.get(p, {}).get('rev', 0.0),
             'at_risk': max(0.0, lytd.get(p, {}).get('rev', 0.0) - ytd.get(p, {}).get('rev', 0.0))}
            for p, v in this.items()]
    rows.sort(key=lambda r: -r['rev_this'])
    _attach_attrs(rows)
    return {'month': ym, 'rows': rows, 'n': len(rows),
            'rev_this': sum(r['rev_this'] for r in rows),
            'rev_ly': sum(r['rev_ly'] for r in rows)}


def monthly_kpis(t=None, team_id=None, n_months=24):
    """Monthly time-series for the acquisition/engagement trends at the top of the page: order
    count, AOV, distinct purchasing customers, and newly-acquired customers (first-ever order that
    month)."""
    t = t or P.today()
    months = P.month_starts(n_months, t)
    win_start = months[0][1]
    dom = sales_mod.order_domain(win_start, t.isoformat(), team_id)

    def _ym(r):
        rng = (r.get('__range') or {}).get('date_order:month') or {}
        return (rng.get('from') or '')[:7]
    mrev, morders = {}, {}
    for r in O.read_group('sale.order', dom, ['amount_untaxed:sum'], ['date_order:month'], lazy=False):
        ym = _ym(r)
        if ym:
            mrev[ym] = r.get('amount_untaxed') or 0.0
            morders[ym] = r.get('__count') or 0
    mbuyers = {}
    for r in O.read_group('sale.order', dom, ['__count'], ['date_order:month', 'partner_id'], lazy=False):
        ym = _ym(r)
        if ym:
            mbuyers[ym] = mbuyers.get(ym, 0) + 1
    # new customers: first-ever order month per partner (all history), bucketed into our window
    mnew = {}
    for r in O.read_group('sale.order', sales_mod.order_domain('2000-01-01', t.isoformat(), team_id),
                          ['date_order:min'], ['partner_id'], lazy=False):
        ym = str(r.get('date_order') or '')[:7]
        if ym:
            mnew[ym] = mnew.get(ym, 0) + 1
    out = []
    for ym, _s, _e in months:
        orders = morders.get(ym, 0)
        out.append({'month': ym, 'orders': orders, 'revenue': mrev.get(ym, 0.0),
                    'aov': (mrev.get(ym, 0.0) / orders) if orders else 0.0,
                    'buyers': mbuyers.get(ym, 0), 'new_customers': mnew.get(ym, 0)})
    return out


def customer_trends(t=None, team_id=None, agent_pids=None):
    """12 months aligned THIS-year vs same-month-LAST-year, with both the monthly value and the
    running CUMULATIVE (this vs last) for orders, AOV, purchasing customers (distinct) and new
    customers. Powers the Trends chart's Monthly-bars / Cumulative-line toggle."""
    t = t or P.today()
    months = P.month_starts(24, t)            # 24 ascending: prior 12 then recent 12
    win_start = months[0][1]
    dom = sales_mod.order_domain(win_start, t.isoformat(), team_id, partner_ids=agent_pids)

    def _ym(r):
        rng = (r.get('__range') or {}).get('date_order:month') or {}
        return (rng.get('from') or '')[:7]
    mrev, morders, mbuyers = {}, {}, {}
    for r in O.read_group('sale.order', dom, ['amount_untaxed:sum'], ['date_order:month'], lazy=False):
        ym = _ym(r)
        if ym:
            mrev[ym] = r.get('amount_untaxed') or 0.0
            morders[ym] = r.get('__count') or 0
    for r in O.read_group('sale.order', dom, ['__count'], ['date_order:month', 'partner_id'], lazy=False):
        ym = _ym(r)
        pid = O.m2o_id(r.get('partner_id'))
        if ym and pid:
            mbuyers.setdefault(ym, set()).add(pid)
    mnew = {}
    for r in O.read_group('sale.order', sales_mod.order_domain('2000-01-01', t.isoformat(), team_id, partner_ids=agent_pids),
                          ['date_order:min'], ['partner_id'], lazy=False):
        ym = str(r.get('date_order') or '')[:7]
        if ym:
            mnew[ym] = mnew.get(ym, 0) + 1

    recent = [m[0] for m in months[12:24]]    # the 12 most recent months (ascending)
    prior = [m[0] for m in months[0:12]]      # the same 12 months a year earlier
    out = []
    cr_t = co_t = cn_t = 0.0
    cr_l = co_l = cn_l = 0.0
    cb_t, cb_l = set(), set()
    for i in range(12):
        ty, ly = recent[i], prior[i]
        ot, rt, nt = morders.get(ty, 0), mrev.get(ty, 0.0), mnew.get(ty, 0)
        ol, rl, nl = morders.get(ly, 0), mrev.get(ly, 0.0), mnew.get(ly, 0)
        bt, bl = mbuyers.get(ty, set()), mbuyers.get(ly, set())
        co_t += ot; cr_t += rt; cn_t += nt; cb_t |= bt
        co_l += ol; cr_l += rl; cn_l += nl; cb_l |= bl
        out.append({
            'month': ty, 'month_ly': ly,
            'orders': ot, 'orders_ly': ol,
            'aov': (rt / ot if ot else 0.0), 'aov_ly': (rl / ol if ol else 0.0),
            'buyers': len(bt), 'buyers_ly': len(bl),
            'new_customers': nt, 'new_customers_ly': nl,
            'orders_cum': co_t, 'orders_cum_ly': co_l,
            'aov_cum': (cr_t / co_t if co_t else 0.0), 'aov_cum_ly': (cr_l / co_l if co_l else 0.0),
            'buyers_cum': len(cb_t), 'buyers_cum_ly': len(cb_l),
            'new_customers_cum': cn_t, 'new_customers_cum_ly': cn_l,
        })
    return out


# ----------------------------------------------------------- segmentation (MECE)
TIERS = [('Whale (≥$25k)', 25000), ('Large ($10–25k)', 10000),
         ('Mid ($2–10k)', 2000), ('Small (<$2k)', 0)]


def _tier(rev):
    for name, lo in TIERS:
        if rev >= lo:
            return name
    return TIERS[-1][0]


def segments(t=None, team_id=None, agent_pids=None):
    """MECE value-tier segmentation over LTM revenue (each active customer in one tier)."""
    t = t or P.today()
    lf, lt = P.ltm(t)
    cust = _cust_rev(lf, lt, team_id, agent_pids)
    agg = {name: {'segment': name, 'customers': 0, 'revenue': 0.0} for name, _ in TIERS}
    for v in cust.values():
        a = agg[_tier(v['rev'])]
        a['customers'] += 1
        a['revenue'] += v['rev']
    total_rev = sum(a['revenue'] for a in agg.values()) or 1.0
    total_n = sum(a['customers'] for a in agg.values()) or 1
    rows = []
    for name, _ in TIERS:
        a = agg[name]
        a['rev_share'] = a['revenue'] / total_rev * 100
        a['cust_share'] = a['customers'] / total_n * 100
        a['avg_rev'] = a['revenue'] / a['customers'] if a['customers'] else 0.0
        rows.append(a)
    return rows


def tier_customers(tier, t=None, team_id=None, agent_pids=None):
    """The customers in ONE LTM value tier — each with LTM revenue, order count, units bought and
    average order value — for the Value-tiers drill. Tier membership matches segments() exactly
    (same LTM window + _tier)."""
    t = t or P.today()
    lf, lt = P.ltm(t)
    cust = _cust_rev(lf, lt, team_id, agent_pids)
    pids = [pid for pid, v in cust.items() if _tier(v['rev']) == tier]
    if not pids:
        return []
    units = {}
    for r in O.read_group('sale.order.line', O.sale_line_domain(lf, lt, team_id, partner_ids=pids),
                          ['product_uom_qty:sum'], ['order_partner_id'], lazy=False):
        pid = O.m2o_id(r.get('order_partner_id'))
        if pid:
            units[pid] = r.get('product_uom_qty') or 0.0
    pdata = {p['id']: p for p in O.search_read('res.partner', [('id', 'in', pids)], ['city', 'state_id'])}
    rows = []
    for pid in pids:
        v = cust[pid]
        o = v.get('orders', 0)
        p = pdata.get(pid, {})
        rows.append({'pid': pid, 'customer': v['name'], 'city': p.get('city') or '(none)',
                     'state': O.m2o_name(p.get('state_id')) or '(none)', 'agent': '(none)',
                     'rev_ltm': v['rev'], 'orders': o, 'units': units.get(pid, 0.0),
                     'aov': (v['rev'] / o) if o else 0.0})
    rows.sort(key=lambda x: -x['rev_ltm'])
    return rows


# ----------------------------------------------------------- at-risk / win-back
def at_risk(t=None, team_id=None, limit=30, min_prior=2000.0, agent_pids=None):
    """Customers who bought materially last year but are down or gone this year, ranked by
    dollars at risk (last − this). The win-back target list."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    rows = []
    for pid, lv in last.items():
        if lv['rev'] < min_prior:
            continue
        tv = this.get(pid, {'rev': 0.0})
        down = lv['rev'] - tv['rev']
        if down <= 0:
            continue
        rows.append({'pid': pid, 'customer': lv['name'], 'rev_ly': lv['rev'], 'rev_ytd': tv['rev'],
                     'orders': tv.get('orders', 0), 'at_risk': down,
                     'status': 'Lost' if pid not in this else 'Declining'})
    rows.sort(key=lambda x: -x['at_risk'])
    return rows[:limit]


def _first_order_dates(team_id=None, t=None, agent_pids=None):
    """{pid: first-ever confirmed order date (ISO)} — one all-history read_group(min)."""
    t = t or P.today()
    g = O.read_group('sale.order', sales_mod.order_domain('2000-01-01', t.isoformat(), team_id, partner_ids=agent_pids),
                     ['date_order:min'], ['partner_id'], lazy=False)
    out = {}
    for r in g:
        pid = O.m2o_id(r.get('partner_id'))
        if pid and r.get('date_order'):
            out[pid] = str(r['date_order'])[:10]
    return out


def new_customers(t=None, team_id=None, limit=30, agent_pids=None):
    """Customers acquired or reactivated this year (active this YTD, not in same-period LY), each
    tagged New (first-ever order this year) vs Reactivated (ordered in a prior year, had lapsed)."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    cand = [pid for pid in this if pid not in last]
    firsts = _first_order_dates(team_id, t, agent_pids) if cand else {}
    rows = []
    for pid in cand:
        v = this[pid]
        first = firsts.get(pid, '')
        status = 'New' if first[:4] == str(t.year) else 'Reactivated'
        rows.append({'pid': pid, 'customer': v['name'], 'rev_ytd': v['rev'], 'orders': v['orders'],
                     'first_order': first, 'status': status})
    rows.sort(key=lambda x: -x['rev_ytd'])
    return rows[:limit]


def lists_bundle(t=None, team_id=None, agent_pids=None):
    """At-risk, New/reactivated and Follow-up lists, all enriched with a CONSISTENT insight column
    set — YTD $, YoY %, orders, last order, days overdue vs the customer's own cadence, agent —
    plus each list's own metric (at-risk $ / status / est. missed $)."""
    t = t or P.today()
    cad = _cadence_bulk(t, team_id, agent_pids=agent_pids)
    ar = at_risk(t, team_id, agent_pids=agent_pids)
    nw = new_customers(t, team_id, agent_pids=agent_pids)
    fu = contact_recommendations(t, team_id, agent_pids=agent_pids)
    pids = {r['pid'] for r in ar} | {r['pid'] for r in nw} | {r['pid'] for r in fu}
    attrs = _partner_attrs(list(pids))

    def enrich(rows):
        for r in rows:
            c = cad.get(r['pid'], {})
            r['last_order'] = r.get('last_order') or c.get('last_order', '')
            r['overdue_days'] = c.get('overdue_days')
            r['agent'] = (attrs.get(r['pid']) or {}).get('agent', '(none)')
            if 'rev_ytd' not in r and 'ltm_rev' in r:
                r['rev_ytd'] = r['ltm_rev']
            if 'yoy_pct' not in r:
                r['yoy_pct'] = P.yoy_pct(r.get('rev_ytd', 0.0), r.get('rev_ly', 0.0))
        return rows
    return {'at_risk': enrich(ar), 'new': enrich(nw), 'followups': enrich(fu)}


def _cadence_bulk(t=None, team_id=None, months=24, agent_pids=None):
    """{pid: {n_orders, last_order, typical_gap_days, days_since, overdue_days, aov}} from ONE
    read_group over the window (count + min/max date + revenue per partner). Bulk approximation of
    per-customer cadence — the basis for the follow-up list and the lists' cadence columns."""
    t = t or P.today()
    months = max(1, months)
    start = dt.date(t.year - (months // 12) - (1 if t.month <= (months % 12) else 0),
                    ((t.month - 1 - (months % 12)) % 12) + 1, 1)
    g = O.read_group('sale.order', sales_mod.order_domain(start.isoformat(), t.isoformat(), team_id, partner_ids=agent_pids),
                     ['amount_untaxed:sum', 'mind:min(date_order)', 'maxd:max(date_order)'],
                     ['partner_id'], lazy=False)
    out = {}
    for r in g:
        pid = O.m2o_id(r.get('partner_id'))
        if not pid:
            continue
        n = r.get('__count') or 0
        rev = r.get('amount_untaxed') or 0.0
        try:
            dmin = dt.date.fromisoformat(str(r.get('mind'))[:10])
            dmax = dt.date.fromisoformat(str(r.get('maxd'))[:10])
        except (TypeError, ValueError):
            continue
        span = (dmax - dmin).days
        gap = (span / (n - 1)) if n >= 2 and span > 0 else None
        days_since = (t - dmax).days
        overdue = (days_since - gap) if gap else None
        out[pid] = {'n_orders': n, 'last_order': dmax.isoformat(), 'typical_gap_days': gap,
                    'days_since': days_since, 'overdue_days': overdue, 'aov': (rev / n) if n else 0.0}
    return out


def contact_recommendations(t=None, team_id=None, limit=40, months=24, agent_pids=None):
    """Prioritised follow-up list: customers overdue against their OWN purchase cadence, ranked by
    estimated missed revenue (how far past due × their average order value). Reps' call list."""
    t = t or P.today()
    cad = _cadence_bulk(t, team_id, months, agent_pids=agent_pids)
    attrs = _partner_attrs(list(cad))
    names = {}
    # names from the LTM revenue map (cheap, already scoped)
    lf, lt = P.ltm(t)
    rev_map = _cust_rev(lf, lt, team_id, agent_pids)
    rows = []
    for pid, c in cad.items():
        gap = c['typical_gap_days']
        if c['n_orders'] < 3 or not gap or gap <= 0:
            continue                      # need a real, repeated cadence
        overdue = c['overdue_days']
        if overdue is None or overdue <= 0:
            continue                      # only those past due
        if c['days_since'] > 365:
            continue                      # long-dead → not a cadence follow-up
        ltm_rev = rev_map.get(pid, {}).get('rev', 0.0)
        cycles_missed = overdue / gap                   # how many reorder cycles past due
        # cap at 3 cycles so a long-churned account (belongs in win-back) doesn't dominate the
        # "call them now" list; this rewards valuable regulars who are RECENTLY due.
        est_missed = min(cycles_missed, 3.0) * c['aov']
        rows.append({
            'pid': pid, 'customer': rev_map.get(pid, {}).get('name', '?'),
            'last_order': c['last_order'], 'typical_gap_days': gap, 'overdue_days': overdue,
            'ltm_rev': ltm_rev, 'orders': c['n_orders'], 'est_missed': est_missed,
            'agent': (attrs.get(pid) or {}).get('agent', '(none)'),
        })
    rows.sort(key=lambda x: -x['est_missed'])
    return rows[:limit]


# ====================================================================== EXPLORER
# Granular (per-customer drill-down) + sum-level (filter/group by city, state, etc).
# All revenue is order-level (sale.order, amount_untaxed) in the same RI+FFS scope as the
# rest of Sales, so every rollup sums back to the headline number (proven in validate()).

# Group-by dimensions for the sum-level rollups (key → display label).
DIMENSIONS = {
    'city': 'City', 'state': 'State / Region', 'country': 'Country',
    'agent': 'Agent', 'segment': 'Value tier',
}


def _partner_attrs(pids):
    """{pid: {street, street2, city, state, country, agent, zip, payment_terms, customer_since,
    tags, pricelist}} — the customer attributes the sum-level rollups slice by and the Customer
    table displays.

    Agent is res.partner.agent_ids (the assigned sales agent; ~one per customer), NOT Odoo's
    user_id 'salesperson' — MEASURED 2026-07-27 at 26 of 1,548 (2%), which is why agent_ids has
    always been the field here. Odoo's `credit_limit` is not read for the same reason: 20 of
    1,548 (1%). Credit EXPOSURE comes from AR (modules/collections.py), not from that field.

    Every value collapses to '(none)' when blank (MECE), so a group-by has no null bucket and a
    filter has something to match. ⚠ `zip` is TEXT, never numeric: postal codes carry leading
    zeros, and 01730 read as a number is 1730 — a different town.
    """
    pids = list(pids)
    if not pids:
        return {}
    rows = O.search_read('res.partner', [('id', 'in', pids)],
                         ['street', 'street2', 'city', 'state_id', 'country_id', 'agent_ids',
                          'zip', 'property_payment_term_id', 'create_date', 'category_id',
                          'property_product_pricelist'])
    aids = {a for r in rows for a in (r.get('agent_ids') or [])}
    anames = ({p['id']: p['name'] for p in O.search_read('res.partner', [('id', 'in', list(aids))], ['name'])}
              if aids else {})
    # Tags are m2m: ids on the partner, names in res.partner.category. One extra read for the
    # whole set rather than one per customer.
    tids = {t for r in rows for t in (r.get('category_id') or [])}
    tnames = ({c['id']: c['name'] for c in
               O.search_read('res.partner.category', [('id', 'in', list(tids))], ['name'])}
              if tids else {})
    out = {}
    for r in rows:
        ag = r.get('agent_ids') or []
        city = (r.get('city') or '').strip()
        tags = [tnames.get(t) for t in (r.get('category_id') or []) if tnames.get(t)]
        # create_date is a DATETIME ('2024-01-15 10:23:45'); the column is a DATE, and the grid
        # compares dates ISO-LEXICALLY, so a trailing time would sort and filter as text noise.
        since = (r.get('create_date') or '')
        out[r['id']] = {
            # ⭐ W30-T33 (carried W29-T51) — the STREET half of the address. `city`/`state`/`zip`/
            # `country` were already here; the two street lines were read nowhere on the customer
            # path, only by `modules/map.py` for geocoding, and a geocoder's input never reached
            # the grid. ⚠ NOT ROUTED THROUGH `map.py`: these come off `res.partner` directly, so a
            # customer the geocoder could not place still shows its address.
            # ⚠ `.strip()` WITHOUT `.title()`, unlike `city`. A street line carries unit numbers,
            # directionals and abbreviations ("123 NW 4TH ST APT 2B"), and title-casing turns
            # those into "123 Nw 4Th St Apt 2B" — a mangling `city` does not risk because a city
            # is one word-set of ordinary nouns.
            'street': (r.get('street') or '').strip() or '(none)',
            'street2': (r.get('street2') or '').strip() or '(none)',
            'city': city.title() if city else '(none)',
            'state': O.m2o_name(r.get('state_id')) or '(none)',
            'country': O.m2o_name(r.get('country_id')) or '(none)',
            # ⭐ WAVE 20 (R3, closes DEBT D-30) — EVERY agent on the partner, not just the first.
            #
            # This was `anames.get(ag[0])`, and it made TWO definitions of "an agent's book" that
            # disagreed: `agent_partner_ids()` (which scopes an agent-LOGIN user's whole app) is
            # m2m-CONTAINS, while this column named only `agent_ids[0]`. So an admin filtering
            # `Agent = X` and X's own login saw different sets — MEASURED book-wide 2026-08-05:
            # Tara Devon Gallager 9 rows, Sang Ching 6, Moishe Rubenstein 1, Martin Pasternak 1.
            #
            # Joined with ', ' rather than kept as a list because the column is a TEXT field the
            # grid groups and filters on; `contains` then matches any agent on a shared account,
            # which is the m2m question asked in the vocabulary the column already speaks.
            'agent': ', '.join(n for n in (anames.get(a) for a in ag) if n) or '(none)',
            'zip': (r.get('zip') or '').strip() or '(none)',
            'payment_terms': O.m2o_name(r.get('property_payment_term_id')) or '(none)',
            'customer_since': str(since)[:10] if since else '',
            # The FULL datetime, for the grid's `created_time` field type (wave-5 item 11) —
            # customer_since above stays the DATE the date-typed column sorts/filters on.
            'created_at': str(since) if since else '',
            'tags': ', '.join(tags) if tags else '(none)',
            'pricelist': O.m2o_name(r.get('property_product_pricelist')) or '(none)',
        }
    return out


def _attach_attrs(rows, key='pid'):
    """Enrich a list of customer rows (each carrying a partner id under `key`) in place with the
    standardized city / state / agent fields — so every customer list (page or drawer) can show the
    same filters/columns. Returns the same list."""
    pids = [r[key] for r in rows if r.get(key) is not None]
    if not pids:
        return rows
    attrs = _partner_attrs(pids)
    for r in rows:
        a = attrs.get(r.get(key), {})
        r.setdefault('city', a.get('city', '(none)'))
        r.setdefault('state', a.get('state', '(none)'))
        r.setdefault('agent', a.get('agent', '(none)'))
    return rows


def _last_order_dates(date_from, date_to, team_id=None, agent_pids=None):
    """{pid: last order date (ISO)} within the window — for the directory's recency column."""
    if USE_STORE:
        try:
            rows = _cust_group_store(date_from, date_to, team_id, agent_pids,
                                     'max(date_order)')
            return {r[0]: str(r[2])[:10] for r in rows if r[0] and r[2]}
        except Exception:
            pass
    g = O.read_group('sale.order', sales_mod.order_domain(date_from, date_to, team_id, partner_ids=agent_pids),
                     ['date_order:max'], ['partner_id'], lazy=False)
    out = {}
    for r in g:
        pid = O.m2o_id(r.get('partner_id'))
        if pid and r.get('date_order'):
            out[pid] = str(r['date_order'])[:10]
    return out


def directory(t=None, team_id=None, limit=None, agent_pids=None):
    """Per-customer table (YTD): revenue, YoY, orders, AOV, last order, city/state/salesperson.
    Basis for the drill-down picker and (indirectly) the rollups. Sorted by revenue desc."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    attrs = _partner_attrs(set(this) | set(last))
    lastord = _last_order_dates(yf, yt, team_id, agent_pids)
    rows = []
    for pid, v in this.items():
        a = attrs.get(pid, {})
        rev, orders = v['rev'], v['orders']
        ly = last.get(pid, {}).get('rev', 0.0)
        rows.append({
            'pid': pid, 'customer': v['name'],
            'revenue': rev, 'revenue_ly': ly, 'yoy_pct': P.yoy_pct(rev, ly),
            'orders': orders, 'aov': (rev / orders) if orders else 0.0,
            'last_order': lastord.get(pid, ''),
            'city': a.get('city', '(none)'), 'state': a.get('state', '(none)'),
            'agent': a.get('agent', '(none)'),
        })
    rows.sort(key=lambda x: -x['revenue'])
    return rows[:limit] if limit else rows


def by_dimension(dim, t=None, team_id=None, agent_pids=None):
    """Sum-level rollup: YTD revenue (vs LY) grouped by a customer attribute. dim is a key of
    DIMENSIONS. MECE — each active customer lands in exactly one group, so Σ(groups) == total
    YTD revenue (verified in validate()). When consolidated (team_id=None) each group also
    carries its Fisch + Royal revenue, mirroring the HQ rollup down to each brand. Sorted desc."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    attrs = _partner_attrs(set(this) | set(last))
    split = team_id is None
    f_rev = _cust_rev(yf, yt, 5, agent_pids) if split else {}
    r_rev = _cust_rev(yf, yt, 6, agent_pids) if split else {}

    def keyfor(pid, rev):
        if dim == 'segment':
            return _tier(rev)
        return (attrs.get(pid) or {}).get(dim) or '(none)'

    agg = {}

    def bucket(k):
        return agg.setdefault(k, {'group': k, 'revenue': 0.0, 'revenue_ly': 0.0,
                                  'fisch': 0.0, 'royal': 0.0, 'customers': 0, 'orders': 0})
    for pid, v in this.items():
        d = bucket(keyfor(pid, v['rev']))
        d['revenue'] += v['rev']; d['customers'] += 1; d['orders'] += v['orders']
        if split:
            d['fisch'] += f_rev.get(pid, {}).get('rev', 0.0)
            d['royal'] += r_rev.get(pid, {}).get('rev', 0.0)
    for pid, v in last.items():
        bucket(keyfor(pid, v['rev']))['revenue_ly'] += v['rev']
    rows = list(agg.values())
    for d in rows:
        d['yoy_pct'] = P.yoy_pct(d['revenue'], d['revenue_ly'])
        d['avg_per_customer'] = d['revenue'] / d['customers'] if d['customers'] else 0.0
    rows.sort(key=lambda x: -x['revenue'])
    return rows


def _order_dom(pid, date_from, date_to, team_id=None):
    return sales_mod.order_domain(date_from, date_to, team_id) + [('partner_id', '=', pid)]


def _line_dom(pid, date_from, date_to, team_id=None):
    return O.sale_line_domain(date_from, date_to, team_id, extra=[('order_partner_id', '=', pid)])


def _monthly_rev(pid, date_from, date_to, team_id=None):
    """{'YYYY-MM': revenue} over a window via ONE month-grouped read_group (was 26 point queries)."""
    g = O.read_group('sale.order', _order_dom(pid, date_from, date_to, team_id),
                     ['amount_untaxed:sum'], ['date_order:month'], lazy=False)
    out = {}
    for r in g:
        rng = (r.get('__range') or {}).get('date_order:month') or {}
        ym = (rng.get('from') or '')[:7]
        if ym:
            out[ym] = r.get('amount_untaxed') or 0.0
    return out


def customer_detail(pid, t=None, team_id=None, n_months=13, top=12):
    """Granular drill-down for one customer: profile, RFM-style KPIs, monthly YoY trend, top SKUs +
    categories bought (LTM, with margin), and recent orders. The ~14 independent Odoo reads run
    CONCURRENTLY (O.parallel_map) so a cold drawer loads in ~1-2s instead of ~6s."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    mf, mt = P.ltm(t)
    range_start = dt.date(t.year - 2, t.month, 1).isoformat()
    # Fisch/Royal mix only when consolidated (team_id=None). When a single BU is selected we never
    # compute or expose the other BU — strict isolation for per-BU permissioning.
    split = team_id is None
    r = O.parallel_map({
        'prof': lambda: O.search_read('res.partner', [('id', '=', pid)], ['name', 'email', 'phone']),
        'attrs': lambda: _partner_attrs([pid]),
        'ytd_rev': lambda: O.sum_field('sale.order', _order_dom(pid, yf, yt, team_id), 'amount_untaxed'),
        'ytd_rev_ly': lambda: O.sum_field('sale.order', _order_dom(pid, lf, lt, team_id), 'amount_untaxed'),
        'ytd_fisch': (lambda: O.sum_field('sale.order', _order_dom(pid, yf, yt, 5), 'amount_untaxed')) if split else (lambda: None),
        'ytd_royal': (lambda: O.sum_field('sale.order', _order_dom(pid, yf, yt, 6), 'amount_untaxed')) if split else (lambda: None),
        'ltm_rev': lambda: O.sum_field('sale.order', _order_dom(pid, mf, mt, team_id), 'amount_untaxed'),
        'orders_ltm': lambda: O.get_odoo().search_count('sale.order', _order_dom(pid, mf, mt, team_id)),
        'orders_ytd': lambda: O.get_odoo().search_count('sale.order', _order_dom(pid, yf, yt, team_id)),
        'lastrow': lambda: O.search_read('sale.order', _order_dom(pid, None, None, team_id), ['date_order'], order='date_order desc', limit=1),
        'firstrow': lambda: O.search_read('sale.order', _order_dom(pid, None, None, team_id), ['date_order'], order='date_order asc', limit=1),
        'mrev': lambda: _monthly_rev(pid, range_start, t.isoformat(), team_id),
        'lg': lambda: O.read_group('sale.order.line', _line_dom(pid, mf, mt, team_id),
                                   ['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'], ['product_id'], lazy=False),
        'recent': lambda: O.search_read('sale.order', _order_dom(pid, None, None, team_id),
                                        ['name', 'date_order', 'amount_untaxed', 'state'], order='date_order desc', limit=10),
    })
    prof = (r['prof'] or [{}])[0]
    a = r['attrs'].get(pid, {})
    ytd_rev, ytd_rev_ly, ltm_rev = r['ytd_rev'], r['ytd_rev_ly'], r['ltm_rev']
    ytd_fisch, ytd_royal = r['ytd_fisch'], r['ytd_royal']
    orders_ltm, orders_ytd = r['orders_ltm'], r['orders_ytd']
    last_order = str(r['lastrow'][0]['date_order'])[:10] if r['lastrow'] else None
    first_order = str(r['firstrow'][0]['date_order'])[:10] if r['firstrow'] else None
    recency = (t - dt.date.fromisoformat(last_order)).days if last_order else None

    mrev = r['mrev']
    monthly = []
    for ym, start, end in P.month_starts(n_months, t):
        y, m = int(ym[:4]) - 1, int(ym[5:7])
        monthly.append({'month': ym, 'revenue': mrev.get(ym, 0.0),
                        'revenue_ly': mrev.get(f'{y:04d}-{m:02d}', 0.0)})

    lg = r['lg']
    skus = [{'product': O.m2o_name(x.get('product_id')), 'revenue': x.get('price_subtotal') or 0.0,
             'qty': x.get('product_uom_qty') or 0.0, 'margin': x.get('margin') or 0.0}
            for x in lg if x.get('product_id')]
    skus.sort(key=lambda x: -x['revenue'])
    line_rev = sum(s['revenue'] for s in skus)
    margin = sum(s['margin'] for s in skus)

    cat = sales_mod._product_cat()
    catagg = {}
    for x in lg:
        prodid = O.m2o_id(x.get('product_id'))
        if not prodid:
            continue
        c = cat.get(prodid, '(uncategorized)')
        catagg[c] = catagg.get(c, 0.0) + (x.get('price_subtotal') or 0.0)
    cats = sorted([{'category': k, 'revenue': v} for k, v in catagg.items()],
                  key=lambda x: -x['revenue'])[:10]

    recent_rows = [{'order': x.get('name'), 'date': str(x.get('date_order'))[:10],
                    'amount': x.get('amount_untaxed') or 0.0,
                    'status': 'Confirmed' if x.get('state') in ('sale', 'done') else x.get('state')}
                   for x in r['recent']]

    return {
        'pid': pid, 'name': prof.get('name') or '(unknown)',
        'city': a.get('city', '(none)'), 'state': a.get('state', '(none)'),
        'country': a.get('country', '(none)'), 'agent': a.get('agent', '(none)'),
        'email': prof.get('email') or '', 'phone': prof.get('phone') or '',
        'ytd_rev': ytd_rev, 'ytd_rev_ly': ytd_rev_ly, 'yoy_pct': P.yoy_pct(ytd_rev, ytd_rev_ly),
        'ytd_fisch': ytd_fisch, 'ytd_royal': ytd_royal,
        'ltm_rev': ltm_rev, 'orders_ytd': orders_ytd, 'orders_ltm': orders_ltm,
        'aov_ltm': (ltm_rev / orders_ltm) if orders_ltm else 0.0,
        'first_order': first_order, 'last_order': last_order, 'recency_days': recency,
        'ltm_margin': margin, 'ltm_gm_pct': (margin / line_rev * 100) if line_rev else 0.0,
        'monthly': monthly, 'top_skus': skus[:top], 'top_categories': cats,
        'recent_orders': recent_rows,
    }


def customer_drawer_bundle(pid, t=None, team_id=None):
    """Everything the customer drawer's first paint needs, in as few round-trips as possible:
    customer_detail (itself parallel) then the other three pulls CONCURRENTLY. One cached unit, so
    switching drawer sections never re-hits Odoo. customer_affinity (the look-alike co-buyer scan —
    3 dependent heavy reads, 4-8s) and customer_stockout stay LAZY, loaded only by their own
    sections, so a cold drawer opens in ~2-3s regardless of how big the customer is."""
    detail = customer_detail(pid, t=t, team_id=team_id)
    decomp, winback, cadence = O.parallel([
        lambda: customer_yoy_decomp(pid, t, team_id),
        lambda: customer_winback(pid, t, team_id),
        lambda: customer_cadence(pid, t, team_id),
    ])
    return {'detail': detail, 'decomp': decomp, 'winback': winback, 'cadence': cadence}


def customer_stockout(pid, t=None, team_id=None, top=25):
    """Stockout exposure for one customer: of the SKUs they buy, which are currently OUT / LOW on
    hand, and how much of their YoY $ decline sits on SKUs that are now out of stock (a likely
    stockout-driven loss) vs other causes. On-hand is a CURRENT snapshot (no historical stock), so
    'decline on a now-out SKU' is a strong proxy, not proof, of a stockout cause."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)

    def skumap(df, dtt):
        # storable goods only — services (delivery charges etc.) can't "stock out"
        dom = O.sale_line_domain(df, dtt, team_id, extra=[('order_partner_id', '=', pid),
                                                          ('product_id.type', '!=', 'service')])
        g = O.read_group('sale.order.line', dom,
                         ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False)
        return {O.m2o_id(r['product_id']): {'name': O.m2o_name(r.get('product_id')),
                'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0}
                for r in g if r.get('product_id')}
    this, last = skumap(yf, yt), skumap(lf, lt)
    pids = list(set(this) | set(last))
    empty = {'rows': [], 'n_out': 0, 'n_low': 0, 'rev_at_risk': 0.0, 'risk_pct': 0.0,
             'drop_stockout': 0.0, 'drop_other': 0.0, 'total_drop': 0.0}
    if not pids:
        return empty
    q = O.read_group('stock.quant', [('location_id.usage', '=', 'internal'), ('product_id', 'in', pids)],
                     ['quantity:sum'], ['product_id'], lazy=False)
    onhand = {O.m2o_id(r['product_id']): (r.get('quantity') or 0.0) for r in q if r.get('product_id')}
    codemap = {}
    for pr in O.search_read('product.product', [('id', 'in', pids)], ['default_code']):
        codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None
    rows, drop_stockout, drop_other = [], 0.0, 0.0
    for p in pids:
        tr = this.get(p, {}).get('rev', 0.0)
        lr = last.get(p, {}).get('rev', 0.0)
        oh = onhand.get(p, 0.0)
        ly_qty = last.get(p, {}).get('qty', 0.0)
        out = oh <= 0
        low = (not out) and oh < max(1.0, ly_qty * 0.25)   # under ~a quarter of their annual usage
        change = tr - lr
        rows.append({'sku': (this.get(p) or last.get(p) or {}).get('name', '?'), 'code': codemap.get(p),
                     'on_hand': oh, 'ly_rev': lr, 'ytd_rev': tr, 'change': change,
                     'status': 'OUT' if out else ('LOW' if low else 'OK')})
        if change < 0:
            if out:
                drop_stockout += -change
            else:
                drop_other += -change
    rows.sort(key=lambda r: ({'OUT': 0, 'LOW': 1, 'OK': 2}[r['status']], -r['ly_rev']))
    this_total = sum(v['rev'] for v in this.values()) or 1.0
    rev_at_risk = sum(r['ytd_rev'] for r in rows if r['status'] in ('OUT', 'LOW'))
    return {'rows': rows[:top], 'n_out': sum(1 for r in rows if r['status'] == 'OUT'),
            'n_low': sum(1 for r in rows if r['status'] == 'LOW'),
            'rev_at_risk': rev_at_risk, 'risk_pct': rev_at_risk / this_total * 100,
            'drop_stockout': drop_stockout, 'drop_other': drop_other,
            'total_drop': drop_stockout + drop_other}


# ====================================================================== ADVANCED FILTERS
# "Find customers who bought X" — resolve a SKU query / category to the set of partner ids that
# purchased it, so the directory can be filtered by purchase behaviour (not just attributes).

def _buyer_window(t):
    """24-month look-back for 'who bought this' — catches recent AND lapsed buyers (win-back)."""
    t = t or P.today()
    return (t - dt.timedelta(days=730)).isoformat(), t.isoformat()


def sku_buyers(query, t=None, team_id=None):
    """Set of partner ids who bought any SKU whose code/name matches `query` (last 24 months).
    Returns None when the query is blank (= no filter)."""
    q = (query or '').strip()
    # A 1-char ilike matches half the catalogue — treat it as "still typing" (no filter yet).
    if len(q) < 2:
        return None
    df, dtt = _buyer_window(t)
    prods = O.search_read('product.product', ['|', ('default_code', 'ilike', q), ('name', 'ilike', q)],
                          ['id'], limit=3000)
    pids = [p['id'] for p in prods]
    if not pids:
        return set()
    g = O.read_group('sale.order.line',
                     O.sale_line_domain(df, dtt, team_id, extra=[('product_id', 'in', pids)]),
                     ['order_partner_id'], ['order_partner_id'], lazy=False)
    return {O.m2o_id(r.get('order_partner_id')) for r in g if r.get('order_partner_id')}


def category_buyers(category, t=None, team_id=None):
    """Set of partner ids who bought from a main category (last 24 months). None = no filter."""
    if not category or category == '(any)':
        return None
    df, dtt = _buyer_window(t)
    catmap = sales_mod._product_cat()
    prod_ids = [pid for pid, c in catmap.items() if c == category]
    if not prod_ids:
        return set()
    g = O.read_group('sale.order.line',
                     O.sale_line_domain(df, dtt, team_id, extra=[('product_id', 'in', prod_ids)]),
                     ['order_partner_id'], ['order_partner_id'], lazy=False)
    return {O.m2o_id(r.get('order_partner_id')) for r in g if r.get('order_partner_id')}


def main_categories():
    """Sorted list of main category names (for the 'bought in category' selector)."""
    return sorted(set(sales_mod._product_cat().values()))


# ====================================================================== WIN-BACK / SKU MOVES
def customer_winback(pid, t=None, team_id=None, top=10):
    """SKU-level YoY moves for one customer (this YTD vs same period last year):
      losers   — SKUs down YoY, each with its share of the customer's TOTAL gross loss (pct_of_loss)
      gainers  — SKUs up YoY (bought more)
      lapsed   — losers gone to zero this year (the re-pitch hooks)
      lapsed_categories — category-level gaps
      total_loss / this_total / last_total / net_change
    'change' is signed (this − last); negative = decline so the UI tints it red."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)

    def sku_map(df, dtt):
        g = O.read_group('sale.order.line', _line_dom(pid, df, dtt, team_id),
                         ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False)
        return {O.m2o_id(r['product_id']): {'name': O.m2o_name(r.get('product_id')),
                'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0}
                for r in g if r.get('product_id')}

    this, last = sku_map(yf, yt), sku_map(lf, lt)
    # product_id -> SKU code, so each row can deep-link to its SKU drawer (code is how the SKU
    # views key products; merged-by-code duplicates resolve to the same drawer)
    pids = list(set(this) | set(last))
    codemap = {}
    if pids:
        for pr in O.search_read('product.product', [('id', 'in', pids)], ['default_code']):
            codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None
    rows = []
    for p in (set(this) | set(last)):
        tv, lv = (this.get(p) or {}), (last.get(p) or {})
        tr, lr = tv.get('rev', 0.0), lv.get('rev', 0.0)
        tq, lq = tv.get('qty', 0.0), lv.get('qty', 0.0)
        name = (this.get(p) or last.get(p) or {}).get('name') or '(?)'
        p_ly = (lr / lq) if lq else None        # realised $/unit last year
        p_ytd = (tr / tq) if tq else None        # realised $/unit this year
        rows.append({'sku': name, 'code': codemap.get(p), 'ly_rev': lr, 'ytd_rev': tr,
                     'change': tr - lr, 'qty_ly': lq, 'qty_ytd': tq, 'qty_change': tq - lq,
                     'price_ly': p_ly, 'price_ytd': p_ytd,
                     'price_chg_pct': (((p_ytd - p_ly) / p_ly * 100) if (p_ly and p_ytd) else None),
                     # split the $ change into volume vs price effects (sum to change)
                     'vol_effect': ((tq - lq) * p_ly) if p_ly is not None else (tr - lr),
                     'price_effect': ((p_ytd - p_ly) * tq) if (p_ly is not None and p_ytd is not None) else 0.0})
    total_loss = sum(-r['change'] for r in rows if r['change'] < 0)
    for r in rows:
        r['pct_of_loss'] = (-r['change'] / total_loss * 100) if (r['change'] < 0 and total_loss) else 0.0
    losers = sorted([r for r in rows if r['change'] < 0], key=lambda x: x['change'])
    gainers = sorted([r for r in rows if r['change'] > 0], key=lambda x: -x['change'])
    lapsed = [r for r in losers if r['ytd_rev'] == 0]

    cat = sales_mod._product_cat()

    def cat_rev(m):
        agg = {}
        for p, v in m.items():
            agg[cat.get(p, '(uncategorized)')] = agg.get(cat.get(p, '(uncategorized)'), 0.0) + v['rev']
        return agg
    tcat, lcat = cat_rev(this), cat_rev(last)
    lapsed_categories = sorted([{'category': c, 'ly_rev': lcat[c], 'ytd_rev': tcat.get(c, 0.0),
                                 'change': tcat.get(c, 0.0) - lcat[c]}
                                for c in lcat if lcat[c] > tcat.get(c, 0.0)],
                               key=lambda x: x['change'])
    this_total = sum(v['rev'] for v in this.values())
    last_total = sum(v['rev'] for v in last.values())
    # 'all_skus' feeds the in-drawer SKU list; cap it so a mega-account doesn't ship thousands
    # of rows to the browser (the list is searchable/filtered, top-by-revenue is what matters).
    all_skus = sorted(rows, key=lambda x: -max(x['ytd_rev'], x['ly_rev']))[:500]
    return {
        'losers': losers[:top], 'gainers': gainers[:top], 'lapsed': lapsed[:top],
        'lapsed_categories': lapsed_categories[:6], 'all_skus': all_skus, 'n_skus': len(rows),
        'n_lapsed': len(lapsed), 'n_losers': len(losers), 'n_gainers': len(gainers),
        'total_loss': total_loss, 'lost_dollars': total_loss,
        'this_total': this_total, 'last_total': last_total, 'net_change': this_total - last_total,
    }


def customer_yoy_decomp(pid, t=None, team_id=None):
    """Decompose the customer's YoY sales change into a volume (order count) effect and a price
    (basket size / AOV) effect. Identity: ΔSales = ΔOrders·AOV_last + Orders_this·ΔAOV."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this_sales = O.sum_field('sale.order', _order_dom(pid, yf, yt, team_id), 'amount_untaxed')
    last_sales = O.sum_field('sale.order', _order_dom(pid, lf, lt, team_id), 'amount_untaxed')
    this_orders = O.get_odoo().search_count('sale.order', _order_dom(pid, yf, yt, team_id))
    last_orders = O.get_odoo().search_count('sale.order', _order_dom(pid, lf, lt, team_id))
    this_aov = (this_sales / this_orders) if this_orders else 0.0
    last_aov = (last_sales / last_orders) if last_orders else 0.0
    volume_effect = (this_orders - last_orders) * last_aov
    price_effect = this_orders * (this_aov - last_aov)
    return {
        'this_sales': this_sales, 'last_sales': last_sales, 'd_sales': this_sales - last_sales,
        'sales_yoy_pct': P.yoy_pct(this_sales, last_sales),
        'this_orders': this_orders, 'last_orders': last_orders, 'd_orders': this_orders - last_orders,
        'this_aov': this_aov, 'last_aov': last_aov, 'd_aov': this_aov - last_aov,
        'aov_yoy_pct': P.yoy_pct(this_aov, last_aov),
        'volume_effect': volume_effect, 'price_effect': price_effect,
    }


def customer_affinity(pid, t=None, team_id=None, top=12, max_cobuyers=60):
    """'Customers who buy similar products also buy …'. Finds the customers who bought this
    customer's SKUs (co-buyers), then ranks the OTHER SKUs those co-buyers buy (that this customer
    doesn't) by spend among them — a cross-sell list. LTM window + capped co-buyers to stay fast."""
    df, dtt = P.ltm(t)
    g = O.read_group('sale.order.line', _line_dom(pid, df, dtt, team_id),
                     ['price_subtotal:sum'], ['product_id'], lazy=False)
    mine = {O.m2o_id(r['product_id']) for r in g if r.get('product_id')}
    if not mine:
        return {'recs': [], 'n_cobuyers': 0, 'mine': 0}
    cg = O.read_group('sale.order.line',
                      O.sale_line_domain(df, dtt, team_id, extra=[('product_id', 'in', list(mine))]),
                      ['price_subtotal:sum'], ['order_partner_id'], lazy=False)
    cobuyers = sorted([(O.m2o_id(r['order_partner_id']), r.get('price_subtotal') or 0.0)
                       for r in cg if r.get('order_partner_id') and O.m2o_id(r['order_partner_id']) != pid],
                      key=lambda x: -x[1])[:max_cobuyers]
    cob_ids = [c[0] for c in cobuyers]
    if not cob_ids:
        return {'recs': [], 'n_cobuyers': 0, 'mine': len(mine)}
    # Single groupby over the co-buyers' lines (one aggregated row per product) — fast.
    pg = O.read_group('sale.order.line',
                      O.sale_line_domain(df, dtt, team_id, extra=[('order_partner_id', 'in', cob_ids)]),
                      ['price_subtotal:sum'], ['product_id'], lazy=False)
    recs = sorted([{'sku': O.m2o_name(r['product_id']), 'pid': O.m2o_id(r['product_id']),
                    'rev': r.get('price_subtotal') or 0.0, 'orders': r.get('__count') or 0}
                   for r in pg if r.get('product_id') and O.m2o_id(r['product_id']) not in mine],
                  key=lambda x: -x['rev'])[:top]
    if recs:
        codemap = {}
        for pr in O.search_read('product.product', [('id', 'in', [r['pid'] for r in recs])], ['default_code']):
            codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None
        for r in recs:
            r['code'] = codemap.get(r['pid'])
    return {'recs': recs, 'n_cobuyers': len(cob_ids), 'mine': len(mine)}


# ====================================================================== CADENCE / CHURN / BENCHMARK
def _order_dates(pid, team_id=None):
    """Recent confirmed order dates (date objects), ascending. Capped at the 3000 most recent so a
    very high-volume buyer can't hang the drawer — cadence/frequency only need recent orders."""
    rows = O.search_read('sale.order', _order_dom(pid, None, None, team_id),
                         ['date_order'], order='date_order desc', limit=3000)
    return sorted(dt.date.fromisoformat(str(r['date_order'])[:10]) for r in rows if r.get('date_order'))


def customer_cadence(pid, t=None, team_id=None):
    """Reorder rhythm from the gaps between orders: typical gap (median of recent), predicted next
    order, days overdue vs that rhythm, and whether the rhythm is slowing (last-4 vs prior-4 gap)."""
    t = t or P.today()
    dates = _order_dates(pid, team_id)
    n = len(dates)
    out = {'n_orders': n, 'median_gap_days': None, 'last_order': dates[-1].isoformat() if dates else None,
           'predicted_next': None, 'overdue_days': None, 'drift_pct': None,
           'recent_gap': None, 'prior_gap': None, 'dates': [d.isoformat() for d in dates]}
    if n < 2:
        return out
    gaps = [(dates[i] - dates[i - 1]).days for i in range(1, n)]
    median_gap = statistics.median(gaps[-8:])
    last = dates[-1]
    predicted = last + dt.timedelta(days=round(median_gap))
    recent_gap = statistics.mean(gaps[-4:]) if len(gaps) >= 4 else statistics.mean(gaps)
    prior_gap = statistics.mean(gaps[-8:-4]) if len(gaps) >= 8 else None
    out.update({'median_gap_days': median_gap, 'predicted_next': predicted.isoformat(),
                'overdue_days': (t - predicted).days, 'recent_gap': recent_gap, 'prior_gap': prior_gap,
                'drift_pct': ((recent_gap - prior_gap) / prior_gap * 100) if prior_gap else None})
    return out


def customer_churn_score(pid, t=None, team_id=None, cad=None, sales_yoy=None):
    """0–100 churn-risk score (higher = more at risk): overdue-vs-cadence (50%) + frequency decay
    last-90 vs prior-90 (30%) + YoY sales trend (20%). Bucketed Healthy/Watch/At-risk/Critical."""
    t = t or P.today()
    cad = cad or customer_cadence(pid, t, team_id)
    mg = cad.get('median_gap_days')
    if mg and cad.get('overdue_days') is not None:
        s_overdue = min(1.0, max(0.0, cad['overdue_days'] / mg) / 2.0)   # 2 cycles late = max
    else:
        s_overdue = 0.5
    dates = [dt.date.fromisoformat(d) for d in cad.get('dates', [])]
    last90 = sum(1 for d in dates if (t - d).days <= 90)
    prior90 = sum(1 for d in dates if 90 < (t - d).days <= 180)
    if prior90 == 0:
        # recent activity but no 90–180d baseline (reactivated / new) → cautious mid risk, not zero
        s_freq = 0.6 if last90 == 0 else 0.3
    else:
        s_freq = min(1.0, max(0.0, (prior90 - last90) / prior90))
    if sales_yoy is None:
        sales_yoy = customer_yoy_decomp(pid, t, team_id)['sales_yoy_pct']
    s_yoy = 0.5 if sales_yoy is None else min(1.0, max(0.0, -sales_yoy / 50.0))   # -50% YoY = max
    score = 100 * (0.5 * s_overdue + 0.3 * s_freq + 0.2 * s_yoy)
    bucket = ('Critical' if score >= 70 else 'At-risk' if score >= 45
              else 'Watch' if score >= 25 else 'Healthy')
    driver = max([('overdue rhythm', s_overdue), ('fewer recent orders', s_freq),
                  ('falling spend', s_yoy)], key=lambda x: x[1])[0]
    return {'score': round(score), 'bucket': bucket, 'driver': driver}


def _pctile(values, x):
    """Percentile rank (0–100) of x within values — mean/midpoint method (ties count as half), so
    the median of a peer set lands at the 50th percentile."""
    if not values:
        return None
    below = sum(1 for v in values if v < x)
    equal = sum(1 for v in values if v == x)
    return (below + 0.5 * equal) / len(values) * 100


def rank_contribution(directory_rows, pid):
    """Rank by YTD revenue + % of BU YTD — pure, from the already-cached directory list."""
    total = sum(r['revenue'] for r in directory_rows) or 1.0
    ranked = sorted(directory_rows, key=lambda r: -r['revenue'])
    rank = next((i + 1 for i, r in enumerate(ranked) if r['pid'] == pid), None)
    me = next((r for r in directory_rows if r['pid'] == pid), None)
    return {'rank': rank, 'n': len(directory_rows),
            'pct_of_bu': (me['revenue'] / total * 100) if me else None}


def peer_benchmark(directory_rows, pid):
    """Percentile rank vs same value-tier peers on revenue, AOV, frequency, YoY — pure."""
    me = next((r for r in directory_rows if r['pid'] == pid), None)
    if not me:
        return None
    tier = _tier(me['revenue'])
    peers = [r for r in directory_rows if _tier(r['revenue']) == tier]
    yoy_vals = [r['yoy_pct'] for r in peers if r['yoy_pct'] is not None]
    return {'cohort_n': len(peers), 'tier': tier,
            'rev_pctile': _pctile([r['revenue'] for r in peers], me['revenue']),
            'aov_pctile': _pctile([r['aov'] for r in peers], me['aov']),
            'freq_pctile': _pctile([r['orders'] for r in peers], me['orders']),
            'yoy_pctile': (_pctile(yoy_vals, me['yoy_pct']) if me['yoy_pct'] is not None else None)}


def customer_whitespace(mine_categories, company_categories, top=6):
    """Pure: breadth (# categories bought / company total) + the biggest company categories this
    customer buys $0 of, ranked by company revenue. `company_categories` = sales.by_category rows."""
    skip = {'(uncategorized)', 'All'}
    mine = {c for c in mine_categories if c not in skip}
    cats = [c for c in company_categories if c['category'] not in skip and c['revenue'] > 0]
    total = len(cats)
    ws = sorted([c for c in cats if c['category'] not in mine], key=lambda x: -x['revenue'])[:top]
    return {'breadth_x': len(mine), 'breadth_y': total, 'whitespace': ws}


# ====================================================================== PAGE-LEVEL: NRR / MIGRATION
def nrr(t=None, team_id=None, agent_pids=None):
    """Net Revenue Retention from the revenue bridge (existing book only; new logos excluded)."""
    b = revenue_bridge(t, team_id, agent_pids=agent_pids)
    last = b['last_total'] or 1.0
    ending = b['last_total'] + b['expansion']['rev'] + b['contraction']['rev'] + b['lost']['rev']
    return {'nrr_pct': ending / last * 100, 'starting': b['last_total'],
            'expansion': b['expansion']['rev'], 'contraction': b['contraction']['rev'],
            'churned': b['lost']['rev'], 'ending_existing': ending, 'new': b['new']['rev']}


def tier_migration(t=None, team_id=None, agent_pids=None):
    """Value-tier flows LTM vs prior-LTM: upgraded / held / downgraded / new / lapsed + net $."""
    lf, lt = P.ltm(t)
    pf, pt = P.prior_ltm(t)
    this = _cust_rev(lf, lt, team_id, agent_pids)
    last = _cust_rev(pf, pt, team_id, agent_pids)
    order = {name: i for i, (name, _) in enumerate(TIERS)}   # 0 = Whale (top) … 3 = Small
    flows = {k: {'flow': k, 'n': 0, 'net': 0.0} for k in
             ['Upgraded', 'Held', 'Downgraded', 'New / reactivated', 'Lapsed']}
    for pid in set(this) | set(last):
        tr = this.get(pid, {}).get('rev', 0.0)
        lr = last.get(pid, {}).get('rev', 0.0)
        if pid in this and pid in last:
            k = ('Upgraded' if order[_tier(tr)] < order[_tier(lr)]
                 else 'Downgraded' if order[_tier(tr)] > order[_tier(lr)] else 'Held')
        elif pid in this:
            k = 'New / reactivated'
        else:
            k = 'Lapsed'
        flows[k]['n'] += 1
        flows[k]['net'] += (tr - lr)
    return {'flows': list(flows.values())}


# ====================================================================== GROUP DRAWER (rollup → drawer)
def group_detail(dim, value, t=None, team_id=None, n_months=13, top=15, agent_pids=None):
    """For one rollup group (e.g. dim='city', value='Brooklyn'): KPIs, monthly trend, the customers
    in the group (clickable), and the top SKUs sold there. dim is a DIMENSIONS key."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    attrs = _partner_attrs(set(this) | set(last))

    def keyfor(pid):
        if dim == 'segment':
            return _tier(this.get(pid, {}).get('rev', last.get(pid, {}).get('rev', 0.0)))
        return (attrs.get(pid) or {}).get(dim) or '(none)'
    pids = [pid for pid in (set(this) | set(last)) if keyfor(pid) == value]
    if not pids:
        return None

    # KPI totals and the customer list cover EVERY member of the group (cheap, in-memory).
    rev_ytd = sum(this.get(p, {}).get('rev', 0.0) for p in pids)
    rev_ly = sum(last.get(p, {}).get('rev', 0.0) for p in pids)
    custs = sorted([{'pid': p, 'customer': (this.get(p) or last.get(p) or {}).get('name', '?'),
                     'revenue': this.get(p, {}).get('rev', 0.0), 'orders': this.get(p, {}).get('orders', 0),
                     'yoy_pct': P.yoy_pct(this.get(p, {}).get('rev', 0.0), last.get(p, {}).get('rev', 0.0)),
                     'city': (attrs.get(p) or {}).get('city', '(none)'),
                     'state': (attrs.get(p) or {}).get('state', '(none)'),
                     'agent': (attrs.get(p) or {}).get('agent', '(none)')}
                    for p in pids], key=lambda x: -x['revenue'])

    # The trend + top-SKU read_groups put every pid in an IN(...) clause, so a huge group
    # (a big city, the '(none)' bucket) would build a slow, oversized query. Cap those two
    # queries to the group's top buyers by revenue; KPIs/customer list above stay complete.
    QCAP = 300
    capped = len(pids) > QCAP
    qpids = sorted(
        pids, key=lambda p: -max(this.get(p, {}).get('rev', 0.0), last.get(p, {}).get('rev', 0.0))
    )[:QCAP] if capped else pids

    range_start = dt.date(t.year - 2, t.month, 1).isoformat()
    g = O.read_group('sale.order',
                     sales_mod.order_domain(range_start, t.isoformat(), team_id) + [('partner_id', 'in', qpids)],
                     ['amount_untaxed:sum'], ['date_order:month'], lazy=False)
    mrev = {}
    for r in g:
        rng = (r.get('__range') or {}).get('date_order:month') or {}
        ym = (rng.get('from') or '')[:7]
        if ym:
            mrev[ym] = r.get('amount_untaxed') or 0.0
    monthly = []
    for ym, _s, _e in P.month_starts(n_months, t):
        y, m = int(ym[:4]) - 1, int(ym[5:7])
        monthly.append({'month': ym, 'revenue': mrev.get(ym, 0.0),
                        'revenue_ly': mrev.get(f'{y:04d}-{m:02d}', 0.0)})

    mf, mt = P.ltm(t)
    lg = O.read_group('sale.order.line',
                      O.sale_line_domain(mf, mt, team_id, extra=[('order_partner_id', 'in', qpids)]),
                      ['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'], ['product_id'], lazy=False)
    skus = sorted([sales_mod._sku_profit(
                   {'product': O.m2o_name(r.get('product_id')), 'pid': O.m2o_id(r.get('product_id')),
                    'rev': r.get('price_subtotal') or 0.0, 'revenue': r.get('price_subtotal') or 0.0,
                    'qty': r.get('product_uom_qty') or 0.0,
                    'margin': r.get('margin') or 0.0, 'lines': r.get('__count') or 0})
                   for r in lg if r.get('product_id')],
                  key=lambda x: -x['rev'])[:top]
    if skus:   # attach SKU code for deep-linking each to its SKU drawer
        scode = {}
        for pr in O.search_read('product.product', [('id', 'in', [s['pid'] for s in skus])], ['default_code']):
            scode[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None
        for s in skus:
            s['code'] = scode.get(s['pid'])
    return {'dim': dim, 'value': value, 'rev_ytd': rev_ytd, 'rev_ly': rev_ly,
            'yoy_pct': P.yoy_pct(rev_ytd, rev_ly), 'n_customers': len([p for p in pids if p in this]),
            'n_total': len(pids), 'monthly': monthly, 'top_skus': skus, 'customers': custs,
            'capped': capped, 'qcap': QCAP}


# ----------------------------------------------------------- KPI-card customer sets
_KPI_SETS = {
    'active': 'Active customers (YTD)', 'new': 'New / reactivated (YTD)',
    'lost': 'Lost customers (YTD)', 'retained': 'Retained customers (YTD)',
    'existing': 'Existing book (NRR base)',
}


def customer_set(kind, t=None, team_id=None, agent_pids=None):
    """Standardized customer list behind a headline KPI card (Active / New / Lost / Retained /
    Existing-book), each row carrying the same shape (pid, customer, rev_ytd, rev_ly, change,
    yoy_pct, orders, city, state, agent) so the drawer renders them with one standardized view."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _cust_rev(yf, yt, team_id, agent_pids)
    last = _cust_rev(lf, lt, team_id, agent_pids)
    tset, lset = set(this), set(last)
    ids = {'active': tset, 'new': tset - lset, 'lost': lset - tset,
           'retained': tset & lset, 'existing': lset}.get(kind, tset)
    attrs = _partner_attrs(list(ids))
    rows = []
    for p in ids:
        tr = this.get(p, {}).get('rev', 0.0)
        lr = last.get(p, {}).get('rev', 0.0)
        a = attrs.get(p, {})
        rows.append({'pid': p, 'customer': (this.get(p) or last.get(p) or {}).get('name', '?'),
                     'rev_ytd': tr, 'rev_ly': lr, 'change': tr - lr, 'yoy_pct': P.yoy_pct(tr, lr),
                     'orders': this.get(p, {}).get('orders', 0),
                     'city': a.get('city', '(none)'), 'state': a.get('state', '(none)'),
                     'agent': a.get('agent', '(none)')})
    rows.sort(key=lambda r: -(r['rev_ly'] if kind == 'lost' else r['rev_ytd']))
    return {'kind': kind, 'label': _KPI_SETS.get(kind, kind), 'rows': rows, 'n': len(rows),
            'rev_ytd': sum(r['rev_ytd'] for r in rows), 'rev_ly': sum(r['rev_ly'] for r in rows)}


# ----------------------------------------------------------- VALIDATION
def validate(t=None, team_id=None):
    """Reconcile every headline number to an independent Odoo aggregate.

    When team_id is set (a single BU is selected) the checks run SCOPED to that BU so the
    validation panel never exposes other-BU figures — BU isolation holds even here. The two
    cross-BU brand-mirror checks (#2, #4b) only make sense consolidated, so they run only
    when team_id is None.
    """
    t = t or P.today()
    checks = []

    # 1. Bridge identity: last + new + expansion + contraction + lost == this
    b = revenue_bridge(t, team_id=team_id)
    recon = (b['last_total'] + b['new']['rev'] + b['expansion']['rev']
             + b['contraction']['rev'] + b['lost']['rev'])
    checks.append({'check': 'Revenue bridge reconciles last→this (New+Exp+Contr+Lost)',
                   'a': round(recon, 2), 'b': round(b['this_total'], 2),
                   'gap': round(recon - b['this_total'], 2),
                   'ok': abs(recon - b['this_total']) <= 1.0})

    # 2. Σ(brand this_total) == company this_total  (consolidated only — cross-BU)
    if team_id is None:
        brand_this = sum(r['this'] for r in bridge_by_brand(t))
        checks.append({'check': 'Σ(brand YTD) == company YTD',
                       'a': round(brand_this, 2), 'b': round(b['this_total'], 2),
                       'gap': round(brand_this - b['this_total'], 2),
                       'ok': abs(brand_this - b['this_total']) <= 1.0})

    # 3. Σ(segment LTM rev) == total LTM rev
    lf, lt = P.ltm(t)
    seg_sum = sum(s['revenue'] for s in segments(t, team_id=team_id))
    ltm_total = sum(v['rev'] for v in _cust_rev(lf, lt, team_id=team_id).values())
    checks.append({'check': 'Σ(segment LTM rev) == total LTM rev',
                   'a': round(seg_sum, 2), 'b': round(ltm_total, 2),
                   'gap': round(seg_sum - ltm_total, 2),
                   'ok': abs(seg_sum - ltm_total) <= 1.0})

    # 4. Sum-level rollup is MECE: Σ(by_dimension('agent')) == total YTD revenue
    yf, yt = P.ytd(t)
    ytd_total = sum(v['rev'] for v in _cust_rev(yf, yt, team_id=team_id).values())
    roll = by_dimension('agent', t, team_id=team_id)
    dim_sum = sum(r['revenue'] for r in roll)
    checks.append({'check': 'Σ(agent rollup) == total YTD revenue',
                   'a': round(dim_sum, 2), 'b': round(ytd_total, 2),
                   'gap': round(dim_sum - ytd_total, 2),
                   'ok': abs(dim_sum - ytd_total) <= 1.0})

    # 4b. Brand mirror reconciles: Σ(Fisch)+Σ(Royal) == total YTD  (consolidated only)
    if team_id is None:
        brand_sum = sum(r['fisch'] + r['royal'] for r in roll)
        checks.append({'check': 'Rollup brand mirror: Σ(Fisch)+Σ(Royal) == total YTD',
                       'a': round(brand_sum, 2), 'b': round(ytd_total, 2),
                       'gap': round(brand_sum - ytd_total, 2),
                       'ok': abs(brand_sum - ytd_total) <= 1.0})

    # 5. Drill-down scope holds: top customer's LTM order-level rev == line-level rev
    top = directory(t, team_id=team_id, limit=1)
    if top:
        pid = top[0]['pid']
        mf, mt = P.ltm(t)
        ord_rev = O.sum_field('sale.order', _order_dom(pid, mf, mt), 'amount_untaxed')
        line_rev = O.sum_field('sale.order.line', _line_dom(pid, mf, mt), 'price_subtotal')
        checks.append({'check': 'Drill-down: top customer LTM order==line revenue',
                       'a': round(ord_rev, 2), 'b': round(line_rev, 2),
                       'gap': round(ord_rev - line_rev, 2),
                       'ok': abs(ord_rev - line_rev) <= max(1.0, 0.005 * ord_rev)})

    # 6. YoY decomposition identity: volume effect + price effect == Δsales (drawer math)
    if top:
        dc = customer_yoy_decomp(top[0]['pid'], t, team_id=team_id)
        recon = dc['volume_effect'] + dc['price_effect']
        checks.append({'check': 'Drawer: volume + price effect == Δsales (top customer)',
                       'a': round(recon, 2), 'b': round(dc['d_sales'], 2),
                       'gap': round(recon - dc['d_sales'], 2),
                       'ok': abs(recon - dc['d_sales']) <= max(1.0, 0.005 * abs(dc['d_sales']) + 1)})

    # 7. NRR ties to the bridge: ending-existing == last + expansion + contraction + churned
    nr = nrr(t, team_id=team_id)
    nrr_recon = nr['starting'] + nr['expansion'] + nr['contraction'] + nr['churned']
    checks.append({'check': 'NRR: ending-existing == start + exp + contr + churn',
                   'a': round(nrr_recon, 2), 'b': round(nr['ending_existing'], 2),
                   'gap': round(nrr_recon - nr['ending_existing'], 2),
                   'ok': abs(nrr_recon - nr['ending_existing']) <= 1.0})

    # 8. Churn score bounded [0,100] with a valid bucket (drawer pill)
    if top:
        cs = customer_churn_score(top[0]['pid'], t, team_id=team_id)
        ok = 0 <= cs['score'] <= 100 and cs['bucket'] in ('Healthy', 'Watch', 'At-risk', 'Critical')
        checks.append({'check': f"Churn score in [0,100] with valid bucket ({cs['bucket']})",
                       'a': cs['score'], 'b': cs['score'], 'gap': 0, 'ok': ok})

    # 9. Tier migration is MECE: Σ(flow customers) == distinct customers active in either window
    tm = tier_migration(t)
    lf, lt = P.ltm(t)
    pf, pt = P.prior_ltm(t)
    active = len(set(_cust_rev(lf, lt)) | set(_cust_rev(pf, pt)))
    flow_n = sum(f['n'] for f in tm['flows'])
    checks.append({'check': 'Tier migration: Σ(flow customers) == active (LTM ∪ prior-LTM)',
                   'a': flow_n, 'b': active, 'gap': flow_n - active, 'ok': flow_n == active})
    return checks