platform/modules/health.py

"""Data Health module — read-only month-end close & reconciliation checks against Odoo.

Built for the bookkeeper / accountant workflow: pick a period, clear unposted items,
reconcile AR and AP, review accruals & cut-off (unbilled revenue, goods-received-not-invoiced),
and catch errors (below-cost sales, duplicate bills, negative stock, costing gaps).

Each check returns the same shape — category / title / severity / scope / count / dollar
impact / a recommended fix / sample rows — so the page can group and render them uniformly.
Checks take a (date_from, date_to) window:
  - scope 'period'  → transactions dated inside the window (what happened this month)
  - scope 'asof'    → open balances / aging as of date_to (month-end position)
  - scope 'current' → live snapshot, period-independent (e.g. on-hand stock)
READ-ONLY — nothing is ever written.
"""
import sys
import datetime as dt
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.ar as ar_mod

CATEGORIES = ['Posting & completeness', 'Accruals & cut-off', 'Receivables (AR)',
              'Payables (AP)', 'Inventory & COGS', 'Data hygiene']


def _check(key, category, title, severity, scope, count, impact, impact_kind, fix,
           sample=None, cols=None):
    return {'key': key, 'category': category, 'title': title, 'severity': severity,
            'scope': scope, 'count': count, 'impact': impact, 'impact_kind': impact_kind,
            'fix': fix, 'sample': sample or [], 'cols': cols or {}}


def _days(a, b):
    try:
        return (dt.date.fromisoformat(str(a)[:10]) - dt.date.fromisoformat(str(b)[:10])).days
    except Exception:
        return 0


def _ex():
    ex = O.excluded_partner_ids()
    return list(ex) if ex else []


def _sum(model, dom, field):
    """sum_field guarded against empty result sets — Odoo's read_group returns None for a
    sum over zero rows, which XML-RPC can't marshal. Return 0.0 instead of erroring."""
    o = O.get_odoo()
    return O.sum_field(model, dom, field) if o.search_count(model, dom) else 0.0


# ---- Posting & completeness -------------------------------------------------
def _draft_moves(df, dt_):
    o = O.get_odoo()
    dom = [('state', '=', 'draft'), ('move_type', 'in',
            ['out_invoice', 'in_invoice', 'out_refund', 'in_refund', 'entry']),
           ('date', '>=', df), ('date', '<=', dt_)]
    n = o.search_count('account.move', dom)
    rows = o.search_read('account.move', dom,
                         ['name', 'move_type', 'partner_id', 'date', 'amount_total'],
                         limit=25, order='date desc')
    TYPE = {'out_invoice': 'cust invoice', 'in_invoice': 'vendor bill',
            'out_refund': 'cust credit', 'in_refund': 'vendor credit', 'entry': 'journal'}
    sample = [{'entry': r.get('name') or '(draft)', 'type': TYPE.get(r['move_type'], r['move_type']),
               'partner': O.m2o_name(r.get('partner_id')), 'date': str(r.get('date'))[:10],
               'amount': r.get('amount_total') or 0} for r in rows]
    return _check('draft', CATEGORIES[0], 'Draft / unposted entries in period',
                  'high' if n else 'low', 'period', n, None, None,
                  'Post or delete these before closing — draft invoices/bills/journals are not in the books yet.',
                  sample, {'amount': 'money'})


# ---- Accruals & cut-off -----------------------------------------------------
def _dni_lines(df, dt_):
    """Delivered-not-invoiced at LINE level (OCA account_cutoff_picking semantics, re-implemented):
    per sale line, (qty_delivered − qty_invoiced) > 0 valued at the line's effective unit price.
    Unlike order-level invoice_status (which counts whole orders and, under order-basis invoicing,
    goods not yet shipped), this measures exactly the shipped-but-unbilled quantity. Domains can't
    compare two fields, so candidates (qty_delivered > 0) are pulled and reconciled client-side."""
    o = O.get_odoo()
    dom = [('order_id.state', 'in', ['sale', 'done']), ('order_id.team_id', 'in', O.TEAM_IDS),
           ('qty_delivered', '>', 0),
           ('order_id.date_order', '>=', f'{df} 00:00:00'),
           ('order_id.date_order', '<=', f'{dt_} 23:59:59')]
    if _ex():
        dom.append(('order_partner_id', 'not in', _ex()))
    lines = o.search_read('sale.order.line', dom,
                          ['order_id', 'order_partner_id', 'qty_delivered', 'qty_invoiced',
                           'product_uom_qty', 'price_unit', 'price_subtotal'])
    out = []
    for r in lines:
        gap = (r.get('qty_delivered') or 0) - (r.get('qty_invoiced') or 0)
        if gap <= 1e-3:
            continue
        qty = r.get('product_uom_qty') or 0
        unit = (r['price_subtotal'] / qty) if qty else (r.get('price_unit') or 0)
        r['_dni_qty'] = gap
        r['_dni_val'] = gap * unit
        out.append(r)
    return lines, out


def _unbilled_revenue(df, dt_):
    lines, dni = _dni_lines(df, dt_)
    total = sum(r['_dni_val'] for r in dni)
    by_order = {}
    for r in dni:
        onm = O.m2o_name(r['order_id'])
        e = by_order.setdefault(onm, {'order': onm, 'customer': O.m2o_name(r['order_partner_id']),
                                      'unbilled': 0.0})
        e['unbilled'] += r['_dni_val']
    sample = sorted(by_order.values(), key=lambda x: -x['unbilled'])[:25]
    return _check('unbilled', CATEGORIES[1], 'Delivered but not invoiced (revenue to go bill)',
                  'high' if total > 25000 else 'medium', 'period', len(by_order), total, 'dollar',
                  'Shipped quantity exceeds invoiced quantity on these orders — bill the gap now '
                  '(or accrue it at cut-off). Line-level: partial invoices are netted correctly.',
                  sample, {'unbilled': 'money'})


def _grni(df, dt_):
    """Goods received not invoiced — accrue the unbilled vendor cost at close."""
    o = O.get_odoo()
    lines = o.search_read('purchase.order.line',
                          [('order_id.invoice_status', '=', 'to invoice'), ('qty_received', '>', 0)],
                          ['qty_received', 'qty_invoiced', 'price_unit', 'product_id', 'order_id'])
    by = {}
    total = 0.0
    for r in lines:
        d = (r.get('qty_received') or 0) - (r.get('qty_invoiced') or 0)
        if d <= 0:
            continue
        val = d * (r.get('price_unit') or 0)
        total += val
        po = O.m2o_name(r.get('order_id'))
        e = by.setdefault(po, {'po': po, 'accrual': 0.0})
        e['accrual'] += val
    rows = sorted(by.values(), key=lambda x: -x['accrual'])[:25]
    return _check('grni', CATEGORIES[1], 'Goods received, not yet invoiced (GRNI accrual)',
                  'high' if total > 25000 else 'medium', 'current', len(by), total, 'dollar',
                  'Accrue this vendor cost at period close (Dr inventory/COGS, Cr GRNI) until the bills arrive.',
                  rows, {'accrual': 'money'})


# ---- Receivables (AR) -------------------------------------------------------
def _ancient_ar(df, dt_):
    o = O.get_odoo()
    cutoff = (dt.date.fromisoformat(dt_) - dt.timedelta(days=365)).isoformat()
    dom = [('move_type', '=', 'out_invoice'), ('state', '=', 'posted'),
           ('payment_state', 'in', ['not_paid', 'partial']), ('invoice_date_due', '<', cutoff)]
    if _ex():
        dom.append(('partner_id', 'not in', _ex()))
    n = o.search_count('account.move', dom)
    amt = _sum('account.move', dom, 'amount_residual_signed')
    rows = o.search_read('account.move', dom,
                         ['name', 'partner_id', 'invoice_date_due', 'amount_residual_signed'],
                         limit=25, order='invoice_date_due asc')
    sample = [{'invoice': r['name'], 'customer': O.m2o_name(r['partner_id']),
               'days_overdue': _days(dt_, r['invoice_date_due']),
               'open': r.get('amount_residual_signed') or 0} for r in rows]
    return _check('ancient_ar', CATEGORIES[2], 'Invoices 365+ days overdue (write-off candidates)',
                  'high' if amt > 25000 else 'medium', 'asof', n, amt, 'dollar',
                  'Decide: escalate, settle, or write off. Provision the doubtful portion at close.',
                  sample, {'open': 'money', 'days_overdue': ('int', 'days overdue')})


def _ar_recon(df, dt_):
    rows = ar_mod.reconciliation_flags(limit=25)
    allr = ar_mod.reconciliation_flags(limit=10 ** 9)
    amt = sum(abs(r['gap']) for r in allr)
    return _check('ar_recon', CATEGORIES[2], 'AR sub-ledger vs partner balance mismatches',
                  'medium', 'current', len(allr), amt, 'dollar',
                  'Reconcile unapplied payments/credits so the AR sub-ledger ties to the partner balance.',
                  rows, {'odoo_receivable': 'money', 'open_docs': 'money', 'gap': 'money'})


def _stale_ar(df, dt_):
    o = O.get_odoo()
    cutoff = (dt.date.fromisoformat(dt_) - dt.timedelta(days=180)).isoformat()
    g = o.read_group('sale.order',
                     [('state', 'in', ['sale', 'done']), ('team_id', 'in', O.TEAM_IDS),
                      ('date_order', '>=', f'{cutoff} 00:00:00'), ('date_order', '<=', f'{dt_} 23:59:59')],
                     ['partner_id'], ['partner_id'], lazy=False)
    active = {O.m2o_id(r['partner_id']) for r in g if r.get('partner_id')}
    parts = o.search_read('res.partner', [('credit', '>', 500)] +
                          ([('id', 'not in', _ex())] if _ex() else []), ['name', 'credit'])
    stale = [p for p in parts if p['id'] not in active]
    amt = sum(p['credit'] for p in stale)
    sample = sorted([{'customer': p['name'], 'owes': p['credit']} for p in stale],
                    key=lambda x: -x['owes'])[:25]
    return _check('stale_ar', CATEGORIES[2], 'Customers owing money but quiet 180+ days',
                  'high' if amt > 25000 else 'medium', 'asof', len(stale), amt, 'dollar',
                  'Push collection and a win-back; provision if uncollectible.', sample, {'owes': 'money'})


# ---- Payables (AP) ----------------------------------------------------------
def _ap_overdue(df, dt_):
    o = O.get_odoo()
    dom = [('move_type', '=', 'in_invoice'), ('state', '=', 'posted'),
           ('payment_state', 'in', ['not_paid', 'partial']), ('invoice_date_due', '<', dt_)]
    n = o.search_count('account.move', dom)
    amt = abs(_sum('account.move', dom, 'amount_residual_signed'))
    rows = o.search_read('account.move', dom,
                         ['name', 'partner_id', 'invoice_date_due', 'amount_residual_signed', 'ref'],
                         limit=25, order='invoice_date_due asc')
    sample = [{'bill': r.get('ref') or r['name'], 'vendor': O.m2o_name(r['partner_id']),
               'days_overdue': _days(dt_, r['invoice_date_due']),
               'open': abs(r.get('amount_residual_signed') or 0)} for r in rows]
    return _check('ap_overdue', CATEGORIES[3], 'Vendor bills overdue (as of period end)',
                  'medium', 'asof', n, amt, 'dollar',
                  'Schedule/clear overdue payables; confirm none are duplicates before paying.',
                  sample, {'open': 'money', 'days_overdue': ('int', 'days overdue')})


def _duplicate_bills(df, dt_):
    o = O.get_odoo()
    dom = [('move_type', '=', 'in_invoice'), ('state', '=', 'posted'),
           ('invoice_date', '>=', df), ('invoice_date', '<=', dt_), ('ref', '!=', False)]
    rows = o.search_read('account.move', dom, ['name', 'partner_id', 'ref', 'amount_total', 'invoice_date'])
    seen = {}
    for r in rows:
        k = (O.m2o_id(r.get('partner_id')), str(r.get('ref')).strip().lower(), round(r.get('amount_total') or 0, 2))
        seen.setdefault(k, []).append(r)
    dups = [v for v in seen.values() if len(v) > 1]
    sample = sorted([{'vendor': O.m2o_name(v[0]['partner_id']), 'ref': v[0]['ref'],
                      'amount': v[0]['amount_total'] or 0, 'copies': len(v)} for v in dups],
                    key=lambda x: -x['amount'])[:25]
    amt = sum((len(v) - 1) * (v[0]['amount_total'] or 0) for v in dups)
    return _check('dup_bills', CATEGORIES[3], 'Possible duplicate vendor bills (same vendor/ref/amount)',
                  'high' if dups else 'low', 'period', len(dups), amt, 'dollar',
                  'Review before paying — duplicate bills cause double payment.', sample, {'amount': 'money'})


# ---- Inventory & COGS -------------------------------------------------------
def _negative_margin(df, dt_):
    o = O.get_odoo()
    dom = O.sale_line_domain(df, dt_, extra=[('margin', '<', 0)])
    n = o.search_count('sale.order.line', dom)
    loss = _sum('sale.order.line', dom, 'margin')
    g = o.read_group('sale.order.line', dom, ['margin:sum', 'product_id'], ['product_id'], lazy=False)
    rows = sorted([{'product': O.m2o_name(r['product_id']), 'margin_lost': r.get('margin') or 0}
                   for r in g if r.get('product_id')], key=lambda x: x['margin_lost'])[:25]
    return _check('neg_margin', CATEGORIES[4], 'Sales below cost (negative margin) in period',
                  'high' if loss < -5000 else 'medium', 'period', n, loss, 'dollar',
                  'Reprice or stop selling these SKUs; check for costing errors driving false losses.',
                  rows, {'margin_lost': 'money'})


def _negative_stock(df, dt_):
    o = O.get_odoo()
    dom = [('location_id.usage', '=', 'internal'), ('quantity', '<', 0)]
    n = o.search_count('stock.quant', dom)
    rows = o.search_read('stock.quant', dom, ['product_id', 'quantity'], limit=25, order='quantity asc')
    sample = [{'product': O.m2o_name(r['product_id']), 'on_hand': r['quantity']} for r in rows]
    return _check('neg_stock', CATEGORIES[4], 'Negative on-hand stock (impossible quantities)',
                  'high' if n > 20 else 'medium', 'current', n, None, None,
                  'Fix receipts/adjustments — negative on-hand distorts inventory valuation and COGS.',
                  sample, {'on_hand': 'num'})


def _product_master(df, dt_):
    o = O.get_odoo()
    prods = o.search_read('product.product', [('active', '=', True), ('default_code', '!=', False)],
                          ['id', 'default_code', 'name', 'standard_price', 'type', 'sale_ok', 'categ_id'])
    q = o.read_group('stock.quant', [('location_id.usage', '=', 'internal')],
                     ['product_id', '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')}
    roots = {c['id'] for c in o.search_read('product.category', [('parent_id', '=', False)], ['id'])}
    uncosted, dup, uncat = [], {}, []
    for p in prods:
        oh = onhand.get(p['id'], 0.0)
        if p.get('type') == 'product' and oh > 0 and (p.get('standard_price') or 0) <= 0:
            uncosted.append({'sku': p['default_code'], 'product': p['name'], 'on_hand': oh})
        dup.setdefault(str(p['default_code']).strip(), []).append(p['name'])
        if p.get('sale_ok') and O.m2o_id(p.get('categ_id')) in roots:
            uncat.append({'sku': p['default_code'], 'product': p['name']})
    dups = [{'sku': k, 'records': len(v)} for k, v in dup.items() if len(v) > 1]
    c_unc = _check('uncosted', CATEGORIES[4], 'In-stock SKUs with zero cost (valuation gap)',
                   'medium', 'current', len(uncosted), None, None,
                   'Set standard cost — these read as $0 inventory and distort margin and valuation.',
                   sorted(uncosted, key=lambda x: -x['on_hand'])[:25], {'on_hand': 'num'})
    c_dup = _check('dup_codes', CATEGORIES[5], 'Duplicate active SKU codes', 'medium', 'current',
                   len(dups), None, None, 'Merge/retire duplicates — shared codes double-count.',
                   sorted(dups, key=lambda x: -x['records'])[:25], {'records': 'int'})
    c_cat = _check('uncat', CATEGORIES[5], 'Sellable products with no real category', 'low', 'current',
                   len(uncat), None, None, 'Assign a category so by-category reporting works.', uncat[:25], {})
    return [c_unc, c_dup, c_cat]


def _orders_no_rep(df, dt_):
    o = O.get_odoo()
    dom = [('state', 'in', ['sale', 'done']), ('team_id', 'in', O.TEAM_IDS), ('user_id', '=', False),
           ('date_order', '>=', f'{df} 00:00:00'), ('date_order', '<=', f'{dt_} 23:59:59')]
    n = o.search_count('sale.order', dom)
    amt = _sum('sale.order', dom, 'amount_untaxed')
    rows = o.search_read('sale.order', dom, ['name', 'partner_id', 'amount_untaxed'],
                         limit=25, order='amount_untaxed desc')
    sample = [{'order': r['name'], 'customer': O.m2o_name(r['partner_id']), 'amount': r['amount_untaxed']}
              for r in rows]
    return _check('no_rep', CATEGORIES[5], 'Orders with no salesperson assigned', 'low', 'period',
                  n, amt, 'dollar', 'Assign a salesperson for correct attribution/commissions.',
                  sample, {'amount': 'money'})


# ---- orchestration ----------------------------------------------------------
def _overdue_activities(df, dt_):
    """Scheduled activities (mail.activity) past their deadline — at probe time ALL 811 open
    activities were overdue: the activity system is dead-lettered, so nothing scheduled there
    can be trusted as a reminder."""
    o = O.get_odoo()
    today = P.today().isoformat()
    n_open = o.search_count('mail.activity', [])
    over = O.search_read('mail.activity', [('date_deadline', '<', today)],
                         ['user_id', 'date_deadline', 'res_model', 'summary'])
    by_user = {}
    for a in over:
        u = O.m2o_name(a.get('user_id')) or '(unassigned)'
        e = by_user.setdefault(u, {'user': u, 'overdue': 0, 'oldest': today})
        e['overdue'] += 1
        d = str(a.get('date_deadline') or today)[:10]
        if d < e['oldest']:
            e['oldest'] = d
    sample = sorted(by_user.values(), key=lambda x: -x['overdue'])[:15]
    return _check('overdue_activities', CATEGORIES[5], 'Scheduled activities past deadline',
                  'medium' if len(over) < 50 else 'high', 'current', len(over), None, 'count',
                  f'{len(over)} of {n_open} open activities are overdue - clear or delete them; '
                  'a reminder system where everything is late reminds nobody of anything.',
                  sample, {'overdue': 'int'})


def _partner_tag_hygiene(df, dt_):
    """Partner tags duplicate the BU concept (Fisch/Royal tags vs team_id) — flag duplicate tag
    names and tag-vs-team mismatches (a Royal-tagged customer on the Fisch team)."""
    o = O.get_odoo()
    tags = O.search_read('res.partner.category', [], ['name'])
    names = {}
    for t in tags:
        names.setdefault((t['name'] or '').strip().upper(), []).append(t['id'])
    dups = {k: v for k, v in names.items() if len(v) > 1}
    issues = [{'issue': f'duplicate tag name "{k}" ({len(v)} tags)', 'count': len(v)} for k, v in dups.items()]
    mism = 0
    sample_m = []
    for tag_name, team in (('FISCH', 6), ('ROYAL', 5)):   # tag says one BU, team says the OTHER
        ids = [i for k, v in names.items() if k == tag_name for i in v]
        if ids:
            rows = O.search_read('res.partner',
                                 [('category_id', 'in', ids), ('team_id', '=', team)],
                                 ['name'], limit=10)
            n = o.search_count('res.partner', [('category_id', 'in', ids), ('team_id', '=', team)])
            mism += n
            sample_m += [{'issue': f'tagged {tag_name.title()} but on the other BU team',
                          'partner': r['name']} for r in rows[:5]]
    total = sum(i['count'] for i in issues) + mism
    return _check('tag_hygiene', CATEGORIES[5], 'Partner tag hygiene (duplicates / BU mismatch)',
                  'low', 'current', total, None, 'count',
                  'Merge duplicate tags; align Fisch/Royal tags with the sales team (tags feed '
                  'segmentation - a mismatch silently mis-buckets the customer).',
                  issues + sample_m, {})


def _no_terms_invoices(df, dt_):
    """Posted customer invoices with NO payment terms — due date defaults silently and dunning
    logic has nothing to anchor on."""
    o = O.get_odoo()
    dom = [('move_type', '=', 'out_invoice'), ('state', '=', 'posted'),
           ('invoice_payment_term_id', '=', False),
           ('invoice_date', '>=', f'{df}'), ('invoice_date', '<=', f'{dt_}')]
    if _ex():
        dom.append(('partner_id', 'not in', _ex()))
    n = o.search_count('account.move', dom)
    amt = _sum('account.move', dom, 'amount_total')
    rows = O.search_read('account.move', dom, ['name', 'partner_id', 'invoice_date', 'amount_total'],
                         limit=15, order='amount_total desc')
    sample = [{'invoice': r['name'], 'customer': O.m2o_name(r['partner_id']),
               'date': str(r['invoice_date'])[:10], 'amount': r['amount_total']} for r in rows]
    return _check('no_terms', CATEGORIES[2], 'Invoices posted without payment terms',
                  'low' if amt < 25000 else 'medium', 'period', n, amt, 'dollar',
                  'Set a default payment term on these customers - no terms means the due date '
                  'and any dunning cadence are meaningless for them.',
                  sample, {'amount': 'money'})


# Economic-nexus (Wayfair) SALES thresholds by state, 2026 — transaction-count tests are mostly
# repealed so only the revenue test is monitored. None = no state sales tax (NH/OR/MT/DE).
# NY is $500k AND 100 sales; AK is local-option (monitored at $100k). Source: Avalara/TaxJar
# state guides — VERIFY WITH THE CPA before registering anywhere; this is a radar, not advice.
NEXUS_THRESHOLDS = {
    'AL': 250000, 'AK': 100000, 'AZ': 100000, 'AR': 100000, 'CA': 500000, 'CO': 100000,
    'CT': 100000, 'DE': None, 'FL': 100000, 'GA': 100000, 'HI': 100000, 'ID': 100000,
    'IL': 100000, 'IN': 100000, 'IA': 100000, 'KS': 100000, 'KY': 100000, 'LA': 100000,
    'ME': 100000, 'MD': 100000, 'MA': 100000, 'MI': 100000, 'MN': 100000, 'MS': 250000,
    'MO': 100000, 'MT': None, 'NE': 100000, 'NV': 100000, 'NH': None, 'NJ': 100000,
    'NM': 100000, 'NY': 500000, 'NC': 100000, 'ND': 100000, 'OH': 100000, 'OK': 100000,
    'OR': None, 'PA': 100000, 'RI': 100000, 'SC': 100000, 'SD': 100000, 'TN': 100000,
    'TX': 500000, 'UT': 100000, 'VT': 100000, 'VA': 100000, 'WA': 100000, 'WV': 100000,
    'WI': 100000, 'WY': 100000, 'DC': 100000}


def nexus(t=None):
    """Economic-nexus radar: trailing-12m invoiced revenue (posted invoices − credit notes) by
    SHIP-TO state vs each state's Wayfair threshold. The company collects ZERO sales tax (all
    resale-exempt) — crossing a threshold unnoticed creates back-liability. Trailing 12m is a
    PROXY (states legally measure current/previous calendar year); status: OVER / >75% warming /
    monitoring. Home state carries physical nexus regardless."""
    t = t or P.today()
    d12 = (t - dt.timedelta(days=365)).isoformat()

    def _by_ship(mt, sign):
        out = {}
        for g in O.read_group('account.move',
                              [('move_type', '=', mt), ('state', '=', 'posted'),
                               ('invoice_date', '>=', d12)],
                              ['amount_untaxed:sum'], ['partner_shipping_id'], lazy=False):
            pid = O.m2o_id(g.get('partner_shipping_id'))
            if pid:
                e = out.setdefault(pid, [0.0, 0])
                e[0] += sign * (g.get('amount_untaxed') or 0.0)
                e[1] += g.get('__count') or 0
        return out
    inv = _by_ship('out_invoice', 1)
    for pid, (v, n) in _by_ship('out_refund', -1).items():
        e = inv.setdefault(pid, [0.0, 0])
        e[0] += v            # refunds reduce state revenue; their doc count isn't a 'sale'
    pids = list(inv.keys())
    pstate = {}
    for i in range(0, len(pids), 5000):
        for p in O.search_read('res.partner', [('id', 'in', pids[i:i + 5000])],
                               ['state_id', 'country_id']):
            code = None
            if p.get('state_id'):
                # state m2o name is the full name; pull the code from the state record below
                code = O.m2o_id(p['state_id'])
            pstate[p['id']] = {'state_rid': code,
                               'country': O.m2o_name(p.get('country_id')) or ''}
    srids = list({v['state_rid'] for v in pstate.values() if v['state_rid']})
    scode = {}
    for i in range(0, len(srids), 5000):
        for s in O.search_read('res.country.state', [('id', 'in', srids[i:i + 5000])],
                               ['code', 'country_id']):
            scode[s['id']] = {'code': s.get('code'), 'country': O.m2o_name(s.get('country_id'))}
    per, unmapped_val, unmapped_n = {}, 0.0, 0
    for pid, (val, n) in inv.items():
        ps = pstate.get(pid) or {}
        sc = scode.get(ps.get('state_rid')) or {}
        code, ctry = sc.get('code'), (sc.get('country') or ps.get('country') or '')
        if code and ('United States' in ctry or ctry == ''):
            e = per.setdefault(code, {'state': code, 'revenue_12m': 0.0, 'n_invoices': 0})
            e['revenue_12m'] += val
            e['n_invoices'] += n
        else:
            unmapped_val += val
            unmapped_n += n
    rows = []
    for e in per.values():
        th = NEXUS_THRESHOLDS.get(e['state'])
        e['threshold'] = th
        e['pct_of_threshold'] = (e['revenue_12m'] / th * 100) if th else None
        e['status'] = ('no sales tax' if th is None
                       else 'OVER' if e['revenue_12m'] >= th
                       else 'warming' if e['revenue_12m'] >= th * 0.75
                       else 'monitor')
        rows.append(e)
    rows.sort(key=lambda x: -(x['pct_of_threshold'] or 0))
    total = sum(e['revenue_12m'] for e in per.values()) + unmapped_val
    return {'rows': rows, 'unmapped_value': unmapped_val, 'unmapped_n': unmapped_n,
            'n_over': sum(1 for r in rows if r['status'] == 'OVER'),
            'n_warming': sum(1 for r in rows if r['status'] == 'warming'),
            '_total_built': total, '_d12': d12}


def nexus_validate(nx):
    """Σ(state revenue) + unmapped == server net invoiced (partition, two aggregation paths)."""
    d12 = nx['_d12']
    srv = (O.sum_field('account.move',
                       [('move_type', '=', 'out_invoice'), ('state', '=', 'posted'),
                        ('invoice_date', '>=', d12)], 'amount_untaxed')
           - O.sum_field('account.move',
                         [('move_type', '=', 'out_refund'), ('state', '=', 'posted'),
                          ('invoice_date', '>=', d12)], 'amount_untaxed'))
    return [{'check': 'Nexus: Σ(state revenue) + unmapped == server net invoiced (12m)',
             'a': round(nx['_total_built'], 2), 'b': round(srv, 2),
             'gap': round(nx['_total_built'] - srv, 2),
             'ok': abs(nx['_total_built'] - srv) <= max(1.0, abs(srv) * 0.001)}]


def _window(date_from=None, date_to=None, t=None):
    t = t or P.today()
    if date_from and date_to:
        return date_from, date_to
    return P.ytd(t)  # default = year to date


def run_all(date_from=None, date_to=None, t=None):
    df, dt_ = _window(date_from, date_to, t)
    checks = [_draft_moves(df, dt_), _unbilled_revenue(df, dt_), _grni(df, dt_),
              _ancient_ar(df, dt_), _ar_recon(df, dt_), _stale_ar(df, dt_),
              _ap_overdue(df, dt_), _duplicate_bills(df, dt_),
              _negative_margin(df, dt_), _negative_stock(df, dt_), _orders_no_rep(df, dt_),
              _overdue_activities(df, dt_), _partner_tag_hygiene(df, dt_),
              _no_terms_invoices(df, dt_)]
    checks += _product_master(df, dt_)
    sev = {'high': 0, 'medium': 1, 'low': 2}
    return sorted([c for c in checks if c],
                  key=lambda c: (sev.get(c['severity'], 9), -abs(c['impact'] or 0)))


def by_category(date_from=None, date_to=None, t=None):
    checks = run_all(date_from, date_to, t)
    return {cat: [c for c in checks if c['category'] == cat] for cat in CATEGORIES}


def summarize(checks):
    """Build the summary KPIs from an already-computed checks list (no extra Odoo calls)."""
    flagged = [c for c in checks if c['count'] > 0]
    return {
        'total_checks': len(checks),
        'issues': len(flagged),
        'high': sum(1 for c in flagged if c['severity'] == 'high'),
        'dollar_at_stake': sum(abs(c['impact']) for c in flagged if c['impact_kind'] == 'dollar'),
        'clean': len(checks) - len(flagged),
    }


def summary(date_from=None, date_to=None, t=None):
    return summarize(run_all(date_from, date_to, t))


def validate(date_from=None, date_to=None, t=None):
    """Light, independent reconciliations (no full run_all): aggregate vs row-by-row."""
    df, dt_ = _window(date_from, date_to, t)
    o = O.get_odoo()
    out = []

    # DNI: row-by-row delivered/invoiced quantity sums vs independent read_group aggregates
    # over the SAME candidate domain (two aggregation paths must agree).
    dom = [('order_id.state', 'in', ['sale', 'done']), ('order_id.team_id', 'in', O.TEAM_IDS),
           ('qty_delivered', '>', 0),
           ('order_id.date_order', '>=', f'{df} 00:00:00'),
           ('order_id.date_order', '<=', f'{dt_} 23:59:59')]
    if _ex():
        dom.append(('order_partner_id', 'not in', _ex()))
    lines, _ = _dni_lines(df, dt_)
    row_qd = sum(r.get('qty_delivered') or 0 for r in lines)
    row_qi = sum(r.get('qty_invoiced') or 0 for r in lines)
    agg_qd = _sum('sale.order.line', dom, 'qty_delivered')
    agg_qi = _sum('sale.order.line', dom, 'qty_invoiced')
    out.append({'check': 'DNI: row Σ qty_delivered == read_group Σ (candidate lines, period)',
                'a': round(row_qd, 2), 'b': round(agg_qd, 2), 'gap': round(row_qd - agg_qd, 2),
                'ok': abs(row_qd - agg_qd) <= 0.01})
    out.append({'check': 'DNI: row Σ qty_invoiced == read_group Σ (candidate lines, period)',
                'a': round(row_qi, 2), 'b': round(agg_qi, 2), 'gap': round(row_qi - agg_qi, 2),
                'ok': abs(row_qi - agg_qi) <= 0.01})

    sdom = [('location_id.usage', '=', 'internal'), ('quantity', '<', 0)]
    n_neg = o.search_count('stock.quant', sdom)
    ids = o.search_read('stock.quant', sdom, ['id'])
    out.append({'check': 'Negative-stock: search_count == len(search_read)',
                'a': n_neg, 'b': len(ids), 'gap': n_neg - len(ids), 'ok': n_neg == len(ids)})
    return out