# -*- coding: utf-8 -*- """ 商机 E2E seed 脚本（票 11 刷新）——API 幂等造数 + DB 级联清残留 票 11 基线升级（在票 05 十二条矩阵之上）： - 关联客户走 API（票 04 customer/add + set-primary；a2 完整剧情：主→普通→切换主要） - 团队成员走 API（票 04 team/add；a2/b1 完整团队：负责人+两角色成员） - 部门专用禁领规则 1 条（冠军团队 allowFreeClaim=0，配合 D-09 禁领回归：b4 领取实测 66014） - 长期暂缓样例（a3 预期重启 2027-01-15，前端按钮态联调） - owner_dept_id 全量落值（创建=负责人主部门，服务端 ownerSnapshotResolver 解析；抛公海保留） DB 豁免仅剩清理级联删一处（关联客户 API 化后，票 05 的补行豁免取消）。 用法: python seed-opportunity.py [--skip-db-clean] 产出: 控制台步骤日志 + seed-manifest.json + seed-data-manifest.md 原则: 全程走真实 API（DB 仅一处豁免：清理级联删，API 无商机删除端点） """ import json import sys import datetime as dt import requests import pymysql if hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') BASE = 'http://localhost:8080' ADMIN_UID = '739564171091247104' # 罗伟健（管理员，无部门） UID = { 'A': '744842565802524672', # 赖永利 · 特战团队（营销中心子） 'B': '744842566742048768', # 肖琴 · 冠军团队（营销中心子，A 兄弟部门） 'C': '744842318024015872', # 曾偲青 · 职员（跨中心） } NAME = { ADMIN_UID: '罗伟健(管理员)', 'A': '赖永利(A·特战)', 'B': '肖琴(B·冠军)', 'C': '曾偲青(C·职员)', } DB = dict(host='8.129.84.155', port=3306, user='root', password='Itc@123456', database='crm', charset='utf8mb4', autocommit=False) PREFIX = 'e2e-' TODAY = dt.date(2026, 8, 30) NOW = '2026-08-30 10:00:00' DEPT_CHAMPION = '744841308677341184' # 冠军团队（B 主部门，票 11 禁领规则专用对象） defects = [] # 非预期响应收集（票面要求：任何一步非预期都记入缺陷草稿） manifest = {'opportunities': [], 'pool_rules': [], 'scheme_templates': [], 'follows': [], 'surveys': [], 'scheme_cards': [], 'saved_views': [], 'customers': [], 'team_members': [], 'workplans': [], 'observations': [], 'checks': {}, 'skipped': []} _seed_customers = {} # opp_id -> (customer_id, customer_name) DB 补的关联客户 def defect(step, msg): line = f' ⚠ [{step}] {msg}' print(line) defects.append({'step': step, 'msg': msg}) def skip(step, msg): print(f' ⊘ [{step}] {msg}') manifest['skipped'].append({'step': step, 'msg': msg}) # ---------------- HTTP ---------------- def get_token(uid): r = requests.get(f'{BASE}/api/auth/debug/token', params={'userId': uid}, timeout=30) r.raise_for_status() data = r.json().get('data') tok = data if isinstance(data, str) else (data or {}).get('token') assert tok, f'debug token 异常: {r.text[:200]}' return tok def sess(uid): s = requests.Session() s.headers['Authorization'] = f'Bearer {get_token(uid)}' return s def api(s, method, path, form=None, json_body=None, params=None, step=''): """统一请求。业务失败/非 200 记 defect 返回 None（不抛异常，由调用方决定降级）。""" url = BASE + path try: if json_body is not None: r = s.post(url, json=json_body, params=params, timeout=30) elif form is not None: r = s.post(url, data=form, params=params, timeout=30) else: r = s.request(method, url, params=params, timeout=30) except Exception as e: defect(step, f'{method} {path} 网络异常: {e}') return None if r.status_code != 200: defect(step, f'{method} {path} HTTP {r.status_code}: {r.text[:200]}') return None try: body = r.json() except ValueError: defect(step, f'{method} {path} 响应非 JSON: {r.text[:200]}') return None code = body.get('code') if code not in (200, '200', 0, '0'): defect(step, f'{method} {path} 业务失败 code={code}: {str(body.get("message"))[:200]}') return None return body.get('data') def page_all(s, path, form, step): """PageResult 帮手：返回 (content, total)。""" d = api(s, 'POST', path, form=form, step=step) if not isinstance(d, dict): return [], None content = d.get('content') or [] return content, d.get('total') def api_void(s, path, form=None, params=None, step=''): """Result 端点专用：code 通过即成功（data=null 不是失败）。 已踩坑：api() 对 data=null 返回 None，会被误判为失败。""" url = BASE + path try: r = s.post(url, data=form, params=params, timeout=30) except Exception as e: defect(step, f'POST {path} 网络异常: {e}') return False if r.status_code != 200: defect(step, f'POST {path} HTTP {r.status_code}: {r.text[:200]}') return False try: body = r.json() except ValueError: defect(step, f'POST {path} 响应非 JSON: {r.text[:200]}') return False code = body.get('code') if code not in (200, '200', 0, '0'): defect(step, f'POST {path} 业务失败 code={code}: {str(body.get("message"))[:200]}') return False return True # ---------------- DB 清理 ---------------- CHILD_TABLES = ['opportunity_follow', 'opportunity_site_survey', 'opportunity_attachment', 'opportunity_team', 'opportunity_customer', 'opportunity_stage_history', 'opportunity_status_history', 'opportunity_focus', 'opportunity_view_log', 'opportunity_oplog', 'opportunity_work_plan', 'opportunity_pending_notice'] def db_clean(cur): """票 04 决议：全量清存量商机。API 无商机删除端点（已核实 8 控制器）→ DB 级联删。 守卫：opportunity 行数 > 100 视为连错库/环境异常，中止。""" cur.execute('SELECT COUNT(*) FROM opportunity') n = cur.fetchone()[0] print(f'[1/6] DB 清残留: opportunity 现有 {n} 行（票 04 决议=全量清 26 条）') if n > 100: raise SystemExit(f' ✋ 行数 {n} 超过守卫阈值 100，疑似连错库，中止（请人工确认 8.129.84.155/crm）') if n == 0: print(' 已是空表（重跑场景），跳过清理') return cur.execute('SELECT id FROM opportunity') ids = [r[0] for r in cur.fetchall()] ph = ','.join(['%s'] * len(ids)) def opp_col(table): cur.execute(f'SHOW COLUMNS FROM {table}') cols = [r[0] for r in cur.fetchall()] for cand in ('opp_id', 'opportunity_id'): if cand in cols: return cand return None # 方案卡值表先于卡表删（按 card 外键） card_col = opp_col('opportunity_scheme_card') if card_col: cur.execute(f'SELECT id FROM opportunity_scheme_card WHERE {card_col} IN ({ph})', ids) card_ids = [r[0] for r in cur.fetchall()] if card_ids: cur.execute('SHOW COLUMNS FROM opportunity_scheme_card_value') vcols = [r[0] for r in cur.fetchall()] for cand in ('card_id', 'scheme_card_id'): if cand in vcols: ph2 = ','.join(['%s'] * len(card_ids)) cur.execute(f'DELETE FROM opportunity_scheme_card_value WHERE {cand} IN ({ph2})', card_ids) break else: cur.execute('DELETE FROM opportunity_scheme_card_value') print(f' ⚠ opportunity_scheme_card_value 无 card 外键列，整表清 {cur.rowcount} 行') cur.execute(f'DELETE FROM opportunity_scheme_card WHERE {card_col} IN ({ph})', ids) print(f' 清 opportunity_scheme_card {cur.rowcount} 行') for t in CHILD_TABLES: col = opp_col(t) if col: cur.execute(f'DELETE FROM {t} WHERE {col} IN ({ph})', ids) else: cur.execute(f'DELETE FROM {t}') print(f' ⚠ {t} 无商机外键列，整表清 {cur.rowcount} 行') print(f' 清 {t} {cur.rowcount} 行') cur.execute('DELETE FROM opportunity') print(f' 清 opportunity 主表 {cur.rowcount} 行') cur.connection.commit() # ---------------- 造数步骤 ---------------- def ensure_pool_rule(admin): print('[2/6] 公海规则前置（F14 领取校验 + 抛公海开关依赖发布中版本）') rows, _ = page_all(admin, '/api/rule/opp-pool-rule/page', {'current': 1, 'size': 50}, step='pool-rule page') published = [r for r in rows if r.get('status') == 2] if published: # 票 11：优先复用通用规则（applyScope=1）；部门专用禁领规则由 ensure_dept_claim_rule 负责， # 免得 page 按创建时间倒序拿到专用规则、manifest 记录语义跑偏 r0 = next((r for r in published if r.get('applyScope') == 1), published[0]) print(f" 复用发布中规则 id={r0['id']} {r0.get('ruleName')} V{r0.get('versionNo')}") manifest['pool_rules'].append(r0) return form = dict(ruleName=PREFIX + 'seed公海规则', applyScope=1, isDefault=1, allowManualPool=1, autoRecycleEnabled=1, recycleDays=30, allowFreeClaim=1, recycleRemindEnabled=1, remindDays=7, ruleDesc='票05 seed 造：允许手动转入/自由领取/30天回收/提前7天提醒') ok = api_void(admin, '/api/rule/opp-pool-rule/publish', form=form, step='pool-rule publish') if not ok: skip('pool-rule', '发布失败，抛公海/领取步骤可能连带失败（defect 已记）') return rows, _ = page_all(admin, '/api/rule/opp-pool-rule/page', {'current': 1, 'size': 50, 'keyword': PREFIX}, step='pool-rule recheck') pub = next((r for r in rows if r.get('status') == 2), None) if pub: print(f" ✔ 新建并发布规则 id={pub['id']} {pub.get('ruleName')}") manifest['pool_rules'].append(pub) else: defect('pool-rule', 'publish 返回成功但 page 查不到发布中版本') def ensure_dept_claim_rule(admin): """票 11：部门专用禁领规则样例（applyScope=2 + 冠军团队 + allowFreeClaim=0）。 配合 D-09 禁领回归：b4 由 B 创建抛公海，E3 不清 owner_dept_id（保留冠军团队）→ 领取时 ActionGuard 按 owner_dept_id 命中本规则 → 66014。 autoRecycleEnabled=0：只干禁领一件事，防 30 天自动回收把公海样例收走。 幂等：已有发布中部门专用规则则复用（不碰他人数据）。""" print('[2b] 部门专用禁领规则前置（D-09 b4 禁领回归依赖发布中版本）') rows, _ = page_all(admin, '/api/rule/opp-pool-rule/page', {'current': 1, 'size': 50}, step='dept-claim-rule page') pub = next((r for r in rows if r.get('status') == 2 and r.get('applyScope') == 2), None) if pub: print(f" 复用发布中部门专用规则 id={pub['id']} {pub.get('ruleName')}") manifest['pool_rules'].append(pub) return form = dict(ruleName=PREFIX + 'seed冠军团队禁领', applyScope=2, isDefault=0, allowManualPool=1, autoRecycleEnabled=0, allowFreeClaim=0, recycleRemindEnabled=0, deptIds=[DEPT_CHAMPION], ruleDesc='票11 seed 造：冠军团队专用禁领（allowFreeClaim=0），不启用自动回收') ok = api_void(admin, '/api/rule/opp-pool-rule/publish', form=form, step='dept-claim-rule publish') if not ok: skip('dept-claim-rule', '发布失败，b4 禁领回归实测可能连带失败（defect 已记）') return rows, _ = page_all(admin, '/api/rule/opp-pool-rule/page', {'current': 1, 'size': 50, 'keyword': PREFIX}, step='dept-claim-rule recheck') pub = next((r for r in rows if r.get('status') == 2 and r.get('applyScope') == 2), None) if pub: print(f" ✔ 新建并发布部门专用规则 id={pub['id']} {pub.get('ruleName')}（dept=冠军团队）") manifest['pool_rules'].append(pub) else: defect('dept-claim-rule', 'publish 返回成功但 page 查不到发布中部门专用版本') def ensure_scheme_template(admin): print('[3/6] 方案卡模板前置（F09 建卡依赖发布中版本）') rows, _ = page_all(admin, '/api/rule/opp-scheme-template/page', {'current': 1, 'size': 50}, step='scheme-template page') published = [r for r in rows if r.get('status') == 2] if published: r0 = published[0] print(f" 复用发布中模板 id={r0['id']} {r0.get('templateName') or r0.get('code')} V{r0.get('versionNo')}") manifest['scheme_templates'].append(r0) return r0['id'] # 优先复用上次 seed copy 出的草稿（templateCode 非种子前缀），避免重跑堆积草稿 draft = next((r for r in rows if r.get('status') == 1 and not str(r.get('templateCode') or '').startswith('OPP_SCHEME_TPL_')), None) if not draft: src = next((r for r in rows if r.get('code') == 'OPP_SCHEME_TPL_01'), None) \ or (rows[0] if rows else None) if not src: skip('scheme-template', '库里无任何方案卡模板可 copy，方案卡样例跳过') return None new_id = api(admin, 'POST', '/api/rule/opp-scheme-template/copy', params={'id': src['id']}, step='scheme-template copy') if new_id is None: skip('scheme-template', 'copy 失败，方案卡样例跳过') return None draft = {'id': new_id, 'copiedFrom': src['id']} detail = api(admin, 'GET', '/api/rule/opp-scheme-template/detail', params={'id': draft['id']}, step='scheme-template detail') if not isinstance(detail, dict): skip('scheme-template', '草稿 detail 失败，方案卡样例跳过') return None # publish=保存并发布：校验的是请求 DTO 里的 fields 列表（库里有字段不够）→ # 用 Spring MVC 索引绑定把 fields 逐项回传（fieldKey/isVisible/isRequired，入参按列表顺序重算） form = {k: v for k, v in detail.items() if k not in ('fields', 'deptNames', 'createBy', 'createTime', 'updateTime') and not isinstance(v, (list, dict))} if isinstance(detail.get('deptIds'), list): form['deptIds'] = detail['deptIds'] for i, f0 in enumerate(detail.get('fields') or []): form[f'fields[{i}].fieldKey'] = f0.get('fieldKey') form[f'fields[{i}].isVisible'] = 0 if f0.get('isVisible') == 0 else 1 form[f'fields[{i}].isRequired'] = 0 if f0.get('isRequired') == 0 else 1 if not any(k.startswith('fields[') for k in form): skip('scheme-template', 'detail 未回传 fields 列表，publish 会被 64022 拒，方案卡样例跳过') return None ok = api_void(admin, '/api/rule/opp-scheme-template/publish', form=form, step='scheme-template publish') if not ok: skip('scheme-template', 'publish 失败，方案卡样例跳过') return None print(f" ✔ 模板发布 id={draft['id']}（fields {len(detail.get('fields') or [])} 项回传）") manifest['scheme_templates'].append({'id': str(draft['id']), 'copiedFrom': draft.get('copiedFrom')}) return draft['id'] def create_opp(s, who, name, **extra): form = dict(opportunityName=name, oppSource='opp_source_02', industryCode='gov', localityType='locality_type_01', bidForm='bid_form_01', # 票 09：D-14 生效后 create 六必填（province/city 未传会被 66001 拒） provinceCode='510000', cityCode='510100', partyAClear=1, partyA='E2E甲方·' + name.replace(PREFIX, ''), remark='票05 seed 造数样例（前端联调数据）') form.update(extra) d = api(s, 'POST', '/api/opportunity', form=form, step=f'create {name}') if d is None: return None, None opp_id = str(d) det = api(s, 'GET', '/api/opportunity/detail', params={'id': d}, step=f'detail {name}') st = det.get('oppStatus') if isinstance(det, dict) else '?' print(f' ✔ {name} id={opp_id} 初始 oppStatus={st}（U01 观察）') manifest['observations'].append(f'{name}: 新建后 oppStatus={st}') row = {'id': opp_id, 'name': name, 'owner': who, 'ownerDept': None, 'status': st, 'samples': [], 'detail': det if isinstance(det, dict) else None} if isinstance(det, dict): row['ownerDept'] = det.get('ownerDeptId') manifest['opportunities'].append(row) return opp_id, det def switch_stage(s, opp_id, seq_no, tag): prog = api(s, 'GET', '/api/opportunity/stage/progress', params={'oppId': opp_id}, step=f'progress {tag}') if not isinstance(prog, dict): skip('stage-switch', f'{tag} progress 失败，停在默认节点') return target = next((n for n in (prog.get('nodes') or []) if n.get('seqNo') == seq_no and not n.get('isFixed')), None) if not target: defect('stage-switch', f'{tag} 找不到 seqNo={seq_no} 的普通节点: {prog.get("nodes")}') return ok = api_void(s, '/api/opportunity/stage/switch', form={'oppId': opp_id, 'toStageId': target['nodeId'], 'remark': f'seed 切阶段到 seq{seq_no}'}, step=f'switch {tag}') if ok: print(f' ✔ {tag} 阶段切到 seq{seq_no}（nodeId={target["nodeId"]}）') def api_add_customer(s, opp_id, cust_id, cust_name, role='customer_role_01', primary=1, tag=''): """票 11：关联客户 API 化（票 04 customer/add，替掉票 05 的 DB 直插豁免）。 幂等：先 customer/list 查重（customer_id 已关联则直接返回已有行 id，--skip-db-clean 重跑不撞 66011）。 add 直设主要时服务端自动降原主要 + 刷主表 primary_customer_id 冗余列（D-17 判定口径）。""" exist = api(s, 'GET', '/api/opportunity/customer/list', params={'oppId': opp_id}, step=f'customer-list {tag}') if isinstance(exist, list): hit = next((r for r in exist if str(r.get('customerId')) == str(cust_id)), None) if hit: print(f" ⊘ 关联客户已在册 opp={opp_id} ← {cust_name}（跳过 add）") _seed_customers[str(opp_id)] = (cust_id, cust_name) manifest['customers'].append({'oppId': str(opp_id), 'customerId': str(cust_id), 'name': cust_name, 'primary': primary}) return hit.get('id') d = api(s, 'POST', '/api/opportunity/customer/add', form={'oppId': opp_id, 'customerId': cust_id, 'customerNameSnapshot': cust_name, 'customerRole': role, 'isPrimaryIntended': primary}, step=f'customer-add {tag}') if d is None: return None _seed_customers[str(opp_id)] = (cust_id, cust_name) manifest['customers'].append({'oppId': str(opp_id), 'customerId': str(cust_id), 'name': cust_name, 'primary': primary}) print(f' ✔ 关联客户 opp={opp_id} ← {cust_name}（行 id={d}，primary={primary}）') return d def api_add_team(s, opp_id, user_ids, project_role, duty, permission, tag): """票 11：团队成员 API 化（票 04 team/add，DTO 批量成员统一角色/职责）。 幂等：先 team/list 查重，目标成员已在册则跳过；快照（姓名）服务端取。""" exist = api(s, 'GET', '/api/opportunity/team/list', params={'oppId': opp_id}, step=f'team-list {tag}') have = {str(r.get('userId')) for r in exist} if isinstance(exist, list) else set() todo = [u for u in user_ids if str(u) not in have] if not todo: print(f' ⊘ 团队成员已在册 opp={opp_id} role={project_role}（跳过 add）') return d = api(s, 'POST', '/api/opportunity/team/add', form={'oppId': opp_id, 'userIds': todo, 'projectRole': project_role, 'duty': duty, 'permission': permission}, step=f'team-add {tag}') if d is not None: print(f' ✔ 团队成员 opp={opp_id} +{len(todo)}人 role={project_role} perm={permission}') manifest['team_members'].append({'oppId': str(opp_id), 'role': project_role, 'duty': duty, 'userIds': [str(u) for u in todo]}) def expect_fail(s, path, form, want_code, step): """票 11：负路径实测——业务失败且 code 命中期望 = 通过（记 checks）； 意外成功或别的错误码都算 defect（D-09 禁领回归主助手）。""" url = BASE + path try: r = s.post(url, data=form, timeout=30) except Exception as e: defect(step, f'POST {path} 网络异常: {e}') return False try: body = r.json() except ValueError: defect(step, f'POST {path} 响应非 JSON: {r.text[:200]}') return False code = str(body.get('code')) if r.status_code == 200 and code == str(want_code): msg = str(body.get('message')) print(f' ✔ 负路径实测 {step}: code={code}（{msg[:60]}）') manifest['checks'][step] = {'code': code, 'message': msg[:120], 'pass': True} return True defect(step, f'期望 code={want_code} 实得 HTTP {r.status_code} code={code}: ' f'{str(body.get("message"))[:120]}') return False def seed_matrix(admin, A, B, C, cur): print('[4/6] 造商机矩阵（12 条：待领取2 / 推进中6 / 暂缓2 / 已关闭2，跨 3 部门归属）') made = {} # --- A（特战团队）5 条 --- a1, _ = create_opp(A, 'A', PREFIX + 'A-推进-智慧园区一期') made['a1'] = a1 a2, _ = create_opp(A, 'A', PREFIX + 'A-推进-智慧园区二期', customerName='园科集团（快照）') made['a2'] = a2 if a2: switch_stage(A, a2, 2, 'A-推进-智慧园区二期') a3, _ = create_opp(A, 'A', PREFIX + 'A-暂缓-数据中心改造') made['a3'] = a3 a4, _ = create_opp(A, 'A', PREFIX + 'A-公海-展厅多媒体项目') made['a4'] = a4 a5, _ = create_opp(A, 'A', PREFIX + 'A-关闭-老机房UPS替换') made['a5'] = a5 # --- B（冠军团队）5 条 --- b1, _ = create_opp(B, 'B', PREFIX + 'B-推进-会议系统扩容') made['b1'] = b1 b2, _ = create_opp(B, 'B', PREFIX + 'B-推进-报告厅音视频') made['b2'] = b2 if b2: switch_stage(B, b2, 3, 'B-推进-报告厅音视频') b3, _ = create_opp(B, 'B', PREFIX + 'B-暂缓-剧场改造') made['b3'] = b3 b4, _ = create_opp(B, 'B', PREFIX + 'B-公海-法院信息化') made['b4'] = b4 b5, _ = create_opp(B, 'B', PREFIX + 'B-关闭-厂房广播改造') made['b5'] = b5 # --- C（职员，跨中心）2 条 --- c1, _ = create_opp(C, 'C', PREFIX + 'C-推进-高校实验室建设') made['c1'] = c1 c2, _ = create_opp(C, 'C', PREFIX + 'C-推进-医院护理呼叫系统') made['c2'] = c2 if c2: switch_stage(C, c2, 2, 'C-推进-医院护理呼叫系统') def row_of(key): return next((r for r in manifest['opportunities'] if r['id'] == str(made[key])), None) # --- 状态流转造 暂缓(3)/公海(1)/关闭(4) --- # 票 11：a3 改长期暂缓样例（预期重启 2027-01-15 远期，前端联调按钮态）；b3 保留近期对照 flows = [ ('a3', 'pause', dict(pauseReason='pause_reason_01', pauseExpectedRestartDate='2027-01-15', pauseRemark='客户预算未批复，长期暂缓样例（票 11：预期重启时间远期）'), A), ('a4', 'release-pool', dict(poolReason='pool_reason_01'), A), ('a5', 'close', dict(closeReason='close_reason_02', closeRemark='客户放弃，seed 关闭样例'), A), ('b3', 'pause', dict(pauseReason='pause_reason_02', pauseExpectedRestartDate='2026-10-15', pauseRemark='预算冻结，seed 暂缓样例（近期重启对照）'), B), ('b4', 'release-pool', dict(poolReason='pool_reason_02'), B), ('b5', 'close', dict(closeReason='close_reason_03', closeRemark='竞争失败，seed 关闭样例'), B), ] for key, action, form, s in flows: if not made.get(key): skip(action, f'{key} 未创建成功，流转跳过') continue ok = api_void(s, f'/api/opportunity/{action}', form={'id': made[key], **form}, step=f'{action} {key}') if ok: print(f' ✔ {key} → {action}') row = row_of(key) if row: row['status'] = {'pause': 3, 'release-pool': 1, 'close': 4}[action] # --- D-09 禁领回归实测（票 11 新增）：b4 命中冠军团队禁领 → 66014；a4 往返对照放行 --- if b4: # b4：B 创建抛公海，E3 不清 owner_dept_id（保留冠军团队）→ A 领取按 owner_dept_id # 命中部门专用规则 allowFreeClaim=0 → 66014（预检拦截，状态不变） expect_fail(A, '/api/opportunity/claim', {'id': b4}, 66014, 'b4_claim_forbidden') if a4: # a4 对照：A 自己抛的公海，owner_dept_id 保留特战团队 → 走通用规则 allowFreeClaim=1 放行； # 领取后立即抛回恢复公海样例（owner/dept 不变，状态复原，多留一条 claim+release 流转痕迹） if api_void(A, '/api/opportunity/claim', params={'id': a4}, step='a4 claim roundtrip'): print(' ✔ a4 领取放行（对照：特战团队走通用默认规则）') manifest['checks']['a4_claim_allowed'] = {'pass': True} api_void(A, '/api/opportunity/release-pool', params={'id': a4, 'poolReason': 'pool_reason_01'}, step='a4 release-pool restore') row = row_of('a4') if row: row['status'] = 1 row['samples'].append('禁领对照：领取放行后已抛回公海') # --- 子表样例：跟进×2（customerId 必填但 A4 客户模块未建 → 占位实测） --- if a2: for i, (content, way) in enumerate([ ('首次电话沟通，确认园区一期扩容意向，约下周现场拜访', 'follow_way_01'), ('上门拜访完成，客户明确预算区间，推进方案卡编制', 'follow_way_02')]): d = api(A, 'POST', '/api/opportunity/follow/add', form={'oppId': a2, 'followContent': content, 'followTime': NOW, 'followWay': way, 'customerId': 1, 'resultTag': 'result_tag_02'}, step=f'follow-add #{i+1}') if d is None: # customerId=1 被拒 → 试 0 d = api(A, 'POST', '/api/opportunity/follow/add', form={'oppId': a2, 'followContent': content, 'followTime': NOW, 'followWay': way, 'customerId': 0, 'resultTag': 'result_tag_02'}, step=f'follow-add #{i+1} retry customerId=0') if d is not None: print(f' ✔ 跟进记录 #{i+1} id={d}') manifest['follows'].append({'id': str(d), 'oppId': str(a2)}) else: skip('follow', f'跟进 #{i+1} 造数失败（customerId 必填 vs A4 未建，见 defect）') # --- 勘察×1（A 的二期） --- if a2: d = api(A, 'POST', '/api/opportunity/site-survey/add', form={'oppId': a2, 'surveyDate': '2026-08-27', 'engineerUserId': UID['A'], 'applyWay': 'apply_way_01', 'surveySeq': 'survey_seq_01', 'applyNo': 'KC-20260827-001', 'surveyDesc': '现场勘察：机房位置/供电/承重确认'}, step='site-survey add') if d is not None: print(f' ✔ 勘察记录 id={d}') manifest['surveys'].append({'id': str(d), 'oppId': str(a2)}) # --- 关联客户（票 04 API 化，替掉票 05 DB 直插）+ 方案卡×1（A 的二期） --- # a2 完整剧情：园科集团为主 → 建工设计院普通 → set-primary 切主要（原主要自动降普通，业务规则 6） # customerId 为占位 id（A4 客户模块未建，add 对 customerId 无存在性校验，仅查重复关联 66011） CUST_A1, CUST_A2 = 920000000000000100, 920000000000000200 a2_primary = None if a2: if api_add_customer(A, int(a2), CUST_A1, '园科集团（seed快照）', primary=1, tag='a2-园科'): api_add_customer(A, int(a2), CUST_A2, '省建工设计院（seed快照）', role='customer_role_04', primary=0, tag='a2-建工') if api_void(A, '/api/opportunity/customer/set-primary', params={'oppId': a2, 'customerId': CUST_A2}, step='customer set-primary a2'): print(f' ✔ a2 主要意向客户切换 → 省建工设计院（customer_id={CUST_A2}）') manifest['checks']['a2_set_primary'] = {'primary': str(CUST_A2), 'pass': True} a2_primary = CUST_A2 else: defect('customer-add a2', 'a2 主关联客户失败，方案卡 customerId 回退占位值') a2_primary = CUST_A1 # 票面口径：每条推进中商机 2-3 个关联客户、恰 1 个主要意向（A4 未建，customerId 用占位 id） if a1: api_add_customer(A, int(a1), 920000000000000500, '云谷科技园（seed快照）', primary=1, tag='a1') api_add_customer(A, int(a1), 920000000000000600, '市规划设计院（seed快照）', role='customer_role_04', primary=0, tag='a1') if b1: api_add_customer(B, int(b1), 920000000000000300, '峰会会展中心（seed快照）', primary=1, tag='b1') api_add_customer(B, int(b1), 920000000000000310, '华堂展陈工程（seed快照）', role='customer_role_05', primary=0, tag='b1') if b2: api_add_customer(B, int(b2), 920000000000000700, '滨江文体中心（seed快照）', primary=1, tag='b2') api_add_customer(B, int(b2), 920000000000000800, '南方声学工程（seed快照）', role='customer_role_05', primary=0, tag='b2') if c1: api_add_customer(C, int(c1), 920000000000000400, '省第一人民医院（seed快照）', primary=1, tag='c1') api_add_customer(C, int(c1), 920000000000000410, '泽康医疗设备（seed快照）', role='customer_role_02', primary=0, tag='c1') if c2: api_add_customer(C, int(c2), 920000000000000900, '仁济医院分院（seed快照）', primary=1, tag='c2') api_add_customer(C, int(c2), 920000000000000950, '安捷机电安装（seed快照）', role='customer_role_05', primary=0, tag='c2') if a2 and a2_primary: tpl_rows, _ = page_all(admin, '/api/rule/opp-scheme-template/page', {'current': 1, 'size': 50}, step='scheme-template page for card') pub = next((r for r in tpl_rows if r.get('status') == 2), None) if not pub: skip('scheme-card', '无发布中模板，方案卡样例跳过') else: sel = api(A, 'GET', '/api/opportunity/scheme-card/selectable-templates', params={'oppId': a2}, step='selectable-templates') tpl_id = None if isinstance(sel, list) and sel: tpl_id = sel[0].get('templateId') or sel[0].get('id') if tpl_id is None: # selectable 为空也允许直接拿发布中模板版本行 id 兜底实测 tpl_id = pub.get('id') defect('selectable-templates', f'商机 {a2} 可选模板列表空（适用范围未命中？），用模板版本行 id={tpl_id} 兜底') # valuesJson 的 fieldKey 只能取「绑定模板版本的显示字段」（66007 校验）， # field-defs 全库字段集比模板显示字段宽，不能用全库 dtl = api(admin, 'GET', '/api/rule/opp-scheme-template/detail', params={'id': tpl_id}, step='scheme-template detail for card') vis = [f for f in ((dtl or {}).get('fields') or []) if f.get('isVisible') == 1] # 结构化字段（招标形式/介入阶段/方案预算）有主表列，禁止进 valuesJson（66007） STRUCTURED = {'scheme_budget', 'bidding_form', 'enter_stage', 'schemeBudget', 'biddingForm', 'enterStage'} dyn = [f for f in vis if f.get('fieldKey') not in STRUCTURED and not any(s in str(f.get('fieldKey')) for s in ('budget', 'bidd', 'stage_'))] values = [] # 必填字段全填（submit 校验），非必填取 2 个点缀 req = [f for f in dyn if f.get('isRequired') == 1] opt = [f for f in dyn if f.get('isRequired') != 1][:2] for f in req + opt: k = str(f.get('fieldKey')) v = 'seed样例-' + str(f.get('fieldName') or k) if 'date' in k or '时间' in str(f.get('fieldName') or ''): v = '2026-09-15' elif any(s in k for s in ('amount', 'budget', 'num', 'count')): v = '100' values.append({'fieldKey': f['fieldKey'], 'fieldValue': v}) d = api(A, 'POST', '/api/opportunity/scheme-card/save', form={'oppId': a2, 'customerId': a2_primary, 'templateId': tpl_id, 'schemeBudget': '880000.00', 'biddingForm': 'bid_form_01', 'enterStage': 'opp_stage_04', 'valuesJson': json.dumps(values, ensure_ascii=False)}, step='scheme-card save') if d is not None: print(f' ✔ 方案卡草稿 id={d}（预算 88 万）') manifest['scheme_cards'].append({'id': str(d), 'oppId': str(a2), 'templateId': str(tpl_id)}) if api_void(A, '/api/opportunity/scheme-card/submit', params={'id': d}, step='scheme-card submit'): print(f' ✔ 方案卡已提交 cardStatus=2') # --- 团队成员（票 04 API 化）：a2/b1 完整团队（负责人本人 + 两角色成员） --- # 角色用 crm-dict「商机项目角色」字典 code（project_role_01~06）； # permission 档：1只读 2可写跟进 3可推进节点；快照（姓名）服务端取 if a2: api_add_team(A, a2, [UID['B']], 'project_role_05', '方案设计与预算编制', 3, 'a2+B') api_add_team(A, a2, [UID['C']], 'project_role_02', '现场勘察配合', 2, 'a2+C') if b1: api_add_team(B, b1, [UID['A']], 'project_role_04', '报价与投标支持', 2, 'b1+A') api_add_team(B, b1, [UID['C']], 'project_role_06', '跨部门协同', 1, 'b1+C') # --- 工作计划（票 12 API 化）：a2 三形态样例（逾期/临期/已完成），逾期实时计算语义 --- # 幂等：先 list 按 planContent 精确匹配复用（--skip-db-clean 重跑不重复 add）； # 复用时校准 deadline（相对当前时间刷新，逾期/临期语义保鲜）与 planStatus（登记完成走 update 回填 finishTime） if a2: now = dt.datetime.now() wp_defs = [ ('e2e-wp-逾期样例（整理现场勘察纪要并回传）', now - dt.timedelta(days=2), 0), ('e2e-wp-临期样例（本周内完成方案卡评审）', now + dt.timedelta(days=3), 0), ('e2e-wp-已完成样例（提交报价初稿）', now + dt.timedelta(days=1), 1), ] exist_wp = api(A, 'GET', '/api/opportunity/workplan/list', params={'oppId': a2}, step='workplan list') or [] for wp_content, wp_deadline, wp_status in wp_defs: hit = next((r for r in exist_wp if r.get('planContent') == wp_content), None) dl = wp_deadline.strftime('%Y-%m-%d %H:%M:%S') if hit: upd = {'id': hit['id'], 'deadline': dl} if hit.get('planStatus') != wp_status: upd['planStatus'] = wp_status if api_void(A, '/api/opportunity/workplan/update', form=upd, step=f'workplan 校准 {wp_content[:14]}'): print(f' ⊘ 工作计划已在册 opp={a2} ← {wp_content[:18]}（deadline/状态校准）') else: d = api(A, 'POST', '/api/opportunity/workplan/add', form={'oppId': a2, 'planContent': wp_content, 'deadline': dl}, step=f'workplan add {wp_content[:14]}') if d is not None: if wp_status == 1: api_void(A, '/api/opportunity/workplan/update', form={'id': d, 'planStatus': 1}, step='workplan 登记完成') hit = {'id': d} print(f' ✔ 工作计划 opp={a2} ← {wp_content[:18]}（行 id={d}，status={wp_status}）') if hit: manifest['workplans'].append({'oppId': str(a2), 'id': str(hit['id']), 'content': wp_content, 'status': wp_status}) a2_row = row_of('a2') if a2_row: a2_row['samples'].append('工作计划×3（逾期/临期/已完成，票 12 Tab 数据源）') # --- 关注×1：A 关注 B 的一条 --- if b1: if api_void(A, '/api/opportunity/focus', params={'oppId': b1}, step='focus'): print(f' ✔ A 已关注 {b1}') row_of('b1')['samples'].append('A(赖永利)已关注') # --- 自定义视图×1（B 保存，F15 样例） --- lv = api(B, 'GET', '/api/preference/view/list', params={'scopeKey': 'opportunity'}, step='view list') existing = None if isinstance(lv, list): existing = next((v for v in lv if str(v.get('name', '')).startswith(PREFIX)), None) body = {'name': PREFIX + '测试视图-我的推进', 'isDefault': True, 'seqNo': 99, 'sortField': 'createTime', 'sortDirection': 'desc', 'conditions': [{'field': 'opportunityName', 'operator': 'like', 'value': PREFIX}, {'field': 'oppStatus', 'operator': 'eq', 'value': '2'}]} if existing: body['viewId'] = existing.get('viewId') d = api(B, 'POST', '/api/preference/view/save', params={'scopeKey': 'opportunity'}, json_body=body, step='saved-view save') if d is not None: print(f' ✔ 自定义视图 viewId={d}（scopeKey=opportunity）') manifest['saved_views'].append({'viewId': str(d), 'owner': 'B', 'scopeKey': 'opportunity'}) return made def verify(admin, A, B, C): print('[5/6] 校验盘点（write checks → manifest）') checks = manifest['checks'] _, t = page_all(admin, '/api/opportunity/page', {'viewType': 'MANAGE', 'current': 1, 'size': 50}, step='verify MANAGE') checks['MANAGE_total'] = t _, t2 = page_all(admin, '/api/opportunity/page', {'viewType': 'PUBLIC_POOL', 'current': 1, 'size': 50}, step='verify POOL') checks['PUBLIC_POOL_total'] = t2 # 票 11：三账号 MINE 计数齐备（前端各账号工作台联调用） for tag, s in (('A', A), ('B', B), ('C', C)): _, tm = page_all(s, '/api/opportunity/page', {'viewType': 'MINE', 'current': 1, 'size': 50}, step=f'verify {tag}-MINE') checks[f'{tag}_MINE_total'] = tm _, t4 = page_all(A, '/api/opportunity/page', {'viewType': 'FOLLOWED', 'current': 1, 'size': 50}, step='verify A-FOLLOWED') checks['A_FOLLOWED_total'] = t4 bs = api(admin, 'POST', '/api/opportunity/board/stage-summary', form={'viewType': 'MANAGE'}, step='verify board-summary') checks['board_stage_summary'] = bs print(f" MANAGE={t} PUBLIC_POOL={t2} A_MINE={checks['A_MINE_total']} " f"B_MINE={checks['B_MINE_total']} C_MINE={checks['C_MINE_total']} A_FOLLOWED={t4}") print(f' board summary: {json.dumps(bs, ensure_ascii=False)[:200] if bs else "FAIL"}') def spot_check_detail(s, opp_id): """票 11 验收抽查：A 登录看推进中商机详情（owner_dept_id 回显 + 团队 + 关联客户齐全）。""" if not opp_id: return print('[5b] 前端联调视角抽查（A 登录 detail）') det = api(s, 'GET', '/api/opportunity/detail', params={'id': opp_id}, step='spot detail') if not isinstance(det, dict): defect('spot detail', '详情拉取失败') return team = api(s, 'GET', '/api/opportunity/team/list', params={'oppId': opp_id}, step='spot team-list') cust = api(s, 'GET', '/api/opportunity/customer/list', params={'oppId': opp_id}, step='spot customer-list') checks = manifest['checks'] checks['spot_a2'] = { 'oppId': str(opp_id), 'ownerDeptId': det.get('ownerDeptId'), 'primaryCustomerId': det.get('primaryCustomerId'), 'team_members': len(team) if isinstance(team, list) else None, 'linked_customers': len(cust) if isinstance(cust, list) else None, 'pass': bool(det.get('ownerDeptId')) and isinstance(team, list) and isinstance(cust, list), } print(f" ✔ a2 详情：ownerDeptId={det.get('ownerDeptId')} " f"团队 {checks['spot_a2']['team_members']} 人 " f"关联客户 {checks['spot_a2']['linked_customers']} 个 " f"primaryCustomer={checks['spot_a2']['primaryCustomerId']}") def write_outputs(): print('[6/6] 落盘 manifest') with open('.scratch/opportunity-e2e/seed-manifest.json', 'w', encoding='utf-8') as f: json.dump(manifest, f, ensure_ascii=False, indent=2) rows = '\n'.join( f"| {r['id']} | {r['name']} | {NAME.get(r['owner'], r['owner'])} | " f"{r.get('status')} | {'；'.join(r['samples']) or '—'} |" for r in manifest['opportunities']) defects_md = ('\n'.join(f"- **[{d['step']}]** {d['msg']}" for d in defects)) or '（无）' skipped_md = ('\n'.join(f"- **[{s['step']}]** {s['msg']}" for s in manifest['skipped'])) or '（无）' md = f"""# 商机 E2E · 测试数据清单（seed-data-manifest） > 票 `11-testdata-refresh` 刷新 · 20260830 · 由 `seed-opportunity.py` 生成（幂等可重跑，重跑=先清后造）。 > 票 `12-workplan-tab` 增补工作计划样例 · 20260831。 > 机读版：`seed-manifest.json`。机读/人读不一致时以 json 为准。 ## 账号矩阵（debug token 用 userId） | 代号 | 姓名 | userId | 部门（dept_id，crm_auth_user.dept_id 实锤） | |---|---|---|---| | 管理员 | 罗伟健 | `{ADMIN_UID}` | —（全量可见） | | A | 赖永利 | `{UID['A']}` | 特战团队 `744841308564094976`（营销中心子） | | B | 肖琴 | `{UID['B']}` | 冠军团队 `744841308677341184`（营销中心子） | | C | 曾偲青 | `{UID['C']}` | 职员 `744841292483133440`（跨中心） | ## 商机矩阵（{len(manifest['opportunities'])} 条） | id | 名称 | 负责人 | status(1待领取/2推进/3暂缓/4关闭) | 附加样例 | |---|---|---|---|---| {rows} ## 规则前置 - 公海规则（{len(manifest['pool_rules'])} 条发布中，含部门专用禁领）：{json.dumps(manifest['pool_rules'], ensure_ascii=False)[:500]} - 方案卡模板：{json.dumps(manifest['scheme_templates'], ensure_ascii=False)[:300]} ## 子表/偏好样例 - 跟进：{len(manifest['follows'])} 条（挂 A-推进-智慧园区二期；customerId 为占位值，A4 客户模块未建） - 勘察：{len(manifest['surveys'])} 条 - 方案卡：{len(manifest['scheme_cards'])} 张（草稿/已提交，预算 88 万，customerId=主要意向客户） - 自定义视图：{len(manifest['saved_views'])} 条（B 名下，scopeKey=opportunity） - 关联客户：API 造 {len(manifest['customers'])} 行（票 04 customer/add；6 条推进中商机每条 2 个恰 1 主要，a2 另含 set-primary 切换剧情） - 团队成员：API 造 {len(manifest['team_members'])} 条（票 04 team/add；a2/b1 完整团队：负责人+方案/现场/报价角色成员） - 工作计划：API 造 {len(manifest['workplans'])} 条（票 12 workplan/add+update；挂 a2：逾期/临期/已完成三形态，逾期=未完成且 now>deadline 前端实时判） - 禁领回归：b4 领取实测 66014（冠军团队专用规则 allowFreeClaim=0）；a4 领取放行对照（走通用规则） - 长期暂缓样例：a3（预期重启 2027-01-15）；b3 近期重启对照（2026-10-15） - 附件：MinIO SK 未配齐，本次未造（票 07 F08 补） ## 校验结果 ```json {json.dumps(manifest['checks'], ensure_ascii=False, indent=2)} ``` ## U01 观察（新建初始状态） {chr(10).join('- ' + o for o in manifest['observations']) or '（无）'} ## 缺陷速报（seed 过程非预期响应，喂票 09 报告） {defects_md} ## 跳过项 {skipped_md} ## 重跑方式 ``` python .scratch/opportunity-e2e/seed-opportunity.py # 先清后造（幂等） python .scratch/opportunity-e2e/seed-opportunity.py --skip-db-clean # 只造不清 ``` """ with open('.scratch/opportunity-e2e/seed-data-manifest.md', 'w', encoding='utf-8') as f: f.write(md) print(' ✔ seed-manifest.json + seed-data-manifest.md') def main(): skip_db_clean = '--skip-db-clean' in sys.argv conn = pymysql.connect(**DB) cur = conn.cursor() try: if not skip_db_clean: db_clean(cur) else: print('[1/6] 跳过 DB 清理（--skip-db-clean）') admin, A, B, C = sess(ADMIN_UID), sess(UID['A']), sess(UID['B']), sess(UID['C']) ensure_pool_rule(admin) ensure_dept_claim_rule(admin) ensure_scheme_template(admin) made = seed_matrix(admin, A, B, C, cur) verify(admin, A, B, C) spot_check_detail(A, made.get('a2')) write_outputs() finally: cur.close() conn.close() print(f"\n===== seed 完成：商机 {len(manifest['opportunities'])} 条 · 缺陷速报 {len(defects)} 条 · 跳过 {len(manifest['skipped'])} 项 =====") if defects: print('缺陷速报（详情见 seed-data-manifest.md）：') for d in defects: print(f" ⚠ [{d['step']}] {d['msg'][:120]}") if __name__ == '__main__': main()