482 lines
18 KiB
Python
482 lines
18 KiB
Python
#!/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)
|