From 82bc6d7078f13036130ce54c5af7b2133595a727 Mon Sep 17 00:00:00 2001 From: fnvtk Date: Mon, 18 May 2026 21:46:23 +0800 Subject: [PATCH] feat: v8 file-path DB - verify_v8.py --- verify_v8.py | 246 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 246 insertions(+) create mode 100644 verify_v8.py diff --git a/verify_v8.py b/verify_v8.py new file mode 100644 index 0000000000..00683666a4 --- /dev/null +++ b/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()