5919659930
nav_cache 模块单例；index 增加 Hub 链接与 hook 探测；SetupActivity 注册 Launcher。 Co-authored-by: Cursor <cursoragent@cursor.com>
443 lines
17 KiB
Python
443 lines
17 KiB
Python
"""
|
||
微信 UI 导航缓存系统 v1.0
|
||
—— 记录所有页面路径、元素坐标，避免重复探索，支持自动纠错
|
||
|
||
核心理念:
|
||
1. 首次探索时录制页面结构到 JSON（持久化）
|
||
2. 后续操作直接查缓存，跳过 uiautomator dump
|
||
3. 缓存失效（点击后页面不符预期）自动触发重新探索
|
||
4. 每个 action 的操作步骤作为 "路径模板" 存储
|
||
"""
|
||
|
||
import json
|
||
import time
|
||
import os
|
||
import re
|
||
import logging
|
||
from typing import Dict, Any, List, Optional, Tuple
|
||
from pathlib import Path
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CACHE_DIR = Path(__file__).parent.parent.parent / "data" / "nav_cache"
|
||
LOG_DIR = Path(__file__).parent.parent.parent / "data" / "operation_logs"
|
||
|
||
|
||
class PageSignature:
|
||
"""页面特征签名——用于快速验证当前在哪个页面"""
|
||
|
||
def __init__(self, name: str, must_have: List[str] = None,
|
||
must_not_have: List[str] = None, title: str = None):
|
||
self.name = name
|
||
self.must_have = must_have or []
|
||
self.must_not_have = must_not_have or []
|
||
self.title = title
|
||
|
||
def match(self, texts: List[str]) -> bool:
|
||
text_set = set(texts)
|
||
for t in self.must_have:
|
||
if t not in text_set:
|
||
return False
|
||
for t in self.must_not_have:
|
||
if t in text_set:
|
||
return False
|
||
if self.title and self.title not in text_set:
|
||
return False
|
||
return True
|
||
|
||
def to_dict(self):
|
||
return {
|
||
"name": self.name,
|
||
"must_have": self.must_have,
|
||
"must_not_have": self.must_not_have,
|
||
"title": self.title,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, d):
|
||
return cls(d["name"], d.get("must_have", []),
|
||
d.get("must_not_have", []), d.get("title"))
|
||
|
||
|
||
class NavStep:
|
||
"""单步导航操作"""
|
||
|
||
def __init__(self, action: str, target: str = None,
|
||
coords: List[int] = None, wait: float = 1.0,
|
||
verify_page: str = None, fallback: str = None):
|
||
self.action = action # tap_text / tap_coord / tap_tab / scroll_up / scroll_down / back / swipe
|
||
self.target = target # 文本目标 or tab名
|
||
self.coords = coords # [x, y] 精确坐标（缓存命中时用）
|
||
self.wait = wait
|
||
self.verify_page = verify_page # 期望到达的页面签名名
|
||
self.fallback = fallback # 失败后备用动作
|
||
|
||
def to_dict(self):
|
||
return {k: v for k, v in {
|
||
"action": self.action, "target": self.target,
|
||
"coords": self.coords, "wait": self.wait,
|
||
"verify_page": self.verify_page, "fallback": self.fallback,
|
||
}.items() if v is not None}
|
||
|
||
@classmethod
|
||
def from_dict(cls, d):
|
||
return cls(**d)
|
||
|
||
|
||
class NavPath:
|
||
"""一条完整的导航路径（从微信首页到目标页面）"""
|
||
|
||
def __init__(self, name: str, steps: List[NavStep] = None,
|
||
target_page: str = None):
|
||
self.name = name
|
||
self.steps = steps or []
|
||
self.target_page = target_page
|
||
self.success_count = 0
|
||
self.fail_count = 0
|
||
self.last_success = None
|
||
|
||
def to_dict(self):
|
||
return {
|
||
"name": self.name,
|
||
"steps": [s.to_dict() for s in self.steps],
|
||
"target_page": self.target_page,
|
||
"success_count": self.success_count,
|
||
"fail_count": self.fail_count,
|
||
"last_success": self.last_success,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, d):
|
||
p = cls(d["name"], [NavStep.from_dict(s) for s in d.get("steps", [])],
|
||
d.get("target_page"))
|
||
p.success_count = d.get("success_count", 0)
|
||
p.fail_count = d.get("fail_count", 0)
|
||
p.last_success = d.get("last_success")
|
||
return p
|
||
|
||
|
||
class WeChatNavCache:
|
||
"""导航缓存管理器"""
|
||
|
||
def __init__(self, wechat_version: str = "8.0.56",
|
||
screen_w: int = 1080, screen_h: int = 2400):
|
||
self.version = wechat_version
|
||
self.screen_w = screen_w
|
||
self.screen_h = screen_h
|
||
|
||
CACHE_DIR.mkdir(parents=True, exist_ok=True)
|
||
self.cache_file = CACHE_DIR / f"nav_v{wechat_version}_{screen_w}x{screen_h}.json"
|
||
|
||
self.pages: Dict[str, Dict] = {}
|
||
self.paths: Dict[str, NavPath] = {}
|
||
self.signatures: Dict[str, PageSignature] = {}
|
||
self.element_cache: Dict[str, Dict[str, Any]] = {}
|
||
|
||
self._load()
|
||
if not self.pages:
|
||
self._init_builtin()
|
||
|
||
def _load(self):
|
||
if self.cache_file.exists():
|
||
try:
|
||
data = json.loads(self.cache_file.read_text(encoding="utf-8"))
|
||
for name, page in data.get("pages", {}).items():
|
||
self.pages[name] = page
|
||
for name, path in data.get("paths", {}).items():
|
||
self.paths[name] = NavPath.from_dict(path)
|
||
for name, sig in data.get("signatures", {}).items():
|
||
self.signatures[name] = PageSignature.from_dict(sig)
|
||
self.element_cache = data.get("element_cache", {})
|
||
logger.info(f"导航缓存已加载: {len(self.pages)} 页面, "
|
||
f"{len(self.paths)} 路径, {len(self.element_cache)} 元素")
|
||
except Exception as e:
|
||
logger.warning(f"加载缓存失败: {e}")
|
||
|
||
def save(self):
|
||
data = {
|
||
"version": self.version,
|
||
"screen": f"{self.screen_w}x{self.screen_h}",
|
||
"updated": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"pages": self.pages,
|
||
"paths": {k: v.to_dict() for k, v in self.paths.items()},
|
||
"signatures": {k: v.to_dict() for k, v in self.signatures.items()},
|
||
"element_cache": self.element_cache,
|
||
}
|
||
self.cache_file.write_text(
|
||
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
def _init_builtin(self):
|
||
"""初始化内置的已知页面数据（基于真机录制 2026-03-14）"""
|
||
|
||
# ─── 底部 Tab 坐标（精确录制） ───
|
||
tab_y = 2148
|
||
self.element_cache["tabs"] = {
|
||
"微信": {"center": [135, tab_y], "bounds": [0, 2119, 270, 2176]},
|
||
"通讯录": {"center": [405, tab_y], "bounds": [270, 2119, 540, 2176]},
|
||
"发现": {"center": [675, tab_y], "bounds": [540, 2119, 810, 2176]},
|
||
"我": {"center": [945, tab_y], "bounds": [810, 2119, 1080, 2176]},
|
||
}
|
||
|
||
# ─── 页面特征签名 ───
|
||
sigs = {
|
||
"home": PageSignature("home", must_have=[], title=None),
|
||
"me": PageSignature("me", must_have=["服务", "收藏", "设置"]),
|
||
"settings": PageSignature("settings", must_have=["账号与安全"],
|
||
title="设置"),
|
||
"account_security": PageSignature(
|
||
"account_security", must_have=["微信号", "手机号", "微信密码"],
|
||
title="账号与安全"),
|
||
"safety_center": PageSignature(
|
||
"safety_center", must_have=["微信安全中心"]),
|
||
"contacts": PageSignature("contacts", must_have=["新的朋友"]),
|
||
"discover": PageSignature("discover", must_have=["朋友圈", "视频号"]),
|
||
}
|
||
for name, sig in sigs.items():
|
||
self.signatures[name] = sig
|
||
|
||
# ─── "我"页面元素坐标（精确录制） ───
|
||
self.pages["me"] = {
|
||
"recorded_at": "2026-03-14 16:32",
|
||
"elements": {
|
||
"头像区": {"center": [608, 298]},
|
||
"微信号": {"center": [643, 398], "text": "微信号：Lytiao1"},
|
||
"状态": {"center": [379, 507]},
|
||
"3个朋友": {"center": [676, 507]},
|
||
"服务": {"center": [197, 706]},
|
||
"收藏": {"center": [197, 883]},
|
||
"朋友圈": {"center": [220, 1037]},
|
||
"视频号": {"center": [220, 1191]},
|
||
"订单与卡包": {"center": [266, 1345]},
|
||
"表情": {"center": [197, 1499]},
|
||
"设置": {"center": [197, 1676]},
|
||
},
|
||
}
|
||
|
||
# ─── 设置页面元素坐标 ───
|
||
self.pages["settings"] = {
|
||
"recorded_at": "2026-03-14 16:32",
|
||
"needs_scroll_to_top": True,
|
||
"elements": {
|
||
"账号与安全": {"center": [250, 300], "note": "顶部滑后第一项"},
|
||
"青少年模式": {"center": [250, 479]},
|
||
"关怀模式": {"center": [158, 631]},
|
||
"新消息通知": {"center": [204, 800]},
|
||
"聊天": {"center": [112, 958]},
|
||
"通用": {"center": [158, 1112]},
|
||
"隐私": {"center": [89, 1257], "section_header": True},
|
||
"朋友权限": {"center": [112, 1423]},
|
||
"个人信息与权限": {"center": [135, 1574]},
|
||
"个人信息收集清单": {"center": [158, 1725]},
|
||
"第三方信息共享清单": {"center": [181, 1876]},
|
||
"插件": {"center": [89, 2018]},
|
||
"关于微信": {"center": [89, 2168]},
|
||
},
|
||
}
|
||
|
||
# ─── 账号与安全页面 ───
|
||
self.pages["account_security"] = {
|
||
"recorded_at": "2026-03-14 16:33",
|
||
"elements": {
|
||
"微信号": {"center": [112, 300], "value_center": [603, 300]},
|
||
"手机号": {"center": [112, 451], "value_center": [603, 451]},
|
||
"微信密码": {"center": [135, 625]},
|
||
"声音锁": {"center": [112, 778]},
|
||
"应急联系人": {"center": [158, 953]},
|
||
"登录过的设备": {"center": [181, 1104]},
|
||
"更多安全设置": {"center": [181, 1257]},
|
||
"微信安全中心": {"center": [513, 1423]},
|
||
},
|
||
}
|
||
|
||
# ─── 常用导航路径 ───
|
||
self.paths["to_me"] = NavPath("to_me", [
|
||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||
], "me")
|
||
|
||
self.paths["to_settings"] = NavPath("to_settings", [
|
||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||
NavStep("scroll_up", None, None, 0.5),
|
||
NavStep("tap_text", "设置", [197, 1676], 2.0, "settings"),
|
||
NavStep("scroll_to_top", None, None, 0.5),
|
||
], "settings")
|
||
|
||
self.paths["to_account_security"] = NavPath("to_account_security", [
|
||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||
NavStep("scroll_up", None, None, 0.5),
|
||
NavStep("tap_text", "设置", [197, 1676], 2.0, "settings"),
|
||
NavStep("scroll_to_top", None, None, 0.5),
|
||
NavStep("tap_text", "账号与安全", [250, 300], 2.0, "account_security"),
|
||
], "account_security")
|
||
|
||
self.paths["to_safety_center"] = NavPath("to_safety_center", [
|
||
NavStep("tap_tab", "我", [945, tab_y], 1.5, "me"),
|
||
NavStep("scroll_up", None, None, 0.5),
|
||
NavStep("tap_text", "设置", [197, 1676], 2.0, "settings"),
|
||
NavStep("scroll_to_top", None, None, 0.5),
|
||
NavStep("tap_text", "账号与安全", [250, 300], 2.0, "account_security"),
|
||
NavStep("tap_text", "微信安全中心", [513, 1423], 3.0, "safety_center"),
|
||
], "safety_center")
|
||
|
||
self.paths["to_discover"] = NavPath("to_discover", [
|
||
NavStep("tap_tab", "发现", [675, tab_y], 1.5, "discover"),
|
||
], "discover")
|
||
|
||
self.paths["to_contacts"] = NavPath("to_contacts", [
|
||
NavStep("tap_tab", "通讯录", [405, tab_y], 1.5, "contacts"),
|
||
], "contacts")
|
||
|
||
self.save()
|
||
logger.info("内置导航缓存已初始化并保存")
|
||
|
||
def get_tab_coord(self, tab_name: str) -> Optional[List[int]]:
|
||
tabs = self.element_cache.get("tabs", {})
|
||
if tab_name in tabs:
|
||
return tabs[tab_name]["center"]
|
||
return None
|
||
|
||
def get_element_coord(self, page: str, element: str) -> Optional[List[int]]:
|
||
page_data = self.pages.get(page, {})
|
||
elements = page_data.get("elements", {})
|
||
if element in elements:
|
||
return elements[element].get("center")
|
||
return None
|
||
|
||
def get_path(self, path_name: str) -> Optional[NavPath]:
|
||
return self.paths.get(path_name)
|
||
|
||
def record_page(self, page_name: str, elements: Dict[str, Dict]):
|
||
"""录制/更新一个页面的元素数据"""
|
||
self.pages[page_name] = {
|
||
"recorded_at": time.strftime("%Y-%m-%d %H:%M"),
|
||
"elements": elements,
|
||
}
|
||
self.save()
|
||
logger.info(f"页面 [{page_name}] 已录制: {len(elements)} 个元素")
|
||
|
||
def record_path_result(self, path_name: str, success: bool):
|
||
"""记录路径执行结果"""
|
||
path = self.paths.get(path_name)
|
||
if path:
|
||
if success:
|
||
path.success_count += 1
|
||
path.last_success = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
else:
|
||
path.fail_count += 1
|
||
self.save()
|
||
|
||
def update_element_coord(self, page: str, element: str,
|
||
new_center: List[int]):
|
||
"""自动纠错：更新元素坐标"""
|
||
if page not in self.pages:
|
||
self.pages[page] = {"elements": {}}
|
||
if "elements" not in self.pages[page]:
|
||
self.pages[page]["elements"] = {}
|
||
if element not in self.pages[page]["elements"]:
|
||
self.pages[page]["elements"][element] = {}
|
||
old = self.pages[page]["elements"][element].get("center")
|
||
self.pages[page]["elements"][element]["center"] = new_center
|
||
self.pages[page]["elements"][element]["updated_at"] = \
|
||
time.strftime("%Y-%m-%d %H:%M")
|
||
self.save()
|
||
logger.info(f"坐标纠错: [{page}].{element} {old} -> {new_center}")
|
||
|
||
|
||
class OperationLogger:
|
||
"""操作日志记录器——每次操作自动记录步骤/耗时/结果"""
|
||
|
||
def __init__(self, device_id: str = ""):
|
||
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
||
self.device_id = device_id
|
||
self.log_file = LOG_DIR / f"ops_{time.strftime('%Y%m%d')}.jsonl"
|
||
self._current_op = None
|
||
|
||
def start_operation(self, action: str, params: dict):
|
||
self._current_op = {
|
||
"action": action,
|
||
"params": {k: v for k, v in params.items()
|
||
if k not in ("password", "pwd", "new_pwd", "old_pwd")},
|
||
"device": self.device_id,
|
||
"started_at": time.strftime("%Y-%m-%d %H:%M:%S"),
|
||
"start_ts": time.time(),
|
||
"steps": [],
|
||
}
|
||
|
||
def log_step(self, step_type: str, detail: str = "",
|
||
coords: List[int] = None, success: bool = True):
|
||
if not self._current_op:
|
||
return
|
||
self._current_op["steps"].append({
|
||
"type": step_type,
|
||
"detail": detail,
|
||
"coords": coords,
|
||
"success": success,
|
||
"ts": round(time.time() - self._current_op["start_ts"], 2),
|
||
})
|
||
|
||
def end_operation(self, success: bool, result: dict = None,
|
||
error: str = None):
|
||
if not self._current_op:
|
||
return
|
||
self._current_op["success"] = success
|
||
self._current_op["duration_ms"] = round(
|
||
(time.time() - self._current_op["start_ts"]) * 1000)
|
||
self._current_op["ended_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
|
||
if result:
|
||
self._current_op["result_summary"] = str(result)[:200]
|
||
if error:
|
||
self._current_op["error"] = error
|
||
|
||
del self._current_op["start_ts"]
|
||
|
||
try:
|
||
with open(self.log_file, "a", encoding="utf-8") as f:
|
||
f.write(json.dumps(self._current_op, ensure_ascii=False) + "\n")
|
||
except Exception as e:
|
||
logger.warning(f"写入操作日志失败: {e}")
|
||
|
||
op = self._current_op
|
||
self._current_op = None
|
||
return op
|
||
|
||
def get_recent_ops(self, limit: int = 20) -> List[dict]:
|
||
"""读取最近的操作记录"""
|
||
if not self.log_file.exists():
|
||
return []
|
||
ops = []
|
||
try:
|
||
with open(self.log_file, "r", encoding="utf-8") as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line:
|
||
ops.append(json.loads(line))
|
||
except Exception:
|
||
pass
|
||
return ops[-limit:]
|
||
|
||
def get_action_stats(self) -> Dict[str, Dict]:
|
||
"""统计每个 action 的成功率和平均耗时"""
|
||
ops = self.get_recent_ops(1000)
|
||
stats: Dict[str, Dict] = {}
|
||
for op in ops:
|
||
action = op.get("action", "unknown")
|
||
if action not in stats:
|
||
stats[action] = {"total": 0, "success": 0,
|
||
"total_ms": 0, "errors": []}
|
||
stats[action]["total"] += 1
|
||
if op.get("success"):
|
||
stats[action]["success"] += 1
|
||
stats[action]["total_ms"] += op.get("duration_ms", 0)
|
||
if op.get("error"):
|
||
stats[action]["errors"].append(op["error"][:50])
|
||
|
||
for action, s in stats.items():
|
||
s["success_rate"] = (
|
||
f"{s['success']/s['total']*100:.0f}%" if s["total"] else "N/A")
|
||
s["avg_ms"] = (
|
||
s["total_ms"] // s["total"] if s["total"] else 0)
|
||
return stats
|
||
|
||
|
||
# 模块级单例，供 wechat_adb_engine 等导入
|
||
nav_cache = WeChatNavCache()
|