# -*- coding: utf-8 -*- """ 毫秒 (1ms.run) 业务适配器 —— 腾讯天御点选验证码凭证获取 协议来源: `/Users/yml/Desktop/抓包/毫秒/点选.har` 从抓包可还原的链路 (本适配器已全部实现): 第一步 毫秒取验证码参数 POST https://1ms.run/api/v1/captcha/params body {"scene": "checkin"} (也可以带 identity_context) -> {"provider":"tencent","params":{"captchaAppId":195672344}, "sessionUuid":"01a12a04-e5e7-726b-b00c-2f54b2e1c167"} 第二步 腾讯天御 cap_union_new 全套纯协议解题 (prehandle -> PoW -> tdc.js -> getcapbysig -> verify) -> {"ticket": "tr03...", "randstr": "@..."} 说明与边界 ---------- * `identity_context` 在抓包中是页面内加密上下文 (emctx1.*), 实测 **不传同样可以** 拿到 `captchaAppId`, 因此本适配器不依赖它; 若后续业务要求, 可通过参数显式传入。 * 抓包中 **不包含** 最终的签到提交请求 (`/api/v1/...` 业务接口), 因此本适配器只负责 产出 `ticket` + `randstr`, 并给出一个可配置的 `submit_business()` 便于对接真实业务 接口 (端点与字段需按实际业务抓包填写, 严禁凭猜测写死)。 """ import logging from dataclasses import dataclass from typing import Dict, Optional, Tuple import httpx from tcaptcha.client import KIND_TEXT_CLICK, TencentCaptchaClient logger = logging.getLogger("one_ms_adapter") ONE_MS_BASE = "https://1ms.run" DEFAULT_SCENE = "checkin" DEFAULT_REFERER = "https://1ms.run/console/personal/benefits/check-in" DEFAULT_USER_AGENT = ("Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/155.0.0.0 Safari/537.36") @dataclass class CaptchaParams: provider: str app_id: str session_uuid: str enabled: bool class OneMsAdapter: """1ms.run 验证码凭证适配器""" def __init__(self, cookie: Optional[str] = None, user_agent: str = DEFAULT_USER_AGENT, timeout: float = 15.0, proxy: Optional[str] = None, proxy_api: Optional[str] = None, telemetry_mode: str = "python"): self.user_agent = user_agent self.timeout = timeout # 腾讯天御按出口 IP 做滚动风控 (errorCode=12); 给了 proxy_api 就能自动换 IP 重试 self.proxy = proxy self.proxy_api = proxy_api self.telemetry_mode = telemetry_mode headers = { "User-Agent": user_agent, "Origin": ONE_MS_BASE, "Referer": DEFAULT_REFERER, "Accept": "application/json, text/plain, */*", "Content-Type": "application/json", } if cookie: headers["Cookie"] = cookie self.http = httpx.Client(base_url=ONE_MS_BASE, headers=headers, timeout=timeout, trust_env=False, follow_redirects=True) # ---------------------------------------------------------- 第一步 def fetch_captcha_params(self, scene: str = DEFAULT_SCENE, identity_context: str = "") -> CaptchaParams: """POST /api/v1/captcha/params —— 取当前场景的验证码配置""" payload: Dict[str, str] = {"scene": scene} if identity_context: payload["identity_context"] = identity_context resp = self.http.post("/api/v1/captcha/params", json=payload) data = resp.json() if data.get("code") != 0: raise RuntimeError(f"取验证码参数失败: {data}") info = data.get("data") or {} params = info.get("params") or {} return CaptchaParams( provider=info.get("provider", ""), app_id=str(params.get("captchaAppId", "")), session_uuid=info.get("sessionUuid", ""), enabled=bool(info.get("enabled")), ) # ---------------------------------------------------------- 第二步 def get_captcha_ticket(self, scene: str = DEFAULT_SCENE, max_retries: int = 3, max_proxy_rotations: int = 3) -> Tuple[Optional[str], Optional[str], Dict]: """ 取得可提交给毫秒业务接口的 (ticket, randstr)。 若配置了 `proxy_api`, 在遇到 `errorCode=12` (f_limit, IP 级滚动风控) 时会 自动换一个出口 IP 重试, 最多 `max_proxy_rotations` 次。 """ params = self.fetch_captcha_params(scene=scene) logger.info("毫秒验证码参数: provider=%s appId=%s enabled=%s sessionUuid=%s", params.provider, params.app_id, params.enabled, params.session_uuid) if not params.enabled or not params.app_id: return None, None, {"error": "该场景未启用验证码"} if params.provider != "tencent": return None, None, {"error": f"暂只支持 tencent provider, 实际为 {params.provider}"} proxy = self.proxy last: Dict = {} for rotation in range(max(1, max_proxy_rotations)): client = TencentCaptchaClient( aid=params.app_id, entry_url=DEFAULT_REFERER, referer=ONE_MS_BASE + "/", wb="1", accept_kinds=(KIND_TEXT_CLICK,), proxy=proxy, telemetry_mode=self.telemetry_mode, ) try: result = client.solve(max_retries=max_retries, max_refreshes=8) finally: client.close() last = result if result.get("success"): return result.get("ticket") or None, result.get("randstr") or None, result if "12" not in str(result.get("error", "")) or not self.proxy_api: break from tcaptcha.proxy import fetch_proxy proxy = fetch_proxy(self.proxy_api) logger.warning("触发 errorCode=12, 换出口 IP 重试 (%d/%d): %s", rotation + 1, max_proxy_rotations, proxy) if not proxy: break return None, None, last # ---------------------------------------------------------- 业务提交 def submit_business(self, path: str, payload: Dict, method: str = "POST") -> Dict: """ 用 ticket/randstr 提交真实业务请求。 注意: 点选.har 中 **没有** 最终的签到业务请求, 因此这里不预设端点与字段, 由调用方按实际业务抓包传入 (通常是 {"vticket": ticket, "vrandstr": randstr, ...})。 """ if method.upper() == "POST": resp = self.http.post(path, json=payload) else: resp = self.http.get(path, params=payload) try: return resp.json() except Exception: return {"code": resp.status_code, "text": resp.text} def close(self): self.http.close() if __name__ == "__main__": import argparse import json logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") ap = argparse.ArgumentParser(description="1ms.run 验证码票据获取") ap.add_argument("--scene", default=DEFAULT_SCENE) ap.add_argument("--cookie", default="") ap.add_argument("--proxy", default=None, help="出口代理, 如 http://1.2.3.4:8080") ap.add_argument("--proxy-api", default=None, help="代理提取 API (遇到 errorCode=12 自动换 IP)") args = ap.parse_args() adapter = OneMsAdapter(cookie=args.cookie or None, proxy=args.proxy, proxy_api=args.proxy_api) try: ticket, randstr, raw = adapter.get_captcha_ticket(scene=args.scene) finally: adapter.close() print(json.dumps({"ticket": ticket, "randstr": randstr, "error": raw.get("error", "")}, ensure_ascii=False, indent=2))