platform/modules/customer_data.py

"""Customer List — build, filter and save customer lists (replaces My Day, owner IA 2026-07-23).

The page is a customer-list WORKBENCH in two parts:
  1. BUILDER — the whole scoped book, one row per customer with a consistent metric surface
     (YTD / LY revenue, at-risk $, cadence, overdue days, estimated missed $ …): filter, sort,
     SELECT customers and add them to a list.
  2. LISTS — saved lists whose membership is a visible, editable FORMULA (a rule set over the
     metric columns) plus hand-picked members. 'Call list' and 'Win-back' ship as TEMPLATES —
     the exact formulas the old My Day queues used — and the user can change or reset them.

Composes already-validated customers-module functions, so the page's counts derive FROM the
lists it renders (rule 8b). ⚠ **This header used to end "no separate validate()" and W30-T33
made that false** — see `validate()` below. The claim held only while every column composed an
already-reconciled customers-module function; the ADDRESS family does not, so it needed its own
tie. Persistence: HF store key 'customer_lists'
({username: {list_name: {rules, sort, members, note}}}); templates are virtual until edited.

The user↔agent link (users.py 'agent' field) scopes the pool to that agent's book; users with
no link (owner/CFO/admin) see the whole book.
"""
import core.odoo as O
import core.periods as P
import core.store as store
import core.table_store as table_store
import modules.ar as ar
import modules.customers as cust
import modules.map as map_mod
import modules.sales as S

LIMIT = 500          # honest builder display cap — always shown with the full count, never silent
KEY = 'customer_lists'
TABLE_KEY = 'customer_table_workspace'

# ------------------------------------------------------------------ the metric pool
# Field metadata for the rule builder: key -> (label, kind). 'num' fields take >=/<=/>/</=,
# 'text' fields take contains/=. Every field is a column of pool() rows.
FIELDS = {
    'customer':         ('Customer', 'text'),
    'status':           ('Status', 'text'),
    'agent':            ('Agent', 'text'),
    'city':             ('City', 'text'),
    'state':            ('State', 'text'),
    'revenue_ytd':      ('YTD $', 'num'),
    'revenue_ly':       ('LY (same period) $', 'num'),
    'yoy_pct':          ('YoY %', 'num'),
    'at_risk':          ('At risk $', 'num'),
    'ltm_rev':          ('LTM $', 'num'),
    'orders_24m':       ('Orders (24M)', 'num'),
    'aov':              ('Avg order $', 'num'),
    'days_since':       ('Days since last order', 'num'),
    'typical_gap_days': ('Typical gap (days)', 'num'),
    'overdue_days':     ('Overdue vs cadence (days)', 'num'),
    'est_missed':       ('Est. missed $', 'num'),
}
NUM_OPS = ('>=', '<=', '>', '<', '=')
TEXT_OPS = ('contains', '=')


def agent_pids(agent_name):
    """Partner ids of an agent's book (None = unscoped/whole book)."""
    return cust.agent_partner_ids(agent_name) if agent_name else None


#: Wave 17 R2 — `ar` bucket label -> the grid's column key. ONE mapping, read by the attribute
#: builder, the empty-row default and validate(), so a bucket cannot exist under two spellings.
AR_BUCKET_FIELDS = {'1-30': 'ar_aged_1_30', '31-60': 'ar_aged_31_60',
                    '61-90': 'ar_aged_61_90', '90+': 'ar_aged_90_plus'}


def _ar_attrs(t=None):
    """{pid: {ar_open, ar_overdue, ar_outstanding, ar_exposure}} — the AR family.

    Wave 21 R1: `ar_open` is the NOT-YET-DUE half of a disjoint split (label "AR current $" —
    key kept so saved views keep working); `ar_outstanding` is the total (current + overdue),
    which is what "open AR" means everywhere else.

    Composed from `modules/ar.credit_exposure()`, NOT re-derived: it is already reconciled to
    Odoo by `ar.validate()`, so the Customer table and the Collections page cannot disagree about
    what a customer owes. Read from the UNFILTERED `_all_rows` — `rows` is a display-truncated
    top-40, and a column built off that would silently blank everybody else.

    ⚠ `credit_limit` is deliberately NOT a column: 20 of 1,548 customers have one. EXPOSURE is
    the number that answers "how much are we out on this customer", and it needs no limit.

    ⚠ `days_to_pay` is NOT COMPUTED here — it is READ from a nightly snapshot. MEASURED
    2026-07-27: `ar.credit_exposure()` costs 9s; `ar.days_to_pay()` costs **301s**, because
    settlement date means walking three years of reconciled receivable lines and their
    full-reconcile groups. Computing it in `pool()` would add five minutes to every Customer-table
    build and every Space container start; computing it LAZILY would just move those five minutes
    onto whoever opened the page first. So the expensive half runs on the app's existing
    background store-sync thread (`ar.refresh_days_to_pay_snapshot`) and this reads the answer
    out of the store for nothing. A missing snapshot yields a blank column, never a slow page.
    """
    out = {}
    try:
        exp = ar.credit_exposure(t)
        for r in exp.get('_all_rows') or []:
            out[r['pid']] = {
                'ar_open': r.get('open', 0.0) or 0.0,
                'ar_overdue': r.get('overdue', 0.0) or 0.0,
                # Wave 21 R1 — the TOTAL owed. Sum of the two halves ar.credit_exposure already
                # reconciles to Odoo's residual read_group (ar.py validate), so no second oracle.
                'ar_outstanding': (r.get('open', 0.0) or 0.0) + (r.get('overdue', 0.0) or 0.0),
                'ar_exposure': r.get('exposure', 0.0) or 0.0,
                # Wave 17 R2 — the aging split, so a saved VIEW can be the collections
                # worklist. Keys mirror `ar.OVERDUE_BUCKETS` through AR_BUCKET_FIELDS; the four
                # sum EXACTLY to `ar_overdue` (gated in validate()).
                **{k: r.get(f'aged_{b}', 0.0) or 0.0
                   for b, k in AR_BUCKET_FIELDS.items()},
            }
    except Exception:
        pass
    for pid, avg in ar.days_to_pay_snapshot().items():
        # A customer with an AR row but no settled invoice in the window keeps `None` — blank,
        # not 0. "Pays in 0 days" is a claim, and a false one.
        out.setdefault(pid, {})['days_to_pay'] = avg
    return out


def _mix_attrs(pids=None, t=None):
    """{pid: {top_category, sku_count, top_sku, top_category_pct}} — the product-mix family.

    ONE read_group over sale.order.line by (partner, product) across the LTM window, then the
    product->category map, then aggregation in Python. Line grain is the only grain that can
    answer "what does this customer buy", and it is deliberately ALL-CHANNEL scope-wise for the
    same reason inventory analysis is (see [[ri-channel-scope-amazon]]): a customer's product mix
    is a fact about the customer, not about a sales team.

    `top_category_pct` is that category's share of the customer's LTM line revenue — the number
    that says whether "top" means dominant or merely first.

    ⚠ There is NO `team_id` parameter, and its absence is the point rather than an omission: for
    a BU-scoped viewer the revenue columns beside this one ARE team-scoped while these are not,
    so "top category" can name products bought through a channel that viewer's revenue figures
    exclude. That is the same trade the inventory modules make, and it is the right one — a
    customer's product mix is a fact about the customer.
    """
    t = t or P.today()
    mf, mt = P.ltm(t)
    # ⚠ `pids` NARROWS THE RESULT, NEVER THE DOMAIN — and that is not a style choice. Putting
    # the caller's ~1,500 partner ids into an `in` clause on a LINE-grain read_group makes Odoo
    # answer **502 Bad Gateway**; the same query unfiltered returns in 22s. Measured 2026-07-27,
    # after the filtered form took the live Customer page past 20 MINUTES without rendering.
    # The grouped result is per-partner anyway, so dropping the extras in Python costs nothing.
    dom = [('order_id.state', 'in', ['sale', 'done']),
           ('order_id.date_order', '>=', str(mf)), ('order_id.date_order', '<=', str(mt))]
    keep = set(pids) if pids else None
    rows = O.read_group('sale.order.line', dom, ['price_subtotal:sum'],
                        ['order_partner_id', 'product_id'], lazy=False)
    prod_ids = {O.m2o_id(r.get('product_id')) for r in rows if r.get('product_id')}
    prod_ids.discard(None)
    cats = {}
    plist = list(prod_ids)
    for i in range(0, len(plist), 5000):
        for p in O.search_read('product.product', [('id', 'in', plist[i:i + 5000])], ['categ_id']):
            # `categ_id`'s display name is the full PATH ("All / Foams & Finishes / Styrofoam").
            # The column takes the LEAF, because a column and a group label have to be readable —
            # and the path's own root ("All") carries no information. ⚠ Two categories that share
            # a leaf name would merge in a group-by; measured against this catalogue that does not
            # happen, and readability is worth more than defending against a rename.
            full = O.m2o_name(p.get('categ_id')) or ''
            cats[p['id']] = (full.split(' / ')[-1].strip() or '(none)') if full else '(none)'

    per = {}
    for r in rows:
        pid = O.m2o_id(r.get('order_partner_id'))
        prod = O.m2o_id(r.get('product_id'))
        if pid is None or prod is None or (keep is not None and pid not in keep):
            continue
        val = r.get('price_subtotal', 0.0) or 0.0
        e = per.setdefault(pid, {'skus': set(), 'by_cat': {}, 'by_sku': {}, 'total': 0.0})
        e['skus'].add(prod)
        e['by_cat'][cats.get(prod, '(none)')] = e['by_cat'].get(cats.get(prod, '(none)'), 0.0) + val
        e['by_sku'][prod] = e['by_sku'].get(prod, 0.0) + val
        e['total'] += val

    top_prod_ids = {max(e['by_sku'], key=e['by_sku'].get) for e in per.values() if e['by_sku']}
    names = {}
    tlist = list(top_prod_ids)
    for i in range(0, len(tlist), 5000):
        for p in O.search_read('product.product', [('id', 'in', tlist[i:i + 5000])], ['name']):
            names[p['id']] = p.get('name') or ''

    out = {}
    for pid, e in per.items():
        cat = max(e['by_cat'], key=e['by_cat'].get) if e['by_cat'] else '(none)'
        sku = max(e['by_sku'], key=e['by_sku'].get) if e['by_sku'] else None
        out[pid] = {
            'top_category': cat,
            'top_category_pct': (e['by_cat'][cat] / e['total']) if e['total'] else None,
            'sku_count': len(e['skus']),
            'top_sku': names.get(sku, '(none)') if sku is not None else '(none)',
        }
    return out


def _salesperson_attrs(pids=None, t=None):
    """{pid: salesperson} — the DOMINANT order-taker over LTM, by order count.

    ⚠ Salesperson is NOT the Agent, and they are not interchangeable (owner, 2026-07-27):
    the SALESPERSON is whoever entered and took the order, the AGENT is the person the customer
    is assigned to and who earns the commission — including when the customer orders through the
    office and somebody else keys it in. So this reads `sale.order.user_id` (the ORDER's taker,
    fully populated: 20 people over $6.2M LTM) and NOT `res.partner.user_id`, which is a
    different, near-empty field.

    A customer can have several over a year, so the column takes the one who took the MOST of
    their orders — a single value the table can group and filter cleanly. Ties break toward the
    larger revenue.
    """
    t = t or P.today()
    mf, mt = P.ltm(t)
    # Same rule as `_mix_attrs`: the pid list narrows the RESULT, not the domain. Order grain is
    # far smaller than line grain so this one never actually 502'd, but a caller's id list has no
    # business being pushed into a query whose grouping already answers per partner.
    keep = set(pids) if pids else None
    dom = S.order_domain(str(mf), str(mt), None)
    rows = O.read_group('sale.order', dom, ['amount_untaxed:sum'],
                        ['partner_id', 'user_id'], lazy=False)
    per = {}
    for r in rows:
        pid = O.m2o_id(r.get('partner_id'))
        if pid is None or (keep is not None and pid not in keep):
            continue
        who = O.m2o_name(r.get('user_id')) or '(none)'
        n, rev = r.get('__count', 0) or 0, r.get('amount_untaxed', 0.0) or 0.0
        cur = per.get(pid)
        if cur is None or (n, rev) > (cur[1], cur[2]):
            per[pid] = (who, n, rev)
    return {pid: v[0] for pid, v in per.items()}


def pool(agent_name=None, team_id=None):
    """One row per customer across the scoped book: the union of YTD buyers, same-period-LY
    buyers and anyone with an order in the 24-month cadence window — so lapsed/win-back
    accounts are in the pool, not just YTD actives. Reuses the customers module's validated
    building blocks (_cust_rev / _cadence_bulk / _partner_attrs) so every metric matches the
    Customers page definitions; est_missed is the Call-list formula (min(cycles missed, 3) × AOV)."""
    return _pool_build(agent_name, team_id, limit=None)['rows']


def pool_first(agent_name=None, team_id=None, limit=100):
    """Wave-7 W1 (C1 as amended): the cold-load FAST SLICE — the pool's first `limit` rows in
    its own default order + the true total, built from the CHEAP families only (revenue ×3,
    cadence, status, est_missed, partner attributes ≈ 12s measured). The AR / product-mix /
    salesperson / coords columns stay BLANK until the full build swaps in: `_mix_attrs` alone
    is a whole-book 20s read_group whose domain must never be pid-narrowed (502 — see its
    docstring), so "all columns on 100 rows" measured 44.6s vs 45.1s full — no win. A partial
    payload is a UI PHASE, never a reporting basis — validate() reconciles the full pool only."""
    return _pool_build(agent_name, team_id, limit=limit, fast=True)


# ------------------------------------------------------------------ the address reconciliation
#: {our pool key: the `res.partner` field an Odoo domain filters on}. The two m2o columns are
#: stored here as RESOLVED NAMES (`O.m2o_name`), so the oracle asks Odoo whether the id is SET
#: while we count whether the name is non-blank — a set id resolving to an empty name is the one
#: place those two questions could come apart, and the per-key leg is what would report it.
ADDRESS_FIELDS = {
    'street': 'street', 'street2': 'street2', 'city': 'city',
    'state': 'state_id', 'country': 'country_id', 'zip': 'zip',
}

#: OUR blank. `_partner_attrs` collapses every partner attribute to '(none)' when blank (MECE, so
#: a group-by has no null bucket), a pool row for a partner with NO attrs takes the same sentinel
#: from the row template, and `_reconcile_ledger`'s revived rows are written '(none)' outright.
_ADDRESS_BLANK = ('(none)', '', None)


def address_blank(value):
    """OUR blank predicate, in ONE place — the fold and the oracle must ask ONE question.

    ⚠ `.strip()` is part of it, and that is the half that diverges from Odoo. A domain
    `('street', '!=', False)` counts a whitespace-only street as PRESENT while this counts it as
    blank, so the two normalizers would disagree by exactly the whitespace-only population.
    MEASURED 2026-08-12 across `street`/`street2`/`city`/`zip` over the whole pool: **0**. The
    leg that keeps it 0 is asserted in `validate()` rather than assumed
    ([[one-question-two-normalizers]]).
    """
    return value in _ADDRESS_BLANK or (isinstance(value, str) and not value.strip())


def validate(team_id=None):
    """Reconcile the ADDRESS family the Customer grid ships to an independent Odoo aggregate.

    ⛔ THE POINT OF THIS FUNCTION IS THAT THE NUMBER SURVIVES THE SESSION THAT TOOK IT. W30-T33
    shipped `street`/`street2` after measuring their coverage with a throwaway shell one-liner —
    a measurement, never a `validate()` — so nothing in the repo re-checked it and the figure
    died with that session. The house law is *every metric gets a `validate()` reconciling to an
    INDEPENDENT Odoo aggregate*, and an oracle nobody re-runs rots exactly like a mode nobody
    runs ([[rules-need-gates]]). `aios-web/api/verify_odoo_relational.py` is what re-runs it.

    ⚠ THE TRAP THAT REDS A CORRECT LEG, and it is why every domain below is scoped to the pool's
    OWN pid set: **the customer pool is the WHOLE BOOK**, not `customer_rank > 0 AND active`
    ([[customer-grid-pool-vs-odoo-count]]). The 3,617/3,478 pair quoted in the wave-30 mailbox is
    the rank>0-active population and is NOT this one; measured here the same day, the pool holds
    3,629 rows of which 3,487 carry a street. Comparing our count against a differently-scoped
    Odoo count is a true measurement of the wrong subject [[measure-the-real-call]].

    ⛔ IT BUILDS THE FULL `pool()`, NOT `pool_first`/`fast=True`, AND THE COST IS REAL — **444 s
    measured 2026-08-12** (the address itself is ~14 s of that; AR, product mix, salesperson,
    coords and DBA are the rest). Reported rather than optimised away, per R6's second sentence.
    `fast=True` is not the shipping basis: it skips `_reconcile_ledger`, so the RETAINED customers
    (I13) never enter and the population is a different one. The recommended fix if this ever has
    to get cheap is a `families=` argument on `_pool_build` so a reconciliation can ask for the
    families it reconciles — NOT a second cheaper pool builder, which would be two evaluators for
    one question [[one-evaluator-per-question]].
    """
    rows = pool(team_id=team_id)
    pids = [int(r['pid']) for r in rows if r.get('pid')]
    odoo = O.get_odoo()
    checks = []

    # ── the population, and the three numbers that keep the address legs honest ───────────────
    # An Odoo `search` excludes archived records by default, so `n_visible` counts the partners
    # that are BOTH live and ours. Three populations therefore ride the pool and only one of them
    # can carry an address: rows Odoo can see, rows `_reconcile_ledger` REVIVED (I13 — Odoo
    # dropped them; every attribute is written '(none)'), and rows that are neither, i.e. archived
    # in Odoo but never retained. The third is the one that grows silently, so it is REPORTED as
    # a number rather than absorbed into a tolerance.
    revived = [r for r in rows if r.get('odoo_status') != 'Active']
    rpids = [int(r['pid']) for r in revived if r.get('pid')]
    n_visible = odoo.search_count('res.partner', [('id', 'in', pids)])
    # ⛔ ASSERTED, NOT ASSUMED. "The retained rows are the ones Odoo lost" is trivially true when
    # checked against the ledger that did the retaining — that is the self-sealing shape. So it is
    # asked of ODOO: a retained pid Odoo can still see means `_reconcile_ledger` buried a LIVE
    # customer's address behind six '(none)' cells.
    still_live = odoo.search_count('res.partner', [('id', 'in', rpids)]) if rpids else 0
    checks.append({
        'check': "every RETAINED customer (I13) is one Odoo genuinely cannot see — asked of "
                 "Odoo, never of the ledger that did the retaining",
        'ours': still_live, 'theirs': 0, 'ok': still_live == 0,
        'detail': {'pool_rows': len(rows), 'odoo_visible': n_visible, 'retained': len(revived),
                   'archived_not_retained': len(pids) - n_visible - len(revived)},
    })

    # ── the six address columns, each against a domain Odoo evaluates server-side ─────────────
    # ⛔ `search_count` WITH A DOMAIN, never a re-read-and-recount: it makes ODOO do the filtering,
    # so the oracle cannot inherit a bug from `_partner_attrs`, which is the code under test.
    for key, field in ADDRESS_FIELDS.items():
        ours = sum(1 for r in rows if not address_blank(r.get(key)))
        theirs = odoo.search_count('res.partner', [('id', 'in', pids), (field, '!=', False)])
        checks.append({
            'check': f"the grid's `{key}` column is non-blank on exactly the pool customers Odoo "
                     f"says carry `{field}` — an INDEPENDENT search_count over the pool's own "
                     f"pid set, never a count of the rows we just built",
            'ours': ours, 'theirs': theirs, 'ok': ours == theirs,
            # `column` is what the GATE matches on. Substring-matching the prose is how a leg
            # silently stops being checked when somebody rewords it, and this file's own twin
            # (`verify_product_pool`) matches on prose precisely because it had no key to use.
            'detail': {'column': key, 'odoo_field': field, 'blank_here': len(rows) - ours},
        })

    # ── the two normalizers are ONE question, and this is the leg that keeps them one ─────────
    # Everything above compares OUR `.strip() or '(none)'` against ODOO'S `!= False`. Those agree
    # on every value except a whitespace-only one, which Odoo reports present and this module
    # renders blank. Measured 0 today; if it ever stops being 0 the per-key legs above go red for
    # a reason that has nothing to do with the address being wrong, so the cause is named HERE.
    text_keys = [k for k, f in ADDRESS_FIELDS.items() if f == k]
    ws = {}
    if pids:
        # `O.search_read`, the same door `_partner_attrs` reads through — the client's own method
        # takes `domain=`/`fields=` as keywords and this wrapper is the one with a fixed shape.
        for got in O.search_read('res.partner', [('id', 'in', pids)], text_keys):
            for k in text_keys:
                v = got.get(k)
                if isinstance(v, str) and v and not v.strip():
                    ws[k] = ws.get(k, 0) + 1
    checks.append({
        'check': "no pool customer carries a WHITESPACE-ONLY address value — the one input on "
                 "which our blank predicate and Odoo's `!= False` provably disagree",
        'ours': sum(ws.values()), 'theirs': 0, 'ok': not ws,
        'detail': {'per_field': ws, 'fields_checked': text_keys},
    })
    return checks


def _dba_attrs(pids, t):
    """{pid: 'Fisch' | 'Royal' | 'Both'} — which brand(s) a customer's confirmed orders carry
    over the pool's own 24-month universe (C-DBA, wave 2026-08-02).

    GIFTWARE DEALS (the Amazon channel) is deliberately NOT a DBA — a customer buying only
    through it stays blank here, exactly as ARCHITECTURE.md §4's wholesale scope treats that
    team. Blank also covers the ledger-retained customers whose activity predates the window:
    blank means "no brand attributable in 24 months", never a guess. One read_group, grouped
    (partner, team); degrades to {} like the other attr families — a blank column beats a
    page that cannot render.
    """
    try:
        import datetime as _dt
        d_from = (_dt.date.fromisoformat(str(t)) - _dt.timedelta(days=731)).isoformat()
        pairs = O.read_group(
            'sale.order',
            [('state', 'in', ('sale', 'done')), ('date_order', '>=', d_from),
             ('partner_id', 'in', list(pids)), ('team_id', 'in', (5, 6))],
            ['partner_id'], ['partner_id', 'team_id'], lazy=False)
        teams = {}
        for r in pairs:
            pid = O.m2o_id(r.get('partner_id'))
            tid = O.m2o_id(r.get('team_id'))
            if pid and tid in (5, 6):
                teams.setdefault(pid, set()).add(tid)
        label = {5: 'Fisch', 6: 'Royal'}
        return {pid: ('Both' if len(ts) == 2 else label[next(iter(ts))])
                for pid, ts in teams.items()}
    except Exception:
        return {}


def _book_pids(team_id=None, agent_pids=None):
    """⭐ WAVE 20 (owner items 8 + 27, ruling R3) — THE WHOLE BOOK, not just who bought recently.

    Owner: *"Martin Pasternak showing only 313 accounts when I filter agent, whereas in Odoo it is
    494 … I want our App to show complete source of truth, even if no sale.order at all or not.
    Its important to see what customer is getting assigned to Martin or whether we can retarget
    them."* The pool was a 24-MONTH SALES universe (YTD ∪ LY-YTD ∪ cadence-window buyers), so a
    customer assigned to an agent who has never ordered — precisely a retargeting target — did not
    exist in the app at all.

    R3's definition, MEASURED against live Odoo 2026-08-05:
        customer_rank>0 active .............. 3,614
        ever ordered (confirmed, team 5/6) ...  1,748
        UNION ............................... 3,723   (vs 1,555 in the old 24-month pool)
    Martin's book under this rule is **494** — the owner's Odoo number, to the account.

    ⛔ **THE "agent-assigned" LEG OF R3 IS DELIBERATELY NOT A THIRD TERM, and that is a
    measurement, not a shortcut.** Taken literally it added 174 partners and pushed Martin to 503;
    every one of the 9 extras on his book was an ODOO ADDRESS RECORD rather than an account —
    `type` in (`delivery`, `other`), most carrying a `parent_id`, and TWO with `name: False`.
    Shipping them would have put nameless rows in the customer table and made "how many customers
    does Martin have" answer 503 against an Odoo screen that says 494.
    Excluding address types wholesale is also wrong (`type not in (delivery, invoice, other,
    private)` measured **488** — it drops 6 genuine accounts that happen to carry a delivery
    type). `customer_rank > 0` is the predicate that means "this partner is a customer record",
    it reproduces the owner's number exactly, and every agent-assigned ACCOUNT already satisfies
    it — so the leg is subsumed rather than dropped. Item 27's actual ask (the 126 assigned
    partners with no `sale.order` at all) is fully served: they are rank>0 and they are in.

    ⚠ ACTIVE ONLY. Archived partners stay out (R3): they are ex-customers, and putting them in
    every count would make "how many customers do we have" unanswerable. The agent-LOGIN scope
    (`customers.agent_partner_ids`) deliberately still includes archived — an agent's own book is
    their whole history — and the two remain compatible because that set is INTERSECTED with this
    pool, so archived rows drop out of the table without narrowing the agent's own permissions.

    ⚠ BU SCOPE IS APPLIED THROUGH ORDERS, NOT THROUGH THE PARTNER. `res.partner` carries no team,
    so a BU-scoped caller gets the partners who have ORDERED in that BU (plus their own agent
    book). Widening it to every partner for a scoped user would cross the BU isolation rule that
    the whole permissioning model rests on.
    """
    try:
        if team_id:
            # Scoped: partners with confirmed orders in THIS BU, ever. `read_group` on partner_id
            # rather than a search_read of orders — the group is the distinct set, and it is one
            # round trip instead of paging tens of thousands of order rows.
            rows = O.read_group('sale.order',
                                [('state', 'in', ('sale', 'done')), ('team_id', '=', team_id)],
                                ['partner_id'], ['partner_id'], lazy=False)
            book = {O.m2o_id(r.get('partner_id')) for r in rows}
            book.discard(None)
        else:
            rank = O.search_read('res.partner',
                                 [('customer_rank', '>', 0), ('active', '=', True)],
                                 ['id'], limit=200000)
            # `ever` is NOT redundant with rank>0: a partner can be archived-then-reactivated, or
            # have had its rank reset, and an account that demonstrably bought from us belongs in
            # the book whatever its flags say now. An ORDER is a fact; a rank is a setting.
            rows = O.read_group('sale.order',
                                [('state', 'in', ('sale', 'done')), ('team_id', 'in', (5, 6))],
                                ['partner_id'], ['partner_id'], lazy=False)
            ever = {O.m2o_id(r.get('partner_id')) for r in rows}
            ever.discard(None)
            book = {r['id'] for r in rank} | ever
        # An agent filter NARROWS the book to that agent's partners — never widens it. The
        # intersection is what keeps an agent-login user inside their own book while still
        # gaining every no-order account assigned to them, which is the whole point of item 27.
        return (book & set(agent_pids)) if agent_pids is not None else book
    except Exception as e:
        # Degrade to the sales-derived universe rather than failing the page — but LOUDLY.
        #
        # ⛔ THIS DEGRADATION IS NOT LIKE THE OTHERS IN THIS MODULE, and the difference is what
        # makes silence wrong here. Every other family (`_ar_attrs`, `_mix_attrs`, …) degrades to
        # a blank COLUMN, which is visible: the user sees an empty column and asks why. This one
        # degrades to absent ROWS. The pool still builds from the sales legs, so the page renders
        # perfectly, the counts look plausible, and Martin is quietly back at 313 with nothing
        # anywhere saying the book leg failed. An invisible degradation of a COUNT is the failure
        # mode this codebase keeps writing rules against ([[no-unverifiable-aggregates]]).
        try:
            import harness.telemetry as _tel
            _tel.error('customer_data:_book_pids', e, fallback='sales-derived pool only')
        except Exception:
            pass
        print(f"[aios] WARNING customer_data._book_pids failed ({type(e).__name__}: {e}) - the "
              f"customer pool is falling back to the 24-month SALES universe, so accounts with "
              f"no recent orders (owner item 27) are MISSING from this build.")
        return set()


def _pool_build(agent_name, team_id, limit, fast=False):
    t = P.today()
    yf, yt = P.ytd(t)
    lf, lt = P.ytd_last_year(t)
    mf, mt = P.ltm(t)
    pids = agent_pids(agent_name)
    this = cust._cust_rev(yf, yt, team_id, pids)
    last = cust._cust_rev(lf, lt, team_id, pids)
    ltm = cust._cust_rev(mf, mt, team_id, pids)
    cad = cust._cadence_bulk(t, team_id, agent_pids=pids)
    all_pids = set(this) | set(last) | set(cad) | _book_pids(team_id, pids)
    total = len(all_pids)
    if limit is not None and total > limit:
        # the SAME ordering the assembled pool ships (see the sort below) — the slice must be
        # the first page of the very list the full build renders, or the swap-in reshuffles.
        ranked = sorted(all_pids,
                        key=lambda p: -((this.get(p) or {}).get('rev', 0.0)
                                        + (last.get(p) or {}).get('rev', 0.0)))
        all_pids = set(ranked[:limit])
    attrs = cust._partner_attrs(list(all_pids))
    if fast:
        # C1 fast phase: the slow families stay empty — their columns render blank behind the
        # toolbar's partial indicator until the full build swaps in.
        ar_a, mix_a, sp_a, geo_a, dba_a = {}, {}, {}, {}, {}
    else:
        # The three families the owner asked for (2026-07-27). Each degrades to {} rather than
        # raising: a customer table that cannot render because AR is momentarily unreachable is
        # a worse failure than one with a blank column.
        ar_a = _ar_attrs(t)
        mix_a = _mix_attrs(all_pids, t)
        sp_a = _salesperson_attrs(all_pids, t)
        # W11 (wave-7): coordinates for the Map VIEW — Odoo coords else the geocode cache,
        # read-only reuse (coords_for never geocodes); missing → null lat/lon, and the
        # client's map view discloses the count.
        geo_a = map_mod.coords_for(list(all_pids))
        # C-DBA (2026-08-02): the brand attribute rides the full build only, like the rest.
        dba_a = _dba_attrs(all_pids, t)
    # customers whose only activity sits in the older half of the 24-month cadence window
    # appear in no revenue map — fetch their names directly (else the lapsed cohort, exactly
    # the win-back targets this union exists for, would render as '?')
    unnamed = [pid for pid in all_pids
               if not ((this.get(pid) or {}).get('name') or (last.get(pid) or {}).get('name')
                       or (ltm.get(pid) or {}).get('name'))]
    extra_names = ({r['id']: r['name'] for r in
                    O.search_read('res.partner', [('id', 'in', unnamed)], ['name'])}
                   if unnamed else {})
    rows = []
    for pid in all_pids:
        tv = this.get(pid) or {}
        lv = last.get(pid) or {}
        c = cad.get(pid) or {}
        rev, ly = tv.get('rev', 0.0), lv.get('rev', 0.0)
        gap = c.get('typical_gap_days')
        overdue = c.get('overdue_days')
        aov = c.get('aov', 0.0)
        est = (min(overdue / gap, 3.0) * aov) if (gap and overdue and overdue > 0) else 0.0
        status = ('New' if rev > 0 and ly <= 0 else
                  'Lost' if ly > 0 and rev <= 0 else
                  'Declining' if 0 < rev < ly else
                  'Growing' if rev > 0 else 'Dormant')
        a = attrs.get(pid) or {}
        rows.append({
            'pid': pid,
            'customer': tv.get('name') or lv.get('name')
                        or (ltm.get(pid) or {}).get('name') or extra_names.get(pid) or '?',
            'status': status,
            'agent': a.get('agent', '(none)'),
            # ⭐ W30-T33 (carried W29-T51) — the street lines beside the city they belong to.
            'street': a.get('street', '(none)'), 'street2': a.get('street2', '(none)'),
            'city': a.get('city', '(none)'),
            # C-DBA: blank (not '(none)') — a select's blank is its own honest empty state.
            'dba': dba_a.get(pid, ''),
            'state': a.get('state', '(none)'),
            # Owner 2026-07-27, the partner-attribute family. `.get(..., default)` is not
            # defensive habit here: search_read returned 1,548 of 1,550 pool pids — two partners
            # in the sales history are archived or deleted — so a customer with NO attrs row is
            # a real case, and it must render as '(none)' rather than KeyError the whole page.
            'country': a.get('country', '(none)'), 'zip': a.get('zip', '(none)'),
            'payment_terms': a.get('payment_terms', '(none)'),
            'customer_since': a.get('customer_since', ''),
            # Row-level datum for the `created_time` field type (wave-5 item 11) — rides every
            # row like pid, no column of its own until a user creates one.
            '_created': a.get('created_at', ''),
            'tags': a.get('tags', '(none)'), 'pricelist': a.get('pricelist', '(none)'),
            # AR / credit exposure — composed from the already-reconciled ar.py blocks, so the
            # Customer table and the Collections page cannot disagree.
            'ar_open': (ar_a.get(pid) or {}).get('ar_open', 0.0),
            'ar_overdue': (ar_a.get(pid) or {}).get('ar_overdue', 0.0),
            'ar_outstanding': (ar_a.get(pid) or {}).get('ar_outstanding', 0.0),
            'ar_exposure': (ar_a.get(pid) or {}).get('ar_exposure', 0.0),
            'days_to_pay': (ar_a.get(pid) or {}).get('days_to_pay'),
            **{k: (ar_a.get(pid) or {}).get(k, 0.0) for k in AR_BUCKET_FIELDS.values()},
            # Product mix, LTM, line grain, ALL-channel.
            'top_category': (mix_a.get(pid) or {}).get('top_category', '(none)'),
            'top_category_pct': (mix_a.get(pid) or {}).get('top_category_pct'),
            'sku_count': (mix_a.get(pid) or {}).get('sku_count', 0),
            'top_sku': (mix_a.get(pid) or {}).get('top_sku', '(none)'),
            # The ORDER-TAKER, not the agent. Different people, different questions.
            'salesperson': sp_a.get(pid, '(none)'),
            # W11: nullable — the Map view pins what it can and counts what it cannot.
            'lat': (geo_a.get(pid) or {}).get('lat'),
            'lon': (geo_a.get(pid) or {}).get('lon'),
            'revenue_ytd': rev, 'revenue_ly': ly,
            'yoy_pct': P.yoy_pct(rev, ly),
            'at_risk': max(ly - rev, 0.0),
            'ltm_rev': (ltm.get(pid) or {}).get('rev', 0.0),
            'orders_24m': c.get('n_orders', 0), 'aov': aov,
            'last_order': c.get('last_order', ''),
            'days_since': c.get('days_since'),
            'typical_gap_days': gap, 'overdue_days': overdue,
            'est_missed': est,
        })
    for r in rows:
        r['odoo_status'] = 'Active'
    # I13 (owner item 13, contract C6): RETAIN customers that disappear from Odoo.
    # Only the FULL build reconciles the ledger — the fast slice is 100 rows by construction,
    # so letting it write would mark ~1,450 living customers archived on every cold load.
    if not fast:
        rows, total = _reconcile_ledger(rows, total, agent_name, team_id)
    rows.sort(key=lambda r: -(r['revenue_ytd'] + r['revenue_ly']))
    return {'rows': rows, 'total': total}


LEDGER_KEY = 'customer_ledger'


def _reconcile_ledger(rows, total, agent_name, team_id):
    """Remember every customer we have ever seen, and keep serving the ones Odoo stopped
    returning (owner item 13: "if it is deleted, that it goes to an archived tag immediately —
    so stored in our platform even if deleted in Odoo").

    Why a ledger at all: this pool is built from SALES HISTORY, not from a partner list, so a
    partner deleted in Odoo usually keeps appearing (their order lines remain) — until the day
    the lines go too, when the row would silently vanish along with every note, tag and cohort
    membership attached to it. Nothing else in the app would notice. The ledger is the only
    record that the customer ever existed, which is why it starts recording now rather than
    when the surfacing is finished: history you did not write down is not recoverable later.

    ⚠ SCOPE-SAFE. The ledger is written ONLY from an UNSCOPED build (no agent, no BU). A scoped
    build legitimately sees a fraction of the book, and reconciling from it would mark every
    customer outside that scope 'Archived' for everyone — the failure would look exactly like
    the data loss this is meant to prevent. Scoped builds READ the ledger and resurrect nothing
    they cannot prove they should see.
    """
    try:
        ledger = dict(store.get(LEDGER_KEY) or {})
    except Exception:
        return rows, total                  # a ledger we cannot read must not break the table
    today = str(P.today())
    live = {r['pid'] for r in rows}
    unscoped = agent_name is None and team_id is None

    if unscoped:
        for r in rows:
            e = ledger.get(str(r['pid'])) or {}
            e.update({'name': r['customer'], 'last_seen': today})
            e.setdefault('first_seen', today)
            ledger[str(r['pid'])] = e

    # Rows the ledger knows and this build did not return. On an unscoped build that means
    # "gone from Odoo"; on a scoped one it usually just means "not in this agent's book", so
    # only the unscoped build may surface them.
    revived = []
    if unscoped:
        for key, e in ledger.items():
            try:
                pid = int(key)
            except (TypeError, ValueError):
                continue
            if pid in live:
                continue
            revived.append({
                'pid': pid, 'customer': e.get('name') or '?', 'status': 'Dormant',
                'odoo_status': 'Archived',
                # Everything else is genuinely UNKNOWN now — the record is gone. Blank is the
                # honest rendering; carrying the last-known numbers forward would state figures
                # as current that nothing can reconcile ([[no-unverifiable-aggregates]]).
                # ⚠ W30-T33: the address keys land HERE TOO, and this is the site a field addition
                # silently skips. `rows_from_pool` projects with `r.get(k)`, so a revived row
                # missing a key renders BLANK rather than raising — archived customers would show
                # an empty address column while live ones showed a full one, and nothing anywhere
                # would report it. Two row templates, one field contract.
                'agent': '(none)', 'street': '(none)', 'street2': '(none)',
                'city': '(none)', 'state': '(none)', 'country': '(none)',
                'zip': '(none)', 'payment_terms': '(none)', 'customer_since': '',
                '_created': '', 'tags': '(none)', 'pricelist': '(none)',
                'ar_open': 0.0, 'ar_overdue': 0.0, 'ar_outstanding': 0.0, 'ar_exposure': 0.0,
                'days_to_pay': None,
                **{k: 0.0 for k in AR_BUCKET_FIELDS.values()},
                'top_category': '(none)', 'top_category_pct': None, 'sku_count': 0,
                'top_sku': '(none)', 'salesperson': '(none)', 'lat': None, 'lon': None,
                'revenue_ytd': 0.0, 'revenue_ly': 0.0, 'yoy_pct': None, 'at_risk': 0.0,
                'ltm_rev': 0.0, 'orders_24m': 0, 'aov': 0.0, 'last_order': '',
                'days_since': None, 'typical_gap_days': None, 'overdue_days': None,
                'est_missed': 0.0,
            })
        if store.available():
            try:
                store.update(LEDGER_KEY, lambda cur: {**cur, **ledger}, flush='async')
            except Exception:
                pass                        # recording is best-effort; the table is not
    return rows + revived, total + len(revived)


# ------------------------------------------------------------------ the rule engine
# A list's formula = [{'field','op','value'}, ...] over FIELDS + a 'sort' key ('-x' = desc).
# The two templates ARE the old My Day queue formulas, expressed as visible rules.
#
# 'note' is a SEED, not the source of truth (owner 2026-07-25). It is what a list reads
# BEFORE anyone edits it; the moment a user saves a description the stored one wins, and
# an empty stored description stays empty — the template must never resurrect. That
# precedence lives in aios_grid.views_from_defs (a saved view replaces its template-derived
# one wholesale) and is pinned by _qa_grid_view_note.py. Edit the prose here only to change
# what a NEW user starts with.
TEMPLATES = {
    'Call list': {
        'rules': [{'field': 'orders_24m', 'op': '>=', 'value': 3},
                  {'field': 'overdue_days', 'op': '>', 'value': 0},
                  {'field': 'days_since', 'op': '<=', 'value': 365}],
        'sort': '-est_missed',
        'note': 'Customers overdue against their OWN reorder cadence (3+ orders, a real gap '
                'pattern, ordered within 365d), ranked by estimated missed revenue.'},
    'Win-back': {
        'rules': [{'field': 'revenue_ly', 'op': '>=', 'value': 2000},
                  {'field': 'at_risk', 'op': '>', 'value': 0}],
        'sort': '-at_risk',
        'note': 'Bought materially last year (over $2,000), down or gone this year, ranked by '
                'dollars at risk.'},
}


def _match(row, rule):
    """One rule against one row. None/missing numeric values fail every numeric comparison
    (a customer with no cadence is never 'overdue'); text ops are case-insensitive."""
    field, op, val = rule.get('field'), rule.get('op'), rule.get('value')
    if field not in FIELDS:
        return True                                   # unknown field → rule is inert, not a crash
    v = row.get(field)
    if FIELDS[field][1] == 'text':
        s, q = str(v or '').lower(), str(val or '').lower()
        return q in s if op == 'contains' else s == q
    try:
        v, val = float(v), float(val)
    except (TypeError, ValueError):
        return False
    return {'>=': v >= val, '<=': v <= val, '>': v > val, '<': v < val,
            '=': v == val}.get(op, False)


def apply_rules(rows, rules, sort=None):
    """Filter the pool through a formula (AND of all rules) and apply the list's sort."""
    out = [r for r in rows if all(_match(r, ru) for ru in (rules or []))]
    if sort:
        key = sort.lstrip('-')
        out.sort(key=lambda r: (r.get(key) is None,
                                -(r.get(key) or 0) if sort.startswith('-') else (r.get(key) or 0))
                 if FIELDS.get(key, ('', 'num'))[1] == 'num'
                 else str(r.get(key) or '').lower(), reverse=False)
    return out


# ------------------------------------------------------------------ persistence (HF store)
def saved_lists(username):
    """{list_name: {'rules','sort','members','note'}} for one user ({} when none / store down)."""
    try:
        return (store.get(KEY) or {}).get(username, {}) or {}
    except Exception:
        return {}


def save_list(username, name, definition):
    def _up(d):
        d.setdefault(username, {})[name] = definition
        return d
    store.update(KEY, _up)


def delete_list(username, name):
    def _up(d):
        (d.get(username) or {}).pop(name, None)
        return d
    store.update(KEY, _up)


# ------------------------------------------------------------------ Airtable-style table workspace
# This is intentionally separate from `customer_lists`: a LIST owns membership/formula
# semantics, while a VIEW owns presentation/query state (filters, multi-sort, grouping,
# visible fields, widths, color, row height). List views can reference the same formula
# without conflating the two persistence contracts.
#
# The store logic MOVED to core/table_store.py 2026-07-27 (the table-page factory): these
# names are the Customer table's instance of the generic per-object workspace store, kept as
# module functions so every existing caller (app.py's host loop, QA gates) is untouched.
# A NEW table object gets its own `core.table_store.make('<its>_table_workspace')`.
_TSTORE = table_store.make(TABLE_KEY)

#: Wave 16 C-TOPIC — the ops OBJECT, for callers that select a table by TOPIC (grid_events'
#: `EventCtx.table`). The module-level re-exports below stay for every existing caller.
TABLE_OPS = _TSTORE

table_workspace = _TSTORE.workspace
shared_table_views = _TSTORE.shared_views      # wave-9 I17: views shared WITH this viewer
shared_table_view = _TSTORE.shared_view        #   ...and one raw, for authorisation only
save_table_view = _TSTORE.save_view
delete_table_view = _TSTORE.delete_view
save_table_field = _TSTORE.save_field
delete_table_field = _TSTORE.delete_field
duplicate_table_field = _TSTORE.duplicate_field
patch_table_overlay = _TSTORE.patch_overlay
save_table_folders = _TSTORE.save_folders
save_table_active_view = _TSTORE.save_active_view   # owner item 3: last-opened view, per user
save_table_record_layout = _TSTORE.save_record_layout  # 2026-08-02 C-LAYOUT: record-detail order


# ------------------------------------------------------------------ digest compatibility
def queues(agent_name=None, team_id=None, limit=200):
    """The two canonical morning lists (call list + win-back) for the daily digest email —
    unchanged formulas via the customers module (the digest always uses the CANONICAL templates,
    not a user's edited copy, so every rep's email means the same thing)."""
    pids = agent_pids(agent_name)
    return {
        'calls': cust.contact_recommendations(team_id=team_id, limit=limit, agent_pids=pids),
        'risk': cust.at_risk(team_id=team_id, limit=limit, agent_pids=pids),
    }