jackyu66gitandCursor
bd22d9dddd
落地 synastry/star/ask API 与 H5 页面，补齐 cece-frontend-re complete-design 证据文档，并加入 Android 模拟器截图抓取脚本。 Co-authored-by: Cursor <cursoragent@cursor.com>
520 lines
17 KiB
Python
520 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""Full-page CeCe screenshot crawl (all reachable screens; paywall shot then back)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
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"
|
|
OUT = ROOT / ".tmp" / "cece-validation" / time.strftime("%Y%m%d") / "all"
|
|
NOTES = OUT / "NOTES.md"
|
|
|
|
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", "")
|
|
)
|
|
|
|
MORE_ITEMS = [
|
|
"星盘报告", "测试", "倾诉", "心情小镇", "幸运地图", "星宿", "爱情树",
|
|
"灵魂伴侣", "聊天分析", "关系网", "商城", "解读", "沙盘", "缘分合盘",
|
|
"星座", "生辰", "紫微", "政余", "灵数", "生肖", "玛雅图腾", "人类图",
|
|
"星盘", "生辰历", "紫微历", "政余历", "星骰", "智慧卡", "日历",
|
|
]
|
|
|
|
HOME_GRID = [
|
|
"I人E人", "星座", "星盘", "生辰", "缘分合盘", "紫微",
|
|
"陪伴小星", "倾诉", "智慧卡", "星盘报告", "生辰历", "灵魂伴侣",
|
|
"AI玩法广场", "更多",
|
|
]
|
|
|
|
MINE_ITEMS = [
|
|
"设置", "订单", "优惠券", "会员", "档案", "测试", "报告", "收藏",
|
|
"帮助", "关于", "钱包", "客服", "消息通知", "签到", "邀请", "客服与帮助",
|
|
]
|
|
|
|
CHART_TABS = ["天象", "本命", "行运", "三限", "次限", "日返", "月返", "法达", "推运", "合盘"]
|
|
|
|
|
|
def sh(args, timeout=120):
|
|
return subprocess.run(args, capture_output=True, text=True, timeout=timeout)
|
|
|
|
|
|
def adb(*args, timeout=120):
|
|
return sh(["adb", *args], timeout=timeout)
|
|
|
|
|
|
def tool(*args, timeout=120):
|
|
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")
|
|
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)
|
|
time.sleep(5)
|
|
|
|
|
|
def dump():
|
|
return tool("ui", "dump")
|
|
|
|
|
|
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 tap_xy(x, y, wait=1.5):
|
|
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.7)
|
|
|
|
|
|
def safe_name(s: str) -> str:
|
|
return re.sub(r"[^\w\u4e00-\u9fff\-]+", "_", s)[:48]
|
|
|
|
|
|
def shot(rel: str, note: str = ""):
|
|
ensure_app()
|
|
path = OUT / rel
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = tool("screenshot", "--out", str(path))
|
|
labs = labels()
|
|
(OUT / (rel.replace(".png", ".labels.txt"))).write_text("\n".join(labs), encoding="utf-8")
|
|
print(f"SHOT {rel} bytes={data.get('bytes')} note={note}", flush=True)
|
|
with NOTES.open("a", encoding="utf-8") as f:
|
|
f.write(f"- `{rel}` {note}\n")
|
|
return path
|
|
|
|
|
|
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)
|
|
first = t.split("\n", 1)[0]
|
|
if first != name and t != name:
|
|
continue
|
|
if not e.get("clickable"):
|
|
continue
|
|
xy = center(e)
|
|
if not xy:
|
|
continue
|
|
if 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):
|
|
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=12) -> bool:
|
|
for i in range(max_rounds):
|
|
ensure_app()
|
|
b = blob()
|
|
labs = labels()
|
|
print(f" escape[{i}] {labs[:4]}", flush=True)
|
|
if any(x.split("\n", 1)[0] == "首页" for x in labs) and not any(
|
|
k in b
|
|
for k in (
|
|
"隐私政策更新",
|
|
"温馨提示",
|
|
"立即领取",
|
|
"打开通知",
|
|
"开通SVIP",
|
|
"向TA提问",
|
|
"确认并支付",
|
|
"专享价购买",
|
|
"未成年人模式",
|
|
)
|
|
):
|
|
return True
|
|
if "同意" in labs and any(k in b for k in ("隐私", "温馨提示")):
|
|
tap_exact("同意")
|
|
continue
|
|
if "不再提醒" in labs:
|
|
tap_exact("不再提醒")
|
|
continue
|
|
if "我知道了" in labs:
|
|
tap_exact("我知道了")
|
|
continue
|
|
if "取消" in labs and "通知" in b:
|
|
tap_exact("取消")
|
|
continue
|
|
if "立即领取" in labs:
|
|
for y in (1680, 1720, 1760, 1800):
|
|
tap_xy(540, y, wait=0.25)
|
|
continue
|
|
# paywall / consult / membership — capture once then back
|
|
if any(k in b for k in ("开通SVIP", "向TA提问", "确认并支付", "专享价购买", "微信支付", "支付宝支付")):
|
|
back()
|
|
continue
|
|
if "关闭" in labs:
|
|
# avoid full-screen close; try small
|
|
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
|
|
back()
|
|
return False
|
|
|
|
|
|
def go_home():
|
|
escape_modals()
|
|
tap_tab("首页", (108, 2255))
|
|
time.sleep(1)
|
|
escape_modals()
|
|
|
|
|
|
def capture_page(rel, note=""):
|
|
"""Shot current page; if cashier, still save as *-cashier then back."""
|
|
b = blob()
|
|
if any(k in b for k in ("确认并支付", "专享价购买", "微信支付", "支付宝支付")):
|
|
shot(rel.replace(".png", "-cashier.png"), note + " [cashier]")
|
|
back(2)
|
|
escape_modals()
|
|
return "cashier"
|
|
shot(rel, note)
|
|
return "ok"
|
|
|
|
|
|
def crawl_tabs():
|
|
tabs = [
|
|
("首页", (108, 2255), "tabs/01-home"),
|
|
("消息", (324, 2255), "tabs/02-message"),
|
|
("问", (540, 2255), "tabs/03-ask"),
|
|
("在线", (756, 2255), "tabs/04-online"),
|
|
("我的", (972, 2255), "tabs/05-mine"),
|
|
]
|
|
for name, fb, base in tabs:
|
|
print("=== TAB", name, flush=True)
|
|
go_home() if name == "首页" else None
|
|
escape_modals()
|
|
tap_tab(name, fb)
|
|
time.sleep(2)
|
|
if name == "在线":
|
|
for _ in range(4):
|
|
b = blob()
|
|
if "我知道了" in b:
|
|
tap_exact("我知道了")
|
|
continue
|
|
if "向TA提问" in b:
|
|
shot(f"{base}-ask-ta.png", "online->向TA提问")
|
|
back()
|
|
continue
|
|
break
|
|
capture_page(f"{base}.png", f"tab {name}")
|
|
if name == "消息":
|
|
if tap_contains("系统通知", ymax=900):
|
|
time.sleep(1.5)
|
|
capture_page(f"{base}-系统通知.png", "系统通知")
|
|
back()
|
|
if tap_contains("小星", ymax=1200):
|
|
time.sleep(1.5)
|
|
capture_page(f"{base}-小星.png", "小星")
|
|
back()
|
|
if name == "问":
|
|
for sub in ("测测AI", "真人1v1"):
|
|
if tap_contains(sub, ymax=500):
|
|
time.sleep(1.5)
|
|
capture_page(f"{base}-{safe_name(sub)}.png", sub)
|
|
# tools sheet
|
|
if tap_exact("工具", ymin=1800) or tap_contains("工具", ymin=1800):
|
|
time.sleep(1.5)
|
|
capture_page(f"{base}-工具sheet.png", "ask tools")
|
|
# tap a few tool chips if present
|
|
for t in ("星盘", "智慧卡", "合盘", "星骰"):
|
|
if tap_exact(t) or tap_contains(t):
|
|
time.sleep(1.5)
|
|
capture_page(f"{base}-工具-{t}.png", f"ask tool {t}")
|
|
back()
|
|
time.sleep(0.8)
|
|
back()
|
|
for btn in ("语音通话", "深度解读", "灵魂伴侣"):
|
|
if tap_exact(btn, ymin=1800) or tap_contains(btn, ymin=1800):
|
|
time.sleep(1.8)
|
|
capture_page(f"{base}-{btn}.png", btn)
|
|
escape_modals()
|
|
back()
|
|
if name == "在线":
|
|
adb("shell", "input", "swipe", "540", "1800", "540", "900", "350")
|
|
time.sleep(1.2)
|
|
capture_page(f"{base}-scroll.png", "online scroll")
|
|
# open filters
|
|
for f in ("综合排序", "全部工具", "筛选"):
|
|
if tap_exact(f) or tap_contains(f):
|
|
time.sleep(1.2)
|
|
capture_page(f"{base}-{f}.png", f)
|
|
back()
|
|
if name == "我的":
|
|
adb("shell", "input", "swipe", "540", "1800", "540", "900", "350")
|
|
time.sleep(1.2)
|
|
capture_page(f"{base}-scroll.png", "mine scroll")
|
|
|
|
|
|
def crawl_home_self_and_search():
|
|
print("=== HOME extras", flush=True)
|
|
go_home()
|
|
if tap_contains("了解TA") or tap_contains("输入生日"):
|
|
time.sleep(1.5)
|
|
capture_page("home/了解TA.png", "了解TA")
|
|
back()
|
|
if tap_exact("签到") or tap_contains("签到"):
|
|
time.sleep(1.5)
|
|
capture_page("home/签到.png", "签到")
|
|
escape_modals()
|
|
back()
|
|
# search bar approx top center
|
|
tap_xy(540, 180)
|
|
time.sleep(1.5)
|
|
capture_page("home/搜索.png", "search")
|
|
back()
|
|
# self card more
|
|
if tap_exact("更多", ymax=800):
|
|
time.sleep(1.5)
|
|
capture_page("home/自己-更多.png", "自己更多")
|
|
# profile tabs if present
|
|
for t in ("沙盘", "星座", "生辰", "星宿", "紫微", "自己"):
|
|
if tap_contains(t, ymax=600):
|
|
time.sleep(1.2)
|
|
capture_page(f"home/档案-{t}.png", f"档案 {t}")
|
|
back(2)
|
|
|
|
|
|
def crawl_grid_and_more():
|
|
for name in HOME_GRID:
|
|
print("=== GRID", name, flush=True)
|
|
go_home()
|
|
ok = tap_exact(name, ymax=2100) or tap_contains(name, ymax=2100)
|
|
if not ok:
|
|
adb("shell", "input", "swipe", "900", "1050", "200", "1050", "280")
|
|
time.sleep(0.8)
|
|
ok = tap_exact(name, ymax=2100) or tap_contains(name, ymax=2100)
|
|
if not ok:
|
|
with NOTES.open("a", encoding="utf-8") as f:
|
|
f.write(f"- MISS grid {name}\n")
|
|
continue
|
|
time.sleep(2)
|
|
capture_page(f"tools/{safe_name(name)}.png", f"grid {name}")
|
|
# chart internal tabs
|
|
if name in ("星盘", "生辰", "紫微", "星座"):
|
|
for t in CHART_TABS:
|
|
if tap_contains(t, ymax=700):
|
|
time.sleep(1.2)
|
|
capture_page(f"tools/{safe_name(name)}-{t}.png", f"{name}/{t}")
|
|
if name == "缘分合盘":
|
|
for t in ("添加档案", "直接选择档案合盘", "示例"):
|
|
if tap_contains(t):
|
|
time.sleep(1.5)
|
|
capture_page(f"tools/合盘-{safe_name(t)}.png", t)
|
|
escape_modals()
|
|
back()
|
|
break
|
|
if name == "更多":
|
|
# scroll more page and open every known item
|
|
seen = set()
|
|
for round_i in range(4):
|
|
data = dump()
|
|
items = []
|
|
for e in data.get("elements") or []:
|
|
t = lab(e)
|
|
if not t or not e.get("clickable"):
|
|
continue
|
|
if area(e) == 0 or area(e) > 360 * 360:
|
|
continue
|
|
first = t.split("\n", 1)[0]
|
|
if first in ("更多", "首页", "消息", "在线", "我的", "推荐", "工具"):
|
|
continue
|
|
if first in seen:
|
|
continue
|
|
items.append(first)
|
|
# prefer known list order
|
|
ordered = [x for x in MORE_ITEMS if x in items or any(x == i for i in items)]
|
|
ordered += [i for i in items if i not in ordered]
|
|
for t in ordered:
|
|
if t in seen:
|
|
continue
|
|
seen.add(t)
|
|
print(" more>", t, flush=True)
|
|
if not (tap_exact(t, ymax=2200) or tap_contains(t, ymax=2200)):
|
|
continue
|
|
time.sleep(2)
|
|
capture_page(f"tools/more/{safe_name(t)}.png", f"more>{t}")
|
|
escape_modals()
|
|
# if left more page, reopen
|
|
if not any(k in blob() for k in MORE_ITEMS[:5] + ["更多", "心情小镇", "沙盘"]):
|
|
go_home()
|
|
tap_exact("更多", ymax=2100) or tap_contains("更多", ymax=2100)
|
|
time.sleep(1.2)
|
|
adb("shell", "input", "swipe", "540", "1900", "540", "900", "350")
|
|
time.sleep(1.0)
|
|
else:
|
|
escape_modals()
|
|
back()
|
|
|
|
|
|
def crawl_mine_children():
|
|
print("=== MINE children", flush=True)
|
|
escape_modals()
|
|
tap_tab("我的", (972, 2255))
|
|
time.sleep(1.5)
|
|
capture_page("mine/00-root.png", "mine root")
|
|
# collect clickable rows
|
|
for round_i in range(3):
|
|
data = dump()
|
|
rows = []
|
|
for e in data.get("elements") or []:
|
|
t = lab(e)
|
|
if not t or not e.get("clickable"):
|
|
continue
|
|
xy = center(e)
|
|
if not xy or xy[1] < 350 or xy[1] > 2100:
|
|
continue
|
|
first = t.split("\n", 1)[0]
|
|
if first in ("首页", "消息", "在线", "我的", "问"):
|
|
continue
|
|
if area(e) > 1080 * 400:
|
|
continue
|
|
rows.append(first)
|
|
# unique
|
|
uniq = []
|
|
for r in rows:
|
|
if r not in uniq:
|
|
uniq.append(r)
|
|
targets = [x for x in MINE_ITEMS if any(x in u for u in uniq)]
|
|
targets += [u for u in uniq if u not in targets][:25]
|
|
for t in targets:
|
|
print(" mine>", t, flush=True)
|
|
escape_modals()
|
|
tap_tab("我的", (972, 2255))
|
|
time.sleep(1)
|
|
if not (tap_contains(t, ymin=350, ymax=2100) or tap_exact(t, ymin=350, ymax=2100)):
|
|
adb("shell", "input", "swipe", "540", "1800", "540", "900", "300")
|
|
time.sleep(0.8)
|
|
if not (tap_contains(t, ymin=350, ymax=2100) or tap_exact(t, ymin=350, ymax=2100)):
|
|
continue
|
|
time.sleep(2)
|
|
capture_page(f"mine/{safe_name(t)}.png", f"mine>{t}")
|
|
# membership sub tabs
|
|
if "会员" in t or t in ("VIP", "SVIP"):
|
|
for sub in ("VIP", "SVIP", "权益", "记录"):
|
|
if tap_exact(sub, ymax=600) or tap_contains(sub, ymax=600):
|
|
time.sleep(1.2)
|
|
capture_page(f"mine/会员-{sub}.png", f"会员/{sub}")
|
|
escape_modals()
|
|
back()
|
|
adb("shell", "input", "swipe", "540", "1800", "540", "900", "350")
|
|
time.sleep(1)
|
|
|
|
|
|
def write_manifest():
|
|
pngs = sorted(str(p.relative_to(OUT)) for p in OUT.rglob("*.png"))
|
|
(OUT / "manifest.json").write_text(
|
|
json.dumps({"count": len(pngs), "screenshots": pngs}, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
print("DONE", len(pngs), flush=True)
|
|
|
|
|
|
def main():
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
NOTES.write_text(f"# full crawl {time.strftime('%F %T')}\n\n", encoding="utf-8")
|
|
adb("shell", "input", "keyevent", "KEYCODE_WAKEUP")
|
|
ensure_app()
|
|
escape_modals()
|
|
crawl_tabs()
|
|
crawl_home_self_and_search()
|
|
crawl_grid_and_more()
|
|
crawl_mine_children()
|
|
write_manifest()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|