feat: v8 file-path DB - verify_v8.py

verify_v8.py

@@ -0,0 +1,246 @@
+#!/usr/bin/env python3
+"""工作手机 SDK v8 - 全量接口验证"""
+import frida, json, time, subprocess, os
+
+PHONE_IP = '127.0.0.1'
+PHONE_PORT = 27042
+WECHAT_PID = 25425
+JS_FILE = '/Users/karuo/Documents/开发/2、私域银行/工作手机/sdk/agent/hook/wechat_api_v8.js'
+SCREENSHOT_DIR = '/Users/karuo/Documents/开发/2、私域银行/工作手机/开发文档/6、测试/'
+ADB = ['adb', '-s', '192.168.110.80:5555']
+
+def screenshot(name):
+    remote = f'/sdcard/v8_{name}.png'
+    local = SCREENSHOT_DIR + f'v8_{name}.png'
+    subprocess.run(ADB + ['shell', f'screencap -p {remote}'], capture_output=True)
+    subprocess.run(ADB + ['pull', remote, local], capture_output=True)
+    return local
+
+def main():
+    print('='*65)
+    print('工作手机 SDK v8 - 全量接口验证（文件路径直接打开DB）')
+    print('='*65)
+
+    device = frida.get_device_manager().add_remote_device(f'{PHONE_IP}:{PHONE_PORT}')
+    session = device.attach(WECHAT_PID)
+    with open(JS_FILE) as f:
+        js = f.read()
+    script = session.create_script(js)
+    script.load()
+    time.sleep(2)
+    e = script.exports_sync
+
+    results = {}
+
+    print('\n[1] 初始化所有DB...')
+    r = e.init_dbs()
+    print(f'    DB状态: {r["status"]}')
+    results['db_init'] = r
+
+    print('\n[2] 账号信息...')
+    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")}')
+    results['account'] = r
+
+    print('\n[3] 全量统计...')
+    r = e.get_all_contact_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")}')
+    results['stats'] = r
+
+    print('\n[4] 联系人列表（前5个）...')
+    r = e.get_contacts(5, 0)
+    for c in (r.get('rows') or [])[:5]:
+        print(f'    {c.get("nickName","?")} | 备注:{c.get("conRemark","无")} | wxid:{c.get("username","?")}')
+    results['contacts'] = r
+
+    print('\n[5] 搜索联系人（搜索"游"）...')
+    r = e.search_contact('游')
+    print(f'    找到 {r.get("count")} 个')
+    for c in (r.get('rows') or [])[:3]:
+        print(f'    {c.get("nickName","?")} ({c.get("username","?")})')
+    results['search'] = r
+
+    print('\n[6] 最近消息（24小时）...')
+    r = e.get_recent_messages(24)
+    print(f'    24小时消息: {r.get("count")} 条')
+    for m in (r.get('rows') or [])[:3]:
+        print(f'    [{m.get("talker","?")}] {str(m.get("content",""))[:40]}')
+    results['recent_msgs'] = r
+
+    print('\n[7] 发送消息到文件传输助手...')
+    content = f'SDK v8 测试消息 {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
+    time.sleep(1)
+    screenshot('after_send_msg')
+
+    print('\n[8] 会话列表（前5个）...')
+    r = e.get_conversations(5)
+    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
+
+    print('\n[9] 群组列表（前5个）...')
+    r = e.get_groups(5)
+    for g in (r.get('rows') or [])[:5]:
+        print(f'    {g.get("chatroomname","?")} | 成员:{g.get("memberCount",0)} | 群主:{g.get("roomowner","?")}')
+    results['groups'] = r
+
+    print('\n[10] 群成员详情（第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")}')
+        print(f'    群主: {r2.get("owner")}')
+        print(f'    群公告: {str(r2.get("announcement",""))[:50] or "无"}')
+        print(f'    我的群名: {r2.get("myName","无")}')
+        print(f'    前3成员: {(r2.get("members") or [])[:3]}')
+        results['group_members'] = r2
+
+    print('\n[11] 标签统计...')
+    r = e.get_labels()
+    print(f'    标签数: {r.get("count")}')
+    print(f'    前10个标签ID: {(r.get("label_ids") or [])[:10]}')
+    results['labels'] = r
+
+    print('\n[12] 朋友圈DB表结构...')
+    r = e.get_sns_db_tables()
+    sns_tables = [t['name'] for t in (r.get('sns',{}).get('rows') or [])]
+    sns_star_tables = [t['name'] for t in (r.get('sns_star',{}).get('rows') or [])]
+    print(f'    SnsMicroMsg.db 表: {sns_tables}')
+    print(f'    sns_star_info.db 表: {sns_star_tables}')
+    results['sns_tables'] = r
+
+    print('\n[13] 朋友圈内容（前5条）...')
+    r = e.get_moments(5)
+    print(f'    朋友圈条数: {r.get("count")}')
+    for m in (r.get('rows') or [])[:5]:
+        print(f'    [{m.get("userName","?")}] {str(m.get("content",""))[:50]}')
+    results['moments'] = r
+
+    print('\n[14] 发布朋友圈...')
+    moment_text = f'SDK v8 后端发布朋友圈 {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 "❌ 未写入"}')
+    results['post_moment'] = r
+    time.sleep(1)
+    screenshot('after_post_moment')
+
+    print('\n[15] 视频号...')
+    r = e.get_finder_info()
+    for db_name, info in r.items():
+        tables = [t['name'] for t in (info.get('rows') or [])]
+        print(f'    {db_name}: {tables[:5]}')
+    results['finder'] = r
+
+    r2 = e.get_finder_contacts()
+    print(f'    视频号联系人: {r2.get("count",0)} 个')
+    results['finder_contacts'] = r2
+
+    r3 = e.get_finder_media()
+    print(f'    视频号媒体缓存: {r3.get("count",0)} 条')
+    results['finder_media'] = r3
+
+    r4 = e.get_finder_accounts()
+    print(f'    视频号账号数据: {list(r4.keys()) if isinstance(r4, dict) else r4}')
+    results['finder_accounts'] = r4
+
+    print('\n[16] 收藏...')
+    r = e.get_favorites(10)
+    print(f'    收藏数: {r.get("count",0)}')
+    for fav in (r.get('rows') or [])[:3]:
+        print(f'    收藏: {str(fav)[:60]}')
+    results['favorites'] = r
+
+    print('\n[17] 钱包信息...')
+    r = e.get_wallet_info()
+    wc_tables = [t['name'] for t in (r.get('WalletCoreDB',{}).get('rows') or [])]
+    wp_tables = [t['name'] for t in (r.get('WalletPay',{}).get('rows') or [])]
+    print(f'    WalletCoreDB表: {wc_tables}')
+    print(f'    WalletPay表: {wp_tables}')
+    results['wallet_info'] = r
+
+    print('\n[18] 钱包账单（前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
+
+    print('\n[19] 红包记录...')
+    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
+
+    print('\n[20] 钱包余额...')
+    r = e.get_wallet_balance()
+    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"][:2]}')
+    results['wallet_balance'] = r
+
+    print('\n[21] 图片/视频统计...')
+    r_img = e.get_image_count()
+    r_vid = e.get_video_count()
+    img_total = r_img.get('rows',[{}])[0].get('total',0) if r_img.get('rows') else 0
+    vid_total = r_vid.get('rows',[{}])[0].get('total',0) if r_vid.get('rows') else 0
+    print(f'    图片总数: {img_total}')
+    print(f'    视频总数: {vid_total}')
+    results['media_stats'] = {'images': img_total, 'videos': vid_total}
+
+    print('\n[22] 文件列表（前3条）...')
+    r = e.get_file_list(3)
+    print(f'    文件记录: {r.get("count",0)} 条')
+    for f_ in (r.get('rows') or [])[:3]:
+        print(f'    {str(f_)[:60]}')
+    results['files'] = r
+
+    print('\n[23] 小程序列表...')
+    r = e.get_app_brand_list()
+    print(f'    小程序表: {r.get("tables",[])}')
+    results['appbrand'] = r
+
+    print('\n[24] 群发消息（发给文件传输助手）...')
+    r = e.mass_message(['filehelper'], f'群发测试 {time.strftime("%H:%M:%S")}')
+    print(f'    群发: 总数={r.get("total")}, 成功={r.get("success_count")}')
+    results['mass_msg'] = r
+
+    print('\n[25] 全局搜索（搜索"游"）...')
+    r = e.global_search('游')
+    print(f'    联系人: {r.get("contacts",{}).get("count",0)} 个')
+    print(f'    消息: {r.get("messages",{}).get("count",0)} 条')
+    results['global_search'] = r
+
+    # 最终截图
+    screenshot('v8_final')
+
+    # 保存结果
+    result_file = '/Users/karuo/Documents/开发/2、私域银行/工作手机/verification_v8_results.json'
+    with open(result_file, 'w', encoding='utf-8') as f:
+        json.dump(results, f, ensure_ascii=False, indent=2, default=str)
+
+    print('\n' + '='*65)
+    print(f'✅ 验证完成！结果保存: {result_file}')
+    print('='*65)
+
+    # 统计
+    errors = sum(1 for v in results.values() if isinstance(v, dict) and v.get('error'))
+    print(f'接口总数: 25 | 有错误: {errors} | 成功: {25-errors}')
+
+if __name__ == '__main__':
+    main()