# -*- coding: utf-8 -*- """ 跟进闭环监控器 v1.0 =========================================================== 目标：让「每个员工或客户登录后填写的任何信息」都做到 1) 跟进 —— 缺跟进的自动建跟进并指派负责人 2) 显示 —— 生成统一看板(HTML)，红/黄/绿一目了然 3) 汇报 —— 生成每日跟进汇报(MD) + 明细(CSV) 原理：直连 智位小猿CRM(工泰合作版) 的 REST API，把各业务实体的 "信息填写"记录聚合成一张「必须跟进」登记表，按 RAG 着色， 并通过各实体专属的 /followup 写接口自动补齐跟进。 重要约束（已实测）： * 当前 CRM 数据文件写入返回 500 EPERM（原子重命名被拒）， 所以 AUTO_FOLLOWUP 默认 False（只报告不写入）。 * 修好 CRM 写入权限后，把 AUTO_FOLLOWUP 改为 True 即可自动建跟进。 * 读接口(显示/汇报)不受 EPERM 影响，始终可运行。 用法： python 跟进闭环监控器.py # 报告模式：生成看板+日报(不写CRM) python 跟进闭环监控器.py --enforce # 强制模式：对红项自动建跟进+指派 python 跟进闭环监控器.py --report # 等价于默认(显式) """ import json, urllib.request, urllib.error, datetime, os, sys, html # ============================ 配置 ============================ CONFIG = { "CRM_BASE": "https://crm.ixgn.cn", "USERNAME": "admin", "PASSWORD": "88073100", "DEFAULT_OWNER": "管理员", # 未指派记录的占位负责人（自动模式下会被分级重排掉） "AUTO_FOLLOWUP": True, # 自动模式：对红项自动建跟进+指派（EPERM已解除，2026-08-09启用） "OUTDIR": ".", # 输出目录(当前工作目录) # 是否把"占位负责人=管理员"的旧记录也按分级重排（True=每天重排；False=只填空缺） "REASSIGN_PLACEHOLDER": True, } # ====================== 业务员分级名册（按盈利大小派单） ====================== # 级别从高到低（min_score 为"盈利大小分"门槛，分数越高盈利越大）。 # owners 为该级别业务员名册（自由文本，可填多人，脚本在级别内轮询均摊）。 # 请按贵司真实人员/级别修改本表；字段取值依据 CRM 实测： # deals.value(14k~1.6M) / target-sites.investmentScore(16~64) / community.reviewScore,Roi / leads.cooperation进度 SALES_TIERS = [ {"level": "L3·经理级", "min_score": 70, "owners": ["王主管"], "desc": "高盈利/大客户/大场站，需经理亲自谈或督办"}, {"level": "L2·中级", "min_score": 40, "owners": ["张伟", "李业务"], "desc": "中盈利/成长期客户与场站"}, {"level": "L1·开发级", "min_score": 0, "owners": ["小李", "李业务"], "desc": "纯开发线索/小客户/观望场站，由开发业务员跟进"}, ] # ====================== 企业微信/个人微信 推送配置 ====================== # 两种推送方式（二选一，填了 webhook_url 优先用群机器人，门槛最低）： # 方式A 群机器人 webhook：在企业微信里建群→添加群机器人→复制 webhook URL 填下方 # （群里所有人：企业微信/个人微信都能收到，无需 corpid/secret） # 方式B 自建应用：需 corpid/agentid/corpsecret，可定向发给指定成员/部门 # 个人微信接收：企业微信消息可在个人微信查看（开启「微信插件」或在企微里允许微信接收） WECOM = { "enabled": False, # 填好下方凭证后改为 True 才会推送 "webhook_url": "", # 方式A：企业微信群机器人 webhook（推荐，门槛最低） "corpid": "", # 方式B：企业ID（我的企业→企业信息） "corpsecret": "", # 方式B：自建应用 Secret（应用管理→自建→查看Secret） "agentid": 0, # 方式B：自建应用 AgentId "to_user": "@all", # 方式B：推送人，"@all" 或 "zhangsan|lisi" "to_party": "", # 方式B：部门ID（逗号分隔） "to_tag": "", # 方式B：标签ID（逗号分隔） "alert_only": False, # True=仅当出现红项/逾期才推送（适时告警）；False=每次运行都推送 "max_red_lines": 12, # 推送文本中最多列出几条红项 } # ====================== Gotify 自建推送配置 ====================== # Gotify 是自建通知服务，把消息推到你手机/桌面的 Gotify App（非个人微信）。 # 你提供过两串 token：T1=应用令牌(app token, ~36位) 用于管理；T2=客户端令牌(client token, 43位) 用于发消息。 # 发消息用 客户端令牌(T2)。base_url 填你的 Gotify 服务地址。 GOTIFY = { "enabled": False, # 填好 base_url 并验证后改为 True 才会推送 "base_url": "", # 你的 Gotify 服务地址，如 https://gotify.example.com 或 http://内网IP:8080 "app_token": "aibT747buG59YjKkMGzWt1x64jyueFGF3c9", # T1 应用令牌(管理用) "client_token": "bEWPD32W1KGBLNTgFpqhZ3iLmTAkNKO66fWN7Qvdsgf", # T2 客户端令牌(发消息用) "priority": 5, # 1-10，越大越醒目 "alert_only": False, # True=仅当出现红项/逾期才推送 } # 超时天数阈值：创建超过该天数且无跟进 -> 视为逾期 OVERDUE_DAYS = 3 # ====================== 实体覆盖定义 ====================== # 每个实体 = 一类「有人填写的信息」。统一抽字段：id/name/owner/ # followups(数组)/followcount/status/date/resolved状态集合/深链 ENTITIES = [ { "key": "leads", "label": "客户线索", "list": "/api/leads?limit=300", "id": "id", "name": "name", "owner": "assignee", "type": "category", "district": "district", "status": "status", "date": "createdAt", "followups": None, "followcount": None, "resolved": ["won", "converted", "closed", "已签约", "已转化", "已成交"], "list_route": "/leads", # leads 的跟进信号：callLog 非空 或 followUpDate 已设 "custom_followed": lambda it: bool(it.get("callLog")) or bool(it.get("followUpDate")), "custom_resolved": lambda it: (it.get("leadStatus") in ["won","converted","已签约","已成交"]) or it.get("status")=="closed", }, { "key": "cooperation-submissions", "label": "场地方合作提交", "list": "/api/cooperation-submissions", "id": "id", "name": "companyName", "owner": "assignedTo", "type": "type", "district": "district", "status": "status", "date": "createdAtText", "followups": "followUps", "followcount": "followCount", "resolved": ["已评审", "已签约", "已通过"], "list_route": "/cooperation", }, { "key": "community-submissions", "label": "小区合作提交", "list": "/api/community-submissions", "id": "id", "name": "communityName", "owner": "assignedTo", "type": "category", "district": "projectAddress", "status": "status", "date": "createdAtText", "followups": "followUps", "followcount": "followCount", "resolved": ["已评审", "已通过", "已签约"], "list_route": "/community", }, { "key": "target-sites", "label": "目标场地", "list": "/api/target-sites?limit=300", "id": "id", "name": "name", "owner": "contactName", "type": "type", "district": "district", "status": "status", "date": "updatedAt", "followups": "followUps", "followcount": "followCount", "resolved": ["已开发", "已签约", "已转化"], "list_route": "/target-sites", }, { "key": "tenders", "label": "招投标", "list": "/api/tenders?limit=300", "id": "id", "name": "name", "owner": "contactName", "type": "type", "district": "district", "status": "status", "date": "updatedAt", "followups": "followUps", "followcount": "followCount", "resolved": ["已投标", "已中标", "已放弃"], "list_route": "/tenders", }, { "key": "tasks", "label": "任务", "list": "/api/tasks?limit=300", "id": "id", "name": "title", "owner": "owner", "type": "type", "district": "", "status": "status", "date": "createdAt", "followups": None, "followcount": None, "resolved": [], "list_route": "/tasks", "custom_followed": lambda it: bool(it.get("done")) or it.get("status") in ["done","completed"], "custom_resolved": lambda it: bool(it.get("done")) or it.get("status") in ["done","completed"], }, { "key": "deals", "label": "商机", "list": "/api/deals?limit=300", "id": "id", "name": "name", "owner": "assignee", "type": "industry", "district": "district", "status": "stage", "date": "createdAt", "followups": None, "followcount": None, "resolved": ["won", "closed-won", "已签约", "已成交"], "list_route": "/deals", "custom_followed": lambda it: bool(it.get("lastActivityAt")) or bool(it.get("timeline")), "custom_resolved": lambda it: it.get("stage") in ["won","closed-won","已签约","已成交"], }, ] # ============================ 工具 ============================ def log(*a): print("[闭环]", *a, flush=True) def api_get(token, path): req = urllib.request.Request(CONFIG["CRM_BASE"]+path, headers={"Authorization":"Bearer "+token}) return json.loads(urllib.request.urlopen(req, timeout=15).read().decode()) def api_write(token, method, path, body): data = json.dumps(body).encode() req = urllib.request.Request(CONFIG["CRM_BASE"]+path, data=data, headers={"Authorization":"Bearer "+token, "Content-Type":"application/json"}, method=method) try: resp = urllib.request.urlopen(req, timeout=15) return resp.status, resp.read().decode() except urllib.error.HTTPError as e: return e.code, e.read().decode() def login(): body = json.dumps({"username":CONFIG["USERNAME"],"password":CONFIG["PASSWORD"]}).encode() req = urllib.request.Request(CONFIG["CRM_BASE"]+"/api/auth/login", data=body, headers={"Content-Type":"application/json"}, method="POST") txt = urllib.request.urlopen(req, timeout=15).read().decode() return json.loads(txt).get("token") # ====================== 企业微信/个人微信 推送 ====================== def wecom_get_token(): u = f"https://qyapi.weixin.qq.com/cgi-bin/gettoken?corpid={WECOM['corpid']}&corpsecret={WECOM['corpsecret']}" try: d = json.loads(urllib.request.urlopen(u, timeout=12).read().decode()) if d.get("errcode", 0) != 0: log(f"企微获取token失败: {d.get('errmsg')}") return None return d.get("access_token") except Exception as ex: log(f"企微获取token异常: {ex}") return None def wecom_send_text(content): # 方式A：群机器人 webhook（门槛最低，优先） if WECOM.get("webhook_url"): body = {"msgtype": "text", "text": {"content": content}} data = json.dumps(body, ensure_ascii=False).encode() try: req = urllib.request.Request(WECOM["webhook_url"], data=data, headers={"Content-Type": "application/json"}, method="POST") resp = json.loads(urllib.request.urlopen(req, timeout=12).read().decode()) if resp.get("errcode", 0) == 0: log("企微群机器人推送成功"); return True log(f"企微webhook推送失败: {resp.get('errmsg')}"); return False except Exception as ex: log(f"企微webhook异常: {ex}"); return False # 方式B：自建应用 tok = wecom_get_token() if not tok: return False body = {"touser": WECOM["to_user"], "toparty": WECOM["to_party"], "totag": WECOM["to_tag"], "msgtype": "text", "agentid": int(WECOM["agentid"]), "text": {"content": content}, "safe": 0} data = json.dumps(body, ensure_ascii=False).encode() req = urllib.request.Request("https://qyapi.weixin.qq.com/cgi-bin/message/send?access_token="+tok, data=data, headers={"Content-Type": "application/json"}, method="POST") try: resp = json.loads(urllib.request.urlopen(req, timeout=12).read().decode()) if resp.get("errcode", 0) == 0: log("企微自建应用推送成功"); return True log(f"企微自建应用推送失败: {resp.get('errmsg')}"); return False except Exception as ex: log(f"企微自建应用推送异常: {ex}"); return False def build_wecom_text(rows, write_blocked): from collections import Counter c = Counter(r["rag"] for r in rows) lv = Counter(r["level"] for r in rows) overdue = [r for r in rows if r["overdue"]] q, nc = alert_rows(rows) # 已开发线索（优质优先 / 无联系方式置底） dev_red = q + nc bl = backlog_count(rows) now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") L = [] L.append("【充电桩CRM·跟进闭环日报】") L.append(f"生成 {now}") L.append(f"总 {len(rows)} 条 | 🔴{c.get('red',0)} 未跟进 | 🟡{c.get('yellow',0)} 跟进中 | 🟢{c.get('green',0)} 已闭环 | ⏰{len(overdue)} 逾期") L.append(f"🔔 每日告警聚焦(已开发线索)：{len(dev_red)} 条；储备池(未开发批量)：{bl} 条") L.append(f"分级派单：L3经理级{lv.get('L3·经理级',0)} · L2中级{lv.get('L2·中级',0)} · L1开发级{lv.get('L1·开发级',0)}") if write_blocked: L.append("\n⚠️ 阻断：CRM写入EPERM，自动跟进未落库，请先修复写入权限") if dev_red: L.append(f"\n🔔 已开发线索待跟进({len(dev_red)}条，列前{WECOM['max_red_lines']}条，优质优先)：") for r in dev_red[:WECOM['max_red_lines']]: L.append(f"· {r['id']} {r['name']}({r['type']}/{r['district']}) 盈利分{r['score']} 负责人:{r['owner'] or '未指派'}") if len(dev_red) > WECOM['max_red_lines']: L.append(f"· …其余{len(dev_red)-WECOM['max_red_lines']}条见看板") else: L.append("\n✅ 无已开发待跟进线索") if nc: L.append(f"\n📞 无联系方式线索(待补充)：{len(nc)} 条（置底，需补全电话）") if overdue: L.append(f"\n⏰ 逾期{len(overdue)}条：") for r in overdue[:5]: L.append(f"· {r['id']} {r['name']} 已{r['age_days']}天无跟进") L.append(f"\n查看完整看板：见本地《跟进闭环看板.html》/《每日跟进汇报.md》") return "\n".join(L) def gotify_send_text(content): if not GOTIFY.get("base_url"): log("Gotify 未配置 base_url，跳过推送"); return False url = GOTIFY["base_url"].rstrip("/") + "/message?token=" + GOTIFY["client_token"] body = {"title": "充电桩CRM·跟进闭环", "message": content, "priority": int(GOTIFY["priority"])} data = json.dumps(body, ensure_ascii=False).encode() try: req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST") d = json.loads(urllib.request.urlopen(req, timeout=12).read().decode()) if "id" in d: log(f"Gotify 推送成功 (id={d.get('id')})"); return True log(f"Gotify 推送返回异常: {d}"); return False except urllib.error.HTTPError as e: log(f"Gotify 推送 HTTP 错误 {e.code}: {e.read().decode()[:160]}"); return False except Exception as ex: log(f"Gotify 推送失败: {ex}"); return False def parse_date(s): if not s: return None for fmt in ("%Y-%m-%dT%H:%M:%S.%fZ","%Y-%m-%dT%H:%M:%SZ","%Y-%m-%d","%Y/%m/%d %H:%M:%S","%Y/%m/%d"): try: return datetime.datetime.strptime(s, fmt) except: pass return None # ============================ 主逻辑 ============================ def collect(token): rows = [] today = datetime.datetime.now() for e in ENTITIES: try: d = api_get(token, e["list"]) except Exception as ex: log(f"拉取 {e['label']} 失败: {ex}") continue # 归一化：有的接口返回数组，有的返回 {items/data:[...]} if isinstance(d, list): items = d elif isinstance(d, dict): items = d.get("items") or d.get("data") or [] else: items = [] for it in items: iid = it.get(e["id"], "") name = it.get(e["name"], "") or it.get("communityName","") or it.get("companyName","") owner = it.get(e["owner"], "") if e["owner"] else "" status = it.get(e["status"], "") etype = it.get(e["type"], "") or it.get("category","") district = it.get(e["district"], "") if e["district"] else "" # 跟进信号 if e.get("custom_followed"): followed = e["custom_followed"](it) else: fArr = it.get(e["followups"]) if e["followups"] else None fc = it.get(e["followcount"]) if e["followcount"] else None followed = bool(fArr) or (isinstance(fc,(int,float)) and fc>0) if e.get("custom_resolved"): resolved = e["custom_resolved"](it) else: resolved = status in e["resolved"] # 时间/逾期 dt = parse_date(it.get(e["date"],"")) age_days = (today-dt).days if dt else None overdue = (not followed) and (age_days is not None and age_days>=OVERDUE_DAYS) # RAG if resolved: rag = "green" elif (not owner and not followed): rag = "red" else: rag = "yellow" rows.append({ "src": e["label"], "key": e["key"], "id": iid, "name": name, "type": etype, "district": district, "owner": owner, "status": status, "followed": followed, "resolved": resolved, "rag": rag, "overdue": overdue, "age_days": age_days, "date": it.get(e["date"],""), "route": e["list_route"], "raw": it, # 保留原始记录，供盈利评分使用 "score": 0, "level": "", "rep": owner, "tip": "", # 派单展示字段 }) return rows # ====================== 分级派单（按开发线索 + 盈利大小） ====================== def calc_value_score(it, ekey): """把各实体的"盈利大小"归一为 0-100 分。""" if ekey == "deals": v = it.get("value") or 0 try: v = float(v) except: v = 0 if v >= 500000: return 90 if v >= 100000: return 60 if v > 0: return 35 return 25 if ekey == "target-sites": lvl = (it.get("investmentLevel") or "") sc = it.get("investmentScore") or 0 try: sc = float(sc) except: sc = 0 base = {"高": 85, "中": 55}.get(lvl, 30) # 高/中/观望(默认30) return max(base, min(100, sc)) if ekey == "community-submissions": rs = it.get("reviewScore") or 0 try: rs = float(rs) except: rs = 0 pc = it.get("estPileCount") or 0 try: pc = float(pc) except: pc = 0 s = 30 if rs >= 70: s = 65 elif rs >= 50: s = 50 if pc >= 12: s += 10 return min(100, s) if ekey == "leads": coop = it.get("cooperation") or {} cs = coop.get("status") if isinstance(coop, dict) else "" shr = coop.get("shareRate") if isinstance(coop, dict) else "" s = 30 if cs in ("assessing", "negotiating", "survey"): s += 30 if shr: s += 10 cj = it.get("contactJourney") or {} if isinstance(cj, dict) and cj.get("hasDeal"): s += 20 if it.get("status") == "no_phone": s = max(s - 15, 10) return min(100, s) return 30 def is_dev_lead(r): """是否处于"开发线索"阶段（需开发型业务员跟进）。""" ekey = r["key"]; it = r["raw"] if ekey == "target-sites": return True # 实测全部 status=待开发 if ekey == "leads": return (it.get("status") in ("new", "no_phone")) or (not it.get("leadStatus")) if ekey == "community-submissions": return (it.get("status") in ("跟进中", "待评审", "新建", "提交")) if ekey == "deals": return (it.get("stage") in ("initial_contact", "site_survey", "proposal")) if ekey == "cooperation-submissions": return (it.get("status") in ("待评审", "新建", "提交")) return False # ====================== 「已开发线索」&「无联系方式」判定（收敛每日告警） ====================== def is_developed(r): """是否已开发：有跟进/合作进度/已联系/有负责人，才视为值得每日告警的已开发线索。 无联系方式的线索(no_phone)视作「待补充」而非「已开发优质」。""" it = r["raw"]; ekey = r["key"] if r["followed"]: return True if ekey == "leads": coop = it.get("cooperation") or {} cs = coop.get("status") if isinstance(coop, dict) else "" if cs in ("assessing", "negotiating", "survey", "signed", "已评估", "洽谈中"): return True cj = it.get("contactJourney") or {} if isinstance(cj, dict) and (cj.get("hasDeal") or cj.get("contacted")): return True if it.get("status") in ("contacted", "negotiating", "assessing"): return True return False # new / no_phone 等未开发 if ekey == "target-sites": return bool(r["owner"]) or bool(it.get("followUps")) or (it.get("status") not in ("待开发", "", None)) if ekey in ("cooperation-submissions", "community-submissions", "tenders", "deals"): return bool(r["owner"]) or bool(it.get("followUps")) return bool(r["owner"]) def is_no_contact(r): """无联系方式的线索：leads 且缺电话/联系方式渠道（如 status=no_phone 或无 phone 字段）。""" if r["key"] != "leads": return False it = r["raw"] if it.get("status") == "no_phone": return True phone = it.get("phone") or it.get("mobile") or it.get("tel") or it.get("contactPhone") or "" return not bool(phone) def alert_rows(rows): """每日告警聚焦：已开发线索（不再对全部未开发批量场站逐条提醒）。 返回 (有联系方式的优质列表[按盈利分降序], 无联系方式线索列表[置底待补充])。""" dev = [r for r in rows if r["key"] == "leads" and is_developed(r)] has_contact = [r for r in dev if not is_no_contact(r)] no_contact = [r for r in dev if is_no_contact(r)] quality = sorted(has_contact, key=lambda x: -x["score"]) # 优质优先 return quality, no_contact def backlog_count(rows): """储备池：未开发的批量记录（红项且非已开发线索），仅持续补充、不逐条告警。""" return sum(1 for r in rows if r["rag"] == "red" and not (r["key"] == "leads" and is_developed(r))) def pick_tier(score): for t in SALES_TIERS: # 已按 min_score 从高到低定义 if score >= t["min_score"]: return t return SALES_TIERS[-1] def pick_rep(tier, rr): owners = tier["owners"] if not owners: return CONFIG["DEFAULT_OWNER"] i = rr.get(tier["level"], 0) % len(owners) rr[tier["level"]] = i + 1 return owners[i] # 各实体用于写入"负责人"的字段 OWNER_FIELD = { "leads": "assignee", "target-sites": "contactName", "cooperation-submissions": "assignedTo", "community-submissions": "assignedTo", "deals": "assignee", } def write_owner(ekey, iid, rep, token): f = OWNER_FIELD.get(ekey) if not f: return False, "no-field" st, txt = api_write(token, "PUT", f"/api/{ekey}/{iid}", {f: rep}) return (200 <= st < 300), txt def score_rows(rows): """为所有记录计算盈利分/级别/提示（仅展示，不写 CRM）。""" rr = {} for r in rows: sc = calc_value_score(r["raw"], r["key"]) tier = pick_tier(sc) dev = is_dev_lead(r) r["score"] = sc r["level"] = tier["level"] r["_tier"] = tier r["_dev"] = dev sug = pick_rep(tier, rr) if tier["owners"] else CONFIG["DEFAULT_OWNER"] r["_suggest"] = sug if r["owner"]: r["rep"] = r["owner"] r["tip"] = f"已指派 {r['owner']}（按{tier['level']}标准）" else: tag = "⚠️开发线索" if dev else "盈利单" r["rep"] = sug r["tip"] = f"{tag} → 建议{tier['level']}·{sug}" def assign_tiered(rows, token): """按分级自动派单 + 对红项建跟进。返回 (动作日志, 是否EPERM阻断)。""" actions = [] wb = False rr = {} for r in rows: ekey = r["key"]; iid = r["id"] owner = r["owner"] or "" should = (owner == "") or (owner == CONFIG["DEFAULT_OWNER"] and CONFIG["REASSIGN_PLACEHOLDER"]) # 1) 对红项自动建跟进 if r["rag"] == "red": body = {"type": "note", "content": "[自动闭环] 系统巡检发现该记录暂无跟进，已自动建跟进并分级指派负责人。", "result": "pending", "operator": CONFIG["DEFAULT_OWNER"], "nextFollowAt": (datetime.datetime.now() + datetime.timedelta(days=2)).strftime("%Y-%m-%d")} st, txt = api_write(token, "POST", f"/api/{ekey}/{iid}/followup", body) ok_fu = (200 <= st < 300) else: ok_fu = True # 2) 分级指派负责人 if not should: actions.append({"id": iid, "src": r["src"], "followup": ok_fu, "assign": False, "level": r["level"], "rep": owner, "detail": "已有人，未重排"}) continue rep = r["_suggest"] ok_as, txt2 = write_owner(ekey, iid, rep, token) if not ok_as: if "EPERM" in txt2: wb = True actions.append({"id": iid, "src": r["src"], "followup": ok_fu, "assign": False, "level": r["level"], "rep": rep, "detail": txt2[:120]}) continue r["owner"] = rep; r["rep"] = rep tag = "⚠️开发线索" if r["_dev"] else "盈利单" r["tip"] = f"{tag} → 已派{r['level']}·{rep}" actions.append({"id": iid, "src": r["src"], "followup": ok_fu, "assign": True, "level": r["level"], "rep": rep, "score": r["score"]}) return actions, wb # ============================ 输出 ============================ def build_html(rows, actions, write_blocked): from collections import Counter c = Counter(r["rag"] for r in rows) lv = Counter(r["level"] for r in rows) total = len(rows) overdue = sum(1 for r in rows if r["overdue"]) now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") def rag_cn(x): return {"red":"🔴 未跟进","yellow":"🟡 跟进中","green":"🟢 已闭环"}.get(x,x) def tier_color(lv_): if lv_.startswith("L3"): return "#c0392b" if lv_.startswith("L2"): return "#e67e22" if lv_.startswith("L1"): return "#2980b9" return "#555" cards = f"""
{total}
总记录
{c.get('red',0)}
🔴 未跟进(须处理)
{c.get('yellow',0)}
🟡 跟进中
{c.get('green',0)}
🟢 已闭环
{overdue}
⏰ 逾期(> {OVERDUE_DAYS}天无跟进)
""" tier_cards = "" for t in SALES_TIERS: n = lv.get(t["level"], 0) col = tier_color(t["level"]) tier_cards += f'
{n}
{html.escape(t["level"])}
{" / ".join(t["owners"]) or "—"}
' trs = "" for r in sorted(rows, key=lambda x:(x["rag"]!="red", x["rag"]!="yellow", -(x["age_days"] or 0))): owner = r["owner"] or "未指派" ov = "⏰逾期" if r["overdue"] else "" lvl = r["level"] or "" lvl_html = f'{html.escape(lvl)}
盈利分{r["score"]}' trs += f""" {html.escape(r['src'])}{html.escape(str(r['id']))} {html.escape(str(r['name']))}{html.escape(str(r['type']))} {html.escape(str(r['district']))}{html.escape(str(r['date']))} {owner}{html.escape(str(r['status']))} {rag_cn(r['rag'])}{ov} {lvl_html}{html.escape(str(r['tip']))}""" action_html = "" if actions: lis = "".join(f"
  • {html.escape(a['id'])} [{html.escape(a['src'])}] 级别:{html.escape(a.get('level',''))} 派:{html.escape(str(a.get('rep','')))} 跟进建:{a['followup']} 指派:{a['assign']} {html.escape(a.get('detail',''))}
    • " for a in actions) action_html = f"

    本次派单动作记录({len(actions)}条)

    " l3 = [r for r in rows if r["level"].startswith("L3")] dev = [r for r in rows if r.get("_dev")] assign_html = f"""

    🎯 分级派单与提示

    按「开发线索 + 盈利大小」自动分级：L3经理级（高盈利/大客户大场站）→ {" / ".join(SALES_TIERS[0]['owners']) or '—'}；L2中级 → {" / ".join(SALES_TIERS[1]['owners'])}；L1开发级 → {" / ".join(SALES_TIERS[2]['owners'])}。

    """ # 🏆 优质线索 Top 10 推荐（已开发·有联系方式·按盈利分） q, nc = alert_rows(rows) top10 = q[:10] if top10: tlis = "".join(f"
  • {html.escape(str(r['id']))} {html.escape(str(r['name']))} · 盈利分{r['score']} · 负责人 {html.escape(str(r['owner'] or '未指派'))} · {html.escape(r['level'])}
    • " for r in top10) top_html = f"

    🏆 优质线索 Top 10 推荐（已开发·按盈利分，每日优先跟进）

      {tlis}

    其余 {backlog_count(rows)} 条为未开发批量场站/提交，归入「储备池」持续补充客户信息，不在每日告警范围；无联系方式线索置底待补录。

    " else: top_html = "" block_html = "" if write_blocked: block_html = """
    ⚠️ 跟进写入被阻断：CRM 数据文件当前返回 500 EPERM（原子重命名被拒）， 自动建跟进/派单未能落库。请先修复 CRM 写入权限（见《CRM写入权限修复指引.md》）， 再将脚本 AUTO_FOLLOWUP 改为 True 重跑。读接口(本看板/日报)不受影响。
    """ return f""" 跟进闭环看板

    🔄 跟进闭环看板

    智位小猿CRM · 生成时间 {now} · 覆盖：每个员工/客户填写的信息均须跟进/显示/汇报
    {cards}
    {tier_cards}
    {block_html} {assign_html} {top_html}

    明细（红→黄→绿 排序，逾期置顶）

    {trs}
    来源ID名称/提交人类型区域/地址填写时间负责人状态跟进逾期派单级别(盈利分)提示
    {action_html}
    跳转处理：在 CRM（{CONFIG['CRM_BASE']}）内用上方 ID 搜索即可定位记录。红项=无负责人且无跟进，必须当天处理。
    """ def build_report(rows, actions, write_blocked): from collections import Counter, defaultdict c = Counter(r["rag"] for r in rows) lv = Counter(r["level"] for r in rows) by_owner = defaultdict(lambda: Counter()) for r in rows: by_owner[r["owner"] or "未指派"][r["rag"]] += 1 now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") red = [r for r in rows if r["rag"]=="red"] overdue = [r for r in rows if r["overdue"]] dev = [r for r in rows if r.get("_dev")] l3 = [r for r in rows if r["level"].startswith("L3")] q, nc = alert_rows(rows) # 已开发线索（优质优先 / 无联系方式置底） bl = backlog_count(rows) # 储备池（未开发批量记录） lines = [] lines.append(f"# 每日跟进闭环汇报\n") lines.append(f"生成时间：{now} ") lines.append(f"CRM：{CONFIG['CRM_BASE']}\n") lines.append("## 一、总览") lines.append(f"- 总记录数：**{len(rows)}**") lines.append(f"- 🔴 未跟进(须处理)：**{c.get('red',0)}**") lines.append(f"- 🟡 跟进中：{c.get('yellow',0)}") lines.append(f"- 🟢 已闭环：{c.get('green',0)}") lines.append(f"- ⏰ 逾期(>{OVERDUE_DAYS}天无跟进)：**{len(overdue)}**") lines.append(f"- 🔔 每日告警聚焦(已开发线索)：**{len(q)+len(nc)}** 条（优质推荐 Top 10）；储备池(未开发批量场站/提交)：**{bl}** 条（仅持续补充，不逐条告警）\n") lines.append("## 二、分级派单与提示（按开发线索 + 盈利大小）") lines.append(f"- 派单规则：L3经理级(高盈利/大客户大场站)→{' / '.join(SALES_TIERS[0]['owners']) or '—'}；L2中级→{' / '.join(SALES_TIERS[1]['owners'])}；L1开发级→{' / '.join(SALES_TIERS[2]['owners'])}。") lines.append(f"- 各级派单量：L3 **{lv.get('L3·经理级',0)}** · L2 **{lv.get('L2·中级',0)}** · L1 **{lv.get('L1·开发级',0)}**") lines.append(f"- ⚠️ 开发线索（待开发/新线索）：**{len(dev)}** 条 → 由开发级业务员优先跟进") lines.append(f"- 🔥 高盈利需经理督办（L3）：**{len(l3)}** 条") if l3: lines.append(" - 高盈利清单（经理级重点）：") for r in l3[:25]: lines.append(f" - [{r['src']}] {r['id']} {r['name']} 盈利分{r['score']} 负责人：{r['owner'] or '未指派'}") lines.append("") if write_blocked: lines.append("## ⚠️ 三、阻断告警（最高优先级）") lines.append("- CRM 数据文件写入返回 **500 EPERM**（原子重命名被拒），自动建跟进/派单**未能落库**。") lines.append("- 根因：CRM 进程对其 JSON 数据目录无写权限（可能被只读属性/杀软/云同步锁定，或运行账户无权）。") lines.append("- 处置：按《CRM写入权限修复指引.md》修复后，将脚本 `AUTO_FOLLOWUP=True` 重跑即可自动补齐跟进。\n") lines.append("## 四、已开发线索·优先跟进（每日告警聚焦，不再逐条提醒全部未开发场站）") lines.append(f"> 仅对**已开发线索**告警并推荐优质 Top 10；其余 {bl} 条未开发批量记录归入「储备池」，持续录入客户信息与联系方式即可，不在每日告警范围。") lines.append(f"- 已开发线索（需跟进/持续跟进）：**{len(q)+len(nc)}** 条") top10 = q[:10] lines.append(f"\n### 🏆 优质线索 Top 10 推荐（按盈利分，已开发·有联系方式）") if top10: for i, r in enumerate(top10, 1): lines.append(f"{i}. [{r['src']}] **{r['id']}** {r['name']} 盈利分{r['score']} 负责人：{r['owner'] or '未指派'} 派单：{r['level']} 提示：{r['tip']}") else: lines.append("- 暂无已开发且已留联系方式的优质线索。") rest = q[10:] if rest: lines.append(f"\n### 其余已开发线索（{len(rest)} 条）") for r in rest: lines.append(f"- [{r['src']}] **{r['id']}** {r['name']}（{r['type']} / {r['district']}） 负责人：{r['owner'] or '未指派'} 派单：{r['level']}") if nc: lines.append(f"\n### 📞 无联系方式线索（置底，待补充）— {len(nc)} 条") for r in nc: lines.append(f"- [{r['src']}] **{r['id']}** {r['name']}（{r['type']} / {r['district']}） 状态：{r['status']} 负责人：{r['owner'] or '未指派'}") lines.append("") if bl: lines.append(f"## 四(b)、储备池（未开发批量记录，持续补充客户信息）") lines.append(f"- 未开发批量场站/提交（无负责人且无跟进）：**{bl}** 条 —— 持续录入客户信息、联系方式与跟进即可，不逐条告警。") lines.append(f"- 其中**无联系方式线索**见上方「📞 无联系方式线索」，请优先补录电话等联系方式。\n") lines.append("\n## 五、逾期清单") if overdue: for r in overdue: lines.append(f"- [{r['src']}] {r['id']} {r['name']} 已 {r['age_days']} 天无跟进") else: lines.append("- 无逾期项。\n") lines.append("\n## 六、按负责人分布") for owner, cc in by_owner.items(): lines.append(f"- {owner}：🔴{cc.get('red',0)} 🟡{cc.get('yellow',0)} 🟢{cc.get('green',0)}") lines.append("\n## 七、行动建议") lines.append("1. 红项当天由派单级别对应业务员跟进（脚本已自动分级派单+建跟进）。") lines.append("2. 逾期项优先回访，更新状态至跟进中/已闭环。") lines.append("3. 场地方/小区提交类（cooperation/community）务必在「红线待核」后进入评审闭环。") lines.append("4. 高盈利(L3)客户/场站由经理级亲自督办，开发线索(L1)由开发业务员持续渗透。") lines.append("5. 建议每日 08:00 自动跑本监控器，沉淀为日报。") if actions: lines.append(f"\n## 八、本次派单/跟进动作({len(actions)}条)") for a in actions: if a.get("assign"): lines.append(f"- ✅ {a['id']} [{a['src']}] 派单→{a.get('level','')}·{a.get('rep','')}（盈利分{a.get('score','')}） 跟进建:{a['followup']}") else: lines.append(f"- ➖ {a['id']} [{a['src']}] 级别{a.get('level','')} 派:{a.get('rep','')} 跟进建:{a['followup']} 指派:{a['assign']} {a.get('detail','')}") return "\n".join(lines) def build_csv(rows): import csv path = os.path.join(CONFIG["OUTDIR"], "跟进闭环明细.csv") with open(path, "w", newline="", encoding="utf-8-sig") as f: w = csv.writer(f) w.writerow(["来源","ID","名称/提交人","类型","区域/地址","填写时间","负责人","状态","跟进状态","RAG","逾期","年龄(天)","派单级别","盈利分","提示"]) for r in rows: w.writerow([r["src"],r["id"],r["name"],r["type"],r["district"],r["date"],r["owner"],r["status"], "已跟进" if r["followed"] else "未跟进", r["rag"], "是" if r["overdue"] else "", r["age_days"], r["level"], r["score"], r["tip"]]) return path # ============================ 入口 ============================ def main(): enforce_mode = "--enforce" in sys.argv AUTO = CONFIG["AUTO_FOLLOWUP"] or enforce_mode log(f"启动 跟进闭环监控器 | enforce={AUTO}") token = login() log("登录成功，开始聚合各实体记录…") rows = collect(token) log(f"聚合完成，共 {len(rows)} 条记录") actions = [] write_blocked = False score_rows(rows) # 先为所有记录算盈利分/级别/提示（纯展示） if AUTO: log("执行分级派单（按开发线索+盈利大小自动指派+对红项建跟进）…") actions, write_blocked = assign_tiered(rows, token) log(f"派单动作 {len(actions)} 条；写入阻断={write_blocked}") else: log("报告模式：仅展示分级建议，不写 CRM（如需落库请用 --enforce 或将 AUTO_FOLLOWUP 置 True）") # 输出 html_path = os.path.join(CONFIG["OUTDIR"], "跟进闭环看板.html") with open(html_path,"w",encoding="utf-8") as f: f.write(build_html(rows, actions, write_blocked)) report_path = os.path.join(CONFIG["OUTDIR"], "每日跟进汇报.md") with open(report_path,"w",encoding="utf-8") as f: f.write(build_report(rows, actions, write_blocked)) csv_path = build_csv(rows) log(f"已生成：\n - {html_path}\n - {report_path}\n - {csv_path}") # 控制台摘要 from collections import Counter c = Counter(r["rag"] for r in rows) q, nc = alert_rows(rows) print(f"\n===== 摘要 =====\n总记录:{len(rows)} 红:{c.get('red',0)} 黄:{c.get('yellow',0)} 绿:{c.get('green',0)} 逾期:{sum(1 for r in rows if r['overdue'])}") print(f" 已开发线索待跟进:{len(q)+len(nc)} (优质Top10见汇报/看板) 储备池(未开发批量):{backlog_count(rows)}") # 企业微信 / 个人微信 推送（“适时反馈数据”） red = [r for r in rows if r["rag"] == "red"] overdue = [r for r in rows if r["overdue"]] if WECOM["enabled"]: if WECOM["alert_only"] and not (red or overdue): log("alert_only 模式且当前无红项/逾期，跳过推送") else: wecom_send_text(build_wecom_text(rows, write_blocked)) # Gotify 自建推送（推到手机/桌面 Gotify App） if GOTIFY["enabled"]: if GOTIFY["alert_only"] and not (red or overdue): log("Gotify alert_only 模式且无红项/逾期，跳过") else: gotify_send_text(build_wecom_text(rows, write_blocked)) if write_blocked: print("⚠️ CRM 写入 EPERM 阻断：自动跟进未落库，请修复写入权限后重跑。") if __name__ == "__main__": main()