# -*- coding: utf-8 -*- """`SolveJob` / `JobManager` —— 把**同步阻塞**的求解包成"可观测的后台任务"。 为什么必须要有这一层（三条都是硬约束，不是设计偏好）: 1. `client.solve()` 是同步阻塞的，一次 7~60s（内含 5.5~7.5s 拟人等待， 换题还要再等 2.8~4.2s）——**不能挂在 HTTP 请求上**，否则前端必超时。 2. ONNX / ddddocr 是阻塞的 CPU 计算，且 `onnxruntime` 的 session **不是线程安全的** -> 必须丢到工作线程，并且**全局串行**（同一时刻只跑一个求解）。 3. 中间态（题型 / 题目图 / 点位 / 段几何 / errorCode）散落在局部变量和日志里 -> 用「**日志 handler 桥接** + 求解器 hook」把它们事件化，前端才能实时看到。 `--` 这里刻意**不修改** `protocol/client.py`：日志桥接零侵入， 求解器的 hook 只在本 job 存活期间（成对安装/卸载），且被串行锁保护。 """ from __future__ import annotations import base64 import importlib import json import logging import threading import time import uuid from typing import Any, Callable, Dict, List, Optional, Tuple from tcaptcha.protocol.client import (KIND_ICON_CLICK, KIND_SLIDER, KIND_TEXT_CLICK, TencentCaptchaClient) from tcaptcha.render import annotate_points, icon_centroid_offset, to_jpeg from tcaptcha.targets import get_target from .events import (EVT_CHALLENGE, EVT_ERROR, EVT_FINISHED, EVT_LOG, EVT_SOLVE, EVT_STARTED, EVT_TELEMETRY, EVT_VERIFY, EventLog) #: 题型 -> accept_kinds KIND_MAP: Dict[str, Tuple[str, ...]] = { "text": (KIND_TEXT_CLICK,), "icon": (KIND_ICON_CLICK,), "slider": (KIND_SLIDER,), "any": (KIND_TEXT_CLICK, KIND_ICON_CLICK, KIND_SLIDER), } #: 要 hook 的求解器： (标签, 模块, 类, 方法) _SOLVER_TARGETS = ( ("icon", "tcaptcha.solvers.icon_click", "CaptchaSolver", "solve"), ("text", "tcaptcha.solvers.text_click", "TextClickSolver", "solve_ex"), ("slider", "tcaptcha.solvers.slider", "SliderSolver", "solve"), ) #: 一次任务最多在内存里留多少张图（超出就丢最早的，保证内存有界） MAX_IMAGES = 24 #: 进程内复用的求解器实例 —— ONNX(YOLO+ResNet18, ~56MB) 与 ddddocr 加载一次要几秒， #: 而 `TencentCaptchaClient` 默认**每次新建**一个 solver；把实例传进去才能真正复用。 #: ⚠️ 复用安全的前提是**求解被全局串行**（见 `JobManager._serial`）—— #: onnxruntime 的 session 不是线程安全的，别在没锁的情况下并发调用。 _SOLVER_CACHE: Dict[str, Any] = {} def cached_solvers() -> Dict[str, Any]: """按需加载并缓存求解器（进程内复用）。供 `app.py` 预热与 `SolveJob` 共用。""" if "icon" not in _SOLVER_CACHE: try: from tcaptcha.solvers.icon_click import CaptchaSolver _SOLVER_CACHE["icon"] = CaptchaSolver() except Exception as exc: # noqa: BLE001 加载失败就退回 client 自身懒加载 logging.getLogger("web.runner").warning("图标求解器加载失败: %s", exc) _SOLVER_CACHE["icon"] = None if "text" not in _SOLVER_CACHE: try: from tcaptcha.solvers.text_click import TextClickSolver _SOLVER_CACHE["text"] = TextClickSolver() except Exception as exc: # noqa: BLE001 logging.getLogger("web.runner").warning("文字点选求解器加载失败: %s", exc) _SOLVER_CACHE["text"] = None return _SOLVER_CACHE class _LogBridge(logging.Handler): """把 tcaptcha 的日志**原样**变成事件 —— 零侵入拿到全过程。 client 里已经有 20+ 条 `logger.info`（抽题 / PoW 命中 / 题型、图片落盘、 遥测密文来源与长度、求解结论、verify 结果…），接上它就等于免费得到一份 "和 CLI 一模一样"的执行轨迹，不必去 client 里逐个加回调。 """ def __init__(self, emit: Callable[[str, str, str], None]) -> None: super().__init__(level=logging.INFO) self._emit = emit def emit(self, record: logging.LogRecord) -> None: try: self._emit(record.levelname, record.name, record.getMessage()) except Exception: # noqa: BLE001 日志桥绝不能反过来炸主流程 pass class _Hooks: """成对安装/卸载的观测 hook（必须被串行锁保护）。 两类： * **求解器 hook** —— 抓题型 / 题目图 / 点位，用来画标注图与量点位质量； * **遥测 hook** —— 抓 `generate_telemetry` 的产物，解出 4 段几何与轨迹点数 （这是判断"包是否正确生成"的关键指标，CLI 里一直靠它做诊断）。 """ def __init__(self, job: "SolveJob") -> None: self._job = job self._saved: List[Tuple[Any, str, Any]] = [] self._gen_saved: Optional[Tuple[Any, Any]] = None def __enter__(self) -> "_Hooks": for key, modname, clsname, fname in _SOLVER_TARGETS: try: cls = getattr(importlib.import_module(modname), clsname) original = getattr(cls, fname) except Exception: # noqa: BLE001 某个求解器不在也不该影响别的 continue setattr(cls, fname, self._job._wrap_solver(key, original)) self._saved.append((cls, fname, original)) try: import tcaptcha.protocol.client as client_mod self._gen_saved = (client_mod, client_mod.generate_telemetry) client_mod.generate_telemetry = self._job._wrap_telemetry(client_mod.generate_telemetry) except Exception: # noqa: BLE001 self._gen_saved = None return self def __exit__(self, *exc_info) -> bool: for cls, fname, original in self._saved: setattr(cls, fname, original) self._saved.clear() if self._gen_saved is not None: setattr(self._gen_saved[0], "generate_telemetry", self._gen_saved[1]) self._gen_saved = None return False class SolveJob: """一次真实求解任务。""" def __init__(self, job_id: str, target_key: str, params: Dict[str, Any], throttle: Optional[Callable[["SolveJob"], float]] = None, on_finish: Optional[Callable[["SolveJob"], None]] = None) -> None: self.id = job_id self.target_key = target_key self.params = dict(params) self.events = EventLog() self.state = "pending" # pending|throttled|running|done|error|cancelled self.result: Optional[Dict[str, Any]] = None self.error = "" self.created_at = time.time() self.finished_at: Optional[float] = None self._throttle = throttle self._on_finish = on_finish self._cancel = threading.Event() self._challenge_no = 0 self._solves: List[Dict[str, Any]] = [] self._rounds: List[Dict[str, Any]] = [] # 每一发的结果（count>1 时） self._count = 1 self._images: Dict[str, bytes] = {} self._img_lock = threading.Lock() # ------------------------------------------------------------------ 图片 def add_image(self, name: str, data: bytes) -> str: """把一张图存进本 job 的内存（返回给前端的 URL）。""" with self._img_lock: while len(self._images) >= MAX_IMAGES: self._images.pop(next(iter(self._images))) self._images[name] = data return "/api/jobs/%s/image/%s" % (self.id, name) def get_image(self, name: str) -> Optional[bytes]: with self._img_lock: return self._images.get(name) # ------------------------------------------------------------- 对外状态 @property def cancelled(self) -> bool: return self._cancel.is_set() def cancel(self) -> None: """标记取消。注意：`solve()` 内部是同步阻塞的， 所以取消只在**节流等待中**或**当前这一发结束后**生效。""" self._cancel.set() self.events.append(EVT_LOG, level="WARNING", logger="web.runner", message="已请求取消（当前这一发结束后生效）") def stats(self) -> Dict[str, Any]: offs = [v for s in self._solves for v in (s.get("off_px") or []) if isinstance(v, (int, float))] ews = [r["errorCode"] for r in self._rounds if r.get("errorCode") is not None] ok_n = sum(1 for r in self._rounds if r.get("success")) return { "challenges": self._challenge_no, "solves": len(self._solves), "points": sum(len(s.get("points") or []) for s in self._solves), "mean_offset_px": round(sum(offs) / len(offs), 1) if offs else None, "count": self._count, "rounds": list(self._rounds), "ok": ok_n, "errorCodes": ews, } def snapshot(self) -> Dict[str, Any]: return { "id": self.id, "target": self.target_key, "params": self.params, "state": self.state, "last_seq": self.events.last_seq, "created_at": self.created_at, "finished_at": self.finished_at, "result": self.result, "stats": self.stats(), "error": self.error, } # ------------------------------------------------------------- 执行流程 def run(self) -> None: """在工作线程里跑（由 JobManager 串行调度）。""" try: self._execute() except BaseException as exc: # noqa: BLE001 兜住一切，别让线程静默死掉 self.state = "error" self.error = "%s: %s" % (type(exc).__name__, exc) self.events.append(EVT_ERROR, message=self.error) logging.getLogger("web.runner").exception("任务 %s 异常", self.id) finally: self.finished_at = time.time() if self._on_finish: self._on_finish(self) def _execute(self) -> None: tgt = get_target(self.target_key) kind = str(self.params.get("kind") or "icon") self._count = max(1, min(20, int(self.params.get("count") or 1))) self.events.append(EVT_STARTED, target=self.target_key, aid=tgt["aid"], label=tgt.get("label", self.target_key), kind=kind, mode="python_values", count=self._count, note=tgt["note"], params={k: v for k, v in self.params.items()}) self.state = "running" bridge = _LogBridge(self._emit_log) loggers = [logging.getLogger("tcaptcha"), logging.getLogger("tcaptcha_client")] for lg in loggers: if lg.level == logging.NOTSET or lg.level > logging.INFO: lg.setLevel(logging.INFO) lg.addHandler(bridge) try: for i in range(1, self._count + 1): if self.cancelled: break if i > 1: self.events.append(EVT_LOG, level="INFO", logger="web.runner", message="—— 第 %d/%d 发 ——" % (i, self._count)) if self._throttle: waited = self._throttle(self) if waited: self.events.append(EVT_LOG, level="INFO", logger="web.runner", message="频控节流结束（等待 %.1fs）" % waited) if self.cancelled: break with _Hooks(self): res = self._call_client(tgt, kind) self._finalize(res, round_no=i) finally: for lg in loggers: lg.removeHandler(bridge) if self.cancelled: self.state = "cancelled" self.events.append(EVT_FINISHED, success=any(r.get("success") for r in self._rounds), errorCode=self._rounds[-1]["errorCode"] if self._rounds else None, ticket=self._rounds[-1].get("ticket") or "", elapsed_ms=self._rounds[-1].get("elapsed_ms"), error=self.error or ("已取消（完成 %d/%d 发）" % (len(self._rounds), self._count) if self.cancelled else ""), stats=self.stats()) def _emit_log(self, level: str, name: str, message: str) -> None: self.events.append(EVT_LOG, level=level, logger=name, message=message) def _call_client(self, tgt, kind: str) -> Dict[str, Any]: solvers = cached_solvers() client = TencentCaptchaClient( aid=tgt["aid"], entry_url=tgt["entry_url"], referer=tgt["referer"], wb=tgt["wb"], accept_kinds=KIND_MAP.get(kind, KIND_MAP["any"]), telemetry_mode="python_values", proxy=None, solver=solvers.get("icon"), # 复用实例：不传的话 client 每发都重建 text_solver=solvers.get("text"), ) try: return client.solve(max_retries=int(self.params.get("max_retries", 2)), max_refreshes=int(self.params.get("max_refreshes", 20))) finally: try: client.close() except Exception: # noqa: BLE001 pass def _finalize(self, res: Dict[str, Any], round_no: int = 1) -> None: raw = res.get("raw_response") or {} ec = raw.get("errorCode") self.result = { "success": bool(res.get("success")), "errorCode": ec, "ticket": res.get("ticket") or "", "randstr": res.get("randstr") or "", "elapsed_ms": res.get("elapsed_ms"), "kind": res.get("kind") or "", "instruction": res.get("instruction") or "", "error": res.get("error") or "", } self.error = res.get("error") or "" self.state = "done" if res.get("success") else ("error" if self.error else "done") self._rounds.append({"round": round_no, "success": bool(res.get("success")), "errorCode": ec, "kind": res.get("kind") or "", "ticket": res.get("ticket") or "", "elapsed_ms": res.get("elapsed_ms"), "error": res.get("error") or ""}) self.events.append(EVT_VERIFY, round=round_no, success=bool(res.get("success")), errorCode=ec, ticket=res.get("ticket") or "", randstr=res.get("randstr") or "", elapsed_ms=res.get("elapsed_ms")) self.events.append(EVT_LOG, level="INFO", logger="web.runner", message="第 %d/%d 发结束：errorCode=%s %s" % (round_no, self._count, ec, "✅" if res.get("success") else (self.error or "未通过"))) # --------------------------------------------------------- 求解器 hook def _wrap_solver(self, key: str, original: Callable) -> Callable: def wrapper(*args, **kwargs): result = original(*args, **kwargs) try: self._capture(key, args, result) except Exception as exc: # noqa: BLE001 捕获失败不能影响求解 self.events.append(EVT_LOG, level="WARNING", logger="web.runner", message="捕获求解中间态失败: %s" % exc) return result return wrapper def _capture(self, key: str, args: tuple, result: Any) -> None: """把「题目图 + 点位」事件化。 ⚠️ `args[0]` 是**求解器实例自身**（hook 的是实例方法，`client` 里是 `self.icon_solver.solve(bg, prompt)` 这种调用），真正的入参从 `args[1]` 开始。 """ if key == "icon": bg = args[1] prompt = args[2] if len(args) > 2 else None points = [(int(p[0]), int(p[1])) for p in (result or [])] self._emit_challenge(KIND_ICON_CLICK, "请依次点击：", bg, prompt) if points: self._emit_solve(KIND_ICON_CLICK, bg, points, prompt, "YOLOv8 召回 + ResNet18 分类 + 全局指派") elif key == "text": bg = args[1] instr = args[2] if len(args) > 2 else "" points = [(int(p[0]), int(p[1])) for p in (getattr(result, "points", None) or [])] self._emit_challenge(KIND_TEXT_CLICK, instr, bg, None) if points: self._emit_solve(KIND_TEXT_CLICK, bg, points, None, _describe(result)) else: # slider bg = args[1] piece = args[2] if len(args) > 2 else None tx = getattr(result, "target_x_image", None) ty = getattr(result, "y_image", None) points = [(int(round(tx)), int(round(ty)))] if tx is not None and ty is not None else [] self._emit_challenge(KIND_SLIDER, "拖动下方滑块完成拼图", bg, piece) if points: self._emit_solve(KIND_SLIDER, bg, points, piece, _describe(result)) def _emit_challenge(self, kind: str, instruction: str, bg: Any, prompt: Any) -> None: self._challenge_no += 1 n = self._challenge_no data: Dict[str, Any] = {"index": n, "kind": kind, "instruction": instruction} data["bg_url"] = self._safe_image("c%d_bg.jpg" % n, bg) if prompt is not None: data["prompt_url"] = self._safe_image("c%d_prompt.jpg" % n, prompt) self.events.append(EVT_CHALLENGE, **data) def _emit_solve(self, kind: str, bg: Any, points, prompt: Any, note: str) -> None: """`off_px` **只对点选题**（图标/文字）有意义 —— 它量的是"点击点离该处 暗色笔画质心多远"。滑块的答案是**缺口位置**（那里通常没有笔画）， 硬套这个量尺会给出十几到几十 px 的噪声，所以滑块一律置 None。 """ n = self._challenge_no measurable = kind in (KIND_ICON_CLICK, KIND_TEXT_CLICK) offsets: List[Optional[float]] = [] centroids: List[Optional[List[float]]] = [] for (x, y) in points: est = icon_centroid_offset(bg, x, y) if measurable else None offsets.append(round(est[2], 1) if est else None) centroids.append([round(est[0], 1), round(est[1], 1)] if est else None) data: Dict[str, Any] = {"index": n, "kind": kind, "points": [list(p) for p in points], "off_px": offsets, "centroids": centroids, "note": note} try: # title 不画（卡片标题里已有），提示图不贴（界面上是独立缩略图， # 贴进去会遮挡右上角的目标图标）—— 详见 render.annotate_points。 # 标注图只给点选题画：红圈+序号表达的是"点击顺序"，滑块答案是 # 缺口位置，画上去只会误导（界面上底图+滑块图已足够核对）。 if measurable: data["annotated_url"] = self.add_image( "c%d_annotated.jpg" % n, annotate_points(bg, points, offsets=offsets, centroids=centroids, embed_prompt=False)) except Exception: # noqa: BLE001 pass self._solves.append(data) self.events.append(EVT_SOLVE, **data) def _safe_image(self, name: str, image: Any) -> str: try: return self.add_image(name, to_jpeg(image)) except Exception: # noqa: BLE001 return "" # ------------------------------------------------------------- 遥测 hook def _wrap_telemetry(self, original: Callable) -> Callable: def wrapper(tdc_code, points, **kwargs): out = original(tdc_code, points, **kwargs) try: self._capture_telemetry(out) except Exception as exc: # noqa: BLE001 解析失败不能影响发出去的包 self.events.append(EVT_LOG, level="WARNING", logger="web.runner", message="解析遥测段几何失败: %s" % exc) return out return wrapper def _capture_telemetry(self, out: Dict[str, Any]) -> None: """解出「4 段几何 + 轨迹点数」—— 判断包是否**按本次会话构建**正确生成的关键指标。""" info: Dict[str, Any] = {"source": out.get("source"), "collect_len": len(out.get("collect") or "")} try: from tcaptcha.telemetry.crypto import split_collect_segments, xtea_decrypt_bytes plain = xtea_decrypt_bytes(base64.b64decode(out["collect"]), out["key"]).decode("latin1").rstrip("\x00") segs = split_collect_segments(plain) info["segments"] = [len(s) for s in segs] info["traj_points"] = len(json.loads(segs[1])) except Exception as exc: # noqa: BLE001 info["segments_err"] = str(exc)[:160] self.events.append(EVT_TELEMETRY, **info) def _describe(result: Any) -> str: """取求解器自带的 describe()（各 solver 都有，但签名不同，统一兜住）。""" try: return str(result.describe()) except Exception: # noqa: BLE001 return "" class JobManager: """任务注册表 + **全局串行**调度 + 按靶子节流。""" def __init__(self, max_jobs: int = 50) -> None: self._jobs: Dict[str, SolveJob] = {} self._order: List[str] = [] self._lock = threading.Lock() self._serial = threading.Lock() # 同一时刻只跑一个求解（ONNX 非线程安全） self._last_end: Dict[str, float] = {} self.max_jobs = max_jobs def submit(self, target_key: str, params: Dict[str, Any]) -> SolveJob: get_target(target_key) # 未知靶子直接抛错 job_id = uuid.uuid4().hex[:12] job = SolveJob(job_id, target_key, params, throttle=self._throttle, on_finish=self._on_finish) with self._lock: self._jobs[job_id] = job self._order.append(job_id) while len(self._order) > self.max_jobs: self._jobs.pop(self._order.pop(0), None) threading.Thread(target=self._worker, args=(job,), name="solve-%s" % job_id, daemon=True).start() return job def get(self, job_id: str) -> Optional[SolveJob]: with self._lock: return self._jobs.get(job_id) def list_jobs(self, limit: int = 30) -> List[Dict[str, Any]]: with self._lock: ids = list(self._order)[-limit:] jobs = [self._jobs[i] for i in ids if i in self._jobs] return [j.snapshot() for j in reversed(jobs)] # ------------------------------------------------------------ 内部调度 def _worker(self, job: SolveJob) -> None: if not self._serial.acquire(timeout=0.1): job.events.append(EVT_LOG, level="INFO", logger="web.runner", message="排队中：同一时刻只跑一个求解（ONNX 非线程安全），" "前面的任务结束后自动开始") self._serial.acquire() try: job.run() finally: self._serial.release() def _throttle(self, job: SolveJob) -> float: """按靶子**实测**的最小间隔节流（1ms=60s，mliev=3s）。返回实际等待秒数。""" tgt = get_target(job.target_key) with self._lock: last = self._last_end.get(job.target_key) if last is None: return 0.0 need = float(tgt["min_interval"]) - (time.time() - last) if need <= 0: return 0.0 job.state = "throttled" job.events.append(EVT_LOG, level="INFO", logger="web.runner", message="频控节流：%s 实测最小间隔 %.0fs，还需等待 %.1fs" "（连打会被判 errorCode=12，不是求解失败）" % (job.target_key, tgt["min_interval"], need)) waited = 0.0 while waited < need: if job.cancelled: return waited step = min(1.0, need - waited) time.sleep(step) waited += step return waited def _on_finish(self, job: SolveJob) -> None: with self._lock: self._last_end[job.target_key] = time.time() def cooldown(self) -> Dict[str, float]: """每个靶子还需等多少秒才能再打（给界面显示倒计时）。""" now = time.time() out: Dict[str, float] = {} with self._lock: for key, last in self._last_end.items(): try: need = float(get_target(key)["min_interval"]) - (now - last) except KeyError: need = 0.0 out[key] = round(max(0.0, need), 1) return out __all__ = ["SolveJob", "JobManager", "KIND_MAP", "EVT_TELEMETRY"]