杨豪
581df23c97
- demo_collect.py: 线上多轮触发验证码 → 本地求解 → 标注图 （绿框目标位 + 黄色 mark 轮廓 + 蓝色拖动箭头 + 信息条） - 每题保存原图标注 PNG 与 mark PNG，index.png 汇总长图 - 用途：给客户演示识别效果；验证码刷新换题，不消耗 verify 配额 - results/ 为演示产物目录，不纳入版本库
433 lines
17 KiB
Python
433 lines
17 KiB
Python
"""演示采集：线上多轮触发验证码 → 本地识别 → results/ 标注图（不拖动）。
|
|
|
|
给客户演示识别效果：每轮采 1 题并求解，在大图上画出
|
|
- 绿色框 + 黄色轮廓：mark 应当重叠的目标位置
|
|
- 蓝色箭头：mark 从左侧初始位拖到目标位的偏移
|
|
- 文字：识别偏移 distance_px（原图 552 系）、缩放、旋转、置信分
|
|
|
|
用法: .venv/bin/python try/demo_collect.py [轮数=5] [每轮采题数=3]
|
|
产物: results/demo_YYYYMMDD_HHMMSS_N.{png,png} + results/index.png（汇总）
|
|
"""
|
|
import asyncio, base64, json, os, random, sys, time, urllib.request as ur
|
|
from pathlib import Path
|
|
|
|
import cv2
|
|
import numpy as np
|
|
import websockets
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT / "src"))
|
|
from captcha_solver import solve_slide
|
|
|
|
CDP = "http://127.0.0.1:9222"
|
|
SCALE = 340 / 552
|
|
OUT = ROOT / "results"
|
|
OUT.mkdir(exist_ok=True)
|
|
|
|
|
|
def random_phone():
|
|
return "1" + random.choice("3458") + "".join(random.choices("0123456789", k=9))
|
|
|
|
|
|
def random_password():
|
|
chars = "abcdefghijkmnpqrstuvwxyz"
|
|
caps = "ABCDEFGHJKLMNPQRSTUVWXYZ"
|
|
digits = "23456789"
|
|
syms = "!@#$%^&*"
|
|
pool = [random.choice(caps), random.choice(digits), random.choice(syms)]
|
|
pool += [random.choice(chars + caps + digits + syms) for _ in range(9)]
|
|
random.shuffle(pool)
|
|
return "".join(pool)
|
|
|
|
|
|
def _json(path):
|
|
try:
|
|
with ur.urlopen(f"{CDP}{path}", timeout=5) as r:
|
|
return json.load(r)
|
|
except (OSError, ValueError):
|
|
return None
|
|
|
|
|
|
async def demo_round(max_questions=3):
|
|
"""一轮：触发验证码 → 采 1 题求解 → 返回 (big_bytes, mark_bytes, result, tip_y) 或 None。"""
|
|
ver = _json("/json/version")
|
|
if not ver or not ver.get("webSocketDebuggerUrl"):
|
|
print("无法连接 CDP", file=sys.stderr)
|
|
return None
|
|
targets = _json("/json/list") or []
|
|
page = next((t for t in targets if t["type"] == "page" and "douyin" in t.get("url", "")), None)
|
|
if not page:
|
|
print("未找到 douyin 页面", file=sys.stderr)
|
|
return None
|
|
async with websockets.connect(ver["webSocketDebuggerUrl"], max_size=64 * 1024 * 1024) as ws:
|
|
mid = [0]
|
|
|
|
async def cdp(method, params=None, sid=None):
|
|
mid[0] += 1
|
|
m = {"id": mid[0], "method": method, "params": params or {}}
|
|
if sid:
|
|
m["sessionId"] = sid
|
|
await ws.send(json.dumps(m))
|
|
while True:
|
|
try:
|
|
r = json.loads(await ws.recv())
|
|
except (json.JSONDecodeError, asyncio.TimeoutError):
|
|
continue
|
|
if r.get("id") == mid[0]:
|
|
if "error" in r:
|
|
raise RuntimeError(f"{method}: {r['error']}")
|
|
return r.get("result", {})
|
|
|
|
psid = (await cdp("Target.attachToTarget", {"targetId": page["id"], "flatten": True}))["sessionId"]
|
|
await cdp("Runtime.enable", sid=psid)
|
|
|
|
async def js(expr):
|
|
r = await cdp("Runtime.evaluate",
|
|
{"expression": expr, "returnByValue": True, "awaitPromise": True}, psid)
|
|
return r.get("result", {}).get("value")
|
|
|
|
async def click(x, y):
|
|
for typ in ("mousePressed", "mouseReleased"):
|
|
await cdp("Input.dispatchMouseEvent",
|
|
{"type": typ, "x": x, "y": y, "button": "left",
|
|
"buttons": 1, "clickCount": 1}, psid)
|
|
|
|
async def type_text(text):
|
|
for ch in text:
|
|
await cdp("Input.dispatchKeyEvent", {"type": "keyDown", "text": ch}, psid)
|
|
await cdp("Input.dispatchKeyEvent", {"type": "keyUp"}, psid)
|
|
await asyncio.sleep(0.03)
|
|
|
|
def iframe_info():
|
|
return _json("/json/list") or []
|
|
|
|
# --- 触发验证码（已有 iframe 则复用） ---
|
|
has_iframe = any("rmc.bytedance.com" in t.get("url", "") for t in iframe_info())
|
|
if not has_iframe:
|
|
modal_raw = await js("""
|
|
(() => {
|
|
const modal = document.querySelector('[class*=login-panel], [id*=login-panel]');
|
|
const phone = [...document.querySelectorAll('input')].find(i => i.placeholder === '请输入手机号');
|
|
return JSON.stringify({open: !!modal, phoneVisible: !!(phone && phone.offsetWidth > 0)});
|
|
})()
|
|
""")
|
|
try:
|
|
modal_state = json.loads(modal_raw) if modal_raw else {}
|
|
except ValueError:
|
|
modal_state = {}
|
|
if not modal_state.get("open"):
|
|
await click(1860, 28)
|
|
await asyncio.sleep(2)
|
|
await js("""
|
|
(() => {
|
|
const other = [...document.querySelectorAll('*')].find(
|
|
e => e.children.length === 0 && /登录其他账号/.test(e.textContent || '') && e.offsetWidth > 0);
|
|
if (other) other.click();
|
|
})()
|
|
""")
|
|
await asyncio.sleep(1.5)
|
|
await js("""
|
|
(() => {
|
|
const els = [...document.querySelectorAll('*')].filter(
|
|
e => e.children.length === 0 && /密码登录/.test(e.textContent || '') && e.offsetWidth > 0);
|
|
if (els.length) els[0].click();
|
|
})()
|
|
""")
|
|
await asyncio.sleep(1.5)
|
|
captcha_ok = False
|
|
for attempt in range(6):
|
|
await js("""
|
|
(() => {
|
|
for (const ph of ['请输入手机号', '请输入密码']) {
|
|
const inp = [...document.querySelectorAll('input')].find(i => i.placeholder === ph);
|
|
if (inp) {
|
|
inp.focus();
|
|
const setter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
|
|
setter.call(inp, '');
|
|
inp.dispatchEvent(new Event('input', {bubbles: true}));
|
|
}
|
|
}
|
|
})()
|
|
""")
|
|
await asyncio.sleep(0.4)
|
|
geo_raw = await js("""
|
|
(() => {
|
|
const phone = [...document.querySelectorAll('input')].find(i => i.placeholder === '请输入手机号');
|
|
const cb = document.querySelector('span.f64a2kjb');
|
|
if (!phone) return 'null';
|
|
const pr = phone.getBoundingClientRect();
|
|
const cr = cb ? cb.getBoundingClientRect() : null;
|
|
return JSON.stringify({phone: [Math.round(pr.x + pr.width / 2), Math.round(pr.y + pr.height / 2)],
|
|
cb: cr ? [Math.round(cr.x + cr.width / 2), Math.round(cr.y + cr.height / 2)] : null});
|
|
})()
|
|
""")
|
|
try:
|
|
geo = json.loads(geo_raw) if geo_raw else None
|
|
except ValueError:
|
|
geo = None
|
|
if not geo:
|
|
print("登录弹窗没弹出")
|
|
return None
|
|
phone_no, pwd_str = random_phone(), random_password()
|
|
print(f"随机凭证: {phone_no}")
|
|
await click(*geo["phone"])
|
|
await asyncio.sleep(0.5)
|
|
await type_text(phone_no)
|
|
pwd_raw = await js("""
|
|
(() => {
|
|
const pwd = [...document.querySelectorAll('input')].find(i => i.placeholder === '请输入密码');
|
|
if (!pwd) return 'null';
|
|
const r = pwd.getBoundingClientRect();
|
|
return JSON.stringify([Math.round(r.x + r.width / 2), Math.round(r.y + r.height / 2)]);
|
|
})()
|
|
""")
|
|
try:
|
|
pwd_xy = json.loads(pwd_raw) if pwd_raw else None
|
|
except ValueError:
|
|
pwd_xy = None
|
|
if pwd_xy:
|
|
await click(*pwd_xy)
|
|
await asyncio.sleep(0.5)
|
|
await type_text(pwd_str)
|
|
if geo.get("cb"):
|
|
await click(*geo["cb"])
|
|
await asyncio.sleep(0.5)
|
|
await click(1165, 630)
|
|
for _ in range(25):
|
|
await asyncio.sleep(1.5)
|
|
if any("rmc.bytedance.com" in t.get("url", "") for t in iframe_info()):
|
|
captcha_ok = True
|
|
break
|
|
err_raw = await js("""
|
|
(() => {
|
|
const els = [...document.querySelectorAll('*')].filter(
|
|
e => e.children.length === 0 && e.textContent && e.offsetWidth > 0 &&
|
|
/仅支持验证码|不存在|格式有误|频繁|繁忙|错误/.test(e.textContent));
|
|
return els.length ? els[0].textContent.trim().slice(0, 60) : '';
|
|
})()
|
|
""")
|
|
if err_raw:
|
|
print(f"提交被拒: {err_raw}，换号重试")
|
|
break
|
|
if captcha_ok:
|
|
break
|
|
if not captcha_ok:
|
|
print("多次换号仍未触发验证码")
|
|
return None
|
|
print("验证码已弹出")
|
|
|
|
# --- 等 iframe 渲染完成 ---
|
|
for _ in range(30):
|
|
targets = iframe_info()
|
|
it = next((t for t in targets if "rmc.bytedance.com" in t.get("url", "")), None)
|
|
if not it:
|
|
await asyncio.sleep(1.0)
|
|
continue
|
|
probe = (await cdp("Target.attachToTarget", {"targetId": it["id"], "flatten": True}))["sessionId"]
|
|
await cdp("Runtime.enable", sid=probe)
|
|
r = await cdp("Runtime.evaluate", {"expression": "document.documentElement.offsetWidth",
|
|
"returnByValue": True}, probe)
|
|
w = r.get("result", {}).get("value") or 0
|
|
await cdp("Target.detachFromTarget", {"sessionId": probe})
|
|
if w >= 380:
|
|
break
|
|
await asyncio.sleep(1.0)
|
|
|
|
targets = iframe_info()
|
|
ifr = next(t for t in targets if "rmc.bytedance.com" in t.get("url", ""))
|
|
isid = (await cdp("Target.attachToTarget", {"targetId": ifr["id"], "flatten": True}))["sessionId"]
|
|
await cdp("Runtime.enable", sid=isid)
|
|
|
|
async def jse(expr, sid=isid):
|
|
r = await cdp("Runtime.evaluate",
|
|
{"expression": expr, "returnByValue": True, "awaitPromise": True}, sid)
|
|
return r.get("result", {}).get("value")
|
|
|
|
geom_raw = await js("""
|
|
(() => {
|
|
const f = [...document.querySelectorAll('iframe')].find(f => /rmc\\.bytedance\\.com/.test(f.src));
|
|
const r = f.getBoundingClientRect();
|
|
return JSON.stringify({x: Math.round(r.x), y: Math.round(r.y), w: f.offsetWidth, h: f.offsetHeight});
|
|
})()
|
|
""")
|
|
try:
|
|
geom = json.loads(geom_raw)
|
|
except ValueError:
|
|
print("iframe 坐标解析失败")
|
|
return None
|
|
IX, IY = geom["x"], geom["y"]
|
|
print("hook:", await jse("""
|
|
(() => {
|
|
window.__lastGet = null;
|
|
const oOpen = XMLHttpRequest.prototype.open, oSend = XMLHttpRequest.prototype.send;
|
|
XMLHttpRequest.prototype.open = function(m, u){ this.__u = u; return oOpen.apply(this, arguments); };
|
|
XMLHttpRequest.prototype.send = function(){
|
|
this.addEventListener('load', () => {
|
|
try {
|
|
if (/captcha\\/get/.test(this.__u || '')) window.__lastGet = JSON.parse(this.responseText);
|
|
} catch (e) {}
|
|
});
|
|
return oSend.apply(this, arguments);
|
|
};
|
|
return 'ok';
|
|
})()
|
|
"""))
|
|
|
|
# --- 采题（refresh 只耗 get，不耗 verify 配额）---
|
|
got = None
|
|
for i in range(max_questions):
|
|
if i > 0 or got is None:
|
|
for typ in ("mousePressed", "mouseReleased"):
|
|
await cdp("Input.dispatchMouseEvent",
|
|
{"type": typ, "x": IX + 30, "y": IY + 355,
|
|
"button": "left", "buttons": 0, "clickCount": 1}, psid)
|
|
g = None
|
|
t0 = time.time()
|
|
while time.time() - t0 < 8 and not g:
|
|
await asyncio.sleep(0.8)
|
|
gg = await jse("JSON.stringify(window.__lastGet)")
|
|
if gg and gg != "null":
|
|
try:
|
|
g = json.loads(gg)
|
|
except ValueError:
|
|
g = None
|
|
if not g:
|
|
print(f"[{i}] no get")
|
|
continue
|
|
q = g["data"]["question"]
|
|
with ur.urlopen(q["url1"]) as r:
|
|
big = r.read()
|
|
with ur.urlopen(q["url2"]) as r:
|
|
mark = r.read()
|
|
res = solve_slide(base64.b64encode(big).decode(), base64.b64encode(mark).decode(),
|
|
tip_y=q["tip_y"])
|
|
if res.get("ok"):
|
|
got = (big, mark, res, q.get("tip_y"))
|
|
break
|
|
print(f"[{i}] 识别失败: {res.get('message')}")
|
|
await jse("window.__lastGet = null;")
|
|
|
|
# 采完关掉验证码（点右上角 ×，避免下一轮复用旧题）——不用 verify，直接清 cookie 导航
|
|
try:
|
|
await cdp("Network.enable", sid=psid)
|
|
await cdp("Network.clearBrowserCookies", sid=psid)
|
|
await cdp("Page.enable", sid=psid)
|
|
await cdp("Page.navigate", {"url": "https://www.douyin.com/jingxuan"}, sid=psid)
|
|
for _ in range(60):
|
|
try:
|
|
ev = json.loads(await asyncio.wait_for(ws.recv(), timeout=2))
|
|
except (asyncio.TimeoutError, json.JSONDecodeError, ValueError):
|
|
continue
|
|
if ev.get("method") == "Page.loadEventFired" and ev.get("sessionId") == psid:
|
|
break
|
|
await asyncio.sleep(3)
|
|
except RuntimeError:
|
|
pass
|
|
return got
|
|
|
|
|
|
def annotate(big_bytes, mark_bytes, res, tip_y):
|
|
"""标注图：绿框目标位 + 黄轮廓 + 蓝箭头拖动轨迹 + 文字。"""
|
|
arr = np.frombuffer(big_bytes, np.uint8)
|
|
big = cv2.imdecode(arr, cv2.IMREAD_COLOR)
|
|
marr = np.frombuffer(mark_bytes, np.uint8)
|
|
mark = cv2.imdecode(marr, cv2.IMREAD_UNCHANGED)
|
|
if big is None or mark is None or mark.ndim < 3 or mark.shape[2] < 4:
|
|
return None
|
|
try:
|
|
x, y = int(res["x"]), int(res["y"])
|
|
s = float(res.get("scale", 1.0))
|
|
except (KeyError, TypeError, ValueError):
|
|
return None
|
|
vis = big.copy()
|
|
h, w = mark.shape[:2]
|
|
# 目标框按识别的尺度画（mark 在大图中的实际覆盖范围）
|
|
th, tw = round(h * s), round(w * s)
|
|
cv2.rectangle(vis, (x, y), (x + tw, y + th), (0, 255, 0), 2)
|
|
# mark alpha 轮廓（黄色）贴到目标位
|
|
alpha = (mark[..., 3] > 128).astype(np.uint8) * 255
|
|
am = alpha
|
|
ys, xs = np.where(am > 0)
|
|
if len(ys):
|
|
try:
|
|
am = am[ys.min():ys.max() + 1, xs.min():xs.max() + 1]
|
|
am_s = cv2.resize(am, (tw, th), interpolation=cv2.INTER_NEAREST)
|
|
edge = cv2.morphologyEx(am_s, cv2.MORPH_GRADIENT, np.ones((2, 2), np.uint8)) > 0
|
|
roi = vis[y:y + th, x:x + tw]
|
|
roi[edge] = (0, 255, 255)
|
|
except (ValueError, cv2.error):
|
|
pass
|
|
# 拖动轨迹箭头：mark 初始画布 x≈0，中心 y 对准目标中心
|
|
cy = y + th // 2
|
|
cv2.arrowedLine(vis, (8, cy), (x - 6, cy), (255, 128, 0), 2, tipLength=0.04)
|
|
cv2.putText(vis, f"drag {res['distance_px']}px", (8, max(14, cy - 10)),
|
|
cv2.FONT_HERSHEY_SIMPLEX, 0.45, (255, 128, 0), 1)
|
|
# 顶部信息条
|
|
bar = np.full((26, vis.shape[1], 3), 30, np.uint8)
|
|
txt = (f"x={res['x']} y={res['y']} scale={s:.3f} rot={res.get('rot', 0)} "
|
|
f"quality={res.get('quality', '-')} {res.get('confidence', '')}"
|
|
+ (f" tip_y={tip_y}" if tip_y is not None else ""))
|
|
cv2.putText(bar, txt, (6, 18), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 255), 1)
|
|
vis = np.vstack([bar, vis])
|
|
return vis
|
|
|
|
|
|
async def main():
|
|
try:
|
|
rounds = int(sys.argv[1]) if len(sys.argv) > 1 else 5
|
|
except ValueError:
|
|
rounds = 5
|
|
n_q = 3
|
|
done = 0
|
|
for i in range(rounds):
|
|
print(f"\n=== 轮 {i + 1}/{rounds} @ {time.strftime('%H:%M:%S')} ===", flush=True)
|
|
try:
|
|
got = await demo_round(n_q)
|
|
except (RuntimeError, OSError) as e:
|
|
print(f"轮 {i + 1} 异常: {e}")
|
|
got = None
|
|
if not got:
|
|
print(f"轮 {i + 1} 未采到题")
|
|
continue
|
|
big, mark, res, tip_y = got
|
|
ts = time.strftime("%Y%m%d_%H%M%S")
|
|
vis = annotate(big, mark, res, tip_y)
|
|
if vis is None:
|
|
print(f"轮 {i + 1} 标注失败（图像解码/字段缺失）")
|
|
continue
|
|
cv2.imwrite(str(OUT / f"demo_{ts}_{i}.png"), vis)
|
|
(OUT / f"demo_{ts}_{i}-mark.png").write_bytes(mark)
|
|
print(f"✓ x={res['x']} y={res['y']} s={res.get('scale')} "
|
|
f"quality={res.get('quality')} {res.get('confidence')} → demo_{ts}_{i}.png")
|
|
done += 1
|
|
await asyncio.sleep(random.uniform(2, 4))
|
|
# 汇总图
|
|
sheets = []
|
|
for p in sorted(OUT.glob("demo_*_*.png")):
|
|
if p.name.endswith("-mark.png"):
|
|
continue
|
|
img = cv2.imread(str(p))
|
|
if img is None:
|
|
continue
|
|
try:
|
|
h, w = img.shape[:2]
|
|
k = 420 / h
|
|
small = cv2.resize(img, (max(1, int(w * k)), 420))
|
|
sheets.append(small)
|
|
except cv2.error:
|
|
continue
|
|
if sheets:
|
|
# 横向拼，超 4 张换行
|
|
rows = [np.hstack(sheets[i:i + 4]) for i in range(0, len(sheets), 4)]
|
|
wmax = max(r.shape[1] for r in rows)
|
|
rows = [np.hstack([r, np.full((r.shape[0], wmax - r.shape[1], 3), 20, np.uint8)])
|
|
for r in rows]
|
|
cv2.imwrite(str(OUT / "index.png"), np.vstack(rows))
|
|
print(f"\n共 {done} 张标注图 → {OUT}/（含 index.png 汇总）")
|
|
else:
|
|
print(f"\n共 {done} 张，无汇总")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|