#!/usr/bin/env python3 # -*- coding: utf-8 -*- """为销售助手生成「按人切分的数据包」—— 只有本人客户范围内的全量数据。 设计原则：数据隔离靠**物理切分**，不靠提示词自觉。 · 每个销售员一个目录：团队共享/销售数据包/<销售员>/ · 目录里只有**属于本人客户**的记录；别人的客户数据根本不在文件里 · 覆盖：飞书活动 / 客诉 / 订单明细（含单号·单价）/ 成交价历史 / 客户特殊要求 / 采购汇总 用法： python3 build_sales_data_pack.py 刘子研 # 单人 python3 build_sales_data_pack.py --all # 全部有数据的销售员 """ import json import os import re import shutil import sys from datetime import datetime import pandas as pd ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' DATA = os.path.join(ROOT, '数据') OUT = os.path.join(ROOT, '团队共享/销售数据包') ERP = os.path.expanduser('~/Desktop/hermas输入-工作台数据库') ATTRIB = os.path.expanduser('~/.hermes/profiles/baorui-sales/customer_attribution.json') MERGE = {'大客户1': '刘子研', '大客户2': '赵云浩', '黄明月': '刘子研', '张立娅': '刘子研', '龙友波': '黄文才', '黄建佳': '韩远怀', '黄自鑫': '韩远怀', '李岚风': '韩远怀', '段玉': '韩远怀', '王慧洁': '吴云', '刘静文': '吴云', '刘斌': '吴云', '陈彩红': '吴云', '梁莹莹': '吴云', '赵以诺': '吴云', '曾琦': '吴云', '曾琪': '吴云', '肖立': '吴云'} def keywords(names): """把归属表里的客户/集团名转成匹配关键词（用于匹配数据里的公司全名）。""" kws = set() for n in names: n = str(n).strip() if not n: continue kws.add(n) base = re.sub(r'集团$', '', n) base = re.sub(r'（.*?）', '', base) if len(base) >= 2: # 中文品牌名 2 字是常态（天隆/丽珠/复星），不能按 3 字卡 kws.add(base) # 去掉常见地域前缀，留核心品牌词（如 深圳市新产业生物医学工程股份有限公司 → 新产业生物） m = re.match(r'^(?:深圳市|珠海市|广州市|上海市|北京市|杭州市|苏州市|西安市|宁波市|苏州|杭州|宁波|上海|北京|广州|西安)?(.*)$', base) if m and len(m.group(1)) >= 3: kws.add(m.group(1)) return kws def matches(name, kws): name = str(name) return any(k in name for k in kws) def load_erp(): frames = [] for p, y in ((os.path.join(ERP, '2024和2025历史数据/2024全年订单.xlsx'), '2024'), (os.path.join(ERP, '2024和2025历史数据/2025全年订单.xlsx'), '2025'), (os.path.join(ERP, '月度销售额数据/销售订单_20260101-20260930.xlsx'), '2026')): d = pd.ExcelFile(p).parse('Sheet1') d.columns = [str(c).strip() for c in d.columns] d['_year'] = y frames.append(d) d = pd.concat(frames, ignore_index=True) d['_sp'] = d['销售员'].astype(str).str.strip().map(lambda x: MERGE.get(x, x)) d['_c'] = d['客户'].astype(str).str.strip() return d def dump(path, obj): os.makedirs(os.path.dirname(path), exist_ok=True) json.dump(obj, open(path, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) def build(person, erp, attrib): names = list((attrib['salespeople'].get(person) or {}).get('customers', [])) own_from_erp = sorted({c for c in erp.loc[erp['_sp'] == person, '_c'] if c and c != 'nan'}) # 严格只用「客户清单」里的名字（含集团关键词），不因为订单经手过就扩范围 kws = keywords(names) d = os.path.join(OUT, person) if os.path.isdir(d): shutil.rmtree(d) os.makedirs(d, exist_ok=True) stat = {} # ① 飞书活动 act = json.load(open(os.path.join(DATA, 'activities_feishu.json'), encoding='utf-8'))['items'] hit = [a for a in act if matches(a.get('customer'), kws)] dump(os.path.join(d, '01_飞书活动.json'), { '_说明': '本人客户范围内的飞书活动记录（全量字段）', '_条数': len(hit), '_口径': '严格按客户清单匹配：客户名命中本人客户/集团；不因「本人提交」而放行范围外客户', 'items': hit}) stat['飞书活动'] = len(hit) # ② 客诉 comp = json.load(open(os.path.join(DATA, 'mcp_complaints.json'), encoding='utf-8'))['items'] hit = [c for c in comp if matches(c.get('customer'), kws)] dump(os.path.join(d, '02_客诉.json'), { '_说明': '本人客户范围内的客诉记录（含紧急度/天数/状态/处理人）', '_条数': len(hit), '_口径': '严格按客户清单匹配；不因「本人创建/处理」而放行范围外客户', 'items': hit}) stat['客诉'] = len(hit) # ③ 订单明细（含单号、单价 —— 按用户要求对本人客户开放） sub = erp[erp['_c'].apply(lambda x: matches(x, kws))] cols = [c for c in ('合同日期', '单据类型', '单据编号', '单据状态', '客户', '销售员', '物料编码', '物料名称', '销售单位', '销售数量', '含税单价', '税率%', '价税合计', '是否赠品', '合同编号#', '货号#', '出货批号#', '产品大类#') if c in sub.columns] rows = sub[cols].copy() rows['_year'] = sub['_year'].values dump(os.path.join(d, '03_订单明细.json'), { '_说明': '本人客户范围内的订单明细（含单据编号、含税单价、批号）', '_条数': int(len(rows)), '_口径': '严格按客户清单匹配；2024/2025 全年 + 2026 1-9 月', 'items': json.loads(rows.to_json(orient='records', force_ascii=False))}) stat['订单明细'] = int(len(rows)) # ④ 成交价历史 ph = json.load(open(os.path.join(DATA, 'customer_price_history.json'), encoding='utf-8')) hist = {c: v for c, v in ph['hist'].items() if matches(c, kws)} dump(os.path.join(d, '04_成交价历史.json'), { '_说明': '本人客户的历史成交价（客户×货号×体系，含最近价/均价/成交明细）', '_客户数': len(hist), '_口径': '按客户名匹配本人客户/集团；源 customer_price_history.json（%s）' % ph.get('generated', ''), 'hist': hist}) stat['成交价客户数'] = len(hist) # ⑤ 客户特殊要求 sr = json.load(open(os.path.join(DATA, 'customer_special_requirements.json'), encoding='utf-8')) hit = [i for i in sr['items'] if matches(i.get('customer'), kws)] dump(os.path.join(d, '05_客户特殊要求.json'), { '_说明': '本人客户范围内的特殊要求（搭配规则/配方体系/包装标签/交付物流/质检文件/价格账期等）', '_条数': len(hit), '_口径': '严格按客户清单匹配', 'categories': sr.get('categories'), 'items': hit}) stat['特殊要求'] = len(hit) # ⑥ 采购汇总（复用按人汇总文件） src = os.path.join(ROOT, '团队共享/客户采购汇总/%s.json' % person) if os.path.exists(src): shutil.copy2(src, os.path.join(d, '06_采购汇总.json')) stat['采购汇总'] = '已含' dump(os.path.join(d, '_索引.json'), { '_说明': '本数据包只包含「%s」本人客户范围内的数据。其他销售员的客户数据不在本包内。' % person, '_授权': '仅供「%s」本人查看；不得读取其他人的数据包目录。' % person, '_生成时间': datetime.now().strftime('%Y-%m-%d %H:%M'), '_客户关键词数': len(kws), '_客户清单': names, '文件': stat, }) return stat def main(): attrib = json.load(open(ATTRIB, encoding='utf-8')) erp = load_erp() args = sys.argv[1:] people = ([p for p in attrib['salespeople'] if p not in ('刘新元',)] if (not args or args[0] == '--all') else args) os.makedirs(OUT, exist_ok=True) for p in people: if p not in attrib['salespeople']: print('跳过（归属表无此人）：', p); continue st = build(p, erp, attrib) print('✅ %-8s %s' % (p, st)) if __name__ == '__main__': main()