落地 synastry/star/ask API 与 H5 页面,补齐 cece-frontend-re complete-design 证据文档,并加入 Android 模拟器截图抓取脚本。 Co-authored-by: Cursor <cursoragent@cursor.com>
585 lines
18 KiB
Python
585 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""Overnight free-page capture for CeCe after login. Avoid paywalls."""
|
|
|
|
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"
|
|
ACTIVITY = f"{PKG}/com.cece.app.MainActivity"
|
|
DATE = time.strftime("%Y%m%d")
|
|
OUT = ROOT / ".tmp" / "cece-validation" / DATE / "free"
|
|
NOTES = OUT / "CAPTURE_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", "")
|
|
)
|
|
|
|
# Strong paywall / cashier signals only (avoid false positive on feed prices)
|
|
PAYWALL_STRONG = (
|
|
"确认支付",
|
|
"微信支付",
|
|
"支付宝支付",
|
|
"苹果支付",
|
|
"立即支付",
|
|
"连续包月",
|
|
"确认协议并支付",
|
|
)
|
|
|
|
# Home tool entries to open (free teaser pages OK)
|
|
HOME_TOOLS = [
|
|
"I人E人",
|
|
"星座",
|
|
"星盘",
|
|
"生辰",
|
|
"缘分合盘",
|
|
"紫微",
|
|
"陪伴小星",
|
|
"倾诉",
|
|
"智慧卡",
|
|
"星盘报告",
|
|
"生辰历",
|
|
"灵魂伴侣",
|
|
"AI玩法广场",
|
|
"更多",
|
|
]
|
|
|
|
|
|
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, "raw": (r.stdout or "")[:500]}
|
|
|
|
|
|
def device_ready():
|
|
return bool(re.search(r"emulator-\d+\s+device", adb("devices").stdout or ""))
|
|
|
|
|
|
def ensure_device():
|
|
if device_ready():
|
|
return
|
|
print("waiting for device…")
|
|
for _ in range(60):
|
|
if device_ready():
|
|
return
|
|
time.sleep(2)
|
|
raise SystemExit("no device")
|
|
|
|
|
|
def label(e):
|
|
return (e.get("text") or e.get("contentDesc") or e.get("contentDescription") or "").strip()
|
|
|
|
|
|
def elements(data):
|
|
return data.get("elements") or []
|
|
|
|
|
|
def dump():
|
|
return tool("ui", "dump")
|
|
|
|
|
|
def shot(rel):
|
|
path = OUT / rel
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = tool("screenshot", "--out", str(path))
|
|
print("SHOT", rel, data.get("bytes"), data.get("error"))
|
|
return path
|
|
|
|
|
|
def save_dump(rel, data=None):
|
|
path = OUT / rel
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = data or dump()
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return data
|
|
|
|
|
|
def blob(data):
|
|
return json.dumps(data, ensure_ascii=False)
|
|
|
|
|
|
def center(e):
|
|
c = e.get("center")
|
|
if isinstance(c, dict):
|
|
return int(c["x"]), int(c["y"])
|
|
b = e.get("bounds") or {}
|
|
if isinstance(b, dict) and "width" in b:
|
|
return int(b["x"] + b["width"] / 2), int(b["y"] + b["height"] / 2)
|
|
return None
|
|
|
|
|
|
def area(e):
|
|
b = e.get("bounds") or {}
|
|
return int(b.get("width") or 0) * int(b.get("height") or 0)
|
|
|
|
|
|
def tap_xy(x, y):
|
|
adb("shell", "input", "tap", str(x), str(y))
|
|
time.sleep(1.8)
|
|
|
|
|
|
def tap_el(e):
|
|
xy = center(e)
|
|
if not xy:
|
|
return False
|
|
print(f" TAP {xy} {label(e)[:60]!r}")
|
|
tap_xy(*xy)
|
|
return True
|
|
|
|
|
|
def find_exact(data, name, clickable_only=True):
|
|
hits = []
|
|
for e in elements(data):
|
|
lab = label(e)
|
|
# tab badges: "消息\n6", "在线\n限免"
|
|
first = lab.split("\n", 1)[0].strip()
|
|
if lab != name and first != name:
|
|
continue
|
|
if clickable_only and not e.get("clickable"):
|
|
continue
|
|
if area(e) > 1080 * 900:
|
|
continue
|
|
hits.append(e)
|
|
hits.sort(key=area)
|
|
return hits
|
|
|
|
|
|
def find_contains(data, needle, clickable_only=True):
|
|
hits = []
|
|
for e in elements(data):
|
|
lab = label(e)
|
|
if needle not in lab and needle not in lab.split("\n", 1)[0]:
|
|
continue
|
|
if clickable_only and not e.get("clickable"):
|
|
continue
|
|
if area(e) > 1080 * 900:
|
|
continue
|
|
hits.append(e)
|
|
hits.sort(key=area)
|
|
return hits
|
|
|
|
|
|
def tap_text(name, exact=True):
|
|
data = dump()
|
|
hits = find_exact(data, name) if exact else find_contains(data, name)
|
|
if not hits:
|
|
print(" MISS", name)
|
|
return False
|
|
return tap_el(hits[0])
|
|
|
|
|
|
def back(n=1):
|
|
for _ in range(n):
|
|
adb("shell", "input", "keyevent", "4")
|
|
time.sleep(1.2)
|
|
|
|
|
|
def is_paywall(data=None):
|
|
data = data or dump()
|
|
b = blob(data)
|
|
if any(h in b for h in PAYWALL_STRONG):
|
|
return True
|
|
# membership cashier sheet
|
|
if ("开通会员" in b or "升级会员" in b) and ("立即开通" in b or "立即购买" in b):
|
|
return True
|
|
if "立即解锁" in b and ("¥" in b or "元" in b):
|
|
return True
|
|
return False
|
|
|
|
|
|
def dismiss_overlays(max_rounds=8):
|
|
for i in range(max_rounds):
|
|
data = dump()
|
|
b = blob(data)
|
|
acted = False
|
|
# Daily login / return gift modal — close via X under sheet (not claim)
|
|
if "立即领取" in b and ("每日登录" in b or "回归福利" in b or "头像挂件" in b or "缘分报告" in b):
|
|
print("dismiss daily-reward via X taps")
|
|
for y in (1680, 1720, 1760, 1800, 1640, 1600):
|
|
tap_xy(540, y)
|
|
data = dump()
|
|
b = blob(data)
|
|
if "立即领取" not in b:
|
|
acted = True
|
|
break
|
|
if "立即领取" in b:
|
|
# last resort: claim free reward to clear
|
|
print("dismiss daily-reward via claim")
|
|
tap_text("立即领取", exact=True)
|
|
time.sleep(2)
|
|
acted = True
|
|
if acted:
|
|
continue
|
|
for name in (
|
|
"不再提醒",
|
|
"我知道了",
|
|
"知道了",
|
|
"暂不开启",
|
|
"以后再说",
|
|
"取消",
|
|
"跳过",
|
|
"允许",
|
|
"使用时允许",
|
|
"仅在使用中允许",
|
|
"仅限这一次",
|
|
"同意",
|
|
):
|
|
hits = find_exact(data, name) or find_contains(data, name)
|
|
hits = [e for e in hits if area(e) < 1080 * 400]
|
|
if hits:
|
|
if name == "同意" and "温馨提示" not in b:
|
|
continue
|
|
print("dismiss", name)
|
|
tap_el(hits[0])
|
|
acted = True
|
|
break
|
|
if not acted:
|
|
break
|
|
time.sleep(1)
|
|
|
|
|
|
def go_home():
|
|
dismiss_overlays()
|
|
data = dump()
|
|
if any(label(e).split("\n", 1)[0] == "首页" and (center(e) or (0, 0))[1] >= 2200 for e in elements(data)):
|
|
tap_bottom_tab("首页", (108, 2255))
|
|
else:
|
|
adb("shell", "am", "start", "-n", ACTIVITY)
|
|
time.sleep(4)
|
|
dismiss_overlays()
|
|
tap_bottom_tab("首页", (108, 2255))
|
|
time.sleep(1.5)
|
|
dismiss_overlays()
|
|
|
|
|
|
def capture_screen(rel, note=""):
|
|
dismiss_overlays()
|
|
data = dump()
|
|
if is_paywall(data):
|
|
print("PAYWALL at", rel, "— back out")
|
|
shot(rel.replace(".png", "-paywall.png"))
|
|
save_dump(rel.replace(".png", "-paywall.json"), data)
|
|
back()
|
|
return "paywall"
|
|
shot(rel)
|
|
save_dump(rel.replace(".png", ".json"), data)
|
|
if note:
|
|
with NOTES.open("a", encoding="utf-8") as f:
|
|
f.write(f"- `{rel}` {note}\n")
|
|
labels = [label(e) for e in elements(data) if label(e)]
|
|
(OUT / rel.replace(".png", ".labels.txt")).write_text(
|
|
"\n".join(labels), encoding="utf-8"
|
|
)
|
|
return "ok"
|
|
|
|
|
|
def tap_bottom_tab(name: str, fb: tuple[int, int]) -> None:
|
|
"""Only tap elements in the bottom tab bar (y>=2200)."""
|
|
data = dump()
|
|
hits = []
|
|
for e in elements(data):
|
|
lab = label(e)
|
|
first = lab.split("\n", 1)[0].strip()
|
|
if first != name and name not in first:
|
|
continue
|
|
xy = center(e)
|
|
if not xy or xy[1] < 2200:
|
|
continue
|
|
if area(e) > 400 * 400:
|
|
continue
|
|
hits.append(e)
|
|
if hits:
|
|
hits.sort(key=lambda e: -(center(e) or (0, 0))[1])
|
|
tap_el(hits[0])
|
|
else:
|
|
print(" tab fallback", name, fb)
|
|
tap_xy(*fb)
|
|
|
|
|
|
def capture_tabs():
|
|
tabs = [
|
|
("首页", "tabs/01-home.png", (108, 2255)),
|
|
("消息", "tabs/02-message.png", (324, 2255)),
|
|
("问", "tabs/03-ask.png", (540, 2255)), # may miss label
|
|
("在线", "tabs/04-online.png", (756, 2255)),
|
|
("我的", "tabs/05-mine.png", (972, 2255)),
|
|
]
|
|
for name, rel, fb in tabs:
|
|
print("=== TAB", name)
|
|
if name == "首页":
|
|
go_home()
|
|
# always leave nested pages first
|
|
back(1)
|
|
dismiss_overlays()
|
|
tap_bottom_tab(name, fb)
|
|
time.sleep(2.2)
|
|
# ask may open sub UI
|
|
if name == "问":
|
|
capture_screen(rel, "ask tab root")
|
|
for sub in ("测测AI", "真人1v1"):
|
|
d = dump()
|
|
if find_contains(d, sub):
|
|
# prefer header tab (y < 400)
|
|
hits = [
|
|
e
|
|
for e in find_contains(d, sub)
|
|
if (center(e) or (0, 9999))[1] < 500
|
|
]
|
|
if hits:
|
|
tap_el(hits[0])
|
|
else:
|
|
tap_text(sub, exact=False)
|
|
time.sleep(2)
|
|
capture_screen(f"tabs/03-ask-{sub}.png", f"ask sub {sub}")
|
|
# tools sheet on ask
|
|
d = dump()
|
|
if find_exact(d, "工具") or find_contains(d, "工具"):
|
|
hits = [e for e in find_contains(d, "工具") if (center(e) or (0, 0))[1] > 1800]
|
|
if hits:
|
|
tap_el(hits[0])
|
|
time.sleep(2)
|
|
if not is_paywall():
|
|
capture_screen("ask/tools-sheet.png", "ask tools sheet")
|
|
back()
|
|
elif name == "消息":
|
|
capture_screen(rel, "message tab")
|
|
# open first conversation if free list item
|
|
d = dump()
|
|
for e in elements(d):
|
|
lab = label(e)
|
|
if lab and e.get("clickable") and (center(e) or (0, 0))[1] < 2000:
|
|
if any(x in lab for x in ("系统", "通知", "客服", "互动", "赞")):
|
|
tap_el(e)
|
|
time.sleep(2)
|
|
if capture_screen("tabs/02-message-detail.png", lab[:40]) != "paywall":
|
|
back()
|
|
break
|
|
elif name == "在线":
|
|
# 限免可能直达「向TA提问」付费页:关掉券弹层并退回列表
|
|
for _ in range(4):
|
|
d = dump()
|
|
b = blob(d)
|
|
if "我知道了" in b:
|
|
tap_text("我知道了", exact=True)
|
|
time.sleep(1)
|
|
continue
|
|
if "向TA提问" in b or "专享价购买" in b:
|
|
print(" online landed on consult — back")
|
|
back()
|
|
continue
|
|
break
|
|
capture_screen(rel, "online tab")
|
|
adb("shell", "input", "swipe", "540", "1800", "540", "900", "350")
|
|
time.sleep(1.5)
|
|
capture_screen("tabs/04-online-scroll.png", "online scrolled")
|
|
elif name == "我的":
|
|
capture_screen(rel, "mine tab")
|
|
adb("shell", "input", "swipe", "540", "1800", "540", "900", "350")
|
|
time.sleep(1.5)
|
|
capture_screen("tabs/05-mine-scroll.png", "mine scrolled")
|
|
# open free settings-like items
|
|
for item in (
|
|
"设置",
|
|
"我的订单",
|
|
"优惠券",
|
|
"我的档案",
|
|
"生命档案",
|
|
"我的测试",
|
|
"我的报告",
|
|
"收藏",
|
|
"帮助",
|
|
"关于",
|
|
"钱包",
|
|
"会员中心",
|
|
):
|
|
go_tab_mine()
|
|
d = dump()
|
|
if find_contains(d, item):
|
|
tap_text(item, exact=False)
|
|
time.sleep(2)
|
|
st = capture_screen(f"mine/{item}.png", item)
|
|
if st == "paywall":
|
|
continue
|
|
back()
|
|
else:
|
|
capture_screen(rel, "home tab")
|
|
|
|
|
|
def go_tab_mine():
|
|
dismiss_overlays()
|
|
tap_bottom_tab("我的", (972, 2255))
|
|
time.sleep(1.5)
|
|
|
|
|
|
def capture_home_tools():
|
|
for name in HOME_TOOLS:
|
|
print("=== TOOL", name)
|
|
go_home()
|
|
time.sleep(1)
|
|
data = dump()
|
|
hits = find_contains(data, name) if name != "更多" else find_exact(data, "更多")
|
|
if not hits:
|
|
# scroll home a bit to reveal
|
|
adb("shell", "input", "swipe", "540", "1600", "540", "900", "300")
|
|
time.sleep(1)
|
|
data = dump()
|
|
hits = find_contains(data, name) or find_exact(data, name)
|
|
if not hits:
|
|
print(" skip missing", name)
|
|
with NOTES.open("a", encoding="utf-8") as f:
|
|
f.write(f"- MISSING tool entry: {name}\n")
|
|
continue
|
|
# prefer smaller icon cell
|
|
hits.sort(key=area)
|
|
tap_el(hits[0])
|
|
time.sleep(2.5)
|
|
safe = re.sub(r"[^\w\u4e00-\u9fff\-]+", "_", name)[:40]
|
|
st = capture_screen(f"tools/{safe}.png", f"tool {name}")
|
|
if st == "paywall":
|
|
back()
|
|
continue
|
|
# one level deeper free exploration for 更多
|
|
if name == "更多":
|
|
d = dump()
|
|
labels = []
|
|
for e in elements(d):
|
|
lab = label(e)
|
|
if lab and e.get("clickable") and 0 < area(e) < 400 * 400:
|
|
labels.append(lab)
|
|
# unique preserve order
|
|
seen = set()
|
|
uniq = []
|
|
for lab in labels:
|
|
if lab in seen or lab in ("更多", "首页", "消息", "在线", "我的", "问"):
|
|
continue
|
|
seen.add(lab)
|
|
uniq.append(lab)
|
|
for lab in uniq[:25]:
|
|
print(" more-item", lab)
|
|
if not tap_text(lab, exact=True):
|
|
if not tap_text(lab, exact=False):
|
|
continue
|
|
time.sleep(2)
|
|
safe2 = re.sub(r"[^\w\u4e00-\u9fff\-]+", "_", lab)[:40]
|
|
st2 = capture_screen(f"tools/more/{safe2}.png", f"more>{lab}")
|
|
back()
|
|
time.sleep(1)
|
|
# ensure still on more page
|
|
d2 = dump()
|
|
if not find_contains(d2, "更多") and not any(
|
|
x in blob(d2) for x in uniq[:5]
|
|
):
|
|
# re-open more
|
|
go_home()
|
|
tap_text("更多")
|
|
time.sleep(2)
|
|
else:
|
|
# optional: capture one sub-tab if present without paying
|
|
d = dump()
|
|
for sub in ("今日", "本周", "概览", "免费", "详情", "解读"):
|
|
if find_exact(d, sub) or find_contains(d, sub):
|
|
# avoid 立即解锁
|
|
if is_paywall(d):
|
|
break
|
|
tap_text(sub, exact=False)
|
|
time.sleep(1.5)
|
|
if is_paywall():
|
|
back()
|
|
break
|
|
capture_screen(f"tools/{safe}-{sub}.png", f"{name}>{sub}")
|
|
break
|
|
back()
|
|
time.sleep(1)
|
|
|
|
|
|
def capture_ask_deep():
|
|
print("=== ASK deep")
|
|
go_home()
|
|
tap_xy(540, 2255)
|
|
time.sleep(2)
|
|
capture_screen("ask/01-root.png", "ask root")
|
|
d = dump()
|
|
for name in ("测测AI", "真人1v1", "历史", "新对话", "换一个"):
|
|
if find_contains(d, name) or find_exact(d, name):
|
|
tap_text(name, exact=False)
|
|
time.sleep(2)
|
|
if is_paywall():
|
|
shot(f"ask/{name}-paywall.png")
|
|
back()
|
|
continue
|
|
capture_screen(f"ask/{name}.png", name)
|
|
d = dump()
|
|
|
|
|
|
def capture_home_feed():
|
|
print("=== HOME feed scroll")
|
|
go_home()
|
|
capture_screen("home/01-top.png", "home top")
|
|
for i in range(1, 4):
|
|
adb("shell", "input", "swipe", "540", "1900", "540", "800", "400")
|
|
time.sleep(1.5)
|
|
capture_screen(f"home/0{i+1}-scroll.png", f"home scroll {i}")
|
|
|
|
|
|
def write_index():
|
|
pngs = sorted(str(p.relative_to(OUT)) for p in OUT.rglob("*.png"))
|
|
manifest = {
|
|
"date": DATE,
|
|
"package": PKG,
|
|
"scope": "free-pages-only",
|
|
"count": len(pngs),
|
|
"screenshots": pngs,
|
|
}
|
|
(OUT / "manifest.json").write_text(
|
|
json.dumps(manifest, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
print("TOTAL PNG", len(pngs))
|
|
|
|
|
|
def main():
|
|
OUT.mkdir(parents=True, exist_ok=True)
|
|
NOTES.write_text(
|
|
f"# CeCe free capture\n\nStarted: {time.strftime('%Y-%m-%d %H:%M:%S')}\n\n",
|
|
encoding="utf-8",
|
|
)
|
|
ensure_device()
|
|
adb("shell", "input", "keyevent", "KEYCODE_WAKEUP")
|
|
dismiss_overlays()
|
|
# prefer 不再提醒 if minor mode
|
|
tap_text("不再提醒")
|
|
dismiss_overlays()
|
|
capture_tabs()
|
|
capture_home_feed()
|
|
capture_home_tools()
|
|
capture_ask_deep()
|
|
write_index()
|
|
with NOTES.open("a", encoding="utf-8") as f:
|
|
f.write(f"\nFinished: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
|
|
print("DONE", OUT)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|