platform/modules/products.py

"""Products / SKU module — SKU health: YoY movers (risers & decliners), zombie SKUs
(catalog rot — sellable, formerly selling, now dead), new winners, coverage collapse
(SKUs losing customer breadth — the FFS-recovery early-warning signal), and velocity leaders.

Per-SKU margin already lives in the Financial module (low_margin_skus); not duplicated here.
Basket / co-purchase (MBA) is intentionally deferred — the existing client app computes it
runtime-side, and a local co-occurrence pull over 74k LTM lines is the kind of heavy job the
project guardrails keep off the local PC. (See BACKLOG.)

All line-level (sale.order.line), RI+FFS scope, excluded accounts removed — reusing sale_line_domain.
Coverage (distinct customers per SKU) uses a 2-level read_group and is a touch slow (~15s/
window), so it's its own function the app calls lazily and caches.
"""
import sys
import datetime as dt
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from functools import lru_cache
import core.odoo as O
import core.periods as P
import modules.sales as sales_mod

# SKU-health views are about real products — exclude service/delivery pseudo-SKUs
# (Delivery Charges otherwise dominate movers/winners). Dot-path FILTER works (groupby
# on the dot-path does not). Validation reconciles on this same non-service universe.
_NO_SVC = [('product_id.type', '!=', 'service')]


@lru_cache(maxsize=1)
def _code_map():
    """product_id → SKU code. Multiple product records can share a default_code (re-SKUing
    /duplicates); keying by code merges them so a re-coded item doesn't read as a fake
    decliner + fake riser. Products without a code fall back to a per-id key. Archived
    products are INCLUDED — re-SKUing typically archives the old record and creates a new
    one under the same code, and the old record still carries last-year sales."""
    prods = O.search_read('product.product', [('active', 'in', [True, False])], ['id', 'default_code'],
                          limit=50000)
    return {p['id']: (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
            for p in prods}


def _sku_rev(date_from, date_to, team_id=None):
    """{sku_code: {'name','rev','qty','orders'}} over a window (services excluded,
    duplicate product records merged by SKU code)."""
    g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC),
                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False)
    codes = _code_map()
    out = {}
    for r in g:
        pid = O.m2o_id(r.get('product_id'))
        if not pid:
            continue
        key = codes.get(pid, f"pid:{pid}")
        e = out.setdefault(key, {'name': O.m2o_name(r.get('product_id')),
                                 'rev': 0.0, 'qty': 0.0, 'orders': 0})
        e['rev'] += r.get('price_subtotal') or 0.0
        e['qty'] += r.get('product_uom_qty') or 0.0
        e['orders'] += r.get('__count') or 0
    return out


def yoy_movers(t=None, limit=20, team_id=None):
    """Top SKU risers and decliners by YTD-vs-same-period-LY revenue change."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _sku_rev(yf, yt, team_id)
    last = _sku_rev(lf, lt, team_id)
    rows = []
    for pid in set(this) | set(last):
        tr = this.get(pid, {'rev': 0.0, 'name': last.get(pid, {}).get('name', '')})
        lr = last.get(pid, {'rev': 0.0})
        name = this.get(pid, {}).get('name') or last.get(pid, {}).get('name') or ''
        rows.append({'code': pid, 'product': name, 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'],
                     'change': tr['rev'] - lr['rev']})
    risers = sorted([r for r in rows if r['change'] > 0], key=lambda x: -x['change'])[:limit]
    decliners = sorted([r for r in rows if r['change'] < 0], key=lambda x: x['change'])[:limit]
    return {'risers': risers, 'decliners': decliners}


def zombie_skus(t=None, limit=30, min_prior=1500.0, team_id=None):
    """Catalog rot: SKUs that sold materially last year but are ~dead this year."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _sku_rev(yf, yt, team_id)
    last = _sku_rev(lf, lt, team_id)
    rows = []
    for pid, lr in last.items():
        if lr['rev'] < min_prior:
            continue
        tr = this.get(pid, {'rev': 0.0})
        if tr['rev'] > 0.05 * lr['rev']:   # still selling at >5% of prior → not a zombie
            continue
        rows.append({'code': pid, 'product': lr['name'], 'rev_ly': lr['rev'], 'rev_ytd': tr['rev'],
                     'lost': lr['rev'] - tr['rev']})
    rows.sort(key=lambda x: -x['lost'])
    return rows[:limit]


def new_winners(t=None, limit=20, min_this=1500.0, team_id=None):
    """SKUs that barely sold last year but are selling well this year."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _sku_rev(yf, yt, team_id)
    last = _sku_rev(lf, lt, team_id)
    rows = []
    for pid, tr in this.items():
        if tr['rev'] < min_this:
            continue
        lr = last.get(pid, {'rev': 0.0})
        if lr['rev'] > 0.05 * tr['rev']:
            continue
        rows.append({'code': pid, 'product': tr['name'], 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'],
                     'gained': tr['rev'] - lr['rev']})
    rows.sort(key=lambda x: -x['gained'])
    return rows[:limit]


def velocity_leaders(t=None, limit=25, team_id=None):
    """Top SKUs by LTM unit velocity (units/month) and revenue."""
    t = t or P.today()
    lf, lt = P.ltm(t)
    sku = _sku_rev(lf, lt, team_id)
    rows = [{'code': k, 'product': v['name'], 'units_ltm': v['qty'], 'units_per_mo': v['qty'] / 12.0,
             'rev_ltm': v['rev'], 'orders_ltm': v['orders']} for k, v in sku.items()]
    rows.sort(key=lambda x: -x['units_ltm'])
    return rows[:limit]


def _code_category():
    """code -> category name. Built from _code_map (pid->code) + the sales category map
    (pid->category); the first record carrying a code sets that code's category."""
    codes = _code_map()
    cats = sales_mod._product_cat()
    out = {}
    for pid, code in codes.items():
        if code not in out and pid in cats:
            out[code] = cats[pid]
    return out


def directory(t=None, team_id=None):
    """Every SKU that sold this YTD or the same period last year, with the fields the SKU
    drill-down filters/sorts on: category, YTD/LY revenue, YoY %, units and orders. Mirrors the
    customer directory — the full list (filtering happens in the UI), sorted by YTD revenue."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this = _sku_rev(yf, yt, team_id)
    last = _sku_rev(lf, lt, team_id)
    code_cat = _code_category()
    rows = []
    for code in set(this) | set(last):
        tr = this.get(code, {})
        lr = last.get(code, {})
        rev_ytd = tr.get('rev', 0.0)
        rev_ly = lr.get('rev', 0.0)
        rows.append({'code': code, 'product': tr.get('name') or lr.get('name') or code,
                     'category': code_cat.get(code, '(uncategorized)'),
                     'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'change': rev_ytd - rev_ly,
                     'yoy_pct': P.yoy_pct(rev_ytd, rev_ly),
                     'qty_ytd': tr.get('qty', 0.0), 'orders_ytd': tr.get('orders', 0)})
    rows.sort(key=lambda r: -r['rev_ytd'])
    return rows


def catalogue():
    """`{sku_code: {'product', 'category'}}` — EVERY ACTIVE product, sold or not.

    ⛔ THIS IS THE CATALOGUE UNIVERSE, AND IT IS DELIBERATELY NOT `directory()`. `directory()`'s
    row set IS the union of two revenue `read_group`s over `sale.order.line` (`:155`), so a SKU
    that never sold cannot exist in it. That is CORRECT for its own callers — `yoy_movers`,
    `zombie_skus`, `new_winners` and `categories` all legitimately want a sales-window universe —
    and it is wrong for the PRODUCT GRID, which is a catalogue and was therefore showing 2,717 of
    5,875 SKUs (wave 29, owner item 22 / ruling R12). The fix is this function plus a LEFT JOIN in
    the consumer, never a window removed from `directory()`: removing the window alone lands at
    3,327 (all-time-sold), because ~2,550 active SKUs have never sold in wholesale scope at all.

    ⚠ NOT BU-SHAPED, and it cannot be: `product.product` carries no team. A catalogue is one
    catalogue. `directory()` stays the BU-shaped half, which is why the two are joined rather than
    merged — a scoped caller gets every SKU with ITS OWN revenue, blank where that BU never sold.

    Keyed exactly like `_code_map()` — the `default_code`, or a `pid:N` fallback for the 33 active
    records that carry none — so the join against `directory()` is code-for-code with no
    normaliser. Names come from `display_name` (`[CODE] NAME`), which is what `O.m2o_name` yields
    off a sale line, so a never-sold row wears the same format as a sold one.

    ⛔ RAISES on a truncated read rather than returning a short catalogue. A silently short pull
    would put the grid back at a plausible wrong number with every gate green — the exact failure
    this function exists to end ([[no-unverifiable-aggregates]], and the same truncation guard
    `modules/backorders.py:161` already uses).
    """
    dom = [('active', '=', True)]
    prods = O.search_read('product.product', dom, ['id', 'default_code', 'display_name', 'name'],
                          limit=50000)
    n = O.get_odoo().search_count('product.product', dom)
    if len(prods) != n:
        raise ValueError(
            f"products.catalogue: the product pull is TRUNCATED — read {len(prods)} rows against "
            f"a search_count of {n}. A short catalogue renders as a plausible smaller grid with "
            f"nothing reporting it; raise the limit before shipping this.")
    code_cat = _code_category()
    out = {}
    for p in prods:
        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
        # First record wins, matching `_code_category`'s own convention. MEASURED 2026-08-11: zero
        # active products share a code, so this branch is a guard, not a merge policy.
        out.setdefault(code, {'product': p.get('display_name') or p.get('name') or code,
                              'category': code_cat.get(code, '(uncategorized)')})
    return out


#: ⭐⭐ WAVE 30 / W30-T34 (the carried W29-T52) — THE PRICELIST STRATUM, PER LIST.
#:
#: MEASURED LIVE 2026-08-12, and every one of these numbers shaped the design rather than
#: decorating it:
#:
#:   · **Five pricelists, three of them material.** Date-valid fixed rules: Fisch 5,139,
#:     Royal 1 2,702, Royal 2 2,605, Public Pricelist 17, Giftware Deals 1. At SKU level that
#:     is Fisch 3,971 · Royal 1 1,920 · Royal 2 1,857 · Public 6 · Giftware 0.
#:   · **3,988 of 5,871 active products carry a usable fixed rule; 1,883 carry NONE.**
#:   · **1,853 SKUs are priced on all three lists AND THE LISTS DISAGREE** — median relative
#:     spread 16%, max 82%. That is why this is three columns rather than one "the price" or a
#:     low/high pair: neither can tell a Fisch rep what Fisch sells the SKU for, which is the
#:     only question the column exists to answer.
#:   · ⛔ **`list_price` is 1.00 on 5,817 of the 5,871** — so the single `3_global` rule
#:     computed over it is meaningless, exactly as W29-T52 said. **This reader never falls
#:     through to it**, and that is the ticket's own negative control: a SKU with no specific
#:     item renders BLANK, never a fallback dressed as a price.
#:
#: ⚠ RESOLVED BY NAME, NOT BY ID. A pricelist renamed or deleted in Odoo makes its column go
#: blank and makes `report["lists_missing"]` name it — a hard-coded id would keep pointing at
#: whatever inherited it and mislabel every cell silently.
PRICELIST_COLUMNS = (
    ("price_fisch", "Fisch"),
    ("price_royal_1", "Royal 1"),
    ("price_royal_2", "Royal 2"),
)


def pricelist_by_code():
    """`({code: {column_key: price}}, report)` — the date-valid FIXED base-tier price for each
    declared pricelist, keyed EXACTLY as `catalogue()` keys its rows.

    ⛔ THE KEYING IS LOAD-BEARING, not a detail. `catalogue()` keys on `default_code` with a
    `pid:{id}` fallback for the ~33 active records that carry none. Keying this map any other
    way would leave those SKUs permanently blank while an oracle counting active products
    counted them — a red gate on a working build, or worse, a silent hole nobody counts.

    **The base tier, deliberately.** `pricecomp._tier_for` picks the highest `min_quantity` at
    or below an order's quantity, because it is pricing a LINE that has one. A catalogue column
    has no quantity in hand, so it takes the LOWEST `min_quantity` — the price at qty 1. Rules
    above that break are a bulk price, and `report["qty_break_only"]` counts the SKUs whose only
    rule sits on one (MEASURED: 10 rules of 10,464 carry a break at all).

    **Variant rules beat template rules**, matching `pricecomp._tier_for`: a `0_product_variant`
    rule is the more specific statement about this exact SKU.

    ⚠ DEGRADES TO `({}, report)` on a read failure, matching `_inventory_by_code` rather than
    `catalogue()`: these are COLUMNS, and a product grid that will not render because pricing is
    momentarily unreachable is a worse failure than one with blank price columns. The blanks are
    not silent — `product_data.validate()`'s coverage leg reconciles each column against a fresh
    Odoo count and goes red at zero.

    ⭐ **The report exists because R6's second sentence is law** (*"if it can't be done, you need
    to explicitly tell me why and recommend a fix"*). Everything this reader CANNOT see is
    counted rather than dropped: rules on pricelists the contract does not declare, non-fixed
    (`formula`/`percent`) rules, the `3_global` fallback, and prices that only exist above a
    quantity break.
    """
    # ⚠ THE COUNTERS ARE DISJOINT AND ORDER-DEPENDENT, and saying so is the difference between a
    # report and a misleading one. A rule is classified ONCE, by the first reason it is skipped:
    # undeclared list → out of date → not fixed → global → zero price. So `rules_not_fixed: 0`
    # means "no formula rule on a list we declare", NOT "this Odoo has no formula rules"
    # (MEASURED 2026-08-12: it has exactly one, and it sits on Public Pricelist, which the
    # contract does not declare — so it lands in `rules_undeclared_list`).
    report = {"lists_missing": [], "rules_total": 0, "rules_undeclared_list": 0,
              "rules_not_fixed": 0, "rules_global": 0, "rules_out_of_date": 0,
              "rules_zero_price": 0, "qty_break_only": 0}
    try:
        pl_rows = O.search_read('product.pricelist', [], ['id', 'name'])
        by_name = {}
        for p in pl_rows:
            by_name.setdefault(str(p.get('name') or '').strip(), p['id'])
        wanted = {}
        for col, name in PRICELIST_COLUMNS:
            pid = by_name.get(name)
            if pid is None:
                report["lists_missing"].append(name)
            else:
                wanted[pid] = col

        # ⭐⭐ FILTERED SERVER-SIDE, AND THE REASON IS A MEASUREMENT, NOT A STYLE PREFERENCE.
        # Reading all 10,464 rules with 9 fields and sorting them in Python costs **30.1s** on
        # this connection; the same rules under a server-side domain with 6 fields cost
        # **13.6s** (measured 2026-08-12, back to back). `pool()` runs this on a scope's
        # first-ever build, so that 16.5s is 16.5s of somebody's page load.
        #
        # ⚠ The domain reproduces the Python predicate EXACTLY, and the null legs are the part
        # that is easy to get wrong: an absent `date_start` is `False`, not a past date, so
        # `('date_start','<=',today)` ALONE would drop every open-ended rule — which is almost
        # all of them.
        today = P.today().isoformat()
        _declared = sorted(wanted)
        _kinds = ['0_product_variant', '1_product']
        _live = [('pricelist_id', 'in', _declared), ('compute_price', '=', 'fixed'),
                 ('applied_on', 'in', _kinds),
                 '|', ('date_start', '=', False), ('date_start', '<=', today),
                 '|', ('date_end', '=', False), ('date_end', '>=', today),
                 ('fixed_price', '>', 0)]
        rules = O.search_read(
            'product.pricelist.item', _live,
            ['pricelist_id', 'product_id', 'product_tmpl_id', 'applied_on', 'fixed_price',
             'min_quantity']) if _declared else []

        # ⭐ R6's SECOND SENTENCE, PAID FOR WITH `search_count` RATHER THAN A WIDER READ. Six
        # counts cost ~1.2s together; the rules they count would cost 16s to read. What this
        # reader cannot see is still REPORTED — it is just no longer transferred.
        def _n(extra):
            try:
                return O.get_odoo().search_count('product.pricelist.item', extra)
            except Exception:
                return -1                      # -1 reads as "not measured", never as zero
        _dated = ['|', ('date_start', '=', False), ('date_start', '<=', today),
                  '|', ('date_end', '=', False), ('date_end', '>=', today)]
        report["rules_total"] = _n([])
        report["rules_undeclared_list"] = _n([('pricelist_id', 'not in', _declared)]) \
            if _declared else report["rules_total"]
        if _declared:
            _on = [('pricelist_id', 'in', _declared)]
            # `percent_price` and formula rules are read NOWHERE in this repo. A rule we cannot
            # price is one the operator is TOLD about, never one that quietly becomes a blank.
            report["rules_not_fixed"] = _n(_on + [('compute_price', '!=', 'fixed')])
            report["rules_global"] = _n(_on + [('compute_price', '=', 'fixed'),
                                               ('applied_on', 'not in', _kinds)])
            report["rules_out_of_date"] = (
                _n(_on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)])
                - _n(_on + [('compute_price', '=', 'fixed'),
                            ('applied_on', 'in', _kinds)] + _dated))
            # A zero price is not "free" — it is an unset rule. Blank says so; 0 does not.
            report["rules_zero_price"] = _n(
                _on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)]
                + _dated + [('fixed_price', '<=', 0)])

        by_var, by_tmpl = {}, {}
        for r in rules:
            col = wanted[O.m2o_id(r.get('pricelist_id'))]
            if r.get('applied_on') == '0_product_variant' and r.get('product_id'):
                by_var.setdefault((col, O.m2o_id(r['product_id'])), []).append(r)
            elif r.get('product_tmpl_id'):
                by_tmpl.setdefault((col, O.m2o_id(r['product_tmpl_id'])), []).append(r)

        dom = [('active', '=', True)]
        prods = O.search_read('product.product', dom,
                              ['id', 'default_code', 'product_tmpl_id'], limit=50000)
        n = O.get_odoo().search_count('product.product', dom)
        if len(prods) != n:
            # Same guard, same reason as `catalogue()`: a short pull renders as a plausible
            # smaller set of priced SKUs with nothing reporting it.
            raise ValueError(
                f"products.pricelist_by_code: the product pull is TRUNCATED — read {len(prods)} "
                f"rows against a search_count of {n}.")
    except Exception as e:
        report["error"] = f"{type(e).__name__}: {str(e)[:200]}"
        return {}, report

    out = {}
    for p in prods:
        code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}")
        tmpl = O.m2o_id(p.get('product_tmpl_id'))
        cells = {}
        for col, _name in PRICELIST_COLUMNS:
            cands = by_var.get((col, p['id'])) or by_tmpl.get((col, tmpl))
            if not cands:
                continue
            base = min(cands, key=lambda r: r.get('min_quantity') or 0.0)
            if (base.get('min_quantity') or 0.0) > 1.0:
                report["qty_break_only"] += 1
            cells[col] = base.get('fixed_price')
        if cells:
            out.setdefault(code, {}).update(cells)
    return out, report


def catalogue_count():
    """The INDEPENDENT population oracle: Odoo's own count of active products.

    Deliberately a bare `search_count` and not a `len()` over anything this module built — the
    2,717 defect shipped silently for a wave because `product_data.validate()` derived BOTH sides
    of its reconciliation from `_sku_rev`, so the oracle could never see a missing row.
    """
    return O.get_odoo().search_count('product.product', [('active', '=', True)])


def categories(t=None, team_id=None):
    """Sorted distinct category names present in the SKU directory (for the drill-down filter)."""
    return sorted({r['category'] for r in directory(t, team_id)})


def _coverage(date_from, date_to, team_id=None):
    """{sku_code: distinct_customer_count} via 2-level read_group (slow-ish), merged by code.
    (A customer buying two records sharing a code can count twice; duplicates are rare, and
    code-level avoids the bigger error of a re-SKUed product reading as full coverage loss.)"""
    g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC),
                     ['__count'], ['product_id', 'order_partner_id'], lazy=False)
    codes = _code_map()
    cov = {}
    names = {}
    for r in g:
        pid = O.m2o_id(r.get('product_id'))
        if not pid:
            continue
        key = codes.get(pid, f"pid:{pid}")
        cov[key] = cov.get(key, 0) + 1
        names.setdefault(key, O.m2o_name(r.get('product_id')))
    return cov, names


def coverage_collapse(t=None, limit=25, min_prior_custs=8, team_id=None):
    """SKUs that lost the most customer breadth YoY — early warning a SKU is dying even if
    revenue hasn't fully cratered yet."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    cov_t, names_t = _coverage(yf, yt, team_id)
    cov_l, names_l = _coverage(lf, lt, team_id)
    rows = []
    for pid, lc in cov_l.items():
        if lc < min_prior_custs:
            continue
        tc = cov_t.get(pid, 0)
        drop = lc - tc
        if drop <= 0:
            continue
        rows.append({'code': pid, 'product': names_l.get(pid) or names_t.get(pid) or '',
                     'custs_ly': lc, 'custs_ytd': tc, 'lost_custs': drop,
                     'pct_drop': drop / lc * 100})
    rows.sort(key=lambda x: (-x['lost_custs'], -x['pct_drop']))
    return rows[:limit]


# ====================================================================== SKU DRAWER (mirror of customer drawer)
def _sku_product_ids(code):
    """All product.product ids sharing this SKU code (merged variants / archived records)."""
    return [pid for pid, c in _code_map().items() if c == code]


def _sku_dom(pids, date_from, date_to, team_id=None):
    return O.sale_line_domain(date_from, date_to, team_id, extra=[('product_id', 'in', pids)] + _NO_SVC)


def _sku_name_category(pids):
    rows = O.search_read('product.product', [('id', 'in', pids)], ['name', 'categ_id'])
    name = rows[0]['name'] if rows else '(unknown)'
    catmap = sales_mod._product_cat()
    main = next((catmap.get(p) for p in pids if catmap.get(p)), '(uncategorized)')
    return name, main


def _sku_buyers(pids, date_from, date_to, team_id=None):
    """{partner_id: {'name','rev','qty'}} for buyers of this SKU over a window."""
    g = O.read_group('sale.order.line', _sku_dom(pids, date_from, date_to, team_id),
                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False)
    out = {}
    for r in g:
        pid = O.m2o_id(r.get('order_partner_id'))
        if pid:
            out[pid] = {'name': O.m2o_name(r.get('order_partner_id')),
                        'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0}
    return out


def sku_detail(code, t=None, team_id=None, n_months=13, allsku=None):
    """KPIs (rev/qty/buyers YoY, GM%), monthly trend, rank & % of BU for one SKU code.
    Pass `allsku` (a cached _sku_rev YTD map) to skip the ~all-SKU rank read."""
    t = t or P.today()
    pids = _sku_product_ids(code)
    if not pids:
        return None
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    mf, mt = P.ltm(t)
    name, category = _sku_name_category(pids)

    def s(df, dtt, field='price_subtotal'):
        return O.sum_field('sale.order.line', _sku_dom(pids, df, dtt, team_id), field)
    rev_ytd, rev_ly = s(yf, yt), s(lf, lt)
    qty_ytd, qty_ly = s(yf, yt, 'product_uom_qty'), s(lf, lt, 'product_uom_qty')
    buyers_ytd = len(_sku_buyers(pids, yf, yt, team_id))
    buyers_ly = len(_sku_buyers(pids, lf, lt, team_id))
    g = O.read_group('sale.order.line', _sku_dom(pids, mf, mt, team_id),
                     ['price_subtotal:sum', 'margin:sum'], [], lazy=False)
    line_rev = (g[0].get('price_subtotal') if g else 0) or 0.0
    margin = (g[0].get('margin') if g else 0) or 0.0

    # Monthly trend: bin lines by their order's month in Python (dot-path month groupby is rejected
    # on sale.order.line), in 2 reads instead of 26 point queries.
    range_start = dt.date(t.year - 2, t.month, 1).isoformat()
    lines = O.search_read('sale.order.line', _sku_dom(pids, range_start, t.isoformat(), team_id),
                          ['price_subtotal', 'order_id'])
    oids = list({O.m2o_id(line['order_id']) for line in lines if line.get('order_id')})
    odate = {}
    for i in range(0, len(oids), 1000):
        for o in O.search_read('sale.order', [('id', 'in', oids[i:i + 1000])], ['date_order']):
            if o.get('date_order'):
                odate[o['id']] = str(o['date_order'])[:7]
    mrev = {}
    for line in lines:
        ym = odate.get(O.m2o_id(line.get('order_id')))
        if ym:
            mrev[ym] = mrev.get(ym, 0.0) + (line.get('price_subtotal') or 0.0)
    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)})

    allsku = allsku if allsku is not None else _sku_rev(yf, yt, team_id)
    total = sum(v['rev'] for v in allsku.values()) or 1.0
    rank = next((i + 1 for i, (c, _v) in enumerate(sorted(allsku.items(), key=lambda kv: -kv[1]['rev']))
                 if c == code), None)
    return {
        'code': code, 'name': name, 'category': category,
        'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'rev_yoy_pct': P.yoy_pct(rev_ytd, rev_ly),
        'qty_ytd': qty_ytd, 'qty_ly': qty_ly, 'qty_yoy_pct': P.yoy_pct(qty_ytd, qty_ly),
        'buyers_ytd': buyers_ytd, 'buyers_ly': buyers_ly, 'buyers_delta': buyers_ytd - buyers_ly,
        'gm_pct': (margin / line_rev * 100) if line_rev else 0.0, 'gm_dollars': margin,
        'rank': rank, 'n_skus': len(allsku), 'pct_of_bu': rev_ytd / total * 100,
        'monthly': monthly,
    }


def sku_buyer_bridge(code, t=None, team_id=None, top=12):
    """Who drives the SKU's YoY: retained / new / churned buyers + a $-ranked churned call list."""
    t = t or P.today()
    pids = _sku_product_ids(code)
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id)
    tset, lset = set(this), set(last)
    retained, new, churned = tset & lset, tset - lset, lset - tset
    churned_list = sorted([{'pid': p, 'customer': last[p]['name'], 'ly_rev': last[p]['rev']}
                           for p in churned], key=lambda x: -x['ly_rev'])[:top]
    return {'retained': {'n': len(retained), 'rev': sum(this[p]['rev'] for p in retained)},
            'new': {'n': len(new), 'rev': sum(this[p]['rev'] for p in new)},
            'churned': {'n': len(churned), 'rev': sum(last[p]['rev'] for p in churned)},
            'buyer_retention_pct': (len(retained) / len(lset) * 100) if lset else 0.0,
            'churned_buyers': churned_list, 'buyers_this': len(tset), 'buyers_last': len(lset),
            'this_total': sum(v['rev'] for v in this.values())}


def sku_concentration(code, t=None, team_id=None):
    """Buyer-concentration risk: top-1/top-3 share, Herfindahl index, effective buyer count (LTM)."""
    t = t or P.today()
    mf, mt = P.ltm(t)
    buyers = _sku_buyers(_sku_product_ids(code), mf, mt, team_id)
    revs = sorted([v['rev'] for v in buyers.values()], reverse=True)
    total = sum(revs) or 1.0
    hhi = sum((r / total) ** 2 for r in revs)
    return {'n_buyers': len(revs), 'top1_pct': (revs[0] / total * 100) if revs else 0.0,
            'top3_pct': (sum(revs[:3]) / total * 100) if revs else 0.0,
            'hhi': hhi, 'eff_buyers': (1 / hhi) if hhi else 0.0}


def sku_price_dispersion(code, t=None, team_id=None, cap=60):
    """Realized $/unit per buyer (LTM) vs the volume-weighted average; recoverable $ on below-VWAP
    accounts. list_price is unreliable, so the dispersion among actual buyers is the margin lever."""
    t = t or P.today()
    mf, mt = P.ltm(t)
    g = O.read_group('sale.order.line', _sku_dom(_sku_product_ids(code), mf, mt, team_id),
                     ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False)
    rows, tot_rev, tot_qty = [], 0.0, 0.0
    for r in g:
        pid = O.m2o_id(r.get('order_partner_id'))
        rev = r.get('price_subtotal') or 0.0
        qty = r.get('product_uom_qty') or 0.0
        if not pid or qty <= 0:
            continue
        rows.append({'pid': pid, 'customer': O.m2o_name(r.get('order_partner_id')),
                     'price': rev / qty, 'qty': qty, 'rev': rev})
        tot_rev += rev
        tot_qty += qty
    vwap = (tot_rev / tot_qty) if tot_qty else 0.0
    for r in rows:
        r['recoverable'] = max(0.0, vwap - r['price']) * r['qty']
    rows.sort(key=lambda x: -x['recoverable'])
    return {'vwap': vwap, 'n_buyers': len(rows),
            'recoverable_total': sum(r['recoverable'] for r in rows), 'rows': rows[:cap]}


def sku_top_buyers(code, t=None, team_id=None, top=15):
    """Ranked buyers of this SKU (YTD) with YoY — clickable to open the customer drawer."""
    t = t or P.today()
    pids = _sku_product_ids(code)
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id)
    rows = [{'pid': p, 'customer': v['name'], 'rev': v['rev'], 'qty': v['qty'],
             'yoy_pct': P.yoy_pct(v['rev'], last.get(p, {}).get('rev', 0.0))} for p, v in this.items()]
    rows.sort(key=lambda x: -x['rev'])
    return rows[:top]


_BUYER_TIERS = [('Whale (≥$25k)', 25000.0), ('Large ($10–25k)', 10000.0),
                ('Mid ($2–10k)', 2000.0), ('Small (<$2k)', 0.0)]


def sku_customer_analysis(code, t=None, team_id=None, top=15):
    """WHO buys this SKU (LTM), as customers: the value-tier mix of its buyers (by each buyer's TOTAL
    spend), and the accounts most DEPENDENT on it (this SKU as a share of their spend — who gets hurt
    most if it stocks out / who to protect)."""
    t = t or P.today()
    lf, lt = P.ltm(t)
    pids = _sku_product_ids(code)
    empty = {'segments': [], 'dependency': [], 'n_buyers': 0, 'avg_dependency': 0.0}
    if not pids:
        return empty
    buyers = _sku_buyers(pids, lf, lt, team_id)        # {pid: {name, rev(this SKU), qty}}
    if not buyers:
        return empty
    bpids = list(buyers)
    g = O.read_group('sale.order', sales_mod.order_domain(lf, lt, team_id) + [('partner_id', 'in', bpids)],
                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
    total = {O.m2o_id(r['partner_id']): (r.get('amount_untaxed') or 0.0) for r in g if r.get('partner_id')}

    def tier(rev):
        for nm, lo in _BUYER_TIERS:
            if rev >= lo:
                return nm
        return _BUYER_TIERS[-1][0]
    seg = {nm: {'tier': nm, 'buyers': 0, 'sku_rev': 0.0} for nm, _ in _BUYER_TIERS}
    dep = []
    for p, v in buyers.items():
        ct = total.get(p, v['rev']) or v['rev']
        s = seg[tier(ct)]
        s['buyers'] += 1
        s['sku_rev'] += v['rev']
        dep.append({'customer': v['name'], 'pid': p, 'sku_rev': v['rev'], 'cust_total': ct,
                    'dependency_pct': (v['rev'] / ct * 100) if ct else None})
    tot = sum(s['sku_rev'] for s in seg.values()) or 1.0
    segments = []
    for nm, _ in _BUYER_TIERS:
        s = seg[nm]
        s['rev_share'] = s['sku_rev'] / tot * 100
        s['avg_per_buyer'] = (s['sku_rev'] / s['buyers']) if s['buyers'] else 0.0
        segments.append(s)
    dep.sort(key=lambda x: -x['sku_rev'])
    deps_known = [d['dependency_pct'] for d in dep if d['dependency_pct'] is not None]
    return {'segments': segments, 'dependency': dep[:top], 'n_buyers': len(buyers),
            'avg_dependency': (sum(deps_known) / len(deps_known)) if deps_known else 0.0}


def sku_whitespace(code, t=None, team_id=None, top=15):
    """Customers who buy this SKU's category but NOT this SKU — ranked prospect list."""
    import modules.customers as cust_mod
    t = t or P.today()
    pids = _sku_product_ids(code)
    _name, category = _sku_name_category(pids)
    cat_buyers = cust_mod.category_buyers(category, team_id=team_id) or set()
    mf, mt = P.ltm(t)
    prospects = list(cat_buyers - set(_sku_buyers(pids, mf, mt, team_id)))
    if not prospects:
        return {'category': category, 'rows': []}
    g = O.read_group('sale.order', sales_mod.order_domain(mf, mt, team_id) + [('partner_id', 'in', prospects)],
                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
    rows = [{'pid': O.m2o_id(r['partner_id']), 'customer': O.m2o_name(r['partner_id']),
             'total_spend': r.get('amount_untaxed') or 0.0} for r in g if r.get('partner_id')]
    rows.sort(key=lambda x: -x['total_spend'])
    return {'category': category, 'rows': rows[:top]}


def sku_drawer_bundle(code, t=None, team_id=None, allsku=None):
    """The whole SKU drawer's first paint in one cached unit: sku_detail first (the not-found gate),
    then the other six pulls CONCURRENTLY (O.parallel). Cuts a ~6-call cold open to ~max(call)."""
    detail = sku_detail(code, t=t, team_id=team_id, allsku=allsku)
    if detail is None:
        return {'detail': None}
    bridge, conc, price, buyers, white, ca = O.parallel([
        lambda: sku_buyer_bridge(code, t, team_id),
        lambda: sku_concentration(code, t, team_id),
        lambda: sku_price_dispersion(code, t, team_id),
        lambda: sku_top_buyers(code, t, team_id),
        lambda: sku_whitespace(code, t, team_id),
        lambda: sku_customer_analysis(code, t, team_id),
    ])
    return {'detail': detail, 'bridge': bridge, 'conc': conc, 'price': price,
            'buyers': buyers, 'white': white, 'ca': ca}


def validate(t=None, team_id=None):
    """Reconcile SKU metrics to independent Odoo aggregates. When team_id is set (a single BU
    selected) every check runs SCOPED to that BU, so the validation panel never reconciles
    against — or exposes — the other BU's numbers. All checks here are BU-scopeable (no cross-BU
    mirror), so team_id threads straight through."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    checks = []
    sku = _sku_rev(yf, yt, team_id)
    sku_sum = sum(v['rev'] for v in sku.values())
    line_total = O.sum_field('sale.order.line',
                             O.sale_line_domain(yf, yt, team_id, extra=_NO_SVC), 'price_subtotal')
    checks.append({'check': 'SKU rev: Σ(per-SKU) == total line revenue, ex-services (YTD)',
                   'a': round(sku_sum, 2), 'b': round(line_total, 2),
                   'gap': round(sku_sum - line_total, 2),
                   'ok': abs(sku_sum - line_total) <= 1.0})

    m = yoy_movers(t, limit=10**9, team_id=team_id)
    movers_sum = sum(r['change'] for r in m['risers']) + sum(r['change'] for r in m['decliners'])
    lf, lt = P.ytd_last_year(t)
    last_total = O.sum_field('sale.order.line',
                             O.sale_line_domain(lf, lt, team_id, extra=_NO_SVC), 'price_subtotal')
    checks.append({'check': 'SKU movers: Σ(Δ) == (YTD − LY) total',
                   'a': round(movers_sum, 2), 'b': round(line_total - last_total, 2),
                   'gap': round(movers_sum - (line_total - last_total), 2),
                   'ok': abs(movers_sum - (line_total - last_total)) <= 1.0})

    # SKU drawer: buyer-bridge reconciles — retained + new buyer revenue == this-period SKU revenue
    top_code = max(sku.items(), key=lambda kv: kv[1]['rev'])[0] if sku else None
    if top_code:
        bb = sku_buyer_bridge(top_code, t, team_id=team_id)
        recon = bb['retained']['rev'] + bb['new']['rev']
        checks.append({'check': 'SKU drawer: retained+new buyer rev == SKU YTD revenue',
                       'a': round(recon, 2), 'b': round(bb['this_total'], 2),
                       'gap': round(recon - bb['this_total'], 2),
                       'ok': abs(recon - bb['this_total']) <= 1.0})
    return checks