# -*- coding: utf-8 -*- """ 商机 E2E seed 脚本（票 05）——API 幂等造数 + DB 级联清残留 决议依据（票 04）：全量清 26 条存量 / 账号=管理员+A(赖永利)+B(肖琴)+C(曾偲青) / 补 待领取(1)+暂缓(3) 不造已转项目(5) / opp_name 前缀 e2e- 幂等 / Python requests（PS5.1 会污染中文） 用法: 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, 28) NOW = '2026-08-28 10:00:00' defects = [] # 非预期响应收集（票面要求：任何一步非预期都记入缺陷草稿） manifest = {'opportunities': [], 'pool_rules': [], 'scheme_templates': [], 'follows': [], 'surveys': [], 'scheme_cards': [], 'saved_views': [], '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: r0 = 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_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', 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 db_add_customer(cur, opp_id, cust_id, cust_name, primary=1): """关联客户 add API 不存在（缺陷候选）→ DB 直插，方案卡 save 依赖本商机关联客户。""" try: cur.execute( 'INSERT INTO opportunity_customer ' '(id, opportunity_id, customer_id, customer_name_snapshot, customer_role, ' ' is_primary_intended, delete_key, create_time, update_time) ' 'VALUES (%s,%s,%s,%s,%s,%s,0,NOW(),NOW())', (920000000000000000 + opp_id % 100000, opp_id, cust_id, cust_name, 'customer_role_01', primary)) cur.connection.commit() _seed_customers[str(opp_id)] = (cust_id, cust_name) print(f' ✔ DB 补关联客户 opp={opp_id} ← {cust_name}（API 无 add 端点，缺陷证据）') return True except Exception as e: defect('db-customer', f'补关联客户失败 opp={opp_id}: {e}') cur.connection.rollback() 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) --- flows = [ ('a3', 'pause', dict(pauseReason='pause_reason_01', pauseExpectedRestartDate='2026-09-30', pauseRemark='客户项目延期，seed 暂缓样例'), 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] # --- 子表样例：跟进×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)}) # --- 关联客户（DB 补）+ 方案卡×1（A 的二期，主要意向客户卡带预算） --- if a2 and db_add_customer(cur, int(a2), 920000000000000100, '园科集团（seed快照）', primary=1): 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': 920000000000000100, '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') # --- 关注×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): 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 _, t3 = page_all(A, '/api/opportunity/page', {'viewType': 'MINE', 'current': 1, 'size': 50}, step='verify A-MINE') checks['A_MINE_total'] = t3 _, 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={t3} A_FOLLOWED={t4}") print(f' board summary: {json.dumps(bs, ensure_ascii=False)[:200] if bs else "FAIL"}') 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） > 票 `05-seed-script` 产出 · 20260828 · 由 `seed-opportunity.py` 生成（幂等可重跑，重跑=先清后造）。 > 机读版：`seed-manifest.json`。机读/人读不一致时以 json 为准。 ## 账号矩阵（debug token 用 userId） | 代号 | 姓名 | userId | 部门 | |---|---|---|---| | 管理员 | 罗伟健 | `{ADMIN_UID}` | 无部门（全量可见） | | A | 赖永利 | `{UID['A']}` | 特战团队（营销中心子） | | B | 肖琴 | `{UID['B']}` | 冠军团队（营销中心子） | | C | 曾偲青 | `{UID['C']}` | 职员（跨中心） | ## 商机矩阵（{len(manifest['opportunities'])} 条） | id | 名称 | 负责人 | status(1待领取/2推进/3暂缓/4关闭) | 附加样例 | |---|---|---|---|---| {rows} ## 规则前置 - 公海规则：{json.dumps(manifest['pool_rules'], ensure_ascii=False)[:300]} - 方案卡模板：{json.dumps(manifest['scheme_templates'], ensure_ascii=False)[:300]} ## 子表/偏好样例 - 跟进：{len(manifest['follows'])} 条（挂 A-推进-智慧园区二期；customerId 为占位值，A4 客户模块未建） - 勘察：{len(manifest['surveys'])} 条 - 方案卡：{len(manifest['scheme_cards'])} 张（草稿/已提交，预算 88 万） - 自定义视图：{len(manifest['saved_views'])} 条（B 名下，scopeKey=opportunity） - 关联客户：DB 直插 1 行（API 无 add 端点，见缺陷速报） - 团队成员：**API 无写端点**（仅 team/list），无法经 API 造 → 缺陷 - 附件：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_scheme_template(admin) made = seed_matrix(admin, A, B, C, cur) verify(admin, A) 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()