#!/usr/bin/env python3 # -*- coding: utf-8 -*- """P1 端到端实测：录入页真提交 → 待办落库 → 拜访视图计数 → 清理。 链路：活动录入页填表 → 生成待办预览 → 逐条确认 → 提交 → 后端 upsert 待办 → 工作台「拜访视图」Tab 计数与下钻 → 全部清理回原状。 """ import json import os import subprocess import sys from playwright.sync_api import sync_playwright ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' BASE = 'http://127.0.0.1:8765/' USER = '刘新元' 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 = sorted([(v['expire_at'], k, v) for k, v in sess.items() if v.get('name') == USER], reverse=True, key=lambda x: x[0]) TOKEN, IDENT = mine[0][1], dict(mine[0][2]) IDENT['open_id'] = 'ou_preview_only' IDENT.pop('expire_at', None) KEY = 'bior_sw_todos_v2_' + USER 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] CUST = ship['customer'] DRAFT = 'TESTP1UI' store0 = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) N0 = sum(len(v) for v in store0['users'].values() if isinstance(v, list)) print('提交前本地待办 %d 条\n' % N0) ok = True def chk(label, cond, detail=''): global ok print(' %s %-36s %s' % ('✅' if cond else '❌', label, detail)) if not cond: ok = False def boot(pg, page_path, extra=None): pg.goto(BASE, wait_until='domcontentloaded') pg.evaluate("""([tk, idn, k, extra]) => { localStorage.setItem('bior_auth_token', tk); localStorage.setItem('bior_sw_user_v2', JSON.stringify(idn)); if (k) localStorage.removeItem(k); if (extra) Object.keys(extra).forEach(function(x){ localStorage.setItem(x, extra[x]); }); }""", [TOKEN, IDENT, KEY, extra or {}]) pg.goto(BASE + page_path, wait_until='load') pg.wait_for_timeout(2600) act_ids = [] try: with sync_playwright() as p: br = p.chromium.launch(channel='chrome', headless=True) pg = br.new_page(viewport={'width': 1400, 'height': 1000}) errs = [] pg.on('pageerror', lambda e: errs.append(str(e))) pg.on('dialog', lambda d: d.accept()) boot(pg, '工具/活动录入.html') # ── ① 新控件就位 ── print('① 录入页新控件') m = pg.evaluate("""() => { const c = document.getElementById('channel'); return {ch: !!c, opts: c ? [...c.querySelectorAll('option')].map(o => o.value) : [], ordRef: !!document.getElementById('ordRefQ'), cardT: !!document.getElementById('cardTodo'), host: !!document.getElementById('todoConfirmHost'), draftId: (ST.draftId || '').slice(0, 12)}; }""") chk('活动形式下拉存在', m['ch'], str(m['opts'])) chk('下拉为 空/线下/线上', set(m['opts']) == {'', 'offline', 'online'}, str(m['opts'])) chk('关联订单输入框存在', m['ordRef']) chk('待办确认卡片存在', m['cardT'] and m['host']) chk('会话有稳定 draftId', bool(m['draftId']), m['draftId']) pg.evaluate("() => { ST.draftId = 'TESTP1UI'; }") # ── ② 未填计划 → 生成预览给提示，不产生草稿 ── pg.evaluate("() => genTodoDrafts(true)") pg.wait_for_timeout(300) n0 = pg.evaluate("() => (ST.todoDrafts||[]).filter(t=>!t.removed).length") chk('无计划时不生成草稿', n0 == 0, '%d 条' % n0) # ── ③ 填内容 → 生成预览 ── print('\n② 填表 → 生成待办预览') pg.evaluate("""([cust, item]) => { document.getElementById('customer').value = cust; document.getElementById('customer').dispatchEvent(new Event('input', {bubbles:true})); const blk = document.querySelector('.oppblock'); const ta = blk.querySelector('[data-f="need"]'); ta.value = '1. P1 端到端测试需求'; ta.dispatchEvent(new Event('input', {bubbles:true})); const iss = blk.querySelector('[data-f="issue"]'); iss.value = 'LOD 与竞品差距 3 倍'; iss.dispatchEvent(new Event('input', {bubbles:true})); addPlan(blk.getAttribute('data-k'), item, '刘新元', '2026-11-20'); }""", [CUST, 'P1 端到端测试待办']) pg.wait_for_timeout(300) pg.evaluate("() => genTodoDrafts(true)") pg.wait_for_timeout(500) d = pg.evaluate("""() => ({ rows: document.querySelectorAll('#todoConfirmHost tbody tr').length, drafts: (ST.todoDrafts||[]).map(t => ({uid:t.uid, type:t.type, from:t.from, conf:t.confirmed})), sum: (document.getElementById('todoSummary')||{}).textContent || '' })""") chk('生成 2 条草稿（计划 + 问题现象）', d['rows'] == 2 and len(d['drafts']) == 2, json.dumps(d['drafts'], ensure_ascii=False)) chk('问题现象派生为客诉跟进', any(x['type'] == 'internal_complaint' for x in d['drafts'])) chk('摘要提示未确认', '未确认' in d['sum'], d['sum']) # ── ③ 未确认 → collect() 不带 todos；validate 报缺活动形式 ── print('\n③ 确认门禁 + 必填门禁') c1 = pg.evaluate("() => collect().todos.length") chk('未确认 → collect().todos 为空', c1 == 0, '%d 条' % c1) v1 = pg.evaluate("""() => { const okv = validate(); return {ok: okv, msg: (document.getElementById('barMsg')||{}).textContent || ''}; }""") chk('缺活动形式 → 校验不通过', (not v1['ok']) and ('活动形式' in v1['msg']), v1['msg'][:70]) pg.evaluate("() => { document.getElementById('channel').value = 'offline'; }") pg.evaluate("""() => { (ST.todoDrafts||[]).forEach(t => { if (!t.removed) tdConfirm(t.uid, true); }); }""") pg.wait_for_timeout(400) c2 = pg.evaluate("() => collect()") chk('确认后 collect().todos=2', len(c2['todos']) == 2, '%d 条' % len(c2['todos'])) chk('collect 带 channel/order_refs/draft_id', c2['channel'] == 'offline' and isinstance(c2['order_refs'], list) and c2['draft_id'] == DRAFT, 'channel=%s draft_id=%s' % (c2['channel'], c2['draft_id'])) v2 = pg.evaluate("() => validate()") chk('信息齐备 → 校验通过', v2 is True) chk('无 JS 错误', not errs, str(errs[:2])) # ── ④ 真提交 ── print('\n④ 真提交（走页面 submitForm）') pg.evaluate("() => submitForm()") pg.wait_for_timeout(6000) res = pg.evaluate("""() => { const t = (document.getElementById('modal')||{}).innerText || ''; return {shown: t.indexOf('提交成功') >= 0, text: t.replace(/\\s+/g,' ').slice(0, 260)}; }""") chk('提交成功并弹结果', res['shown'], res['text'][:120]) chk('结果含待办落库行', '待办落库' in res['text'], res['text'][-90:]) st1 = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) mine_td = [t for t in st1['users'][USER] if (t.get('source') or {}).get('draft_id') == DRAFT] chk('本地待办 +2 且带 source', len(mine_td) == 2, '%d 条' % len(mine_td)) acts = [f for f in os.listdir(ROOT + '/数据/activities') if f.startswith('ACT')] chk('活动已落盘', len(acts) == 1, str(acts)) aid = acts[0].replace('.json', '') if acts else '' act_ids.append(aid) if aid: rec = json.load(open('%s/数据/activities/%s.json' % (ROOT, aid), encoding='utf-8')) chk('活动 channel=offline', rec.get('channel') == 'offline') # ── ⑤ 拜访视图 ── print('\n⑤ 拜访视图 Tab（两个入口）') boot(pg, '工具/index.html') pg.evaluate("""() => { window.state = window.state || {}; window.state.user = JSON.parse(JSON.stringify(%s)); if (typeof showApp === 'function') showApp(); }""" % json.dumps(IDENT, ensure_ascii=False)) pg.wait_for_timeout(1200) for entry in ['工具/index.html', '工具/app.html']: if entry.endswith('app.html'): pg.goto(BASE + entry, wait_until='load'); pg.wait_for_timeout(2200) pg.evaluate("""() => { window.state = window.state || {}; window.state.user = JSON.parse(JSON.stringify(%s)); if (typeof showApp === 'function') showApp(); }""" % json.dumps(IDENT, ensure_ascii=False)) pg.wait_for_timeout(900) has = pg.evaluate("() => (window.TABS||[]).some(t => t.id === 'visitStats')") chk('%s 有 visitStats Tab' % entry.split('/')[-1], has) pg.evaluate("() => { state.tab='visitStats'; render(); }") pg.wait_for_timeout(2500) v = pg.evaluate("""() => { const T = e => (e ? e.innerText.replace(/\\s+/g,' ').trim() : ''); const b = document.getElementById('visitBody'); return {body: T(b).slice(0, 150), vals: [...document.querySelectorAll('#visitBody .value')].map(T)}; }""") chk(' 线下拜访次数=1', v['vals'][:2] == ['1', '0'], str(v['vals'])) if entry.endswith('index.html'): # 下钻：展开客户 + 看详情 pg.evaluate("() => visitToggle(0)") pg.wait_for_timeout(400) pg.evaluate("() => { const a = document.querySelector('#visitActs0 a'); if (a) a.click(); }") pg.wait_for_timeout(2500) det = pg.evaluate("() => { const p = document.querySelector('#visitActs0 pre'); return p ? p.innerText.replace(/\\s+/g,' ').slice(0,90) : ''; }") chk(' 客户下钻 → 活动详情', len(det) > 20, det[:70]) br.close() finally: print('\n=== 清理测试数据 ===') for aid in act_ids: p = '%s/数据/activities/%s.json' % (ROOT, aid) if os.path.exists(p): os.remove(p) ix = ROOT + '/数据/activities/_index.json' i0 = json.load(open(ix, encoding='utf-8')) i1 = [x for x in i0 if x.get('id') not in act_ids] open(ix, 'w', encoding='utf-8').write(json.dumps(i1, ensure_ascii=False, indent=1)) st = json.load(open(ROOT + '/数据/todos.json', encoding='utf-8')) rm = 0 for u, l in st['users'].items(): if not isinstance(l, list): continue k = [t for t in l if (t.get('source') or {}).get('draft_id') != DRAFT] rm += len(l) - len(k) st['users'][u] = k open(ROOT + '/数据/todos.json', 'w', encoding='utf-8').write(json.dumps(st, ensure_ascii=False, indent=1)) cmd = json.dumps({'update': 'todos', 'updates': [{'q': {'user': USER}, 'u': {'$set': {'todos': st['users'].get(USER, [])}}}]}, 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 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) n_after = sum(len(v) for v in st['users'].values() if isinstance(v, list)) left = [t for u, l in st['users'].items() if isinstance(l, list) for t in l if (t.get('source') or {}).get('draft_id') == DRAFT] leftover = [f for f in os.listdir(ROOT + '/数据/activities') if f.startswith('ACT')] print(' 本地待办 %d → %d（清 %d）｜残留测试待办 %d｜残留活动 %d → %s' % (n_after + rm, n_after, rm, len(left), len(leftover), '✅ 干净' if not left and not leftover else '❌')) chk('清理干净', not left and not leftover and n_after == N0) print('\n%s' % ('✅ P1 端到端全部通过' if ok else '❌ 存在失败项')) sys.exit(0 if ok else 1)