yml2213
5c4ff8c4a1
- targets.py 增加 label 字段（1ms 签到 / mliev 登录），下拉框、说明、历史表统一用友好名 - POST /api/jobs 支持 count(1~20)：一次任务连打 N 发，发间自动按靶子频控节流、可取消 - 每发结果记入 stats.rounds，EVT_VERIFY 带 round，结果区展示「每发结果（通过 x/y）」彩色一览 - 滑块不再生成标注图（红圈+序号只表达点击顺序，对滑块是误导） - 观测台 README 补齐启动/后台运行/端口占用/自检说明
546 lines
25 KiB
Python
546 lines
25 KiB
Python
# -*- 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"]
|