feat: v8-fix Java.choose DB - verify_v8_fix.py

verify_v8_fix.py

@@ -0,0 +1,336 @@
+#!/usr/bin/env python3
+"""工作手机 SDK v8-fix - 全量接口验证"""
+import frida, json, time, subprocess, os
+
+PHONE_PORT = 27042
+WECHAT_PID = 25425
+JS_FILE = '/Users/karuo/Documents/开发/2、私域银行/工作手机/sdk/agent/hook/wechat_api_v8_fix.js'
+SCREENSHOT_DIR = '/Users/karuo/Documents/开发/2、私域银行/工作手机/开发文档/6、测试/'
+ADB = ['adb', '-s', '192.168.110.80:5555']
+RESULT_FILE = '/Users/karuo/Documents/开发/2、私域银行/工作手机/verification_v8fix_results.json'
+
+def screenshot(name):
+    remote = f'/sdcard/v8fix_{name}.png'
+    local = SCREENSHOT_DIR + f'v8fix_{name}.png'
+    subprocess.run(ADB + ['shell', f'screencap -p {remote}'], capture_output=True)
+    time.sleep(0.5)
+    subprocess.run(ADB + ['pull', remote, local], capture_output=True)
+    return local
+
+def main():
+    print('='*65)
+    print('工作手机 SDK v8-fix - 全量接口验证')
+    print('='*65)
+
+    device = frida.get_device_manager().add_remote_device(f'127.0.0.1:{PHONE_PORT}')
+    session = device.attach(WECHAT_PID)
+    with open(JS_FILE) as f:
+        js = f.read()
+    script = session.create_script(js)
+    
+    log_msgs = []
+    def on_msg(msg, data):
+        if msg['type'] == 'send':
+            log_msgs.append(msg['payload'])
+        elif msg['type'] == 'log':
+            print(f'  [JS] {msg["payload"]}')
+    script.on('message', on_msg)
+    script.load()
+    time.sleep(3)
+    e = script.exports_sync
+
+    results = {}
+    passed = 0
+    failed = 0
+
+    def check(name, r, key=None):
+        nonlocal passed, failed
+        if r.get('error') and not r.get('success'):
+            print(f'    ❌ {name}: {r.get("error","")[:60]}')
+            failed += 1
+        else:
+            passed += 1
+        return r
+
+    print('\n[1] Ping...')
+    r = e.ping()
+    print(f'    ✅ {r}')
+    passed += 1
+
+    print('\n[2] 初始化所有DB...')
+    r = e.init_dbs()
+    status = r.get('status', {})
+    for k, v in status.items():
+        print(f'    {"✅" if v else "❌"} {k}')
+    print(f'    总DB实例: {r.get("total_dbs")}')
+    results['db_init'] = r
+
+    print('\n[3] 列出所有DB...')
+    r = e.list_all_dbs()
+    print(f'    找到 {r.get("count")} 个DB')
+    for db in (r.get('dbs') or []):
+        print(f'    {db["name"]}: {db["path"].split("/")[-1]}')
+    results['all_dbs'] = r
+
+    print('\n[4] 账号信息...')
+    r = e.get_account_info()
+    print(f'    微信号: {r.get("wxid")}')
+    print(f'    昵称: {r.get("nickname")}')
+    print(f'    手机: {r.get("phone")}')
+    print(f'    地区: {r.get("region")}')
+    print(f'    性别: {r.get("sex")}')
+    results['account'] = r
+    if r.get('wxid'): passed += 1
+    else: failed += 1
+
+    print('\n[5] 全量统计...')
+    r = e.get_all_stats()
+    print(f'    ✅ 好友数: {r.get("friends")}')
+    print(f'    ✅ 群组数: {r.get("groups")}')
+    print(f'    ✅ 消息总数: {r.get("messages")}')
+    print(f'    ✅ 图片总数: {r.get("images")}')
+    print(f'    ✅ 视频总数: {r.get("videos")}')
+    print(f'    ✅ 会话数: {r.get("conversations")}')
+    print(f'    ✅ 文件数: {r.get("files")}')
+    results['stats'] = r
+    passed += 1
+
+    print('\n[6] 联系人数量...')
+    r = e.get_contact_count()
+    print(f'    ✅ 真实好友: {r.get("friends")} | 总记录: {r.get("total")}')
+    results['contact_count'] = r
+    passed += 1
+
+    print('\n[7] 联系人列表（前10个）...')
+    r = e.get_contacts(10, 0)
+    for c in (r.get('rows') or [])[:10]:
+        print(f'    {c.get("nickName","?")} | 备注:{c.get("conRemark","无")} | {c.get("username","")}')
+    results['contacts'] = r
+    passed += 1
+
+    print('\n[8] 搜索联系人（搜索"游"）...')
+    r = e.search_contact('游')
+    print(f'    找到 {r.get("count")} 个')
+    for c in (r.get('rows') or [])[:5]:
+        print(f'    {c.get("nickName","?")} ({c.get("username","?")})')
+    results['search'] = r
+    passed += 1
+
+    print('\n[9] 最近消息（24小时）...')
+    r = e.get_recent_messages(24)
+    print(f'    24小时消息: {r.get("count")} 条')
+    for m in (r.get('rows') or [])[:5]:
+        print(f'    [{m.get("talker","?")}] {str(m.get("content",""))[:40]}')
+    results['recent_msgs'] = r
+    passed += 1
+
+    print('\n[10] 发送消息到文件传输助手...')
+    content = f'SDK v8-fix 测试消息 {time.strftime("%Y-%m-%d %H:%M:%S")}'
+    r = e.send_message('filehelper', content)
+    print(f'    发送: success={r.get("success")}, msgId={r.get("msgId")}')
+    print(f'    DB验证: {"✅ 已写入" if r.get("db_verified") else "❌ 未写入"}')
+    results['send_msg'] = r
+    if r.get('db_verified'): passed += 1
+    else: failed += 1
+    time.sleep(1)
+    screenshot('after_send_msg')
+
+    print('\n[11] 会话列表（前10个）...')
+    r = e.get_conversations(10)
+    print(f'    会话数: {r.get("count")}')
+    for c in (r.get('rows') or [])[:5]:
+        print(f'    [{c.get("username","?")}] 未读:{c.get("unReadCount",0)} | {str(c.get("lastMsg",""))[:30]}')
+    results['conversations'] = r
+    passed += 1
+
+    print('\n[12] 群组列表（前10个）...')
+    r = e.get_groups(10)
+    print(f'    群组数: {r.get("count")}')
+    for g in (r.get('rows') or [])[:5]:
+        print(f'    {g.get("chatroomname","?")} | 成员:{g.get("memberCount",0)} | 群主:{g.get("roomowner","?")}')
+    results['groups'] = r
+    passed += 1
+
+    print('\n[13] 群成员详情（第1个群）...')
+    if r.get('rows'):
+        gid = r['rows'][0].get('chatroomname')
+        r2 = e.get_group_members(gid)
+        print(f'    群ID: {gid}')
+        print(f'    成员数: {r2.get("count")} | memberCount字段: {r2.get("memberCount")}')
+        print(f'    群主: {r2.get("owner")}')
+        print(f'    群公告: {str(r2.get("announcement",""))[:60] or "无"}')
+        print(f'    我的群名: {r2.get("myName","无")}')
+        print(f'    前3成员: {(r2.get("members") or [])[:3]}')
+        results['group_members'] = r2
+        passed += 1
+
+    print('\n[14] 标签统计...')
+    r = e.get_labels()
+    print(f'    标签数: {r.get("count")}')
+    top = sorted((r.get('usage') or {}).items(), key=lambda x: -int(x[1]))[:10]
+    print(f'    使用最多的标签: {top}')
+    results['labels'] = r
+    passed += 1
+
+    print('\n[15] 朋友圈DB信息...')
+    r = e.get_sns_db_info()
+    for k, v in r.items():
+        print(f'    {k}: available={v.get("available")}, tables={v.get("tables",[])}')
+    results['sns_info'] = r
+
+    print('\n[16] 朋友圈内容（前5条）...')
+    r = e.get_moments(5)
+    print(f'    朋友圈条数: {r.get("count")} | DB: {r.get("db","?")}')
+    for m in (r.get('rows') or [])[:5]:
+        print(f'    [{m.get("userName","?")}] {str(m.get("content",""))[:60]}')
+    results['moments'] = r
+    if not r.get('error'): passed += 1
+    else: failed += 1
+
+    print('\n[17] 发布朋友圈...')
+    moment_text = f'SDK v8-fix 后端发布 {time.strftime("%H:%M:%S")} #测试'
+    r = e.post_moment(moment_text)
+    print(f'    发布: success={r.get("success")}, snsId={r.get("snsId")}, db={r.get("db")}')
+    print(f'    DB验证: {"✅ 已写入" if r.get("db_verified") else "❌ 未写入"} | error={r.get("error","无")}')
+    results['post_moment'] = r
+    if r.get('db_verified'): passed += 1
+    else: failed += 1
+    time.sleep(1)
+    screenshot('after_post_moment')
+
+    print('\n[18] 视频号信息...')
+    r = e.get_finder_info()
+    for k, v in r.items():
+        print(f'    {k}: available={v.get("available")}, tables={v.get("tables",[])}')
+    results['finder_info'] = r
+
+    print('\n[19] 视频号联系人...')
+    r = e.get_finder_contacts(10)
+    print(f'    视频号联系人: {r.get("count",0)} 个')
+    for c in (r.get('rows') or [])[:3]:
+        print(f'    {str(c)[:60]}')
+    results['finder_contacts'] = r
+    if not r.get('error'): passed += 1
+    else: failed += 1
+
+    print('\n[20] 视频号媒体缓存...')
+    r = e.get_finder_media(5)
+    print(f'    视频号媒体: {r.get("count",0)} 条')
+    results['finder_media'] = r
+    if not r.get('error'): passed += 1
+    else: failed += 1
+
+    print('\n[21] 视频号账号...')
+    r = e.get_finder_accounts()
+    print(f'    视频号账号表: {r.get("tables",[])}')
+    for k, v in r.items():
+        if k != 'tables' and isinstance(v, dict) and v.get('rows'):
+            print(f'    {k}: {v["count"]} 条')
+    results['finder_accounts'] = r
+
+    print('\n[22] 收藏...')
+    r = e.get_favorites(10)
+    print(f'    收藏: {r.get("count",0)} 条 | error={r.get("error","无")}')
+    for fav in (r.get('rows') or [])[:3]:
+        print(f'    {str(fav)[:60]}')
+    results['favorites'] = r
+    if not r.get('error'): passed += 1
+    else: failed += 1
+
+    print('\n[23] 钱包信息...')
+    r = e.get_wallet_info()
+    print(f'    钱包表: {r.get("tables",[])}')
+    results['wallet_info'] = r
+    if not r.get('error'): passed += 1
+    else: failed += 1
+
+    print('\n[24] 钱包账单（前5条）...')
+    r = e.get_wallet_ledger(5)
+    print(f'    账单: {r.get("count",0)} 条')
+    for ledger in (r.get('rows') or [])[:5]:
+        print(f'    用户:{ledger.get("tradeUser","?")} 金额:{ledger.get("tradeAmount","?")} 类型:{ledger.get("tradeType","?")} 日期:{ledger.get("msgDate","?")}')
+    results['wallet_ledger'] = r
+    if not r.get('error'): passed += 1
+    else: failed += 1
+
+    print('\n[25] 钱包用户信息...')
+    r = e.get_wallet_user_info()
+    print(f'    钱包表: {r.get("tables",[])}')
+    for k, v in r.items():
+        if k != 'tables' and isinstance(v, dict) and v.get('rows'):
+            print(f'    {k}: {v["rows"][:1]}')
+    results['wallet_user'] = r
+
+    print('\n[26] 红包记录...')
+    r = e.get_lucky_money(5)
+    print(f'    红包: {r.get("count",0)} 条')
+    for lm in (r.get('rows') or [])[:3]:
+        print(f'    {str(lm)[:60]}')
+    results['lucky_money'] = r
+    passed += 1
+
+    print('\n[27] 图片统计...')
+    r = e.get_image_count()
+    total = r.get('rows',[{}])[0].get('total',0) if r.get('rows') else 0
+    print(f'    ✅ 图片总数: {total}')
+    results['image_count'] = r
+    passed += 1
+
+    print('\n[28] 视频统计...')
+    r = e.get_video_count()
+    total = r.get('rows',[{}])[0].get('total',0) if r.get('rows') else 0
+    print(f'    ✅ 视频总数: {total}')
+    results['video_count'] = r
+    passed += 1
+
+    print('\n[29] 文件列表...')
+    r = e.get_file_list(5)
+    print(f'    文件: {r.get("count",0)} 条')
+    results['files'] = r
+    passed += 1
+
+    print('\n[30] 小程序列表...')
+    r = e.get_app_brand_list()
+    print(f'    小程序表: {r.get("tables",[])}')
+    for k, v in r.items():
+        if k != 'tables' and isinstance(v, dict) and v.get('count',0) > 0:
+            print(f'    {k}: {v["count"]} 条')
+    results['appbrand'] = r
+    if not r.get('error'): passed += 1
+    else: failed += 1
+
+    print('\n[31] 群发消息（发给文件传输助手）...')
+    r = e.mass_message(['filehelper'], f'群发测试 {time.strftime("%H:%M:%S")}')
+    print(f'    群发: 总数={r.get("total")}, 成功={r.get("success_count")}')
+    results['mass_msg'] = r
+    if r.get('success_count',0) > 0: passed += 1
+    else: failed += 1
+
+    print('\n[32] 全局搜索（搜索"游"）...')
+    r = e.global_search('游')
+    print(f'    联系人: {r.get("contacts",{}).get("count",0)} 个')
+    print(f'    消息: {r.get("messages",{}).get("count",0)} 条')
+    results['global_search'] = r
+    passed += 1
+
+    print('\n[33] 原始SQL查询...')
+    r = e.raw_query('main', 'SELECT COUNT(*) as n FROM rcontact WHERE type=3')
+    print(f'    好友数（原始SQL）: {r.get("rows",[{}])[0].get("n","?")}')
+    results['raw_query'] = r
+    passed += 1
+
+    # 最终截图
+    screenshot('final')
+
+    # 保存结果
+    with open(RESULT_FILE, 'w', encoding='utf-8') as f:
+        json.dump(results, f, ensure_ascii=False, indent=2, default=str)
+
+    print('\n' + '='*65)
+    total = passed + failed
+    print(f'✅ 验证完成！通过: {passed}/{total} ({int(passed/total*100)}%)')
+    print(f'结果保存: {RESULT_FILE}')
+    print('='*65)
+
+if __name__ == '__main__':
+    main()