#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 票据 05：公海池与线索种子数据铺设脚本（幂等可重跑） 覆盖目标（frontend-integration/issues/05-testdata-seeding.md）： 1. 公海池 ×2（不同部门、不同省份、不同领取规则），各配负责人 2. 线索种子：待领取 / 已领取 / 跟进中 / 已转商机（尽力，商机模块未实现时记录缺口） 3. 「我的关注」有数据（follow 一条线索） 4. 验收：四视图分页均有数据，打印统计 用法： python seed-testdata.py # 默认 http://localhost:8081 SEED_BASE=http://x.x.x.x python seed-testdata.py python seed-testdata.py --user 739564171091247104 前置：verify profile 运行中（/api/auth/debug/token 可用）；目标用户具备 ROLE_ADMIN。 幂等策略：部门/公海池按名称匹配，线索按 leadName 精确匹配，流转按当前状态守卫。 """ import json import os import sys import urllib.parse import urllib.request BASE = os.environ.get("SEED_BASE", "http://localhost:8081") USER_ID = os.environ.get("SEED_USER_ID", "739564171091247104") STATUS_NAMES = {1: "未分发", 2: "待领取", 3: "已领取", 4: "跟进中", 5: "已转商机", 6: "过期失效", 7: "线索作废"} DEPT_NAME = "华南销售部(seed)" POOL_A = { "poolName": "华南公海池(seed)", "claimRule": 1, "recycleDays": 7, "expireDays": 180, "dailyClaimLimit": 10, "holdLimit": 200, "provinceCode": "440000", "cityCode": "440100", } POOL_B = { "poolName": "华北公海池(seed)", "claimRule": 2, "recycleDays": 10, "expireDays": 90, "dailyClaimLimit": 5, "holdLimit": 100, "provinceCode": "110000", "cityCode": None, } # 线索种子：target 流转目标（claim=已领取 follow=跟进中 convert=已转商机），follow_me=当前用户关注 LEADS = [ {"leadName": "网络奇迹有限公司", "phone": "15611229615", "pool": "A", "target": None, "provinceCode": "440000", "cityCode": "440100", "channelCode": "exhibition", "brandCode": "itc", "productCode": "led_display", "sceneCode": "meeting_room", "consultContent": "展会现场咨询LED显示屏，意向会议室改造项目", "follow_me": True}, {"leadName": "广州智联科技有限公司", "phone": "13800138001", "pool": "A", "target": "claim", "provinceCode": "440000", "cityCode": "440100", "channelCode": "jd", "brandCode": "shichang", "productCode": "conference_system", "sceneCode": "auditorium", "consultContent": "京东渠道咨询会议系统", "follow_me": False}, {"leadName": "深圳华彩光电有限公司", "phone": "13800138002", "pool": "A", "target": "follow", "provinceCode": "440000", "cityCode": "440300", "channelCode": "zhihu", "brandCode": "itc", "productCode": "video_conference", "sceneCode": "command_center", "consultContent": "知乎渠道咨询视频会议+指挥中心场景", "follow_me": False}, {"leadName": "东莞声视电子有限公司", "phone": "13800138003", "pool": "A", "target": "convert", "provinceCode": "440000", "cityCode": "441900", "channelCode": "digital_av", "brandCode": "three_a", "productCode": "sound_system", "sceneCode": "sports_venue", "consultContent": "数字音视渠道咨询扩声系统（文体场馆）", "follow_me": False}, {"leadName": "北京京彩传媒有限公司", "phone": "13800138004", "pool": "B", "target": None, "provinceCode": "110000", "cityCode": None, "channelCode": "netease_news", "brandCode": "itc", "productCode": "record_broadcast", "sceneCode": "theater", "consultContent": "网易新闻渠道咨询录播系统（剧院）", "follow_me": True}, ] FINDINGS = [] # 铺设过程发现的阻塞/异常（票据第 4 项） def http(method, path, params=None, form=None, token=None): url = BASE + path if params: url += "?" + urllib.parse.urlencode(params, doseq=True) data = None if form is not None: form = {k: v for k, v in form.items() if v is not None} data = urllib.parse.urlencode(form, doseq=True).encode("utf-8") req = urllib.request.Request(url, data=data, method=method) if token: req.add_header("Authorization", "Bearer " + token) try: with urllib.request.urlopen(req, timeout=60) as resp: body = json.loads(resp.read().decode("utf-8")) except urllib.error.HTTPError as e: body = json.loads(e.read().decode("utf-8") or "{}") if body.get("success") is not True: raise RuntimeError(f"{method} {path} -> code={body.get('code')} msg={body.get('message')}") return body.get("data") def main(): user_id = USER_ID for i, a in enumerate(sys.argv[1:]): if a == "--user": user_id = sys.argv[i + 2] if a == "--base": global BASE BASE = sys.argv[i + 2] print(f"== 票据05 测试数据铺设 @ {BASE} (操作人 {user_id}) ==") # ---- 0. 调试 token（verify profile 后门，ADR-0014）---- token = http("GET", "/api/auth/debug/token", {"userId": user_id}) print("[ok] 获取调试 token") # ---- 1. 行政区划 ID 定位（sys_region 数据库主键，供公海池区域关联）---- provinces = http("GET", "/api/rule/region/level", {"level": 1}, token=token) prov_by_code = {p["code"]: p for p in provinces} region_ids = {} # poolKey -> {"province": id, "city": id|None} first_city_by_prov = {} # 省 code -> 首个下级市 code（直辖市等无明确市的线索回填用） for key, spec in (("A", POOL_A), ("B", POOL_B)): prov = prov_by_code.get(spec["provinceCode"]) if not prov: raise RuntimeError(f"省份 {spec['provinceCode']} 未入库（票据01 区划种子缺失）") city_id = None city_code = spec["cityCode"] children = http("GET", "/api/rule/region/list", {"parentId": prov["id"]}, token=token) if children: first_city_by_prov[spec["provinceCode"]] = children[0]["code"] if city_code is None: # 直辖市：取第一个下级作为市 if children: city_code = children[0]["code"] city_id = children[0]["id"] spec["cityCode"] = city_code else: hit = next((c for c in children if c["code"] == city_code), None) if not hit: raise RuntimeError(f"市 {city_code} 不在省 {spec['provinceCode']} 下") city_id = hit["id"] region_ids[key] = {"province": prov["id"], "city": city_id} print(f"[ok] 池{key} 区域定位：省 {prov['name']}({prov['id']})" + (f" 市 {city_code}({city_id})" if city_id else " 仅省级")) # ---- 2. 部门：保证 ≥2 个（一部门一池约束）---- def flatten(nodes, out): for n in nodes or []: out.append(n) flatten(n.get("children"), out) return out depts = flatten(http("GET", "/api/system/depts/tree", token=token), []) seed_dept = next((d for d in depts if d.get("deptName") == DEPT_NAME), None) if seed_dept is None: http("POST", "/api/system/depts/save", form={"deptName": DEPT_NAME, "sort": 90}, token=token) depts = flatten(http("GET", "/api/system/depts/tree", token=token), []) seed_dept = next(d for d in depts if d.get("deptName") == DEPT_NAME) print(f"[ok] 创建部门 {DEPT_NAME}({seed_dept['id']})") else: print(f"[skip] 部门 {DEPT_NAME} 已存在({seed_dept['id']})") if len(depts) < 2: raise RuntimeError("部门数不足 2，无法建两个池") dept_a = seed_dept["id"] dept_b = next(d["id"] for d in depts if d["id"] != dept_a) print(f"[ok] 池A部门={dept_a}({seed_dept.get('deptName')}) 池B部门={dept_b}") # ---- 3. 公海池 ×2（按名称幂等）---- pool_page = http("POST", "/api/rule/pool/page", form={"current": 1, "size": 100}, token=token) pools_by_name = {p["poolName"]: p for p in pool_page.get("content", [])} pool_ids = {} for key, spec, dept_id in (("A", POOL_A, dept_a), ("B", POOL_B, dept_b)): exist = pools_by_name.get(spec["poolName"]) if exist: pool_ids[key] = exist["id"] print(f"[skip] 公海池 {spec['poolName']} 已存在({exist['id']})") continue form = { "poolName": spec["poolName"], "deptId": dept_id, "claimRule": spec["claimRule"], "recycleDays": spec["recycleDays"], "expireDays": spec["expireDays"], "dailyClaimLimit": spec["dailyClaimLimit"], "holdLimit": spec["holdLimit"], "ownerUserId": user_id, "provinceRegionIds": [region_ids[key]["province"]], } if region_ids[key]["city"]: form["cityRegionIds"] = [region_ids[key]["city"]] http("POST", "/api/rule/pool/saveOrUpdate", form=form, token=token) pool_page = http("POST", "/api/rule/pool/page", form={"current": 1, "size": 100}, token=token) pools_by_name = {p["poolName"]: p for p in pool_page.get("content", [])} pool_ids[key] = pools_by_name[spec["poolName"]]["id"] print(f"[ok] 创建公海池 {spec['poolName']}({pool_ids[key]}) " f"规则={spec['claimRule']} 部门={dept_id}") # ---- 4. 线索种子（按 leadName 幂等 + 状态守卫流转）---- lead_ids = {} for spec in LEADS: page = http("POST", "/api/lead/page", form={"current": 1, "size": 20, "viewType": "MANAGE", "keyword": spec["leadName"]}, token=token) hit = next((l for l in page.get("content", []) if l.get("leadName") == spec["leadName"]), None) if hit: lead_ids[spec["leadName"]] = hit["id"] print(f"[skip] 线索 {spec['leadName']} 已存在({hit['id']} 状态={STATUS_NAMES.get(hit.get('status'))})") continue form = { "leadName": spec["leadName"], "phone": spec["phone"], "provinceCode": spec["provinceCode"], "cityCode": spec["cityCode"] or first_city_by_prov.get(spec["provinceCode"]), "channelCode": spec["channelCode"], "brandCode": spec["brandCode"], "productCode": spec["productCode"], "sceneCode": spec["sceneCode"], "consultContent": spec["consultContent"], "poolId": pool_ids[spec["pool"]], } claim_on_create = spec["target"] in ("claim", "follow", "convert") lid = http("POST", "/api/lead/create", params={"claimOnCreate": str(claim_on_create).lower()}, form=form, token=token) lead_ids[spec["leadName"]] = lid print(f"[ok] 创建线索 {spec['leadName']}({lid}) claimOnCreate={claim_on_create}") # ---- 5. 流转（状态守卫：已到位则跳过）---- def status_of(lid): return http("GET", "/api/lead/detail", {"id": lid}, token=token).get("status") for spec in LEADS: lid = lead_ids[spec["leadName"]] target = spec["target"] if target in ("follow", "convert"): if status_of(lid) == 3: # 已领取 → 提交反馈 → 跟进中 http("POST", "/api/lead/feedback-submit", form={ "leadId": lid, "feedbackStatus": 1, "content": "种子铺设自动反馈：客户意向明确", "productCode": spec["productCode"], }, token=token) print(f"[ok] {spec['leadName']} 提交反馈 → 跟进中") if target == "convert": if status_of(lid) in (3, 4): try: http("POST", "/api/lead/convert", params={"id": lid}, form={ "opportunityName": spec["leadName"] + "商机", "industryCode": "it", "intendedCustomer": spec["leadName"], "regionCode": spec["cityCode"] or spec["provinceCode"], }, token=token) print(f"[ok] {spec['leadName']} 转商机 → 已转商机") except RuntimeError as e: FINDINGS.append(f"转商机被阻塞：{spec['leadName']} -> {e}") print(f"[!!] {spec['leadName']} 转商机失败（记录为发现项）：{e}") # ---- 6. 关注（我的关注视图有数据）---- for spec in LEADS: if spec.get("follow_me"): lid = lead_ids[spec["leadName"]] http("POST", "/api/lead/follow", params={"id": lid}, form={}, token=token) print(f"[ok] 关注线索 {spec['leadName']}（幂等）") # ---- 7. 验收：四视图 + 统计 ---- print("\n== 验收 ==") for view in ("PUBLIC_POOL", "MY_LEAD", "MY_FOLLOW", "MANAGE"): page = http("POST", "/api/lead/page", form={"current": 1, "size": 10, "viewType": view}, token=token) total = page.get("total") names = [l.get("leadName") for l in page.get("content", [])[:5]] print(f"[view] {view}: total={total} 样例={names}") stats = http("POST", "/api/lead/stats", form={"viewType": "MANAGE"}, token=token) print(f"[stats] MANAGE 状态分布: {stats}") pool_page = http("POST", "/api/rule/pool/page", form={"current": 1, "size": 10}, token=token) print(f"[pools] {[(p['poolName'], p.get('deptName'), p.get('ownerUserName')) for p in pool_page.get('content', [])]}") if FINDINGS: print("\n== 发现项（阻塞/异常，需跟进） ==") for f in FINDINGS: print(" - " + f) print("\n== 铺设完成 ==") if __name__ == "__main__": main()