""" 腾讯天御点选验证码 (Tencent Turing Captcha) 样本纯协议采集器 通过模拟前端两步 HTTP 接口直接拉取高清原图： 1. GET /cap_union_prehandle 获取初始会话 sess 2. POST /cap_union_new_getsig 获取题目配置与背景/提示图 URL 3. 并行下载 672x480 背景大图与 170x50 提示图标图 """ import argparse import asyncio import json import logging import os import random import re import time from pathlib import Path from typing import Optional, Tuple import httpx logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S", ) logger = logging.getLogger("tcaptcha_collector") class TencentCaptchaCollector: BASE_URL = "https://turing.captcha.qcloud.com" DEFAULT_AID = "2039519451" def __init__(self, aid: str = DEFAULT_AID, output_dir: str = "tcaptcha/data/raw", timeout: float = 10.0): self.aid = aid self.output_dir = Path(output_dir) self.output_dir.mkdir(parents=True, exist_ok=True) self.index_file = self.output_dir / "index.jsonl" self.timeout = timeout self.client = httpx.Client( base_url=self.BASE_URL, trust_env=False, headers={ "User-Agent": ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/152.0.0.0 Safari/537.36" ), "Referer": "https://turing.captcha.qcloud.com/", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", }, timeout=self.timeout, follow_redirects=True, ) def _fetch_session(self) -> Tuple[Optional[str], Optional[str]]: """第一步：预检请求获取 sess""" cb_name = f"_aq_{random.randint(100000, 999999)}" tkid = str(random.randint(100000000, 999999999)) ua_b64 = "TW96aWxsYS81LjAgKE1hY2ludG9zaDsgSW50ZWwgTWFjIE9TIFggMTBfMTVfNykgQXBwbGVXZWJLaXQvNTM3LjM2IChLSFRNTCwgbGlrZSBHZWNrbykgQ2hyb21lLzE1Mi4wLjAuMCBTYWZhcmkvNTM3LjM2" params = { "aid": self.aid, "protocol": "https", "accver": "1", "showtype": "popup", "ua": ua_b64, "noheader": "1", "fb": "1", "grayscale": "1", "clientype": "2", "lang": "zh-cn", "js": "/tcaptcha-frame.ba387dd1.js", "wb": "2", "tkid": tkid, "subsid": "1", "callback": cb_name, } resp = self.client.get("/cap_union_prehandle", params=params) if resp.status_code != 200: logger.error("prehandle failed with status %d", resp.status_code) return None, None match = re.search(rf"{cb_name}\((.*)\)", resp.text) if not match: logger.error("Failed to parse prehandle response: %s", resp.text[:100]) return None, None try: data = json.loads(match.group(1)) sess = data.get("sess") return sess, tkid except Exception as e: logger.error("Failed to decode JSON from prehandle: %s", e) return None, None def _fetch_sig_and_images(self, sess: str, tkid: str) -> Optional[dict]: """第二步：请求题目信息并下载图片对""" resp = self.client.post( "/cap_union_new_getsig", data={"sess": sess, "tkid": tkid}, headers={"Content-Type": "application/x-www-form-urlencoded"}, ) if resp.status_code != 200: logger.error("getsig failed with status %d", resp.status_code) return None try: res_json = resp.json() except Exception as e: logger.error("Failed to decode getsig JSON: %s", e) return None if str(res_json.get("ret")) != "0": logger.warning("getsig returned non-zero ret: %s", res_json) return None cfg = res_json.get("data", {}) instruction = cfg.get("instruction", "请依次点击：") bg_cfg = cfg.get("bg_elem_cfg", {}) img_url_bg = bg_cfg.get("img_url", "") if not img_url_bg: logger.warning("No img_url found in getsig response") return None img_url_prompt = img_url_bg.replace("img_index=1", "img_index=0") # 下载背景大图 (672x480) 与 提示图 (170x50) bg_resp = self.client.get(img_url_bg) prompt_resp = self.client.get(img_url_prompt) if bg_resp.status_code != 200 or prompt_resp.status_code != 200: logger.error("Failed to download image pairs") return None return { "instruction": instruction, "bg_bytes": bg_resp.content, "prompt_bytes": prompt_resp.content, "img_url_bg": img_url_bg, "img_url_prompt": img_url_prompt, "sess": sess, "tkid": tkid, } def fetch_one(self) -> Optional[str]: """采集单组样本并保存到本地""" sess, tkid = self._fetch_session() if not sess or not tkid: return None data = self._fetch_sig_and_images(sess, tkid) if not data: return None sample_id = f"{int(time.time() * 1000)}_{random.randint(1000, 9999)}" bg_filename = f"{sample_id}_bg.png" prompt_filename = f"{sample_id}_prompt.png" meta_filename = f"{sample_id}_meta.json" bg_path = self.output_dir / bg_filename prompt_path = self.output_dir / prompt_filename meta_path = self.output_dir / meta_filename bg_path.write_bytes(data["bg_bytes"]) prompt_path.write_bytes(data["prompt_bytes"]) meta_info = { "sample_id": sample_id, "timestamp": time.time(), "aid": self.aid, "instruction": data["instruction"], "bg_file": bg_filename, "prompt_file": prompt_filename, "bg_size": len(data["bg_bytes"]), "prompt_size": len(data["prompt_bytes"]), "sess": data["sess"], "tkid": data["tkid"], } meta_path.write_text(json.dumps(meta_info, ensure_ascii=False, indent=2), encoding="utf-8") # 写入 index.jsonl with open(self.index_file, "a", encoding="utf-8") as f: f.write(json.dumps(meta_info, ensure_ascii=False) + "\n") logger.info( "✓ 采集成功 [%s]: 大图 %d KB, 提示图 %d KB | 提示: %s", sample_id, len(data["bg_bytes"]) // 1024, len(data["prompt_bytes"]) // 1024, data["instruction"], ) return sample_id def collect_batch(self, count: int = 10, delay: float = 0.8) -> int: """批量采集样本""" logger.info("开始批量采集点选验证码样本，目标数量: %d, 间隔: %.2fs", count, delay) success_count = 0 for i in range(count): try: sid = self.fetch_one() if sid: success_count += 1 else: logger.warning("第 %d/%d 个样本采集失败，准备重试", i + 1, count) except Exception as e: logger.error("采集异常: %s", e) if i < count - 1 and delay > 0: # 随机微弱浮动，模拟正常采集 actual_delay = delay + random.uniform(-0.2, 0.3) time.sleep(max(0.2, actual_delay)) logger.info("批量采集完成: 成功 %d/%d (存储目录: %s)", success_count, count, self.output_dir) return success_count def close(self): self.client.close() def main(): parser = argparse.ArgumentParser(description="腾讯天御点选验证码纯协议批量采集器") parser.add_argument("--count", type=int, default=10, help="采集样本数量 (默认: 10)") parser.add_argument("--delay", type=float, default=0.8, help="每次请求间隔秒数 (默认: 0.8)") parser.add_argument("--aid", type=str, default=TencentCaptchaCollector.DEFAULT_AID, help="业务 AppID") parser.add_argument("--output-dir", type=str, default="tcaptcha/data/raw", help="保存路径") args = parser.parse_args() collector = TencentCaptchaCollector(aid=args.aid, output_dir=args.output_dir) try: collector.collect_batch(count=args.count, delay=args.delay) finally: collector.close() if __name__ == "__main__": main()