# -*- coding: utf-8 -*- """票 02 · 真值 specimens 补全采集：9 未覆盖端点 + 7 退化端点（去重后 10 个采集流程）。 夹具前缀 e2c-r3d-（DB 正门清扫）；偏好保存后原值恢复；transfer 夹具只用自有客户。 产物：specimens-doc-truth.json（'METHOD /path' → {request, response}，与套件 specimens 同构）。 """ from __future__ import annotations import io import json import sys import time from pathlib import Path import pymysql import requests sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8') BASE = 'http://localhost:8080' ADMIN = '739564171091247104' # 罗伟健 BUDDY = '744842318024015872' # 曾偲青 PFX = 'e2c-r3d' DEPT_SALES = 744700334353416192 # 华南销售部（seed，F-13 同款：接收总监推导部门） BUDDY_HOME_DEPT = 744841292483133440 # BUDDY 原部门（职员部，手术还原用） OUT = Path('.scratch/customer-integration-ready') XLSX_DIR = OUT / 'xlsx' XLSX_DIR.mkdir(parents=True, exist_ok=True) S = requests.Session() S.headers['Authorization'] = 'Bearer placeholder' specimens: dict[str, dict] = {} def capture(method, path, req, resp): specimens[f'{method} {path}'] = {'request': req, 'response': resp} def get_token(uid): r = requests.get(f'{BASE}/api/auth/debug/token', params={'userId': uid}, timeout=10) r.raise_for_status() return r.json()['data'] def api(method, path, form=None, params=None, files=None, step='', uid=ADMIN): headers = {'Authorization': f'Bearer {get_token(uid)}'} kw = {'headers': headers, 'timeout': 60} if files: kw['files'] = files kw['data'] = form or {} elif method == 'POST': kw['data'] = form or {} else: kw['params'] = params or {} if form: kw['params'].update(form) r = S.request(method, BASE + path, **kw) try: body = r.json() except Exception: body = {'_raw': r.text[:300], '_status': r.status_code} print(f" [{step or path}] http={r.status_code} code={body.get('code') if isinstance(body, dict) else '?'}") return r, body def db(): return pymysql.connect(host='8.129.84.155', port=3306, user='root', password='Itc@123456', database='crm', charset='utf8mb4', autocommit=True, connect_timeout=10, cursorclass=pymysql.cursors.DictCursor) def dbq(sql, args=None): with db() as conn, conn.cursor() as cur: cur.execute(sql, args) return cur.fetchall() def dbx(sql, args=None): with db() as conn, conn.cursor() as cur: cur.execute(sql, args) return cur.rowcount def sweep(): rows = dbq("SELECT id FROM customer WHERE customer_name LIKE %s", (PFX + '%',)) for row in rows: cid = row['id'] dbx("DELETE FROM contact_import_fail WHERE task_id IN " "(SELECT id FROM contact_import_task WHERE customer_id=%s)", (cid,)) for t in ['customer_contact_edge', 'customer_contact_graph', 'customer_contact_reveal_log', 'contact_import_task', 'customer_contact', 'customer_oplog', 'customer_transfer_detail', 'customer_transfer', 'team_member', 'customer_focus', 'customer_import_fail', 'customer_import_task']: try: dbx(f"DELETE FROM {t} WHERE customer_id=%s", (cid,)) except Exception: pass dbx("DELETE FROM customer WHERE id=%s", (cid,)) # 交割头表（无 customer_id 列）：本轮 BUDDY 发起的当日交割单 try: n = dbx("DELETE FROM customer_transfer WHERE from_user_id=%s AND create_time >= CURDATE()", (int(BUDDY),)) print(f'sweep: 清扫 BUDDY 当日交割单 {n} 张') except Exception: pass print(f'sweep: 清扫夹具客户 {len(rows)} 组') return len(rows) def mk_customer(name, **extra): form = {'customerName': name, 'customerType': 'customer_type_01', 'provinceCode': '440000', 'cityCode': '440100', 'districtCode': '440103', 'industryCode': 'other', 'customerStarLevel': 3, 'relationStarLevel': 3, 'isBizNegotiated': 0, 'isChild': 0, 'ownerUserId': ADMIN} form.update(extra) r, body = api('POST', '/api/customer/create', form=form, step=f'create {name}') d = body.get('data') or {} if d.get('needConfirm'): r, body = api('POST', '/api/customer/create', form=dict(form, confirmSimilar='true'), step='create confirm') d = body.get('data') or {} assert d.get('id'), f'夹具创建失败: {str(body)[:200]}' return {'id': str(d['id']), 'no': d.get('customerNo', ''), 'name': name} def mk_xlsx(path, sheet_name, header, rows): from openpyxl import Workbook wb = Workbook() ws = wb.active ws.title = sheet_name ws.append(header) for row in rows: ws.append(row) wb.save(path) def h_graph_and_contacts(): """H1+H2: 图谱有数据读 + 联系人详情。""" a = mk_customer(f'{PFX}-图谱-{int(time.time())%100000}') _, qa = api('POST', '/api/customer/contact/quickAdd', form={ 'customerId': a['id'], 'rows[0].name': '张顶层', 'rows[0].jobTitleCode': 'other_job', 'rows[0].phone': '13800000101', 'rows[0].source': 'referral', 'rows[0].isKeyContact': '1', 'rows[1].name': '李中层', 'rows[1].jobTitleCode': 'other_job', 'rows[1].phone': '13800000102', 'rows[1].source': 'other', 'rows[2].name': '王基层', 'rows[2].jobTitleCode': 'other_job', 'rows[2].phone': '13800000103'}, step='quickAdd 3') _, gd0 = api('GET', '/api/customer/contact/graph/detail', params={'customerId': a['id']}, step='graph pre-read') nodes = sorted(((gd0.get('data') or {}).get('nodes') or []), key=lambda x: str(x.get('contactId'))) cids = [str(n['contactId']) for n in nodes] assert len(cids) == 3, f'graph nodes 异常: {str(gd0)[:200]}' # 图谱：张顶层=10 李中层=6 王基层=3，两条手动边 _, gd = api('GET', '/api/customer/contact/graph/detail', params={'customerId': a['id']}, step='graph pre-read') ver = int((gd.get('data') or {}).get('version') or 0) form = {'customerId': a['id'], 'version': str(ver), 'nodes[0].contactId': cids[0], 'nodes[0].level': '10', 'nodes[1].contactId': cids[1], 'nodes[1].level': '6', 'nodes[2].contactId': cids[2], 'nodes[2].level': '3', 'edges[0].parentId': cids[0], 'edges[0].childId': cids[1], 'edges[1].parentId': cids[1], 'edges[1].childId': cids[2]} api('POST', '/api/customer/contact/graph/save', form=form, step='graph save 2 edges') _, body = api('GET', '/api/customer/contact/graph/detail', params={'customerId': a['id']}, step='H1 图谱全量读(有数据)') d = body.get('data') or {} assert d.get('nodes') and d.get('edges'), '图谱读仍退化' capture('GET', '/api/customer/contact/graph/detail', {'query': {'customerId': a['id']}}, body) # H2 联系人详情 _, body = api('GET', '/api/customer/contact/detail', params={'id': cids[0]}, step='H2 联系人详情') capture('GET', '/api/customer/contact/detail', {'query': {'id': cids[0]}}, body) return a, cids def h_quickadd_failed(a): """H3: quickAdd 部分失败行（同批重复手机号 → 查重键冲突行失败）。""" form = {'customerId': a['id'], 'rows[0].name': '赵好行', 'rows[0].jobTitleCode': 'other_job', 'rows[0].phone': '13800000104', 'rows[1].name': '钱坏行', 'rows[1].jobTitleCode': 'other_job', 'rows[1].phone': '13800000104'} _, body = api('POST', '/api/customer/contact/quickAdd', form=form, step='H3 quickAdd 部分失败') d = body.get('data') or {} print(f" successCount={d.get('successCount')} failedRows={len(d.get('failedRows') or [])}") capture('POST', '/api/customer/contact/quickAdd', {'form': form}, body) def h_batchedit_failed(a, cids): """H4: batchEdit 失败行（含已删联系人 id → 行级失败）。""" _, d0 = api('GET', '/api/customer/contact/detail', params={'id': cids[0]}, step='batchEdit pre-read') ver = int(((d0.get('data') or {}).get('version')) or 0) form = {'customerId': a['id'], 'rows[0].id': cids[1], 'rows[0].name': '李中层改', 'rows[1].id': cids[2], 'rows[1].phone': '13800000102'} # 改成他人手机号 → 行级查重失败 _, body = api('POST', '/api/customer/contact/batchEdit', form=form, step='H4 batchEdit 失败行') d = body.get('data') or {} print(f" code={body.get('code')} failedRows={len(d.get('failedRows') or []) if isinstance(d, dict) else '-'}") capture('POST', '/api/customer/contact/batchEdit', {'form': form}, body) def h_customer_import(): """H5: 客户导入三段式（含 FAIL/SUSPECT 行）→ upload/result/failures/page/template。""" good = ['', f'{PFX}-导入好行', 'customer_type_01', '', '440000', '440100', 'other', '3', 'r3d'] bad = ['', f'{PFX}-导入坏行', '', '', '440000', '440100', 'other', '3', ''] # 缺客户类型 → 预检 FAIL（P1-1） dup = ['', 'e2c-恒信达科技有限公司', 'customer_type_01', '', '440000', '440100', 'other', '3', ''] # 同名已有 → FAIL/SUSPECT path = XLSX_DIR / 'cust-import.xlsx' mk_xlsx(path, '客户', ['客户编号', '客户名称', '客户类型', '统一社会信用代码', '省份编码', '城市编码', '行业编码', '客户星级(1-5)', '备注'], [good, bad, dup]) with open(path, 'rb') as f: r, body = api('POST', '/api/customer/import/upload', form={'importMode': 'UPSERT', 'duplicateStrategy': 'SKIP'}, files={'file': (path.name, f, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')}, step='H5 upload 预校验') capture('POST', '/api/customer/import/upload', {'form': {'importMode': 'UPSERT', 'duplicateStrategy': 'SKIP'}, 'file': path.name}, body) d = body.get('data') or {} task_id = d.get('taskId') assert task_id, f'upload 无 taskId: {str(body)[:200]}' _, body = api('POST', '/api/customer/import/confirm', form={'taskId': str(task_id)}, step='H5 confirm') capture('POST', '/api/customer/import/confirm', {'form': {'taskId': str(task_id)}}, body) for _ in range(30): _, body = api('GET', '/api/customer/import/result', params={'taskId': str(task_id)}, step='result 轮询') st = (body.get('data') or {}).get('status') if st is not None and int(st) >= 2: # 0=DRAFT 1=RUNNING 2=DONE 3=FAILED break time.sleep(1) capture('GET', '/api/customer/import/result', {'query': {'taskId': str(task_id)}}, body) _, body = api('GET', '/api/customer/import/failures', params={'taskId': str(task_id)}, step='H5 failures(有失败行)') assert body.get('data'), 'failures 仍空集' capture('GET', '/api/customer/import/failures', {'query': {'taskId': str(task_id)}}, body) _, body = api('GET', '/api/customer/import/page', params={'current': '1', 'size': '10'}, step='H5 import/page') capture('GET', '/api/customer/import/page', {'query': {'current': '1', 'size': '10'}}, body) r = requests.get(BASE + '/api/customer/import/template', headers={'Authorization': f'Bearer {get_token(ADMIN)}'}, timeout=60) capture('GET', '/api/customer/import/template', {'query': {}}, {'_binary': True, 'contentType': r.headers.get('Content-Type'), 'bytes': len(r.content), 'note': 'xlsx 导入模板二进制流（EasyExcel 读取，不走 JSON 信封）'}) print(f' template: {r.headers.get("Content-Type")} {len(r.content)}B') return task_id def h_contact_import(a): """H6: 联系人导入三段式（1 坏行）→ contact/import/page 有任务。""" path = XLSX_DIR / 'contact-import.xlsx' mk_xlsx(path, '联系人', ['姓名', '职务（字典编码）', '手机号', '来源（字典编码）', '是否关键联系人(是/否)'], [['孙导入', 'other_job', '13800000105', 'referral', '是'], ['周坏行', 'other_job', '', 'referral', '']]) with open(path, 'rb') as f: _, body = api('POST', '/api/customer/contact/import/upload', form={'customerId': a['id'], 'duplicateStrategy': 'SKIP', 'importMode': 'UPSERT'}, files={'file': (path.name, f, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')}, step='H6 contact upload') capture('POST', '/api/customer/contact/import/upload', {'form': {'customerId': a['id'], 'duplicateStrategy': 'SKIP', 'importMode': 'UPSERT'}, 'file': path.name}, body) task_id = ((body.get('data') or {}).get('taskId')) if task_id: api('POST', '/api/customer/contact/import/confirm', form={'taskId': str(task_id)}, step='H6 confirm') _, body = api('GET', '/api/customer/contact/import/page', params={'customerId': a['id'], 'current': '1', 'size': '10'}, step='H6 contact/import/page') capture('GET', '/api/customer/contact/import/page', {'query': {'customerId': a['id'], 'current': '1', 'size': '10'}}, body) def h_customer_page(): """H7: 客户分页（旧版 POST /api/customer/page）。""" form = {'current': '1', 'size': '10', 'keyword': PFX} _, body = api('POST', '/api/customer/page', form=form, step='H7 customer/page') capture('POST', '/api/customer/page', {'form': form}, body) def h_preference(): """H8: 列偏好 save + get（contact scope；原值双保险恢复）。""" scope = 'contact.list' _, before = api('GET', '/api/preference/get', params={'scopeKey': scope}, step='偏好 pre-read') form = {'scopeKey': scope, 'visibleKeys': ['name', 'phone', 'jobTitleName'], 'columnOrder': ['name', 'phone', 'jobTitleName']} _, body = api('POST', '/api/preference/save', form=form, step='H8 preference/save') capture('POST', '/api/preference/save', {'form': form}, body) _, body = api('GET', '/api/preference/get', params={'scopeKey': scope}, step='H8 preference/get') capture('GET', '/api/preference/get', {'query': {'scopeKey': scope}}, body) # 原值恢复 bd = (before.get('data') or {}) if isinstance(before.get('data'), dict) else {} if bd: vis = bd.get('visibleKeys') or [] order = bd.get('columnOrder') or [] rform = {'scopeKey': scope} for i, k in enumerate(vis): rform[f'visibleKeys[{i}]'] = str(k) for i, k in enumerate(order): rform[f'columnOrder[{i}]'] = str(k) api('POST', '/api/preference/save', form=rform, step='偏好原值恢复') else: try: dbx("DELETE FROM user_column_preference WHERE scope_key=%s AND user_id=%s", (scope, int(ADMIN))) except Exception as e: print('偏好原值清理跳过:', e) def h_batch_ops(): """H9: assign/claim/release-pool 批量端点（各含失败行）。""" c1 = mk_customer(f'{PFX}-批A-{int(time.time())%100000}') c2 = mk_customer(f'{PFX}-批B-{int(time.time())%100000}') c3 = mk_customer(f'{PFX}-批C-{int(time.time())%100000}') # assign-batch：C1/C3 → BUDDY 成功 + 非法 id 失败（C3 同时为 H10 交割夹具铺路） form = {'ids': f"{c1['id']},{c3['id']},999888777666555444", 'userId': BUDDY} _, body = api('POST', '/api/customer/assign-batch', form=form, step='H9 assign-batch') capture('POST', '/api/customer/assign-batch', {'form': form}, body) d = body.get('data') or {} print(f" assign-batch: success={d.get('successCount')} fail={d.get('failCount')}") # release-pool-batch：把 C2（admin 名下）抛公海 + 非法 id 失败 form = {'ids': f"{c2['id']},999888777666555444"} _, body = api('POST', '/api/customer/release-pool-batch', form=form, step='H9 release-pool-batch') capture('POST', '/api/customer/release-pool-batch', {'form': form}, body) d = body.get('data') or {} print(f" release-pool-batch: success={d.get('successCount')} fail={d.get('failCount')}") # claim-batch：C2 在公海可领 + C1 已归属（BUDDY）失败 form = {'ids': f"{c2['id']},{c1['id']}"} _, body = api('POST', '/api/customer/claim-batch', form=form, step='H9 claim-batch') capture('POST', '/api/customer/claim-batch', {'form': form}, body) d = body.get('data') or {} print(f" claim-batch: success={d.get('successCount')} fail={d.get('failCount')}") return c1, c2, c3 def h_transfer_assign(c1, c2): """H10: 交割采集——**BUDDY 发起**（其名下仅本轮夹具客户，圈定不涉共享种子，F-13 陷阱规避）。 D22 手术（原值先查先存，finally 还原）：华南销售部 leader=admin + BUDDY 划入该部， resolveDirector 才有「启用的直属销售总监」可推导（此前 resign 直采 67011 根因）。 reason=transfer_post 传 toDirectorId=ADMIN（调动/区域调整路径，同时覆盖 toDirectorId 范围校验分支）； assign 用 [自有客户成功 + 非法 id 失败] 展示 CustomerBatchFailItem 行级失败形态（BatchRunner 部分成功语义）。 """ # 0) 圈定安全前置：BUDDY 名下活跃客户必须全为本轮夹具（整批原子换主防扫走他人客户） left = dbq("SELECT id, customer_name FROM customer WHERE owner_user_id=%s AND deleted=0 " "AND archive_status=1", (int(BUDDY),)) strangers = [r for r in left if not str(r['customer_name'] or '').startswith(PFX)] if strangers: print(f' 交割采集放弃：BUDDY 名下存在非夹具客户 {[str(r["customer_name"]) for r in strangers][:5]}') return # 1) D22 手术（原值先查先存） orig = dbq("SELECT leader_user_id FROM sys_dept WHERE id=%s", (DEPT_SALES,)) orig_leader = orig[0]['leader_user_id'] if orig else None orig_u = dbq("SELECT dept_id FROM crm_auth_user WHERE id=%s", (int(BUDDY),)) orig_buddy_dept = orig_u[0]['dept_id'] if orig_u else None try: dbx("UPDATE sys_dept SET leader_user_id=%s WHERE id=%s", (int(ADMIN), DEPT_SALES)) dbx("UPDATE crm_auth_user SET dept_id=%s WHERE id=%s", (DEPT_SALES, int(BUDDY))) _, pv = api('GET', '/api/customer/transfer/preview', uid=BUDDY, step='transfer preview(BUDDY)') capture('GET', '/api/customer/transfer/preview', {'query': {}}, pv) _, body = api('POST', '/api/customer/transfer/initiate', form={'reason': 'transfer_post', 'toDirectorId': ADMIN, 'remark': 'r3d 真值采集'}, uid=BUDDY, step='H10 initiate(BUDDY transfer_post)') capture('POST', '/api/customer/transfer/initiate', {'form': {'reason': 'transfer_post', 'toDirectorId': ADMIN, 'remark': 'r3d 真值采集'}}, body) tid = ((body.get('data') or {}).get('id')) if not tid: print(f' initiate 未返回 id: {str(body)[:160]}') return _, asg = api('GET', '/api/customer/transfer/assignable', params={'id': str(tid)}, uid=BUDDY, step='assignable') capture('GET', '/api/customer/transfer/assignable', {'query': {'id': str(tid)}}, asg) cand = ((asg.get('data') or [{}])[0] or {}).get('userId') or BUDDY form = {'id': str(tid), 'customerIds': f"{c1['id']},999888777666555444", 'assignUserId': str(cand)} _, body = api('POST', '/api/customer/transfer/assign', form=form, uid=BUDDY, step='H10 assign 失败行') capture('POST', '/api/customer/transfer/assign', {'form': form}, body) d = body.get('data') or {} print(f" transfer/assign: success={d.get('successCount')} fail={d.get('failCount')} " f"fails={[(x.get('id'), x.get('reason'), x.get('message')) for x in (d.get('failures') or [])]}") finally: # 2) 还原组织原值（夹具客户归属由 sweep 硬删兜底，无需逐条还原） if orig_leader is None: dbx("UPDATE sys_dept SET leader_user_id=NULL WHERE id=%s", (DEPT_SALES,)) else: dbx("UPDATE sys_dept SET leader_user_id=%s WHERE id=%s", (orig_leader, DEPT_SALES)) if orig_buddy_dept is not None: dbx("UPDATE crm_auth_user SET dept_id=%s WHERE id=%s", (orig_buddy_dept, int(BUDDY))) def h_quick_create(): """H11: quick-create needConfirm=true 分支（相似名命中 e2c-恒信达）。""" form = {'customerName': f'{PFX}恒信达科技有限公司'} _, body = api('POST', '/api/customer/quick-create', form=form, step='H11 quick-create needConfirm') capture('POST', '/api/customer/quick-create', {'form': form}, body) d = body.get('data') or {} print(f" needConfirm={d.get('needConfirm')} hits={len(d.get('similarHits') or [])}") def main(): sweep() a, cids = h_graph_and_contacts() h_quickadd_failed(a) h_batchedit_failed(a, cids) h_customer_import() h_contact_import(a) h_customer_page() h_preference() c1, c2, c3 = h_batch_ops() h_transfer_assign(c1, c2) h_quick_create() sweep() (OUT / 'specimens-doc-truth.json').write_text( json.dumps(specimens, ensure_ascii=False, indent=1), encoding='utf-8') print(f'\n== 采集完成：{len(specimens)} 端点 specimens → specimens-doc-truth.json ==') for k in sorted(specimens): print(' ', k) if __name__ == '__main__': main()