#!/usr/bin/env python3 # -*- coding: utf-8 -*- """给 三部门逐月_2026.json 的 salespersons 数组补「个人目标 / 达成率」。 数据源（逐人目标）—— 已探明，唯一权威： 数据/kpi_scheme.json → people[]： · h2_months = [7,8,9,10,11,12] 月个人目标（元） · annual_target = 个人年度目标（元） · dept / role / conf（分配依据：按人年度目标占比分摊部门 H2 余额） 产出（就地写回，幂等）： salespersons[i]['target'] 当月个人目标（万元）—— 仅 7-12 月有源；1-6 月无逐人月目标 → null salespersons[i]['achieve'] 当月达成率 = amt / target × 100（%）；无目标 → null salespersons[i]['target_annual'] 个人年度目标（万元）；无源 → null salespersons[i]['achieve_ytd'] 个人 1-9 月累计实际 / 年度目标 × 100（%）；同一人各月行相同 meta.salesperson_targets 口径说明 + 覆盖率 meta.salesperson_target_gap 无逐人目标源 / 缺 1-6 月逐人月目标的如实记录 铁律：只加字段、不改任何金额；不编造目标 —— 找不到就是 null 并在 meta 里写明。 用法：env -u PYTHONPATH -u VIRTUAL_ENV /usr/bin/python3 脚本/apply_salesperson_targets.py """ import io import json import os import shutil import sys from collections import defaultdict sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import sp_merge as _spm # 销售员归并映射（唯一权威源，见 脚本/sp_merge.py） import target_source as _ts # 目标数据唯一权威源（见 脚本/target_source.py） HOME = os.path.expanduser("~") W = f"{HOME}/Desktop/Hermes输出-工作类" DATA = f"{W}/数据/三部门逐月_2026.json" TARGET_AUTH = f"{W}/数据/目标基础数据_2026.json" # 唯一权威源 SCHEME = f"{W}/数据/kpi_scheme.json" # 旧源（deprecation，仅回退） # 逐人月目标：唯一权威源含全年 12 个月；旧源仅 H2（7-12 月） Q_MONTHS = ['2026-07', '2026-08', '2026-09', '2026-10', '2026-11', '2026-12'] ALL_MONTHS = ['2026-%02d' % m for m in range(1, 13)] WAN = 1e4 def load_scheme(): """返回 {姓名: {'annual_wan': x, 'mon_wan': {月: 万元}}, 'src': 源名}。 优先读**唯一权威源** 数据/目标基础数据_2026.json（people[].target_annual / target_monthly，全年 12 月）； 缺失/损坏 → 回退旧源 kpi_scheme.json（仅 H2）并打印告警。 """ sd = _ts.load() if sd: out = {} for p in (sd.get('people') or []): nm = (p.get('name') or '').strip() if not nm: continue at = p.get('target_annual') tm = p.get('target_monthly') or {} mon = {m: round(float(v), 2) for m, v in tm.items() if isinstance(v, (int, float))} if at is None and not mon: continue # 无任何目标 → 仍视为「无源」 out[nm] = {'annual_wan': (round(float(at), 2) if at is not None else None), 'mon_wan': mon} return out, '数据/目标基础数据_2026.json（唯一权威源）' print('⚠️ 未找到权威源 数据/目标基础数据_2026.json → 回退旧源 kpi_scheme.json（已 deprecation）') if not os.path.exists(SCHEME): return {}, None d = json.load(io.open(SCHEME, encoding="utf-8")) out = {} for p in (d.get("people") or []): nm = (p.get("name") or "").strip() if not nm: continue at = p.get("annual_target") hm = p.get("h2_months") or [] mon = {} for i, m in enumerate(Q_MONTHS): if i < len(hm) and isinstance(hm[i], (int, float)): mon[m] = round(float(hm[i]) / WAN, 2) out[nm] = { "annual_wan": round(float(at) / WAN, 2) if isinstance(at, (int, float)) and at else None, "mon_wan": mon, } return out, '数据/kpi_scheme.json（旧源·deprecation 回退）' def merged_target(scheme, name): """把「归并到 name 名下」的全部成员在逐人目标源里的 月/年目标 相加。 · 月目标：逐月累加，只累加有值的成员（某人某月无值则不加）；全无 → 空 dict → 该月 null · 年度目标：累加成员年度目标；全无 → None 归并后的大客户1岗（刘子研）年度目标由 _spm.TARGET_ANNUAL_WAN 强制覆盖（用户指定）。 区域经理（吴云/韩远怀）年度目标由 _spm.MGR_TARGET_OVERRIDE 覆盖（缺口按区域摊入）， 且其 7-12 月个人月目标按「现有月目标占比」等比缩放，使 6 个月合计精确 = 新年目标 （末月吸收舍入差；无月目标的月份仍然缺失/null，绝不写 0）。 """ members = _spm.group_members(name) mon, annuals = {}, [] for mem in members: s = scheme.get(mem) if not s: continue for m, v in (s["mon_wan"] or {}).items(): if isinstance(v, (int, float)): mon[m] = round(mon.get(m, 0.0) + v, 2) if s["annual_wan"]: annuals.append(s["annual_wan"]) annual = round(sum(annuals), 2) if annuals else None if name == _spm.MERGE_TARGET: annual = _spm.TARGET_ANNUAL_WAN # 大客户1岗全年目标：用户 2026-10-06 指定 ov = _spm.MGR_TARGET_OVERRIDE.get(name) if ov is not None: annual = ov # 区域经理：缺口按区域摊入后的一年目标 tot = round(sum(mon.values()), 2) if mon and tot > 0: # 按现有月目标占比等比缩放，合计精确 = 新年目标 keys = sorted(mon.keys()) vals = {k: round(mon[k] * ov / tot, 2) for k in keys} vals[keys[-1]] = round(ov - sum(vals[k] for k in keys[:-1]), 2) mon = vals return mon, annual def main(): if not os.path.exists(DATA): print(f"❌ 找不到 {DATA}") return 1 scheme, src_name = load_scheme() if not scheme: print(f"❌ 找不到逐人目标源 {TARGET_AUTH} / {SCHEME}") return 1 J = json.load(io.open(DATA, encoding="utf-8")) sps = J.get("salespersons") or [] # 个人 1-9 月累计实际（来自本数据源的 salespersons 行；口径与 customers/depts 一致） ytd = defaultdict(float) for r in sps: ytd[r.get("sp")] += r.get("amt") or 0.0 n_hit_mon = n_hit_annual = 0 no_source = set() # 有 ERP 金额但完全无逐人目标 no_month_h1 = defaultdict(set) # 无 1-6 月逐人月目标的销售员 _tgt_cache = {} # 归并后姓名 -> (mon_wan, annual_wan)；非归并人取本人（与旧逻辑逐值一致） for r in sps: sp = r.get("sp") s = scheme.get(sp) m = r.get("month") r["target"] = None r["achieve"] = None r["target_annual"] = None r["achieve_ytd"] = None if sp not in _tgt_cache: _tgt_cache[sp] = merged_target(scheme, sp) mon, annual = _tgt_cache[sp] in_scheme = (s is not None) or bool(mon) or (annual is not None) if not in_scheme: no_source.add(sp) continue # 当月目标（7-12 月有源；归并后取组成员月目标之和，无值不计；无源月份留 null） tgt = mon.get(m) if tgt: r["target"] = tgt r["achieve"] = round((r.get("amt") or 0.0) / tgt * 100, 1) n_hit_mon += 1 elif m and m < '2026-07': no_month_h1[sp].add(m) # 年度目标 + YTD 达成率 if annual: r["target_annual"] = annual r["achieve_ytd"] = round(ytd.get(sp, 0.0) / annual * 100, 1) n_hit_annual += 1 meta = J.setdefault("meta", {}) meta["salesperson_targets"] = { "source": f"{src_name}（唯一权威目标源 → 逐人年度目标 + 12 个月度目标）", "monthly_caliber": ("target = 当月个人目标（万元），来自唯一权威源（全年 12 个月，按 monthly_rule 摊自年度、合计精确=年度）。" "achieve = 当月实际 / 当月目标 × 100（%）"), "annual_caliber": ("target_annual = 个人年度目标（万元）；" "achieve_ytd = 个人 1-9 月累计实际 / 个人年度目标 × 100（%），同一人各月行相同"), "rows": len(sps), "rows_with_monthly_target": n_hit_mon, "rows_with_annual_target": n_hit_annual, "people_with_source": len(scheme), "people_no_source": sorted(no_source), } meta["salesperson_target_gap"] = { "monthly_1_6_missing": ("逐人月目标现由唯一权威源（数据/目标基础数据_2026.json）提供全年 12 个月；" "旧源 kpi_scheme.json 仅 H2（7-12 月），已 deprecation、仅回退。"), "people_with_rows_but_no_target": sorted(no_source), "note": "未在权威源中登记年度目标者（无 KPI 目标的新人/离职/助理）无逐人目标，字段为 null。", } # 人员归并口径（唯一权威源 脚本/sp_merge.py）：黄明月/张立娅/刘新元 → 刘子研 meta["person_merge"] = _spm.meta_block() # 离职人员业绩划入区域经理 + 缺口按区域摊给区域经理（唯一权威源 脚本/sp_merge.py） meta["mgr_merge"] = _spm.mgr_merge_meta_block() # 目标数据唯一权威源（lineage） _sd = _ts.load() or {} meta["target_source"] = { "authoritative": "数据/目标基础数据_2026.json", "supersedes": (_sd.get("meta") or {}).get("supersedes"), "monthly_rule": (_sd.get("meta") or {}).get("monthly_rule"), "generated": (_sd.get("meta") or {}).get("generated"), "note": "逐人/部门/区域目标均取自唯一权威源；改目标＝管理台导出 JSON 落盘后重跑 脚本/refresh_sandept.sh。", } # ── 自检：只加字段，不动金额 ── s_amt = round(sum(r.get("amt") or 0 for r in sps), 2) print("================ 逐人目标门禁 ================") print(f" 逐人源: kpi_scheme.json people[]（{len(scheme)} 人）") print(f" salespersons 行数={len(sps)} 当月目标命中={n_hit_mon} 年度目标命中={n_hit_annual}") print(f" 无逐人目标源的销售员 {len(no_source)} 人: {sorted(no_source)[:20]}") print(f" Σamt（应不变）= {s_amt:,.2f} 万") shutil.copy2(DATA, DATA + ".bak-pre-sptarget") io.open(DATA, "w", encoding="utf-8").write( json.dumps(J, ensure_ascii=False, separators=(",", ":"))) print(f"\n✅ 已写回 {DATA}（备份 .bak-pre-sptarget）") return 0 if __name__ == "__main__": sys.exit(main())