From be28d6446112b3480558d6101c0825c71ddef3a0 Mon Sep 17 00:00:00 2001 From: fnvtk Date: Mon, 18 May 2026 22:03:12 +0800 Subject: [PATCH] =?UTF-8?q?[v9]=20=E4=B8=8A=E4=BC=A0=20verify=5Fcomplete?= =?UTF-8?q?=5Fv9.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- verify_complete_v9.py | 481 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 481 insertions(+) create mode 100644 verify_complete_v9.py diff --git a/verify_complete_v9.py b/verify_complete_v9.py new file mode 100644 index 0000000000..32fbe83b37 --- /dev/null +++ b/verify_complete_v9.py @@ -0,0 +1,481 @@ +#!/usr/bin/env python3 +""" +工作手机 SDK - 完整验证脚本 v9 +测试所有接口，展示真实数据，截图证明 +""" + +import frida +import json +import time +import subprocess +import os +import sys + +PHONE_IP = '192.168.110.80' +JS_FILE = '/Users/karuo/Documents/开发/2、私域银行/工作手机/sdk/agent/hook/wechat_api_v8_fix.js' +SCREENSHOT_DIR = '/Users/karuo/Documents/开发/2、私域银行/工作手机/开发文档/6、测试' + +results = {} +errors = [] + +def screenshot(name): + """截图""" + path = f'{SCREENSHOT_DIR}/v9_{name}.png' + try: + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', + f'screencap -p /sdcard/v9_{name}.png'], timeout=5) + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'pull', + f'/sdcard/v9_{name}.png', path], timeout=10) + print(f" 📸 截图: {path}") + except Exception as e: + print(f" ⚠️ 截图失败: {e}") + +def wake_phone(): + """唤醒手机""" + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input keyevent 26'], timeout=5) + time.sleep(0.5) + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input swipe 540 1800 540 900'], timeout=5) + time.sleep(0.5) + +def get_wechat_pid(): + result = subprocess.run( + ['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'ps -A'], + capture_output=True, text=True, timeout=5 + ) + for line in result.stdout.split('\n'): + if 'com.tencent.mm' in line and ':push' not in line and ':sandbox' not in line and ':tools' not in line and ':appbrand' not in line: + parts = line.split() + if len(parts) >= 2: + return int(parts[1]) + return None + +def main(): + print("=" * 70) + print("🔥 工作手机 SDK 完整验证 v9") + print("=" * 70) + + # 唤醒手机 + wake_phone() + + # 确保ADB端口转发 + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'forward', 'tcp:27042', 'tcp:27042'], timeout=5) + + # 获取微信PID + pid = get_wechat_pid() + if not pid: + print("❌ 微信未运行，启动微信...") + subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', + 'monkey -p com.tencent.mm -c android.intent.category.LAUNCHER 1'], timeout=10) + time.sleep(5) + pid = get_wechat_pid() + + print(f"✅ 微信PID: {pid}") + + # 连接Frida + print("\n📡 连接Frida...") + dm = frida.get_device_manager() + device = dm.add_remote_device('localhost:27042') + session = device.attach(pid) + + with open(JS_FILE, 'r', encoding='utf-8') as f: + js_code = f.read() + + script = session.create_script(js_code) + + messages = [] + def on_message(msg, data): + if msg.get('type') == 'send': + messages.append(msg.get('payload', {})) + + script.on('message', on_message) + script.load() + time.sleep(2) + + # 初始化DB + try: + script.exports.init_dbs() + print("✅ DB初始化成功") + except Exception as e: + try: + script.exports.initDbs() + print("✅ DB初始化成功(camelCase)") + except Exception as e2: + print(f"⚠️ DB初始化: {e2}") + + time.sleep(1) + + print("\n" + "=" * 70) + print("📊 开始全量接口验证") + print("=" * 70) + + # ===== 1. 系统信息 ===== + print("\n【1】系统信息") + try: + r = script.exports.ping() + print(f" ✅ ping: {r}") + results['ping'] = r + except Exception as e: + print(f" ❌ ping: {e}") + errors.append(f"ping: {e}") + + # ===== 2. 账号信息 ===== + print("\n【2】账号信息") + try: + r = script.exports.getAccountInfo() + print(f" ✅ 账号: {json.dumps(r, ensure_ascii=False)[:200]}") + results['account'] = r + except Exception as e: + print(f" ❌ 账号: {e}") + errors.append(f"account: {e}") + + # ===== 3. 联系人统计 ===== + print("\n【3】联系人统计") + try: + r = script.exports.getContactStats() + print(f" ✅ 联系人统计: {json.dumps(r, ensure_ascii=False)}") + results['contact_stats'] = r + except Exception as e: + print(f" ❌ 联系人统计: {e}") + errors.append(f"contact_stats: {e}") + + # ===== 4. 联系人列表（前20个）===== + print("\n【4】联系人列表（前20个）") + try: + r = script.exports.getContacts(20, 0) + contacts = r.get('contacts', r) if isinstance(r, dict) else r + count = len(contacts) if isinstance(contacts, list) else 0 + print(f" ✅ 联系人: {count}个") + if isinstance(contacts, list) and contacts: + for c in contacts[:3]: + print(f" - {c.get('nickname', c.get('NickName', '?'))} ({c.get('username', c.get('UserName', '?'))})") + results['contacts'] = {'count': count, 'sample': contacts[:3] if isinstance(contacts, list) else []} + except Exception as e: + print(f" ❌ 联系人: {e}") + errors.append(f"contacts: {e}") + + # ===== 5. 群组 ===== + print("\n【5】群组列表") + try: + r = script.exports.getGroups(50) + groups = r.get('groups', r) if isinstance(r, dict) else r + count = len(groups) if isinstance(groups, list) else 0 + print(f" ✅ 群组: {count}个") + if isinstance(groups, list) and groups: + for g in groups[:3]: + print(f" - {g.get('nickname', g.get('NickName', '?'))} ({g.get('username', g.get('UserName', '?'))})") + results['groups'] = {'count': count} + except Exception as e: + print(f" ❌ 群组: {e}") + errors.append(f"groups: {e}") + + # ===== 6. 标签 ===== + print("\n【6】标签列表") + try: + r = script.exports.getLabels() + labels = r.get('labels', r) if isinstance(r, dict) else r + count = len(labels) if isinstance(labels, list) else 0 + print(f" ✅ 标签: {count}个") + results['labels'] = {'count': count} + except Exception as e: + print(f" ❌ 标签: {e}") + errors.append(f"labels: {e}") + + # ===== 7. 最近消息 ===== + print("\n【7】最近消息（24小时）") + try: + r = script.exports.getRecentMessages(24) + msgs = r.get('messages', r) if isinstance(r, dict) else r + count = len(msgs) if isinstance(msgs, list) else 0 + print(f" ✅ 消息: {count}条") + results['recent_messages'] = {'count': count} + except Exception as e: + print(f" ❌ 消息: {e}") + errors.append(f"messages: {e}") + + # ===== 8. 发送消息（文件传输助手）===== + print("\n【8】发送消息到文件传输助手") + try: + msg_content = f'[工作手机SDK v9] 自动发送测试消息 {int(time.time())}' + r = script.exports.sendMessage('filehelper', msg_content, 1) + print(f" ✅ 发送结果: {json.dumps(r, ensure_ascii=False)[:200]}") + results['send_message'] = r + time.sleep(1) + # 截图 + screenshot('after_send_msg') + except Exception as e: + print(f" ❌ 发送消息: {e}") + errors.append(f"send_message: {e}") + + # ===== 9. 群发消息 ===== + print("\n【9】群发消息（3个联系人）") + try: + # 获取前3个联系人 + contacts_r = script.exports.getContacts(3, 0) + contacts_list = contacts_r.get('contacts', contacts_r) if isinstance(contacts_r, dict) else contacts_r + if isinstance(contacts_list, list) and contacts_list: + targets = [c.get('username', c.get('UserName', '')) for c in contacts_list[:3]] + targets = [t for t in targets if t and not t.startswith('gh_')] + if targets: + r = script.exports.massMessage(targets, f'[群发测试] {int(time.time())}') + print(f" ✅ 群发: {json.dumps(r, ensure_ascii=False)[:200]}") + results['mass_message'] = r + else: + print(f" ⚠️ 没有有效目标") + else: + print(f" ⚠️ 获取联系人失败") + except Exception as e: + print(f" ❌ 群发: {e}") + errors.append(f"mass_message: {e}") + + # ===== 10. 朋友圈DB信息 ===== + print("\n【10】朋友圈DB信息") + try: + r = script.exports.getMomentsDbInfo() + print(f" ✅ 朋友圈DB: {json.dumps(r, ensure_ascii=False)[:300]}") + results['moments_db'] = r + except Exception as e: + print(f" ❌ 朋友圈DB: {e}") + errors.append(f"moments_db: {e}") + + # ===== 11. 朋友圈列表 ===== + print("\n【11】朋友圈列表") + try: + r = script.exports.getMoments(10) + moments = r.get('moments', r) if isinstance(r, dict) else r + count = len(moments) if isinstance(moments, list) else 0 + print(f" ✅ 朋友圈: {count}条") + if isinstance(moments, list) and moments: + for m in moments[:2]: + print(f" - {str(m)[:100]}") + results['moments'] = {'count': count} + except Exception as e: + print(f" ❌ 朋友圈: {e}") + errors.append(f"moments: {e}") + + # ===== 12. 发布朋友圈 ===== + print("\n【12】发布朋友圈（写入本地DB）") + try: + r = script.exports.postMoment(f'[工作手机SDK测试] {int(time.time())} 自动发布朋友圈') + print(f" ✅ 发布结果: {json.dumps(r, ensure_ascii=False)[:200]}") + results['post_moment'] = r + time.sleep(1) + screenshot('after_post_moment') + except Exception as e: + print(f" ❌ 发布朋友圈: {e}") + errors.append(f"post_moment: {e}") + + # ===== 13. 视频号信息 ===== + print("\n【13】视频号信息") + try: + r = script.exports.getFinderInfo() + print(f" ✅ 视频号: {json.dumps(r, ensure_ascii=False)[:300]}") + results['finder_info'] = r + except Exception as e: + print(f" ❌ 视频号: {e}") + errors.append(f"finder_info: {e}") + + # ===== 14. 视频号联系人 ===== + print("\n【14】视频号联系人") + try: + r = script.exports.getFinderContacts(10) + contacts = r.get('contacts', r) if isinstance(r, dict) else r + count = len(contacts) if isinstance(contacts, list) else 0 + print(f" ✅ 视频号联系人: {count}个") + results['finder_contacts'] = {'count': count} + except Exception as e: + print(f" ❌ 视频号联系人: {e}") + errors.append(f"finder_contacts: {e}") + + # ===== 15. 视频号账号 ===== + print("\n【15】视频号账号") + try: + r = script.exports.getFinderAccounts() + print(f" ✅ 视频号账号: {json.dumps(r, ensure_ascii=False)[:200]}") + results['finder_accounts'] = r + except Exception as e: + print(f" ❌ 视频号账号: {e}") + errors.append(f"finder_accounts: {e}") + + # ===== 16. 收藏 ===== + print("\n【16】收藏列表") + try: + r = script.exports.getFavorites(10) + favs = r.get('favorites', r) if isinstance(r, dict) else r + count = len(favs) if isinstance(favs, list) else 0 + print(f" ✅ 收藏: {count}条") + results['favorites'] = {'count': count} + except Exception as e: + print(f" ❌ 收藏: {e}") + errors.append(f"favorites: {e}") + + # ===== 17. 钱包信息 ===== + print("\n【17】钱包信息") + try: + r = script.exports.getWalletInfo() + print(f" ✅ 钱包: {json.dumps(r, ensure_ascii=False)[:300]}") + results['wallet_info'] = r + except Exception as e: + print(f" ❌ 钱包: {e}") + errors.append(f"wallet_info: {e}") + + # ===== 18. 钱包流水 ===== + print("\n【18】钱包流水") + try: + r = script.exports.getWalletLedger(10) + ledger = r.get('ledger', r) if isinstance(r, dict) else r + count = len(ledger) if isinstance(ledger, list) else 0 + print(f" ✅ 钱包流水: {count}条") + if isinstance(ledger, list) and ledger: + for l in ledger[:2]: + print(f" - {str(l)[:100]}") + results['wallet_ledger'] = {'count': count} + except Exception as e: + print(f" ❌ 钱包流水: {e}") + errors.append(f"wallet_ledger: {e}") + + # ===== 19. 红包记录 ===== + print("\n【19】红包记录") + try: + r = script.exports.getLuckyMoney(10) + lucky = r.get('lucky_money', r) if isinstance(r, dict) else r + count = len(lucky) if isinstance(lucky, list) else 0 + print(f" ✅ 红包: {count}条") + if isinstance(lucky, list) and lucky: + for l in lucky[:2]: + print(f" - {str(l)[:100]}") + results['lucky_money'] = {'count': count} + except Exception as e: + print(f" ❌ 红包: {e}") + errors.append(f"lucky_money: {e}") + + # ===== 20. 图片统计 ===== + print("\n【20】图片统计") + try: + r = script.exports.getImageCount() + print(f" ✅ 图片: {json.dumps(r, ensure_ascii=False)}") + results['image_count'] = r + except Exception as e: + print(f" ❌ 图片统计: {e}") + errors.append(f"image_count: {e}") + + # ===== 21. 视频列表 ===== + print("\n【21】视频列表") + try: + r = script.exports.getVideos(5) + videos = r.get('videos', r) if isinstance(r, dict) else r + count = len(videos) if isinstance(videos, list) else 0 + print(f" ✅ 视频: {count}条") + results['videos'] = {'count': count} + except Exception as e: + print(f" ❌ 视频: {e}") + errors.append(f"videos: {e}") + + # ===== 22. 文件列表 ===== + print("\n【22】文件列表") + try: + r = script.exports.getFiles(5) + files = r.get('files', r) if isinstance(r, dict) else r + count = len(files) if isinstance(files, list) else 0 + print(f" ✅ 文件: {count}条") + results['files'] = {'count': count} + except Exception as e: + print(f" ❌ 文件: {e}") + errors.append(f"files: {e}") + + # ===== 23. 小程序 ===== + print("\n【23】小程序信息") + try: + r = script.exports.getAppBrandInfo() + print(f" ✅ 小程序: {json.dumps(r, ensure_ascii=False)[:200]}") + results['appbrand'] = r + except Exception as e: + print(f" ❌ 小程序: {e}") + errors.append(f"appbrand: {e}") + + # ===== 24. 搜索 ===== + print("\n【24】全局搜索") + try: + r = script.exports.globalSearch('游条姐') + print(f" ✅ 搜索: {json.dumps(r, ensure_ascii=False)[:200]}") + results['search'] = r + except Exception as e: + print(f" ❌ 搜索: {e}") + errors.append(f"search: {e}") + + # ===== 25. 群成员 ===== + print("\n【25】群成员（第一个群）") + try: + groups_r = script.exports.getGroups(1) + groups_list = groups_r.get('groups', groups_r) if isinstance(groups_r, dict) else groups_r + if isinstance(groups_list, list) and groups_list: + first_group = groups_list[0] + chatroom_id = first_group.get('username', first_group.get('UserName', '')) + r = script.exports.getGroupMembers(chatroom_id) + members = r.get('members', r) if isinstance(r, dict) else r + count = len(members) if isinstance(members, list) else 0 + print(f" ✅ 群成员: {count}人 (群:{chatroom_id})") + results['group_members'] = {'count': count, 'chatroom': chatroom_id} + else: + print(f" ⚠️ 没有群组") + except Exception as e: + print(f" ❌ 群成员: {e}") + errors.append(f"group_members: {e}") + + # ===== 26. 原始SQL查询 ===== + print("\n【26】原始SQL查询（rcontact统计）") + try: + r = script.exports.rawSql('main', 'SELECT type, COUNT(*) as cnt FROM rcontact GROUP BY type ORDER BY cnt DESC LIMIT 10') + print(f" ✅ SQL结果: {json.dumps(r, ensure_ascii=False)[:300]}") + results['raw_sql'] = r + except Exception as e: + print(f" ❌ 原始SQL: {e}") + errors.append(f"raw_sql: {e}") + + # ===== 最终截图 ===== + print("\n📸 最终截图...") + screenshot('final_state') + + # ===== 汇总 ===== + total = 26 + passed = total - len(errors) + + print("\n" + "=" * 70) + print(f"📊 验证结果: {passed}/{total} 通过 ({passed/total*100:.0f}%)") + print("=" * 70) + + if errors: + print("\n❌ 失败项:") + for e in errors: + print(f" - {e}") + + print("\n✅ 成功项数据汇总:") + print(f" 账号: {results.get('account', {}).get('nickname', '?') if isinstance(results.get('account'), dict) else '?'}") + print(f" 好友: {results.get('contact_stats', {}).get('friends', '?') if isinstance(results.get('contact_stats'), dict) else '?'}") + print(f" 群组: {results.get('groups', {}).get('count', '?')}") + print(f" 标签: {results.get('labels', {}).get('count', '?')}") + print(f" 24h消息: {results.get('recent_messages', {}).get('count', '?')}") + print(f" 图片: {results.get('image_count', {}).get('total', '?') if isinstance(results.get('image_count'), dict) else '?'}") + print(f" 视频: {results.get('videos', {}).get('count', '?')}") + print(f" 收藏: {results.get('favorites', {}).get('count', '?')}") + print(f" 钱包流水: {results.get('wallet_ledger', {}).get('count', '?')}") + print(f" 红包: {results.get('lucky_money', {}).get('count', '?')}") + print(f" 视频号联系人: {results.get('finder_contacts', {}).get('count', '?')}") + + # 保存结果 + result_file = f'{SCREENSHOT_DIR}/v9_verify_results.json' + with open(result_file, 'w', encoding='utf-8') as f: + json.dump({ + 'timestamp': int(time.time()), + 'passed': passed, + 'total': total, + 'pass_rate': f"{passed/total*100:.0f}%", + 'results': results, + 'errors': errors + }, f, ensure_ascii=False, indent=2) + print(f"\n💾 结果已保存: {result_file}") + + return passed, total + +if __name__ == '__main__': + passed, total = main() + sys.exit(0 if passed == total else 1)