#!/usr/bin/env python3 # -*- coding: utf-8 -*- """P1 后端端到端实测（真实 HTTP）+ 跑完彻底清理。 验证： ① channel 必填门禁（无 channel → 400 且报明原因） ② order_refs 服务端按 is_test 分类（发货单/测试单） ③ 生成≠落库：confirmed=false 的条目必须被跳过 ④ 幂等：同一 draft_id 二次提交 → created=0 / updated=N，不新增 ⑤ 绝不抹掉人的动作：命中已有条目时只更新文本，done/conclusion 原样保留 ⑥ 拜访视图计数与提交结果一致 自带 finally 清理：活动文件 + 索引 + 待办（本地与云端）+ 云端活动文档。 """ import io import json import os import subprocess import sys import urllib.request ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' BASE = 'http://127.0.0.1:8765' DRAFT = 'TESTP1DRY' OWNER = '刘新元' CLI = '/Users/liuxinyuan/.npm-global/lib/node_modules/@cloudbase/cli/dist/standalone/cli.js' ENV = 'bier-sales-d0gatbvlx288724e9' sess = json.load(open(ROOT + '/数据/auth_sessions.json', encoding='utf-8')) mine = [(v['expire_at'], k, v) for k, v in sess.items() if v.get('name') == OWNER] mine.sort(reverse=True, key=lambda x: x[0]) TOKEN = mine[0][1] print('用会话：%s / role=%s\n' % (mine[0][2].get('name'), mine[0][2].get('role'))) idx = json.load(open(ROOT + '/数据/orders_2026_index.json', encoding='utf-8'))['orders'] ship = [o for o in idx if not o.get('is_test') and o.get('customer')][0] test = [o for o in idx if o.get('is_test') and o.get('customer')][0] print('测试用订单：发货单 %s (%s) / 测试单 %s (%s)\n' % (ship['no'], ship['customer'], test['no'], test['customer'])) ok = True def chk(label, cond, detail=''): global ok print(' %s %-38s %s' % ('✅' if cond else '❌', label, detail)) if not cond: ok = False def call(path, body=None, method=None): req = urllib.request.Request(BASE + path, method=method or ('POST' if body is not None else 'GET')) req.add_header('Authorization', 'Bearer ' + TOKEN) data = None if body is not None: data = json.dumps(body).encode('utf-8') req.add_header('Content-Type', 'application/json') try: with urllib.request.urlopen(req, data, timeout=60) as r: return r.status, json.load(r) except urllib.error.HTTPError as e: try: return e.code, json.load(e) except Exception: return e.code, {'raw': e.read()[:200].decode('utf-8', 'ignore')} def plan_row(uid, item, owner=OWNER, due='2026-11-15', tech=False): return {'uid': uid, 'item': item, 'owner': owner, 'due': due, 'tech_support': tech, 'confirmed': True} def payload(plan_text='回传 LOD 复测数据', confirmed_issue=True): return { 'type': '项目跟进', 'channel': 'offline', 'draft_id': DRAFT, 'customer': ship['customer'], 'contact': '', 'date': '2026-10-07', 'order_refs': [ship['no'], test['no'], 'NOSUCHORDER123'], 'opportunities': [], 'blocks': [{'opp': '', 'need': 'LOD 不稳定，需复测', 'params': [{'sku': 'FMD5104', 'desc': 'LOD 复测'}], 'issue': 'LOD 与竞品差距 3 倍', 'plan': [{'item': plan_text, 'owner': OWNER, 'due': '2026-11-15'}, {'item': '输出对比报告', 'owner': OWNER, 'due': '2026-11-20'}]}], 'todos': [ plan_row('r-1', plan_text), plan_row('r-2', '输出对比报告', tech=True), {'uid': 'r-3', 'item': '问题现象派生条', 'owner': OWNER, 'due': '2026-11-10', 'type': 'internal_complaint', 'confirmed': confirmed_issue}, {'uid': 'r-4', 'item': '未确认不该落库', 'owner': OWNER, 'due': '2026-11-01', 'confirmed': False}, ], } created_act_ids = [] store_before = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) n_before = sum(len(v) for v in store_before['users'].values() if isinstance(v, list)) print('提交前本地待办 %d 条\n' % n_before) try: # ① 必填门禁 print('① channel 必填门禁') p = payload(); p.pop('channel') st, r = call('/api/activity/submit', p) chk('无 channel → 400', st == 400 and not r.get('ok'), '%s %s' % (st, r.get('error', '')[:50])) # ②③ 正常提交 print('\n② order_refs 服务端分类 + ③ confirmed 门禁') st, r = call('/api/activity/submit', payload()) chk('提交成功', st == 200 and r.get('ok'), '%s %s' % (st, r.get('error', ''))) if not r.get('ok'): raise SystemExit('提交失败，终止') aid = r['id'] created_act_ids.append(aid) refs = {x['no']: x['kind'] for x in r.get('order_refs') or []} chk('发货单判为 ship', refs.get(ship['no']) == 'ship', '%s → %s' % (ship['no'], refs.get(ship['no']))) chk('测试单判为 test', refs.get(test['no']) == 'test', '%s → %s' % (test['no'], refs.get(test['no']))) chk('不存在的订单被剔除并回报', r.get('unmatched_orders') == ['NOSUCHORDER123'], str(r.get('unmatched_orders'))) td = r.get('todos') or {} chk('落库 3 条（第 4 条 confirmed=false 被跳过）', td.get('created') == 3 and td.get('skipped') == 1, json.dumps(td, ensure_ascii=False)) rec = json.load(open('%s/数据/activities/%s.json' % (ROOT, aid), encoding='utf-8')) chk('活动落 channel', rec.get('channel') == 'offline', rec.get('channel')) chk('索引带 channel', any(x.get('channel') == 'offline' for x in json.load(open(ROOT + '/数据/activities/_index.json', encoding='utf-8')) if x.get('id') == aid)) store = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) mine_td = [t for t in store['users'][OWNER] if (t.get('source') or {}).get('draft_id') == DRAFT] chk('本地待办 +3 且带 source', len(mine_td) == 3 and all(t['source']['kind'] == 'activity' for t in mine_td), '%d 条' % len(mine_td)) chk('类型/scope 正确', {t['type'] for t in mine_td} == {'external_project', 'internal_complaint'}, str(sorted(t['type'] for t in mine_td))) chk('技术协同标记落库', any(t.get('tech_support') for t in mine_td)) # ⑤ 人的动作不被抹掉 print('\n⑤ 已勾选完成的条目不被人重提交抹掉') doc = [t for t in mine_td if t['text'] == '输出对比报告'][0] doc['done'] = True doc['conclusion'] = '已回传，客户确认' doc['closed'] = '2026-10-07' io.open(ROOT + '/数据/todos.json', 'w', encoding='utf-8').write( json.dumps(store, ensure_ascii=False, indent=1)) # ④ 幂等 + ⑤ 保留动作 st, r2 = call('/api/activity/submit', payload(plan_text='回传 LOD 复测数据（含原始数据）')) created_act_ids.append(r2['id']) td2 = r2.get('todos') or {} chk('二次提交不新增', td2.get('created') == 0 and td2.get('updated') == 3, json.dumps(td2, ensure_ascii=False)) store2 = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) mine2 = [t for t in store2['users'][OWNER] if (t.get('source') or {}).get('draft_id') == DRAFT] chk('条数仍为 3（未膨胀）', len(mine2) == 3, '%d 条' % len(mine2)) d2 = [t for t in mine2 if t.get('conclusion') == '已回传，客户确认'] chk('done/conclusion 原样保留', len(d2) == 1 and d2[0]['done'] is True, json.dumps({k: d2[0].get(k) for k in ('done', 'conclusion', 'closed')}, ensure_ascii=False) if d2 else '丢了') chk('文本按新值更新', any('含原始数据' in t['text'] for t in mine2)) # ⑥ 拜访视图 print('\n⑥ 拜访视图统计') st, v = call('/api/visit/stats?month=2026-10') chk('接口可用', st == 200 and v.get('ok'), '%s' % st) chk('线下拜访次数计入 2', v.get('offline_count') == 2, json.dumps({k: v.get(k) for k in ('offline_count', 'online_count', 'total')})) chk('客户下钻到活动', any(c['customer'] == ship['customer'] and len(c['acts']) == 2 for c in v.get('customers') or []), str([(c['customer'], c['total']) for c in v.get('customers') or []])) st, v2 = call('/api/visit/stats?month=2020-01') chk('空月份返回 0', v2.get('offline_count') == 0 and v2.get('online_count') == 0) finally: print('\n=== 清理测试数据 ===') # 活动文件 + 索引 for aid in created_act_ids: p = '%s/数据/activities/%s.json' % (ROOT, aid) if os.path.exists(p): os.remove(p) ix = ROOT + '/数据/activities/_index.json' idx0 = json.load(open(ix, encoding='utf-8')) idx1 = [x for x in idx0 if x.get('id') not in created_act_ids] io.open(ix, 'w', encoding='utf-8').write(json.dumps(idx1, ensure_ascii=False, indent=1)) print(' 活动文件与索引已清（%d → %d）' % (len(idx0), len(idx1))) # 待办：去掉本次 draft 产生的条目（本地 + 云端） st3 = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) removed = 0 for u, lst in st3['users'].items(): if not isinstance(lst, list): continue keep = [t for t in lst if (t.get('source') or {}).get('draft_id') != DRAFT] removed += len(lst) - len(keep) st3['users'][u] = keep io.open(ROOT + '/数据/todos.json', 'w', encoding='utf-8').write( json.dumps(st3, ensure_ascii=False, indent=1)) n_after = sum(len(v) for v in st3['users'].values() if isinstance(v, list)) print(' 本地待办 %d → %d（清掉 %d）' % (n_before + 3 if False else n_after + removed, n_after, removed)) # 云端同步 cmd = json.dumps({'update': 'todos', 'updates': [{'q': {'user': OWNER}, 'u': {'$set': {'todos': st3['users'].get(OWNER, [])}}}]}, ensure_ascii=False) subprocess.run(['node', CLI, 'db', 'nosql', 'execute', '--command', json.dumps([{'TableName': 'todos', 'CommandType': 'UPDATE', 'Command': cmd}], ensure_ascii=False), '--json', '-e', ENV], capture_output=True, text=True, timeout=120) # 云端活动文档 for aid in created_act_ids: dc = json.dumps({'delete': 'activities', 'deletes': [{'q': {'id': aid}, 'limit': 0}]}, ensure_ascii=False) subprocess.run(['node', CLI, 'db', 'nosql', 'execute', '--command', json.dumps([{'TableName': 'activities', 'CommandType': 'DELETE', 'Command': dc}], ensure_ascii=False), '--json', '-e', ENV], capture_output=True, text=True, timeout=120) print(' 云端待办与活动文档已清') # 复核 st4 = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) left = [t for u, l in st4['users'].items() if isinstance(l, list) for t in l if (t.get('source') or {}).get('draft_id') == DRAFT] print(' 残留测试待办 %d 条 → %s' % (len(left), '✅ 干净' if not left else '❌ 有残留')) print(' 活动文件残留 %d 个 → %s' % (len([f for f in os.listdir(ROOT + '/数据/activities') if f.startswith('ACT')]), '✅ 干净' if not [f for f in os.listdir(ROOT + '/数据/activities') if f.startswith('ACT')] else '❌')) print('\n%s' % ('✅ P1 后端全部通过' if ok else '❌ 存在失败项')) sys.exit(0 if ok else 1)