#!/usr/bin/env python3 # -*- coding: utf-8 -*- """前端下钻 UI 实测：部门 → 销售 → 客户 → 活动明细（注入测试活动，跑完清理）。""" import io import json import os import sys import time from playwright.sync_api import sync_playwright ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' ACT = os.path.join(ROOT, '数据', 'activities') IDX = os.path.join(ACT, '_index.json') BASE = 'http://127.0.0.1:8765/' M = time.strftime('%Y-%m') IDS = ['ACTTESTDR1', 'ACTTESTDR2', 'ACTTESTDR3'] roles = json.load(io.open(ROOT + '/数据/user_roles.json', encoding='utf-8')) bd = {} for u in roles: bd.setdefault((u.get('dept') or '').strip(), []).append(u.get('name')) P_BIG, P_ZJ = bd['大客户部'][0], bd['生命科学-浙江'][0] 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') == '刘新元'], 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) TEST = [(IDS[0], P_BIG, '测试客户甲', 'offline', M + '-05'), (IDS[1], P_BIG, '测试客户乙', 'online', M + '-06'), (IDS[2], P_ZJ, '测试客户丙', 'offline', M + '-07')] ok = True def chk(label, cond, detail=''): global ok print(' %s %-40s %s' % ('✅' if cond else '❌', label, detail)) if not cond: ok = False idx0 = json.load(io.open(IDX, encoding='utf-8')) n0 = len(idx0) try: idx = list(idx0) for aid, owner, cust, ch, date in TEST: rec = {'id': aid, 'created': date + ' 10:00:00', 'submitted_by': owner, 'sales': owner, 'type': '项目跟进', 'subtype': '', 'customer': cust, 'contact': '', 'date': date, 'channel': ch, 'order_refs': [], 'order_ref_nos': [], 'opportunities': [], 'opp_names': [], 'opp_count': 0, 'blocks': [{'opp': '', 'need': 'x', 'plan': []}], 'block_count': 1, 'order_no': '', 'sku_count': 0, 'att_count': 0, 'feishu_text': '【测试】拜访记录'} io.open(os.path.join(ACT, aid + '.json'), 'w', encoding='utf-8').write(json.dumps(rec, ensure_ascii=False, indent=1)) idx.append({k: rec.get(k) for k in ('id', 'created', 'submitted_by', 'sales', 'type', 'subtype', 'customer', 'contact', 'date', 'channel', 'order_no', 'sku_count', 'att_count', 'opp_count', 'block_count')}) io.open(IDX, 'w', encoding='utf-8').write(json.dumps(idx, ensure_ascii=False, indent=1)) print('已注入 %d 条（%d → %d）｜样本 大客户部=%s 生命科学-浙江=%s\n' % (len(TEST), n0, len(idx), P_BIG, P_ZJ)) with sync_playwright() as p: br = p.chromium.launch(channel='chrome', headless=True) pg = br.new_page(viewport={'width': 1500, 'height': 1050}) errs = [] pg.on('pageerror', lambda e: errs.append(str(e))) pg.on('dialog', lambda d: d.accept()) pg.goto(BASE, wait_until='domcontentloaded') pg.evaluate("""([tk, idn]) => { localStorage.setItem('bior_auth_token', tk); localStorage.setItem('bior_sw_user_v2', JSON.stringify(idn)); }""", [TOKEN, IDENT]) pg.goto(BASE + '工具/index.html', wait_until='load') pg.wait_for_timeout(2500) 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) pg.evaluate("() => { visitDim='dept'; visitDrill={dept:'',sales:''}; visitHubState='see'; state.tab='visitStats'; render(); }") pg.wait_for_timeout(3000) def snap(): return pg.evaluate("""() => { const b=document.getElementById('visitBody'); const T=e=>e?e.innerText.replace(/\\s+/g,' ').trim():''; return {txt:T(b), rows:[...b.querySelectorAll('tbody tr')].map(tr=>T(tr)), crumb:T(document.getElementById('visitCrumb'))}; }""") print('① 一级：按部门') s = snap() chk('出现部门行（大客户部/生命科学销售部）', '大客户部' in s['txt'] and '生命科学销售部' in s['txt'], s['rows'][:2]) chk('部门行带数字', any('大客户部' in r and '1' in r for r in s['rows']), s['rows'][0][:60] if s['rows'] else '') pg.screenshot(path='/tmp/drill_1_dept.png') print('\n② 二级：下钻到销售') pg.evaluate("() => visitDrillTo('大客户部','')") pg.wait_for_timeout(900) s = snap() chk('面包屑 全部部门 › 大客户部', '大客户部' in s['crumb'], s['crumb'][:60]) chk('列出该部门销售', P_BIG in s['txt'], s['rows'][0][:70] if s['rows'] else '') pg.screenshot(path='/tmp/drill_2_sales.png') print('\n③ 三级：下钻到客户') pg.evaluate("(n) => visitDrillTo('大客户部', n)", P_BIG) pg.wait_for_timeout(900) s = snap() chk('只列该销售的客户', '测试客户甲' in s['txt'] and '测试客户乙' in s['txt'], str(len(s['rows'])) + ' 行') chk('面包屑到销售级', P_BIG in s['crumb'], s['crumb'][:70]) print('\n④ 活动明细 + 看详情') pg.evaluate("() => visitToggle(0)") pg.wait_for_timeout(500) n_acts = pg.evaluate("() => document.querySelectorAll('#visitActs0 td div').length") chk('展开出活动明细', n_acts >= 1, '%d 条' % n_acts) pg.evaluate("() => { const a=document.querySelector('#visitActs0 a'); if(a) a.click(); }") pg.wait_for_timeout(2200) det = pg.evaluate("() => { const p=document.querySelector('#visitActs0 pre'); return p?p.innerText.slice(0,40):''; }") chk('看详情渲染出活动文本', len(det) > 5, det[:40]) pg.screenshot(path='/tmp/drill_3_cust.png') print('\n⑤ 面包屑回上级 + 维度切换') pg.evaluate("() => { const a=document.querySelector('#visitCrumb a'); if(a) a.click(); }") pg.wait_for_timeout(900) s = snap() chk('点面包屑回到一级', '大客户部' in s['txt'] and '销售数' in s['txt']) pg.evaluate("() => visitSetDim('cust')") pg.wait_for_timeout(2500) s = snap() chk('按客户维度：出现客户行', '测试客户甲' in s['txt'], str(len(s['rows'])) + ' 行') chk('按客户维度不再有部门语义', '销售数' not in s['txt']) pg.evaluate("() => visitSetDim('dept')") pg.wait_for_timeout(2500) chk('全程无 JS 错误', not errs, str(errs[:2])) br.close() finally: idx = json.load(io.open(IDX, encoding='utf-8')) idx = [x for x in idx if x.get('id') not in IDS] io.open(IDX, 'w', encoding='utf-8').write(json.dumps(idx, ensure_ascii=False, indent=1)) for aid in IDS: pth = os.path.join(ACT, aid + '.json') if os.path.exists(pth): os.remove(pth) left = [f for f in os.listdir(ACT) if f.startswith('ACTTEST')] print('\n=== 清理 === 索引 %d → %d ｜ 残留 %d → %s' % (n0, len(idx), len(left), '✅ 干净' if not left else '❌')) ok = ok and not left and len(idx) == n0 print('\n%s' % ('✅ 前端下钻管道实测通过' if ok else '❌ 有失败项')) sys.exit(0 if ok else 1)