# -*- coding: utf-8 -*- """ 每日优质客户智能推荐引擎（recommend_daily.py） ------------------------------------------------ 目标：每天动态推荐一批「最值得拨打」的客户，且每天推荐不同的客户（轮换）。 五维质量评分（各 0-100，加权合成综合分）： 1) 投入回报 ROI 权重 25% —— 行业回报基数 + 规模/等级加成 2) 位置 Location 权重 20% —— 城市能级 + 是否城市核心区 3) 投资安全性 Safety 权重 20% —— 可联系(有电话) + 地址可核验 + 等级 4) 周边价值 Surround 权重 20% —— 行业可建桩/引流价值（蓝海代理） 5) 成交难易 Deal 权重 15% —— 有电话易触达 + 本地业主易决策 - 大集团难啃 数据来源：customer_abcd_store.json（已去重/评分/上传的客户池）。 说明：store 仅持久化 name/industry/region/address/phone/source/score/grade， 无坐标/蓝海字段，故位置与周边价值以「城市能级 / 行业可建桩价值」作为可解释代理。 若后续 store 补齐 lng/lat 与 blue_ocean，本引擎可无缝升级为更精确的距离/竞争评分。 轮换机制：recommend_history.json 记录每家推荐日期与次数； 被推荐过的客户 30 天内不再进入「新推」，优先把新鲜客户推到前面， 保证每天动态推荐不同客户；池子耗尽才回退「复推」。 """ import sys, os, json, datetime, collections import customer_abcd_pipeline as P # 复用 is_fujian / has_phone / STORE HERE = os.path.dirname(os.path.abspath(__file__)) STORE = os.path.join(HERE, P.STORE) HISTORY = os.path.join(HERE, "recommend_history.json") OUT_HTML = os.path.join(HERE, "每日优质客户推荐.html") OUT_MD = os.path.join(HERE, "每日优质客户推荐.md") OUT_CSV = os.path.join(HERE, "每日优质客户推荐.csv") # ---------------- 可调参数 ---------------- DEFAULT_PROVINCE = "福建" # 仅推荐该省（与「选对省份」口径一致） DEFAULT_N = 30 # 每天推荐数量 ROTATION_DAYS = 30 # 推荐后锁定天数（期间不重复新推） INDUSTRY_CAP_RATIO = 0.40 # 单一行业最多占每日推荐比例（保证行业内交叉多样） # 品牌锚点租户（自带高 EV 流量 / 高信任，ROI 与周边价值加成） BRAND_ANCHOR = ["小米","比亚迪","蔚来","理想","小鹏","特斯拉","普利司通","米其林","驰加", "顺丰","京东","中通","德邦","国家电网","特来电","星星充电","宁德时代"] import re def phone_info(p): """返回 (是否手机号直拨, 是否多号码)。手机号更易触达、更安全。""" p = str(p or "") nums = re.findall(r"1\d{10}|\d{3,4}-?\d{7,8}", p) mob = any(n.startswith("1") and len(n.replace("-","")) == 11 for n in nums) multi= len(nums) > 1 return mob, multi def brand_anchor(name): return any(b in (name or "") for b in BRAND_ANCHOR) # 城市能级（福建内） CITY_TIER = {"福州":100,"厦门":100,"泉州":85,"漳州":85,"莆田":75,"宁德":72, "龙岩":72,"平潭":75,"三明":70,"南平":70} # 行业 → 投入回报基数 INDUSTRY_ROI = {"重卡超充":95,"充电桩地锁":90,"网约车":88,"物流园":82,"制造业":78, "停车场":72,"酒店":70} # 行业 → 周边价值/可建桩引流价值 INDUSTRY_HOST = {"重卡超充":95,"充电桩地锁":90,"网约车":88,"物流园":85,"制造业":80, "停车场":78,"酒店":75} # 大集团/难啃品牌（成交难度 -） BIG_BRAND = ["顺丰","京东","中通","圆通","申通","韵达","百世","菜鸟","德邦","邮政","EMS", "美团","滴滴","货拉拉","国家电网","南方电网","中石化","中石油","集团","股份", "有限公司","国企","连锁","万达","万科","碧桂园","韵达","跨越"] # 本地小业主（成交难度 +，业主直接决策） SMALL_LOCAL = ["经营部","分部","门市","店","个体","便民","社区","服务部","经营"] # ---------------- 共享模式匹配 ---------------- # 按场站行业预匹配最适用的充电共享部署模式（用于打单话术与方案推荐） SHARING_MODE = { "物流园":"多车一桩", "重卡超充":"多车一桩", "制造业":"多车一桩", "充电桩地锁":"多车一桩", "酒店":"社区分时", "社区物业":"社区分时", "商业综合体":"社区分时", "商务办公":"社区分时", "停车场":"临近车位共享", "网约车":"临近车位共享", "充电桩":"临近车位共享", } MODE_INFO = { "多车一桩": "一台多枪直流桩/轮充排插服务多车，功率动态分配；适配长停车队、重卡超充，单桩周转最高。", "社区分时": "分时预约错峰充电，居民夜间优先、白天开放营运车；谷电占比高、利用率破60%。", "临近车位共享": "1台交流桩覆盖2-3相邻车位，App预约+地锁联动；少建桩、降Capex、提利用率。", } def sharing_mode(r): m = SHARING_MODE.get(r.get("industry",""), "临近车位共享") return m, MODE_INFO.get(m,"") WEIGHTS = {"roi":.25,"loc":.20,"safety":.20,"surround":.20,"deal":.15} def clamp(x): return max(0, min(100, x)) def city_of(region): region = region or "" return region.split("·")[0] if "·" in region else region def district_of(region): region = region or "" return region.split("·")[-1] if "·" in region else "" CORE_DISTRICTS = ["鼓楼","思明","湖里","仓山","晋安","台江","马尾","芗城","鲤城","丰泽", "洛江","新罗","梅列","三元","龙文","海沧","集美","同安","翔安","蕉城"] # ---------------- 五维评分 ---------------- def score_roi(r): base = INDUSTRY_ROI.get(r.get("industry",""), 65) bonus = 0 if r.get("grade") == "A": bonus += 5 nm = r.get("name","") if any(k in nm for k in ["物流园","产业园","工业园","基地","中心","分拨","仓储","超充"]): bonus += 5 if brand_anchor(nm): bonus += 5 # 品牌锚点自带高 EV 流量 return clamp(base + bonus) def score_location(r): tier = CITY_TIER.get(city_of(r.get("region","")), 70) bonus = 0 d = district_of(r.get("region","")) if "区" in d: bonus += 10 if any(cd in d for cd in CORE_DISTRICTS): bonus += 5 addr = str(r.get("address") or "").strip() if len(addr) >= 25: bonus += 5 # 地址越精确越易定位 elif addr: bonus += 2 return clamp(tier + bonus) def score_safety(r): s = 50 if P.has_phone(r.get("phone")): s += 25 if str(r.get("address") or "").strip(): s += 10 if r.get("grade") == "A": s += 10 elif r.get("grade") == "B": s += 5 mob, multi = phone_info(r.get("phone")) if mob: s += 5 # 手机直拨更易核验、更安全 if multi: s += 5 # 多号码冗余联系 return clamp(s) def score_surround(r): base = INDUSTRY_HOST.get(r.get("industry",""), 65) bonus = 5 if r.get("grade") == "A" else 0 if brand_anchor(r.get("name","")): bonus += 5 # 锚点租户周边引流价值更高 return clamp(base + bonus) def score_deal(r): s = 60 if P.has_phone(r.get("phone")): s += 20 if r.get("grade") == "A": s += 10 nm = r.get("name","") if any(b in nm for b in BIG_BRAND): s -= 30 if any(sm in nm for sm in SMALL_LOCAL): s += 5 if phone_info(r.get("phone"))[0]: s += 5 # 手机直拨成交更快 return clamp(s) def composite(sc): return round(sum(WEIGHTS[k]*sc[k] for k in WEIGHTS), 1) # ---------------- 推荐理由 / 拨打建议 ---------------- def build_reason(r, sc): parts = [] if sc["roi"] >= 85: parts.append(f"回报高({r['industry']})") if sc["loc"] >= 90: parts.append("核心区位") elif sc["loc"] >= 75: parts.append("区位良好") if sc["safety"] >= 85: parts.append("信息可核验·安全") if sc["surround"] >= 85: parts.append("周边建桩价值高") if sc["deal"] >= 85: parts.append("易成交(本地业主·有电话)") elif sc["deal"] <= 50: parts.append("大集团·需走采购") if not parts: parts.append("综合优质") return "；".join(parts) def build_tip(r): nm = r.get("name","") if any(b in nm for b in BIG_BRAND): return "集团/连锁客户：先找能源管理或行政负责人，走采购/集采流程，主推充电桩分成+光储充一体化方案。" if P.has_phone(r.get("phone")): return "直接拨打，业主决策快：主推轻资产合作（场地提供+分成），强调零投入、480KW超充与回本模型。" return "暂无电话，建议地图/企查查补全联系方式后再触达。" # ---------------- 主流程 ---------------- def main(): args = sys.argv[1:] dry = "--dry-run" in args n = DEFAULT_N prov = DEFAULT_PROVINCE rot = ROTATION_DAYS for i,a in enumerate(args): if a == "--n" and i+1 < len(args): try: n = int(args[i+1]) except: pass if a == "--province" and i+1 < len(args): prov = args[i+1] if a == "--days" and i+1 < len(args): try: rot = int(args[i+1]) except: pass today = datetime.date.today() now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") # 1) 载入客户池 with open(STORE, encoding="utf-8") as f: store = json.load(f) recs = list(store.values()) # 2) 资格过滤：A/B + 有电话 + 目标省份 def in_prov(r): return P.is_fujian(r.get("region","")) if prov == "福建" else (prov in (r.get("region","") or "")) eligible = [r for r in recs if r.get("grade") in ("A","B") and P.has_phone(r.get("phone")) and in_prov(r)] print(f"[1] 合格池(A/B+有电话+{prov}): {len(eligible)} 家 | 每日推荐 {n} 家 | 轮换锁定 {rot} 天 | dry_run={dry}") # 3) 五维评分 + 综合分 for r in eligible: sc = {"roi":score_roi(r),"loc":score_location(r),"safety":score_safety(r), "surround":score_surround(r),"deal":score_deal(r)} r["_sc"] = sc r["_comp"] = composite(sc) eligible.sort(key=lambda r:(-r["_comp"], -r.get("score",0), r.get("name","") or "")) # 4) 轮换选择（先新鲜；行业上限保证多样） history = {} if os.path.exists(HISTORY): try: history = json.load(open(HISTORY, encoding="utf-8")) except: history = {} def fresh(r): h = history.get(r["key"]) if not h or not h.get("last"): return True try: last = datetime.date.fromisoformat(h["last"]) except: return True return (today - last).days >= rot cap = max(1, round(n * INDUSTRY_CAP_RATIO)) # 单行业上限 picked, ind_cnt, used = [], collections.Counter(), set() # 第一遍：新鲜 + 行业上限（优先把各行业最优推上来，保证多样） for r in eligible: if len(picked) >= n: break if not fresh(r): continue if ind_cnt[r.get("industry","?")] >= cap: continue picked.append((r, "新推")); used.add(id(r)) ind_cnt[r.get("industry","?")] += 1 # 第二遍：新鲜但超出行业上限的，补足名额（仍只推新鲜） if len(picked) < n: for r in eligible: if len(picked) >= n: break if fresh(r) and id(r) not in used: picked.append((r, "新推")); used.add(id(r)) # 第三遍：池子不足时回退复推 if len(picked) < n: for r in eligible: if len(picked) >= n: break if id(r) not in used: picked.append((r, "复推")); used.add(id(r)) new_cnt = sum(1 for _,t in picked if t == "新推") re_cnt = sum(1 for _,t in picked if t == "复推") print(f"[2] 今日推荐 {len(picked)} 家（新推 {new_cnt} / 复推 {re_cnt}）｜ 单行业上限 {cap}") # 5) 组装推荐列表 rows = [] for idx,(r,t) in enumerate(picked, 1): sc = r["_sc"] m, mnote = sharing_mode(r) rows.append({ "rank": idx, "tag": t, "name": r.get("name",""), "industry": r.get("industry",""), "region": r.get("region",""), "phone": r.get("phone",""), "grade": r.get("grade",""), "roi": sc["roi"], "loc": sc["loc"], "safety": sc["safety"], "surround": sc["surround"], "deal": sc["deal"], "comp": r["_comp"], "mode": m, "mode_note": mnote, "reason": build_reason(r, sc), "tip": build_tip(r), }) # 6) 写历史（非 dry-run） if not dry: for r,t in picked: h = history.get(r["key"], {"last":None,"count":0}) h["last"] = today.isoformat(); h["count"] = h.get("count",0) + 1 history[r["key"]] = h json.dump(history, open(HISTORY,"w",encoding="utf-8"), ensure_ascii=False, indent=1) print(f"[3] 已更新推荐历史：累计已推荐 {len(history)} 家") # 7) 汇总 city_dist = collections.Counter(city_of(r["region"]) for r,_ in picked) ind_dist = collections.Counter(r["industry"] for r,_ in picked) mode_dist = collections.Counter(sharing_mode(r)[0] for r,_ in picked) summary = { "date": now, "province": prov, "total": len(picked), "new": new_cnt, "re": re_cnt, "city_dist": dict(city_dist), "ind_dist": dict(ind_dist), "mode_dist": dict(mode_dist), "avg_comp": round(sum(x["comp"] for x in rows)/max(1,len(rows)),1) if rows else 0, "eligible_pool": len(eligible), } # 8) 输出 write_html(rows, summary) write_md(rows, summary) write_csv(rows) print(f"[4] 输出：\n {OUT_HTML}\n {OUT_MD}\n {OUT_CSV}") return rows, summary # ---------------- 输出：HTML ---------------- def _bar(v, color): return (f'
' f'
{v}
') def write_html(rows, s): cards = [] for x in rows: cards.append(f""" {x['rank']}{x['tag']} {x['name']}
{x['industry']} · {x['region']} · {x['grade']}级 · {x['mode']}
{x['phone']} {_bar(x['roi'],'#e74c3c')}回报 {_bar(x['loc'],'#3498db')}位置 {_bar(x['safety'],'#27ae60')}安全 {_bar(x['surround'],'#9b59b6')}周边 {_bar(x['deal'],'#f39c12')}难易 {x['comp']} {x['reason']}
🔗 {x['mode']}：{x['mode_note']}
{x['tip']} """) city_items = " ".join(f"{k} {v}" for k,v in s["city_dist"].items()) ind_items = " ".join(f"{k} {v}" for k,v in s["ind_dist"].items()) mode_items = " ".join(f"{k} {v}" for k,v in s["mode_dist"].items()) html = f""" 每日优质客户智能推荐 {s['date']}

📞 每日优质客户智能推荐

生成时间 {s['date']} ｜ 省份 {s['province']} ｜ 合格池 {s['eligible_pool']} 家 ｜ 轮换锁定 {ROTATION_DAYS} 天
{s['total']}今日推荐(家)
{s['new']}新推
{s['re']}复推
{s['avg_comp']}平均综合分
城市分布：{city_items}
行业分布：{ind_items}
共享模式分布：{mode_items}
{''.join(cards)}
序客户电话五维评分(回报/位置/安全/周边/难易)综合推荐理由 / 拨打建议
""" open(OUT_HTML,"w",encoding="utf-8").write(html) # ---------------- 输出：MD ---------------- def write_md(rows, s): L = [] L.append(f"# 每日优质客户智能推荐（{s['date']}）\n") L.append(f"- 省份口径：{s['province']} ｜ 合格池(A/B+有电话)：{s['eligible_pool']} 家") L.append(f"- 今日推荐 **{s['total']}** 家（新推 {s['new']} / 复推 {s['re']}），平均综合分 {s['avg_comp']}") L.append(f"- 城市分布：{ '、'.join(f'{k}{v}' for k,v in s['city_dist'].items()) }") L.append(f"- 行业分布：{ '、'.join(f'{k}{v}' for k,v in s['ind_dist'].items()) }\n") L.append("## 推荐清单\n") L.append("| # | 客户 | 行业/区域 | 电话 | 回报 | 位置 | 安全 | 周边 | 难易 | 综合 | 建议共享模式 | 推荐理由 |") L.append("|---|------|-----------|------|------|------|------|------|------|------|------------|----------|") for x in rows: L.append(f"| {x['rank']} | {x['name']} | {x['industry']}·{x['region']} | {x['phone']} | " f"{x['roi']} | {x['loc']} | {x['safety']} | {x['surround']} | {x['deal']} | " f"**{x['comp']}** | {x['mode']} | {x['reason']} |") L.append("\n## 拨打建议（按共享模式）\n") for x in rows[:8]: L.append(f"- **{x['name']}**（{x['industry']} · {x['mode']}）：{x['mode_note']} {x['tip']}") open(OUT_MD,"w",encoding="utf-8").write("\n".join(L)) # ---------------- 输出：CSV ---------------- def write_csv(rows): import csv with open(OUT_CSV,"w",encoding="utf-8-sig",newline="") as f: w = csv.writer(f) w.writerow(["序号","客户","行业","区域","电话","等级","投入回报","位置","投资安全性", "周边价值","成交难易","综合分","建议共享模式","共享模式说明","推荐理由","拨打建议","标签"]) for x in rows: w.writerow([x["rank"],x["name"],x["industry"],x["region"],x["phone"],x["grade"], x["roi"],x["loc"],x["safety"],x["surround"],x["deal"],x["comp"], x["mode"],x["mode_note"],x["reason"],x["tip"],x["tag"]]) if __name__ == "__main__": main()