siyu-bank/workphone-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[v9-fix] 修复exports_sync调用方式

Browse Source
This commit is contained in:
fnvtk
2026-05-18 22:05:41 +08:00
parent a94d8cb842
commit a8bce7f469
1 changed files with 405 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

405
verify_v9_fixed.py Normal file
View File

@@ -0,0 +1,405 @@
#!/usr/bin/env python3
"""
工作手机 SDK - 完整验证脚本 v9-fixed
使用exports_sync，正确调用所有接口
"""
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 = []
passed_count = 0
total_count = 0
def screenshot(name):
path = f'{SCREENSHOT_DIR}/v9fix_{name}.png'
try:
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell',
f'screencap -p /sdcard/v9fix_{name}.png'], timeout=5, capture_output=True)
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'pull',
f'/sdcard/v9fix_{name}.png', path], timeout=10, capture_output=True)
print(f" 📸 截图已保存: v9fix_{name}.png")
except Exception as e:
print(f" ⚠️ 截图失败: {e}")
def wake_phone():
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input keyevent 26'], timeout=5, capture_output=True)
time.sleep(0.5)
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'shell', 'input swipe 540 1800 540 900'], timeout=5, capture_output=True)
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 test(name, func):
global passed_count, total_count
total_count += 1
try:
r = func()
passed_count += 1
return r
except Exception as e:
errors.append(f"{name}: {e}")
print(f" ❌ {name}: {e}")
return None
def main():
global passed_count, total_count
print("=" * 70)
print("🔥 工作手机 SDK 完整验证 v9-fixed")
print(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
print("=" * 70)
# 唤醒手机
wake_phone()
# 确保ADB端口转发
subprocess.run(['adb', '-s', f'{PHONE_IP}:5555', 'forward', 'tcp:27042', 'tcp:27042'],
timeout=5, capture_output=True)
# 获取微信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, capture_output=True)
time.sleep(5)
pid = get_wechat_pid()
if not pid:
print("❌ 无法启动微信")
return 0, 0
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)
script.load()
time.sleep(2)
# 使用exports_sync
ex = script.exports_sync
# 初始化DB
try:
r = ex.initDbs()
print(f"✅ DB初始化: {json.dumps(r, ensure_ascii=False)[:200]}")
except Exception as e:
print(f"⚠️ DB初始化: {e}")
time.sleep(1)
print("\n" + "=" * 70)
print("📊 开始全量接口验证（26项）")
print("=" * 70)
# ===== 1. ping =====
print("\n【1】系统ping")
r = test("ping", lambda: ex.ping())
if r: print(f" ✅ {r}")
results['ping'] = r
# ===== 2. 账号信息 =====
print("\n【2】账号信息")
r = test("account", lambda: ex.getAccountInfo())
if r:
print(f" ✅ 账号: {r.get('nickname', '?')} | wxid: {r.get('wxid', '?')}")
print(f" 手机: {r.get('phone', '?')} | 地区: {r.get('region', '?')}")
results['account'] = r
# ===== 3. 全量统计 =====
print("\n【3】全量数据统计")
r = test("stats", lambda: ex.getAllStats())
if r:
print(f" ✅ 好友: {r.get('friends', 0)} | 群组: {r.get('groups', 0)} | 消息: {r.get('messages', 0)}")
print(f" 图片: {r.get('images', 0)} | 视频: {r.get('videos', 0)} | 文件: {r.get('files', 0)}")
results['stats'] = r
# ===== 4. 联系人列表 =====
print("\n【4】联系人列表（前20个）")
r = test("contacts", lambda: ex.getContacts(20, 0))
if r:
rows = r.get('rows', [])
print(f" ✅ 联系人: {len(rows)}个")
for c in rows[:3]:
print(f" - {c.get('nickName', '?')} ({c.get('username', '?')})")
results['contacts'] = {'count': len(rows), 'sample': rows[:3]}
# ===== 5. 联系人数量 =====
print("\n【5】联系人数量")
r = test("contact_count", lambda: ex.getContactCount())
if r:
print(f" ✅ 好友: {r.get('friends', 0)} | 总计: {r.get('total', 0)}")
results['contact_count'] = r
# ===== 6. 群组列表 =====
print("\n【6】群组列表")
r = test("groups", lambda: ex.getGroups(50))
if r:
rows = r.get('rows', [])
print(f" ✅ 群组: {len(rows)}个")
for g in rows[:3]:
print(f" - {g.get('nickName', g.get('NickName', '?'))} ({g.get('username', '?')})")
results['groups'] = {'count': len(rows), 'sample': rows[:3]}
# ===== 7. 标签 =====
print("\n【7】标签列表")
r = test("labels", lambda: ex.getLabels())
if r:
rows = r.get('rows', [])
print(f" ✅ 标签: {len(rows)}个")
for l in rows[:5]:
print(f" - {l.get('labelName', l.get('name', '?'))} (id:{l.get('labelId', l.get('id', '?'))})")
results['labels'] = {'count': len(rows)}
# ===== 8. 最近消息 =====
print("\n【8】最近消息（24小时）")
r = test("recent_messages", lambda: ex.getRecentMessages(24))
if r:
rows = r.get('rows', [])
print(f" ✅ 消息: {len(rows)}条")
results['recent_messages'] = {'count': len(rows)}
# ===== 9. 发送消息 =====
print("\n【9】发送消息到文件传输助手")
msg_content = f'[工作手机SDK v9] 自动测试消息 {time.strftime("%H:%M:%S")}'
r = test("send_message", lambda: ex.sendMessage('filehelper', msg_content, 1))
if r:
print(f" ✅ 发送结果: {json.dumps(r, ensure_ascii=False)[:200]}")
results['send_message'] = r
time.sleep(1)
screenshot('after_send_msg')
# ===== 10. 群发消息 =====
print("\n【10】群发消息")
try:
contacts_r = ex.getContacts(5, 0)
rows = contacts_r.get('rows', [])
targets = [c.get('username', '') for c in rows if c.get('username', '') and not c.get('username', '').startswith('gh_')][:3]
if targets:
r = test("mass_message", lambda: ex.massMessage(targets, f'[群发测试] {time.strftime("%H:%M:%S")}'))
if r:
print(f" ✅ 群发: 目标{len(targets)}人, 结果: {json.dumps(r, ensure_ascii=False)[:150]}")
results['mass_message'] = r
else:
print(f" ⚠️ 没有有效目标")
except Exception as e:
print(f" ❌ 群发: {e}")
errors.append(f"mass_message: {e}")
# ===== 11. 群成员 =====
print("\n【11】群成员（第一个群）")
try:
groups_r = ex.getGroups(1)
rows = groups_r.get('rows', [])
if rows:
chatroom_id = rows[0].get('username', '')
r = test("group_members", lambda: ex.getGroupMembers(chatroom_id))
if r:
members = r.get('rows', [])
print(f" ✅ 群: {chatroom_id} | 成员: {len(members)}人")
results['group_members'] = {'count': len(members), 'chatroom': chatroom_id}
except Exception as e:
print(f" ❌ 群成员: {e}")
errors.append(f"group_members: {e}")
# ===== 12. 朋友圈DB信息 =====
print("\n【12】朋友圈DB信息")
r = test("moments_db", lambda: ex.getMomentsDbInfo())
if r:
print(f" ✅ 朋友圈DB: {json.dumps(r, ensure_ascii=False)[:300]}")
results['moments_db'] = r
# ===== 13. 朋友圈列表 =====
print("\n【13】朋友圈列表")
r = test("moments", lambda: ex.getMoments(10))
if r:
rows = r.get('rows', [])
print(f" ✅ 朋友圈: {len(rows)}条")
for m in rows[:2]:
print(f" - {str(m)[:100]}")
results['moments'] = {'count': len(rows)}
# ===== 14. 发布朋友圈 =====
print("\n【14】发布朋友圈（写入本地DB）")
moment_content = f'[工作手机SDK测试] {time.strftime("%Y-%m-%d %H:%M:%S")} 自动发布'
r = test("post_moment", lambda: ex.postMoment(moment_content))
if r:
print(f" ✅ 发布结果: {json.dumps(r, ensure_ascii=False)[:200]}")
results['post_moment'] = r
time.sleep(1)
screenshot('after_post_moment')
# ===== 15. 视频号信息 =====
print("\n【15】视频号信息")
r = test("finder_info", lambda: ex.getFinderInfo())
if r:
print(f" ✅ 视频号: {json.dumps(r, ensure_ascii=False)[:300]}")
results['finder_info'] = r
# ===== 16. 视频号联系人 =====
print("\n【16】视频号联系人")
r = test("finder_contacts", lambda: ex.getFinderContacts(10))
if r:
rows = r.get('rows', [])
print(f" ✅ 视频号联系人: {len(rows)}个")
results['finder_contacts'] = {'count': len(rows)}
# ===== 17. 视频号账号 =====
print("\n【17】视频号账号")
r = test("finder_accounts", lambda: ex.getFinderAccounts())
if r:
print(f" ✅ 视频号账号: {json.dumps(r, ensure_ascii=False)[:200]}")
results['finder_accounts'] = r
# ===== 18. 收藏 =====
print("\n【18】收藏列表")
r = test("favorites", lambda: ex.getFavorites(10))
if r:
rows = r.get('rows', [])
print(f" ✅ 收藏: {len(rows)}条")
results['favorites'] = {'count': len(rows)}
# ===== 19. 钱包信息 =====
print("\n【19】钱包信息")
r = test("wallet_info", lambda: ex.getWalletInfo())
if r:
print(f" ✅ 钱包: {json.dumps(r, ensure_ascii=False)[:300]}")
results['wallet_info'] = r
# ===== 20. 钱包流水 =====
print("\n【20】钱包流水")
r = test("wallet_ledger", lambda: ex.getWalletLedger(10))
if r:
rows = r.get('rows', [])
print(f" ✅ 钱包流水: {len(rows)}条")
for l in rows[:2]:
print(f" - {str(l)[:100]}")
results['wallet_ledger'] = {'count': len(rows)}
# ===== 21. 红包记录 =====
print("\n【21】红包记录")
r = test("lucky_money", lambda: ex.getLuckyMoney(10))
if r:
rows = r.get('rows', [])
print(f" ✅ 红包: {len(rows)}条")
for l in rows[:2]:
print(f" - {str(l)[:100]}")
results['lucky_money'] = {'count': len(rows)}
# ===== 22. 图片统计 =====
print("\n【22】图片统计")
r = test("image_count", lambda: ex.getImageCount())
if r:
print(f" ✅ 图片: {json.dumps(r, ensure_ascii=False)}")
results['image_count'] = r
# ===== 23. 视频列表 =====
print("\n【23】视频列表")
r = test("videos", lambda: ex.getVideos(5))
if r:
rows = r.get('rows', [])
print(f" ✅ 视频: {len(rows)}条")
results['videos'] = {'count': len(rows)}
# ===== 24. 文件列表 =====
print("\n【24】文件列表")
r = test("files", lambda: ex.getFiles(5))
if r:
rows = r.get('rows', [])
print(f" ✅ 文件: {len(rows)}条")
results['files'] = {'count': len(rows)}
# ===== 25. 小程序 =====
print("\n【25】小程序信息")
r = test("appbrand", lambda: ex.getAppBrandInfo())
if r:
print(f" ✅ 小程序: {json.dumps(r, ensure_ascii=False)[:200]}")
results['appbrand'] = r
# ===== 26. 原始SQL =====
print("\n【26】原始SQL查询")
r = test("raw_sql", lambda: ex.rawSql('main', 'SELECT type, COUNT(*) as cnt FROM rcontact GROUP BY type ORDER BY cnt DESC LIMIT 10'))
if r:
print(f" ✅ SQL结果: {json.dumps(r, ensure_ascii=False)[:300]}")
results['raw_sql'] = r
# ===== 最终截图 =====
print("\n📸 最终截图...")
screenshot('final_state')
# ===== 汇总 =====
print("\n" + "=" * 70)
print(f"📊 验证结果: {passed_count}/{total_count} 通过 ({passed_count/total_count*100:.0f}%)")
print("=" * 70)
if errors:
print("\n❌ 失败项:")
for e in errors:
print(f" - {e}")
print("\n✅ 真实数据汇总:")
acc = results.get('account', {})
stats = results.get('stats', {})
cc = results.get('contact_count', {})
print(f" 📱 账号: {acc.get('nickname', '?')} | wxid: {acc.get('wxid', '?')}")
print(f" 👥 好友: {cc.get('friends', stats.get('friends', '?'))} | 总联系人: {cc.get('total', '?')}")
print(f" 💬 群组: {results.get('groups', {}).get('count', stats.get('groups', '?'))}")
print(f" 🏷️ 标签: {results.get('labels', {}).get('count', '?')}")
print(f" 📨 24h消息: {results.get('recent_messages', {}).get('count', '?')}")
print(f" 🖼️ 图片: {stats.get('images', '?')}")
print(f" 🎬 视频: {stats.get('videos', '?')}")
print(f" 📁 文件: {stats.get('files', '?')}")
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}/v9fix_verify_results.json'
with open(result_file, 'w', encoding='utf-8') as f:
json.dump({
'timestamp': int(time.time()),
'time': time.strftime('%Y-%m-%d %H:%M:%S'),
'passed': passed_count,
'total': total_count,
'pass_rate': f"{passed_count/total_count*100:.0f}%",
'results': results,
'errors': errors
}, f, ensure_ascii=False, indent=2)
print(f"\n💾 结果已保存: {result_file}")
return passed_count, total_count
if __name__ == '__main__':
passed, total = main()
print(f"\n🎯 最终: {passed}/{total} ({passed/total*100:.0f}%)")
sys.exit(0 if passed >= total * 0.8 else 1)