#!/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()