| """Price compliance — selling below the customer's pricelist tier (pocket-price floor). |
| |
| Discount FIELDS are unused in this Odoo (verified dead), but reps can still key a low |
| price_unit directly — leakage that is invisible today. Per (customer × SKU) over the LTM: |
| realized unit price (Σ subtotal ÷ Σ qty — discount-true) vs the customer's applicable |
| pricelist tier (fixed-price rules, qty break matched on the average order quantity). |
| Exceptions below THRESHOLD of tier form the worklist, each valued at its annualized leak |
| (= (tier − realized) × LTM qty). Pricing canon: ~1% realization ≈ 8-9% operating profit. |
| |
| v1 simplifications (documented, deliberate): |
| - qty break matched on AVG order qty per (customer, SKU) — not per order line; |
| - fixed-price rules only (this instance prices via fixed tiers, like the legacy SKU app); |
| - LTM window, wholesale scope. |
| """ |
| import datetime as dt |
|
|
| import core.odoo as O |
| import core.periods as P |
| import modules.customers as cust |
|
|
| THRESHOLD = 0.97 |
|
|
|
|
| def _rules(): |
| """All date-valid fixed-price pricelist rules, indexed by (pricelist, product-variant) |
| and (pricelist, product-template). Highest min_quantity ≤ qty wins at match time.""" |
| today = P.today().isoformat() |
| rows = O.search_read( |
| 'product.pricelist.item', |
| [('compute_price', '=', 'fixed'), |
| ('applied_on', 'in', ['0_product_variant', '1_product'])], |
| ['pricelist_id', 'product_id', 'product_tmpl_id', 'min_quantity', |
| 'fixed_price', 'date_start', 'date_end', 'applied_on']) |
| ok = [] |
| for r in rows: |
| ds = str(r.get('date_start') or '')[:10] |
| de = str(r.get('date_end') or '')[:10] |
| if (ds and ds > today) or (de and de < today): |
| continue |
| ok.append(r) |
| by_var, by_tmpl = {}, {} |
| for r in ok: |
| pl = O.m2o_id(r['pricelist_id']) |
| if r['applied_on'] == '0_product_variant' and r.get('product_id'): |
| by_var.setdefault((pl, O.m2o_id(r['product_id'])), []).append(r) |
| elif r.get('product_tmpl_id'): |
| by_tmpl.setdefault((pl, O.m2o_id(r['product_tmpl_id'])), []).append(r) |
| return ok, by_var, by_tmpl |
|
|
|
|
| def _tier_for(pl, pid, tmpl, qty, by_var, by_tmpl): |
| """The applicable fixed tier price: variant rules beat template rules; within a scope, |
| the highest qty break ≤ qty wins.""" |
| for cands in (by_var.get((pl, pid)), by_tmpl.get((pl, tmpl))): |
| if not cands: |
| continue |
| eligible = [r for r in cands if (r.get('min_quantity') or 0) <= max(qty, 1)] |
| if eligible: |
| best = max(eligible, key=lambda r: r.get('min_quantity') or 0) |
| return best.get('fixed_price') or None |
| return None |
|
|
|
|
| def matched_pairs(team_id=None, t=None): |
| """The shared substrate: every (customer × SKU) pair bought in the LTM with its applicable |
| fixed tier attached (tier=None when the customer has no pricelist / no rule matches). |
| Used by the compliance sweep here AND by the price corridor + pocket waterfall |
| (modules/reprice.py) — one definition of 'realized vs tier' everywhere.""" |
| t = t or P.today() |
| lf, lt = P.ltm(t) |
|
|
| |
| g = O.read_group('sale.order.line', O.sale_line_domain(lf, lt, team_id), |
| ['price_subtotal:sum', 'product_uom_qty:sum'], |
| ['order_partner_id', 'product_id'], lazy=False) |
| pairs = [] |
| for r in g: |
| pid = O.m2o_id(r.get('order_partner_id')) |
| prod = O.m2o_id(r.get('product_id')) |
| qty = r.get('product_uom_qty') or 0.0 |
| rev = r.get('price_subtotal') or 0.0 |
| n = r.get('__count') or 1 |
| if pid and prod and qty > 0 and rev > 0: |
| pairs.append({'pid': pid, 'customer': O.m2o_name(r.get('order_partner_id')), |
| 'prod': prod, 'product': O.m2o_name(r.get('product_id')), |
| 'qty': qty, 'rev': rev, 'lines': n, |
| 'unit': rev / qty, 'avg_order_qty': qty / n}) |
|
|
| |
| pids = list({p['pid'] for p in pairs}) |
| plists = {} |
| for i in range(0, len(pids), 2000): |
| for r in O.search_read('res.partner', [('id', 'in', pids[i:i + 2000])], |
| ['property_product_pricelist']): |
| plists[r['id']] = O.m2o_id(r.get('property_product_pricelist')) |
| prods = list({p['prod'] for p in pairs}) |
| pinfo = {} |
| for i in range(0, len(prods), 2000): |
| for r in O.search_read('product.product', |
| [('id', 'in', prods[i:i + 2000]), |
| ('active', 'in', [True, False])], |
| ['product_tmpl_id', 'default_code']): |
| pinfo[r['id']] = {'tmpl': O.m2o_id(r.get('product_tmpl_id')), |
| 'code': (r.get('default_code') or '').strip()} |
|
|
| all_rules, by_var, by_tmpl = _rules() |
| for p in pairs: |
| info = pinfo.get(p['prod'], {}) |
| p['sku'] = info.get('code', '') |
| pl = plists.get(p['pid']) |
| tier = (_tier_for(pl, p['prod'], info.get('tmpl'), p['avg_order_qty'], |
| by_var, by_tmpl) if pl else None) |
| p['tier'] = tier if (tier and tier > 0) else None |
| return {'pairs': pairs, 'checked': len(pairs), 'n_rules': len(all_rules), |
| 'window': (lf, lt)} |
|
|
|
|
| def build(team_id=None, t=None, pre=None): |
| """The compliance sweep. Returns dict(rows[exceptions], checked, matched, kpis…).""" |
| mp = pre or matched_pairs(team_id, t) |
| matched_list = [p for p in mp['pairs'] if p.get('tier')] |
| exceptions = [p for p in matched_list if p['unit'] / p['tier'] < THRESHOLD] |
| attrs = cust._partner_attrs(list({p['pid'] for p in exceptions})) |
| rows = [] |
| for p in exceptions: |
| tier = p['tier'] |
| rows.append({ |
| 'pid': p['pid'], 'customer': p['customer'], |
| 'sku': p['sku'], 'product': p['product'], |
| 'agent': (attrs.get(p['pid']) or {}).get('agent', '(none)'), |
| 'tier_price': tier, 'realized': p['unit'], |
| 'pct_of_tier': p['unit'] / tier * 100.0, |
| 'qty_ltm': p['qty'], 'rev_ltm': p['rev'], |
| 'leak': (tier - p['unit']) * p['qty'], 'lines': p['lines'], |
| }) |
| rows.sort(key=lambda x: -x['leak']) |
| return { |
| 'rows': rows, 'checked': mp['checked'], 'matched': len(matched_list), |
| 'n_rules': mp['n_rules'], |
| 'leak_total': sum(r['leak'] for r in rows), |
| 'exception_rate': (len(rows) / len(matched_list) * 100.0) if matched_list else 0.0, |
| 'window': mp['window'], |
| } |
|
|
|
|
| def validate(t=None, team_id=None): |
| """(1) rule pull complete vs search_count; (2) the TOP exception re-derived row-level: |
| its realized unit price recomputed from its raw order lines (independent path) must match |
| the read_group aggregate the sweep used.""" |
| b = build(team_id, t) |
| n = O.get_odoo().search_count( |
| 'product.pricelist.item', |
| [('compute_price', '=', 'fixed'), |
| ('applied_on', 'in', ['0_product_variant', '1_product'])]) |
| checks = [{'check': 'fixed pricelist rules — pull complete (pre date-filter)', |
| 'a': b['n_rules'], 'b': f'<= {n}', 'gap': '', |
| 'ok': b['n_rules'] <= n}] |
| if b['rows']: |
| top = b['rows'][0] |
| lf, lt = b['window'] |
| lines = O.search_read('sale.order.line', |
| O.sale_line_domain(lf, lt, team_id, |
| partner_ids=[top['pid']], |
| extra=[('product_id.default_code', '=', |
| top['sku'])]), |
| ['price_subtotal', 'product_uom_qty']) |
| qty = sum(l.get('product_uom_qty') or 0 for l in lines) |
| rev = sum(l.get('price_subtotal') or 0 for l in lines) |
| unit = rev / qty if qty else 0.0 |
| checks.append({'check': f"top exception realized price — {top['sku']}", |
| 'a': round(unit, 4), 'b': round(top['realized'], 4), |
| 'gap': round(unit - top['realized'], 4), |
| 'ok': abs(unit - top['realized']) < 0.01}) |
| return checks |
|
|
