# -*- 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())