diff --git a/scripts/stock_deal_report.py b/scripts/stock_deal_report.py new file mode 100644 index 0000000..56ef18d --- /dev/null +++ b/scripts/stock_deal_report.py @@ -0,0 +1,309 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +备货单成交统计报告生成器（固定脚本） + +口径： + 成交 = 有备货单（stock_orders）；时间按单据日期 biz_date； + 剔除 草稿/未提交(approval_status=0)、已拒绝(3)、已作废(4)；排除软删除； + 初版与关联终版（object_id 互指）同月并存时只计终版，跨月各计各月； + 车况：明细 mileage>0 为二手车、=0 为新车；国家取清关目的国（去英文括号后缀）。 + +用法： + python3 scripts/stock_deal_report.py --year 2026 --month 7 # 单月 + python3 scripts/stock_deal_report.py --from 2026-01 --to 2026-06 # 区间 + 可选：--file runtime/fdesk-20260813.sql（默认） + +输出：runtime/<期间>成交统计报告.pdf + runtime/<期间>成交详单.csv +依赖：python3 + weasyprint（首次对每个 dump 建解析缓存，之后秒级重跑） +""" +import argparse, re, csv, html, os, pickle +from collections import Counter, defaultdict +from datetime import datetime, timezone, timedelta +from weasyprint import HTML + +TZ = timezone(timedelta(hours=8)) +EXCLUDE = {0, 3, 4} # 草稿/拒绝/作废 +DEFAULT_DUMP = 'runtime/fdesk-20260813.sql' + +# ---------------- SQL dump 解析 ---------------- +def pvl(s, i): + vals = []; cur = ''; instr = False; n = len(s) + while i < n: + c = s[i] + if instr: + if c == '\\' and i + 1 < n: cur += c + s[i + 1]; i += 2; continue + if c == "'": + if i + 1 < n and s[i + 1] == "'": cur += "''"; i += 2; continue + cur += "'"; i += 1; instr = False; continue + cur += c; i += 1; continue + else: + if c == "'": instr = True; cur += "'"; i += 1; continue + if c == ',': vals.append(cur.strip()); cur = ''; i += 1; continue + if c == ')': vals.append(cur.strip()); i += 1; break + cur += c; i += 1 + return vals, i + +def unesc(raw): + raw = raw.strip() + if raw.upper() == 'NULL': return None + if len(raw) >= 2 and raw[0] == "'" and raw[-1] == "'": + inner = raw[1:-1]; out = []; i = 0; L = len(inner) + mp = {'n': '\n', 'r': '\r', 't': '\t', '0': '\x00', '\\': '\\', "'": "'", '"': '"', 'b': '\b', 'Z': '\x1a'} + while i < L: + c = inner[i] + if c == '\\' and i + 1 < L: out.append(mp.get(inner[i + 1], inner[i + 1])); i += 2 + elif c == "'" and i + 1 < L and inner[i + 1] == "'": out.append("'"); i += 2 + else: out.append(c); i += 1 + return ''.join(out) + try: return int(raw) + except: + try: return float(raw) + except: return raw + +def load_dump(path): + """返回 (orders: id->dict, items: [(order_id, model, qty, mileage)])""" + cache = path + '.cache.pkl' + if os.path.exists(cache): + with open(cache, 'rb') as f: + return pickle.load(f) + cols = {}; cur_create = None + orders = {}; items = [] + with open(path, encoding='utf-8', errors='replace') as f: + for line in f: + if cur_create is not None: + m = re.match(r'\s*`(\w+)`\s', line) + if m: cols[cur_create].append(m.group(1)) + if re.search(r'\)\s*ENGINE', line): cur_create = None + continue + if line.startswith('CREATE TABLE `'): + t = line[14:line.index('`', 14)] + if t in ('stock_orders', 'stock_order_items'): + cur_create = t; cols[t] = [] + continue + if line.startswith('INSERT INTO `stock_orders` VALUES ('): + v = [unesc(x) for x in pvl(line, line.index('VALUES (') + 8)[0]] + ci = {c: i for i, c in enumerate(cols['stock_orders'])} + orders[v[ci['id']]] = dict( + order_no=v[ci['order_no']], biz=v[ci['biz_date']] or 0, + fin=v[ci['is_final']], oid=v[ci['object_id']] or 0, + ap=v[ci['approval_status']], deleted=v[ci['deleted_at']] is not None, + country=str(v[ci['clearance_country_snapshot']] or '').strip(), + sales=str(v[ci['salesperson_snapshot']] or '').strip(), + ) + elif line.startswith('INSERT INTO `stock_order_items` VALUES ('): + v = [unesc(x) for x in pvl(line, line.index('VALUES (') + 8)[0]] + ci = {c: i for i, c in enumerate(cols['stock_order_items'])} + items.append((v[ci['order_id']], str(v[ci['model_snapshot']] or '').strip(), + int(v[ci['quantity']] or 0), int(v[ci['mileage']] or 0))) + with open(cache, 'wb') as f: + pickle.dump((orders, items), f) + return orders, items + +# ---------------- 统计 ---------------- +_PAREN = re.compile(r'\((?:[^()]|\([^()]*\))*\)\s*$') + +def clean_country(c): + while True: + c2 = _PAREN.sub('', c).strip() + if c2 == c: return c + c = c2 + +def month_key(ts): + return datetime.fromtimestamp(ts, TZ).strftime('%Y-%m') + +def build_report(orders, items, months): + """months: [(y, m), ...] 连续月份列表""" + for o in orders.values(): + o['country'] = clean_country(o['country']) + + t0 = int(datetime(months[0][0], months[0][1], 1, tzinfo=TZ).timestamp()) + ly, lm = months[-1] + t1 = int((datetime(ly, lm, 1, tzinfo=TZ) + timedelta(days=32)).replace(day=1).timestamp()) + + def in_range(o): + return (not o['deleted']) and t0 <= o['biz'] < t1 + + cand = {i: o for i, o in orders.items() if in_range(o) and o['ap'] not in EXCLUDE} + excluded = {a: sum(1 for o in orders.values() if in_range(o) and o['ap'] == a) for a in sorted(EXCLUDE)} + + # 初终版去重（按月）：初版的关联终版与初版同月 -> 跳过初版 + sel = {}; skipped = []; cross_month_fin = 0 + for i, o in cand.items(): + if o['fin'] == 0 and o['oid'] > 0: + fin = orders.get(o['oid']) + if fin and fin['fin'] == 1 and not fin['deleted']: + if month_key(fin['biz']) == month_key(o['biz']): + skipped.append(i); continue + cross_month_fin += 1 + sel[i] = o + + order_ids = set(sel) + items_by_order = defaultdict(list) + for oid, model, qty, mile in items: + if oid in order_ids: + items_by_order[oid].append((model, qty, mile)) + + by_model = Counter(); by_mc = defaultdict(Counter); by_country = Counter() + by_sales = Counter(); by_sales_orders = Counter(); by_cond = Counter() + by_model_cond = defaultdict(Counter); by_country_cond = defaultdict(Counter) + by_sales_cond = defaultdict(Counter); by_month = defaultdict(lambda: Counter()) + detail = [] + + for oid, o in sel.items(): + sp = o['sales'] or '(未填)' + c = o['country'] or '(未填)' + mk = month_key(o['biz']) + by_sales_orders[sp] += 1 + by_month[mk]['orders'] += 1 + for model, qty, mile in items_by_order.get(oid, []): + m = model or '(未填车型)' + cd = '二手车' if (mile or 0) > 0 else '新车' + by_model[m] += qty; by_mc[m][c] += qty; by_country[c] += qty; by_sales[sp] += qty + by_cond[cd] += qty + by_model_cond[m][cd] += qty; by_country_cond[c][cd] += qty; by_sales_cond[sp][cd] += qty + by_month[mk][cd] += qty; by_month[mk]['veh'] += qty + detail.append(dict(model=m, country=c, qty=qty, sales=sp, month=mk, + cond=cd if cd == '新车' else f'二手车({mile:,}km)', + order_no=o['order_no'], date=datetime.fromtimestamp(o['biz'], TZ).strftime('%m-%d'))) + detail.sort(key=lambda r: (r['month'], r['sales'], r['order_no'])) + return dict(sel=sel, excluded=excluded, skipped=skipped, cross=cross_month_fin, + by_model=by_model, by_mc=by_mc, by_country=by_country, by_sales=by_sales, + by_sales_orders=by_sales_orders, by_cond=by_cond, + by_model_cond=by_model_cond, by_country_cond=by_country_cond, + by_sales_cond=by_sales_cond, by_month=by_month, detail=detail) + +# ---------------- 渲染 ---------------- +def esc(x): return html.escape(str(x)) +def tr(rows): + return ''.join('' + ''.join(esc(c) for c in r) + '' for r in rows) + +def render_pdf(R, months, label, out_pdf): + total = sum(R['by_model'].values()) + new_n = R['by_cond'].get('新车', 0); used_n = R['by_cond'].get('二手车', 0) + multi = len(months) > 1 + + model_rows = [(m, R['by_model_cond'][m].get('新车', 0), R['by_model_cond'][m].get('二手车', 0), n, + '、'.join(f'{c}×{k}' for c, k in R['by_mc'][m].most_common())) + for m, n in sorted(R['by_model'].items(), key=lambda x: -x[1])] + country_rows = [(c, R['by_country_cond'][c].get('新车', 0), R['by_country_cond'][c].get('二手车', 0), n, f'{n / total * 100:.1f}%') + for c, n in R['by_country'].most_common()] + sales_rows = [(s, R['by_sales_orders'][s], R['by_sales_cond'][s].get('新车', 0), R['by_sales_cond'][s].get('二手车', 0), + R['by_sales'][s], f'{R["by_sales"][s] / total * 100:.1f}%') + for s, _ in sorted(R['by_sales'].items(), key=lambda x: -R['by_sales'][x[0]])] + month_rows = [(mk, c['orders'], c['veh'], c.get('新车', 0), c.get('二手车', 0)) + for mk, c in sorted(R['by_month'].items())] + dcol = [(r['sales'], r['month'], r['model'], r['country'], r['cond'], r['qty'], r['order_no'], r['date']) for r in R['detail']] \ + if multi else \ + [(r['sales'], r['model'], r['country'], r['cond'], r['qty'], r['order_no'], r['date']) for r in R['detail']] + + sec = [] + if multi: + sec.append(('月度分布', '' + '' + f'{tr(month_rows)}
月份单数车辆数新车二手车
')) + sec.append(('车型成交排行', '' + '' + f'{tr(model_rows)}
车型新车二手合计成交国家（国家×辆数）
')) + sec.append(('国家成交排行', '' + '' + f'{tr(country_rows)}
国家（清关目的国）新车二手合计占比
')) + sec.append(('业务员成交排行', '' + '' + '' + f'{tr(sales_rows)}
业务员单数新车二手辆数合计占车辆数比
')) + dhead = ('业务员月份车型国家' + '车况数量单号单据日期') if multi else \ + ('业务员车型国家车况' + '数量单号单据日期') + sec.append(('成交详单（按月份、业务员）', f'{dhead}{tr(dcol)}
')) + sections = ''.join(f'

{i}、{t}

{b}' for i, (t, b) in enumerate(sec, 1)) + + ex = R['excluded'] + doc = f""" +

备货单成交统计报告 · {label}

+
数据来源：stock_orders / stock_order_items · 生成日期 {datetime.now(TZ).date().isoformat()} · 统计口径：有备货单即成交
+
口径说明：单据日期（biz_date）落在 {label}；已剔除 草稿/未提交（{ex.get(0, 0)} 单）、已拒绝（{ex.get(3, 0)} 单）、已作废（{ex.get(4, 0)} 单）； +初版与关联终版（object_id 互指）同月并存时只计终版（本期跳过初版 {len(R['skipped'])} 单）；跨月关联终版 {R['cross']} 单按所属月份计入。 +车况判定：系统无显式新车/二手车字段，按明细 mileage 判定：里程>0 为二手车、=0 为新车。国家按清关目的国。
+ + + + + + + +
{len(R['sel'])}
成交单数
{total}
成交车辆数
{new_n}
新车（辆）
{used_n}
二手车（辆）
{len(R['by_model'])}
成交车型数
{len(R['by_country'])}
覆盖国家数
+{sections} +

注：明细按 stock_order_items 逐行展开（数量取 quantity）；车况括号内为二手车明细里程（km）；业务员为建单时快照（salesperson_snapshot）。

+""" + HTML(string=doc).write_pdf(out_pdf) + +def write_csv(R, label, multi): + path = f'runtime/{label}成交详单.csv' + with open(path, 'w', newline='', encoding='utf-8-sig') as f: + w = csv.writer(f) + w.writerow(['业务员', '月份', '车型', '国家', '车况', '数量', '单号', '单据日期'] if multi + else ['业务员', '车型', '国家', '车况', '数量', '单号', '单据日期']) + for r in R['detail']: + row = [r['sales'], r['month'], r['model'], r['country'], r['cond'], r['qty'], r['order_no'], r['date']] if multi \ + else [r['sales'], r['model'], r['country'], r['cond'], r['qty'], r['order_no'], r['date']] + w.writerow(row) + return path + +def main(): + ap = argparse.ArgumentParser(description='备货单成交统计报告') + ap.add_argument('--file', default=DEFAULT_DUMP, help='Navicat 全库 dump 路径') + g = ap.add_mutually_exclusive_group(required=True) + g.add_argument('--year', type=int); ap.add_argument('--month', type=int, default=None) + g.add_argument('--frm', '--from', dest='frm', default=None, help='起始月 2026-01') + ap.add_argument('--to', default=None, help='结束月 2026-06') + args = ap.parse_args() + + if args.frm: + y1, m1 = map(int, args.frm.split('-')) + y2, m2 = map(int, (args.to or args.frm).split('-')) + months = [] + y, m = y1, m1 + while (y, m) <= (y2, m2): + months.append((y, m)); m += 1 + if m == 13: y, m = y + 1, 1 + else: + months = [(args.year, args.month or datetime.now(TZ).month)] + + if len(months) == 1: + y, m = months[0]; label = f'{y}年{m}月' + else: + (y1, m1), (y2, m2) = months[0], months[-1] + label = f'{y1}年{m1}-{m2}月' if y1 == y2 else f'{y1}年{m1}月-{y2}年{m2}月' + + orders, items = load_dump(args.file) + print(f'dump: {args.file}（orders {len(orders)} / items {len(items)}），期间: {label}') + R = build_report(orders, items, months) + total = sum(R['by_model'].values()) + print(f"入选 {len(R['sel'])} 单 / {total} 辆（新车 {R['by_cond'].get('新车', 0)} / 二手 {R['by_cond'].get('二手车', 0)}），" + f"车型 {len(R['by_model'])}，国家 {len(R['by_country'])}，剔除 {R['excluded']}，同月终版跳过初版 {len(R['skipped'])} 单") + pdf = f'runtime/{label}成交统计报告.pdf' + render_pdf(R, months, label, pdf) + csvp = write_csv(R, label, len(months) > 1) + print('已生成:', pdf, '|', csvp) + +if __name__ == '__main__': + main()