You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
332 lines
17 KiB
332 lines
17 KiB
3 days ago
|
# -*- coding: utf-8 -*-
|
||
|
|
"""seed-customer.py — 客户模块 E2E 联调数据种子（票 03）
|
||
|
|
|
||
|
|
前缀约定：
|
||
|
|
- 客户/商机/视图名统一 `e2c-` 前缀（客户域专属命名空间，**不撞商机域 seed 的 `e2e-`**）
|
||
|
|
- 幂等 = 先物理清 e2c- 命名空间再重建；他人数据与 KH20260903xxx 冒烟残留不动
|
||
|
|
|
||
|
|
造数通道：API 优先（POST /api/customer 等），API 不可达的行 DB 直插/直拨（stage 2/3、
|
||
|
|
owner 换绑、过期 follow、超期锚点）——每处直插在 seed-ids.json 登记。
|
||
|
|
|
||
|
|
守卫：customer < 3 行或 e2c- 命中 > 60 行 → 疑似连错库，中止。
|
||
|
|
用法：py -X utf8 .scratch/customer-e2e/seed-customer.py [--keep] # --keep 跳过清理只补造
|
||
|
|
"""
|
||
|
|
import sys, io, json, time
|
||
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
|
import requests
|
||
|
|
import pymysql
|
||
|
|
|
||
|
|
BASE = 'http://localhost:8080'
|
||
|
|
ADMIN = '739564171091247104' # 罗伟健 ROLE_ADMIN dept=744841292348915712
|
||
|
|
BUDDY = '744842318024015872' # 曾偲青（职员部 744841292483133440，无角色——作协同/成员对象）
|
||
|
|
BUDDY_NAME, BUDDY_DEPT, BUDDY_DEPT_NAME = '曾偲青', '744841292483133440', '职员'
|
||
|
|
PFX = 'e2c-'
|
||
|
|
IDS_FILE = '.scratch/customer-e2e/seed-ids.json'
|
||
|
|
|
||
|
|
CONF = dict(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)
|
||
|
|
|
||
|
|
ids = {'admin': ADMIN, 'buddy': BUDDY, 'customers': {}, 'contacts': {},
|
||
|
|
'opportunities': {}, 'follows': {}, 'dbPatches': [], 'notes': []}
|
||
|
|
|
||
|
|
|
||
|
|
def db():
|
||
|
|
return pymysql.connect(**CONF)
|
||
|
|
|
||
|
|
|
||
|
|
def dbq(sql, args=None, fetch=True):
|
||
|
|
with db() as conn, conn.cursor() as cur:
|
||
|
|
cur.execute(sql, args)
|
||
|
|
return cur.fetchall() if fetch else cur.rowcount
|
||
|
|
|
||
|
|
|
||
|
|
def get_token(uid):
|
||
|
|
r = requests.get(f'{BASE}/api/auth/debug/token', params={'userId': uid}, timeout=30)
|
||
|
|
r.raise_for_status()
|
||
|
|
d = r.json().get('data')
|
||
|
|
tok = d if isinstance(d, str) else (d or {}).get('token')
|
||
|
|
assert tok, f'debug token 异常: {r.text[:200]}'
|
||
|
|
return tok
|
||
|
|
|
||
|
|
|
||
|
|
S = requests.Session()
|
||
|
|
S.headers['Authorization'] = f'Bearer {get_token(ADMIN)}'
|
||
|
|
|
||
|
|
|
||
|
|
def api(method, path, form=None, json_body=None, params=None, step=''):
|
||
|
|
"""统一请求；业务失败返回 None 并打印（不中断，末尾汇总决定 exit code）。"""
|
||
|
|
url = BASE + path
|
||
|
|
try:
|
||
|
|
if json_body is not None:
|
||
|
|
r = S.post(url, json=json_body, params=params, timeout=30)
|
||
|
|
elif method == 'POST':
|
||
|
|
r = S.post(url, data=form, params=params, timeout=30)
|
||
|
|
elif method == 'PUT':
|
||
|
|
r = S.put(url, data=form, params=params, timeout=30)
|
||
|
|
else:
|
||
|
|
r = S.request(method, url, params=params, timeout=30)
|
||
|
|
except Exception as e:
|
||
|
|
print(f' ✘ {step}: 网络异常 {e}')
|
||
|
|
return None
|
||
|
|
if r.status_code != 200:
|
||
|
|
print(f' ✘ {step}: HTTP {r.status_code} {r.text[:160]}')
|
||
|
|
return None
|
||
|
|
body = r.json()
|
||
|
|
code = body.get('code')
|
||
|
|
if code not in (0, '0', 200, '200'):
|
||
|
|
print(f' ✘ {step}: code={code} {str(body.get("message"))[:160]}')
|
||
|
|
return None
|
||
|
|
return body.get('data')
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------- 字典 code 动态取 ----------------
|
||
|
|
def dict_codes(group, limit=5):
|
||
|
|
rows = dbq(
|
||
|
|
"SELECT i.code FROM dict_item i JOIN dict_group g ON i.group_id=g.id "
|
||
|
|
"WHERE g.code=%s AND i.deleted=0 ORDER BY i.sort_no LIMIT %s", (group, limit))
|
||
|
|
assert rows, f'字典组 {group} 为空'
|
||
|
|
return [r['code'] for r in rows]
|
||
|
|
|
||
|
|
|
||
|
|
print('== 0. 字典准备 ==')
|
||
|
|
CTYPES = dict_codes('customer_type')
|
||
|
|
IND1 = dbq("SELECT i.code, i.id 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 NULL ORDER BY i.sort_no LIMIT 3")
|
||
|
|
IND_CHILD = dbq("SELECT i.code 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=%s ORDER BY i.sort_no LIMIT 1",
|
||
|
|
(IND1[0]['id'],))
|
||
|
|
FOLLOW_WAY = dict_codes('follow_way')
|
||
|
|
print(f' customer_type={CTYPES[:3]}... industry1={IND1[0]["code"]} child={IND_CHILD or "无"} follow_way={FOLLOW_WAY}')
|
||
|
|
|
||
|
|
# ---------------- 1. 清理 ----------------
|
||
|
|
if '--keep' not in sys.argv:
|
||
|
|
print('== 1. 清理 e2c- 命名空间 ==')
|
||
|
|
n_all = dbq('SELECT COUNT(*) c FROM customer')[0]['c']
|
||
|
|
assert n_all >= 3, f'customer 仅 {n_all} 行，疑似连错库，中止'
|
||
|
|
cust = dbq("SELECT id, customer_name FROM customer WHERE customer_name LIKE %s", (PFX + '%',))
|
||
|
|
assert len(cust) <= 60, f'e2c- 命中 {len(cust)} 行超守卫阈值 60，中止'
|
||
|
|
cids = [str(r['id']) for r in cust]
|
||
|
|
ph = ','.join(['%s'] * len(cids)) if cids else None
|
||
|
|
with db() as conn, conn.cursor() as cur:
|
||
|
|
if cids:
|
||
|
|
# 商机侧：先绑定行后商机（仅 e2c- 商机与绑定到 e2c 客户的行）
|
||
|
|
cur.execute(f"DELETE FROM opportunity_customer WHERE customer_id IN ({ph})", cids)
|
||
|
|
print(f' 清 opportunity_customer {cur.rowcount}')
|
||
|
|
cur.execute("DELETE FROM opportunity WHERE opp_name LIKE %s", (PFX + '%',))
|
||
|
|
print(f' 清 opportunity(e2c-) {cur.rowcount}')
|
||
|
|
# 客户从表
|
||
|
|
for t in ['customer_contact', 'customer_oplog', 'customer_team_member',
|
||
|
|
'customer_follow', 'customer_pending_notice']:
|
||
|
|
cur.execute(f'DELETE FROM {t} WHERE customer_id IN ({ph})', cids)
|
||
|
|
print(f' 清 {t} {cur.rowcount}')
|
||
|
|
for t in ['customer_focus', 'customer_view_log']:
|
||
|
|
cur.execute(f'DELETE FROM {t} WHERE customer_id IN ({ph})', cids)
|
||
|
|
print(f' 清 {t} {cur.rowcount}')
|
||
|
|
# 交割单（经 detail 反查 e2c 客户）
|
||
|
|
cur.execute(f'SELECT DISTINCT transfer_id FROM customer_transfer_detail WHERE customer_id IN ({ph})', cids)
|
||
|
|
tids = [r['transfer_id'] for r in cur.fetchall()]
|
||
|
|
if tids:
|
||
|
|
ph2 = ','.join(['%s'] * len(tids))
|
||
|
|
cur.execute(f'DELETE FROM customer_transfer_detail WHERE transfer_id IN ({ph2})', tids)
|
||
|
|
cur.execute(f'DELETE FROM customer_transfer WHERE id IN ({ph2})', tids)
|
||
|
|
print(f' 清 customer_transfer {len(tids)} 单')
|
||
|
|
cur.execute(f'DELETE FROM customer WHERE id IN ({ph})', cids)
|
||
|
|
print(f' 清 customer {cur.rowcount}')
|
||
|
|
cur.execute("DELETE FROM user_saved_view WHERE name LIKE %s AND user_id=%s", (PFX + '%', ADMIN))
|
||
|
|
print(f' 清 user_saved_view(e2c-) {cur.rowcount}')
|
||
|
|
print(' 清理完成')
|
||
|
|
else:
|
||
|
|
print('== 1. --keep 跳过清理 ==')
|
||
|
|
|
||
|
|
# ---------------- 2. 造客户（API） ----------------
|
||
|
|
print('== 2. 造客户 ==')
|
||
|
|
CITY = ('440000', '440100', '440103') # 广东 广州 荔湾（sys_region 3432 条国标码）
|
||
|
|
|
||
|
|
|
||
|
|
def make_customer(key, name, **extra):
|
||
|
|
form = dict(customerName=name, customerType=CTYPES[0], provinceCode=CITY[0],
|
||
|
|
cityCode=CITY[1], districtCode=CITY[2], industryCode=IND1[0]['code'],
|
||
|
|
customerStarLevel=3, relationStarLevel=3, isBizNegotiated=0, isChild=0,
|
||
|
|
ownerUserId=ADMIN, confirmSimilar='true', remark='票03 seed 造数样例（前端联调数据）')
|
||
|
|
if IND_CHILD:
|
||
|
|
form['industryChildCode'] = IND_CHILD[0]['code']
|
||
|
|
form.update(extra)
|
||
|
|
d = api('POST', '/api/customer', form=form, step=f'create {name}')
|
||
|
|
if d is None:
|
||
|
|
return None
|
||
|
|
if isinstance(d, dict) and d.get('needConfirm'):
|
||
|
|
print(f' ✘ {name}: 命中相似需确认（confirmSimilar 未生效）: {str(d.get("similarHits"))[:120]}')
|
||
|
|
return None
|
||
|
|
cid = str(d.get('id')) if isinstance(d, dict) else str(d)
|
||
|
|
no = (d.get('customerNo') if isinstance(d, dict) else None) or \
|
||
|
|
dbq('SELECT customer_no FROM customer WHERE id=%s', (cid,))[0]['customer_no']
|
||
|
|
print(f' ✔ {name} id={cid} no={no}')
|
||
|
|
ids['customers'][key] = {'id': cid, 'no': no, 'name': name}
|
||
|
|
return cid
|
||
|
|
|
||
|
|
|
||
|
|
make_customer('stage1', PFX + '阶段-潜在-甲')
|
||
|
|
make_customer('stage2', PFX + '阶段-重潜-乙', customerStarLevel=4)
|
||
|
|
make_customer('stage3', PFX + '阶段-成交-丙', customerStarLevel=5, relationStarLevel=4)
|
||
|
|
make_customer('star5', PFX + '星级-五星', customerStarLevel=5, relationStarLevel=5)
|
||
|
|
make_customer('full', PFX + '全字段-科技', unifiedCreditCode='91440101E2CTEST001',
|
||
|
|
legalRepresentative='王测试', establishedDate='2018-05-20',
|
||
|
|
registeredCapital='1000万元人民币', businessScope='软件开发；信息系统集成',
|
||
|
|
staffSize='50-99人', annualRevenue='5000万元', businessAddress='广州市荔湾区测试路1号',
|
||
|
|
companyDecisionMaker='王总', joinedPresidentClass=1, presidentClassPerson='王总',
|
||
|
|
presidentClassPhone='13800000001', networkUnits=json.dumps(['标杆客户', '同行推荐']),
|
||
|
|
cooperationSystem='OA对接', isBizNegotiated=1)
|
||
|
|
make_customer('simA', PFX + '恒信达科技有限公司', unifiedCreditCode='91440101E2CHXDA001')
|
||
|
|
make_customer('simB', PFX + '恒信达科技有限责任公司', unifiedCreditCode='91440101E2CHXDB001')
|
||
|
|
make_customer('followFuture', PFX + '跟进-待办')
|
||
|
|
make_customer('followOld', PFX + '提醒-超期')
|
||
|
|
make_customer('collab', PFX + '协同-样例')
|
||
|
|
make_customer('member', PFX + '成员-样例')
|
||
|
|
make_customer('transfer1', PFX + '交割-甲')
|
||
|
|
make_customer('transfer2', PFX + '交割-乙')
|
||
|
|
make_customer('transfer3', PFX + '交割-丙')
|
||
|
|
# 公海（不传 ownerUserId → enter_pool_time 记当下）
|
||
|
|
make_customer('pool1', PFX + '公海-甲', ownerUserId=None)
|
||
|
|
make_customer('pool2', PFX + '公海-乙', ownerUserId=None)
|
||
|
|
# 归档样例
|
||
|
|
arch = make_customer('archived', PFX + '已归档')
|
||
|
|
if arch:
|
||
|
|
ok = api('POST', '/api/customer/archive', params={'id': arch}, step='archive 已归档')
|
||
|
|
if ok is not None:
|
||
|
|
print(' ✔ archived 归档完成')
|
||
|
|
|
||
|
|
# ---------------- 3. DB 直拨（API 不可达行） ----------------
|
||
|
|
print('== 3. DB 直拨 ==')
|
||
|
|
C = ids['customers']
|
||
|
|
if C.get('stage2'):
|
||
|
|
dbq('UPDATE customer SET customer_stage=2 WHERE id=%s', (C['stage2']['id'],), fetch=False)
|
||
|
|
ids['dbPatches'].append('stage2 customer_stage=2')
|
||
|
|
if C.get('stage3'):
|
||
|
|
dbq('UPDATE customer SET customer_stage=3 WHERE id=%s', (C['stage3']['id'],), fetch=False)
|
||
|
|
ids['dbPatches'].append('stage3 customer_stage=3')
|
||
|
|
if C.get('collab'):
|
||
|
|
dbq('UPDATE customer SET owner_user_id=%s, owner_user_name_snapshot=%s, owner_dept_id=%s, '
|
||
|
|
'owner_dept_name_snapshot=%s WHERE id=%s',
|
||
|
|
(BUDDY, BUDDY_NAME, BUDDY_DEPT, BUDDY_DEPT_NAME, C['collab']['id']), fetch=False)
|
||
|
|
ok = api('POST', f"/api/customer/{C['collab']['id']}/members",
|
||
|
|
form={'memberUserIds': ADMIN}, step='members collab+admin')
|
||
|
|
ids['dbPatches'].append(f"collab owner→曾偲青 + admin 成员(ok={ok is not None})")
|
||
|
|
if C.get('followOld'):
|
||
|
|
dbq('UPDATE customer SET last_valid_follow_time=NOW()-INTERVAL 30 DAY WHERE id=%s',
|
||
|
|
(C['followOld']['id'],), fetch=False)
|
||
|
|
ids['dbPatches'].append('followOld last_valid_follow_time=30天前（Job 超期样例）')
|
||
|
|
|
||
|
|
# ---------------- 4. 跟进（API 未来 + DB 直插过期） ----------------
|
||
|
|
print('== 4. 跟进样例 ==')
|
||
|
|
if C.get('followFuture'):
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
# LocalDateTime form 绑定按 ISO_LOCAL_DATE_TIME（T 分隔），空格分隔会 400
|
||
|
|
nxt = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%dT%H:%M:%S')
|
||
|
|
ok = api('POST', f"/api/customer/{C['followFuture']['id']}/follow",
|
||
|
|
form={'followWay': FOLLOW_WAY[0], 'followContent': 'seed 待办样例：明天回访确认需求',
|
||
|
|
'nextFollowTime': nxt}, step='follow followFuture')
|
||
|
|
if ok is not None:
|
||
|
|
print(f' ✔ followFuture 待跟进（next={nxt}）')
|
||
|
|
if C.get('followOld'):
|
||
|
|
# 直插过期 follow（供 FOLLOW_UP_DUE 视图 EXISTS 命中）
|
||
|
|
with db() as conn, conn.cursor() as cur:
|
||
|
|
cur.execute('SELECT COUNT(*) c FROM customer_follow WHERE customer_id=%s',
|
||
|
|
(C['followOld']['id'],))
|
||
|
|
if cur.fetchone()['c'] == 0:
|
||
|
|
import time as _t
|
||
|
|
fid = int(_t.time() * 1000) * 1000 + 7
|
||
|
|
cur.execute(
|
||
|
|
'INSERT INTO customer_follow (id, customer_id, follow_way, follow_content, '
|
||
|
|
'next_follow_time, follow_by, follow_by_name, follow_dept_name, '
|
||
|
|
'creator_id, create_time, updater_id, update_time, deleted) VALUES '
|
||
|
|
'(%s,%s,%s,%s,NOW()-INTERVAL 2 DAY,%s,%s,%s,%s,NOW()-INTERVAL 7 DAY,%s,NOW()-INTERVAL 7 DAY,0)',
|
||
|
|
(fid, C['followOld']['id'], FOLLOW_WAY[0], 'seed 直插：已过期下次跟进（FOLLOW_UP_DUE 视图素材）',
|
||
|
|
ADMIN, '罗伟健', '广东保伦电子股份有限公司', ADMIN, ADMIN))
|
||
|
|
print(f' ✔ 直插过期 follow id={fid}')
|
||
|
|
ids['follows']['followOldExpired'] = str(fid)
|
||
|
|
ids['dbPatches'].append('followOld 直插过期 follow（next_follow_time=2天前）')
|
||
|
|
else:
|
||
|
|
print(' ⊘ followOld 已有 follow（--keep 重跑），跳过')
|
||
|
|
|
||
|
|
# ---------------- 5. 商机关联样例（API，商机域） ----------------
|
||
|
|
print('== 5. 商机关联样例 ==')
|
||
|
|
|
||
|
|
|
||
|
|
def make_opp(key, name, bind_key=None, role='customer_role_01', primary=1):
|
||
|
|
form = dict(opportunityName=name, oppSource='opp_source_02', industryCode=IND1[0]['code'],
|
||
|
|
localityType='locality_type_01', bidForm='bid_form_01',
|
||
|
|
provinceCode=CITY[0], cityCode=CITY[1], partyAClear=1,
|
||
|
|
partyA='e2c甲方·' + name, remark='票03 seed 商机关联样例')
|
||
|
|
d = api('POST', '/api/opportunity', form=form, step=f'opp {name}')
|
||
|
|
if d is None:
|
||
|
|
return None
|
||
|
|
oid = str(d)
|
||
|
|
print(f' ✔ {name} id={oid}')
|
||
|
|
ids['opportunities'][key] = {'id': oid, 'name': name}
|
||
|
|
if bind_key and C.get(bind_key):
|
||
|
|
d2 = api('POST', '/api/opportunity/customer/add',
|
||
|
|
form={'oppId': oid, 'customerId': C[bind_key]['id'],
|
||
|
|
'customerNameSnapshot': C[bind_key]['name'],
|
||
|
|
'customerRole': role, 'isPrimaryIntended': primary},
|
||
|
|
step=f'bind {name}←{C[bind_key]["name"]}')
|
||
|
|
if d2 is not None:
|
||
|
|
print(f' ✔ 绑定 {C[bind_key]["name"]}（primary={primary}）')
|
||
|
|
return oid
|
||
|
|
|
||
|
|
|
||
|
|
# 交割-乙 绑进行中商机 → F-11 抛公海 67007 拦截样例；交割-甲 绑 → F-13 owner 联动样例
|
||
|
|
make_opp('oppBlock', PFX + '联动-进行中', bind_key='transfer2')
|
||
|
|
make_opp('oppTransfer', PFX + '联动-交割', bind_key='transfer1')
|
||
|
|
|
||
|
|
# ---------------- 6. 联系人（API） ----------------
|
||
|
|
print('== 6. 联系人样例 ==')
|
||
|
|
|
||
|
|
|
||
|
|
def make_contact(key, cust_key, name, **extra):
|
||
|
|
if not C.get(cust_key):
|
||
|
|
return None
|
||
|
|
form = dict(customerId=C[cust_key]['id'], name=name, jobTitleName=extra.pop('jobTitleName', '经理'),
|
||
|
|
isKeyContact=0, isInternal=0)
|
||
|
|
form.update(extra)
|
||
|
|
d = api('POST', '/api/customer/contact', form=form, step=f'contact {name}')
|
||
|
|
if d is None:
|
||
|
|
return None
|
||
|
|
print(f' ✔ {name} → {C[cust_key]["name"]} id={d}')
|
||
|
|
ids['contacts'][key] = {'id': str(d), 'name': name, 'customerKey': cust_key}
|
||
|
|
return d
|
||
|
|
|
||
|
|
|
||
|
|
make_contact('key1', 'full', '张关键', phone='13812340001', isKeyContact=1, isInternal=1,
|
||
|
|
jobTitleName='采购总监', giftRemark='偏好茶叶', source='contact_source_01')
|
||
|
|
make_contact('normal1', 'full', '李普通', phone='13812340002', jobTitleName='工程师')
|
||
|
|
make_contact('dupPhone', 'full', '王重复', phone='13812340001', jobTitleName='副总') # check-phone 命中样例
|
||
|
|
make_contact('other1', 'simA', '赵外线', phone='13812340003', jobTitleName='法人')
|
||
|
|
|
||
|
|
# ---------------- 7. 库检验证 + 落盘 ----------------
|
||
|
|
print('== 7. 库检验证 ==')
|
||
|
|
chk = []
|
||
|
|
chk.append(('e2c- 客户总数', dbq("SELECT COUNT(*) c FROM customer WHERE customer_name LIKE %s", (PFX + '%',))[0]['c'], 17))
|
||
|
|
chk.append(('公海(owner空)', dbq("SELECT COUNT(*) c FROM customer WHERE customer_name LIKE %s AND owner_user_id IS NULL", (PFX + '%',))[0]['c'], 2))
|
||
|
|
chk.append(('归档', dbq("SELECT COUNT(*) c FROM customer WHERE customer_name LIKE %s AND archive_status=2", (PFX + '%',))[0]['c'], 1))
|
||
|
|
chk.append(('stage 分布', sorted(r['customer_stage'] for r in dbq(
|
||
|
|
"SELECT customer_stage FROM customer WHERE customer_name LIKE %s AND deleted=0", (PFX + '%',))), [1] * 15 + [2, 3]))
|
||
|
|
chk.append(('联系人', dbq("SELECT COUNT(*) c FROM customer_contact ct JOIN customer c ON c.id=ct.customer_id WHERE c.customer_name LIKE %s", (PFX + '%',))[0]['c'], 4))
|
||
|
|
chk.append(('e2c- 商机', dbq("SELECT COUNT(*) c FROM opportunity WHERE opp_name LIKE %s AND deleted=0", (PFX + '%',))[0]['c'], 2))
|
||
|
|
chk.append(('商机绑定', dbq("SELECT COUNT(*) c FROM opportunity_customer oc JOIN customer cu ON cu.id=oc.customer_id WHERE cu.customer_name LIKE %s", (PFX + '%',))[0]['c'], 2))
|
||
|
|
fails = 0
|
||
|
|
for label, got, want in chk:
|
||
|
|
mark = '✔' if got == want else '✘'
|
||
|
|
if got != want:
|
||
|
|
fails += 1
|
||
|
|
print(f' {mark} {label}: {got}（期望 {want}）')
|
||
|
|
|
||
|
|
with open(IDS_FILE, 'w', encoding='utf-8') as f:
|
||
|
|
json.dump(ids, f, ensure_ascii=False, indent=2)
|
||
|
|
print(f'\n== seed-ids.json 落盘 → {IDS_FILE} ==')
|
||
|
|
print('数据矩阵摘要：')
|
||
|
|
for k, v in C.items():
|
||
|
|
print(f" {k}: {v['name']} id={v['id']} no={v['no']}")
|
||
|
|
print(f'\n{"ALL GREEN" if fails == 0 else f"{fails} 项校验不符"}')
|
||
|
|
sys.exit(0 if fails == 0 else 1)
|