File size: 7,587 Bytes
c14ceee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """Assortment module — facet-level performance from the curated product taxonomy the team
already maintains IN Odoo (x_main/x_sub/x_color/x_material/x_occasion/x_collection +
x_studio_season), which no other module used until 2026-07-05.
Governing question (dashboard standard): WHICH PARTS OF THE ASSORTMENT EARN THEIR KEEP — AND IS
THE SEASONAL BUY READY? Revenue/margin/YoY per facet value (BU-scoped via the standard wholesale
line domain), plus season readiness: units the coming 6 months sold LAST year per season tag vs
units on hand today. validate() reconciles facet partitions to independent totals.
"""
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
FACETS = [('x_main', 'Main category'), ('x_sub', 'Sub-category'), ('x_color', 'Color'),
('x_material', 'Material'), ('x_occasion', 'Occasion'),
('x_collection', 'Collection'), ('x_studio_season', 'Season')]
UNTAGGED = '(untagged)'
def _products():
"""pid -> {code, facets..., on_hand, cost}. One pull, all facet fields."""
fields = ['default_code', 'qty_available', 'standard_price'] + [f for f, _ in FACETS]
out = {}
for p in O.search_read('product.product',
[('default_code', '!=', False), ('active', 'in', [True, False])],
fields):
out[p['id']] = p
return out
def _rev_by_product(df, dt_, team_id=None):
"""product_id -> {rev, qty, margin} over the window, standard wholesale scope."""
out = {}
for g in O.read_group('sale.order.line', O.sale_line_domain(df, dt_, team_id),
['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'],
['product_id'], lazy=False):
if g.get('product_id'):
out[O.m2o_id(g['product_id'])] = {
'rev': g.get('price_subtotal') or 0.0,
'qty': g.get('product_uom_qty') or 0.0,
'margin': g.get('margin') or 0.0}
return out
def build(team_id=None, t=None):
t = t or P.today()
yf, yt = P.ytd(t)
lf, lt = P.ytd_last_year(t)
prods = _products()
now = _rev_by_product(yf, yt, team_id)
ly = _rev_by_product(lf, lt, team_id)
# ---- facet rollups (every dimension partitions the same YTD revenue) -------------------
dims = {}
for fkey, flabel in FACETS:
agg = {}
for pid, r in now.items():
p = prods.get(pid)
val = ((p or {}).get(fkey) or UNTAGGED) if p else UNTAGGED
val = str(val).strip() or UNTAGGED
e = agg.setdefault(val, {'value': val, 'rev': 0.0, 'rev_ly': 0.0, 'margin': 0.0,
'qty': 0.0, 'skus': set()})
e['rev'] += r['rev']
e['margin'] += r['margin']
e['qty'] += r['qty']
e['skus'].add(pid)
for pid, r in ly.items():
p = prods.get(pid)
val = ((p or {}).get(fkey) or UNTAGGED) if p else UNTAGGED
val = str(val).strip() or UNTAGGED
agg.setdefault(val, {'value': val, 'rev': 0.0, 'rev_ly': 0.0, 'margin': 0.0,
'qty': 0.0, 'skus': set()})['rev_ly'] += r['rev']
rows = []
for e in agg.values():
e['n_skus'] = len(e['skus'])
del e['skus']
e['gm_pct'] = (e['margin'] / e['rev'] * 100) if e['rev'] else None
e['yoy_pct'] = ((e['rev'] - e['rev_ly']) / e['rev_ly'] * 100) if e['rev_ly'] else None
e['yoy_abs'] = e['rev'] - e['rev_ly']
rows.append(e)
rows.sort(key=lambda x: -x['rev'])
dims[fkey] = {'label': flabel, 'rows': rows}
# ---- coverage: how much of revenue is on FACETED (x_main-tagged) SKUs ------------------
total_rev = sum(r['rev'] for r in now.values())
faceted_rev = sum(r['rev'] for pid, r in now.items()
if prods.get(pid, {}).get('x_main'))
n_faceted = sum(1 for p in prods.values() if p.get('x_main'))
# ---- season readiness: units sold in [today, +180d] LAST YEAR per season vs on hand ----
nf, ntt = (t - dt.timedelta(days=365)).isoformat(), (t + dt.timedelta(days=180) - dt.timedelta(days=365)).isoformat()
ahead_ly = _rev_by_product(nf, ntt, team_id)
seasons = {}
for pid, r in ahead_ly.items():
p = prods.get(pid)
s = str((p or {}).get('x_studio_season') or '').strip()
if not s:
continue
e = seasons.setdefault(s, {'season': s, 'demand_units_ly': 0.0, 'demand_rev_ly': 0.0,
'on_hand_units': 0.0, 'on_hand_value': 0.0, 'skus': set()})
e['demand_units_ly'] += r['qty']
e['demand_rev_ly'] += r['rev']
e['skus'].add(pid)
for pid, p in prods.items():
s = str(p.get('x_studio_season') or '').strip()
if s and s in seasons:
seasons[s]['on_hand_units'] += p.get('qty_available') or 0.0
seasons[s]['on_hand_value'] += (p.get('qty_available') or 0.0) * (p.get('standard_price') or 0.0)
season_rows = []
for e in seasons.values():
e['n_skus'] = len(e['skus'])
e['codes'] = sorted((prods.get(pid, {}).get('default_code') or '').strip()
for pid in e['skus'] if prods.get(pid, {}).get('default_code'))
del e['skus']
e['cover_pct'] = (e['on_hand_units'] / e['demand_units_ly'] * 100) if e['demand_units_ly'] else None
season_rows.append(e)
season_rows.sort(key=lambda x: -x['demand_rev_ly'])
# suspect facet hygiene: values carried by <3 SKUs across ALL products (typos like 'vgsd')
suspects = []
for fkey, flabel in FACETS:
vals = {}
for p in prods.values():
v = str(p.get(fkey) or '').strip()
if v:
vals[v] = vals.get(v, 0) + 1
suspects += [{'facet': flabel, 'value': v, 'skus': n} for v, n in vals.items() if n < 3]
return {'dims': dims, 'season': season_rows, 'suspects': sorted(suspects, key=lambda x: x['skus']),
'total_rev': total_rev, 'faceted_rev': faceted_rev,
'faceted_share': (faceted_rev / total_rev * 100) if total_rev else None,
'n_faceted': n_faceted, 'n_products': len(prods),
'ytd': (yf, yt), 'ly': (lf, lt)}
def validate(t=None, team_id=None, pre=None):
t = t or P.today()
b = pre or build(team_id, t)
yf, yt = b['ytd']
total = O.sum_field('sale.order.line', O.sale_line_domain(yf, yt, team_id), 'price_subtotal')
checks = []
for fkey, flabel in FACETS[:2]: # every dimension partitions the SAME revenue; check two
s = sum(r['rev'] for r in b['dims'][fkey]['rows'])
checks.append({'check': f'Assortment: Σ({flabel} facet rev) == total line revenue (YTD)',
'a': round(s, 2), 'b': round(total, 2), 'gap': round(s - total, 2),
'ok': abs(s - total) <= max(1.0, abs(total) * 0.001)})
fs = b['faceted_rev'] + sum(r['rev'] for r in b['dims']['x_main']['rows']
if r['value'] == UNTAGGED)
checks.append({'check': 'Assortment: faceted rev + untagged bucket == total (partition)',
'a': round(fs, 2), 'b': round(sum(r['rev'] for r in b['dims']['x_main']['rows']), 2),
'gap': round(fs - sum(r['rev'] for r in b['dims']['x_main']['rows']), 2),
'ok': abs(fs - sum(r['rev'] for r in b['dims']['x_main']['rows'])) <= 1.0})
return checks
|