#!/usr/bin/env python3 # -*- coding: utf-8 -*- """客户 2 年同比（集团口径，无集团按客户）—— 大客户部为主 口径：2026 前三季度(1-9月) vs 2025 同期；标准销售订单 + 非赠品 + 剔除宝锐系 集团：CRM 权威映射（数据/pricing/_key_customers.json 的 group）优先；无集团 → 按客户本体 部门：客户 → 销售员（2026 归属优先，回退 2025 销售员）→ 人员表部门 校验：自算 2025 逐月金额 必须等于文件自带透视表 Sheet8 的值 输出：数据/客户2年同比_集团口径.json """ import io import json import warnings from collections import defaultdict warnings.filterwarnings('ignore') import openpyxl # noqa: E402 W = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' H = '/Users/liuxinyuan/Desktop/hermas输入-工作台数据库/2024和2025历史数据' CUR = '/Users/liuxinyuan/Desktop/hermas输入-工作台数据库/月度销售额数据/销售订单_20260101-20261015.xlsx' MONTHS = ['2025-01', '2025-02', '2025-03', '2025-04', '2025-05', '2025-06', '2025-07', '2025-08', '2025-09'] MONTHS26 = ['2026-%02d' % m for m in range(1, 10)] BAORUI = ('宝锐', '宝泰', '横琴') def sf(v): try: s = str(v).strip() if s in ('', '-', 'None', 'nan'): return None return float(s.replace(',', '')) except Exception: return None def ok_row(dtype, gift, cust): dt = str(dtype or '') if '标准销售订单' not in dt and '标准' not in dt: return False if str(gift or '').strip() == '是': return False if any(k in str(cust or '') for k in BAORUI): return False return True # ── ① 2025（历史文件 Sheet1）── wb = openpyxl.load_workbook(H + '/2025全年订单.xlsx', data_only=True, read_only=True) ws = wb['Sheet1'] hdr = None p25 = defaultdict(lambda: defaultdict(float)) # 客户 → 月 → 金额 m25 = defaultdict(float) # 月 → 金额（校验用） sp25 = {} # 客户 → 销售员（全年金额最大者） sp25amt = defaultdict(float) for i, row in enumerate(ws.iter_rows(min_row=1, values_only=True), 1): if hdr is None: hdr = [str(x or '').strip() for x in row] ix = {k: hdr.index(k) for k in ('合同月份', '客户', '销售员', '价税合计', '是否赠品', '单据类型') if k in hdr} continue if '客户' not in ix: break mon = str(row[ix['合同月份']] or '').strip()[:7] if mon not in MONTHS: continue cust = str(row[ix['客户']] or '').strip() if not ok_row(row[ix['单据类型']] if '单据类型' in ix else '', row[ix['是否赠品']] if '是否赠品' in ix else '', cust): continue amt = sf(row[ix['价税合计']]) or 0.0 p25[cust][mon] += amt m25[mon] += amt s = str(row[ix['销售员']] or '').strip() if '销售员' in ix else '' if s: sp25amt[(cust, s)] += amt wb.close() # 校验：文件自带透视 Sheet8 wb = openpyxl.load_workbook(H + '/2025全年订单.xlsx', data_only=True, read_only=True) piv = {} for row in wb['Sheet8'].iter_rows(min_row=4, values_only=True): k, v = str(row[0] or '').strip()[:7], sf(row[1]) if k.startswith('2025-') and v: piv[k] = round(v, 2) wb.close() print('█ 2025 校验（自算 vs 文件透视表 Sheet8，元）') ok_all = True for k in MONTHS: a, b = round(m25.get(k, 0), 2), piv.get(k, 0) flag = '✅' if abs(a - b) < 1 else '❌' if flag == '❌': ok_all = False print(' %s %-8s 自算 %14s ｜ 透视 %14s ｜ 差 %10s' % (flag, k, format(a, ',.2f'), format(b, ',.2f'), format(a - b, ',.2f'))) print(' 2025 1-9月合计 自算 %s 元' % format(sum(m25.values()), ',.2f')) assert ok_all, '2025 口径校验未通过——先修口径再继续' # ── ② 2026（主档）── wb = openpyxl.load_workbook(CUR, data_only=True, read_only=True) ws = wb[wb.sheetnames[0]] hdr = None p26 = defaultdict(lambda: defaultdict(float)) m26 = defaultdict(float) sp26amt = defaultdict(float) for row in ws.iter_rows(min_row=1, values_only=True): if hdr is None: hdr = [str(x or '').strip() for x in row] ix = {k: hdr.index(k) for k in ('合同日期', '客户', '销售员', '价税合计', '是否赠品', '单据类型', '单据状态') if k in hdr} continue if '合同日期' not in ix: break d = str(row[ix['合同日期']])[:7] cust = str(row[ix['客户']] or '').strip() if not ok_row(row[ix['单据类型']] if '单据类型' in ix else '', row[ix['是否赠品']] if '是否赠品' in ix else '', cust): continue if '单据状态' in ix and str(row[ix['单据状态']] or '').strip() not in ('已审核', ''): continue amt = sf(row[ix['价税合计']]) or 0.0 p26[cust][d] += amt m26[d] += amt s = str(row[ix['销售员']] or '').strip() if '销售员' in ix else '' if s: sp26amt[(cust, s)] += amt wb.close() print() print('█ 2026 逐月（元）:', {k: round(v) for k, v in sorted(m26.items())}) print(' 2026 1-9月合计 %s 元 ｜ 10月 %s 元' % (format(sum(v for k, v in m26.items() if k in MONTHS26), ',.2f'), format(m26.get('2026-10', 0), ',.2f'))) # ── ③ 归属：集团 + 部门 ── kc = json.load(io.open(W + '/数据/pricing/_key_customers.json', encoding='utf-8')) grp = {r['name']: r['group'] for r in kc if str(r.get('group') or '').strip()} dept_of_cust = {r['name']: r.get('dept', '') for r in kc if r.get('dept')} crm = json.load(io.open(W + '/数据/mcp_customers.json', encoding='utf-8'))['customers'] for r in crm: g = str(r.get('group') or '').strip() if g and r.get('name') not in grp: grp[r['name']] = g pers = json.load(io.open(W + '/数据/pricing/_personnel.json', encoding='utf-8')) pp = pers.get('personnel') if isinstance(pers, dict) else pers dept_of_sp = {} for p in pp: d = str(p.get('dept') or '') if '大客户' in d: dept_of_sp[p['name']] = '大客户部' elif '拓展' in d: dept_of_sp[p['name']] = '销售拓展部' elif '生命科学' in d: dept_of_sp[p['name']] = '生命科学部' def cust_dept(c): if dept_of_cust.get(c): return dept_of_cust[c] best, ba = '', 0.0 for (cc, s), a in sp26amt.items(): if cc == c and a > ba and s in dept_of_sp: best, ba = dept_of_sp[s], a if best: return best best, ba = '', 0.0 for (cc, s), a in sp25amt.items(): if cc == c and a > ba and s in dept_of_sp: best, ba = dept_of_sp[s], a return best # ── ④ 集团级 2 年同比 ── def roll(periods, pk): g = defaultdict(lambda: {'amt': 0.0, 'custs': set()}) for c, ms in pk.items(): key = grp.get(c) or c # 有集团按集团，无集团按客户 v = sum(ms.get(m, 0) for m in periods) if v: g[key]['amt'] += v g[key]['custs'].add(c) return g G25 = roll(MONTHS, p25) G26 = roll(MONTHS26, p26) gdept, g26c = {}, {} for c, ms in p26.items(): key = grp.get(c) or c v = sum(ms.get(m, 0) for m in MONTHS26) d = cust_dept(c) if v and (key not in g26c or v > g26c[key][0]): g26c[key] = (v, c, d) for key, (v, c, d) in g26c.items(): gdept[key] = d for c, ms in p25.items(): key = grp.get(c) or c if key not in gdept: gdept[key] = cust_dept(c) rows = [] for key in set(list(G25) + list(G26)): a26 = G26.get(key, {}).get('amt', 0.0) a25 = G25.get(key, {}).get('amt', 0.0) if a26 < 1 and a25 < 1: continue rows.append({'集团': key, '部门': gdept.get(key, '') or '未标注', '2025_1-9月': round(a25, 2), '2026_1-9月': round(a26, 2), '同比': (round((a26 - a25) / a25 * 100, 1) if a25 > 0 else None), '增减额': round(a26 - a25, 2), '2025实体数': len(G25.get(key, {}).get('custs', [])), '2026实体数': len(G26.get(key, {}).get('custs', []))}) rows.sort(key=lambda r: -r['2026_1-9月']) json.dump({'meta': {'gen': '2026-10-16', 'period': '1-9月', 'caliber': '标准销售订单+非赠品+剔除宝锐系', 'yx': '集团=CRM权威映射，无集团按客户本体'}, 'rows': rows}, io.open(W + '/数据/客户2年同比_集团口径.json', 'w', encoding='utf-8'), ensure_ascii=False, indent=1) dk = [r for r in rows if r['部门'] == '大客户部'] print() print('█ 大客户部 · 集团级 2 年同比（2026 vs 2025，1-9月，元）｜ 共 %d 个集团/客户' % len(dk)) print(' %-28s %14s %14s %9s %13s' % ('集团', '2025', '2026', '同比', '增减额')) for r in dk[:20]: y = ('%.1f%%' % r['同比']) if r['同比'] is not None else '新增' print(' %-28s %14s %14s %9s %13s' % (r['集团'][:28], format(r['2025_1-9月'], ',.0f'), format(r['2026_1-9月'], ',.0f'), y, format(r['增减额'], ',.0f'))) print() print(' 大客户部小计: 2025 %s ｜ 2026 %s ｜ 同比 %s' % (format(sum(r['2025_1-9月'] for r in dk), ',.0f'), format(sum(r['2026_1-9月'] for r in dk), ',.0f'), '%.1f%%' % ((sum(r['2026_1-9月'] for r in dk) - sum(r['2025_1-9月'] for r in dk)) / max(1, sum(r['2025_1-9月'] for r in dk)) * 100))) print('✅ JSON: 数据/客户2年同比_集团口径.json')