# -*- coding: utf-8 -*- """ 宝锐分子诊断产品清单：父项 → 子项 完整映射（品名 / 货号 / 物料编码 三层齐全） 数据源（唯一权威件）：数据分析/分子检测类试剂产品清单-YYYYMMDD.xlsx ① 产品清单-生产部内容 25 列：父项（产品编码/名称/货号/对内货号/生产货号/规格+单位/有效期） + 子项（组分编码/旧编码/货号/对内货号/生产货号/名称/单位/用量/规格） ② 产品清单-分子试剂产品部内容 23 列：父项（编码/名称/货号/对内货号/规格）+ 子项（编码/货号/对内货号/名称/规格/装量/单位） ③ 产品清单-科研产品 同上 ④ 研发对产品目录修改记录 变更台账（不进映射） 口径要点（来自 baoyu-erp-product-catalog 技能）： - 三套编码并存：产品货号 / 产品对内货号 / 产品生产货号 —— 全部保留，不做归一，别名层才能查得到。 - 父项字段是「合并单元格」：只 ffill 父项**身份列**（编码/名称/货号/对内货号/生产货号/规格/单位）。 - **不要 ffill 销售分组**：该列会跨产品块填充，会把 LAMP 类错标成直扩 qRT-PCR。BOM分组同理按「父项块首行取值」处理。 - 子项可能没有货号 → 用「组分编码」（物料编码）做唯一标识。 产物：数据分析/宝锐分子诊断产品清单_父项子项映射_YYYYMMDD.xlsx Sheet1 父项-子项全映射 一行一对，可直接 vlookup Sheet2 父项清单 唯一父项（货号+物料编码） Sheet3 子项清单 唯一子项（货号+物料编码）+ 被哪些父项引用 Sheet4 数据说明与缺口 覆盖率 / 缺失清单 / 口径说明 """ import os, re, glob, io, json, warnings from datetime import datetime import pandas as pd warnings.filterwarnings('ignore') WORK = os.path.expanduser('~/Desktop/Hermes输出-工作类') SRC_DIR = os.path.join(WORK, '经营分析', '数据分析') PARENT_FILL = ['产品编码', '产品名称', '产品货号', '产品对内货号', '产品生产货号', '产品规格型号'] BLOCK_COLS = ['BOM分组', '销售分组'] # 合并单元格但会跨块 → 只取块首行，不 ffill OUT_COLS = ['父项物料编码', '父项货号', '父项对内货号', '父项生产货号', '父项品名', '父项规格', '父项单位', 'BOM分组', '销售分组', '项次', '子项物料编码', '子项旧编码', '子项货号', '子项对内货号', '子项生产货号', '子项品名', '子项存货名称', '子项规格', '子项用量', '子项单位', '是否已转产', '有效期/月', '来源Sheet', 'Excel行号'] def _s(v): """统一成字符串：NaN/None → ''，去掉首尾空格，干掉 Excel 里常见的全角空格""" if v is None: return '' if isinstance(v, float) and pd.isna(v): return '' t = str(v).strip().replace('\u3000', ' ') return '' if t.lower() in ('nan', 'none', 'nat') else t def _num(v): """数值列容错：非数值（如「根据客户订单量和分装要求」）保留原文，别丢信息""" if v is None or (isinstance(v, float) and pd.isna(v)): return '' s = str(v).strip() if s.lower() in ('nan', 'none', ''): return '' try: f = float(s) return int(f) if f == int(f) else f except (TypeError, ValueError): return s def find_src(): fs = [f for f in glob.glob(os.path.join(SRC_DIR, '分子检测类试剂产品清单-*.xlsx'))] if not fs: raise SystemExit('未找到源件：数据分析/分子检测类试剂产品清单-*.xlsx') fs.sort(key=lambda p: (re.search(r'-(\d{8})\.xlsx$', p) or [None, '0'])[1]) return fs[-1] def load_sheet(path, sn): """读一个 Sheet，返回标准化后的 DataFrame（父项块首行的 BOM/销售分组单独挂回父项）""" df = pd.read_excel(path, sheet_name=sn, dtype=object) df.columns = [_s(c) for c in df.columns] n0 = len(df) # 父项块首行 = 产品货号非空的行（其余行是子项续行） first = df['产品货号'].map(_s) != '' blocks = {} for i, r in df[first].iterrows(): key = (_s(r.get('产品货号')), _s(r.get('产品编码'))) blocks[key] = {c: _s(r.get(c)) for c in BLOCK_COLS if c in df.columns} for c in PARENT_FILL: if c in df.columns: df[c] = df[c].ffill() for c in PARENT_FILL + ['产品单位']: if c not in df.columns: df[c] = '' # 逐行取块首行的 BOM分组/销售分组 cur = {} bom, sale = [], [] for _, r in df.iterrows(): if _s(r.get('产品货号')) != '': cur = blocks.get((_s(r.get('产品货号')), _s(r.get('产品编码'))), {}) bom.append(cur.get('BOM分组', '')) sale.append(cur.get('销售分组', '')) out = pd.DataFrame({ '父项物料编码': df['产品编码'].map(_s), '父项货号': df['产品货号'].map(_s), '父项对内货号': df['产品对内货号'].map(_s) if '产品对内货号' in df else '', '父项生产货号': df['产品生产货号'].map(_s) if '产品生产货号' in df else '', '父项品名': df['产品名称'].map(_s), '父项规格': df['产品规格型号'].map(_s) if '产品规格型号' in df else '', '父项单位': df['产品单位'].map(_s) if '产品单位' in df else '', 'BOM分组': bom, '销售分组': sale, '项次': df['项次'].map(_num) if '项次' in df else '', '子项物料编码': df['组分编码'].map(_s) if '组分编码' in df else '', '子项旧编码': df['组分旧编码'].map(_s) if '组分旧编码' in df else '', '子项货号': df['组分货号'].map(_s) if '组分货号' in df else '', '子项对内货号': df['组分对内货号'].map(_s) if '组分对内货号' in df else '', '子项生产货号': df['组分生产货号'].map(_s) if '组分生产货号' in df else '', '子项品名': df['组分名称'].map(_s) if '组分名称' in df else '', '子项存货名称': df['子项存货名称(标记不一样名称)'].map(_s) if '子项存货名称(标记不一样名称)' in df else '', '子项规格': df['组分规格型号'].map(_s) if '组分规格型号' in df else '', '子项用量': df['子项用量'].map(_num) if '子项用量' in df else (df['装量'].map(_num) if '装量' in df else ''), '子项单位': df['组分单位'].map(_s) if '组分单位' in df else (df['单位'].map(_s) if '单位' in df else ''), '是否已转产': df['是否已经转产'].map(_s) if '是否已经转产' in df else '', '有效期/月': df['有效期/月'].map(_num) if '有效期/月' in df else '', '来源Sheet': sn, 'Excel行号': list(range(2, n0 + 2)), }) # 父项货号/编码都没有的行（表尾说明行、空行）丢掉 out = out[(out['父项货号'] != '') | (out['父项物料编码'] != '')] return out, n0 def main(): src = find_src() print('源件：%s' % os.path.basename(src)) sheets = ['产品清单-生产部内容', '产品清单-分子试剂产品部内容', '产品清单-科研产品'] xl = pd.ExcelFile(src) sheets = [s for s in sheets if s in xl.sheet_names] print('参与映射的 Sheet：%s' % ' / '.join(sheets)) frames, raw_n = [], 0 for sn in sheets: d, n = load_sheet(src, sn) raw_n += n print(' %-28s 原始 %4d 行 → 映射 %4d 行 ｜ 父项 %3d ｜ 子项 %3d' % (sn, n, len(d), d['父项货号'].nunique(), d['子项货号'].nunique())) frames.append(d) df = pd.concat(frames, ignore_index=True) df = df.drop_duplicates(subset=['父项货号', '父项物料编码', '子项货号', '子项物料编码', '项次', '来源Sheet'], keep='first').reset_index(drop=True) df = df[OUT_COLS] # 覆盖率 n = len(df) def cov(col): return int((df[col] != '').sum()) stats = { '源件': os.path.basename(src), 'Sheet数': len(sheets), '原始行': raw_n, '映射行(父项×子项)': n, '父项数(货号+编码)': df.groupby(['父项货号', '父项物料编码']).ngroups, '父项货号唯一': df['父项货号'].nunique(), '子项数(货号+编码)': df.groupby(['子项货号', '子项物料编码']).ngroups, '子项货号唯一': df['子项货号'].nunique(), '父项物料编码覆盖': '%d/%d' % (cov('父项物料编码'), n), '子项物料编码覆盖': '%d/%d' % (cov('子项物料编码'), n), '子项货号覆盖': '%d/%d' % (cov('子项货号'), n), '子项品名覆盖': '%d/%d' % (cov('子项品名'), n), '子项规格覆盖': '%d/%d' % (cov('子项规格'), n), 'BOM分组覆盖': '%d/%d' % (cov('BOM分组'), n), } # 父项清单 par = (df.groupby(['父项货号', '父项物料编码'], as_index=False) .agg(父项品名=('父项品名', 'first'), 父项对内货号=('父项对内货号', 'first'), 父项规格=('父项规格', 'first'), 父项单位=('父项单位', 'first'), BOM分组=('BOM分组', 'first'), 销售分组=('销售分组', 'first'), 子项数=('子项物料编码', 'size'), 子项物料编码=('子项物料编码', lambda x: '、'.join(sorted({i for i in x if i}))), 来源Sheet=('来源Sheet', lambda x: '/'.join(sorted(set(x))))) .sort_values(['父项货号', '父项物料编码'])[ ['父项货号', '父项物料编码', '父项品名', '父项规格', '父项单位', '父项对内货号', 'BOM分组', '销售分组', '子项数', '子项物料编码', '来源Sheet']]) if '销售分组' in par.columns: par['销售分组'] = par['销售分组'].replace({'基础物料数据（勿选）': '（勿选）'}) # 子项清单 chi = (df.groupby(['子项货号', '子项物料编码'], as_index=False) .agg(子项品名=('子项品名', 'first'), 子项对内货号=('子项对内货号', 'first'), 子项规格=('子项规格', 'first'), 子项单位=('子项单位', 'first'), 被引用次数=('父项货号', 'size'), 所属父项=('父项货号', lambda x: '、'.join(sorted(set(x))[:8])), 来源Sheet=('来源Sheet', lambda x: '/'.join(sorted(set(x)))))) # 缺口清单 gaps = df[(df['子项物料编码'] == '') | (df['子项货号'] == '') | (df['子项品名'] == '')] gaps = gaps[['来源Sheet', 'Excel行号', '父项货号', '父项物料编码', '父项品名', '子项货号', '子项物料编码', '子项品名']] note = pd.DataFrame({'项目': list(stats.keys()), '值': [str(v) for v in stats.values()]}) note = pd.concat([note, pd.DataFrame({ '项目': ['—— 口径说明 ——', '', '父项物料编码', '父项货号', '子项物料编码', '子项货号', '父项物料编码=子项物料编码', '本项目已含', '未含'], '值': ['', '', '源表「产品编码」= ERP / MES 侧物料编码', '源表「产品货号」= 业务主货号（对外）', '源表「组分编码」= 子项物料编码', '源表「组分货号」= 子项货号，空则只能靠物料编码识别', '单组分产品（父项即子项）会出现两列同值，属正常', '生产部 / 分子试剂产品部 / 科研产品 三个 Sheet 的父项—子项映射', '研发变更台账 Sheet（只记手工改动，非全量台账）、价格、库存、客户专属后缀（-DB/-TP/-NA 等）需回 ERP 取'], })], ignore_index=True) stamp = re.search(r'-(\d{8})\.xlsx$', src).group(1) out_path = os.path.join(SRC_DIR, '宝锐分子诊断产品清单_父项子项映射_%s.xlsx' % stamp) with pd.ExcelWriter(out_path, engine='openpyxl') as w: df.to_excel(w, sheet_name='父项-子项全映射', index=False) par.to_excel(w, sheet_name='父项清单', index=False) chi.to_excel(w, sheet_name='子项清单', index=False) note.to_excel(w, sheet_name='数据说明与缺口', index=False) for ws in w.book.worksheets: # 冻结首行 + 筛选器 + 列宽 ws.freeze_panes = 'A2' ws.auto_filter.ref = ws.dimensions for col in ws.columns: L = max((len(_s(c.value)) for c in col[:200]), default=8) ws.column_dimensions[col[0].column_letter].width = min(max(L * 1.5 + 2, 9), 46) print() print('═══ 产物 ═══') print(' %s' % out_path) print(' 大小 %.2f MB' % (os.path.getsize(out_path) / 1048576.0)) print() for k, v in stats.items(): print(' %-22s %s' % (k, v)) print() print(' 子项编码/货号/品名 缺口行：%d' % len(gaps)) # JSON 侧车（供工作台/检索用） js = { 'generated': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'source': os.path.basename(src), 'stats': stats, 'parents': par.to_dict('records'), 'children': chi.to_dict('records'), 'pairs': df[['父项货号', '父项物料编码', '子项货号', '子项物料编码']].to_dict('records'), } jp = os.path.join(WORK, '数据', 'product_parent_child_map.json') io.open(jp, 'w', encoding='utf-8').write(json.dumps(js, ensure_ascii=False)) print(' JSON 侧车：%s（%.2f MB）' % (jp, os.path.getsize(jp) / 1048576.0)) # ── 紧凑索引：供「历史成交价查询」等前端按货号/物料编码 O(1) 反查 ── # children_of[货号] / children_of_key[货号||编码]：该货号（该规格）由哪些子项组成 # parents_of[货号] / parents_of_key[货号||编码] ：该子项（该规格）属于哪些父项 kids_of, kids_of_key = {}, {} par_of, par_of_key = {}, {} for r in df.to_dict('records'): ps, pc = r['父项货号'], r['父项物料编码'] cs, cc = r['子项货号'], r['子项物料编码'] kid = {'c': cc, 's': cs, 'n': r['子项品名'], 'sp': r['子项规格'], 'q': r['子项用量'], 'u': r['子项单位'], 'pc': pc, 'ps': ps, 'psp': r['父项规格']} for key in filter(None, [ps, '%s||%s' % (ps, pc) if (ps and pc) else '']): tgt = kids_of if key == ps else kids_of_key row = tgt.setdefault(key, []) if not any((x['c'] == cc and x['s'] == cs and str(x['q']) == str(kid['q'])) for x in row): row.append(kid) par = {'ps': ps, 'pc': pc, 'pn': r['父项品名'], 'pp': r['父项规格'], 'bom': r['BOM分组'], 'sale': r['销售分组']} for key in filter(None, [cs, '%s||%s' % (cs, cc) if (cs and cc) else '']): tgt = par_of if key == cs else par_of_key row = tgt.setdefault(key, []) if not any((x['ps'] == ps and x['pc'] == pc) for x in row): row.append(par) bom = { 'generated': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'source': os.path.basename(src), 'n_pairs': n, 'n_parent_sku': len(kids_of), 'n_child_sku': len(par_of), 'children_of': kids_of, 'children_of_key': kids_of_key, 'parents_of': par_of, 'parents_of_key': par_of_key, } bp = os.path.join(WORK, '数据', 'product_bom_map.json') io.open(bp, 'w', encoding='utf-8').write(json.dumps(bom, ensure_ascii=False)) print(' BOM 索引：%s（%.2f MB ｜ 父项 %d ｜ 子项 %d ｜ 精准键 %d/%d）' % (bp, os.path.getsize(bp) / 1048576.0, len(kids_of), len(par_of), len(kids_of_key), len(par_of_key))) if __name__ == '__main__': main()