platform/modules/sales.py

"""Sales module — revenue, YoY (same-period), seasonality, rep/customer/SKU breakdowns.

Scope: confirmed orders (state sale/done) on Fisch+Royal teams, excluded accounts removed.
Order-level metrics come from sale.order (amount_untaxed); SKU-level from
sale.order.line (price_subtotal). Each public metric has a paired validate_* that
reconciles against an independent Odoo aggregate.
"""
import sys
from functools import lru_cache
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))  # platform/
import core.odoo as O
import core.periods as P
import modules.inventory as inv_mod   # reuse the main-category resolver


def order_domain(date_from=None, date_to=None, team_id=None, partner_ids=None):
    """Confirmed sale.order domain (RI+FFS scope, house accounts excluded). partner_ids (a
    collection, possibly empty) restricts to those customers — the Customer-module Agent filter
    rides on this; None = no partner restriction (an empty collection matches no orders)."""
    dom = [('state', 'in', ['sale', 'done']), ('team_id', 'in', TEAMS(team_id))]
    if date_from:
        dom.append(('date_order', '>=', f'{date_from} 00:00:00'))
    if date_to:
        dom.append(('date_order', '<=', f'{date_to} 23:59:59'))
    ex = O.excluded_partner_ids()
    if ex:
        dom.append(('partner_id', 'not in', list(ex)))
    if partner_ids is not None:
        dom.append(('partner_id', 'in', list(partner_ids)))
    if O.doc_mode() == 'invoice':
        dom.append(('invoice_status', '=', 'invoiced'))
    return dom


def TEAMS(team_id):
    return [team_id] if team_id is not None else O.TEAM_IDS


# ---- order-level primitives: STORE-backed (OM-2 retrofit 2026-07-12) with LIVE fallback ------
# headline/scorecard/trends make ~50 of these per build; the store serves them in milliseconds
# (kept minutes-fresh by the app's auto-sync). The SQL mirrors order_domain() EXACTLY — same
# states, teams, exclusions, date bounds and the Orders/Invoiced basis (invoice_status is
# synced). validate() stays on live Odoo = the standing parity proof. Any store problem falls
# back to the live reads.
USE_STORE = True


def _order_where_store(date_from, date_to, team_id):
    """(where_sql, params) mirroring order_domain() for the store's sale_order table."""
    params, w = [], ["state IN ('sale','done')"]
    teams = 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 O.doc_mode() == 'invoice':
        w.append("invoice_status = 'invoiced'")
    return " AND ".join(w), params


def _order_agg_store(expr, date_from, date_to, team_id):
    import harness.datastore as DS
    where, params = _order_where_store(date_from, date_to, team_id)
    r = DS.ro_con().execute(
        f"SELECT {expr} FROM sale_order WHERE {where}", params).fetchone()
    return r[0] or 0


def _order_partner_groups_store(date_from, date_to, team_id):
    """[(partner_id, n_orders, revenue)] — feeds cadence + concentration."""
    import harness.datastore as DS
    where, params = _order_where_store(date_from, date_to, team_id)
    return DS.ro_con().execute(
        f"SELECT partner_id, count(*), sum(amount_untaxed) FROM sale_order WHERE {where} "
        "GROUP BY 1", params).fetchall()


def _orev(date_from, date_to, team_id=None):
    if USE_STORE:
        try:
            return float(_order_agg_store('sum(amount_untaxed)', date_from, date_to, team_id))
        except Exception:
            pass
    return O.sum_field('sale.order', order_domain(date_from, date_to, team_id), 'amount_untaxed')


def _orders(date_from, date_to, team_id=None):
    if USE_STORE:
        try:
            return int(_order_agg_store('count(*)', date_from, date_to, team_id))
        except Exception:
            pass
    return O.get_odoo().search_count('sale.order', order_domain(date_from, date_to, team_id))


def _custs(date_from, date_to, team_id=None):
    if USE_STORE:
        try:
            return int(_order_agg_store('count(DISTINCT partner_id)',
                                        date_from, date_to, team_id))
        except Exception:
            pass
    return O.distinct_count('sale.order', order_domain(date_from, date_to, team_id), 'partner_id')


# ---------------------------------------------------------------- headline
def headline(t=None, team_id=None):
    """Headline scoped to a DBA (team_id) or consolidated (None). by_team always shows both."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    rev = _orev(yf, yt, team_id)
    rev_ly = _orev(lf, lt, team_id)
    orders = _orders(yf, yt, team_id)
    custs = _custs(yf, yt, team_id)
    out = {
        'as_of': yt,
        'ytd_revenue': rev,
        'ytd_revenue_ly': rev_ly,
        'yoy_pct': P.yoy_pct(rev, rev_ly),
        'ytd_orders': orders,
        'ytd_customers': custs,
        'aov': (rev / orders) if orders else 0.0,
        'by_team': {},
    }
    for tid in O.TEAM_IDS:
        out['by_team'][O.TEAM_NAMES[tid]] = {
            'ytd': _orev(yf, yt, tid),
            'ytd_ly': _orev(lf, lt, tid),
        }
    return out


# ---------------------------------------------------------------- period scorecard
# (label, key, weekday-aligned-LY?) — Today/WTD compare to 52 weeks ago (same weekday); the
# month/quarter/year periods compare to the same calendar window last year.
_PERIODS = [('Today', 'today', True), ('Week to date', 'wtd', True), ('Month to date', 'mtd', False),
            ('Quarter to date', 'qtd', False), ('Year to date', 'ytd', False)]


def _period_window(key, t):
    import datetime as dt
    if key == 'today':
        return P._d(t), P._d(t)
    return {'wtd': P.wtd, 'mtd': P.mtd, 'qtd': P.qtd, 'ytd': P.ytd}[key](t)


def period_scorecard(t=None, team_id=None):
    """The headline period scorecard: revenue (+ YoY same-period), orders, AOV and active customers
    for Today / WTD / MTD / QTD / YTD. Each entry carries its date window so the UI can make every
    number click through to the decomposition drawer."""
    t = t or P.today()
    out = []
    for label, key, wk in _PERIODS:
        f, tt = _period_window(key, t)
        cf, ct = P.shift_year(f, tt, weeks=wk)
        rev = _orev(f, tt, team_id)
        rev_ly = _orev(cf, ct, team_id)
        orders = _orders(f, tt, team_id)
        out.append({
            'key': key, 'label': label, 'date_from': f, 'date_to': tt,
            'cmp_from': cf, 'cmp_to': ct,
            'revenue': rev, 'revenue_ly': rev_ly, 'yoy_pct': P.yoy_pct(rev, rev_ly),
            'orders': orders, 'customers': _custs(f, tt, team_id),
            'aov': (rev / orders) if orders else 0.0,
        })
    return out


# ---------------------------------------------------------------- seasonality / trend
def weekly_trend(n_weeks=13, t=None, team_id=None):
    """Per-week revenue for the last n_weeks (Mon–Sun) vs the same week 52 weeks earlier (weekday-
    aligned YoY). Each row carries start/end so a clicked week decomposes to its exact window."""
    t = t or P.today()
    rows = []
    for label, start, end in P.week_starts(n_weeks, t):
        this = _orev(start, end, team_id)
        cf, ct = P.shift_year(start, end, weeks=True)
        last = _orev(cf, ct, team_id)
        rows.append({'week': label, 'start': start, 'end': end, 'cmp_from': cf, 'cmp_to': ct,
                     'revenue': this, 'revenue_ly': last, 'yoy_pct': P.yoy_pct(this, last)})
    return rows


def monthly_trend(n=13, t=None, team_id=None):
    """Per-month revenue for the last n months + same month one year earlier (YoY)."""
    t = t or P.today()
    rows = []
    for ym, start, end in P.month_starts(n, t):
        this = _orev(start, end, team_id)
        # same month last year
        y, m = int(ym[:4]) - 1, int(ym[5:7])
        import datetime as dt
        ly_start = dt.date(y, m, 1).isoformat()
        ly_end = (dt.date(y + (m // 12), (m % 12) + 1, 1) - dt.timedelta(days=1)).isoformat()
        last = _orev(ly_start, ly_end, team_id)
        rows.append({'month': ym, 'start': start, 'end': end, 'cmp_from': ly_start, 'cmp_to': ly_end,
                     'revenue': this, 'revenue_ly': last, 'yoy_pct': P.yoy_pct(this, last)})
    return rows


# ---------------------------------------------------------------- breakdowns
def by_team(t=None):
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    return [{'team': O.TEAM_NAMES[tid], 'ytd': _orev(yf, yt, tid),
             'ytd_ly': _orev(lf, lt, tid),
             'yoy_pct': P.yoy_pct(_orev(yf, yt, tid), _orev(lf, lt, tid))}
            for tid in O.TEAM_IDS]


def by_rep(t=None, limit=25, team_id=None):
    t = t or P.today()
    yf, yt = P.ytd(t)
    g = O.read_group('sale.order', order_domain(yf, yt, team_id),
                     ['amount_untaxed:sum'], ['user_id'], lazy=False)
    rows = [{'rep': O.m2o_name(r.get('user_id')) or '(none)', 'uid': O.m2o_id(r.get('user_id')),
             'revenue': r.get('amount_untaxed') or 0.0,
             'orders': r.get('__count') or r.get('user_id_count') or 0}
            for r in g]
    rows.sort(key=lambda x: -x['revenue'])
    return rows[:limit]


def category_product_ids(category):
    """The product ids belonging to a main category — for decomposing a category number."""
    return [pid for pid, c in _product_cat().items() if c == category]


def _top_customers_store(yf, yt, team_id, limit):
    import harness.datastore as DS
    params, w = [], ["state IN ('sale','done')"]
    teams = TEAMS(team_id)
    w.append("team_id IN (" + ",".join("?" * len(teams)) + ")")
    params += list(teams)
    w.append("CAST(date_order AS TIMESTAMP) >= CAST(? AS TIMESTAMP)")
    params.append(f"{yf} 00:00:00")
    w.append("CAST(date_order AS TIMESTAMP) <= CAST(? AS TIMESTAMP)")
    params.append(f"{yt} 23:59:59")
    ex = O.excluded_partner_ids()
    if ex:
        w.append("partner_id NOT IN (" + ",".join("?" * len(ex)) + ")")
        params += list(ex)
    if O.doc_mode() == 'invoice':
        w.append("invoice_status = 'invoiced'")
    rows = DS.ro_con().execute(
        "SELECT o.partner_id, coalesce(p.name, '#' || o.partner_id), "
        "sum(o.amount_untaxed), count(*) "
        "FROM sale_order o LEFT JOIN res_partner p ON p.id = o.partner_id "
        "WHERE " + " AND ".join(w) + " GROUP BY 1, 2 ORDER BY 3 DESC LIMIT ?",
        params + [int(limit)]).fetchall()
    return [{'pid': r[0], 'customer': r[1], 'revenue': r[2] or 0.0, 'orders': r[3]}
            for r in rows if r[0]]


def top_customers(t=None, limit=25, team_id=None):
    t = t or P.today()
    yf, yt = P.ytd(t)
    if USE_STORE:
        try:
            return _top_customers_store(yf, yt, team_id, limit)
        except Exception:
            pass
    g = O.read_group('sale.order', order_domain(yf, yt, team_id),
                     ['amount_untaxed:sum'], ['partner_id'], lazy=False)
    rows = [{'pid': O.m2o_id(r.get('partner_id')),
             'customer': O.m2o_name(r.get('partner_id')),
             'revenue': r.get('amount_untaxed') or 0.0,
             'orders': r.get('__count') or 0}
            for r in g if r.get('partner_id')]
    rows.sort(key=lambda x: -x['revenue'])
    return rows[:limit]


def _top_skus_store(yf, yt, team_id, limit):
    """Mirrors sale_line_domain: confirmed states, order-side teams/dates/invoice-mode,
    line-side partner exclusion, product_id set. Names + SKU codes join in-store (the live
    path needs an extra search_read for codes)."""
    import harness.datastore as DS
    params, w = [], ["o.state IN ('sale','done')", "l.product_id IS NOT NULL"]
    teams = TEAMS(team_id)
    w.append("o.team_id IN (" + ",".join("?" * len(teams)) + ")")
    params += list(teams)
    w.append("CAST(o.date_order AS TIMESTAMP) >= CAST(? AS TIMESTAMP)")
    params.append(f"{yf} 00:00:00")
    w.append("CAST(o.date_order AS TIMESTAMP) <= CAST(? AS TIMESTAMP)")
    params.append(f"{yt} 23:59:59")
    ex = O.excluded_partner_ids()
    if ex:
        w.append("l.order_partner_id NOT IN (" + ",".join("?" * len(ex)) + ")")
        params += list(ex)
    if O.doc_mode() == 'invoice':
        w.append("o.invoice_status = 'invoiced'")
    rows = DS.ro_con().execute(
        "SELECT l.product_id, coalesce(p.name, '#' || l.product_id), p.default_code, "
        "sum(l.price_subtotal), sum(l.product_uom_qty), sum(l.margin), count(*) "
        "FROM sale_order_line l JOIN sale_order o ON o.id = l.order_id "
        "LEFT JOIN product_product p ON p.id = l.product_id "
        "WHERE " + " AND ".join(w) + " GROUP BY 1, 2, 3 ORDER BY 4 DESC LIMIT ?",
        params + [int(limit)]).fetchall()
    return [_sku_profit({'product': r[1], 'pid': r[0],
                         'code': (str(r[2]).strip() if r[2] else None),
                         'revenue': r[3] or 0.0, 'qty': r[4] or 0.0,
                         'margin': r[5] or 0.0, 'lines': r[6]})
            for r in rows if r[0]]


def top_skus(t=None, limit=25, team_id=None):
    t = t or P.today()
    yf, yt = P.ytd(t)
    if USE_STORE:
        try:
            return _top_skus_store(yf, yt, team_id, limit)
        except Exception:
            pass
    g = O.read_group('sale.order.line', O.sale_line_domain(yf, yt, team_id),
                     ['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'], ['product_id'], lazy=False)
    rows = [_sku_profit({'product': O.m2o_name(r.get('product_id')), 'pid': O.m2o_id(r.get('product_id')),
             '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 g if r.get('product_id')]
    rows.sort(key=lambda x: -x['revenue'])
    rows = rows[:limit]
    if rows:   # attach SKU code so the UI can deep-link each to its SKU drawer
        codemap = {}
        for pr in O.search_read('product.product', [('id', 'in', [r['pid'] for r in rows])], ['default_code']):
            codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None
        for r in rows:
            r['code'] = codemap.get(r['pid'])
    return rows


def _sku_profit(s):
    """Attach profit/order + margin% to a SKU row (uses the Margin module's `margin` sum). `lines`
    (order lines) ~= the number of orders containing the SKU (one line per SKU per order in
    practice), so profit/order = total margin ÷ lines."""
    lines = s.get('lines') or 0
    s['profit_per_order'] = (s['margin'] / lines) if lines else 0.0
    s['margin_pct'] = (s['margin'] / s['revenue'] * 100.0) if s.get('revenue') else 0.0
    return s


def _attach_sku_codes(rows, pid_key='pid'):
    """Attach SKU `code` (default_code) to product rows so each can deep-link to its SKU drawer."""
    ids = [r[pid_key] for r in rows if r.get(pid_key)]
    if not ids:
        return rows
    codemap = {}
    for pr in O.search_read('product.product', [('id', 'in', ids)], ['default_code']):
        codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None
    for r in rows:
        r['code'] = codemap.get(r.get(pid_key))
    return rows


def decompose(date_from, date_to, team_id=None, line_extra=None, order_extra=None,
              product_ids=None, compare=None, top=30):
    """Universal decomposition of ANY sales number into its contributors over a window. Line-level
    (sale.order.line) so it breaks down by customer, SKU and category consistently and ties to the
    headline (line == order revenue, proven in validate()).

      line_extra   extra sale.order.line domain clauses (e.g. a rep via order_id.user_id)
      order_extra  the order-level translation of the same scope (for the orders count)
      product_ids  restrict to a category's products
      compare      (cmp_from, cmp_to) for the same-period-last-year total (YoY headline)

    Returns totals + ranked `customers`, `skus`, `categories` (each with revenue + % share)."""
    ex = list(line_extra or [])
    if product_ids is not None:
        ex.append(('product_id', 'in', list(product_ids)))
    dom = O.sale_line_domain(date_from, date_to, team_id, extra=ex)

    gc = O.read_group('sale.order.line', dom, ['price_subtotal:sum', 'product_uom_qty:sum'],
                      ['order_partner_id'], lazy=False)
    customers = [{'pid': O.m2o_id(r.get('order_partner_id')), 'customer': O.m2o_name(r.get('order_partner_id')),
                  'revenue': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0,
                  'lines': r.get('__count') or 0} for r in gc if r.get('order_partner_id')]

    gs = O.read_group('sale.order.line', dom, ['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'],
                      ['product_id'], lazy=False)
    skus = [_sku_profit({'pid': O.m2o_id(r.get('product_id')), 'product': O.m2o_name(r.get('product_id')),
             '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 gs if r.get('product_id')]

    cat = _product_cat()
    cagg = {}
    for s in skus:
        c = cat.get(s['pid'], '(uncategorized)')
        e = cagg.setdefault(c, {'category': c, 'revenue': 0.0, 'qty': 0.0, 'skus': 0})
        e['revenue'] += s['revenue']; e['qty'] += s['qty']; e['skus'] += 1
    categories = sorted(cagg.values(), key=lambda x: -x['revenue'])

    total = sum(c['revenue'] for c in customers)
    units = sum(s['qty'] for s in skus)
    for lst in (customers, skus, categories):
        for r in lst:
            r['pct'] = (r['revenue'] / total * 100.0) if total else 0.0
    customers.sort(key=lambda x: -x['revenue'])
    skus.sort(key=lambda x: -x['revenue'])
    _attach_sku_codes(skus)

    # orders count (order-level, single cheap aggregate) + AOV
    odom = order_domain(date_from, date_to, team_id) + list(order_extra or [])
    if product_ids is not None and not order_extra:
        odom = odom + [('order_line.product_id', 'in', list(product_ids))]
    orders = O.get_odoo().search_count('sale.order', odom)

    total_ly = None
    if compare:
        lex = list(line_extra or [])
        if product_ids is not None:
            lex.append(('product_id', 'in', list(product_ids)))
        total_ly = O.sum_field('sale.order.line',
                               O.sale_line_domain(compare[0], compare[1], team_id, extra=lex), 'price_subtotal')

    return {
        'window': f'{date_from} → {date_to}', 'date_from': date_from, 'date_to': date_to,
        'total': total, 'total_ly': total_ly, 'yoy_pct': (P.yoy_pct(total, total_ly) if total_ly is not None else None),
        'orders': orders, 'units': units, 'n_customers': len(customers), 'n_skus': len(skus),
        'aov': (total / orders) if orders else 0.0,
        'customers': customers[:top], 'skus': skus[:top], 'categories': categories,
        'all_customers': customers, 'all_skus': skus,
    }


_ORDER_STATE = {'draft': 'Quote', 'sent': 'Quote sent', 'sale': 'Confirmed', 'done': 'Locked',
                'cancel': 'Cancelled'}
_ORDER_INV = {'upselling': 'Upselling', 'invoiced': 'Invoiced', 'to invoice': 'To invoice',
              'no': 'Nothing to invoice'}


def orders_in_scope(date_from, date_to, team_id=None, line_extra=None, order_extra=None,
                    product_ids=None):
    """The per-ORDER list behind any sales number over a window — the raw sale.order rows that make
    up a decomposition, so a chart click drills all the way down to the individual orders (each
    exportable to Excel). Revenue / units / margin are the IN-SCOPE line contribution (e.g. for a
    clicked SKU, only that SKU's lines), so Σ(order revenue) ties to the decomposition headline.
    Returns ALL orders (no silent cap — the export must be complete), newest-revenue first."""
    ex = list(line_extra or [])
    if product_ids is not None:
        ex.append(('product_id', 'in', list(product_ids)))
    dom = O.sale_line_domain(date_from, date_to, team_id, extra=ex)
    g = O.read_group('sale.order.line', dom,
                     ['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'],
                     ['order_id'], lazy=False)
    by_id = {}
    for r in g:
        oid = O.m2o_id(r.get('order_id'))
        if not oid:
            continue
        by_id[oid] = {'oid': oid, 'order': O.m2o_name(r.get('order_id')),
                      '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}
    if not by_id:
        return []
    # one order-level pull for the human columns (date / customer / status)
    for m in O.search_read('sale.order', [('id', 'in', list(by_id))],
                           ['name', 'date_order', 'partner_id', 'state', 'invoice_status']):
        e = by_id.get(m['id'])
        if not e:
            continue
        e['order'] = m.get('name') or e['order']
        e['date'] = (m.get('date_order') or '')[:10]
        e['customer'] = O.m2o_name(m.get('partner_id'))
        e['status'] = _ORDER_STATE.get(m.get('state'), m.get('state') or '')
        e['invoiced'] = _ORDER_INV.get(m.get('invoice_status'), m.get('invoice_status') or '')
    return sorted(by_id.values(), key=lambda x: -x['revenue'])


@lru_cache(maxsize=1)
def _product_cat():
    """product_id → main category name (reuses inventory's category resolver).
    Cached for the process — the category taxonomy is structural and rarely changes; this avoids
    re-reading the whole product master on every customer drill-down / win-back."""
    catmap = inv_mod._cat_main_map()
    prods = O.search_read('product.product', [('default_code', '!=', False)], ['id', 'categ_id'])
    return {p['id']: (catmap.get(O.m2o_id(p.get('categ_id'))) or '(uncategorized)') for p in prods}


def _line_rev_by_product_store(date_from, date_to, team_id):
    """[(product_id, Σ price_subtotal)] mirroring sale_line_domain (order-side scope join,
    line-side exclusions, product set)."""
    import harness.datastore as DS
    params, w = [], ["o.state IN ('sale','done')", "l.product_id IS NOT NULL"]
    teams = TEAMS(team_id)
    w.append("o.team_id IN (" + ",".join("?" * len(teams)) + ")")
    params += list(teams)
    w.append("CAST(o.date_order AS TIMESTAMP) >= CAST(? AS TIMESTAMP)")
    params.append(f"{date_from} 00:00:00")
    w.append("CAST(o.date_order AS TIMESTAMP) <= CAST(? AS TIMESTAMP)")
    params.append(f"{date_to} 23:59:59")
    ex = O.excluded_partner_ids()
    if ex:
        w.append("l.order_partner_id NOT IN (" + ",".join("?" * len(ex)) + ")")
        params += list(ex)
    if O.doc_mode() == 'invoice':
        w.append("o.invoice_status = 'invoiced'")
    return DS.ro_con().execute(
        "SELECT l.product_id, sum(l.price_subtotal) FROM sale_order_line l "
        "JOIN sale_order o ON o.id = l.order_id WHERE " + " AND ".join(w) + " GROUP BY 1",
        params).fetchall()


def _cat_rev(date_from, date_to, cat, team_id=None):
    """Revenue per main category over a window (line-level, brand-aware)."""
    if USE_STORE:
        try:
            out = {}
            for pid, amt in _line_rev_by_product_store(date_from, date_to, team_id):
                c = cat.get(pid, '(uncategorized)')
                out[c] = out.get(c, 0.0) + (amt or 0.0)
            return out
        except Exception:
            pass
    g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id),
                     ['price_subtotal:sum'], ['product_id'], lazy=False)
    out = {}
    for r in g:
        pid = O.m2o_id(r.get('product_id'))
        if not pid:
            continue
        c = cat.get(pid, '(uncategorized)')
        out[c] = out.get(c, 0.0) + (r.get('price_subtotal') or 0.0)
    return out


def by_category(t=None, limit=20, team_id=None):
    """Revenue by main category, YTD vs same-period last year — which categories drive (or
    drag) the number. Brand-filterable, so you can see e.g. which categories Fisch is losing."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    cat = _product_cat()
    this = _cat_rev(yf, yt, cat, team_id)
    last = _cat_rev(lf, lt, cat, team_id)
    rows = [{'category': c, 'revenue': this.get(c, 0.0), 'revenue_ly': last.get(c, 0.0),
             'change': this.get(c, 0.0) - last.get(c, 0.0),
             'yoy_pct': P.yoy_pct(this.get(c, 0.0), last.get(c, 0.0))}
            for c in (set(this) | set(last))]
    rows.sort(key=lambda x: -x['revenue'])
    return rows[:limit]


def cadence(t=None, team_id=None):
    """Reorder behaviour over LTM: repeat-purchase rate, avg orders/customer, and the
    order-frequency distribution. A wholesale-health signal (are customers coming back?)."""
    t = t or P.today()
    lf, lt = P.ltm(t)
    counts = None
    if USE_STORE:
        try:
            counts = [r[1] for r in _order_partner_groups_store(lf, lt, team_id) if r[0]]
        except Exception:
            counts = None
    if counts is None:
        g = O.read_group('sale.order', order_domain(lf, lt, team_id),
                         ['partner_id'], ['partner_id'], lazy=False)
        counts = [r.get('__count') or 0 for r in g if r.get('partner_id')]
    n = len(counts)
    total_orders = sum(counts)
    repeat = sum(1 for c in counts if c >= 2)
    buckets = [('1 order', lambda c: c == 1), ('2-3 orders', lambda c: 2 <= c <= 3),
               ('4-9 orders', lambda c: 4 <= c <= 9), ('10+ orders', lambda c: c >= 10)]
    dist = [{'frequency': label, 'customers': sum(1 for c in counts if fn(c)),
             'pct': (sum(1 for c in counts if fn(c)) / n * 100) if n else 0.0}
            for label, fn in buckets]
    return {
        'customers': n,
        'total_orders': total_orders,
        'avg_orders': (total_orders / n) if n else 0.0,
        'repeat_customers': repeat,
        'repeat_rate': (repeat / n * 100) if n else 0.0,
        'distribution': dist,
        'ltm_window': f'{lf} → {lt}',
    }


def concentration(t=None, team_id=None):
    """Top-N customer share of YTD revenue."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    revs = None
    if USE_STORE:
        try:
            revs = sorted([r[2] or 0.0 for r in _order_partner_groups_store(yf, yt, team_id)
                           if r[0]], reverse=True)
        except Exception:
            revs = None
    if revs is None:
        g = O.read_group('sale.order', order_domain(yf, yt, team_id),
                         ['amount_untaxed:sum'], ['partner_id'], lazy=False)
        revs = sorted([r.get('amount_untaxed') or 0.0 for r in g if r.get('partner_id')],
                      reverse=True)
    total = sum(revs) or 1.0
    def share(n):
        return sum(revs[:n]) / total * 100.0
    return {'n_customers': len(revs), 'total': total,
            'top10_pct': share(10), 'top25_pct': share(25),
            'top50_pct': share(50), 'top100_pct': share(100)}


# ---------------------------------------------------------------- returns (credit notes)
# Returns = posted customer credit notes (account.move, move_type='out_refund'). They carry a
# partner and salesperson but sit on the generic 'Sales' team (not Fisch/Royal team 5/6), so
# returns are reported CONSOLIDATED and sliced by agent (via the customer's res.partner.agent_ids)
# and by period — never BU-split. amount_untaxed_signed is negative for out_refund; we flip it so
# a "return" reads as a positive dollar figure everywhere.
def returns_domain(date_from=None, date_to=None, partner_ids=None):
    dom = [('move_type', '=', 'out_refund'), ('state', '=', 'posted')]
    if date_from:
        dom.append(('invoice_date', '>=', date_from))
    if date_to:
        dom.append(('invoice_date', '<=', date_to))
    ex = O.excluded_partner_ids()
    if ex:
        dom.append(('partner_id', 'not in', list(ex)))
    if partner_ids is not None:
        dom.append(('partner_id', 'in', list(partner_ids)))
    return dom


def _returns_amt(date_from, date_to, partner_ids=None):
    """Total returns $ (positive) over a window."""
    v = O.sum_field('account.move', returns_domain(date_from, date_to, partner_ids), 'amount_untaxed_signed')
    return -(v or 0.0)


def returns_monthly(n=13, t=None, partner_ids=None):
    """Returns $ per month for the last n months + same month one year earlier (YoY). Shaped like
    the revenue trend (`revenue`/`revenue_ly` keys) so it renders through the same chart_yoy_bars.
    ONE month-grouped read over a ~2-year span (was 2n sequential sum queries)."""
    t = t or P.today()
    months = P.month_starts(n, t)
    span_from = f"{int(months[0][0][:4]) - 1:04d}-{months[0][0][5:7]}-01"   # 1 year before the first month
    g = O.read_group('account.move', returns_domain(span_from, t.isoformat(), partner_ids),
                     ['amount_untaxed_signed:sum'], ['invoice_date:month'], lazy=False)
    mret = {}
    for r in g:
        ym = ((r.get('__range') or {}).get('invoice_date:month') or {}).get('from', '')[:7]
        if ym:
            mret[ym] = -(r.get('amount_untaxed_signed') or 0.0)
    rows = []
    for ym, start, end in months:
        y, m = int(ym[:4]) - 1, int(ym[5:7])
        this, last = mret.get(ym, 0.0), mret.get(f'{y:04d}-{m:02d}', 0.0)
        rows.append({'month': ym, 'start': start, 'end': end,
                     'revenue': this, 'revenue_ly': last, 'yoy_pct': P.yoy_pct(this, last)})
    return rows


def returns_by_partner(t=None, partner_ids=None):
    """{partner_id: returns$ YTD} (positive) — the raw per-customer credit-note total. The agent
    rollup maps these to res.partner.agent_ids in the Customers module."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    g = O.read_group('account.move', returns_domain(yf, yt, partner_ids),
                     ['amount_untaxed_signed:sum'], ['partner_id'], lazy=False)
    return {O.m2o_id(r.get('partner_id')): -(r.get('amount_untaxed_signed') or 0.0)
            for r in g if r.get('partner_id')}


def returns_headline(t=None, partner_ids=None):
    """YTD returns $ + YoY + return rate (returns / gross revenue). Consolidated."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    ret = _returns_amt(yf, yt, partner_ids)
    ret_ly = _returns_amt(lf, lt, partner_ids)
    gross = _orev(yf, yt, None) if partner_ids is None else O.sum_field(
        'sale.order', order_domain(yf, yt, None, partner_ids=partner_ids), 'amount_untaxed')
    return {'ytd': ret, 'ytd_ly': ret_ly, 'yoy_pct': P.yoy_pct(ret, ret_ly),
            'rate_pct': (ret / gross * 100.0) if gross else 0.0}


# ---------------------------------------------------------------- VALIDATION
def validate(t=None, team_id=None):
    """Reconcile metrics against independent Odoo aggregates. Returns list of checks.

    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. The one
    cross-BU check (#2, Σ teams == total) only makes sense consolidated, so it runs only when
    team_id is None."""
    t = t or P.today()
    yf, yt = P.ytd(t)
    checks = []

    # 1. Order-level revenue (sale.order) vs line-level revenue (sale.order.line)
    order_rev = _orev(yf, yt, team_id)
    line_rev = O.sum_field('sale.order.line', O.sale_line_domain(yf, yt, team_id), 'price_subtotal')
    gap = order_rev - line_rev
    checks.append({
        'check': 'YTD revenue: order-level == line-level',
        'a': round(order_rev, 2), 'b': round(line_rev, 2),
        'gap': round(gap, 2),
        'ok': abs(gap) <= max(1.0, 0.001 * order_rev)})

    # 2. Sum of per-team revenue == total  (consolidated only — cross-BU)
    if team_id is None:
        team_sum = sum(_orev(yf, yt, tid) for tid in O.TEAM_IDS)
        checks.append({
            'check': 'YTD revenue: Σ(team) == total',
            'a': round(team_sum, 2), 'b': round(order_rev, 2),
            'gap': round(team_sum - order_rev, 2),
            'ok': abs(team_sum - order_rev) <= 1.0})

    # 3. Sum of per-customer revenue == total
    cust_sum = sum(c['revenue'] for c in top_customers(t, limit=10**9, team_id=team_id))
    checks.append({
        'check': 'YTD revenue: Σ(customer) == total',
        'a': round(cust_sum, 2), 'b': round(order_rev, 2),
        'gap': round(cust_sum - order_rev, 2),
        'ok': abs(cust_sum - order_rev) <= 1.0})

    # 4. Σ(category revenue) == line-level total (YTD) — SAME-SOURCE on purpose (wave-14 debt
    # sweep). `by_category` reads the DATASTORE mirror when USE_STORE while `line_rev` above is
    # LIVE Odoo, so this check used to compare two sources and went red on a pending sync
    # (a stable +2,499.90 measured twice 75s apart, 2026-08-02 — the same mixed-source
    # signature as the wave-13 ±$8.5k triple, which healed on resync). The decomposition's
    # claim — every line lands in exactly one category, none lost, none doubled — must be
    # judged against the SAME rows the decomposition consumed; order-vs-line freshness is
    # check 1's job, on one source.
    cat_sum = sum(c['revenue'] for c in by_category(t, limit=10**9, team_id=team_id))
    cat_line_total = line_rev
    if USE_STORE:
        try:
            cat_line_total = sum((amt or 0.0) for _pid, amt in
                                 _line_rev_by_product_store(yf, yt, team_id))
        except Exception:
            pass
    checks.append({
        'check': 'YTD revenue: Σ(category) == line-level total',
        'a': round(cat_sum, 2), 'b': round(cat_line_total, 2),
        'gap': round(cat_sum - cat_line_total, 2),
        'ok': abs(cat_sum - cat_line_total) <= 1.0})

    # 6. Cadence: Σ(frequency-bucket customers) == total LTM customers
    cad = cadence(t, team_id=team_id)
    bucket_n = sum(d['customers'] for d in cad['distribution'])
    checks.append({
        'check': 'Cadence: Σ(frequency buckets) == LTM customers',
        'a': bucket_n, 'b': cad['customers'],
        'gap': bucket_n - cad['customers'],
        'ok': bucket_n == cad['customers']})

    # 7. Decompose ties out: the YTD decomposition total == line-level YTD, and its customer / SKU /
    #    category breakdowns each sum back to that total (the drill-any-number drawer is trustworthy).
    dec = decompose(yf, yt, team_id=team_id)
    checks.append({
        'check': 'Decompose YTD total == line-level revenue',
        'a': round(dec['total'], 2), 'b': round(line_rev, 2),
        'gap': round(dec['total'] - line_rev, 2),
        'ok': abs(dec['total'] - line_rev) <= max(1.0, 0.001 * line_rev)})
    cust_d = sum(c['revenue'] for c in dec['all_customers'])
    sku_d = sum(s['revenue'] for s in dec['all_skus'])
    cat_d = sum(c['revenue'] for c in dec['categories'])
    checks.append({
        'check': 'Decompose: Σ(customers)=Σ(SKUs)=Σ(categories)=total',
        'a': round(cust_d, 2), 'b': round(dec['total'], 2),
        'gap': round(max(abs(cust_d - dec['total']), abs(sku_d - dec['total']), abs(cat_d - dec['total'])), 2),
        'ok': all(abs(x - dec['total']) <= 1.0 for x in (cust_d, sku_d, cat_d))})

    return checks