File size: 6,465 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 | """Order Book — pre-season booking coverage (the forward view sitting unused in sale.order).
Giftware pre-books: confirmed orders carry FUTURE delivery windows (commitment_date), so the
order book for the season is known months ahead. The curve compares THIS year's cumulative
booked $ for season delivery against last year's same-week curve — 20% under with 12 weeks to
go = cut the container program NOW; over = expand POs / push laggard agents.
Season window: deliveries Aug 1 – Dec 31 (the Q4 giftware peak; env-overridable later if the
business adds a spring season). x-axis = weeks before Oct 1 (nominal peak).
"""
import datetime as dt
import core.odoo as O
import core.periods as P
SEASON_FROM = (8, 1) # deliveries from Aug 1
SEASON_TO = (12, 31) # through Dec 31
PEAK = (10, 1) # nominal peak for the weeks-before axis
def _season_orders(year, team_id=None, cutoff=None):
"""Confirmed orders with commitment_date inside `year`'s season window, booked up to
`cutoff` (order date). Returns [(order_date, amount)]."""
ex = O.excluded_partner_ids()
dom = [('state', 'in', ['sale', 'done']),
('commitment_date', '>=', f'{year}-{SEASON_FROM[0]:02d}-{SEASON_FROM[1]:02d} 00:00:00'),
('commitment_date', '<=', f'{year}-{SEASON_TO[0]:02d}-{SEASON_TO[1]:02d} 23:59:59')]
dom.append(('team_id', '=', team_id) if team_id is not None
else ('team_id', 'in', O.TEAM_IDS))
if cutoff:
dom.append(('date_order', '<=', f'{cutoff} 23:59:59'))
if ex:
dom.append(('partner_id', 'not in', list(ex)))
rows = O.search_read('sale.order', dom, ['date_order', 'amount_untaxed'])
return [(str(r.get('date_order') or '')[:10], r.get('amount_untaxed') or 0.0)
for r in rows if r.get('date_order')]
def curve(team_id=None, t=None, years_back=2):
"""Cumulative booked-$ curves, one per season year, aligned on weeks-before-peak.
The current year's curve stops at today; prior years run to their season end."""
t = t or P.today()
ty = t.year
out_rows, totals = [], {}
for yr in range(ty - years_back, ty + 1):
peak = dt.date(yr, *PEAK)
cutoff = t.isoformat() if yr == ty else None
orders = _season_orders(yr, team_id, cutoff)
orders.sort()
cum = 0.0
weekly = {}
for d, amt in orders:
cum += amt
wk = (peak - dt.date.fromisoformat(d)).days // 7
weekly[wk] = cum # last write per week = cumulative at week end
for wk, v in sorted(weekly.items(), reverse=True):
out_rows.append({'year': str(yr), 'weeks_before_peak': -wk, 'booked': v})
totals[str(yr)] = cum
# same-week-LY comparison for the headline
wk_now = (dt.date(ty, *PEAK) - t).days // 7
ly_same = 0.0
for r in out_rows:
if r['year'] == str(ty - 1) and r['weeks_before_peak'] <= -wk_now:
ly_same = max(ly_same, r['booked'])
return {'rows': out_rows, 'totals': totals,
'ty': str(ty), 'ly': str(ty - 1),
'booked_ty': totals.get(str(ty), 0.0), 'ly_same_week': ly_same,
'weeks_to_peak': wk_now}
def by_category(team_id=None, t=None):
"""Booked $ per product category, this season TY vs LY-as-of-the-same-date.
(read_group cannot group by a dot-path — group by product, map to category locally.)"""
t = t or P.today()
ty = t.year
per_prod = {}
for yr, cut in ((ty, t.isoformat()),
(ty - 1, t.replace(year=ty - 1).isoformat())):
ex = O.excluded_partner_ids()
dom = [('order_id.state', 'in', ['sale', 'done']),
('order_id.commitment_date', '>=',
f'{yr}-{SEASON_FROM[0]:02d}-{SEASON_FROM[1]:02d} 00:00:00'),
('order_id.commitment_date', '<=',
f'{yr}-{SEASON_TO[0]:02d}-{SEASON_TO[1]:02d} 23:59:59'),
('order_id.date_order', '<=', f'{cut} 23:59:59'),
('product_id', '!=', False)]
dom.append(('order_id.team_id', '=', team_id) if team_id is not None
else ('order_id.team_id', 'in', O.TEAM_IDS))
if ex:
dom.append(('order_partner_id', 'not in', list(ex)))
for g in O.read_group('sale.order.line', dom, ['price_subtotal:sum'],
['product_id'], lazy=False):
pid = O.m2o_id(g.get('product_id'))
if not pid:
continue
e = per_prod.setdefault(pid, {'ty': 0.0, 'ly': 0.0})
e['ty' if yr == ty else 'ly'] += g.get('price_subtotal') or 0.0
cats = {}
pids = list(per_prod)
for i in range(0, len(pids), 2000):
for p in O.search_read('product.product',
[('id', 'in', pids[i:i + 2000]),
('active', 'in', [True, False])], ['categ_id']):
cats[p['id']] = O.m2o_name(p.get('categ_id')) or '(none)'
out = {}
for pid, v in per_prod.items():
cat = cats.get(pid, '(none)')
e = out.setdefault(cat, {'category': cat, 'ty': 0.0, 'ly': 0.0})
e['ty'] += v['ty']
e['ly'] += v['ly']
rows = list(out.values())
for r in rows:
r['delta_pct'] = ((r['ty'] / r['ly'] - 1) * 100.0) if r['ly'] else None
rows.sort(key=lambda x: -x['ty'])
return rows
def validate(t=None, team_id=None):
"""The client-side cumulative end point ties one server-side aggregate over the exact
same domain (sum_field) — the independent arithmetic path."""
t = t or P.today()
b = curve(team_id, t)
ty = int(b['ty'])
ex = O.excluded_partner_ids()
dom = [('state', 'in', ['sale', 'done']),
('commitment_date', '>=', f'{ty}-{SEASON_FROM[0]:02d}-{SEASON_FROM[1]:02d} 00:00:00'),
('commitment_date', '<=', f'{ty}-{SEASON_TO[0]:02d}-{SEASON_TO[1]:02d} 23:59:59'),
('date_order', '<=', f'{t.isoformat()} 23:59:59')]
dom.append(('team_id', '=', team_id) if team_id is not None
else ('team_id', 'in', O.TEAM_IDS))
if ex:
dom.append(('partner_id', 'not in', list(ex)))
srv = O.sum_field('sale.order', dom, 'amount_untaxed')
return [{'check': f'{ty} season booked $ — client cumulative vs server sum',
'a': round(b['booked_ty'], 2), 'b': round(srv, 2),
'gap': round(b['booked_ty'] - srv, 2),
'ok': abs(b['booked_ty'] - srv) < 1.0}]
|