#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 客户ABCD智能优化 + 优选上传管线 v2（累积式存储，跨运行不丢数据） ==================================================================== 目标：各行业增量客户 → 去重合并 → 复合评分定 ABCD → 优选上传生产CRM(crm.ixgn.cn)。 - A/B 级：直接 POST 到 /api/target-sites（生产真实写入，带 1 条落库验证） - C 级（无联系方式/待补充）：进「待补录池」CSV，不传生产，供后续补录电话 - D 级（低价值/重复/黑名单）：丢弃 - 累积式 store：所有候选并入 customer_abcd_store.json，已传标 uploaded； 报告(明细/待补录/看板)从【全量store】生成 → 跨运行累积、不丢数据 - 本地检查点即 store：已处理 key 不再重复上传 → 支持「不间断增量」 安全： - token TTL≈1min，遇 401 自动重登录重试 - 批量上传前，先用第 1 条 A/B 做 POST+GET 落库验证（防 slice bug 丢数据） - target-sites 无 DELETE，上传即新增（unshift 到最前） 依赖：仅 Python 标准库。 用法： python customer_abcd_pipeline.py # 默认：福州/厦门爬取 + 喂入本地CSV + 上传A/B python customer_abcd_pipeline.py --city 漳州市 # 追加城市 python customer_abcd_pipeline.py --dry-run # 只分级不写生产 python customer_abcd_pipeline.py --max 50 # 单次最多上传50条 """ import urllib.request, urllib.parse, json, csv, time, sys, os, datetime, glob, re, ssl, collections # ============================ CONFIG ============================ CRM_BASE = "https://crm.ixgn.cn" USER, PW = "admin", "88073100" AMAP_KEYS = ["8ea4f0ac00dfb092d3e14d5776b1def9", ""] # ← 高德 WebService Key 池（免费个人Key，支持多Key轮换扩容）。第1个已填；把第2个空串替换成你的第二个高德Key即可自动轮换；腾讯仍作最终备用。 TENCENT_KEY = "LHOBZ-KKNWT-YMOXB-VSMJC-VJJJQ-BBB6P" # 腾讯地图 WebService Key（高德配额耗尽/未配置时作备用） FUJIAN = ["福州","厦门","漳州","泉州","三明","莆田","龙岩","南平","宁德","平潭"] PROMISING = ["物流","重卡","充电","停车","储能","光伏","网约车","货运","仓储","产业园","酒店","地锁","制造","工厂"] DEFAULT_CITIES = ["福州市","厦门市","泉州市","漳州市"] MAX_PAGE = 2 SCAN_COMP = False # 竞品扫描(耗配额)，默认关 MAX_UPLOAD_PER_RUN = 80 STORE = "customer_abcd_store.json" INDUSTRY_KW = { "物流园": ["物流园","物流园区","货运站","仓储物流","快递分拨中心","产业园"], "重卡超充": ["货车停车场","卡车停车场","卡车服务中心","加气站","物流基地","重卡充电"], "充电桩地锁": ["充电站","充电桩","地锁","车位锁","机械车位","新能源汽车充电","超充站"], "停车场": ["停车场","停车楼","立体车库","停车库"], "酒店": ["酒店","商务酒店","连锁酒店","度假酒店","宾馆","民宿"], "制造业": ["工业园","产业园区","制造厂","汽车制造","生产工厂","重卡制造"], "网约车": ["网约车司机驿站","出租车充电站","客运站","交通枢纽","滴滴驿站"], } CTX = ssl.create_default_context() # ============================ CRM 鉴权 ============================ _token = None def login(): global _token req = urllib.request.Request(CRM_BASE+"/api/auth/login", data=json.dumps({"username":USER,"password":PW}).encode(), headers={"Content-Type":"application/json"}) with urllib.request.urlopen(req, context=CTX, timeout=15) as r: _token = json.load(r).get("token") return _token def _hdr(): global _token if not _token: login() return {"Authorization":"Bearer "+_token, "Content-Type":"application/json"} def api_get(path): global _token for _ in range(3): try: req = urllib.request.Request(CRM_BASE+path, headers=_hdr()) with urllib.request.urlopen(req, context=CTX, timeout=25) as r: return json.load(r) except urllib.error.HTTPError as e: if e.code == 401: login(); continue raise raise Exception("GET failed "+path) def api_post(path, body): global _token for _ in range(3): try: req = urllib.request.Request(CRM_BASE+path, data=json.dumps(body).encode(), headers=_hdr(), method="POST") with urllib.request.urlopen(req, context=CTX, timeout=25) as r: return json.load(r) except urllib.error.HTTPError as e: if e.code == 401: login(); continue raise raise Exception("POST failed "+path) def api_post_batch(records): """批量新增：POST 数组到 /api/target-sites（生产已支持数组批量，单次数百条）。""" global _token for _ in range(3): try: req = urllib.request.Request(CRM_BASE+"/api/target-sites", data=json.dumps(records).encode(), headers=_hdr(), method="POST") with urllib.request.urlopen(req, context=CTX, timeout=60) as r: return json.load(r) except urllib.error.HTTPError as e: if e.code == 401: login(); continue raise raise Exception("BATCH POST failed") def as_list(d): if isinstance(d, list): return d if isinstance(d, dict): for k in ("items","records","data","list"): if isinstance(d.get(k), list): return d[k] return [] # ============================ 工具 ============================ def norm(s): return re.sub(r"\s+", "", str(s or "")).lower() def dedup_key(name, region): return norm(name) + "|" + norm(region) def is_fujian(region): return any(c in (region or "") for c in FUJIAN) def has_phone(p): p = str(p or "").strip() return bool(re.search(r"1\d{10}", p)) or bool(re.search(r"0\d{2,3}-?\d{7,8}", p)) def load_store(): if os.path.exists(STORE): try: return json.load(open(STORE, encoding="utf-8")) except: pass return {} def save_store(s): json.dump(s, open(STORE, "w", encoding="utf-8"), ensure_ascii=False, indent=2) def pull_crm_keys(): """拉取 CRM 现有 target-sites 的 key 集合（pageSize 足够大，覆盖全量）。""" keys = set() try: d = api_get("/api/target-sites?pageSize=20000&page=1") for t in as_list(d): keys.add(dedup_key(t.get("name"), t.get("region"))) except Exception as e: print(f" [CRM索引拉取失败] {e}（仅用本地store去重）") return keys # ============================ 来源：腾讯地图爬取 ============================ def tencent_search(keyword, city, page): params = {"keyword":keyword, "boundary":f"region({city},0)", "page_size":20, "page_index":page} url = "https://apis.map.qq.com/ws/place/v1/search?" + urllib.parse.urlencode(params) + "&key=" + TENCENT_KEY try: d = json.loads(urllib.request.urlopen(url, timeout=12).read().decode()) except Exception as e: return [], str(e) if d.get("status") != 0: return [], d.get("message","") return d.get("data", []), "" def _is_quota(err): """判断地图接口返回是否为配额/限流类错误（高德/腾讯共用）。""" return any(k in err for k in ("配额","121","INVALID","上限","调用量","每日", "frequency","LIMIT","QPS","OVER_LIMIT","QUOTA", "OVER_QUOTA","INVALID_USER_KEY","10017","10018", "EXCEED","DAY","PLAT_NOMATCH","OUT_OF_SERVICE")) def amap_search(keyword, city, page, key): """高德 WebService 文本搜索（指定key，支持多Key轮换）。成功返回 pois 列表，失败返回 ([], err)。""" if not key: return [], "NO_AMAP_KEY" params = {"keywords": keyword, "city": city.replace("市", ""), "page": page, "offset": 20} url = "https://restapi.amap.com/v3/place/text?" + urllib.parse.urlencode(params) + "&key=" + key try: d = json.loads(urllib.request.urlopen(url, timeout=12).read().decode()) except Exception as e: return [], str(e) if str(d.get("status")) != "1": # 附带 infocode，便于精确识别配额/限流（10017=超日配额, 10018=超QPS） return [], f"{d.get('info','')}|{d.get('infocode','')}" return d.get("pois", []), "" def _amap_key_pool(): """返回非空的高德Key列表（去除占位空串）。多Key用于额度耗尽时轮换。""" return [k for k in AMAP_KEYS if k and k.strip()] def scrape_poi(cities): """多源爬取：高德多Key轮换优先，全部高德Key额度耗尽时自动 fallback 腾讯。""" rows = [] keys = _amap_key_pool() amap_dead_idx = set() # 已耗尽额度的高德Key索引（进程内，本次运行有效） tencent_dead = False def next_amap_idx(exclude): """返回下一个未耗尽且未本轮尝试过的高德Key索引；全部耗尽返回 None。""" for i in range(len(keys)): if i not in amap_dead_idx and i not in exclude: return i return None for city in cities: for industry, kws in INDUSTRY_KW.items(): for kw in kws: for page in range(1, MAX_PAGE+1): src, pois, err = "", [], "" tried = set() ki = next_amap_idx(tried) while ki is not None and not pois: # 依次尝试每个高德Key，全部失败才转腾讯 tried.add(ki) pois, err = amap_search(kw, city, page, keys[ki]); src = "高德地图POI" if err: if _is_quota(err): amap_dead_idx.add(ki) remain = [i for i in range(len(keys)) if i not in amap_dead_idx] if remain: print(f" [高德Key{ki}配额耗尽] {kw}@{city}: {err} → 轮换至Key{remain[0]}") else: print(f" [高德全部Key配额耗尽] {kw}@{city}: {err} → 切腾讯备用") else: print(f" [高德告警] {kw}@{city}: {err}（转腾讯）") pois = [] ki = next_amap_idx(tried) if not pois and not tencent_dead: pois, err = tencent_search(kw, city, page); src = "腾讯地图POI" if err: if _is_quota(err): print(f" [腾讯配额耗尽] {kw}@{city}: {err} → 停止爬取"); tencent_dead = True; break else: print(f" [腾讯告警] {kw}@{city}: {err}（跳过关键词）"); break if not pois: break for p in pois: loc = p.get("location", {}) if isinstance(loc, str) and "," in loc: lng, lat = loc.split(",")[:2] else: lng = str(loc.get("lng", "")); lat = str(loc.get("lat", "")) ad = p.get("ad_info", {}).get("district", "") or p.get("adname", "") rows.append({ "name": p.get("title") or p.get("name", ""), "industry": industry, "type": p.get("category") or p.get("type", ""), "city": city.replace("市", ""), "district": ad, "region": f"{city.replace('市','')}·{ad}" if ad else city.replace("市", ""), "address": p.get("address", "") or "", "lng": str(lng), "lat": str(lat), "phone": p.get("tel", "") or "", "source": src, "blue_ocean": None, }) time.sleep(0.15) time.sleep(0.1) print(f" ✓ 城市 {city} 爬取累计 {len(rows)} 条") return rows # ============================ 来源：本地 CSV 喂入 ============================ def ingest_csv(): rows = [] files = sorted(set(glob.glob("物流园业主名单.csv") + glob.glob("候选场地_*.csv") + glob.glob("停车场可建桩明细.csv"))) for fp in files: try: with open(fp, encoding="utf-8-sig") as f: for r in csv.DictReader(f): rows.append(csv_to_record(r, fp)) except Exception as e: print(f" [CSV读取失败] {fp}: {e}") if files: print(f" ✓ CSV 喂入 {len(files)} 个文件 → {len(rows)} 条") return rows def _g(d, *keys): for k in keys: if k in d and str(d[k]).strip(): return d[k] return "" def csv_to_record(r, fp): name = _g(r, "公司名","名称","name") phone = _g(r, "电话","tel","contactPhone","手机") addr = _g(r, "注册地址","地址","address") city = _g(r, "城市","city") dist = _g(r, "区县","district","区") if not city: for c in FUJIAN: if c in (addr+name): city = c; break region = f"{city}·{dist}" if (city and dist) else (city or "") typ = _g(r, "类型","type","category","经营范围") industry = "物流园" if any(k in (name+typ) for k in ["物流","仓储","产业园"]) else "充电桩地锁" if any(k in (name+typ) for k in ["重卡","货车","卡车"]): industry="重卡超充" if any(k in (name+typ) for k in ["停车"]): industry="停车场" if any(k in (name+typ) for k in ["酒店","宾馆","民宿"]): industry="酒店" if any(k in (name+typ) for k in ["制造","工厂","工业园"]): industry="制造业" return {"name":name,"industry":industry,"type":typ,"city":city,"district":dist, "region":region,"address":addr,"lng":"","lat":"","phone":phone, "source":f"CSV·{os.path.basename(fp)}","blue_ocean":None} # ============================ 评分 → ABCD ============================ def score_record(rec): reasons = []; s = 0 if has_phone(rec.get("phone")): s += 30; reasons.append("有联系方式+30") else: reasons.append("无联系方式(上限C)") if any(k in (rec.get("industry","")+rec.get("type","")) for k in PROMISING): s += 20; reasons.append("行业匹配+20") if is_fujian(rec.get("region")): s += 15; reasons.append("福建区域+15") if str(rec.get("address") or "").strip(): s += 10; reasons.append("有地址+10") if str(rec.get("lng") or "") and str(rec.get("lat") or ""): s += 10; reasons.append("有坐标+10") bo = rec.get("blue_ocean") if bo is not None: if bo <= 1: s += 15; reasons.append("蓝海(周边≤1充电站)+15") elif bo <= 3: s += 8; reasons.append("竞争较少+8") if has_phone(rec.get("phone")) and str(rec.get("address") or "").strip() and is_fujian(rec.get("region")): s += 5; reasons.append("信息完整+5") s = min(100, s) no_phone = not has_phone(rec.get("phone")) if no_phone: grade = "C" if s >= 40 else "D" elif s >= 80: grade = "A" elif s >= 60: grade = "B" elif s >= 40: grade = "C" else: grade = "D" return s, grade, "；".join(reasons) # ============================ 主流程 ============================ def main(): args = sys.argv[1:] dry_run = "--dry-run" in args all_up = "--all" in args max_up = MAX_UPLOAD_PER_RUN for i,a in enumerate(args): if a == "--city" and i+1 < len(args): DEFAULT_CITIES.append(args[i+1]) if a == "--max" and i+1 < len(args): try: max_up = int(args[i+1]) except: pass if all_up: max_up = 10**9 # --all：解除单次上限，全部导入 cities = list(dict.fromkeys(DEFAULT_CITIES)) now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") print(f"\n===== 客户ABCD管线 {now} | 城市:{cities} | dry_run={dry_run} | max_upload={max_up} =====") # 1) 采集 print("[1/5] 采集增量客户...") recs = scrape_poi(cities) + ingest_csv() print(f" 原始候选: {len(recs)} 条") # 2) 累积式 store upsert print("[2/5] 合并入累积 store（去重/更新）...") store = load_store() crm_keys = pull_crm_keys() for r in recs: k = dedup_key(r["name"], r["region"]) if not k: continue s, g, reason = score_record(r) rec = store.get(k) or {} rec.update({"key":k, "name":r["name"], "industry":r["industry"], "region":r["region"], "address":r["address"], "phone":r["phone"], "source":r["source"], "score":s, "grade":g, "reason":reason, "first_seen": rec.get("first_seen", now)}) rec["last_seen"] = now # 已在 CRM 或已上传 → 标 uploaded if rec.get("uploaded") or k in crm_keys: rec["uploaded"] = True else: rec.setdefault("uploaded", False) store[k] = rec print(f" store 总量: {len(store)} 条（CRM已有 {len(crm_keys)} 条已标上传）") # 3) 分级统计（来自全量 store） print("[3/5] 评分定级 ABCD...") gc = collections.Counter(r["grade"] for r in store.values()) print(f" 分级: A={gc.get('A',0)} B={gc.get('B',0)} C={gc.get('C',0)} D={gc.get('D',0)}") # 4) 上传 A/B/C（全部导入；先验证落库，再批量提速） grade_set = ("A","B","C") # A/B/C 全部导入；D(低价值)丢弃 print(f"[4/5] 上传 A/B/C 到生产CRM（{'全部导入' if all_up else f'单次上限{max_up}'}）...") ab = sorted([r for r in store.values() if r["grade"] in grade_set and not r.get("uploaded")], key=lambda x:-x["score"]) uploaded, failed = [], [] if not ab: print(" 无待上传 A/B/C（全部已传或落库验证未过）。") elif dry_run: print(f" [dry-run] 拟上传 {len(ab)} 条 A/B/C，未写生产。") else: # 落库验证（用第1条做单条 POST+GET） probe = ab[0] ok = verify_upload(probe) verified = set() if not ok: print(" ⚠️ 落库验证失败，中止上传（保护数据，不丢不重）。请检查生产写入权限。") ab = [] else: verified.add(probe["key"]); store[probe["key"]]["uploaded"] = True pending = [r for r in ab[:max_up] if r["key"] not in verified and r["key"] not in crm_keys] # 批量上传：每批 200 条 BATCH = 200 for i in range(0, len(pending), BATCH): chunk = pending[i:i+BATCH] bodies = [{ "name": r["name"], "type": r["industry"], "region": r["region"], "address": r["address"] or r["region"], "contactName": "", "contactPhone": r["phone"] or "", "contact": r["source"], "status": "待开发", "transformerStatus": "待确认", "priority": "高" if r["grade"]=="A" else ("中" if r["grade"]=="B" else "低"), "note": f"【ABCD·{r['grade']}·{r['score']}分】{r['reason']}\n来源:{r['source']}", } for r in chunk] try: res = api_post_batch(bodies) cnt = (res or {}).get("count", len(chunk)) for r in chunk: store[r["key"]]["uploaded"] = True uploaded.extend(r["key"] for r in chunk) print(f" 批量 {i//BATCH+1}: 提交 {len(chunk)} 条，服务端返回新增 {cnt}") except Exception as e: for r in chunk: failed.append((r["name"], str(e))) print(f" ✅ 上传 {len(uploaded)} 条 | 失败 {len(failed)} 条") c_pool = [r for r in store.values() if r["grade"] == "C" and not r.get("uploaded")] d_drop = [r for r in store.values() if r["grade"] == "D"] # 5) 产出报告（全量 store） print("[5/5] 生成报告...") write_reports(store, ab, uploaded, c_pool, d_drop, now, dry_run) save_store(store) print(f"\n===== 完成 ===== A={gc.get('A',0)} B={gc.get('B',0)} C(待上传/未上传)={len(c_pool)} D(丢弃)={len(d_drop)} | 本次上传={len(uploaded)} | store={len(store)}") def verify_upload(rec): body = { "name": rec["name"], "type": rec["industry"], "region": rec["region"], "address": rec["address"] or rec["region"], "contactName": "", "contactPhone": rec["phone"] or "", "contact": rec["source"]+"·落库验证", "status":"待开发", "transformerStatus":"待确认", "priority":"高", "note": f"【ABCD验证·{rec['grade']}·{rec['score']}分】{rec['reason']}", } try: res = api_post("/api/target-sites", body) rid = (res or {}).get("record",{}).get("id") or (res or {}).get("id") if not rid: print(f" [验证] POST返回但无id: {res}"); return False time.sleep(0.3) got = api_get("/api/target-sites/"+rid) if got and (got.get("id")==rid or got.get("name")==rec["name"]): print(f" ✅ 落库验证通过（{rid} 已持久化）→ 继续批量上传"); return True print(f" ⚠️ 落库验证失败：GET {rid} 未返回记录（疑似 slice bug 丢数据）"); return False except Exception as e: print(f" ⚠️ 落库验证异常: {e}"); return False def write_reports(store, ab, uploaded, c_pool, d_drop, now, dry_run): up_set = set(uploaded) allr = sorted(store.values(), key=lambda x:(-{"A":3,"B":2,"C":1,"D":0}.get(x["grade"],0), -x["score"])) # ---- 明细 CSV（全量） ---- with open("客户ABCD明细.csv","w",encoding="utf-8-sig",newline="") as f: w = csv.writer(f) w.writerow(["等级","评分","名称","行业","区域","地址","电话","来源","评分依据","是否已上传"]) for r in allr: w.writerow([r["grade"], r["score"], r["name"], r["industry"], r["region"], r["address"], r["phone"], r["source"], r["reason"], "是" if r.get("uploaded") else "否"]) # ---- 待补录池 CSV（累积，全量 C） ---- with open("待补录客户_无联系方式.csv","w",encoding="utf-8-sig",newline="") as f: w = csv.writer(f) w.writerow(["名称","行业","区域","地址","来源","评分","建议"]) for r in sorted(c_pool, key=lambda x:-x["score"]): w.writerow([r["name"], r["industry"], r["region"], r["address"], r["source"], r["score"], "补录联系电话后可升为B/A级上传"]) # ---- MD 报告 ---- gc = collections.Counter(r["grade"] for r in store.values()) L = [f"# 客户ABCD智能分级报告\n生成：{now} CRM：{CRM_BASE}\n", "## 一、概览", f"- 累积客户池（store）：**{len(store)}**", f"- 🅰️ A级（优质·上传）：**{gc.get('A',0)}**", f"- 🅱️ B级（良·上传）：**{gc.get('B',0)}**", f"- 🅲️ C级（无联系方式·待补录）：**{len(c_pool)}**", f"- 🅳️ D级（低价值/重复·丢弃）：**{len(d_drop)}**", f"- 本次上传生产：{'dry-run未上传' if dry_run else len(uploaded)} 条（A/B/C 全部导入，D 丢弃）\n", "## 二、🅰️ A级 Top 推荐（已/将上传）"] for r in [x for x in allr if x["grade"]=="A"][:10]: tag = "✅已上传" if r.get("uploaded") else ("(dry-run)" if dry_run else "待上传") L.append(f"- **{r['name']}** （{r['industry']} / {r['region']}） {r['score']}分 · {r['phone'] or '无电话'} · {tag}") L.append("\n## 三、🅲️ C级待补录（无联系方式，置底供补充）") for r in sorted(c_pool, key=lambda x:-x["score"])[:15]: L.append(f"- {r['name']} （{r['industry']} / {r['region']}） {r['score']}分 · 来源:{r['source']}") if len(c_pool) > 15: L.append(f"- …其余 {len(c_pool)-15} 条见《待补录客户_无联系方式.csv》") L.append("\n## 四、评分规则（可解释/智能化）") L.append("- 有联系方式 +30 ｜ 行业匹配 +20 ｜ 福建区域 +15 ｜ 有地址 +10 ｜ 有坐标 +10 ｜ 蓝海(周边≤1充电站) +15 ｜ 信息完整 +5") L.append("- A≥80(须有电话) ｜ B 60-79(须有电话) ｜ C 40-59或无电话(待补录) ｜ D<40(丢弃)") L.append("\n> 不间断增量：已处理/已上传记录写入 store，下次运行自动跳过，仅上传新客户。C级补录电话后可升A/B。") open("客户ABCD分级报告.md","w",encoding="utf-8").write("\n".join(L)) write_kanban(store, ab, uploaded, c_pool, d_drop, now, dry_run) def write_kanban(store, ab, uploaded, c_pool, d_drop, now, dry_run): up_set = set(uploaded) def card(r): up = "✅已上传" if r.get("uploaded") else ("(dry-run)" if dry_run else "待上传") return f'
{r["name"]}
{r["industry"]} · {r["region"]}
{r["score"]}分 · {r["phone"] or "无电话"}
{up}
' gc = collections.Counter(r["grade"] for r in store.values()) a_cards = "".join(card(r) for r in sorted([x for x in store.values() if x["grade"]=="A"], key=lambda x:-x["score"])[:12]) b_cards = "".join(card(r) for r in sorted([x for x in store.values() if x["grade"]=="B"], key=lambda x:-x["score"])[:12]) c_cards = "".join(card(r) for r in sorted(c_pool, key=lambda x:-x["score"])[:12]) d_cards = "".join(card(r) for r in sorted(d_drop, key=lambda x:-x["score"])[:12]) html = f""" 客户ABCD分级看板

客户ABCD智能分级看板

生成 {now} · CRM {CRM_BASE} · 累积store {len(store)} 条 · {'[dry-run 未上传]' if dry_run else 'A/B/C 已上传生产'}
🅰️ A {gc.get('A',0)}🅱️ B {gc.get('B',0)} 🅲️ C {len(c_pool)}🅳️ D {len(d_drop)} 本次上传 {len(uploaded)}

🅰️ A级 · 优质

{a_cards or '无'}

🅱️ B级 · 良

{b_cards or '无'}

🅲️ C级 · 待补录（无联系方式）

{c_cards or '无'}

🅳️ D级 · 丢弃

{d_cards or '无'}

不间断增量：已处理记录写入 store，下次运行自动跳过，仅上传新客户。C级补录电话后可升A/B。

""" open("客户ABCD分级看板.html","w",encoding="utf-8").write(html) if __name__ == "__main__": main()