#!/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