yml2213
5c4ff8c4a1
- targets.py 增加 label 字段（1ms 签到 / mliev 登录），下拉框、说明、历史表统一用友好名 - POST /api/jobs 支持 count(1~20)：一次任务连打 N 发，发间自动按靶子频控节流、可取消 - 每发结果记入 stats.rounds，EVT_VERIFY 带 round，结果区展示「每发结果（通过 x/y）」彩色一览 - 滑块不再生成标注图（红圈+序号只表达点击顺序，对滑块是误导） - 观测台 README 补齐启动/后台运行/端口占用/自检说明
237 lines
9.6 KiB
Python
237 lines
9.6 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""tcaptcha 的 Web 界面（形态 A：观测 / 调试台）。
|
||
|
||
定位：**把 CLI 的黑盒变透明** —— 启一次任务，实时看到
|
||
「抽到哪道题 → PoW 命中 → 求解出点位（带标注图）→ collect 段几何 → verify 的 errorCode/ticket」。
|
||
|
||
它**不是**新造一套求解逻辑：后端就是 `TencentCaptchaClient.solve()`，
|
||
与 `scripts/verify/send_test.py` / `send_campaign.py` 跑的是同一条链路。
|
||
|
||
启动:
|
||
uv run --extra web python -m tcaptcha.web.app # 默认 127.0.0.1:8770
|
||
uv run --extra web python -m tcaptcha.web.app --port 9000 --host 0.0.0.0
|
||
|
||
接口:
|
||
GET / 单页界面
|
||
GET /api/targets 靶子列表 + 频控冷却倒计时
|
||
POST /api/jobs 建任务 {target, kind, max_retries, max_refreshes}
|
||
GET /api/jobs 最近任务列表
|
||
GET /api/jobs/{id} 任务快照
|
||
GET /api/jobs/{id}/events **SSE** 事件流（支持 Last-Event-ID 续传）
|
||
GET /api/jobs/{id}/image/{n} 题目图 / 标注图（内存里，任务结束仍在）
|
||
POST /api/jobs/{id}/cancel 取消（当前这一发结束后生效）
|
||
GET /api/health 健康检查
|
||
|
||
⚠️ 频控：靶子自带实测最小间隔（1ms = 60s）。后端会**强制节流**并在事件流里说明，
|
||
所以界面上看到"等 60s"是设计行为，不是卡住。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any, Dict, Optional
|
||
|
||
from fastapi import FastAPI, HTTPException, Request
|
||
from fastapi.responses import HTMLResponse, Response, StreamingResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel, Field
|
||
|
||
from tcaptcha.web.events import TERMINAL_TYPES
|
||
from tcaptcha.web.runner import KIND_MAP, JobManager
|
||
|
||
from tcaptcha.targets import TARGETS, describe as describe_targets
|
||
|
||
logger = logging.getLogger("web.app")
|
||
|
||
STATIC_DIR = Path(__file__).resolve().parent / "static"
|
||
MANAGER = JobManager(max_jobs=50)
|
||
|
||
app = FastAPI(title="tcaptcha 观测台", version="0.1.0")
|
||
|
||
|
||
# --------------------------------------------------------------------- 模型
|
||
class JobRequest(BaseModel):
|
||
target: str = Field(default="mliev", description="靶子 key")
|
||
kind: str = Field(default="icon", description="text | icon | slider | any")
|
||
count: int = Field(default=1, ge=1, le=20, description="连打次数（发间自动按靶子频控节流）")
|
||
max_retries: int = Field(default=2, ge=1, le=10)
|
||
max_refreshes: int = Field(default=20, ge=1, le=60)
|
||
|
||
|
||
# --------------------------------------------------------------------- 页面
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def index() -> HTMLResponse:
|
||
html = (STATIC_DIR / "index.html").read_text(encoding="utf-8")
|
||
return HTMLResponse(html)
|
||
|
||
|
||
@app.get("/api/health")
|
||
def health() -> Dict[str, Any]:
|
||
return {"status": "online", "targets": list(TARGETS), "cooldown": MANAGER.cooldown()}
|
||
|
||
|
||
@app.get("/api/targets")
|
||
def targets() -> Dict[str, Any]:
|
||
"""靶子列表 + 当前冷却剩余秒数（前端据此倒计时、禁用按钮）。"""
|
||
return {"targets": describe_targets(), "cooldown": MANAGER.cooldown(),
|
||
"kinds": sorted(KIND_MAP)}
|
||
|
||
|
||
# --------------------------------------------------------------------- 任务
|
||
@app.post("/api/jobs")
|
||
def create_job(req: JobRequest) -> Dict[str, Any]:
|
||
if req.target not in TARGETS:
|
||
raise HTTPException(400, "未知靶子 %r（可选：%s）" % (req.target, ", ".join(TARGETS)))
|
||
if req.kind not in KIND_MAP:
|
||
raise HTTPException(400, "未知题型 %r（可选：%s）" % (req.kind, ", ".join(sorted(KIND_MAP))))
|
||
job = MANAGER.submit(req.target, {"kind": req.kind, "count": req.count,
|
||
"max_retries": req.max_retries,
|
||
"max_refreshes": req.max_refreshes})
|
||
return job.snapshot()
|
||
|
||
|
||
@app.get("/api/jobs")
|
||
def list_jobs(limit: int = 30) -> Dict[str, Any]:
|
||
return {"jobs": MANAGER.list_jobs(limit=limit)}
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}")
|
||
def get_job(job_id: str) -> Dict[str, Any]:
|
||
job = MANAGER.get(job_id)
|
||
if job is None:
|
||
raise HTTPException(404, "任务不存在或已被淘汰")
|
||
return job.snapshot()
|
||
|
||
|
||
@app.post("/api/jobs/{job_id}/cancel")
|
||
def cancel_job(job_id: str) -> Dict[str, Any]:
|
||
job = MANAGER.get(job_id)
|
||
if job is None:
|
||
raise HTTPException(404, "任务不存在或已被淘汰")
|
||
job.cancel()
|
||
return job.snapshot()
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/image/{name}")
|
||
def get_image(job_id: str, name: str) -> Response:
|
||
job = MANAGER.get(job_id)
|
||
if job is None:
|
||
raise HTTPException(404, "任务不存在或已被淘汰")
|
||
data = job.get_image(name)
|
||
if data is None:
|
||
raise HTTPException(404, "图片不存在或已被内存上限淘汰")
|
||
return Response(content=data, media_type="image/jpeg",
|
||
headers={"Cache-Control": "public, max-age=600"})
|
||
|
||
|
||
@app.get("/api/jobs/{job_id}/events")
|
||
async def job_events(job_id: str, request: Request, since: int = 0) -> StreamingResponse:
|
||
"""SSE：先补发历史（断线重连/中途打开页面都不丢），再持续推送直到终态。"""
|
||
job = MANAGER.get(job_id)
|
||
if job is None:
|
||
raise HTTPException(404, "任务不存在或已被淘汰")
|
||
# 浏览器 EventSource 重连时自动带 Last-Event-ID，用它续传
|
||
header_since = request.headers.get("last-event-id")
|
||
if header_since and header_since.isdigit():
|
||
since = max(since, int(header_since))
|
||
|
||
async def stream():
|
||
cursor = since
|
||
idle_ticks = 0
|
||
while True:
|
||
if await request.is_disconnected():
|
||
return
|
||
batch = job.events.since(cursor)
|
||
for evt in batch:
|
||
cursor = evt.seq
|
||
payload = json.dumps(evt.to_dict(), ensure_ascii=False)
|
||
# 刻意不发 `event:` 行 —— 让前端统一走 onmessage 按 data.type 分派，
|
||
# 少一处"类型名要和 addEventListener 对齐"的隐患。
|
||
yield "id: %d\ndata: %s\n\n" % (evt.seq, payload)
|
||
if evt.type in TERMINAL_TYPES:
|
||
return
|
||
if not batch:
|
||
if job.finished_at is not None:
|
||
return # 终态且已推完
|
||
idle_ticks += 1
|
||
if idle_ticks % 40 == 0: # 每 ~10s 一次心跳，防止中间层掐连接
|
||
yield ": keep-alive\n\n"
|
||
else:
|
||
idle_ticks = 0
|
||
await asyncio.sleep(0.25)
|
||
|
||
return StreamingResponse(stream(), media_type="text/event-stream",
|
||
headers={"Cache-Control": "no-cache",
|
||
"X-Accel-Buffering": "no",
|
||
"Connection": "keep-alive"})
|
||
|
||
|
||
# 静态资源（app.js / style.css）
|
||
if STATIC_DIR.is_dir():
|
||
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
||
|
||
|
||
# --------------------------------------------------------------- 启动入口
|
||
def port_in_use(host: str, port: int) -> bool:
|
||
"""端口是否已被占用（用来给中文提示，而不是把 uvicorn 的 errno 48 直接甩给用户）。"""
|
||
import socket
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
||
try:
|
||
sock.bind((host, port))
|
||
return False
|
||
except OSError:
|
||
return True
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="tcaptcha Web 观测台")
|
||
parser.add_argument("--host", default="127.0.0.1",
|
||
help="监听地址（默认 127.0.0.1 仅本机；0.0.0.0 供局域网访问，注意无鉴权）")
|
||
parser.add_argument("--port", type=int, default=8770, help="监听端口（默认 8770）")
|
||
parser.add_argument("--no-warmup", action="store_true", help="不预热 AI 推理引擎")
|
||
args = parser.parse_args()
|
||
|
||
logging.basicConfig(level=logging.INFO,
|
||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||
datefmt="%H:%M:%S")
|
||
|
||
if port_in_use(args.host, args.port):
|
||
print("\n ✗ 端口 %d 已被占用，服务没起来。\n"
|
||
" 查占用进程：lsof -nP -iTCP:%d -sTCP:LISTEN\n"
|
||
" 停掉它 ：kill <上面的 PID>\n"
|
||
" 或换端口 ：--port 8771\n" % (args.port, args.port), flush=True)
|
||
return 2
|
||
|
||
if args.host not in ("127.0.0.1", "localhost"):
|
||
print("\n ⚠️ 监听地址是 %s —— 界面**没有任何鉴权**，别暴露到公网。\n" % args.host, flush=True)
|
||
|
||
if not args.no_warmup:
|
||
def _warm() -> None:
|
||
try:
|
||
from tcaptcha.web.runner import cached_solvers
|
||
cached_solvers() # 预加载 ONNX/ddddocr，避免第一次任务等几秒
|
||
logging.getLogger("web.app").info("AI 推理引擎预热完成（进程内复用）")
|
||
except Exception as exc: # noqa: BLE001
|
||
logging.getLogger("web.app").warning("预热失败（不影响使用）: %s", exc)
|
||
|
||
import threading
|
||
threading.Thread(target=_warm, name="warmup", daemon=True).start()
|
||
|
||
import uvicorn
|
||
print("\n tcaptcha 观测台已启动 → http://%s:%d\n"
|
||
" （Ctrl+C 停止；靶子带实测最小间隔，1ms 为 60s，后端会自动节流）\n"
|
||
% (args.host, args.port), flush=True)
|
||
try:
|
||
uvicorn.run(app, host=args.host, port=args.port, log_level="info")
|
||
except KeyboardInterrupt:
|
||
pass
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|