腾讯天御点选验证码

""" 极简本地 Web 点选验证码打标工具启动方式: uv run python tcaptcha/web_labeler.py --port 8765 访问地址: http://127.0.0.1:8765 """ import argparse import base64 import json import logging from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from io import BytesIO from pathlib import Path from urllib.parse import parse_qs, urlparse from PIL import Image try: from tcaptcha.slicer import slice_prompt_image except ImportError: from slicer import slice_prompt_image logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") logger = logging.getLogger("web_labeler") RAW_DIR = Path("tcaptcha/data/raw") LABEL_FILE = Path("tcaptcha/data/labels.jsonl") CROPS_DIR = Path("tcaptcha/data/crops") def get_labeled_sample_ids() -> set: if not LABEL_FILE.exists(): return set() labeled = set() with open(LABEL_FILE, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: try: record = json.loads(line) labeled.add(record.get("sample_id")) except Exception: pass return labeled def get_all_samples() -> list: if not RAW_DIR.exists(): return [] meta_files = sorted(RAW_DIR.glob("*_meta.json")) samples = [] for mf in meta_files: try: meta = json.loads(mf.read_text(encoding="utf-8")) samples.append(meta) except Exception: pass return samples def get_sample_data(sample_id: str): bg_path = RAW_DIR / f"{sample_id}_bg.png" prompt_path = RAW_DIR / f"{sample_id}_prompt.png" meta_path = RAW_DIR / f"{sample_id}_meta.json" if not bg_path.exists() or not prompt_path.exists() or not meta_path.exists(): return None meta = json.loads(meta_path.read_text(encoding="utf-8")) # 切割提示图标 crops = slice_prompt_image(prompt_path) sliced_b64 = [] for idx, (crop_img, _) in enumerate(crops): buf = BytesIO() crop_img.save(buf, format="PNG") b64 = base64.b64encode(buf.getvalue()).decode("utf-8") sliced_b64.append(f"data:image/png;base64,{b64}") bg_b64 = base64.b64encode(bg_path.read_bytes()).decode("utf-8") prompt_b64 = base64.b64encode(prompt_path.read_bytes()).decode("utf-8") return { "sample_id": sample_id, "instruction": meta.get("instruction", "请依次点击："), "bg_b64": f"data:image/png;base64,{bg_b64}", "prompt_b64": f"data:image/png;base64,{prompt_b64}", "sliced_prompts": sliced_b64, } HTML_TEMPLATE = """ 腾讯天御点选验证码 - 标注工作台

请依次点击：

快捷键: 点击大图标记坐标 | [Z] 撤销 | [R] 重置 | [S] 跳过 | [Enter/Space] 保存

""" class LabelHandler(BaseHTTPRequestHandler): def do_GET(self): url = urlparse(self.path) if url.path in ["/", "/index.html"]: self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() self.wfile.write(HTML_TEMPLATE.encode("utf-8")) return if url.path == "/api/next": all_samples = get_all_samples() labeled_ids = get_labeled_sample_ids() unlabeled = [s for s in all_samples if s.get("sample_id") not in labeled_ids] if not unlabeled: resp_data = {"finished": True, "stats": {"total": len(all_samples), "labeled": len(labeled_ids)}} else: next_sample = unlabeled[0] sample_data = get_sample_data(next_sample["sample_id"]) resp_data = { "finished": False, "sample": sample_data, "stats": {"total": len(all_samples), "labeled": len(labeled_ids)}, } self.send_response(200) self.send_header("Content-Type", "application/json; charset=utf-8") self.end_headers() self.wfile.write(json.dumps(resp_data).encode("utf-8")) return self.send_response(404) self.end_headers() def do_POST(self): url = urlparse(self.path) if url.path == "/api/save": content_len = int(self.headers.get("Content-Length", 0)) body = self.rfile.read(content_len).decode("utf-8") data = json.loads(body) sample_id = data.get("sample_id") points = data.get("points", []) if not sample_id or len(points) != 3: self.send_response(400) self.end_headers() return LABEL_FILE.parent.mkdir(parents=True, exist_ok=True) CROPS_DIR.mkdir(parents=True, exist_ok=True) # 保存切片作为正样本 bg_path = RAW_DIR / f"{sample_id}_bg.png" if bg_path.exists(): bg_img = Image.open(bg_path).convert("RGBA") for i, pt in enumerate(points): cx, cy = pt["x"], pt["y"] # 裁剪 54x54 小图块 crop_box = (max(0, cx - 27), max(0, cy - 27), min(672, cx + 27), min(480, cy + 27)) crop_patch = bg_img.crop(crop_box) crop_patch.save(CROPS_DIR / f"{sample_id}_target_{i+1}.png") # 写入 labels.jsonl record = { "sample_id": sample_id, "points": points, "ans_format": [{"elem_id": i + 1, "type": "DynAnswerType_POS", "data": f"{p['x']},{p['y']}"} for i, p in enumerate(points)], } with open(LABEL_FILE, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") logger.info("✓ 成功保存样本标注: %s -> %s", sample_id, points) self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(b'{"status":"ok"}') return self.send_response(404) self.end_headers() def main(): parser = argparse.ArgumentParser(description="腾讯点选验证码轻量 Web 打标工具") parser.add_argument("--port", type=int, default=8765, help="本地服务端口 (默认: 8765)") args = parser.parse_args() server = ThreadingHTTPServer(("127.0.0.1", args.port), LabelHandler) logger.info("🚀 打标工作台已启动: http://127.0.0.1:%d", args.port) logger.info("在浏览器打开上述地址即可开始打标，按 Ctrl+C 停止") try: server.serve_forever() except KeyboardInterrupt: logger.info("打标服务已停止") if __name__ == "__main__": main()