#!/usr/bin/env python3 # -*- coding: utf-8 -*- """系列级资料归档 v3（含客户维度） v3 变更（2026-10-26 用户确认）： 1. 归集改「货号编码法」：剥离 P/F 前缀、X 档位、配方号、形态与组分后缀，同键必同系列 - 修正 M5244-ZJ4-TP 误入 MD2104 - 合并 MD104 / MD084 / MD071 / M244 / M274 / M134 / M211 / E04 等同键拆分 2. 新增客户维度：客户数、单一客户货号数、单客金额占比、TOP1 客户与占比、集中度 3. 未归类兜底：所有有销货号均按货号键落系列，不再出现「未归类」 """ import difflib import json import os import re import sys from collections import Counter, defaultdict import openpyxl from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter sys.path.insert(0, '/Users/liuxinyuan/Desktop/Hermes输出-工作类/脚本') import erp_sales_core as core ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' DOC = '/Users/liuxinyuan/.hermes/profiles/zone-b/cache/documents' F2 = os.path.join(DOC, 'doc_a1cb1a53ca50_2026-10-24-附件2（已有数据思维导图清单）.xlsx') OFFICIAL = os.path.join(ROOT, '数据', '客户清单_官方_20260930.json') OUT = os.path.join(ROOT, '销售分析', '产品系列资料归档与缺口清单_v3_2026-10-26.xlsx') CORE = ['冻融稳定性', '加速稳定性', '效期/长期/实时稳定性', '运输/物流稳定性'] FORM = ['NA', 'DG', 'DB', 'TP', 'FP', '5P', 'O20', 'O25', 'S50', 'S70', 'P01', 'P02', 'P03'] INTERNAL = ('宝锐', '宝泰', '横琴') DG_PARAMS = ('冻干前后活性回收率、冻干粉复溶时间与澄清度、冻干保护剂兼容性、冻干态加速稳定性、冻干球含水量、常温运输耐受') NA_PARAMS = ('宿主核酸残留、DNase/RNase 残留、阴性对照 Ct、低拷贝检出限、批间背景一致性') def code_key(sku): x = re.sub(r'\s+', '', sku or '').upper() x = re.sub(r'-\d$', '', x) for t in FORM: x = re.sub(r'-%s(?=-|$)' % t, '', x) x = re.sub(r'-{2,}', '-', x).strip('-') if re.match(r'^P[A-Z]', x): x = x[1:] if re.match(r'^\d', x): return re.sub(r'-\d+$', '', x.split('-')[0] + '-' + (x.split('-')[1] if '-' in x else '')) or '其他/数字编码' m = re.match(r'^([A-Z]+)(\d+)(.*)$', x) if not m: return x L, D = m.group(1), m.group(2) if L.startswith('F') and len(L) > 1: L = L[1:] if L in ('M', 'MD') and len(D) >= 4: D = D[-3:] return L + D def family(name): n = re.sub(r'^\s*\d+(?:\.\d+)?\s*[×xX]\s*', '', name or '') n = re.sub(r'\s*[\(（].*$', '', n).strip().replace('Ⅱ', 'II').replace('Ⅲ', 'III').replace('Ⅳ', 'IV') return re.sub(r'\s+', ' ', n) or '未命名' def kind(name): n = name or '' if re.search(r'Premix|Mix|预混', n, re.I): return '预混液' if re.search(r'Taq|RTase|逆转录|UNG|Bst|酶|Enzyme|Polymerase|Tth|Inhibitor|EXase', n, re.I): return '单酶' return '其他' def xfold(name): m = re.search(r'(\d+(?:\.\d+)?)\s*[×xX]', name or '') return m.group(1) + '×' if m else '' def norm_cust(s): s = re.sub(r'（.*?）|\(.*?\)', '', s or '').strip() for t in ('股份有限公司', '有限责任公司', '有限公司', '公司', '集团'): s = s.replace(t, '') return re.sub(r'\s+', '', s) def docs_of(sku, f2): if sku in f2: return f2[sku] for k, v in f2.items(): if sku.startswith(k) or k.startswith(sku): return v return {'docs': 0, 'core': {}} def main(): off = json.load(open(OFFICIAL, encoding='utf-8'))['customers'] off_names = {x['cust'] for x in off} canon = {} for x in off: canon.setdefault(norm_cust(x['cust']), x['cust']) names_list = list(canon) cmap = {} def match(c): if c in cmap: return cmap[c] if c in off_names: r = c else: n = norm_cust(c) if n in canon: r = canon[n] else: m = difflib.get_close_matches(n, names_list, n=1, cutoff=0.88) r = canon[m[0]] if m else c cmap[c] = r return r # ERP 聚合 p = core.find_latest_erp() wb = openpyxl.load_workbook(p, data_only=True, read_only=True) ws = wb.worksheets[0] it = ws.iter_rows(values_only=True) H = [str(x or '').strip() for x in next(it)] ix = {h: i for i, h in enumerate(H)} sku_amt, sku_cust, sku_name = defaultdict(float), defaultdict(lambda: defaultdict(float)), {} for r in it: sku = str(r[ix['货号#']] or '').strip() cust = str(r[ix['客户']] or '').strip() if not sku or not cust: continue if str(r[ix['单据状态']]) != '已审核' or str(r[ix['单据类型']]) != '标准销售订单': continue if str(r[ix['是否赠品']] or '') in ('是', 'Y', '1') or any(k in cust for k in INTERNAL): continue a = float(r[ix['价税合计']] or 0) if a <= 0: continue sku_amt[sku] += a sku_cust[sku][match(cust)] += a nm = str(r[ix['物料名称']] or '').strip() if len(nm) > len(sku_name.get(sku, '')): sku_name[sku] = nm # 附件2 资料 f2ws = openpyxl.load_workbook(F2, data_only=True)['覆盖矩阵'] f2 = {} for i in range(3, f2ws.max_row + 1): s = str(f2ws.cell(i, 2).value or '').strip() if s: f2[re.sub(r'\s+', '', s)] = {'docs': int(f2ws.cell(i, 4).value or 0), 'core': {CORE[j]: str(f2ws.cell(i, 5 + j).value or '').strip() for j in range(4)}} # 货号 → 系列键 bykey = defaultdict(list) for sku in sku_amt: bykey[code_key(sku)].append(sku) for sku in f2: if sku not in sku_amt: bykey[code_key(sku)].append(sku) # 代表货号与系列标签 key_label, key_rep = {}, {} used = Counter() f2_upper = {k.upper() for k in f2} PROC = re.compile(r'-(P\d+|R\d+|O\d+|S\d+|V\d+|T\d+|SCS\w*|FPO)\b.*$') for k, skus in bykey.items(): def label_of(s): x = re.sub(r'\s+', '', s).upper() x = PROC.sub('', x) x = re.sub(r'-\d{6}.*$', '', x) x = re.sub(r'-\d$', '', x) return x cands = [s for s in skus if re.sub(r'\s+', '', s).upper() in f2_upper] pool = cands or skus plain = [s for s in pool if not re.match(r'^[PF][A-Z]', s.upper())] base = max(plain or pool, key=lambda s: sku_amt.get(s, 0)) lab = label_of(base) if lab in used: used[lab] += 1 lab = '%s-%d' % (lab, used[lab]) key_label[k], key_rep[k] = lab, base # SKU 明细 rows = [] for k, skus in bykey.items(): for sku in skus: cs = sku_cust.get(sku, {}) tot = sum(cs.values()) or sku_amt.get(sku, 0) top = max(cs.items(), key=lambda x: x[1]) if cs else ('', 0) nm = sku_name.get(sku, '') cat = ('动物诊断' if re.match(r'^(AHM|AHRM)', sku) else '客户定制' if (re.match(r'^(2FM|2AM|A3$)', sku) or '核酸检测试剂' in nm or re.search(r'-SCS', sku.upper())) else '标准') rows.append({'系列': key_label[k], '代表货号': key_rep[k], '货号': sku, '名称': nm, '类型': kind(nm), 'X数': xfold(nm), '组分': '组分' if re.search(r'-\d$', sku) else '成品', 'DG': 'Y' if (nm and ('(DG)' in nm or '(DG)' in nm) or re.search(r'-DG', sku, re.I)) else '', 'NA': 'Y' if (re.search(r'\(NA\)', nm or '') or re.search(r'-NA', sku, re.I)) else '', '类别': cat, '销售额': round(tot, 2), '客户数': len(cs), 'TOP1': top[0], 'TOP1额': round(top[1], 2), 'TOP1占比': (top[1] / tot * 100) if tot else 0}) # 系列汇总 byseries = defaultdict(list) for r in rows: byseries[r['系列']].append(r) srows = [] for lab, rs in byseries.items(): tot = sum(r['销售额'] for r in rs) finished = [r for r in rs if r['组分'] == '成品'] comps = [r for r in rs if r['组分'] == '组分'] sole = [r for r in rs if r['客户数'] == 1] top = max(rs, key=lambda r: r['TOP1额']) custs = {r['TOP1'] for r in rs if r['TOP1']} docs = sum(docs_of(r['货号'], f2)['docs'] for r in rs) have = set() for r in rs: for kk, vv in docs_of(r['货号'], f2)['core'].items(): if vv and vv not in ('0', ''): have.add(kk) miss = [c for c in CORE if c not in have] kd = Counter(r['类型'] for r in (finished or comps)).most_common(1)[0][0] if (finished or comps) else '其他' cat = Counter(r['类别'] for r in rs).most_common(1)[0][0] fam_amt = defaultdict(float) for r in rs: if r['名称']: fam_amt[family(r['名称'])] += r['销售额'] fam_name = max(fam_amt.items(), key=lambda x: x[1])[0] if fam_amt else '未命名' srows.append({'代表货号': lab, '产品系列': fam_name, '类别': cat, '类型': kd, '主货号数': len({re.sub(r'-\d{6}.*$', '', r['货号']) for r in rs}), '成品数': len(finished), '组分数': len(comps), 'X数': '、'.join(sorted({r['X数'] for r in rs if r['X数']})) or '—', 'DG': 'Y' if any(r['DG'] for r in rs) else '', 'NA': 'Y' if any(r['NA'] for r in rs) else '', '销售额': round(tot, 2), '客户数': len(custs), '单客货号数': len(sole), '单客金额占比': (sum(r['销售额'] for r in sole) / tot * 100) if tot else 0, 'TOP1客户': top['TOP1'], 'TOP1占比': top['TOP1额'] / tot * 100 if tot else 0, '资料份数': docs, '资料状态': '未收录' if docs == 0 else ('齐全' if not miss else '部分'), '缺失核心项': '、'.join(miss) or '无', 'DG专项': DG_PARAMS if any(r['DG'] for r in rs) else '', 'NA专项': NA_PARAMS if any(r['NA'] for r in rs) else ''}) srows.sort(key=lambda r: -r['销售额']) wb2 = openpyxl.Workbook() Hf, Ff = Font(color='FFFFFF', bold=True), PatternFill('solid', fgColor='1F4E79') ws1 = wb2.active ws1.title = '系列总表' h1 = ['代表货号', '产品系列', '类别', '类型', '主货号数', '成品数', '组分数', 'X数', 'DG', 'NA', '2026销售额', '客户数', '单客货号数', '单客金额占比%', 'TOP1客户', 'TOP1占比%', '客户集中度', '资料份数', '资料状态', '缺失核心项', '资料路径建议'] ws1.append(h1) for r in srows: lvl = ('单一客户' if r['客户数'] == 1 else '高度集中' if r['TOP1占比'] >= 70 else '较集中' if r['TOP1占比'] >= 40 else '分散') path = '客户定制资料包' if (r['单客金额占比'] >= 80 and r['类别'] == '客户定制') else ( '客户定制资料包' if r['单客金额占比'] >= 80 else '标准品资料') ws1.append([r['代表货号'], r['产品系列'], r['类别'], r['类型'], r['主货号数'], r['成品数'], r['组分数'], r['X数'], r['DG'], r['NA'], r['销售额'], r['客户数'], r['单客货号数'], round(r['单客金额占比'], 1), r['TOP1客户'], round(r['TOP1占比'], 1), lvl, r['资料份数'], r['资料状态'], r['缺失核心项'], path]) for c in ws1[1]: c.font = Hf; c.fill = Ff; c.alignment = Alignment(horizontal='center', wrap_text=True) for i, w in enumerate([13, 30, 9, 8, 8, 7, 7, 12, 5, 5, 13, 8, 10, 11, 28, 9, 10, 9, 9, 30, 16], 1): ws1.column_dimensions[get_column_letter(i)].width = w ws1.freeze_panes = 'C2' ws2 = wb2.create_sheet('变体明细') h2 = ['系列', '代表货号', '货号', '物料名称', '类型', 'X数', '组分', 'DG', 'NA', '类别', '2026销售额', '客户数', 'TOP1客户', 'TOP1占比%'] ws2.append(h2) for r in sorted(rows, key=lambda x: (-x['销售额'], x['系列'])): ws2.append([r['系列'], r['代表货号'], r['货号'], r['名称'], r['类型'], r['X数'], r['组分'], r['DG'], r['NA'], r['类别'], r['销售额'], r['客户数'], r['TOP1'], round(r['TOP1占比'], 1)]) for c in ws2[1]: c.font = Hf; c.fill = Ff; c.alignment = Alignment(horizontal='center', wrap_text=True) for i, w in enumerate([14, 12, 26, 48, 8, 7, 7, 5, 5, 9, 13, 8, 28, 10], 1): ws2.column_dimensions[get_column_letter(i)].width = w ws2.freeze_panes = 'C2' ws3 = wb2.create_sheet('系列缺口任务清单') h3 = ['任务ID', '代表货号', '产品系列', '类别', '成品数', '2026销售额(万)', '客户数', '单客金额占比%', '资料份数', '缺失核心项', 'DG/NA专项', '优先级', '建议动作', '责任部门', '预计人日', '计划开始', '计划完成'] ws3.append(h3) n = 0 for r in srows: if r['缺失核心项'] == '无' or r['类别'] != '标准' or r['销售额'] < 100000: continue n += 1 sales = r['销售额'] pri = 'P0' if sales >= 1000000 else ('P1' if sales >= 300000 else 'P2') sp = [] if r['DG']: sp.append('DG可冻干') if r['NA']: sp.append('NA低背景') days = 2 * len(r['缺失核心项'].split('、')) + (2 if r['资料份数'] == 0 else 0) + 2 * len(sp) act = ('先建系列档案，再补：' if r['资料份数'] == 0 else '补充：') + r['缺失核心项'] if sp: act += '；另需' + '、'.join(sp) + '专项参数' ws3.append([f'T{n:03d}', r['代表货号'], r['产品系列'], r['类别'], r['成品数'], round(sales / 1e4, 1), r['客户数'], round(r['单客金额占比'], 1), r['资料份数'], r['缺失核心项'], '、'.join(sp) or '—', pri, act, '产品部/研发', days, '', '']) for c in ws3[1]: c.font = Hf; c.fill = Ff; c.alignment = Alignment(horizontal='center', wrap_text=True) for i, w in enumerate([8, 13, 30, 9, 7, 13, 8, 11, 9, 28, 16, 7, 48, 12, 8, 11, 11], 1): ws3.column_dimensions[get_column_letter(i)].width = w ws3.freeze_panes = 'A2' ws4 = wb2.create_sheet('单一客户货号') ws4.append(['系列', '货号', '物料名称', '类别', '唯一客户', '2026销售额']) for r in sorted([x for x in rows if x['客户数'] == 1], key=lambda x: -x['销售额']): ws4.append([r['系列'], r['货号'], r['名称'], r['类别'], r['TOP1'], r['销售额']]) for c in ws4[1]: c.font = Hf; c.fill = Ff; c.alignment = Alignment(horizontal='center', wrap_text=True) for i, w in enumerate([14, 26, 48, 9, 30, 13], 1): ws4.column_dimensions[get_column_letter(i)].width = w ws4.freeze_panes = 'A2' ws5 = wb2.create_sheet('口径说明') for line in [ ['归集方法', '货号编码法：剥 P/F 前缀、X 档位（M2181=2×/M0181=10×、MD2104=2×/MD5104=5×）、6 位配方号、形态(-NA/-DG/-FP…)与组分(-1…-6)后缀，同键同系列'], ['系列命名', '同键内取 2× 基准、无 F/P 前缀、长度最短、销售额最高者为代表货号，再去掉代次与配方号'], ['类别', '标准 / 客户定制（2FM、A3）/ 动物诊断（AHM、AHRM）；后两类不派标准品资料任务'], ['客户口径', '《客户清单_官方_20260930》1635 家归一；单一客户 = 该货号 2026 年仅 1 家采购'], ['集中度', 'TOP1占比 ≥70% 高度集中、≥40% 较集中；单客金额占比 ≥80% 建议走客户定制资料包'], ['资料口径', '附件2 覆盖矩阵 56 个货号 / 167 份资料，按前缀映射到系列；四项核心=冻融/加速/效期/运输'], ['优先级', 'P0 ≥100 万、P1 ≥30 万、P2 其余'], ['待确认', '责任部门、人日与时间节点待定，确认后出甘特图'], ]: ws5.append(line) ws5.column_dimensions['A'].width = 14 ws5.column_dimensions['B'].width = 130 wb2.save(OUT) std = [r for r in srows if r['类别'] == '标准'] print('系列 %d（标准 %d / 定制 %d / 动物诊断 %d）｜货号 %d' % ( len(srows), len(std), sum(1 for r in srows if r['类别'] == '客户定制'), sum(1 for r in srows if r['类别'] == '动物诊断'), len(rows))) print('销售额 %.0f 万 ｜任务 %d ｜人日 %.0f' % ( sum(r['销售额'] for r in srows) / 1e4, n, sum(float(x[14] or 0) for x in ws3.iter_rows(min_row=2, values_only=True)))) print('\nTOP12 系列:') for r in srows[:12]: print(' %-12s %-30s 客户%3d 单客%3d %5.1f%% TOP1 %-14s %5.1f%% %7.1f万 %s' % ( r['代表货号'], r['产品系列'][:30], r['客户数'], r['单客货号数'], r['单客金额占比'], (r['TOP1客户'] or '')[:14], r['TOP1占比'], r['销售额'] / 1e4, r['资料状态'])) print('✅', OUT) if __name__ == '__main__': main()