feat(try): 演示采集脚本（不拖动，标注图输出 results/）
- demo_collect.py: 线上多轮触发验证码 → 本地求解 → 标注图 （绿框目标位 + 黄色 mark 轮廓 + 蓝色拖动箭头 + 信息条） - 每题保存原图标注 PNG 与 mark PNG，index.png 汇总长图 - 用途：给客户演示识别效果；验证码刷新换题，不消耗 verify 配额 - results/ 为演示产物目录，不纳入版本库
This commit is contained in:
杨豪
+2
-27
@@ -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/
|
||||
|
||||
@@ -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())
|
||||
Reference in New Issue
Block a user