jackyu66gitandCursor
53eb4577b3
对齐测测交互：首页「+」支持邀请填档案/添加档案/邀请合盘；各 Tab 与工具页按同一视觉与功能标准落地；本地对照截图目录显式忽略。 Co-authored-by: Cursor <cursoragent@cursor.com>
1959 lines
67 KiB
Python
Executable File
1959 lines
67 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Module-whitelist deep crawl for CeCe (depth 3–4).
|
||
|
||
Shallow full crawl (cece_capture_all.py) only hits L1/L2.
|
||
This script walks known product modules with explicit child labels,
|
||
shots paywalls then backs out, and skips UGC/feed noise.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
|
||
ROOT = Path("/Users/jack/Project/digital-psychology")
|
||
TOOLS = ROOT / "tools" / "android"
|
||
ANDROID_HOME = Path("/opt/homebrew/share/android-commandlinetools")
|
||
PKG = "com.xxwolo.cc5"
|
||
ACT = f"{PKG}/com.cece.app.MainActivity"
|
||
STAMP = time.strftime("%Y%m%d-%H%M%S")
|
||
OUT = ROOT / ".tmp" / "cece-validation" / time.strftime("%Y%m%d") / f"deep-{STAMP}"
|
||
NOTES = OUT / "NOTES.md"
|
||
MANIFEST = OUT / "manifest.json"
|
||
|
||
os.environ["ANDROID_HOME"] = str(ANDROID_HOME)
|
||
os.environ["ANDROID_SDK_ROOT"] = str(ANDROID_HOME)
|
||
os.environ["PATH"] = (
|
||
f"{ANDROID_HOME}/emulator:{ANDROID_HOME}/platform-tools:"
|
||
f"/opt/homebrew/bin:/usr/bin:/bin:" + os.environ.get("PATH", "")
|
||
)
|
||
|
||
# ─── module whitelist ───────────────────────────────────────────────
|
||
# entry: ("home"|"more"|"mine"|"tab", label)
|
||
# children: labels allowed at every depth (exact first-line match preferred)
|
||
# tabs: always try as horizontal/top chips when visible
|
||
# max_depth: L1=entry shot, L2=children, L3/L4=deeper
|
||
|
||
COMMON_DEEP = [
|
||
"开始", "开始测试", "立即开始", "立即体验", "免费体验", "继续", "下一步",
|
||
"查看详情", "深度解读", "解锁", "示例", "示例报告",
|
||
"添加档案", "选择档案", "新建档案", "编辑档案", "直接选择档案合盘",
|
||
"今日", "本周", "本月", "日运", "周运", "月运",
|
||
"概览", "详情", "解读",
|
||
]
|
||
|
||
ZODIAC = [
|
||
"白羊座", "金牛座", "双子座", "巨蟹座", "狮子座", "处女座",
|
||
"天秤座", "天蝎座", "射手座", "摩羯座", "水瓶座", "双鱼座",
|
||
]
|
||
|
||
CHART_TABS = ["天象", "本命", "行运", "三限", "次限", "日返", "月返", "法达", "推运", "合盘"]
|
||
NATAL_INNER = ["概览", "行星", "宫位", "相位", "解读", "深度解读"]
|
||
SYNASTRY_INNER = [
|
||
"添加档案", "直接选择档案合盘", "示例", "恋爱", "友情", "亲子",
|
||
"比较盘", "组合盘", "组合中点", "时空中点", "组合星座", "推运",
|
||
"查看报告", "深度报告", "邀请合盘",
|
||
]
|
||
|
||
MODULES = [
|
||
{
|
||
"id": "xingpan",
|
||
"entry": ("home", "星盘"),
|
||
"max_depth": 4,
|
||
"tabs": CHART_TABS,
|
||
"children": CHART_TABS + NATAL_INNER + COMMON_DEEP + [
|
||
"档案", "自己", "基础解读", "Deepseek解读", "本命盘", "现代", "参数",
|
||
"日", "周", "月", "测测达人在线解读星盘",
|
||
],
|
||
},
|
||
{
|
||
"id": "xingzuo",
|
||
"entry": ("home", "星座"),
|
||
"max_depth": 4,
|
||
# Hub 解读页：无障碍常泄漏成首页；勿用首页心情分「爱情/事业」当 Tab
|
||
"tabs": ["查看星盘", "简单模式"],
|
||
"fallback_xy": {
|
||
"查看星盘": (920, 420),
|
||
"简单模式": (200, 420),
|
||
},
|
||
"children": ["星座密码", "整体分析一下我的星图", "...全文"] + ZODIAC[:4] + COMMON_DEEP,
|
||
},
|
||
{
|
||
"id": "shengchen",
|
||
"entry": ("home", "生辰"),
|
||
"max_depth": 4,
|
||
# Hub Compose 页 a11y 常空/泄漏首页；用实测坐标深挖
|
||
"tabs": ["查看生辰历", "Ai 解读", "校正"],
|
||
"fallback_xy": {
|
||
"查看生辰历": (920, 420),
|
||
"Ai 解读": (930, 700),
|
||
"校正": (280, 1080),
|
||
},
|
||
"children": [
|
||
"合盘", "流盘", "参数", "设置", "开始校正", "直接选择类型", "...全文",
|
||
] + COMMON_DEEP,
|
||
},
|
||
{
|
||
"id": "hepan",
|
||
"entry": ("home", "缘分合盘"),
|
||
"max_depth": 4,
|
||
"tabs": [],
|
||
"children": SYNASTRY_INNER + COMMON_DEEP + ZODIAC[:4],
|
||
},
|
||
{
|
||
"id": "ziwei",
|
||
"entry": ("home", "紫微"),
|
||
"max_depth": 4,
|
||
"tabs": ["身宫", "命宫", "来因宫", "官禄宫", "迁移宫", "财帛宫", "夫妻宫"],
|
||
"fallback_xy": {
|
||
"查看紫微历": (920, 480),
|
||
"身宫": (200, 720),
|
||
"命宫": (540, 720),
|
||
"来因宫": (880, 720),
|
||
"官禄宫": (200, 980),
|
||
"迁移宫": (420, 980),
|
||
"财帛宫": (660, 980),
|
||
"夫妻宫": (880, 980),
|
||
},
|
||
"children": ["查看紫微历", "...全文"] + COMMON_DEEP,
|
||
},
|
||
{
|
||
"id": "iren",
|
||
"entry": ("home", "I人E人"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": COMMON_DEEP + ["开始测试", "重新测试"],
|
||
},
|
||
{
|
||
"id": "xiaoxing",
|
||
"entry": ("home", "陪伴小星"),
|
||
"max_depth": 3,
|
||
"tabs": ["聊天", "日记", "成长"],
|
||
"children": COMMON_DEEP + ["聊天", "日记", "成长", "设置", "语音"],
|
||
},
|
||
{
|
||
"id": "qingsu",
|
||
"entry": ("home", "倾诉"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": COMMON_DEEP + ["开始倾诉", "匹配达人", "语音通话", "文字"],
|
||
},
|
||
{
|
||
"id": "zhihuika",
|
||
"entry": ("home", "智慧卡"),
|
||
"max_depth": 3,
|
||
"tabs": ["智慧卡", "三张牌", "深度解读"],
|
||
"children": COMMON_DEEP + ["智慧卡", "三张牌", "深度解读", "灵魂伴侣", "抽卡", "再抽一次"],
|
||
},
|
||
{
|
||
"id": "xingpan_report",
|
||
"entry": ("home", "星盘报告"),
|
||
"max_depth": 3,
|
||
"tabs": ["全部", "解读", "情感", "财富", "自我探索"],
|
||
"children": COMMON_DEEP + ["全部", "解读", "情感", "财富", "自我探索", "购买", "示例", "目录"],
|
||
},
|
||
{
|
||
"id": "shengchenli",
|
||
"entry": ("home", "生辰历"),
|
||
"max_depth": 3,
|
||
"tabs": ["生辰", "合盘", "流盘"],
|
||
"children": COMMON_DEEP + ["生辰", "合盘", "流盘", "参数", "设置", "星盘", "紫微", "政余"],
|
||
},
|
||
{
|
||
"id": "linghun",
|
||
"entry": ("home", "灵魂伴侣"),
|
||
"max_depth": 3,
|
||
"tabs": ["男生", "女生", "不限"],
|
||
"children": COMMON_DEEP + ["男生", "女生", "不限", "开始匹配", "查看报告"],
|
||
},
|
||
{
|
||
"id": "ai_plaza",
|
||
"entry": ("home", "AI玩法广场"),
|
||
"max_depth": 3,
|
||
"tabs": ["热门", "上新", "测试", "心动站", "工具"],
|
||
"children": ["搜索", "七宗罪与七美德", "DIY水晶手串", "印盘", "人生说明书", "正缘画像"] + COMMON_DEEP,
|
||
},
|
||
# from 更多
|
||
{
|
||
"id": "ceshi",
|
||
"entry": ("more", "测试"),
|
||
"max_depth": 4,
|
||
"tabs": ["全部", "情感", "职场", "自我", "健康", "趣味"],
|
||
"children": COMMON_DEEP + [
|
||
"全部", "情感", "职场", "自我", "健康", "趣味",
|
||
"MBTI", "九型人格", "开始测试", "继续答题", "提交",
|
||
],
|
||
},
|
||
{
|
||
"id": "shapan",
|
||
"entry": ("more", "沙盘"),
|
||
"max_depth": 3,
|
||
# Unity Game view：无障碍几乎为空，靠底部工具栏坐标
|
||
"tabs": ["完成", "道具", "保存"],
|
||
"fallback_xy": {
|
||
"完成": (180, 2200),
|
||
"道具": (400, 2200),
|
||
"保存": (620, 2200),
|
||
"解读": (840, 2200),
|
||
},
|
||
"children": COMMON_DEEP + ["完成", "道具", "保存", "解读"],
|
||
},
|
||
{
|
||
"id": "moodtown",
|
||
"entry": ("more", "心情小镇"),
|
||
"max_depth": 3,
|
||
# 「和我聊聊」按钮无障碍缺失；按角色名 Y+230 实测
|
||
"tabs": ["甄心", "志远", "明朗"],
|
||
"fallback_xy": {
|
||
"甄心": (900, 1118),
|
||
"志远": (900, 1665),
|
||
"明朗": (900, 2210),
|
||
},
|
||
"children": [],
|
||
},
|
||
{
|
||
"id": "xingxiu",
|
||
"entry": ("more", "星宿"),
|
||
"max_depth": 3,
|
||
"tabs": ["本命星宿", "值日星宿", "星宿关系", "全部"],
|
||
"children": COMMON_DEEP + ["本命星宿", "值日星宿", "星宿关系", "全部", "...全文"],
|
||
},
|
||
{
|
||
"id": "aiqingshu",
|
||
"entry": ("more", "爱情树"),
|
||
"max_depth": 3,
|
||
"tabs": ["开始测试", "重新测试"],
|
||
"children": COMMON_DEEP + [
|
||
"开始测试", "重新测试", "伴侣相处阻力", "成人依恋类型", "家谱图",
|
||
"依恋类型", "完成测试题目后即可查看详细解读",
|
||
],
|
||
},
|
||
{
|
||
"id": "liaotian",
|
||
"entry": ("more", "聊天分析"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": COMMON_DEEP + ["上传", "开始分析", "示例"],
|
||
},
|
||
{
|
||
"id": "guanxiwang",
|
||
"entry": ("more", "关系网"),
|
||
"max_depth": 3,
|
||
"tabs": ["添加", "编辑"],
|
||
"fallback_xy": {
|
||
"添加": (980, 200),
|
||
"编辑": (880, 200),
|
||
},
|
||
"children": COMMON_DEEP + ["添加", "编辑"],
|
||
},
|
||
{
|
||
"id": "jiedu",
|
||
# 「更多」里的「解读」已变成分区标题不可点；达人解读实际在底栏「在线」
|
||
"entry": ("tab", "在线"),
|
||
"max_depth": 3,
|
||
"tabs": ["关注", "全部", "私人专线", "综合排序", "筛选", "甄选达人"],
|
||
"fallback_xy": {
|
||
"关注": (101, 205),
|
||
"全部": (274, 205),
|
||
"私人专线": (487, 205),
|
||
"综合排序": (125, 309),
|
||
"筛选": (990, 309),
|
||
"甄选达人": (154, 400),
|
||
},
|
||
"children": [],
|
||
},
|
||
{
|
||
"id": "xingzhi",
|
||
"entry": ("more", "星骰"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": COMMON_DEEP + ["投掷", "再来一次"],
|
||
},
|
||
{
|
||
"id": "rishu",
|
||
"entry": ("more", "灵数"),
|
||
"max_depth": 3,
|
||
"tabs": ["财富", "忠诚", "远见", "挑剔"],
|
||
"children": COMMON_DEEP + ["财富", "忠诚", "远见", "挑剔", "...全文"],
|
||
},
|
||
{
|
||
"id": "shengxiao",
|
||
"entry": ("more", "生肖"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": COMMON_DEEP + [
|
||
"鼠", "牛", "虎", "兔", "龙", "蛇", "马", "羊", "猴", "鸡", "狗", "猪",
|
||
],
|
||
},
|
||
{
|
||
"id": "mayan",
|
||
"entry": ("more", "玛雅图腾"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": COMMON_DEEP,
|
||
},
|
||
{
|
||
"id": "renleitu",
|
||
"entry": ("more", "人类图"),
|
||
"max_depth": 3,
|
||
"tabs": ["设计"],
|
||
"children": COMMON_DEEP + ["设计", "...全文"],
|
||
},
|
||
{
|
||
"id": "ziweili",
|
||
"entry": ("more", "紫微历"),
|
||
"max_depth": 3,
|
||
"tabs": ["大限", "参数", "设置"],
|
||
"children": COMMON_DEEP + ["大限", "参数", "设置", "4~13岁", "14~23岁", "24~33岁"],
|
||
},
|
||
{
|
||
"id": "rili",
|
||
"entry": ("more", "日历"),
|
||
"max_depth": 3,
|
||
# 点一个日期进详情
|
||
"tabs": ["立秋", "2", "3", "7"],
|
||
"children": COMMON_DEEP + ["立秋"],
|
||
},
|
||
# 我的关键子树
|
||
{
|
||
"id": "mine_dangan",
|
||
"entry": ("mine", "档案"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": COMMON_DEEP + ["添加", "新建", "编辑", "删除", "自己"],
|
||
},
|
||
{
|
||
"id": "mine_huiyuan",
|
||
"entry": ("mine", "VIP会员陪伴中"),
|
||
"max_depth": 4,
|
||
"tabs": ["VIP", "SVIP", "记录"],
|
||
"children": [
|
||
"VIP", "SVIP", "记录", "开通", "续费", "活动须知", "遇到问题?",
|
||
"连续包月", "连续包季", "连续包年", "12个月",
|
||
"工具内容解读权益", "确认并支付", "首次开通送",
|
||
"新用户限时特惠", "开启一段更懂你的陪伴", "升级SVIP",
|
||
] + COMMON_DEEP,
|
||
"entry_alts": [
|
||
"VIP会员陪伴中", "升级SVIP", "¥9.9开启陪伴", "新用户限时特惠",
|
||
"开通会员", "VIP会员", "测测会员", "会员",
|
||
],
|
||
},
|
||
{
|
||
"id": "mine_duihuan",
|
||
"entry": ("mine", "兑换码"),
|
||
"max_depth": 2,
|
||
"tabs": [],
|
||
"children": ["兑换", "确定", "取消"] + COMMON_DEEP,
|
||
},
|
||
{
|
||
"id": "mine_dingdan",
|
||
"entry": ("mine_mall_order", "全部订单"),
|
||
"max_depth": 3,
|
||
"tabs": ["全部", "待付款", "待收货", "退款售后"],
|
||
"fallback_xy": {
|
||
"全部": (150, 280),
|
||
"待付款": (380, 280),
|
||
"待收货": (620, 280),
|
||
"退款售后": (900, 280),
|
||
},
|
||
"children": ["全部", "待付款", "待收货", "退款售后", "全部订单", "商城订单", "详情"] + COMMON_DEEP,
|
||
},
|
||
{
|
||
"id": "mine_shezhi",
|
||
"entry": ("mine_gear", "设置"),
|
||
"max_depth": 3,
|
||
"tabs": [],
|
||
"children": [
|
||
"个人资料", "账号安全", "免密支付", "通知设置", "隐私设置", "语言设置",
|
||
"智能体设置", "黑名单", "未成年人模式", "夜间模式", "震动提醒", "推荐给好友",
|
||
],
|
||
},
|
||
{
|
||
"id": "mine_baogao",
|
||
"entry": ("mine", "报告与测试"),
|
||
"max_depth": 3,
|
||
"tabs": ["已购买", "我的报告", "我的测试"],
|
||
"children": ["已购买", "我的报告", "我的测试", "购买须知"] + COMMON_DEEP,
|
||
"entry_alts": ["报告与测试", "报告", "我的报告", "0\n报告与测试"],
|
||
},
|
||
{
|
||
"id": "mine_ceshi",
|
||
"entry": ("mine", "测试"),
|
||
"max_depth": 3,
|
||
"tabs": ["我的测试", "我的报告", "已购买"],
|
||
"children": ["我的测试", "我的报告", "已购买", "继续", "重新测试"] + COMMON_DEEP,
|
||
"entry_alts": ["我的测试", "报告与测试"],
|
||
},
|
||
]
|
||
|
||
# 底栏 Tab（补齐）
|
||
MODULES = [
|
||
{
|
||
"id": "tab_message",
|
||
"entry": ("tab", "消息"),
|
||
"max_depth": 2,
|
||
"tabs": [],
|
||
# 不进小星/互动聊天，只截系统通知
|
||
"children": ["系统通知"],
|
||
},
|
||
{
|
||
"id": "tab_ask",
|
||
"entry": ("tab", "问"),
|
||
"max_depth": 2,
|
||
# 不截真人1v1
|
||
"tabs": ["测测AI"],
|
||
"children": ["测测AI", "工具", "深度解读"],
|
||
},
|
||
{
|
||
"id": "tab_online",
|
||
"entry": ("tab", "在线"),
|
||
"max_depth": 2,
|
||
"tabs": [],
|
||
"children": ["综合排序", "筛选", "全部工具"],
|
||
},
|
||
] + MODULES
|
||
|
||
BOTTOM_TABS = {"首页", "消息", "在线", "我的", "问"}
|
||
TAB_FALLBACK = {
|
||
"首页": (108, 2255),
|
||
"消息": (324, 2255),
|
||
"问": (540, 2255),
|
||
"在线": (756, 2255),
|
||
"我的": (972, 2255),
|
||
}
|
||
|
||
PAYWALL_KEYS = (
|
||
"确认并支付", "专享价购买", "微信支付", "支付宝支付", "苹果支付",
|
||
"立即支付", "连续包月", "确认协议并支付", "开通SVIP", "向TA提问",
|
||
)
|
||
|
||
BLACKLIST_EXACT = {
|
||
"首页", "消息", "在线", "我的", "问", "推荐", "工具", "更多",
|
||
"点赞", "评论", "关注", "粉丝", "私信", "直播",
|
||
"最热", "最新", "全部评论",
|
||
}
|
||
BLACKLIST_RE = re.compile(
|
||
r"^(\d+|U\d+|第\s*\d+|$|"
|
||
r".{0,2}穿搭|梨形|财神爷|米咪嘟|竹木鑫|赵莹|天使|"
|
||
r"广告|下载|打开测测|"
|
||
r"全部评论|条评论|写评论|查看评论)"
|
||
)
|
||
|
||
MAX_CHILDREN_PER_PAGE = 16
|
||
MAX_SHOTS = 600
|
||
|
||
# Long screenshot: finger swipe UP (往上滑) to reveal content below, then stitch
|
||
LONG_MAX_SCROLLS = 6 # extra frames after the first viewport
|
||
LONG_BOTTOM_CROP = 260 # drop bottom tab bar from stitch frames
|
||
LONG_TOP_CROP_CONT = 160 # drop status/sticky header on continuation frames
|
||
LONG_MIN_OVERLAP = 60
|
||
LONG_MAX_OVERLAP = 520
|
||
LONG_SCROLL_BACK = 5 # after stitch, pull page back to top (finger down)
|
||
|
||
# 上滑长图时折叠块入口（点开后再截）
|
||
EXPAND_LABELS = (
|
||
"展开更多",
|
||
"展开全部",
|
||
"查看更多",
|
||
"点击展开",
|
||
"显示更多",
|
||
"展开",
|
||
)
|
||
|
||
|
||
# ─── adb / ui helpers ───────────────────────────────────────────────
|
||
|
||
def sh(args, timeout=120):
|
||
try:
|
||
return subprocess.run(args, capture_output=True, text=True, timeout=timeout)
|
||
except subprocess.TimeoutExpired:
|
||
print(f" TIMEOUT {' '.join(str(a) for a in args[:4])}", flush=True)
|
||
return subprocess.CompletedProcess(args, 124, "", "timeout")
|
||
|
||
|
||
def adb(*args, timeout=30):
|
||
return sh(["adb", *args], timeout=timeout)
|
||
|
||
|
||
def tool(*args, timeout=25):
|
||
r = sh([str(TOOLS), *args, "--json"], timeout=timeout)
|
||
try:
|
||
return json.loads(r.stdout or "{}")
|
||
except Exception:
|
||
return {"error": (r.stderr or "")[:300]}
|
||
|
||
|
||
def fg_pkg() -> str:
|
||
r = adb("shell", "dumpsys", "window", timeout=15)
|
||
m = re.search(r"mCurrentFocus=Window\{[^ ]+ u0 ([^/\}]+)", r.stdout or "")
|
||
return m.group(1) if m else ""
|
||
|
||
|
||
def ensure_app():
|
||
if PKG not in fg_pkg():
|
||
print("RELAUNCH", fg_pkg(), flush=True)
|
||
adb("shell", "am", "start", "-n", ACT, timeout=15)
|
||
time.sleep(5)
|
||
|
||
|
||
def dump():
|
||
"""UI dump with short timeout; never hang the crawl."""
|
||
try:
|
||
data = tool("ui", "dump", timeout=18)
|
||
if isinstance(data, dict) and data.get("elements") is not None:
|
||
return data
|
||
except Exception as ex:
|
||
print(f" dump err: {ex}", flush=True)
|
||
return {"elements": []}
|
||
|
||
|
||
def lab(e):
|
||
return (e.get("contentDesc") or e.get("text") or "").strip()
|
||
|
||
|
||
def labels(data=None):
|
||
data = data or dump()
|
||
return [lab(e) for e in (data.get("elements") or []) if lab(e)]
|
||
|
||
|
||
def blob(data=None):
|
||
return "\n".join(labels(data))
|
||
|
||
|
||
def area(e):
|
||
b = e.get("bounds") or {}
|
||
return int(b.get("width") or 0) * int(b.get("height") or 0)
|
||
|
||
|
||
def center(e):
|
||
c = e.get("center")
|
||
if isinstance(c, dict):
|
||
return int(c["x"]), int(c["y"])
|
||
return None
|
||
|
||
|
||
def first_line(t: str) -> str:
|
||
return t.split("\n", 1)[0].strip()
|
||
|
||
|
||
def safe_name(s: str) -> str:
|
||
return re.sub(r"[^\w\u4e00-\u9fff\-]+", "_", s)[:40]
|
||
|
||
|
||
def tap_xy(x, y, wait=1.4):
|
||
adb("shell", "input", "tap", str(x), str(y))
|
||
time.sleep(wait)
|
||
|
||
|
||
def back(n=1):
|
||
for _ in range(n):
|
||
adb("shell", "input", "keyevent", "4")
|
||
time.sleep(0.65)
|
||
if PKG not in fg_pkg():
|
||
print(" back exited app → relaunch", flush=True)
|
||
ensure_app()
|
||
break
|
||
|
||
|
||
def swipe_up():
|
||
"""手指往上滑 → 页面内容上移，露出下方。
|
||
|
||
起点避开底部手势条（过低易触发 Home / 返回）。
|
||
"""
|
||
adb("shell", "input", "swipe", "540", "1680", "540", "720", "380")
|
||
time.sleep(0.95)
|
||
|
||
|
||
def swipe_down():
|
||
"""手指往下滑 → 回到上方（仅用于长图后复位）。"""
|
||
adb("shell", "input", "swipe", "540", "750", "540", "1850", "380")
|
||
time.sleep(0.7)
|
||
|
||
|
||
def scroll_page_top(n: int = 3):
|
||
"""After long shot, gently return toward top (fewer swipes = less stall)."""
|
||
for _ in range(n):
|
||
swipe_down()
|
||
|
||
|
||
def expand_more_if_present(max_taps: int = 3) -> int:
|
||
"""上滑过程中若出现「展开更多」等按钮，点开再继续截。返回点击次数。"""
|
||
taps = 0
|
||
for _ in range(max_taps):
|
||
data = dump()
|
||
blob_txt = "\n".join(lab(e) for e in (data.get("elements") or []) if lab(e))
|
||
if not any(k in blob_txt for k in ("展开更多", "展开全部", "查看更多", "显示更多", "点击展开")) and "展开" not in blob_txt:
|
||
break
|
||
hit = None
|
||
for e in data.get("elements") or []:
|
||
t = lab(e)
|
||
if not t:
|
||
continue
|
||
fl = first_line(t)
|
||
if fl in ("收起", "收起更多", "收起全部"):
|
||
continue
|
||
matched = fl in EXPAND_LABELS or any(
|
||
k in fl for k in ("展开更多", "展开全部", "查看更多", "显示更多", "点击展开")
|
||
)
|
||
if not matched and fl == "展开":
|
||
matched = True
|
||
if not matched:
|
||
continue
|
||
xy = center(e)
|
||
if not xy or xy[1] < 200 or xy[1] > 2150:
|
||
continue
|
||
hit = (fl, xy)
|
||
break
|
||
if not hit:
|
||
break
|
||
name, xy = hit
|
||
print(f" long: 展开 '{name}' {xy}", flush=True)
|
||
tap_xy(*xy, wait=1.2)
|
||
taps += 1
|
||
time.sleep(0.35)
|
||
return taps
|
||
|
||
|
||
def swipe_left_mid():
|
||
adb("shell", "input", "swipe", "900", "1050", "200", "1050", "280")
|
||
time.sleep(0.8)
|
||
|
||
|
||
def swipe_right_mid():
|
||
"""Reset home tool carousel to first page (I人E人 side)."""
|
||
adb("shell", "input", "swipe", "200", "1050", "900", "1050", "280")
|
||
time.sleep(0.8)
|
||
|
||
|
||
def fingerprint(labs: list[str]) -> str:
|
||
# include head + mid content so tab chips alone don't collapse distinct pages
|
||
fls = [first_line(x) for x in labs]
|
||
key = "|".join(fls[:18] + fls[18:36])
|
||
return hashlib.md5(key.encode()).hexdigest()[:12]
|
||
|
||
|
||
def is_blacklisted(name: str) -> bool:
|
||
if name in BLACKLIST_EXACT or name in BOTTOM_TABS:
|
||
return True
|
||
if BLACKLIST_RE.search(name):
|
||
return True
|
||
if len(name) > 16 and not any(k in name for k in ("报告", "测试", "解读", "档案", "合盘")):
|
||
return True
|
||
return False
|
||
|
||
|
||
def allowed(name: str, allow: set[str]) -> bool:
|
||
"""Strict whitelist: exact first-line match only (no fuzzy substring)."""
|
||
if is_blacklisted(name):
|
||
return False
|
||
return name in allow
|
||
|
||
# ─── gestures / nav ─────────────────────────────────────────────────
|
||
|
||
def tap_exact(name: str, ymin=0, ymax=9999, prefer_small=True) -> bool:
|
||
data = dump()
|
||
hits = []
|
||
for e in data.get("elements") or []:
|
||
t = lab(e)
|
||
fl = first_line(t)
|
||
if fl != name and t != name:
|
||
continue
|
||
if not e.get("clickable"):
|
||
continue
|
||
xy = center(e)
|
||
if not xy or xy[1] < ymin or xy[1] > ymax:
|
||
continue
|
||
if area(e) > 1080 * 700:
|
||
continue
|
||
hits.append(e)
|
||
if not hits:
|
||
return False
|
||
hits.sort(key=lambda e: area(e) if prefer_small else -area(e))
|
||
xy = center(hits[0])
|
||
print(f" TAP '{name}' {xy}", flush=True)
|
||
tap_xy(*xy)
|
||
return True
|
||
|
||
|
||
def tap_contains(needle: str, ymin=0, ymax=9999) -> bool:
|
||
data = dump()
|
||
hits = []
|
||
for e in data.get("elements") or []:
|
||
t = lab(e)
|
||
if needle not in t:
|
||
continue
|
||
if not e.get("clickable"):
|
||
continue
|
||
xy = center(e)
|
||
if not xy or xy[1] < ymin or xy[1] > ymax:
|
||
continue
|
||
if area(e) > 1080 * 700:
|
||
continue
|
||
hits.append(e)
|
||
if not hits:
|
||
return False
|
||
hits.sort(key=area)
|
||
xy = center(hits[0])
|
||
print(f" TAP~ '{needle}' {xy}", flush=True)
|
||
tap_xy(*xy)
|
||
return True
|
||
|
||
|
||
def tap_tab(name: str):
|
||
fb = TAB_FALLBACK.get(name, (540, 2255))
|
||
if not tap_exact(name, ymin=2180) and not tap_contains(name, ymin=2180):
|
||
print(f" tab fallback {name} {fb}", flush=True)
|
||
tap_xy(*fb)
|
||
|
||
|
||
def escape_modals(max_rounds=10) -> None:
|
||
for i in range(max_rounds):
|
||
ensure_app()
|
||
labs = labels()
|
||
b = "\n".join(labs)
|
||
fls = [first_line(x) for x in labs]
|
||
print(f" escape[{i}] {fls[:4]}", flush=True)
|
||
if "首页" in fls and not any(
|
||
k in b
|
||
for k in (
|
||
"隐私政策更新", "温馨提示", "立即领取", "打开通知",
|
||
"开通SVIP", "向TA提问", "确认并支付", "专享价购买", "未成年人模式",
|
||
)
|
||
):
|
||
return
|
||
if "同意" in fls and any(k in b for k in ("隐私", "温馨提示")):
|
||
tap_exact("同意")
|
||
continue
|
||
if "不再提醒" in fls:
|
||
tap_exact("不再提醒")
|
||
continue
|
||
if "我知道了" in fls:
|
||
tap_exact("我知道了")
|
||
continue
|
||
if "取消" in fls and "通知" in b:
|
||
tap_exact("取消")
|
||
continue
|
||
if "立即领取" in fls:
|
||
for y in (1680, 1720, 1760, 1800):
|
||
tap_xy(540, y, wait=0.25)
|
||
continue
|
||
if any(k in b for k in PAYWALL_KEYS):
|
||
# leave to caller to shot; just stop escaping into back-loop
|
||
return
|
||
if "关闭" in fls:
|
||
data = dump()
|
||
for e in data.get("elements") or []:
|
||
if lab(e) == "关闭" and area(e) < 200 * 200:
|
||
tap_xy(*center(e))
|
||
break
|
||
else:
|
||
back()
|
||
continue
|
||
# stuck overlay
|
||
if i >= 3:
|
||
back()
|
||
continue
|
||
return
|
||
|
||
|
||
def scroll_home_top():
|
||
"""Pull homepage content back to the tool grid (avoid feed / mid-scroll)."""
|
||
for _ in range(3):
|
||
adb("shell", "input", "swipe", "540", "700", "540", "1900", "280")
|
||
time.sleep(0.45)
|
||
|
||
|
||
def reset_home_tool_row():
|
||
"""Horizontal carousel often sits on page 2 — swipe right back to I人E人."""
|
||
for _ in range(2):
|
||
swipe_right_mid()
|
||
|
||
|
||
def is_more_wall(labs=None) -> bool:
|
||
"""「更多」工具墙：顶栏「更多/推荐」+ 墙内专属入口，避免首页信息流误判。"""
|
||
fls = [first_line(x) for x in (labs or labels())]
|
||
if "更多" in fls and "推荐" in fls and (
|
||
"心情小镇" in fls or "幸运地图" in fls or "聊天分析" in fls or "关系网" in fls
|
||
):
|
||
return True
|
||
return False
|
||
|
||
|
||
def is_home_grid(labs=None) -> bool:
|
||
fls = [first_line(x) for x in (labs or labels())]
|
||
# inside chart chrome → not home
|
||
if "天象" in fls and "本命" in fls and "行运" in fls:
|
||
return False
|
||
# 「更多」工具墙也有星座/生辰/紫微，不能当首页
|
||
if is_more_wall(labs):
|
||
return False
|
||
# 玩法广场 hub
|
||
if "玩法广场" in fls and "热门" in fls:
|
||
return False
|
||
# 首页宫格：完整第一屏，或横向滑过后的工具行
|
||
if "I人E人" in fls:
|
||
return True
|
||
if "星座" in fls and "星盘" in fls and "生辰" in fls:
|
||
return True
|
||
if "紫微" in fls and ("恋爱能力" in fls or "星盘报告" in fls or "生辰历" in fls or "倾诉" in fls):
|
||
return True
|
||
if "AI玩法广场" in fls and ("星盘报告" in fls or "灵魂伴侣" in fls or "智慧卡" in fls):
|
||
return True
|
||
# 底栏在首页 + 可见档案「自己」工具带
|
||
if "首页" in fls and "自己" in fls and "示例" in fls and ("紫微" in fls or "星座" in fls or "星盘" in fls):
|
||
return True
|
||
return False
|
||
|
||
|
||
def tap_top_right_close() -> bool:
|
||
"""在线房间等全屏页：右上角 × 关闭（BACK/底栏常无效）。"""
|
||
try:
|
||
data = dump()
|
||
except Exception as ex:
|
||
print(f" close-X dump fail: {ex}", flush=True)
|
||
data = {}
|
||
hits = []
|
||
for e in data.get("elements") or []:
|
||
if not e.get("clickable"):
|
||
continue
|
||
xy = center(e)
|
||
if not xy:
|
||
continue
|
||
x, y = xy
|
||
if y > 300 or x < 820:
|
||
continue
|
||
if area(e) > 60000:
|
||
continue
|
||
t = (lab(e) or "").strip()
|
||
fl = first_line(t)
|
||
score = 0
|
||
if fl in ("×", "X", "x", "关闭"):
|
||
score += 10
|
||
if "关闭" in t:
|
||
score += 5
|
||
if not fl:
|
||
score += 2
|
||
hits.append((score, area(e), xy))
|
||
hits.sort(key=lambda h: (-h[0], h[1]))
|
||
if hits and hits[0][0] > 0:
|
||
xy = hits[0][2]
|
||
print(f" TAP-XY close-X {xy}", flush=True)
|
||
tap_xy(*xy, wait=1.2)
|
||
return True
|
||
for xy in ((1002, 215), (990, 160), (1010, 180)):
|
||
print(f" TAP-XY close-X fallback {xy}", flush=True)
|
||
tap_xy(*xy, wait=1.0)
|
||
try:
|
||
fls = [first_line(x) for x in labels()]
|
||
if not any(k in fls for k in ("说点什么...", "说点什么", "公开连麦", "麦序")):
|
||
return True
|
||
except Exception:
|
||
return True
|
||
return False
|
||
|
||
|
||
def go_home():
|
||
"""Leave subpages via BACK, then land on 首页 tool grid."""
|
||
try:
|
||
escape_modals(max_rounds=3)
|
||
except Exception:
|
||
pass
|
||
prev_sig = None
|
||
stuck = 0
|
||
for i in range(10):
|
||
try:
|
||
labs = labels()
|
||
except Exception as ex:
|
||
print(f" go_home dump fail: {ex}", flush=True)
|
||
back()
|
||
continue
|
||
fls = [first_line(x) for x in labs]
|
||
b = "\n".join(labs)
|
||
print(f" go_home[{i}] {fls[:5]}", flush=True)
|
||
sig = tuple(fls[:4])
|
||
if sig == prev_sig:
|
||
stuck += 1
|
||
else:
|
||
stuck = 0
|
||
prev_sig = sig
|
||
if stuck >= 3:
|
||
print(" go_home: stuck → force-stop relaunch", flush=True)
|
||
adb("shell", "am", "force-stop", "com.xxwolo.cc5", timeout=15)
|
||
time.sleep(1.2)
|
||
adb("shell", "am", "start", "-n", ACT, timeout=15)
|
||
time.sleep(3.5)
|
||
tap_tab("首页")
|
||
time.sleep(1)
|
||
break
|
||
# Unity / WebView 沙盘等：无原生控件，BACK 无效 → 强杀重开
|
||
# 注意：空 dump 不能当成 Game view（否则会误强杀死循环）
|
||
if fls == ["Game view"] or (len(fls) == 1 and fls[0] == "Game view"):
|
||
print(" go_home: Game view → force-stop relaunch", flush=True)
|
||
adb("shell", "am", "force-stop", "com.xxwolo.cc5", timeout=15)
|
||
time.sleep(1)
|
||
adb("shell", "am", "start", "-n", ACT, timeout=15)
|
||
time.sleep(3.5)
|
||
tap_tab("首页")
|
||
time.sleep(1)
|
||
break
|
||
if not fls:
|
||
print(" go_home: empty dump → BACK + 首页", flush=True)
|
||
back()
|
||
time.sleep(0.5)
|
||
tap_tab("首页")
|
||
time.sleep(0.8)
|
||
if i >= 2:
|
||
adb("shell", "am", "start", "-n", ACT, timeout=15)
|
||
time.sleep(2.5)
|
||
tap_tab("首页")
|
||
break
|
||
continue
|
||
if is_home_grid(labs):
|
||
break
|
||
# 测测AI 聊天浮层：必须先关 ×（点首页无效）
|
||
if "测测AI" in fls and any(
|
||
k in b
|
||
for k in (
|
||
"让我来解答你的问题吧",
|
||
"内容由AI生成",
|
||
"正在生成",
|
||
"已思考",
|
||
"真人1v1",
|
||
)
|
||
) and (
|
||
"让我来解答你的问题吧" in b
|
||
or "内容由AI生成" in b
|
||
or "正在生成" in b
|
||
or "已思考" in fls
|
||
or any(len(first_line(x)) > 36 for x in labs[:10])
|
||
):
|
||
print(" go_home: 测测AI sheet → close-X", flush=True)
|
||
tap_top_right_close()
|
||
time.sleep(0.8)
|
||
continue
|
||
# 向TA提问 / 达人咨询收银：BACK
|
||
if "向TA提问" in fls or (
|
||
"文字沟通" in fls and "语音沟通" in fls and any(
|
||
k in b for k in ("请输入问题", "支付方式", "选择档案", "VIP专享价")
|
||
)
|
||
):
|
||
print(" go_home: 向TA提问 → BACK", flush=True)
|
||
back()
|
||
time.sleep(0.7)
|
||
continue
|
||
# 达人主页（文字/语音沟通入口）
|
||
if "文字沟通" in fls and "语音沟通" in fls:
|
||
print(" go_home: 达人主页 → BACK", flush=True)
|
||
back()
|
||
time.sleep(0.7)
|
||
continue
|
||
# 在线直播间 / 达人房间：点右上角 ×
|
||
if any(k in fls for k in ("说点什么...", "说点什么", "公开连麦", "麦序", "更多在线")):
|
||
print(" go_home: live room → close-X", flush=True)
|
||
tap_top_right_close()
|
||
time.sleep(0.8)
|
||
continue
|
||
# 「我的」个人主页：BACK 无效，直接切首页 Tab
|
||
if "VIP会员陪伴中" in fls or (
|
||
"ESTP" in fls and any(k in fls for k in ("兑换码", "订单", "设置", "报告与测试"))
|
||
) or (
|
||
any(k in fls for k in ("兑换码", "我的订单", "设置")) and "首页" in fls and "我的" in fls
|
||
):
|
||
print(" go_home: on 我的 → tap 首页", flush=True)
|
||
tap_tab("首页")
|
||
time.sleep(1.0)
|
||
continue
|
||
# 消息 / 问 / 在线 底栏页：BACK 停在 Tab 内
|
||
if ("系统通知" in fls and "小星" in fls) or (
|
||
"消息" in fls and any(k in fls for k in ("系统通知", "互动消息", "去开启"))
|
||
):
|
||
print(" go_home: on 消息 → tap 首页", flush=True)
|
||
tap_tab("首页")
|
||
time.sleep(1.0)
|
||
continue
|
||
# 在线列表须先于泛化的「测测AI」判断
|
||
if "综合排序" in fls and ("筛选" in fls or "私人专线" in fls or "全部工具" in fls):
|
||
print(" go_home: on 在线 → tap 首页", flush=True)
|
||
tap_tab("首页")
|
||
time.sleep(1.0)
|
||
continue
|
||
if "测测AI" in fls or "真人1v1" in fls:
|
||
print(" go_home: on 问 → tap 首页", flush=True)
|
||
tap_tab("首页")
|
||
time.sleep(1.0)
|
||
continue
|
||
if is_comment_feed(labs):
|
||
print(" go_home: leave comment feed", flush=True)
|
||
back()
|
||
time.sleep(0.7)
|
||
continue
|
||
# 档案列表也在我的树下
|
||
if any(k in fls for k in ("档案列表", "按关系排序", "按昵称排序", "添加档案")):
|
||
print(" go_home: leave 档案 → 首页", flush=True)
|
||
back()
|
||
time.sleep(0.5)
|
||
tap_tab("首页")
|
||
time.sleep(0.8)
|
||
continue
|
||
if any(
|
||
k in fls
|
||
for k in (
|
||
"天象", "本命", "行运", "三限", "基础解读", "Deepseek解读",
|
||
)
|
||
):
|
||
back()
|
||
time.sleep(0.7)
|
||
continue
|
||
if any(k in b for k in PAYWALL_KEYS):
|
||
back()
|
||
time.sleep(0.5)
|
||
continue
|
||
# Prefer BACK over tab tap when unsure (tab tap often no-ops / hangs dumps)
|
||
back()
|
||
time.sleep(0.6)
|
||
if i >= 3:
|
||
tap_tab("首页")
|
||
time.sleep(0.8)
|
||
try:
|
||
escape_modals(max_rounds=2)
|
||
except Exception:
|
||
pass
|
||
scroll_home_top()
|
||
try:
|
||
labs = labels()
|
||
if is_home_grid(labs) and "I人E人" not in [first_line(x) for x in labs]:
|
||
reset_home_tool_row()
|
||
labs = labels()
|
||
if not is_home_grid(labs):
|
||
print(" go_home: relaunch main", flush=True)
|
||
adb("shell", "am", "force-stop", "com.xxwolo.cc5", timeout=15)
|
||
time.sleep(1)
|
||
adb("shell", "am", "start", "-n", ACT, timeout=15)
|
||
time.sleep(3.5)
|
||
tap_tab("首页")
|
||
time.sleep(1)
|
||
scroll_home_top()
|
||
reset_home_tool_row()
|
||
except Exception as ex:
|
||
print(f" go_home final: {ex}", flush=True)
|
||
adb("shell", "am", "force-stop", "com.xxwolo.cc5", timeout=15)
|
||
time.sleep(1)
|
||
adb("shell", "am", "start", "-n", ACT, timeout=15)
|
||
time.sleep(3.5)
|
||
|
||
|
||
def open_more():
|
||
"""Open 首页宫格「更多」——常在第二屏，需左右滑。
|
||
|
||
禁止 tap_contains('更多')：信息流大卡片文案里常含「更多」，会点到 (540,642) 假入口。
|
||
"""
|
||
go_home()
|
||
scroll_home_top()
|
||
reset_home_tool_row()
|
||
for _ in range(5):
|
||
try:
|
||
# 只认宫格小图标：文案精确为「更多」，且在工具带 Y 范围
|
||
if tap_exact("更多", ymin=900, ymax=1700, prefer_small=True):
|
||
time.sleep(1.6)
|
||
if is_more_wall():
|
||
return True
|
||
print(" open_more: tap 更多 but not wall, back", flush=True)
|
||
back()
|
||
time.sleep(0.6)
|
||
except Exception as ex:
|
||
print(f" open_more tap err: {ex}", flush=True)
|
||
swipe_left_mid()
|
||
# 坐标兜底（Pixel7 模拟器第二屏右下宫格）
|
||
print(" open_more TAP-XY (984,1366)", flush=True)
|
||
swipe_left_mid()
|
||
tap_xy(984, 1366, wait=1.8)
|
||
try:
|
||
if is_more_wall():
|
||
return True
|
||
b = blob()
|
||
if any(k in b for k in ("心情小镇", "幸运地图", "星骰")):
|
||
return True
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
def find_and_tap_in_scroll(name: str, rounds=5, ymin=300, ymax=2100) -> bool:
|
||
for i in range(rounds):
|
||
if tap_exact(name, ymin=ymin, ymax=ymax):
|
||
return True
|
||
if tap_force(name, ymin=ymin, ymax=ymax):
|
||
return True
|
||
# 短文案禁用 contains，避免误点信息流
|
||
if len(name) >= 3 and tap_contains(name, ymin=ymin, ymax=ymax):
|
||
return True
|
||
if i < rounds - 1:
|
||
# 更多墙内轻滑，避免整页上滑把墙滑走
|
||
adb("shell", "input", "swipe", "540", "1700", "540", "1100", "280")
|
||
time.sleep(0.45)
|
||
return False
|
||
|
||
|
||
def tap_home_tool(label: str) -> bool:
|
||
"""Tap a home-grid tool; prefer exact match, then known coordinates."""
|
||
# Known Pixel-7-ish grid centers (from successful captures)
|
||
COORDS = {
|
||
"I人E人": (150, 974),
|
||
"星座": (320, 974),
|
||
"星盘": (490, 974),
|
||
"生辰": (660, 974),
|
||
"缘分合盘": (830, 974),
|
||
"紫微": (1000, 974),
|
||
"陪伴小星": (150, 1200),
|
||
"倾诉": (320, 1200),
|
||
"智慧卡": (490, 1200),
|
||
"星盘报告": (660, 1200),
|
||
"生辰历": (830, 1200),
|
||
"灵魂伴侣": (1000, 1200),
|
||
"更多": (984, 1366),
|
||
"AI玩法广场": (274, 1583),
|
||
}
|
||
scroll_home_top()
|
||
reset_home_tool_row()
|
||
for attempt in range(3):
|
||
try:
|
||
if tap_exact(label, ymin=350, ymax=1700):
|
||
return True
|
||
except Exception as ex:
|
||
print(f" tap_exact err: {ex}", flush=True)
|
||
if attempt == 0:
|
||
scroll_home_top()
|
||
elif attempt == 1:
|
||
swipe_left_mid()
|
||
# last resort: fixed coordinates (avoids another failing dump loop)
|
||
if label in COORDS:
|
||
x, y = COORDS[label]
|
||
print(f" TAP-XY '{label}' ({x},{y})", flush=True)
|
||
if label == "更多":
|
||
swipe_left_mid()
|
||
tap_xy(x, y, wait=1.8)
|
||
return True
|
||
try:
|
||
return tap_contains(label, ymin=350, ymax=1700)
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
# ─── capture state ──────────────────────────────────────────────────
|
||
|
||
class State:
|
||
def __init__(self):
|
||
self.shots = 0
|
||
self.seen_fps: set[str] = set()
|
||
self.log: list[dict] = []
|
||
|
||
|
||
STATE = State()
|
||
|
||
|
||
def _pil_open(path: Path):
|
||
from PIL import Image
|
||
|
||
return Image.open(path).convert("RGB")
|
||
|
||
|
||
def _strip_mse(a, b) -> float:
|
||
"""Subsampled mean squared error between two same-size RGB images."""
|
||
wa, ha = a.size
|
||
wb, hb = b.size
|
||
if (wa, ha) != (wb, hb) or wa == 0 or ha == 0:
|
||
return 1e18
|
||
pa, pb = a.load(), b.load()
|
||
err = 0.0
|
||
n = 0
|
||
step = 6
|
||
for y in range(0, ha, step):
|
||
for x in range(0, wa, step):
|
||
ca, cb = pa[x, y], pb[x, y]
|
||
err += (ca[0] - cb[0]) ** 2 + (ca[1] - cb[1]) ** 2 + (ca[2] - cb[2]) ** 2
|
||
n += 1
|
||
return err / max(n, 1)
|
||
|
||
|
||
def _frame_sig(img) -> str:
|
||
"""Cheap content signature to detect end-of-scroll (no movement)."""
|
||
w, h = img.size
|
||
band = img.crop((0, h // 3, w, (2 * h) // 3)).resize((64, 32))
|
||
return hashlib.md5(band.tobytes()).hexdigest()[:16]
|
||
|
||
|
||
def find_overlap_height(prev, curr, min_ov=LONG_MIN_OVERLAP, max_ov=LONG_MAX_OVERLAP) -> int:
|
||
"""How many top pixels of curr overlap the bottom of prev — coarse + fast."""
|
||
w, h = prev.size
|
||
max_ov = min(max_ov, h - 20, curr.size[1] - 20)
|
||
min_ov = min(min_ov, max_ov)
|
||
# fixed-step search only (avoid per-pixel refine — was hanging the crawl)
|
||
best_ov, best_err = min_ov, 1e18
|
||
for ov in range(min_ov, max_ov + 1, 24):
|
||
prev_strip = prev.crop((0, h - ov, w, h)).resize((w // 8, max(1, ov // 8)))
|
||
curr_strip = curr.crop((0, 0, w, ov)).resize((w // 8, max(1, ov // 8)))
|
||
err = _strip_mse(prev_strip, curr_strip)
|
||
if err < best_err:
|
||
best_err, best_ov = err, ov
|
||
return best_ov
|
||
|
||
|
||
def stitch_long(frames: list[Path], out_path: Path) -> Path | None:
|
||
"""Vertical stitch with overlap detection; crops chrome on continuation frames."""
|
||
if not frames:
|
||
return None
|
||
from PIL import Image
|
||
|
||
imgs = [_pil_open(p) for p in frames]
|
||
w, h0 = imgs[0].size
|
||
# first frame: keep full width, drop bottom tab for cleaner long page
|
||
pieces = [imgs[0].crop((0, 0, w, max(1, h0 - LONG_BOTTOM_CROP)))]
|
||
for im in imgs[1:]:
|
||
w2, h2 = im.size
|
||
top = min(LONG_TOP_CROP_CONT, h2 // 5)
|
||
bottom = max(top + 1, h2 - LONG_BOTTOM_CROP)
|
||
body = im.crop((0, top, w2, bottom))
|
||
if body.size[0] != pieces[-1].size[0]:
|
||
body = body.resize((pieces[-1].size[0], body.size[1]))
|
||
ov = find_overlap_height(pieces[-1], body)
|
||
pieces.append(body.crop((0, ov, body.size[0], body.size[1])))
|
||
|
||
total_h = sum(p.size[1] for p in pieces)
|
||
canvas = Image.new("RGB", (pieces[0].size[0], total_h), (255, 255, 255))
|
||
y = 0
|
||
for p in pieces:
|
||
canvas.paste(p, (0, y))
|
||
y += p.size[1]
|
||
out_path.parent.mkdir(parents=True, exist_ok=True)
|
||
canvas.save(out_path, "PNG", optimize=True)
|
||
return out_path
|
||
|
||
|
||
def screencap_to(path: Path) -> Path:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
data = tool("screenshot", "--out", str(path))
|
||
if path.exists() and path.stat().st_size >= 1000:
|
||
return path
|
||
# binary fallback (avoid text-mode corruption)
|
||
with path.open("wb") as f:
|
||
subprocess.run(
|
||
["adb", "exec-out", "screencap", "-p"],
|
||
stdout=f,
|
||
timeout=60,
|
||
check=False,
|
||
)
|
||
if not path.exists() or path.stat().st_size < 1000:
|
||
raise RuntimeError(f"screencap failed: {path} tool={data}")
|
||
return path
|
||
|
||
|
||
def shot(rel: str, note: str = "", long: bool = True) -> Path:
|
||
"""Capture viewport; optionally scroll-stitch a *-long.png and scroll back to top."""
|
||
ensure_app()
|
||
path = OUT / rel
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
print(f" shot:begin {rel}", flush=True)
|
||
# screencap first so we always have a file even if dump hangs
|
||
screencap_to(path)
|
||
STATE.shots += 1
|
||
print(f"SHOT [{STATE.shots}] {rel} bytes={path.stat().st_size} note={note}", flush=True)
|
||
try:
|
||
labs = labels()
|
||
(OUT / (rel.replace(".png", ".labels.txt"))).write_text("\n".join(labs), encoding="utf-8")
|
||
except Exception as ex:
|
||
print(f" labels fail: {ex}", flush=True)
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- `{rel}` {note}\n")
|
||
STATE.log.append({"path": rel, "note": note, "bytes": path.stat().st_size})
|
||
|
||
if long and not rel.endswith("-cashier.png"):
|
||
try:
|
||
long_path = shot_long_from(path, rel, note)
|
||
if long_path:
|
||
STATE.log.append({"path": str(long_path.relative_to(OUT)), "note": note + " [long]"})
|
||
except Exception as ex:
|
||
print(f" long-shot fail: {ex}", flush=True)
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- LONG_FAIL `{rel}` {ex}\n")
|
||
finally:
|
||
# 只复位滚动，不再 dump（dump 是卡死主因之一）
|
||
try:
|
||
scroll_page_top(2)
|
||
except Exception as ex:
|
||
print(f" scroll_page_top fail: {ex}", flush=True)
|
||
return path
|
||
|
||
|
||
def shot_long_from(first: Path, rel: str, note: str):
|
||
"""往上滑多屏截取并拼成 *-long.png。
|
||
|
||
策略：先滑再截（保证能动）；每隔一帧再尝试点「展开更多」，dump 超时则跳过。
|
||
"""
|
||
parts_dir = first.parent / (first.stem + "_parts")
|
||
parts_dir.mkdir(parents=True, exist_ok=True)
|
||
frames = [first]
|
||
part0 = parts_dir / "0.png"
|
||
if not part0.exists():
|
||
part0.write_bytes(first.read_bytes())
|
||
|
||
prev_sig = _frame_sig(_pil_open(frames[0]))
|
||
for i in range(1, LONG_MAX_SCROLLS + 1):
|
||
print(f" long: 往上滑 frame={i}", flush=True)
|
||
try:
|
||
(OUT / ".heartbeat").write_text(f"{time.strftime('%T')} {rel} f{i}\n", encoding="utf-8")
|
||
except Exception:
|
||
pass
|
||
swipe_up()
|
||
# expand 先关掉：ui dump 在长图循环里会偶发把进程卡死/杀掉
|
||
# （单独冒烟正常，全量爬里 frame=1 就挂）。需要时再开 LONG_EXPAND=1
|
||
if os.environ.get("LONG_EXPAND") == "1" and i % 2 == 1:
|
||
try:
|
||
n_exp = expand_more_if_present(max_taps=2)
|
||
if n_exp:
|
||
print(f" long: expanded x{n_exp}", flush=True)
|
||
time.sleep(0.4)
|
||
except Exception as ex:
|
||
print(f" long: expand skip {ex}", flush=True)
|
||
part = parts_dir / f"{i}.png"
|
||
try:
|
||
screencap_to(part)
|
||
except Exception as ex:
|
||
print(f" long: screencap fail {ex}", flush=True)
|
||
break
|
||
try:
|
||
sig = _frame_sig(_pil_open(part))
|
||
except Exception as ex:
|
||
print(f" long: sig fail {ex}", flush=True)
|
||
break
|
||
if sig == prev_sig:
|
||
print(f" long: 已到底 frame={i}", flush=True)
|
||
part.unlink(missing_ok=True)
|
||
break
|
||
prev_sig = sig
|
||
frames.append(part)
|
||
|
||
if len(frames) <= 1:
|
||
print(" long: 单屏无需拼接", flush=True)
|
||
return None
|
||
|
||
long_rel = rel.replace(".png", "-long.png")
|
||
out = OUT / long_rel
|
||
print(f" long: stitching {len(frames)} frames…", flush=True)
|
||
try:
|
||
stitch_long(frames, out)
|
||
except Exception as ex:
|
||
print(f" long: stitch fail {ex}, fallback concat", flush=True)
|
||
# emergency: simple vertical concat with fixed crop
|
||
from PIL import Image
|
||
imgs = [_pil_open(p) for p in frames]
|
||
w, h = imgs[0].size
|
||
crop_b = LONG_BOTTOM_CROP
|
||
pieces = [imgs[0].crop((0, 0, w, h - crop_b))]
|
||
for im in imgs[1:]:
|
||
body = im.crop((0, LONG_TOP_CROP_CONT, w, h - crop_b))
|
||
ov = min(280, body.size[1] // 4)
|
||
pieces.append(body.crop((0, ov, w, body.size[1])))
|
||
th = sum(p.size[1] for p in pieces)
|
||
canvas = Image.new("RGB", (w, th), (255, 255, 255))
|
||
y = 0
|
||
for p in pieces:
|
||
canvas.paste(p, (0, y))
|
||
y += p.size[1]
|
||
canvas.save(out, "PNG")
|
||
STATE.shots += 1
|
||
print(
|
||
f"SHOT-LONG [{STATE.shots}] {long_rel} frames={len(frames)} "
|
||
f"bytes={out.stat().st_size} note={note}",
|
||
flush=True,
|
||
)
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- `{long_rel}` {note} [long x{len(frames)} 往上滑+展开]\n")
|
||
return out
|
||
|
||
|
||
def is_comment_feed(labs=None) -> bool:
|
||
"""UGC 评论流：不截图，应直接返回。"""
|
||
b = "\n".join(labs or labels())
|
||
return ("全部评论" in b) or ("写评论" in b) or ("条评论" in b and "最热" in b)
|
||
|
||
|
||
def is_result_card(labs=None) -> bool:
|
||
"""结果图 / 分享海报页：不重复截。"""
|
||
b = "\n".join(labs or labels())
|
||
keys = (
|
||
"结果图", "生成结果图", "分享结果", "分享卡片", "生成卡片",
|
||
"保存到相册", "保存图片", "生成海报", "长图分享",
|
||
)
|
||
return any(k in b for k in keys)
|
||
|
||
|
||
def capture_here(rel: str, note: str = "", long: bool = True) -> str:
|
||
"""Shot current (viewport + optional long); cashier → shot then stop-deeper."""
|
||
try:
|
||
labs = labels()
|
||
b = "\n".join(labs)
|
||
except Exception as ex:
|
||
print(f" blob fail: {ex}", flush=True)
|
||
labs, b = [], ""
|
||
if is_comment_feed(labs):
|
||
print(f" skip comment feed: {note}", flush=True)
|
||
back()
|
||
return "skip"
|
||
if is_result_card(labs):
|
||
print(f" skip result card: {note}", flush=True)
|
||
back()
|
||
return "skip"
|
||
if any(k in b for k in PAYWALL_KEYS):
|
||
shot(rel.replace(".png", "-cashier.png"), note + " [cashier]", long=False)
|
||
return "cashier"
|
||
shot(rel, note, long=long)
|
||
return "ok"
|
||
|
||
|
||
def collect_candidates(allow: set[str], tabs: list[str], ymin=280, ymax=2100) -> list[str]:
|
||
data = dump()
|
||
found: list[str] = []
|
||
seen: set[str] = set()
|
||
# prefer tabs first
|
||
for t in tabs:
|
||
for e in data.get("elements") or []:
|
||
fl = first_line(lab(e))
|
||
if fl == t or t in lab(e):
|
||
if fl not in seen and e.get("clickable"):
|
||
seen.add(fl if fl == t else t)
|
||
found.append(t)
|
||
break
|
||
for e in data.get("elements") or []:
|
||
if not e.get("clickable"):
|
||
continue
|
||
t = lab(e)
|
||
if not t:
|
||
continue
|
||
fl = first_line(t)
|
||
xy = center(e)
|
||
if not xy or xy[1] < ymin or xy[1] > ymax:
|
||
continue
|
||
if area(e) == 0 or area(e) > 900 * 500:
|
||
continue
|
||
if fl in seen:
|
||
continue
|
||
if not allowed(fl, allow):
|
||
continue
|
||
seen.add(fl)
|
||
found.append(fl)
|
||
if len(found) >= MAX_CHILDREN_PER_PAGE:
|
||
break
|
||
return found
|
||
|
||
|
||
# ─── module exploration (flat, not deep recursive DFS) ──────────────
|
||
# 递归 dive + 每层长图 + 密集 ui dump 会导致进程在 candidates 后静默死亡。
|
||
# 改为：入口长图 → 每个 tab 各截一页（可长图）→ 少量白名单按钮各截一页。
|
||
|
||
SKIP_ACTIONS = {
|
||
"测测达人在线解读星盘", "现代", "参数", "日", "周", "月",
|
||
"分享", "保存", "生成卡片", "历史", "记录", "帮助", "说明",
|
||
"评论", "全部评论", "写评论", "点赞", "最热", "最新",
|
||
# 结果图 / 分享海报：不重复截
|
||
"结果图", "查看结果", "生成结果图", "分享结果", "分享卡片",
|
||
"保存图片", "保存到相册", "海报", "长图分享", "生成海报",
|
||
# 倾诉 / 真人1v1 / 聊天类：不点进去
|
||
"倾诉", "开始倾诉", "匹配达人",
|
||
"真人1v1", "真人1V1", "语音通话", "文字",
|
||
"聊天", "小星", "陪伴小星", "聊天分析", "互动消息",
|
||
}
|
||
|
||
# 整模块跳过（不进入）
|
||
SKIP_MODULES = {
|
||
"qingsu", # 倾诉
|
||
"liaotian", # 聊天分析
|
||
"xiaoxing", # 陪伴小星（聊天）
|
||
"tab_online", # 在线达人（真人咨询）
|
||
}
|
||
|
||
|
||
def tap_force(name: str, ymin=0, ymax=9999) -> bool:
|
||
"""Tap by first-line text even when node is not marked clickable (settings rows etc.)."""
|
||
try:
|
||
data = dump()
|
||
except Exception as ex:
|
||
print(f" tap_force dump fail: {ex}", flush=True)
|
||
return False
|
||
hits = []
|
||
for e in data.get("elements") or []:
|
||
fl = first_line(lab(e))
|
||
if fl != name:
|
||
continue
|
||
xy = center(e)
|
||
if not xy or xy[1] < ymin or xy[1] > ymax:
|
||
continue
|
||
if area(e) > 1080 * 900:
|
||
continue
|
||
hits.append((0 if e.get("clickable") else 1, area(e), xy))
|
||
if not hits:
|
||
return False
|
||
hits.sort()
|
||
xy = hits[0][2]
|
||
print(f" TAP! '{name}' {xy}", flush=True)
|
||
tap_xy(*xy)
|
||
return True
|
||
|
||
|
||
def tap_label(name: str, ymin=180, ymax=2150) -> bool:
|
||
"""Exact / force tap only — never tap_contains (feed cards false-hit e.g. 540,642)."""
|
||
print(f" try tap '{name}'", flush=True)
|
||
try:
|
||
if tap_exact(name, ymin=ymin, ymax=ymax):
|
||
return True
|
||
if tap_force(name, ymin=ymin, ymax=ymax):
|
||
return True
|
||
except Exception as ex:
|
||
print(f" tap fail '{name}': {ex}", flush=True)
|
||
return False
|
||
|
||
|
||
def visible_allow_hits(allow: set[str], ymin=160, ymax=2050) -> list[tuple[str, tuple[int, int]]]:
|
||
"""One dump → (label, xy) for whitelist items on screen, top-to-bottom."""
|
||
try:
|
||
data = dump()
|
||
except Exception as ex:
|
||
print(f" visible dump fail: {ex}", flush=True)
|
||
return []
|
||
hits = []
|
||
seen = set()
|
||
for e in data.get("elements") or []:
|
||
fl = first_line(lab(e))
|
||
if not fl or fl in seen or fl not in allow:
|
||
continue
|
||
if fl in BLACKLIST_EXACT or fl in BOTTOM_TABS or fl in SKIP_ACTIONS:
|
||
continue
|
||
xy = center(e)
|
||
if not xy or xy[1] < ymin or xy[1] > ymax:
|
||
continue
|
||
if area(e) > 1080 * 850:
|
||
continue
|
||
seen.add(fl)
|
||
hits.append((xy[1], fl, xy))
|
||
hits.sort()
|
||
return [(fl, xy) for _, fl, xy in hits]
|
||
|
||
|
||
def is_home_leak(data=None) -> bool:
|
||
"""判断 dump 是否仍是首页（而非当前模块页）。
|
||
|
||
注意：AI玩法广场等页卡片标题也可能含「七宗罪与七美德」，不能单靠该文案判定。
|
||
"""
|
||
try:
|
||
labs = labels(data) if data is not None else labels()
|
||
except Exception:
|
||
return True
|
||
fls = [first_line(x) for x in labs]
|
||
b = "\n".join(labs)
|
||
# 首页宫格特征：底栏 + 工具行 + 示例/自己切换
|
||
tool_row = ("I人E人" in fls and "星盘" in fls and "生辰" in fls) or (
|
||
"倾诉" in fls and "智慧卡" in fls and "星盘报告" in fls
|
||
)
|
||
home_chrome = "首页" in fls and "我的" in fls
|
||
if "今日心情" in b and tool_row and home_chrome:
|
||
return True
|
||
if tool_row and home_chrome and "示例" in fls and "自己" in fls:
|
||
return True
|
||
if "部分标题由AI生成，仅供参考" in b and "AI玩法广场" in fls and tool_row and home_chrome:
|
||
return True
|
||
return False
|
||
|
||
|
||
def leave_detail(prefer_close: bool = False) -> None:
|
||
"""离开详情：直播间/测测AI 用右上角 ×，其余 BACK。"""
|
||
try:
|
||
labs = labels()
|
||
fls = [first_line(x) for x in labs]
|
||
b = "\n".join(labs)
|
||
if prefer_close or any(
|
||
k in b
|
||
for k in ("测测AI", "让我来解答你的问题吧", "说点什么...", "公开连麦", "麦序")
|
||
):
|
||
tap_top_right_close()
|
||
elif any(k in fls for k in ("说点什么...", "公开连麦", "麦序")):
|
||
tap_top_right_close()
|
||
else:
|
||
back()
|
||
time.sleep(0.55)
|
||
except Exception:
|
||
ensure_app()
|
||
|
||
|
||
def explore_module(mod: dict) -> None:
|
||
mid = mod["id"]
|
||
tabs = list(mod.get("tabs") or [])
|
||
children = [c for c in (mod.get("children") or []) if c not in SKIP_ACTIONS]
|
||
allow: set[str] = set(tabs) | set(children)
|
||
fb: dict[str, tuple[int, int]] = dict(mod.get("fallback_xy") or {})
|
||
|
||
print(f" explore flat tabs={tabs[:10]} children={len(children)} fb={list(fb)[:6]}", flush=True)
|
||
|
||
# 先截 L1；dump 放在截图后重试，避免入口动画导致空 dump
|
||
st = capture_here(f"{mid}/L1.png", mid, long=True)
|
||
if st == "cashier":
|
||
print(" cashier noted — still try on-screen targets", flush=True)
|
||
|
||
for _ in range(3):
|
||
adb("shell", "input", "swipe", "540", "700", "540", "1700", "260")
|
||
time.sleep(0.35)
|
||
|
||
leak = is_home_leak()
|
||
if not leak:
|
||
onscreen = visible_allow_hits(allow)
|
||
if not onscreen:
|
||
time.sleep(1.0)
|
||
onscreen = visible_allow_hits(allow)
|
||
leak = is_home_leak()
|
||
else:
|
||
onscreen = []
|
||
# Hub Compose：dump 过稀或仍像首页时，只信 fallback，避免点到「示例」等首页节点
|
||
if fb and (leak or len(onscreen) <= 2 or is_home_leak()):
|
||
if not leak or len(onscreen) <= 2:
|
||
print(
|
||
f" a11y weak/home-leak (onscreen={len(onscreen)}) → fallback XY only",
|
||
flush=True,
|
||
)
|
||
leak = True
|
||
onscreen = []
|
||
print(f" onscreen: {[fl for fl, _ in onscreen[:14]]}", flush=True)
|
||
|
||
on_names = {fl for fl, _ in onscreen}
|
||
ordered: list[tuple[str, tuple[int, int] | None]] = []
|
||
for t in tabs:
|
||
xy = next((xy for fl, xy in onscreen if fl == t), None)
|
||
if xy is None and t in fb:
|
||
xy = fb[t]
|
||
ordered.append((t, xy))
|
||
if not leak:
|
||
for fl, xy in onscreen:
|
||
if fl not in tabs:
|
||
ordered.append((fl, xy))
|
||
extra_search = 0
|
||
for c in children:
|
||
if c in on_names or c in tabs:
|
||
continue
|
||
ordered.append((c, None))
|
||
extra_search += 1
|
||
if extra_search >= 4:
|
||
break
|
||
# fallback 里声明但未进 tabs 的入口也补上
|
||
for name, xy in fb.items():
|
||
if name not in {n for n, _ in ordered}:
|
||
ordered.append((name, xy))
|
||
|
||
targets = ordered[:14]
|
||
print(f" targets({len(targets)}): {[n for n, _ in targets[:12]]}", flush=True)
|
||
|
||
for name, xy in targets:
|
||
if STATE.shots >= MAX_SHOTS:
|
||
break
|
||
for _ in range(2):
|
||
adb("shell", "input", "swipe", "540", "700", "540", "1700", "240")
|
||
time.sleep(0.25)
|
||
if xy is None and name in tabs and not leak:
|
||
refreshed = visible_allow_hits({name}, ymin=150, ymax=1200)
|
||
if refreshed:
|
||
xy = refreshed[0][1]
|
||
if xy is None and name in fb:
|
||
xy = fb[name]
|
||
if xy is not None:
|
||
print(f" TAP-XY '{name}' {xy}", flush=True)
|
||
tap_xy(*xy, wait=1.3)
|
||
elif name in tabs or name in fb:
|
||
# 声明的 tabs 即使 dump 弱也要再试一次精确点击
|
||
if name in fb:
|
||
print(f" TAP-XY '{name}' {fb[name]}", flush=True)
|
||
tap_xy(*fb[name], wait=1.3)
|
||
elif not tap_label(name, ymin=120, ymax=2100):
|
||
print(f" skip tab miss '{name}'", flush=True)
|
||
continue
|
||
else:
|
||
print(f" skip offscreen '{name}'", flush=True)
|
||
continue
|
||
try:
|
||
escape_modals(max_rounds=2)
|
||
except Exception:
|
||
pass
|
||
if name in tabs or name in fb:
|
||
rel = f"{mid}/{safe_name(name)}/L2.png"
|
||
do_long = True
|
||
else:
|
||
rel = f"{mid}/act_{safe_name(name)}/L2.png"
|
||
do_long = False
|
||
capture_here(rel, f"{mid}>{name}", long=do_long)
|
||
leave_detail(prefer_close=(name in ("Ai 解读", "测测AI")))
|
||
|
||
|
||
def dive(module_id: str, path_parts: list[str], depth: int, max_depth: int,
|
||
allow: set[str], tabs: list[str]) -> str:
|
||
"""Deprecated recursive path — kept as thin wrapper to flat explore for safety."""
|
||
print(" dive: redirected to flat explore", flush=True)
|
||
return "ok"
|
||
|
||
|
||
# ─── module entry ───────────────────────────────────────────────────
|
||
|
||
def enter_module(mod: dict) -> bool:
|
||
kind, label = mod["entry"]
|
||
print(f"\n=== MODULE {mod['id']} entry={kind}/{label}", flush=True)
|
||
escape_modals()
|
||
if kind == "home":
|
||
go_home()
|
||
ok = tap_home_tool(label)
|
||
if not ok:
|
||
# try 更多 as fallback
|
||
if open_more() and find_and_tap_in_scroll(label, ymin=200, ymax=2200):
|
||
time.sleep(1.8)
|
||
return True
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- MISS entry home/{label}\n")
|
||
return False
|
||
time.sleep(1.8)
|
||
return True
|
||
if kind == "more":
|
||
if not open_more():
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- MISS open 更多 for {label}\n")
|
||
return False
|
||
if not find_and_tap_in_scroll(label, rounds=8, ymin=150, ymax=2350):
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- MISS entry more/{label}\n")
|
||
return False
|
||
time.sleep(1.8)
|
||
return True
|
||
if kind == "mine":
|
||
# 离开评论流再进我的
|
||
try:
|
||
if is_comment_feed():
|
||
back()
|
||
time.sleep(0.6)
|
||
except Exception:
|
||
pass
|
||
tap_tab("我的")
|
||
time.sleep(1.2)
|
||
escape_modals(max_rounds=3)
|
||
if not find_and_tap_in_scroll(label, rounds=6, ymin=300, ymax=2200):
|
||
alts = list(mod.get("entry_alts") or [])
|
||
alts += {
|
||
"会员": ["VIP会员", "开通会员", "我的会员", "VIP会员陪伴中", "升级SVIP"],
|
||
"VIP会员陪伴中": ["升级SVIP", "¥9.9开启陪伴", "新用户限时特惠", "VIP会员"],
|
||
"¥9.9开启陪伴": ["VIP会员陪伴中", "升级SVIP", "新用户限时特惠"],
|
||
"报告": ["报告与测试", "我的报告", "报告中心"],
|
||
"报告与测试": ["报告", "我的报告"],
|
||
"测试": ["我的测试", "报告与测试"],
|
||
"订单": ["我的订单"],
|
||
"设置": ["设置", "通用"],
|
||
}.get(label, [])
|
||
hit = False
|
||
for a in alts:
|
||
if find_and_tap_in_scroll(a, rounds=4, ymin=200, ymax=2200):
|
||
hit = True
|
||
break
|
||
if not hit:
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- MISS entry mine/{label}\n")
|
||
return False
|
||
time.sleep(1.8)
|
||
return True
|
||
if kind == "mine_gear":
|
||
# 设置齿轮无障碍文案，固定点右上角
|
||
try:
|
||
if is_comment_feed():
|
||
back()
|
||
time.sleep(0.6)
|
||
except Exception:
|
||
pass
|
||
tap_tab("我的")
|
||
time.sleep(1.2)
|
||
escape_modals(max_rounds=3)
|
||
print(" TAP-XY mine gear (1010,180)", flush=True)
|
||
tap_xy(1010, 180, wait=1.5)
|
||
try:
|
||
fls = [first_line(x) for x in labels()]
|
||
if "设置" in fls or "个人资料" in fls or "账号安全" in fls:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- MISS entry mine_gear/设置\n")
|
||
return False
|
||
if kind == "mine_mall_order":
|
||
# 订单入口在测测商城 → 个人中心（H5，BACK/底栏常无效）
|
||
try:
|
||
fls0 = [first_line(x) for x in labels()]
|
||
b0 = "\n".join(labels())
|
||
if any(k in b0 for k in ("订单列表", "测测商城", "个人中心", "weimob")):
|
||
print(" leave mall H5 → force-stop", flush=True)
|
||
adb("shell", "am", "force-stop", "com.xxwolo.cc5", timeout=15)
|
||
time.sleep(1)
|
||
adb("shell", "am", "start", "-n", ACT, timeout=15)
|
||
time.sleep(3.5)
|
||
except Exception:
|
||
pass
|
||
try:
|
||
if is_comment_feed():
|
||
back()
|
||
time.sleep(0.6)
|
||
except Exception:
|
||
pass
|
||
tap_tab("我的")
|
||
time.sleep(1.2)
|
||
escape_modals(max_rounds=3)
|
||
# 「商城」入口 Y 会随头部运营位变化，放宽范围
|
||
if not (tap_exact("商城", ymin=700, ymax=1600) or tap_contains("商城", ymin=700, ymax=1600)):
|
||
# 动态找中心，失败再用兜底
|
||
xy = None
|
||
try:
|
||
for e in dump().get("elements") or []:
|
||
if first_line(lab(e)) == "商城" and e.get("clickable"):
|
||
xy = center(e)
|
||
break
|
||
except Exception:
|
||
pass
|
||
if xy:
|
||
print(f" TAP-XY 商城 {xy}", flush=True)
|
||
tap_xy(*xy, wait=2.0)
|
||
else:
|
||
print(" TAP-XY 商城 fallback (782,1250)", flush=True)
|
||
tap_xy(782, 1250, wait=2.0)
|
||
time.sleep(2.5)
|
||
ok_mall = False
|
||
for _ in range(4):
|
||
try:
|
||
fls = [first_line(x) for x in labels()]
|
||
if "测测商城" in fls or "分类" in fls or "购物车" in fls or "个人中心" in fls:
|
||
ok_mall = True
|
||
break
|
||
except Exception:
|
||
pass
|
||
time.sleep(1.0)
|
||
if not ok_mall:
|
||
# 仍继续：部分机型 WebView dump 延迟，坐标进个人中心再校验订单
|
||
print(" warn: mall labels not confirmed, continue XY", flush=True)
|
||
# 商城底栏「我的」常不可 clickable
|
||
print(" TAP-XY mall 我的 (946,2280)", flush=True)
|
||
tap_xy(946, 2280, wait=1.8)
|
||
# 点全部订单 / 商城订单
|
||
if not (
|
||
tap_exact("全部订单", ymin=700, ymax=1400)
|
||
or tap_contains("全部订单", ymin=700, ymax=1400)
|
||
or tap_exact("商城订单", ymin=700, ymax=1400)
|
||
or tap_contains("商城订单", ymin=700, ymax=1400)
|
||
):
|
||
print(" TAP-XY 全部订单 (926,884)", flush=True)
|
||
tap_xy(926, 884, wait=1.5)
|
||
time.sleep(1.2)
|
||
try:
|
||
b = "\n".join(labels())
|
||
if any(k in b for k in ("订单", "待付款", "待收货", "退款", "订单列表")):
|
||
return True
|
||
except Exception:
|
||
pass
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write("- MISS entry mine_mall_order/订单\n")
|
||
return False
|
||
if kind == "tab":
|
||
tap_tab(label)
|
||
time.sleep(1.5)
|
||
return True
|
||
return False
|
||
|
||
|
||
def run_module(mod: dict):
|
||
if STATE.shots >= MAX_SHOTS:
|
||
return
|
||
if not enter_module(mod):
|
||
return
|
||
try:
|
||
explore_module(mod)
|
||
except Exception as ex:
|
||
import traceback
|
||
print(f" explore ERR: {ex}", flush=True)
|
||
traceback.print_exc()
|
||
# leave module — 底栏 Tab / WebView 上 BACK 常无效，直接回首页
|
||
try:
|
||
tap_tab("首页")
|
||
time.sleep(0.8)
|
||
go_home()
|
||
except Exception as ex:
|
||
print(f" leave module fail: {ex}", flush=True)
|
||
ensure_app()
|
||
|
||
|
||
def write_manifest():
|
||
pngs = sorted(str(p.relative_to(OUT)) for p in OUT.rglob("*.png"))
|
||
MANIFEST.write_text(
|
||
json.dumps(
|
||
{
|
||
"count": len(pngs),
|
||
"shots_logged": STATE.shots,
|
||
"modules": [m["id"] for m in MODULES],
|
||
"screenshots": pngs,
|
||
"log": STATE.log,
|
||
},
|
||
ensure_ascii=False,
|
||
indent=2,
|
||
),
|
||
encoding="utf-8",
|
||
)
|
||
print("DONE", len(pngs), "pngs", flush=True)
|
||
|
||
|
||
def main():
|
||
import faulthandler
|
||
faulthandler.enable()
|
||
OUT.mkdir(parents=True, exist_ok=True)
|
||
NOTES.write_text(
|
||
f"# deep whitelist crawl {time.strftime('%F %T')}\n\n"
|
||
f"modules={len(MODULES)} mode=flat max_shots={MAX_SHOTS}\n\n",
|
||
encoding="utf-8",
|
||
)
|
||
print(f"START out={OUT} mode=flat", flush=True)
|
||
adb("shell", "input", "keyevent", "KEYCODE_WAKEUP")
|
||
ensure_app()
|
||
escape_modals(max_rounds=4)
|
||
only = set(sys.argv[1:]) if len(sys.argv) > 1 else None
|
||
for mod in MODULES:
|
||
if only and mod["id"] not in only:
|
||
continue
|
||
if mod["id"] in SKIP_MODULES:
|
||
print(f"\n>>> skip {mod['id']} (SKIP_MODULES)", flush=True)
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- SKIP module {mod['id']}\n")
|
||
continue
|
||
try:
|
||
print(f"\n>>> begin {mod['id']} shots={STATE.shots}", flush=True)
|
||
run_module(mod)
|
||
print(f"<<< end {mod['id']} shots={STATE.shots}", flush=True)
|
||
except Exception as ex:
|
||
import traceback
|
||
print(f"ERR {mod['id']}: {ex}", flush=True)
|
||
traceback.print_exc()
|
||
with NOTES.open("a", encoding="utf-8") as f:
|
||
f.write(f"- ERR {mod['id']}: {ex}\n")
|
||
ensure_app()
|
||
try:
|
||
go_home()
|
||
except Exception:
|
||
ensure_app()
|
||
write_manifest()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
main()
|
||
except Exception:
|
||
import traceback
|
||
traceback.print_exc()
|
||
raise
|