#!/usr/bin/env python3 # -*- coding: utf-8 -*- """官方通报「考核分数通报」-> 看板用真实数据包 输入: 数据/kpi_actual.json (由 _parse_kpi_actual.py 生成) + 数据/kpi_scheme_raw.json 输出: 数据/kpi_actual_pack.json 结构: { "roles": { rk: { month: {people, items[], sup[], kt{}, kt70{}, total{}, coef{}, rank{}} } }, "map": { rk: { month: {schemeItemNo: [actItemIdx,...]} } } } """ import json, re, os BASE = "/Users/liuxinyuan/Desktop/Hermes输出-工作类" A = json.load(open(os.path.join(BASE, "数据/kpi_actual.json"))) RAW = json.load(open(os.path.join(BASE, "数据/kpi_scheme_raw.json"))) SCH = json.load(open(os.path.join(BASE, "数据/kpi_scheme.json"))) # 富化规则（type/tiers/tierLabel） # 角色 -> (raw srcTag, {月份: 协议书sheet名}) ROLES = { "dk": ("大客户部", {"2026-08": "大客户部-8月", "2026-09": "大客户部-9月"}), "tz": ("拓展部", {"2026-08": "销售工程师-8月", "2026-09": "销售工程师-9月"}), "ls_eng": ("科研_销售工程师", {"2026-08": "销售工程师-2026年8月", "2026-09": "销售工程师-2026年9月"}), "ls_mgr": ("科研_区域经理", {"2026-08": "科研区域销售经理8月", "2026-09": "科研区域销售经理9月"}), } # 关键任务项：按关键词把「协议书项」匹配到「通报项」 KW = { "有效客户拜访次数": ["客户拜访次数"], "协助拜访课题组数量": ["协助拜访课题组"], "新客户开发数量": ["新客户开发"], "召开销售周会和复盘会议": ["周会"], "课题组关键决策人拜访数量": ["关键决策人"], "测试单发放数量": ["测试单发放数量"], "当月测试单反馈率": ["测试反馈率", "反馈率"], "回款及时率": ["回款金额的比例", "回款"], "成交客户数量": ["客户版块", "高校课题组"], "月度关键任务": ["workshop"], "促销成果": ["促销"], } DROP_PREFIX = ("日常过程管理", "执行能力", "工作态度", "月度销售额") # 上级评分原则子表，非考核项 def clean(n): for p in DROP_PREFIX: if n.startswith(p): return None return n def build_role(rk, act_key): d = A[act_key] out = {"months": {}, "map": {}} for ym, mo in sorted(d["months"].items()): items = [it for it in mo["items"] if clean(it["name"])] names = [it["name"] for it in items] used = set() mapping = {} # 先匹配协议书里的关键任务项 for sheet in ROLES[rk][1].values(): pass for sheet_name, sh in RAW[ROLES[rk][0]]["sheets"].items(): pass # 用当前月对应的协议书项（8/9月规则相同顺序） sheet = ROLES[rk][1].get(ym) if not sheet: # 历史月份：按通报项自身顺序（无协议书）→ 顺序映射 recs = [] for i, it in enumerate(items): recs.append((i + 1, it)) mapping = {} for i, it in enumerate(items): mapping[str(i + 1)] = [i] else: scheme_items = RAW[ROLES[rk][0]]["sheets"][sheet]["items"] for si in scheme_items: nm = si["item"] if nm.startswith("指派任务"): continue kws = KW.get(nm, [nm]) hit = None for kw in kws: for i, an in enumerate(names): if i in used: continue if kw in an: hit = i break if hit is not None: break if hit is None: # 兜底：字符重合度 best, bi = 0, None for i, an in enumerate(names): if i in used: continue common = len(set(nm) & set(an)) / max(1, len(set(nm))) if common > best: best, bi = common, i hit = bi if best >= 0.5 else None if hit is not None: used.add(hit) mapping[str(si["no"])] = [hit] # 指派任务 → 通报里剩余项 for si in scheme_items: if si["item"].startswith("指派任务"): mapping[str(si["no"])] = [i for i in range(len(items)) if i not in used] # 分档项（A/B/C）在通报里门槛单元格是空的 → 从协议书的 tierLabel/tiers 生成门槛文案 tl_map = {} try: rich = SCH.get("roles", {}).get(rk, {}).get("months", {}).get(ym, {}).get("items", []) or scheme_items for si in rich: t = si.get("tiers") or {} if si.get("type") != "abc" or not isinstance(t, dict) or not t: continue def seq(col): vals = [(t.get(k) or {}).get(col) for k in ("T1", "T2", "T3")] if any(v is None for v in vals): return None return "/".join(str(int(round(float(v) * 100))) for v in vals) parts = ["%s档 %s%%" % (c, seq(c)) for c in ("A", "B", "C") if seq(c)] lab = " · ".join(parts) for ix in mapping.get(str(si.get("no")), []): tl_map[ix] = lab except NameError: tl_map = {} people = mo["people"] pack_items = [] for i, it in enumerate(items): pack_items.append({ "n": it["name"], "w": it["weight"], "T": it["tiers"], "v": {p: it["rawvals"].get(p) for p in people}, "s": {p: it["score"].get(p) for p in people}, "d": {p: it["disc"].get(p) for p in people}, "tl": tl_map.get(i, ""), }) sup = [] for s in mo["superiors"]: if not s["scores"]: continue full = s.get("full") if full is None: m = re.search(r"满分\s*(\d+)", s["name"]) full = float(m.group(1)) if m else None sup.append({"n": s["name"], "full": full, "s": s["scores"]}) kt = {} kt70 = {} tot = {} coef = {} for k, v in mo["totals"].items(): if "折算70" in k: kt70 = v elif "关键任务" in k: kt = v elif "系数" in k: coef = v else: tot = v rank = {p: i + 1 for i, p in enumerate(sorted([x for x in tot], key=lambda n: -tot[n]))} wsum = sum([(it.get("weight") or 0) for it in items]) or None # 官方表内部自洽性：Σ折算得分 是否等于 官方关键任务得分 selfcheck = {} for p in people: ds = sum([(it["disc"].get(p) or 0) for it in items]) ktv = kt.get(p) if ktv is not None and abs(ds - ktv) > 0.051: selfcheck[p] = {"sum_disc": round(ds, 2), "kt": ktv} out["months"][ym] = {"people": people, "items": pack_items, "sup": sup, "kt": kt, "kt70": kt70, "total": tot, "coef": coef, "rank": rank, "wsum": (round(wsum, 4) if wsum else None), "selfcheck": selfcheck, "rule_text": mo.get("rule_text"), "title": mo["title"]} out["map"][ym] = mapping return out def main(): res = {"roles": {}, "map": {}} for rk, act_key in [("dk", "dk"), ("tz", "tz"), ("ls_eng", "ls_eng"), ("ls_mgr", "ls_mgr")]: r = build_role(rk, act_key) res["roles"][rk] = r["months"] res["map"][rk] = r["map"] p = os.path.join(BASE, "数据/kpi_actual_pack.json") json.dump(res, open(p, "w"), ensure_ascii=False, separators=(",", ":")) print("WROTE", p, os.path.getsize(p)) for rk in res["roles"]: for ym in sorted(res["roles"][rk]): m = res["roles"][rk][ym] print(f"{rk:7s} {ym} 人={len(m['people'])} 项={len(m['items'])} 上级={len(m['sup'])} " f"关键任务={m['kt']} 总分={m['total']}") if __name__ == "__main__": main()