# -*- coding: utf-8 -*-
"""e2e-heavy.py — 客户重流程 API 级实测（票 05，F-13 交割 / F-14 导入 / F-15 提醒规则 / F-16 查重消费侧）

产出：report-heavy.md + specimens-heavy.json（票 06 合并素材）+ e2e-heavy-checks.json
输入：seed-ids.json（票 03）；e2e-core.py 同款框架（票 04 已验证基建）
纪律：flat 实路径；Result<Void> 判 code；401 重取 token；缺陷从 D-04 起；
      判定依据 原型 > spec > 问用户；库检每步新连接 autocommit=True。
"""
import sys, io, json, time, re, os, random, subprocess, datetime
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
import requests
import pymysql

BASE = 'http://localhost:8080'
ADMIN = '739564171091247104'   # 罗伟健（根部门「广东保伦」，data_scope=4）
BUDDY = '744842318024015872'   # 曾偲青（无角色；f13 交割发起人）
IDS = json.load(open('.scratch/customer-e2e/seed-ids.json', encoding='utf-8'))
C = {k: v['id'] for k, v in IDS['customers'].items()}
NO = {k: v['no'] for k, v in IDS['customers'].items()}
OPP = {k: v['id'] for k, v in IDS['opportunities'].items()}

DEPT_SEED = 744700334353416192        # 华南销售部(seed)：f13 接收总监推导部门
DEPT_BUDDY_ORIG = 744841292483133440  # 职员部：BUDDY 原部门（teardown 还原）
SALES_ID = '760000000000000001'        # e2c-销售甲 固定 id（幂等复用；字符串便于与 API/DB 比对）
SALES_NAME = 'e2c-销售甲'
JAVA_EXE = r'D:\IntelliJ IDEA 2025.1.3\jbr\bin\java.exe'  # corretto-17 起服 >120s（第三轮实测），JBR 实测 20-40s
PFX = 'e2c-'
TS = datetime.datetime.now().strftime('%H%M%S')

checks, defects, specimens = [], [], {}
_skips = set()


def check(flow, case, verdict, detail=''):
    checks.append({'flow': flow, 'case': case,
                   'verdict': {'pass': '✅', 'warn': '⚠', 'fail': '❌'}[verdict], 'detail': detail})
    print(f"  {verdict} [{flow}] {case}" + (f' — {detail}' if detail else ''))


def defect(did, severity, title, detail):
    defects.append({'id': did, 'severity': severity, 'title': title, 'detail': detail})
    print(f"  ⚑ 登记 {did}({severity}): {title}")


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 get_token(uid, tries=3):
    # debug/token 偶发瞬时故障（Redis Connection reset）→ 重试 + fail fast（票 04 教训）。
    last = None
    for _ in range(tries):
        try:
            r = requests.get(f'{BASE}/api/auth/debug/token', params={'userId': uid}, timeout=30)
            if r.status_code == 200:
                d = r.json().get('data')
                tok = d if isinstance(d, str) else (d or {}).get('token')
                if tok:
                    return tok
                last = f'body={r.text[:120]}'
            else:
                last = f'HTTP {r.status_code} {r.text[:120]}'
        except Exception as e:
            last = str(e)
        time.sleep(2)
    raise RuntimeError(f'debug/token 获取失败 userId={uid}: {last}')


S = requests.Session()
S.headers['Authorization'] = f'Bearer {get_token(ADMIN)}'
B = requests.Session()
B.headers['Authorization'] = f'Bearer {get_token(BUDDY)}'


def _capture(method, path, req, resp_body):
    path = re.sub(r'/\d{15,}', '/{id}', path)  # 路径 id 归一化：票 06 按端点归档
    key = f'{method} {path}'
    if key in _skips:
        return
    _skips.add(key)
    is_json = isinstance(req, (dict, list)) and req is not None and method != 'FORM'
    specimens[key] = {
        'request': {'contentType': 'application/json' if is_json else 'application/x-www-form-urlencoded',
                    'body': req if is_json else req},
        'response': resp_body,
    }


def api(sess, method, path, form=None, json_body=None, params=None, step='', retry401=True):
    url = BASE + path
    try:
        if json_body is not None:
            r = sess.post(url, json=json_body, params=params, timeout=30) if method == 'POST' \
                else sess.put(url, json=json_body, params=params, timeout=30)
        elif method == 'POST':
            r = sess.post(url, data=form, params=params, timeout=30)
        elif method == 'PUT':
            r = sess.put(url, data=form, params=params, timeout=30)
        elif method == 'DELETE':
            r = sess.delete(url, params=params, timeout=30)
        else:
            r = sess.request(method, url, params=params, timeout=30)
    except Exception as e:
        print(f'  ✘ {step}: 网络异常 {e}')
        return None
    if r.status_code == 401 and retry401:
        uid = BUDDY if sess is B else ADMIN
        sess.headers['Authorization'] = f'Bearer {get_token(uid)}'
        return api(sess, method, path, form, json_body, params, step, retry401=False)
    if r.status_code != 200:
        _capture(method, path, json_body if json_body is not None else form,
                 {'_http': r.status_code, '_raw': r.text[:400]})
        print(f'  ✘ {step}: HTTP {r.status_code} {r.text[:140]}')
        return {'_http': r.status_code, 'code': r.status_code, 'message': r.text[:200]}
    try:
        body = r.json()
    except ValueError:
        _capture(method, path, json_body if json_body is not None else form, {'_raw': r.text[:400]})
        print(f'  ✘ {step}: 响应非 JSON')
        return None
    _capture(method, path, json_body if json_body is not None else form, body)
    return body


def api_raw(sess, method, path, **kw):
    """不经 _capture 的裸请求（二进制/表单文件上传/自定义头检查用）。"""
    return sess.request(method, BASE + path, timeout=60, **kw)


def data_of(resp):
    return resp.get('data') if isinstance(resp, dict) else None


def code_of(resp):
    return resp.get('code') if isinstance(resp, dict) else None


def expect_code(resp, want, flow, case):
    got = code_of(resp)
    if got is not None and int(got) == want:
        check(flow, case, 'pass', f'code={got} 如预期')
        return True
    check(flow, case, 'fail', f'期望 code={want}，实际 code={got} msg={str(resp.get("message") if isinstance(resp, dict) else resp)[:120]}')
    return False


def dict_one(group, parent_id=None):
    try:
        if parent_id:
            rows = dbq("SELECT i.code c FROM dict_item i JOIN dict_group g ON i.group_id=g.id "
                       "WHERE g.code=%s AND i.deleted=0 AND i.parent_id=%s ORDER BY i.sort_no LIMIT 1",
                       (group, parent_id))
        else:
            rows = dbq("SELECT i.code c FROM dict_item i JOIN dict_group g ON i.group_id=g.id "
                       "WHERE g.code=%s AND i.deleted=0 AND i.parent_id IS NULL ORDER BY i.sort_no LIMIT 1",
                       (group,))
        return rows[0]['c'] if rows else None
    except Exception:
        return None


CTYPE = dict_one('customer_type') or 'customer_type_01'
GOV = dict_one('industry') or 'gov'
FOLLOW_WAY = dict_one('follow_way') or 'phone'


# ==================== setup / teardown ====================

_g = {'sales_id': SALES_ID, 'buddy_leftover': None, 'transfer_id': None, 'transfer_no': None,
      'sales_assignable': None}


def setup_heavy():
    """f13 前提：华南销售部 leader=admin；seed 销售甲；BUDDY 划入销售部；甲乙丙(+BUDDY 残留)圈定。"""
    print('== setup_heavy ==')
    # 1) 华南销售部 leader=admin（resolveDirector 数据前提，teardown 还原 NULL）
    dbx("UPDATE sys_dept SET leader_user_id=%s WHERE id=%s", (int(ADMIN), DEPT_SEED))
    check('SETUP', '华南销售部 leader=admin', 'pass')
    # 2) seed 销售甲（幂等：固定 id 先删后插，动态补齐非空列）
    cols = dbq("SHOW COLUMNS FROM crm_auth_user")
    row = {c['Field']: None for c in cols}
    row.update({'id': int(SALES_ID), 'username': SALES_NAME, 'dept_id': DEPT_SEED,
                'enabled': 1, 'employment_status': 'active', 'deleted': 0})
    notnull = [c['Field'] for c in cols if c['Null'] == 'NO' and c['Default'] is None and c['Extra'] == '']
    missing = [c for c in notnull if row.get(c) is None]
    if missing:
        raise RuntimeError(f'crm_auth_user 缺不可空列无法 seed 销售甲: {missing}')
    dbx("DELETE FROM crm_auth_user WHERE id=%s", (SALES_ID,))
    dbx(f"INSERT INTO crm_auth_user ({','.join(row)}) VALUES ({','.join(['%s'] * len(row))})",
        tuple(row.values()))
    check('SETUP', 'seed 销售甲(华南销售部/在职/启用)', 'pass', f'id={SALES_ID}')
    # 3) BUDDY 划入华南销售部（teardown 还原职员部）
    dbx("UPDATE crm_auth_user SET dept_id=%s WHERE id=%s", (DEPT_SEED, BUDDY))
    # 4) 甲乙丙 assign 给 BUDDY（API 路径，owner→BUDDY 才能由 BUDDY 发起交割）
    for k in ('transfer1', 'transfer2', 'transfer3'):
        r = api(S, 'POST', '/api/customer/assign',
                params={'id': C[k], 'userId': BUDDY}, step=f'assign {k}→BUDDY')
        if code_of(r) != 0:
            raise RuntimeError(f'setup assign {k} 失败: {r}')
    check('SETUP', '甲乙丙 assign→BUDDY', 'pass')
    # 5) BUDDY 名下全量（=交接单圈定范围：甲乙丙 + 残留）
    left = dbq("SELECT id, customer_name FROM customer WHERE owner_user_id=%s AND deleted=0 "
               "AND archive_status=1", (BUDDY,))
    _g['buddy_leftover'] = [{'id': str(r['id']), 'name': r['customer_name']} for r in left]
    print(f"  BUDDY 名下 {len(left)} 条：{[x['name'] for x in _g['buddy_leftover']]}")


def teardown_heavy():
    """还原：客户归属/商机 owner→admin；BUDDY 部门；部门 leader。交接单/销售甲留痕（报告注明）。"""
    print('== teardown_heavy ==')
    n = 0
    for k in ('transfer1', 'transfer2', 'transfer3'):
        n += dbx("UPDATE customer SET owner_user_id=%s, owner_user_name_snapshot='罗伟健', "
                 "owner_dept_id=744841292348915712, owner_dept_name_snapshot='广东保伦电子股份有限公司', "
                 "enter_pool_time=NULL WHERE id=%s", (ADMIN, C[k]))
    if _g.get('buddy_leftover'):
        for x in _g['buddy_leftover']:
            n += dbx("UPDATE customer SET owner_user_id=%s, owner_user_name_snapshot='罗伟健', "
                     "owner_dept_id=744841292348915712, owner_dept_name_snapshot='广东保伦电子股份有限公司', "
                     "enter_pool_time=NULL WHERE id=%s", (ADMIN, x['id']))
    dbx("UPDATE opportunity o JOIN opportunity_customer oc ON oc.opportunity_id=o.id "
        "SET o.owner_user_id=%s, o.owner_name_snapshot='罗伟健', o.owner_dept_id=744841292348915712 "
        "WHERE oc.customer_id IN (%s,%s,%s) AND oc.delete_key=0",
        (ADMIN, C['transfer1'], C['transfer2'], C['transfer3']))
    dbx("UPDATE crm_auth_user SET dept_id=%s WHERE id=%s", (DEPT_BUDDY_ORIG, BUDDY))
    dbx("UPDATE sys_dept SET leader_user_id=NULL WHERE id=%s", (DEPT_SEED,))
    print(f'  还原客户 owner {n} 条；商机 owner 已还原 admin；BUDDY 部门/leader 已还原')


# ==================== F-13 客户交割 ====================

def f13():
    print('== F-13 客户交割 ==')
    F = 'F-13'
    total_scope = len(_g['buddy_leftover'])  # 交接单圈定 = BUDDY 名下全量（甲乙丙 + 残留）

    # 13.1 preview：发起前名下可交接客户
    r = api(B, 'GET', '/api/customer/transfer/preview', step='preview')
    rows = data_of(r) or []
    ids_prev = {str(x.get('customerId')) for x in rows}
    ok = len(rows) == total_scope and C['transfer1'] in ids_prev and C['transfer3'] in ids_prev
    check(F, f'preview 圈定 {total_scope} 条(甲乙丙+残留)', 'pass' if ok else 'fail',
          f'实际 {len(rows)} 条')

    # 13.2 reason 缺失 → 67011
    r = api(B, 'POST', '/api/customer/transfer/initiate', form={}, step='initiate 无 reason')
    expect_code(r, 67011, F, 'initiate reason 缺失 → 67011')

    # 13.3 toDirectorId 越范围（传 BUDDY 自己，≠leader）→ 67011（D22 范围校验）
    r = api(B, 'POST', '/api/customer/transfer/initiate',
            form={'reason': 'region_adjust', 'toDirectorId': BUDDY}, step='initiate 越范围总监')
    expect_code(r, 67011, F, 'toDirectorId 不在可交接范围 → 67011')

    # 13.4 resign 自动推导唯一接收总监（D22）→ JG 编号 + PENDING
    r = api(B, 'POST', '/api/customer/transfer/initiate', form={'reason': 'resign', 'remark': 'e2e 票05'},
            step='initiate resign 自动推导')
    d = data_of(r) or {}
    tno = str(d.get('transferNo') or '')
    today = datetime.date.today().strftime('%Y%m%d')
    ok = code_of(r) == 0 and tno.startswith('JG' + today) and len(tno) == 14
    _g['transfer_id'] = d.get('id')
    _g['transfer_no'] = tno
    check(F, f'initiate resign → 交接单 {tno}（D22 自动推导罗伟健）',
          'pass' if ok else 'fail', f'status={d.get("status")} totalCount={d.get("totalCount")}')
    check(F, 'totalCount=圈定全量（无 customerIds 字段=名下全量口径）',
          'pass' if int(d.get('totalCount') or 0) == total_scope else 'warn',
          f'totalCount={d.get("totalCount")} vs 预览 {total_scope}')
    tid = _g['transfer_id']

    # 13.5 重入 initiate → 67010
    r = api(B, 'POST', '/api/customer/transfer/initiate', form={'reason': 'resign'}, step='重入')
    expect_code(r, 67010, F, '存在未完成交接单重入 → 67010')

    # 13.6 DB：整批原子——名下客户 owner→总监（admin）四件套
    row = dbq("SELECT owner_user_id, owner_user_name_snapshot, owner_dept_id, enter_pool_time "
              "FROM customer WHERE id=%s", (C['transfer1'],))
    ok = row and str(row[0]['owner_user_id']) == ADMIN and row[0]['enter_pool_time'] is None
    check(F, 'initiate 后客户 owner→接收总监(整批原子)', 'pass' if ok else 'fail',
          f"owner={row[0]['owner_user_id'] if row else '?'}")

    # 13.7 DB：商机 owner 三列联动（真源同步）
    orow = dbq("SELECT o.owner_user_id, o.owner_name_snapshot, o.owner_dept_id FROM opportunity o "
               "JOIN opportunity_customer oc ON oc.opportunity_id=o.id "
               "WHERE oc.customer_id=%s AND oc.delete_key=0", (C['transfer1'],))
    ok = orow and str(orow[0]['owner_user_id']) == ADMIN and orow[0]['owner_name_snapshot'] == '罗伟健'
    check(F, '商机 owner 三列联动(owner/快照/部门)', 'pass' if ok else 'fail',
          f"e2c-联动-交割 owner={orow[0]['owner_user_id'] if orow else '?'}")

    # 13.8 DB：oplog 系统叙事含交接编号
    lrow = dbq("SELECT detail FROM customer_oplog WHERE customer_id=%s AND action='TRANSFER' "
               "ORDER BY id DESC LIMIT 1", (C['transfer1'],))
    ok = lrow and tno in (lrow[0]['detail'] or '') and '发起交接' in (lrow[0]['detail'] or '')
    check(F, 'oplog 含交接编号+发起交接叙事', 'pass' if ok else 'fail',
          (lrow[0]['detail'] or '')[:60] if lrow else '无记录')

    # 13.9 detail 明细
    r = api(B, 'GET', f'/api/customer/transfer/{tid}', step='detail')
    d = data_of(r) or {}
    det = d.get('details') or []
    ok = len(det) == total_scope and all(int(x.get('assignStatus') or 0) == 0 for x in det)
    check(F, 'detail 明细齐且全待分配', 'pass' if ok else 'fail', f'minutes={len(det)}')
    detail_ids = [str(x.get('customerId')) for x in det]

    # 13.10 assignable（D-07：根部门全树查询稳定超时 → 缺陷验证模式，双次重试留证据）
    users, uids = [], set()
    for _ in range(2):
        r = api(B, 'GET', f'/api/customer/transfer/{tid}/assignable', step='assignable')
        users = data_of(r) or []
        uids = {str(x.get('userId')) for x in users}
        if uids:
            break
        time.sleep(2)
    _g['sales_assignable'] = bool(uids)
    if uids:
        ok = SALES_ID in uids and BUDDY not in uids and ADMIN not in uids
        check(F, 'assignable 含销售甲且排除发起人/总监', 'pass' if ok else 'fail',
              f'共 {len(users)} 人')
    else:
        check(F, 'assignable 含销售甲且排除发起人/总监', 'warn',
              '接口 30s×2 稳定超时未返回 → 登记 D-07')
        defect('D-07', 'P1', '交割可分配销售列表接口超时',
               '复现：发起交割后 GET /api/customer/transfer/{id}/assignable → '
               '30s（E2E）与 60s（手动 probe）均稳定超时，两轮各复现 2 次。'
               '前端「选择分配对象」无法加载。疑因 getChildDeptIds 根部门全树递归 + '
               '用户过滤全表扫描；assign 本体不受影响（直接指定 userId 成功）。待复核。')

    # 13.11a assign 单条（甲→销售甲，逐条独立事务）
    r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
            form=[('customerIds', C['transfer1']), ('assignUserId', SALES_ID)],
            step='assign 甲→销售甲')
    br = data_of(r) or {}
    check(F, 'assign 单条 甲→销售甲', 'pass' if int(br.get('successCount') or 0) == 1 else 'fail',
          f'fails={br.get("failures")}')

    # 13.11b 重复 assign 已分配明细 → 67012（失败行）
    r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
            form=[('customerIds', C['transfer1']), ('assignUserId', SALES_ID)],
            step='重复 assign 甲')
    br = data_of(r) or {}
    fails = br.get('failures') or []
    ok = int(br.get('successCount') or 0) == 0 and fails
    check(F, '重复分配已分配明细 → 67012(失败行)', 'pass' if ok else 'fail',
          f'{fails[0] if fails else br}')

    # 13.11c 分配给当前负责人 → 67012（乙明细仍 PENDING，临时把乙 owner 指向 BUDDY）
    dbx("UPDATE customer SET owner_user_id=%s WHERE id=%s", (BUDDY, C['transfer2']))
    r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
            form=[('customerIds', C['transfer2']), ('assignUserId', BUDDY)],
            step='分配给当前负责人')
    br = data_of(r) or {}
    fails = br.get('failures') or []
    ok = int(br.get('successCount') or 0) == 0 and fails
    check(F, '分配给当前负责人 → 67012(失败行)', 'pass' if ok else 'warn',
          f'{fails[0] if fails else br}')
    dbx("UPDATE customer SET owner_user_id=%s WHERE id=%s", (ADMIN, C['transfer2']))

    # 13.11d 无效分配对象 → 67012（spec 写 66001，实现统一收口 67012 = 契约发现）
    bad = 999999999999
    r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
            form=[('customerIds', C['transfer2']), ('assignUserId', bad)], step='分配无效用户')
    got = code_of(r)
    br = data_of(r) or {}
    fails = br.get('failures') or []
    ok = (got == 67012) or (got == 0 and fails and int(fails[0].get('code') or fails[0].get('errorCode') or 0) == 67012)
    check(F, '分配对象不存在 → 67012(spec 66001 偏差记录)', 'pass' if ok else 'warn',
          f'code={got} fails={fails[:1]}')

    # 13.11e 剩余明细批量 assign → 全部完成
    rest = [i for i in detail_ids if i != str(C['transfer1'])]
    r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
            form=[('customerIds', i) for i in rest] + [('assignUserId', SALES_ID)],
            step='assign 剩余→销售甲')
    br = data_of(r) or {}
    check(F, f'批量 assign 剩余成功 {br.get("successCount")}/{len(rest)}',
          'pass' if int(br.get('successCount') or 0) == len(rest) else 'fail',
          f'fails={br.get("failures")}')

    # 13.12 D23 自动完成
    r = api(B, 'GET', f'/api/customer/transfer/{tid}', step='detail after assign')
    d = data_of(r) or {}
    ok = int(d.get('assignedCount') or 0) == int(d.get('totalCount') or -1) and int(d.get('status') or -1) == 1
    check(F, 'assignedCount==totalCount → 自动置已完成(D23)', 'pass' if ok else 'fail',
          f"assigned={d.get('assignedCount')}/{d.get('totalCount')} status={d.get('status')}")

    # 13.13 已完成单再 assign → 状态校验（实现在 BatchResult 失败行内收口，非抛 67004）
    r = api(S, 'POST', f'/api/customer/transfer/{tid}/assign',
            form=[('customerIds', C['transfer1']), ('assignUserId', SALES_ID)],
            step='已完成单再分配')
    got = code_of(r)
    br = data_of(r) or {}
    fails = br.get('failures') or []
    ok = (got == 67004) or (got == 0 and fails and '不可再分配' in str(fails[0].get('message', '')))
    check(F, '交接单已完成不可再分配（67004 或失败行收口）', 'pass' if ok else 'fail',
          f'code={got} fails={fails[:1]}')

    # 13.16 DB：分配后客户 owner=销售甲 + 商机二次联动
    row = dbq("SELECT owner_user_id, owner_user_name_snapshot FROM customer WHERE id=%s", (C['transfer1'],))
    orow = dbq("SELECT o.owner_user_id, o.owner_name_snapshot FROM opportunity o "
               "JOIN opportunity_customer oc ON oc.opportunity_id=o.id "
               "WHERE oc.customer_id=%s AND oc.delete_key=0", (C['transfer1'],))
    ok = row and str(row[0]['owner_user_id']) == SALES_ID and orow and str(orow[0]['owner_user_id']) == SALES_ID
    check(F, 'assign 后客户/商机 owner→销售甲', 'pass' if ok else 'fail',
          f"cust={row[0]['owner_user_id'] if row else '?'} opp={orow[0]['owner_user_id'] if orow else '?'}")

    # 13.17 page 按 transferNo 过滤
    r = api(B, 'GET', '/api/customer/transfer/page', params={'transferNo': tno, 'pageSize': 10},
            step='page 交接编号过滤')
    recs = (data_of(r) or {}).get('content') or []
    ok = any(str(x.get('transferNo')) == tno for x in recs)
    check(F, 'page 交接编号过滤命中', 'pass' if ok else 'fail', f'命中 {len(recs)} 条')

    # 13.18 交接编号当日续号（DB 断言当日 max 即本单）
    mx = dbq("SELECT MAX(transfer_no) m FROM customer_transfer WHERE transfer_no LIKE %s",
             ('JG' + today + '%',))
    check(F, '交接编号当日续号不重号', 'pass' if mx and str(mx[0]['m']) == tno else 'fail',
          f"max={mx[0]['m'] if mx else '?'} vs 本单 {tno}")


# ==================== F-14 客户导入 ====================

def _mk_xlsx(path, cust_rows, contact_rows=None, header_override=None):
    """openpyxl 造 EasyExcel 兼容 xlsx：sheet0=客户(9 列 index 绑定) sheet1=联系人(5 列)。"""
    from openpyxl import Workbook
    wb = Workbook()
    ws = wb.active
    ws.title = '客户'
    headers = header_override or ['客户编号', '客户名称', '客户类型', '统一社会信用代码',
                                  '省份编码', '城市编码', '行业编码', '客户星级(1-5)', '备注']
    ws.append(headers)
    for row in cust_rows:
        ws.append([(row[i] if i < len(row) else None) for i in range(9)])
    ws2 = wb.create_sheet('联系人')
    ws2.append(['客户编号', '联系人姓名', '联系电话', '职务名称', '是否关键联系人(是/否)'])
    for row in (contact_rows or []):
        ws2.append([(row[i] if i < len(row) else None) for i in range(5)])
    wb.save(path)


def _upload(sess, path, import_mode, step):
    with open(path, 'rb') as f:
        r = api_raw(sess, 'POST', '/api/customer/import/upload',
                    files={'file': (os.path.basename(path), f,
                                    'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')},
                    data={'importMode': importMode_fix(import_mode)})
    if r.status_code != 200:
        _capture('POST', '/api/customer/import/upload', {'importMode': import_mode},
                 {'_http': r.status_code, '_raw': r.text[:400]})
        print(f'  ✘ {step}: HTTP {r.status_code} {r.text[:140]}')
        return {'_http': r.status_code, 'code': r.status_code, 'message': r.text[:200]}
    body = r.json()
    _capture('POST', '/api/customer/import/upload', {'importMode': import_mode}, body)
    return body


def importMode_fix(m):
    return m


def f14():
    print('== F-14 客户导入 ==')
    F = 'F-14'
    XLSX = '.scratch/customer-e2e/_import-heavy.xlsx'
    new_name = f'{PFX}导入-新增-{TS}'

    # 14.1 模板下载：RFC 5987 + xlsx 魔数
    r = api_raw(S, 'GET', '/api/customer/import/template')
    cd = r.headers.get('Content-Disposition', '')
    ok = (r.status_code == 200 and r.content[:2] == b'PK'
          and ("filename*=UTF-8''" in cd or 'filename*=' in cd))
    check(F, '模板下载(RFC5987 文件名+xlsx 魔数)', 'pass' if ok else 'fail', f'CD={cd[:80]}')

    ind = dbq("SELECT i.code c FROM dict_item i JOIN dict_group g ON i.group_id=g.id "
              "WHERE g.code='industry' AND i.deleted=0 AND i.parent_id IS NOT NULL "
              "ORDER BY i.sort_no LIMIT 1")
    ind_code = ind[0]['c'] if ind else ''
    # 14.2 构造 7 行客户 + 3 行联系人
    #  r1 带行内编号（联系人 sheet 需按编号挂接）；r2/r3/r4 必须带名称（名称必填先于匹配/重复组判定）；
    #  r5 备注列非空防 EasyExcel 忽略全空行
    new_no = f'E2C-IMP-{TS}'
    cust = [
        [new_no, new_name, CTYPE, None, '440000', '440100', ind_code, '3', '票05 新增行'],  # r1 INSERT
        [NO['transfer2'], PFX + '交割-乙', None, None, None, None, None, '4', None],        # r2 按编号 UPDATE 乙
        [NO['transfer3'], PFX + '交割-丙', None, None, None, None, None, None, None],       # r3 重复组A
        [NO['transfer3'], PFX + '交割-丙', None, None, None, None, None, None, None],       # r4 重复组B（D24）
        [None, None, None, None, None, None, None, None, '缺名称行'],                       # r5 缺名称
        [None, f'{PFX}导入-星级-{TS}', CTYPE, None, None, None, None, '6', None],           # r6 星级越界
        [None, f"{dbq('SELECT customer_name n FROM customer WHERE id=%s', (IDS['customers']['simA']['id'],))[0]['n']}",
         None, None, None, None, None, None, None],                                        # r7 与 simA 同名 → SUSPECT
    ]
    contacts = [
        [new_no, 'e2c-联系人甲', '13800007701', '项目经理', '是'],   # 挂 r1（文件内新增客户）
        [None, 'e2c-联系人乙', None, None, None],                  # 缺编号
        ['E2C-NOT-EXIST', 'e2c-联系人丙', '13800007702', '组长', None],  # 编号不存在
    ]
    _mk_xlsx(XLSX, cust, contacts)

    # 14.3 非法 importMode → 67013
    r = _upload(S, XLSX, 'BAD_MODE', '非法 mode')
    expect_code(r, 67013, F, 'importMode 非法 → 67013')

    # 14.4 表头错列 → 67013（模板版本锚点逐列比对）
    _mk_xlsx(XLSX, cust[:1], header_override=['客户编号', '客户名称X', '客户类型', '统一社会信用代码',
                                              '省份编码', '城市编码', '行业编码', '客户星级(1-5)', '备注'])
    r = _upload(S, XLSX, 'APPEND_ONLY', '表头错列')
    expect_code(r, 67013, F, '模板表头不符 → 67013')

    # 14.5 空数据行 → 67013
    _mk_xlsx(XLSX, [])
    r = _upload(S, XLSX, 'APPEND_ONLY', '空数据')
    expect_code(r, 67013, F, '文件内无数据行 → 67013')

    # 14.6 合法文件 APPEND_ONLY 上传 → DRAFT 预检计数
    _mk_xlsx(XLSX, cust, contacts)
    r = _upload(S, XLSX, 'APPEND_ONLY', 'APPEND_ONLY 上传')
    d = data_of(r) or {}
    task_id = d.get('taskId')
    ok = (code_of(r) == 0 and str(d.get('totalCount')) == '7'
          and str(d.get('insertCount')) == '1' and str(d.get('updateCount')) == '0'
          and str(d.get('failCount')) == '5' and str(d.get('suspectCount')) == '1')
    # APPEND_ONLY：r2 已存在 → fail「客户已存在」；insert=r1；suspect=r7
    check(F, 'APPEND_ONLY 预检：t=7 insert=1 fail=5(含已存在) suspect=1',
          'pass' if ok else 'fail', f"t={d.get('totalCount')} i={d.get('insertCount')} "
          f"u={d.get('updateCount')} f={d.get('failCount')} s={d.get('suspectCount')}")

    # 14.7 行级 verdict 语义抽查（预览 rows）
    pmap = {(x.get('sheetName'), str(x.get('rowNum'))): x for x in (d.get('rows') or [])
            if x.get('sheetName') != '联系人'}
    v2 = pmap.get(('客户', '2'), {}).get('verdict')
    v3 = pmap.get(('客户', '3'), {}).get('reason', '')
    v7 = pmap.get(('客户', '7'), {})
    ok = ('FAIL' in str(v2) or 'fail' in str(v2)) and '重复时不导入' in v3 \
        and 'SUSPECT' in str(v7.get('verdict', ''))
    check(F, '行级判定：已存在 FAIL / 文件内重复组 FAIL(D24) / 相似 SUSPECT',
          'pass' if ok else 'fail', f"r2={v2} r3={v3[:24]} r7={v7.get('verdict')}")

    # 14.8 UPDATE_ONLY：r1 新客户行 → fail「未找到可更新客户」；r2 按编号 UPDATE
    r = _upload(S, XLSX, 'UPDATE_ONLY', 'UPDATE_ONLY')
    d2 = data_of(r) or {}
    ok = code_of(r) == 0 and str(d2.get('updateCount')) == '1' and str(d2.get('insertCount')) == '0'
    check(F, 'UPDATE_ONLY 预检：update=1(按编号) insert=0', 'pass' if ok else 'fail',
          f"u={d2.get('updateCount')} i={d2.get('insertCount')}")

    # 14.9 UPSERT 独立文件：新编号 + 已有客户同名 → 不按名称匹配（SUSPECT 而非 UPDATE）
    up_no = f'E2C-UP-{TS}'
    _mk_xlsx(XLSX, [[up_no, f"{dbq('SELECT customer_name n FROM customer WHERE id=%s', (IDS['customers']['simA']['id'],))[0]['n']}",
                      None, None, None, None, None, None, None]])
    r = _upload(S, XLSX, 'UPSERT', 'UPSERT 同名新编号')
    d3 = data_of(r) or {}
    pmap3 = {(x.get('sheetName'), str(x.get('rowNum'))): x for x in (d3.get('rows') or [])
             if x.get('sheetName') != '联系人'}
    v1u = pmap3.get(('客户', '1'), {})
    ok = ('SUSPECT' in str(v1u.get('verdict', ''))
          and '疑似重复' in str(v1u.get('reason', '')))
    check(F, 'UPSERT 同名(新编号)→SUSPECT 不按名称匹配', 'pass' if ok else 'warn',
          f"r1={v1u.get('verdict')} {str(v1u.get('reason'))[:30]}")

    # 14.10 confirm 提交 + 状态机 67014
    r = api(S, 'POST', f'/api/customer/import/{task_id}/confirm', step='confirm')
    ok = code_of(r) == 0
    check(F, 'confirm DRAFT→RUNNING 异步执行', 'pass' if ok else 'fail', f'taskId={task_id}')
    r = api(S, 'POST', f'/api/customer/import/{task_id}/confirm', step='重复 confirm')
    expect_code(r, 67014, F, '重复 confirm → 67014')

    # 14.11 轮询 result → 终态（D-05 阻断下预期 FAILED；修复后应为 DONE 行级部分成功）
    status, result = None, {}
    for _ in range(30):
        time.sleep(1)
        r = api(S, 'GET', f'/api/customer/import/{task_id}', step='poll result')
        result = data_of(r) or {}
        status = result.get('status')
        if str(status) in ('2', '3'):
            break
    if str(status) == '3':
        check(F, '执行终态', 'warn',
              f"status=FAILED fail_reason={result.get('failReason')}（D-05 导入 INSERT 缺省列阻断）")
        defect('D-05', 'P1',
               '导入新增客户行必失败：customer.is_biz_negotiated NOT NULL 无默认值，createCustomer 未赋值',
               '复现：APPEND_ONLY/UPSERT 上传含新增行文件 → confirm → 单行 INSERT 报 '
               "Field 'is_biz_negotiated' doesn't have a default value（日志 23:37:05 task-4）→ "
               'catch 后 insertFail 把完整 MyBatis 异常文本写入 customer_import_fail.fail_reason '
               'varchar(500) → Data too long → 整任务 FAILED、行级明细全丢（fail 表 0 行）。'
               '期望：实体默认值或 INSERT 前补 0；insertFail reason 截断防放大。')
    else:
        ok = str(status) == '2' and str(result.get('insertCount')) == '1' \
            and str(result.get('updateCount')) == '1'
        check(F, '执行完成 DONE：insert=1 update=1', 'pass' if ok else 'fail',
              f"status={status} i={result.get('insertCount')} u={result.get('updateCount')} "
              f"f={result.get('failCount')} s={result.get('suspectCount')}")

    # 14.12 failures 明细（D-05 阻断下 fail 表 0 行 = 明细丢失实证）
    r = api(S, 'GET', f'/api/customer/import/{task_id}/failures', step='failures')
    fails = data_of(r) or []
    check(F, f'失败明细 {len(fails)} 行', 'warn' if str(status) == '3' and not fails else 'pass',
          '；'.join(str(x.get('reason'))[:20] for x in fails[:3]) or 'D-05 阻断：明细丢失')

    # 14.13 DB：任务态落库证据（r1 未落库 = D-05 实证；r2 未更新 = 阻断实证）
    row = dbq("SELECT id FROM customer WHERE customer_no=%s AND deleted=0", (new_no,))
    row2 = dbq("SELECT customer_star_level FROM customer WHERE id=%s", (C['transfer2'],))
    check(F, 'r1 新客户未落库（D-05）/r2 未更新（阻断）', 'warn',
          f"r1={'未落库' if not row else '落库'} r2星={row2[0]['customer_star_level'] if row2 else '?'}")
    if row:
        _g['import_new_id'] = str(row[0]['id'])

    # 14.14 page 仅当前用户可见（PageResult 序列化字段=content，非 records）
    r = api(S, 'GET', '/api/customer/import/page', params={'pageSize': 50}, step='page admin')
    mine = {str(x.get('taskId')) for x in (data_of(r) or {}).get('content') or []}
    r = api(B, 'GET', '/api/customer/import/page', params={'pageSize': 50}, step='page buddy')
    theirs = {str(x.get('taskId')) for x in (data_of(r) or {}).get('content') or []}
    if str(task_id) not in mine:
        check(F, '最近导入仅提交人可见(creator 过滤)', 'warn',
              f'admin page 未含本 task（buddy 有={str(task_id) in theirs}）——需复核 creator_id 过滤')
        defect('D-06', 'P1',
               '导入任务分页查不到本人任务',
               '复现：任意导入任务（DB creator_id 正确=' + str(ADMIN) + '）→ GET '
               '/api/customer/import/page 以提交人身份查询 → content 不含该任务。'
               '前端「最近导入记录」列表将恒为空。待复核 eq(creatorId, getRequiredUserId) 条件。')
    else:
        ok = str(task_id) not in theirs
        check(F, '最近导入仅提交人可见(creator 过滤)', 'pass' if ok else 'fail',
              f'admin 有={str(task_id) in mine} buddy 有={str(task_id) in theirs}')


def f14_cleanup():
    print('== f14_cleanup ==')
    nid = _g.get('import_new_id')
    if nid:
        dbx("DELETE FROM customer_oplog WHERE customer_id=%s", (nid,))
        dbx("DELETE FROM customer_contact WHERE customer_id=%s", (nid,))
        dbx("DELETE FROM customer WHERE id=%s", (nid,))
        print(f'  已清导入新客户 {nid}')
    dbx("UPDATE customer SET customer_name=%s, customer_star_level=3 WHERE id=%s",
        (PFX + '交割-乙', C['transfer2']))
    print('  乙名称/星级已还原')


# ==================== 服务重启（f15 Job 观测用，U-02 拍板：env 注入 cron 实测） ====================

JAR = r'e:\code\crm-backend-matt\crm-app\target\crm-app-1.0.0-SNAPSHOT.jar'


def _pid_on_8080():
    out = subprocess.run('netstat -ano | findstr :8080 | findstr LISTENING',
                         shell=True, capture_output=True, text=True).stdout
    for line in out.splitlines():
        parts = line.split()
        if len(parts) >= 5 and parts[3] == 'LISTENING':
            return int(parts[4])
    return None


def restart_service(extra_env=None, wait=300):
    pid = _pid_on_8080()
    if pid:
        subprocess.run(f'taskkill /F /PID {pid}', shell=True, capture_output=True)
        for _ in range(15):  # 等 LISTENING 消失（端口释放）再起，避免绑定失败
            time.sleep(1)
            if not _pid_on_8080():
                break
    env = dict(os.environ)
    env.update({'SPRING_PROFILES_ACTIVE': 'verify', 'CRM_MINIO_AK': 'admin',
                'CRM_MINIO_SK': 'Itc@123456'})
    env.update(extra_env or {})
    print(f'  -- 起服 {os.path.basename(JAVA_EXE)} extra_env={extra_env or "默认"} --', flush=True)
    subprocess.Popen([JAVA_EXE, '-jar', JAR], env=env, cwd=r'e:\code\crm-backend-matt',
                     stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
                     creationflags=subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS)
    t0 = time.time()
    while time.time() - t0 < wait:
        time.sleep(5)
        try:
            r = requests.post(f'{BASE}/api/customer/workspace/pool/page',
                              data={'current': 1, 'size': 1}, timeout=5,
                              headers={'Authorization': S.headers['Authorization']})
            if r.status_code == 200:
                print(f'  服务已起（{int(time.time() - t0)}s）', flush=True)
                return True
            if r.status_code == 401:
                S.headers['Authorization'] = f'Bearer {get_token(ADMIN)}'
        except Exception:
            print('.', end='', flush=True)
    raise RuntimeError('服务重启后未就绪')


def f15():
    print('== F-15 提醒规则族 ==')
    F = 'F-15'
    URL = '/api/rule/customer/reminder'
    RULE_ORIG = {'masterEnabled': 1, 'firstTriggerEnabled': 1, 'firstTriggerDays': 30,
                 'secondIntervalEnabled': 1, 'secondIntervalDays': 7}

    # 15.1 GET 单例回显
    r = api(S, 'GET', URL, step='current')
    d = data_of(r) or {}
    ok = code_of(r) == 0 and d.get('masterEnabled') is not None
    check(F, 'GET 单例可读（记录当前值，不预设——重跑时规则可能留测试态）', 'pass' if ok else 'fail',
          f"master={d.get('masterEnabled')} first={d.get('firstTriggerEnabled')}/{d.get('firstTriggerDays')} "
          f"second={d.get('secondIntervalEnabled')}/{d.get('secondIntervalDays')}")

    # 15.2 PUT 全量覆盖（测试态：first=1/second=1 全开）
    r = api(S, 'PUT', URL, json_body={'masterEnabled': 1, 'firstTriggerEnabled': 1,
                                      'firstTriggerDays': 1, 'secondIntervalEnabled': 1,
                                      'secondIntervalDays': 1}, step='PUT 测试态')
    d = data_of(r) or {}
    ok = code_of(r) == 0 and str(d.get('firstTriggerDays')) == '1'
    check(F, 'PUT 全量覆盖保存即生效', 'pass' if ok else 'fail', f"first={d.get('firstTriggerDays')}")

    # 15.3/15.4 64023：天数非正整数（校验失败不落库）
    for case, days2 in (('firstTriggerDays=0 → 64023', 0), ('secondIntervalDays=-1 → 64023', -1)):
        body = {'masterEnabled': 1, 'firstTriggerEnabled': 1, 'firstTriggerDays': 1,
                'secondIntervalEnabled': 1, 'secondIntervalDays': 1}
        if 'first' in case:
            body['firstTriggerDays'] = days2
        else:
            body['secondIntervalDays'] = days2
        r = api(S, 'PUT', URL, json_body=body, step=case)
        expect_code(r, 64023, F, case)
    d = data_of(api(S, 'GET', URL, step='复查')) or {}
    ok = str(d.get('firstTriggerDays')) == '1' and str(d.get('secondIntervalDays')) == '1'
    check(F, '64023 校验失败未落库（1/1 保持）', 'pass' if ok else 'fail',
          f"first={d.get('firstTriggerDays')} second={d.get('secondIntervalDays')}")

    # 15.5 开关缺省=0（orDisabled：全量覆盖不静默保留旧值）
    r = api(S, 'PUT', URL, json_body={'firstTriggerDays': 2, 'secondIntervalDays': 2}, step='PUT 缺开关')
    d = data_of(r) or {}
    ok = code_of(r) == 0 and str(d.get('masterEnabled')) == '0' \
        and str(d.get('firstTriggerEnabled')) == '0' and str(d.get('secondIntervalEnabled')) == '0'
    check(F, '开关缺省 → 0（缺省禁用口径）', 'pass' if ok else 'fail',
          f"master={d.get('masterEnabled')} first={d.get('firstTriggerEnabled')} "
          f"second={d.get('secondIntervalEnabled')}")

    # 15.6 恢复测试态（Job 观测前提）
    api(S, 'PUT', URL, json_body={'masterEnabled': 1, 'firstTriggerEnabled': 1,
                                  'firstTriggerDays': 1, 'secondIntervalEnabled': 1,
                                  'secondIntervalDays': 1}, step='恢复测试态')

    # 15.7 Job 观测（U-02 拍板：重启注入 cron 每 30s 实测；对称商机 U03 仅 SKIP 的升级做法）
    print('  -- 重启服务带 CRM_CUSTOMER_REMINDERCRON=*/30 * * * * ? --')
    restart_service({'CRM_CUSTOMER_REMINDERCRON': '*/30 * * * * ?'})
    time.sleep(70)  # 等至少两轮 Job
    old_anchor = dbq("SELECT last_valid_follow_time a FROM customer WHERE id=%s", (C['followOld'],))[0]['a']
    plan_expect = (old_anchor + datetime.timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S')
    notices = dbq("SELECT * FROM customer_pending_notice WHERE customer_id=%s", (C['followOld'],))
    n = notices[0] if notices else {}
    ok = (len(notices) == 1 and str(n.get('notice_status')) == '0'
          and 'FOLLOW' in str(n.get('notice_type'))
          and str(n.get('target_user_id')) == ADMIN
          and str(n.get('plan_notify_time')).startswith(plan_expect[:10]))
    check(F, f'Job 落待发：超期样例 1 条 plan={plan_expect[:10]}（锚点+first=1）',
          'pass' if ok else 'fail',
          f"status={n.get('notice_status')} type={n.get('notice_type')} "
          f"target={n.get('target_user_id')} plan={n.get('plan_notify_time')}")
    payload = json.loads(n.get('payload_json') or '{}') if n else {}
    ok = payload.get('remindSeq') == 0 and 'anchor' in payload and payload.get('firstTriggerDays') == 1
    check(F, 'payload 快照（remindSeq/anchor/规则参数）', 'pass' if ok else 'warn',
          str(payload)[:100])

    # 15.8 未到期不落 + 公海/归档不参与
    f2 = dbq("SELECT COUNT(*) c FROM customer_pending_notice WHERE customer_id=%s", (C['followFuture'],))
    pool_ids = ','.join(str(C[k]) for k in ('pool1', 'pool2') if k in C)
    arc = dbq("SELECT COUNT(*) c FROM customer_pending_notice WHERE customer_id=%s", (C['archived'],))
    ok = f2[0]['c'] == 0 and arc[0]['c'] == 0
    check(F, '未到期/公海/归档客户均未落待发', 'pass' if ok else 'fail',
          f"待办={f2[0]['c']} 公海+归档={arc[0]['c']}")

    # 15.9 幂等：第二轮 Job 同 plan 不重写
    time.sleep(70)
    n2 = dbq("SELECT COUNT(*) c FROM customer_pending_notice WHERE customer_id=%s", (C['followOld'],))
    check(F, '幂等：同计划时刻任意状态占位不重写', 'pass' if n2[0]['c'] == 1 else 'fail',
          f'条数 {n2[0]["c"]}')

    # 15.10 锚点刷新 → 旧锚点未发提醒置失效
    r = api(S, 'POST', f"/api/customer/{C['followOld']}/follow",
            form={'followWay': FOLLOW_WAY, 'followContent': f'票05 锚点刷新 {TS}'},
            step='follow 超期客户')
    if code_of(r) == 0:
        time.sleep(70)
        rows = dbq("SELECT notice_status FROM customer_pending_notice WHERE customer_id=%s", (C['followOld'],))
        invalidated = all(str(r0['notice_status']) != '0' for r0 in rows)
        check(F, '锚点被跟进刷新 → 旧未发提醒置已失效', 'pass' if invalidated else 'fail',
              f'status={[str(r0["notice_status"]) for r0 in rows]}')
    else:
        check(F, '锚点被跟进刷新 → 旧未发提醒置已失效', 'fail', f'follow 失败 {r}')

    # 15.11 还原：规则回原值 + 服务还原默认 cron
    api(S, 'PUT', URL, json_body=RULE_ORIG, step='规则还原')
    d = data_of(api(S, 'GET', URL, step='规则复查')) or {}
    ok = str(d.get('firstTriggerDays')) == '30' and str(d.get('secondIntervalDays')) == '7'
    check(F, '提醒规则还原出厂值（30/7）', 'pass' if ok else 'fail', f'{d}')
    print('  -- 还原服务（默认 cron 凌晨 4 点） --')
    restart_service()
    d = data_of(api(S, 'POST', '/api/customer/workspace/pool/page',
                    form={'current': 1, 'size': 1}, step='smoke')) or {}
    check(F, '服务还原后 smoke', 'pass' if d else 'fail')


def f16():
    print('== F-16 查重设置消费侧（出厂默认：单例未落地，DefaultCustomerDedupSettingProvider 兜底） ==')
    F = 'F-16'
    simA_name = dbq("SELECT customer_name n FROM customer WHERE id=%s",
                    (IDS['customers']['simA']['id'],))[0]['n']
    simA_code = dbq("SELECT unified_credit_code c FROM customer WHERE id=%s",
                    (IDS['customers']['simA']['id'],))[0]['c']

    # 16.1 信用代码硬拦（出厂 creditCodeEnabled=true：开关关也拦——单例缺省无法验关闭态）
    r = api(S, 'GET', '/api/customer/check-credit-code', params={'creditCode': simA_code}, step='check-credit')
    hit = data_of(r)
    check(F, 'check-credit-code 命中已有信用代码', 'pass' if hit else 'fail', str(hit)[:80])
    form = {'customerName': f'{PFX}查重-撞码-{TS}', 'customerType': CTYPE,
            'unifiedCreditCode': simA_code, 'provinceCode': '440000',
            'cityCode': '440100', 'districtCode': '440103', 'industryCode': GOV,
            'customerStarLevel': 3, 'relationStarLevel': 3,
            'isBizNegotiated': '0', 'isChild': '0', 'ownerUserId': ADMIN}
    r = api(S, 'POST', '/api/customer', form=form, step='创建撞码')
    expect_code(r, 67003, F, '创建撞已有信用代码 → 67003 硬拦')

    # 16.2 名称相似（出厂 nameSimilarEnabled=true + threshold=80）
    d = data_of(api(S, 'GET', '/api/customer/check-name', params={'name': simA_name}, step='check-name'))
    check(F, 'check-name 相似命中（出厂阈值 80 生效）', 'pass' if d else 'fail', str(d)[:80])
    form = {'customerName': simA_name, 'customerType': CTYPE, 'provinceCode': '440000',
            'cityCode': '440100', 'districtCode': '440103', 'industryCode': GOV,
            'customerStarLevel': 3, 'relationStarLevel': 3,
            'isBizNegotiated': '0', 'isChild': '0', 'ownerUserId': ADMIN}
    r = api(S, 'POST', '/api/customer', form=form, step='创建相似名')
    d = data_of(r) or {}
    ok = code_of(r) == 0 and d.get('needConfirm') is True
    check(F, '创建名称相似 → needConfirm 弹窗未落库', 'pass' if ok else 'fail',
          f'needConfirm={d.get("needConfirm")}')

    # 16.3 电话软提示（出厂 phoneSoftCheckEnabled=true；D26 关闭路径依赖单例=票 09，观察点）
    d = data_of(api(S, 'GET', '/api/customer/contact/check-phone',
                    params={'phone': '13812340001'}, step='check-phone'))
    check(F, 'check-phone 跨客户软提示命中', 'pass' if d else 'fail', str(d)[:80])

    # 16.4 导入消费点（f14 已实测：UPSERT 同名→SUSPECT 不写入）
    check(F, '导入侧消费：名称相似 → SUSPECT 不写入（f14 14.9 实测引用）', 'pass', '见 F-14')

    # 16.5 结论：单例未落地
    check(F, 'customer_dedup_rule 单例未落地（票 09 范围）→ 出厂默认全开生效',
          'pass', 'DefaultCustomerDedupSettingProvider: 名称相似开/80/信用码开/电话软提示开')


# ==================== main ====================

def main():
    print(f'==== e2e-heavy 开始 TS={TS} ====')
    flows = []
    try:
        setup_heavy()
        flows.append(('F-13', f13))
        flows.append(('F-14', f14))
        flows.append(('F-15', f15))
        flows.append(('F-16', f16))
        for name, fn in flows:
            print(f'\n—— {name} ——')
            try:
                fn()
            except Exception as e:
                import traceback
                traceback.print_exc()
                check(name, f'流程异常中断: {e}', 'fail')
        f14_cleanup()
        teardown_heavy()
    except Exception as e:
        import traceback
        traceback.print_exc()
        check('MAIN', f'骨架异常: {e}', 'fail')
        try:
            teardown_heavy()
        except Exception as e2:
            print(f'  teardown 也失败: {e2}')

    # 统计与落盘
    n_pass = sum(1 for c in checks if c['verdict'] == '✅')
    n_warn = sum(1 for c in checks if c['verdict'] == '⚠')
    n_fail = sum(1 for c in checks if c['verdict'] == '❌')
    print(f'\n==== 结果：✅{n_pass} ⚠{n_warn} ❌{n_fail} ====')
    for d in defects:
        print(f"  缺陷 {d['id']}({d['severity']}): {d['title']}")
    out = '.scratch/customer-e2e/'
    json.dump(checks, open(out + 'e2e-heavy-checks.json', 'w', encoding='utf-8'),
              ensure_ascii=False, indent=1)
    json.dump(specimens, open(out + 'specimens-heavy.json', 'w', encoding='utf-8'),
              ensure_ascii=False, indent=1, default=str)
    json.dump(defects, open(out + 'e2e-heavy-defects.json', 'w', encoding='utf-8'),
              ensure_ascii=False, indent=1)
    print(f'  checks/specimens/defects 已落盘（specimens {len(specimens)} 端点）')


if __name__ == '__main__':
    main()