diff --git a/.gitignore b/.gitignore index 12b699c..c2786e6 100644 --- a/.gitignore +++ b/.gitignore @@ -1,29 +1,4 @@ -# Python __pycache__/ -*.py[cod] -*.egg-info/ +*.pyc .venv/ -venv/ - -# IDE -.idea/ -.vscode/ -*.swp - -# 样本图片（不提交原始验证码） -captchas/ -!captchas/.gitkeep - -# 实验产物 -try/out/ -try/visual/ -src/out/ -*.png -*.jpeg - -# 系统文件 -.DS_Store -Thumbs.db - -# CodeGraph 本地索引 -.codegraph/ +results/ diff --git a/try/demo_collect.py b/try/demo_collect.py new file mode 100644 index 0000000..07884d7 --- /dev/null +++ b/try/demo_collect.py @@ -0,0 +1,432 @@ +"""演示采集：线上多轮触发验证码 → 本地识别 → 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())