#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ERP 真实成交价索引构建（宝锐工作台「客户 Tab · 💰 价格查询」数据底座） ===================================================================== 产出：数据/price_index.json 数据源（2024-2026 三年） 2026：~/Desktop/hermas输入-工作台数据库/月度销售额数据/销售订单_20260101-YYYYMMDD.xlsx （YYYYMMDD 每次导出都变 —— 代码取该目录里日期最新的 销售订单_2026*.xlsx， 禁止在代码或注释里写死具体日期） 2025：~/Desktop/hermas输入-工作台数据库/2024和2025历史数据/2025全年订单.xlsx（Sheet1＝全年，已验证是 1-5月/6月 Sheet 的超集） 2024：~/Desktop/hermas输入-工作台数据库/2024和2025历史数据/2024全年订单.xlsx（Sheet1＝全年，列名与 2025 完全一致， 另多一列「合同日期」；2024 用日粒度日期） · 已核对：2024 文件仅含 2024-01..2024-12、2025 文件仅含 2025-01..2025-12， 两文件「单据编号」交集为 0 → 无跨年重复入库 口径（硬）： · 含税单价 > 0 才算有效价格 · 排除 是否赠品 = 是 · 排除预测单（单据编号 FO 开头） · 货号为空的记录不计入价格索引 货号 → 产品信息：数据/product_master.json（事实卡机器可读版） 命中主数据 SKU → source = product_master 命中组分(BOM)码 → source = component_of:<父SKU> 两者都不中 → source = erp_only（用 ERP 物料名称兜底，标「未收录」） 事实卡里同产品名推断出的品类，单独标 cat_inferred=true（不改主数据口径） 可重跑：换新增 ERP 文件后直接重跑本脚本即可。 用法：/usr/bin/python3 脚本/build_price_index.py """ import glob import json import os import re import sys import time import pandas as pd ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) DATA_DIR = os.path.join(ROOT, '数据') MASTER_FILE = os.path.join(DATA_DIR, 'product_master.json') OUT_FILE = os.path.join(DATA_DIR, 'price_index.json') FACT_CARD = os.path.join(ROOT, '知识库-copilot/诊断原料/产品资料/' '产品_诊断原料货号主数据_分子检测试剂_事实卡_da.md') ERP_2026_DIR = os.path.expanduser('~/Desktop/hermas输入-工作台数据库/月度销售额数据') ERP_2025_FILE = os.path.expanduser('~/Desktop/hermas输入-工作台数据库/2024和2025历史数据/2025全年订单.xlsx') ERP_2024_FILE = os.path.expanduser('~/Desktop/hermas输入-工作台数据库/2024和2025历史数据/2024全年订单.xlsx') COL_DATE, COL_ORDER, COL_CUST, COL_SALES = '合同日期', '单据编号', '客户', '销售员' COL_SKU, COL_MNAME, COL_QTY, COL_PRICE, COL_TOTAL = '货号#', '物料名称', '销售数量', '含税单价', '价税合计' COL_UNIT = '销售单位' COL_GIFT = '是否赠品' # ─────────────────────────── 工具 ─────────────────────────── def norm_sku(s): """货号归一：去全部空白（含全角），保留大小写（ERP 里存在大小写不同的同族货号）""" if s is None: return '' t = str(s) if t.lower() in ('nan', 'none', '/', '-', ''): return '' return re.sub(r'\s+', '', t.replace('\u3000', '')).strip() def num(v): try: f = float(v) except (TypeError, ValueError): return None if f != f: # NaN return None return f def pct(sorted_vals, q): """线性插值分位数（避免额外依赖）""" if not sorted_vals: return None if len(sorted_vals) == 1: return round(sorted_vals[0], 6) pos = (len(sorted_vals) - 1) * q lo = int(pos) hi = min(lo + 1, len(sorted_vals) - 1) frac = pos - lo v = sorted_vals[lo] * (1 - frac) + sorted_vals[hi] * frac return round(v, 6) def agg(records): """对一组明细算价格统计（records 元素含 price/qty/total/date/customer）""" ps = sorted(r['price'] for r in records if r.get('price') is not None) if not ps: return None qsum = sum((r.get('qty') or 0) for r in records) tsum = sum((r.get('total') or 0) for r in records) custs = set(r['customer'] for r in records if r.get('customer')) units = sorted(set((r.get('unit') or '') for r in records if r.get('unit'))) latest = max(records, key=lambda r: (str(r.get('date') or ''), str(r.get('order') or ''))) return { 'min': round(ps[0], 6), 'p25': pct(ps, 0.25), 'p50': pct(ps, 0.50), 'p75': pct(ps, 0.75), 'max': round(ps[-1], 6), 'avg': round(sum(ps) / len(ps), 6), 'weighted_avg': round(tsum / qsum, 6) if qsum else None, 'deal_count': len(records), 'customer_count': len(custs), 'qty_total': round(qsum, 3), 'amount_total': round(tsum, 2), 'units': units, 'last': {'price': round(latest['price'], 6), 'date': latest.get('date'), 'qty': latest.get('qty'), 'order': latest.get('order')}, } def year_stats(records): """年度分布：2024/2025/2026 各自的成交次数与 均/最低/最高 价（供前端看价格趋势） ⚠️ 注意：同一货号可能同时以 U（酶活单位，单价极小）与 ml/瓶（单价大）成交， 此时跨单位的 均/最低/最高 没有可比性 —— 故同时给出该年的 units 清单， 前端须在同一 unit 内比价（by_unit 下也各自带 by_year）。 """ out = {} for y in sorted(set(r.get('year') for r in records if r.get('year'))): rs = [r for r in records if r.get('year') == y and r.get('price') is not None] if not rs: continue ps = sorted(r['price'] for r in rs) out[str(y)] = { 'deal_count': len(rs), 'avg': round(sum(ps) / len(ps), 6), 'min': round(ps[0], 6), 'max': round(ps[-1], 6), 'customer_count': len(set(r['customer'] for r in rs if r.get('customer'))), 'qty_total': round(sum((r.get('qty') or 0) for r in rs), 3), 'units': sorted(set((r.get('unit') or '') for r in rs if r.get('unit'))), } return out def read_erp_2026(): """2026：取文件名日期最新的 销售订单_*.xlsx；日期列＝合同日期（日粒度）""" cands = glob.glob(os.path.join(ERP_2026_DIR, '销售订单_2026*.xlsx')) if not cands: cands = glob.glob(os.path.join(ERP_2026_DIR, '销售订单*.xlsx')) if not cands: return None, [] def keyf(p): mt = re.findall(r'(\d{8})', os.path.basename(p)) return (max(mt) if mt else '', os.path.getmtime(p)) cands.sort(key=keyf) srcs = [] for path in cands[-1:]: # 只取最新一个（全量导出） df = pd.read_excel(path) srcs.append({'file': path, 'sheet': 'Sheet1', 'year': 2026, 'date_field': COL_DATE, 'date_granularity': '日', 'raw_rows': int(len(df))}) yield_rows = df return yield_rows, srcs def read_erp_file(path, year, sheet='Sheet1'): """通用：读一个「全年订单」源文件（2024/2025 同结构）""" if not os.path.exists(path): return None, [] df = pd.read_excel(path, sheet_name=sheet) srcs = [{'file': path, 'sheet': sheet, 'year': year, 'date_field': '合同日期' if COL_DATE in df.columns else '合同月份', 'date_granularity': '日' if COL_DATE in df.columns else '月', 'raw_rows': int(len(df))}] return df, srcs def read_erp_2025(): """2025：Sheet1 为全年（已验证含 1-5月/6月 Sheet 全部单据号）；日期列＝合同月份（月粒度）""" return read_erp_file(ERP_2025_FILE, 2025) def read_erp_2024(): """2024：Sheet1 为全年（12 个月齐备）；该文件已有「合同日期」列 → 用日粒度""" return read_erp_file(ERP_2024_FILE, 2024) def extract(df, year, dcol=None): """按口径抽取有效价格明细（2024/2026 日期＝合同日期：日粒度；2025＝合同月份：月粒度）""" if dcol is None: dcol = '合同月份' if year == 2025 else COL_DATE elif year == 2025: dcol = '合同月份' out = [] for _, r in df.iterrows(): gift = str(r.get(COL_GIFT, '') or '').strip() if gift == '是': continue order = str(r.get(COL_ORDER, '') or '').strip() if order.upper().startswith('FO'): continue sku = norm_sku(r.get(COL_SKU)) if not sku: continue price = num(r.get(COL_PRICE)) if price is None or price <= 0: continue qty = num(r.get(COL_QTY)) or 0 total = num(r.get(COL_TOTAL)) if total is None: total = round(price * qty, 4) dv = r.get(dcol) if year == 2025: d = '' if (dv is None or str(dv) in ('nan', 'NaT', 'None')) else str(dv)[:7] else: d = pd.to_datetime(dv, errors='coerce') d = d.strftime('%Y-%m-%d') if pd.notna(d) else '' out.append({ 'date': d, 'customer': str(r.get(COL_CUST, '') or '').strip(), 'sales': str(r.get(COL_SALES, '') or '').strip(), 'sku': sku, 'unit': str(r.get(COL_UNIT, '') or '').strip(), 'qty': round(qty, 4), 'price': round(price, 6), 'total': round(total, 4), 'order': order, 'year': year, 'mname': str(r.get(COL_MNAME, '') or '').strip(), }) return out def load_master(): with open(MASTER_FILE, encoding='utf-8') as f: m = json.load(f) sku2 = {} comp2parent = {} for it in m['skus']: sku = norm_sku(it.get('sku')) if not sku: continue sku2[sku] = it for c in (it.get('components') or []): cs = norm_sku(c.get('code')) if cs: comp2parent.setdefault(cs, []) if it['sku'] not in comp2parent[cs]: comp2parent[cs].append(it['sku']) return m, sku2, comp2parent def load_fact_card_inferred_cat(): """事实卡里「同产品名同源推断」的品类（19 个货号），仅作检索/展示补充，标 cat_inferred""" out = {} if not os.path.exists(FACT_CARD): return out txt = open(FACT_CARD, encoding='utf-8').read() for sku, cat in re.findall(r'([A-Za-z0-9][A-Za-z0-9\-\.]{2,})\s*→\s*([0-9]{2}\.[0-9]{2}\.[0-9]{2})\s*\(', txt): if sku in ('E03', 'FE03', 'FE07', 'FE20', 'FE23', 'E13', 'FM5091', 'FMD5091', 'M5101', 'M5244', 'PM05'): continue out[norm_sku(sku)] = cat return out def main(): t0 = time.time() df26, src26 = read_erp_2026() df25, src25 = read_erp_2025() df24, src24 = read_erp_2024() recs = [] for dfy, srcy, yy in ((df26, src26, 2026), (df25, src25, 2025), (df24, src24, 2024)): if dfy is None: print('⚠️ %d 源文件缺失，跳过' % yy) continue r = extract(dfy, yy) srcy[0]['valid_rows'] = len(r) recs += r print('有效价格明细：%d 条（2026 %s / 2025 %s / 2024 %s）' % ( len(recs), src26[0].get('valid_rows') if src26 else '-', src25[0].get('valid_rows') if src25 else '-', src24[0].get('valid_rows') if src24 else '-')) master, sku2, comp2parent = load_master() inferred_cat = load_fact_card_inferred_cat() by_sku = {} for r in recs: by_sku.setdefault(r['sku'], []).append(r) all_skus = set(sku2) | set(by_sku) skus_out = {} for sku in all_skus: mi = sku2.get(sku) recs_s = by_sku.get(sku, []) if mi: src = 'product_master' parents = [] prod = { 'name': mi.get('name') or '', 'spec': mi.get('spec') or '', 'cat': mi.get('cat') or '', 'life_months': mi.get('life_months'), 'bom_group': mi.get('bom_group') or '', 'line': mi.get('line') or '', 'unit': mi.get('unit') or '', 'components': [{'code': c.get('code', ''), 'name': c.get('name', ''), 'spec': c.get('spec', ''), 'vol': c.get('vol', ''), 'unit2': c.get('unit2', '')} for c in (mi.get('components') or [])], } elif sku in comp2parent: src = 'component_of' parents = comp2parent[sku] pm = sku2.get(parents[0], {}) prod = { 'name': pm.get('name') or '', 'spec': pm.get('spec') or '', 'cat': pm.get('cat') or '', 'life_months': pm.get('life_months'), 'bom_group': pm.get('bom_group') or '', 'line': pm.get('line') or '', 'unit': '', 'components': [{'code': c.get('code', ''), 'name': c.get('name', ''), 'spec': c.get('spec', ''), 'vol': c.get('vol', ''), 'unit2': c.get('unit2', '')} for c in (pm.get('components') or [])], } prod['parent_name'] = pm.get('name') or '' prod['as_component'] = True else: src = 'erp_only' parents = [] erp_names = sorted(set(x['mname'] for x in recs_s if x.get('mname'))) prod = { 'name': (erp_names[0] if erp_names else ''), 'spec': '', 'cat': '', 'life_months': None, 'bom_group': '', 'line': '', 'unit': '', 'components': [], } # ERP 里出现过的物料名称（便于按名称检索/交叉核对） prod['erp_names'] = sorted(set(x['mname'] for x in recs_s if x.get('mname')))[:6] cat_inf = False if not prod.get('cat') and sku in inferred_cat: prod['cat'] = inferred_cat[sku] cat_inf = True prod['cat_inferred'] = cat_inf prod['source'] = src prod['parent_skus'] = parents st = agg(recs_s) if recs_s else None if st: st['by_year'] = year_stats(recs_s) # 年度分布（价格趋势） if st and len(st.get('units') or []) > 1: by_unit = {} for u in st['units']: a = agg([x for x in recs_s if (x.get('unit') or '') == u]) if a: by_unit[u] = {k: a[k] for k in ('min', 'p50', 'max', 'deal_count', 'qty_total')} by_unit[u]['by_year'] = year_stats( [x for x in recs_s if (x.get('unit') or '') == u]) st['by_unit'] = by_unit # 按客户聚合（价格敏感：接口层再按权限裁剪） bc = {} for x in recs_s: c = x['customer'] or '—' bc.setdefault(c, []).append(x) by_customer = [] for c, rs in bc.items(): a = agg(rs) a['customer'] = c a['salespersons'] = sorted(set(v['sales'] for v in rs if v.get('sales'))) by_customer.append(a) by_customer.sort(key=lambda z: -(z['amount_total'] or 0)) skus_out[sku] = { 'sku': sku, 'product': prod, 'stats': st, 'by_customer': by_customer, 'records': sorted(recs_s, key=lambda x: (str(x['date']), str(x['order'])), reverse=True), } # 名称检索索引（含主数据名 / ERP 物料名 / 规格 / 品类）——供接口做关键词匹配 search = {} for sku, o in skus_out.items(): names = [o['product'].get('name') or '', o['product'].get('spec') or '', o['product'].get('cat') or '', sku] names += o['product'].get('erp_names') or [] key = ' '.join(names).lower() search[sku] = re.sub(r'\s+', ' ', key)[:400] salepersons = {} for r in recs: s = r['sales'] or '—' d = salepersons.setdefault(s, {'deal_count': 0, 'customers': set(), 'skus': set()}) d['deal_count'] += 1 if r['customer']: d['customers'].add(r['customer']) d['skus'].add(r['sku']) sales_out = {k: {'deal_count': v['deal_count'], 'customer_count': len(v['customers']), 'sku_count': len(v['skus'])} for k, v in salepersons.items()} meta = { 'generated': time.strftime('%Y-%m-%d %H:%M:%S'), 'year_range': '2024-2026', 'sources': src26 + src25 + src24, 'criteria': { 'price': '含税单价 > 0', 'exclude_gift': '是否赠品 = 是 已排除', 'exclude_forecast': '单据编号 FO 开头（预测单）已排除', 'empty_sku': '货号为空不计入', 'amount': '金额口径＝价税合计（ERP 原文）', 'note': '2025 数据日期为月粒度（ERP 导出列为「合同月份」），2024/2026 为日粒度（合同日期）', }, 'sku_count': len(skus_out), 'sku_with_price': len(by_sku), 'record_count': len(recs), 'by_year': {str(y): { 'record_count': sum(1 for r in recs if r['year'] == y), 'sku_count': len(set(r['sku'] for r in recs if r['year'] == y)), 'customer_count': len(set(r['customer'] for r in recs if r['year'] == y and r['customer'])), 'sales_count': len(set(r['sales'] for r in recs if r['year'] == y and r['sales'])), 'order_count': len(set(r['order'] for r in recs if r['year'] == y and r['order'])), } for y in (2024, 2025, 2026)}, 'customer_count': len(set(r['customer'] for r in recs if r['customer'])), 'sales_count': len(salepersons), 'master': {'file': '数据/product_master.json', 'generated': master.get('generated'), 'total_skus': master.get('total_skus')}, 'fact_card': {'file': os.path.relpath(FACT_CARD, ROOT), 'inferred_cat_used': sorted(inferred_cat.keys())}, 'record_fields': ['date(日期)', 'customer(客户)', 'sales(销售员)', 'sku(货号)', 'qty(销售数量)', 'price(含税单价)', 'total(价税合计)', 'order(单据编号)'], 'build_seconds': round(time.time() - t0, 2), } out = {'meta': meta, 'skus': skus_out, 'search': search, 'sales': sales_out} tmp = OUT_FILE + '.tmp' with open(tmp, 'w', encoding='utf-8') as f: json.dump(out, f, ensure_ascii=False, separators=(',', ':')) os.replace(tmp, OUT_FILE) size = os.path.getsize(OUT_FILE) / 1024.0 / 1024.0 print('✅ %s（%.2f MB）SKU %d（有价格 %d） 明细 %d 条 客户 %d 销售员 %d 耗时 %.1fs' % ( OUT_FILE, size, meta['sku_count'], meta['sku_with_price'], meta['record_count'], meta['customer_count'], meta['sales_count'], meta['build_seconds'])) if __name__ == '__main__': sys.exit(main())