feat(web): 新增 Web 观测台(形态 A)+ 抽出 render/targets 共用层

- tcaptcha/web/:FastAPI + SSE + 零构建单页。把同步阻塞的 solve() 包成可观测
  后台任务(工作线程 + 全局串行,因 onnxruntime session 非线程安全),用
  logging.Handler 零侵入把 client 日志变成事件流;按靶子实测 min_interval 强制节流。
- tcaptcha/render.py:标注图(点击顺序编号 + 质心偏差)与点位量尺,CLI 与 web 共用。
- tcaptcha/targets.py:靶子表提为单一来源(含实测频控间隔 1ms=60s / mliev=3s)。
- fastapi/uvicorn 走 optional extra `web`,核心包零 web 依赖(带回归断言)。
- 修两个真 bug:hook 求解器时 args[0] 是实例自身(真入参从 args[1] 起);
  client 默认每次新建 solver,预热实例未被复用(每发重载 56MB ONNX)。
- 复验:icon/text/slider 三题型各拿到 errorCode=0;selftest 全绿;依赖隔离通过。

用法:uv run --extra web python -m tcaptcha.web.app
This commit is contained in:
yml2213
2026-10-14 19:10:54 +08:00
parent 74a863c8b3
commit 99565f65f0
15 changed files with 1827 additions and 1189 deletions
+29 -64
View File
@@ -28,13 +28,11 @@ from __future__ import annotations
import argparse
import base64
import json
import os
import sys
import time
from pathlib import Path
from typing import Dict, List, Optional, Sequence, Tuple
from typing import Dict, List, Optional
import cv2
import numpy as np
ROOT = Path(__file__).resolve().parents[2]
@@ -45,14 +43,9 @@ import tcaptcha.protocol.client as client_mod # noqa: E
import tcaptcha.solvers.icon_click as icon_mod # noqa: E402
from tcaptcha.protocol.client import (KIND_ICON_CLICK, KIND_SLIDER, # noqa: E402
KIND_TEXT_CLICK, TencentCaptchaClient)
from tcaptcha.render import annotate_points, icon_centroid_offset # noqa: E402
from tcaptcha.targets import TARGETS, get_target # noqa: E402
# 内置靶子:aid / entry_url / referer / wb
TARGETS: Dict[str, Tuple[str, str, str, str]] = {
"1ms": ("195672344", "https://1ms.run/console/personal/benefits/check-in",
"https://1ms.run/", "1"),
"mliev": ("190758559", "https://mliev.login.wang/auth/#/oidc/authorize",
"https://mliev.login.wang/", "1"),
}
KIND_MAP = {
"text": (KIND_TEXT_CLICK,),
"icon": (KIND_ICON_CLICK,),
@@ -60,53 +53,8 @@ KIND_MAP = {
"any": (KIND_TEXT_CLICK, KIND_ICON_CLICK, KIND_SLIDER),
}
# --------------------------------------------------------------------------- #
# 「点得准不准」的独立量尺:点击点 vs 该处图标的**暗色笔画质心**
# --------------------------------------------------------------------------- #
def icon_centroid_offset(bg_pil, x: int, y: int, half: int = 32) -> Optional[Tuple[float, float, float]]:
"""返回 (质心x, 质心y, 离点击点的距离px);估不出来返回 None。
原理:图标是近黑的线条画,背景是彩色的(天空/水面/木纹…)。在点击点周围截
(2*half)² 的窗口,用 Otsu 分出暗色笔画,取笔画质心当作"图标视觉中心"。
**这是独立于我求解器的量尺**(求解器用的是 YOLO box 中心),所以能回答
"我们到底点偏了多少"。
"""
w, h = bg_pil.size
box = (max(0, x - half), max(0, y - half), min(w, x + half), min(h, y + half))
patch = np.array(bg_pil.crop(box).convert("L"))
if patch.size == 0:
return None
thr, binimg = cv2.threshold(patch.astype(np.uint8), 0, 255,
cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU)
ys, xs = np.nonzero(binimg)
if len(xs) < 20:
return None
cx, cy = float(xs.mean()), float(ys.mean())
lx, ly = x - box[0], y - box[1] # 点击点在窗口内的坐标
return (cx + box[0], cy + box[1], float(np.hypot(cx - lx, cy - ly)))
def annotate(bg_bytes: bytes, points: Sequence[Sequence[int]], prompt_bytes: Optional[bytes],
offsets: Sequence[Optional[float]]) -> bytes:
"""底图上画**按点击顺序编号**的红圈;右上角贴提示图;标注每点偏离质心的 px。"""
bgr = cv2.imdecode(np.frombuffer(bg_bytes, np.uint8), cv2.IMREAD_COLOR)
for i, (x, y) in enumerate(points, 1):
cv2.circle(bgr, (int(x), int(y)), 18, (0, 0, 255), 2)
cv2.circle(bgr, (int(x), int(y)), 2, (0, 0, 255), -1)
cv2.putText(bgr, str(i), (int(x) - 8, int(y) - 22),
cv2.FONT_HERSHEY_SIMPLEX, 0.85, (0, 255, 0), 2)
if prompt_bytes:
pr = cv2.imdecode(np.frombuffer(prompt_bytes, np.uint8), cv2.IMREAD_COLOR)
if pr is not None:
ph = bgr.shape[0] // 4
pw = max(1, int(pr.shape[1] * ph / pr.shape[0]))
pr = cv2.resize(pr, (pw, ph))
y0, x0 = 4, bgr.shape[1] - pw - 4
if x0 >= 0:
bgr[y0:y0 + ph, x0:x0 + pw] = pr
cv2.rectangle(bgr, (x0, y0), (x0 + pw, y0 + ph), (255, 255, 255), 1)
return cv2.imencode(".jpg", bgr)[1].tobytes()
# 点位的"量尺"(点击点 vs 图标暗色笔画质心)与标注图渲染统一在 tcaptcha.render,
# 与 Web 界面共用同一份实现 —— 别在这里再抄一遍。
def main() -> int:
@@ -118,11 +66,14 @@ def main() -> int:
help="给答案整体平移 DX,DY 像素(容差标定用)")
ap.add_argument("--swap", action="store_true", help="故意交换前两个点(负对照,必错)")
ap.add_argument("--max-refreshes", type=int, default=40)
ap.add_argument("--interval", type=float, default=1.0,
help="两次求解之间的间隔秒数(1ms 连打会触发频控,建议 3~5)")
ap.add_argument("--outdir", default=None)
ap.add_argument("--max-retries", type=int, default=1)
args = ap.parse_args()
aid, entry, referer, wb = TARGETS[args.target]
tgt = get_target(args.target)
aid, entry, referer, wb = tgt["aid"], tgt["entry_url"], tgt["referer"], tgt["wb"]
dx, dy = (int(v) for v in args.offset.split(","))
tag = "%s_%s_n%d_off%s%s" % (args.target, args.kind, args.n, args.offset,
"_swap" if args.swap else "")
@@ -245,7 +196,8 @@ def main() -> int:
offsets.append(None if est is None else round(est[2], 1))
centroids.append(None if est is None else [round(est[0], 1), round(est[1], 1)])
(d / "bg.jpg").write_bytes(bg_raw)
(d / "annotated.jpg").write_bytes(annotate(bg_raw, pts, prompt_raw, offsets))
(d / "annotated.jpg").write_bytes(
annotate_points(bg_raw, pts, prompt=prompt_raw, offsets=offsets))
if prompt_raw is not None:
(d / "prompt.jpg").write_bytes(prompt_raw)
(d / "points.json").write_text(json.dumps({
@@ -257,26 +209,36 @@ def main() -> int:
"prehandles": counters["prehandle"],
}, ensure_ascii=False, indent=1), encoding="utf-8")
row = {"run": n, "errorCode": ec, "points": [list(p) for p in pts],
row = {"run": n, "errorCode": ec, "kind": res.get("kind"),
"instruction": res.get("instruction"),
"points": [list(p) for p in pts],
"collect_len": last_gen.get("collect_len"),
"off_px": offsets, "secs": round(time.time() - t0, 1),
"prehandles": counters["prehandle"]}
rows.append(row)
print("run%-2d ec=%-4s 偏离质心px=%-22s prehandle=%-3d points=%s %.0fs" % (
n, ec, str(offsets), counters["prehandle"], pts, time.time() - t0))
time.sleep(1.0)
print("run%-2d ec=%-4s kind=%-10s 偏离质心px=%-20s prehandle=%-3d points=%s %.0fs" % (
n, ec, res.get("kind") or "?", str(offsets), counters["prehandle"], pts,
time.time() - t0))
time.sleep(max(0.0, args.interval))
finally:
c.close()
icon_mod.CaptchaSolver.solve = real_icon_solve
client_mod.generate_telemetry = real_gen
counts: Dict[str, int] = {}
by_kind: Dict[str, Dict[str, int]] = {}
for r in rows:
counts[str(r.get("errorCode"))] = counts.get(str(r.get("errorCode")), 0) + 1
ec_s = str(r.get("errorCode"))
counts[ec_s] = counts.get(ec_s, 0) + 1
k = str(r.get("kind") or "?")
by_kind.setdefault(k, {})
by_kind[k][ec_s] = by_kind[k].get(ec_s, 0) + 1
all_off = [v for r in rows for v in (r.get("off_px") or []) if isinstance(v, (int, float))]
summary = {
"target": args.target, "kind": args.kind, "n": args.n,
"offset_applied": [dx, dy], "swapped": args.swap,
"errorCode_counts": counts,
"by_kind": by_kind,
"centroid_offset_px": {
"count": len(all_off),
"min": min(all_off) if all_off else None,
@@ -291,6 +253,9 @@ def main() -> int:
print("靶子=%s 题型=%s 次数=%d 答案平移=%s 交换=%s" % (args.target, args.kind, args.n,
args.offset, args.swap))
print("errorCode 计数: %s" % json.dumps(counts, ensure_ascii=False))
if len(by_kind) > 1 or "?" not in by_kind:
for k in sorted(by_kind):
print(" 按题型 %-12s %s" % (k, json.dumps(by_kind[k], ensure_ascii=False)))
if all_off:
print("点击点偏离图标笔画质心: 均值 %.1fpx / 最小 %.1f / 最大 %.1f(%d 个点)"
% (np.mean(all_off), min(all_off), max(all_off), len(all_off)))