diff --git a/.gitignore b/.gitignore index f76dbe6..f915d70 100644 --- a/.gitignore +++ b/.gitignore @@ -24,3 +24,14 @@ tcaptcha/data/crops/ data/ runs/ yolov8n.pt + +# 本地环境变量配置(含真实账号密码,防止泄露) +.env +.env.* + +# 抓包文件与临时压缩包 +har/ +*.har +*.zip + + diff --git a/dist/rainyun_qinglong/README.md b/dist/rainyun_qinglong/README.md new file mode 100644 index 0000000..7c991c5 --- /dev/null +++ b/dist/rainyun_qinglong/README.md @@ -0,0 +1,53 @@ +# 雨云 (Rainyun) 自动登录与每日签到 (青龙面板平铺单目录部署包) + +本目录为纯净的**单层平铺目录**结构,无需嵌套子文件夹,无需 `uv` 或 `venv`,可直接压缩上传至青龙面板或服务器运行。 + +--- + +## 目录文件清单 + +```text +rainyun_qinglong/ +├── rainyun_checkin.py # 核心签到脚本 (青龙定时任务入口) +├── client.py # 腾讯验证码协议客户端 +├── crypto.py # 腾讯 TDC 混沌虚拟机 32 轮魔改 XTEA 纯 Python 还原 +├── solver.py # AI 视觉检测推理引擎 (YOLOv8 + ResNet18) +├── slicer.py # 提示图投影切片 +├── server.py # (可选) 独立 HTTP 验证码微服务 +├── yolo_detector.pt # YOLO 图标候选定位权重文件 +├── icon_classifier.pt # ResNet18 图标分类识别权重文件 +├── .env # 本地配置文件 (格式: RAINYUN_ACCOUNT="账号#密码") +├── .env.example # 环境变量示例文件 +└── README.md # 本说明文档 +``` + +--- + +## 青龙面板极简部署步骤 + +### 1. 上传文件 +将本目录下的所有文件上传至青龙面板的脚本目录: +`/ql/data/scripts/` (新版青龙)或 `/ql/scripts/` (旧版青龙)。 + +### 2. 在青龙后台安装依赖 +进入青龙面板 ->【依赖管理】->【Python3】-> 点击【添加依赖】,依次添加并安装: +```text +httpx +torch +torchvision +ultralytics +opencv-python-headless +pillow +numpy +``` + +### 3. 配置账号环境变量 +在青龙面板【环境变量】添加变量: +- **名称**:`RAINYUN_ACCOUNT` +- **值**:`账号#密码` (多个账号用 `&` 分隔,例如 `13700000000#pwd1&13800000000#pwd2`) + +### 4. 添加定时任务 +在青龙面板【定时任务】点击【新建任务】: +- **任务名**:`雨云每日签到` +- **命令**:`task rainyun_checkin.py` +- **定时规则**:`30 8 * * *` (每天早上 8:30 自动执行) diff --git a/dist/rainyun_qinglong/client.py b/dist/rainyun_qinglong/client.py new file mode 100644 index 0000000..d06adc2 --- /dev/null +++ b/dist/rainyun_qinglong/client.py @@ -0,0 +1,323 @@ +""" +腾讯天御点选验证码工业级纯协议客户端 (Tencent Turing Captcha Protocol Client) +实现端到端全自动纯协议闭环: +1. GET /cap_union_prehandle 获取初始会话与 PoW 题目 +2. 毫秒级求解 Proof of Work 哈希碰撞 +3. 并发拉取动态混淆的 tdc.js 脚本 +4. POST /cap_union_new_getsig 获取当前轮次题目与图片 +5. YOLOv8n + ResNet18 模型进行视觉检测、分类与全局非冲突点选排序 +6. 纯算法还原 (Custom XTEA + 贝塞尔人手微颤) 生成行为轨迹密文 collect 与会话密钥 eks (100% 纯 Python) +7. POST /cap_union_new_verify 提交验证,获取最终 ticket 与 randstr +""" + +import hashlib +import json +import logging +import random +import re +import sys +import time +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +import httpx + +CURRENT_DIR = Path(__file__).resolve().parent +if str(CURRENT_DIR) not in sys.path: + sys.path.insert(0, str(CURRENT_DIR)) + +try: + from crypto import extract_eks, generate_collect + from solver import CaptchaSolver +except ImportError: + from tcaptcha.crypto import extract_eks, generate_collect + from tcaptcha.solver import CaptchaSolver + +logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") +logger = logging.getLogger("tcaptcha_client") + + +class TencentCaptchaClient: + BASE_URL = "https://turing.captcha.qcloud.com" + DEFAULT_AID = "2039519451" + + def __init__( + self, + aid: str = DEFAULT_AID, + solver: Optional[CaptchaSolver] = None, + timeout: float = 15.0, + user_agent: Optional[str] = None, + ): + self.aid = aid + self.timeout = timeout + self.user_agent = user_agent or ( + "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" + ) + self.solver = solver or CaptchaSolver() + + self.http_client = httpx.Client( + base_url=self.BASE_URL, + trust_env=False, # 屏蔽环境代理干扰 + headers={ + "User-Agent": self.user_agent, + "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 _solve_pow(self, prefix: str, target_md5: str) -> Tuple[str, int]: + """求解 Proof of Work 哈希碰撞""" + t0 = time.time() + for i in range(2000000): + candidate = f"{prefix}{i}" + if hashlib.md5(candidate.encode("utf-8")).hexdigest() == target_md5: + calc_time = max(1, round((time.time() - t0) * 1000)) + return candidate, calc_time + raise RuntimeError(f"PoW MD5 collision search failed for prefix {prefix}") + + def _generate_telemetry(self, tdc_code: str, click_points: List[Tuple[int, int]]) -> Dict[str, str]: + """纯 Python 算法生成真实行为轨迹与遥测密文 collect 及 eks (100% 纯算免 Node.js)""" + eks = extract_eks(tdc_code) + collect = generate_collect(click_points, aid=self.aid, user_agent=self.user_agent) + return { + "collect": collect, + "tlg": str(len(collect)), + "eks": eks, + } + + def solve(self, max_retries: int = 2) -> Dict: + """ + 端到端全自动解题并提交验证 + 返回格式: + { + "success": bool, + "ticket": str, + "randstr": str, + "elapsed_ms": int, + "points": [(x1, y1), (x2, y2), (x3, y3)], + "raw_response": dict, + "error": str + } + """ + start_time = time.time() + active_sess = None + tkid = None + tdc_code = None + pow_answer = None + pow_calc_time = None + last_raw = {} + last_error = "Max retries exceeded or verification rejected" + + for attempt in range(max_retries): + try: + # 1. 预检请求 (Prehandle) 或在会话内刷新 + cb_name = f"_aq_{random.randint(100000, 999999)}" + if not tkid: + tkid = str(random.randint(100000000, 999999999)) + + if not active_sess: + # 首次进入:执行 Prehandle + ua_b64 = "TW96aWxsYS81LjAgKE1hY2ludG9zaDsgSW50ZWwgTWFjIE9TIFggMTBfMTVfNykgQXBwbGVXZWJLaXQvNTM3LjM2IChLSFRNTCwgbGlrZSBHZWNrbykgQ2hyb21lLzE1Mi4wLjAuMCBTYWZhcmkvNTM3LjM2" + params = { + "aid": self.aid, + "protocol": "https", + "accver": "1", + "showtype": "popup", + "ua": ua_b64, + "noheader": "1", + "fb": "1", + "aged": "0", + "enableAged": "0", + "enableDarkMode": "0", + "grayscale": "1", + "clientype": "2", + "lang": "zh-cn", + "entry_url": "https://app.rainyun.com/account/reward/earn#", + "elder_captcha": "0", + "js": "/tcaptcha-frame.ba387dd1.js", + "wb": "2", + "tkid": tkid, + "subsid": str(random.randint(1, 5)), + "callback": cb_name, + } + + pre_resp = self.http_client.get( + "/cap_union_prehandle", + params=params, + headers={"Referer": "https://app.rainyun.com/"}, + ) + match = re.search(rf"{cb_name}\((.*)\)", pre_resp.text) + if not match: + logger.warning("Failed to parse prehandle callback, retrying...") + continue + + pre_data = json.loads(match.group(1)) + initial_sess = pre_data.get("sess") + comm_cfg = pre_data.get("data", {}).get("comm_captcha_cfg", {}) + + # 2. 计算 PoW + pow_cfg = comm_cfg.get("pow_cfg", {}) + pow_answer, pow_calc_time = self._solve_pow(pow_cfg["prefix"], pow_cfg["md5"]) + + # 3. 拉取当前会话专属动态 tdc.js + tdc_path = comm_cfg.get("tdc_path", "") + tdc_resp = self.http_client.get( + tdc_path, + headers={"Referer": "https://turing.captcha.gtimg.com/"}, + ) + tdc_code = tdc_resp.text + current_sess = initial_sess + else: + # 会话内刷新:直接使用前一轮返回的 sess + current_sess = active_sess + + # 4. 获取题目信息与图片 URL (POST /cap_union_new_getsig) + sig_resp = self.http_client.post( + "/cap_union_new_getsig", + data={"sess": current_sess, "tkid": tkid}, + headers={ + "Content-Type": "application/x-www-form-urlencoded", + "Origin": "https://turing.captcha.gtimg.com", + "Referer": "https://turing.captcha.gtimg.com/", + }, + ) + sig_json = sig_resp.json() + if str(sig_json.get("ret")) != "0": + logger.warning(f"getsig returned non-zero ret: {sig_json}") + active_sess = None + continue + + dyn_info = sig_json.get("data", {}) + bg_url = dyn_info.get("bg_elem_cfg", {}).get("img_url", "") + prompt_url = bg_url.replace("img_index=1", "img_index=0") + + # 提取图片 URL 中的最新有效 sess + sess_match = re.search(r"sess=([^&]+)", bg_url) + active_sess = sess_match.group(1) if sess_match else current_sess + + # 5. 下载背景大图与提示图标图 + bg_bytes = self.http_client.get( + bg_url, + headers={"Referer": "https://turing.captcha.gtimg.com/"}, + ).content + prompt_bytes = self.http_client.get( + prompt_url, + headers={"Referer": "https://turing.captcha.gtimg.com/"}, + ).content + + # 6. AI 模型高精度求解 + pts = self.solver.solve(bg_bytes, prompt_bytes) + if len(pts) < 3: + logger.warning("Solver failed to find 3 valid points, retrying...") + continue + + # 7. 格式化答案 ans + ans_list = [ + {"elem_id": 1, "type": "DynAnswerType_POS", "data": f"{pts[0][0]},{pts[0][1]}"}, + {"elem_id": 2, "type": "DynAnswerType_POS", "data": f"{pts[1][0]},{pts[1][1]}"}, + {"elem_id": 3, "type": "DynAnswerType_POS", "data": f"{pts[2][0]},{pts[2][1]}"}, + ] + ans_str = json.dumps(ans_list, separators=(",", ":")) + + # 8. 纯算法生成真实行为轨迹 collect 与会话密钥 eks (100% 纯 Python) + telemetry = self._generate_telemetry(tdc_code, pts) + + # 9. 提交验证 POST /cap_union_new_verify + verify_body = { + "ans": ans_str, + "collect": telemetry["collect"], + "tlg": telemetry["tlg"], + "eks": telemetry["eks"], + "sess": active_sess, + "tkid": tkid, + "pow_answer": pow_answer, + "pow_calc_time": str(pow_calc_time), + } + + verify_resp = self.http_client.post( + "/cap_union_new_verify", + data=verify_body, + headers={ + "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8", + "Origin": "https://turing.captcha.gtimg.com", + "Referer": "https://turing.captcha.gtimg.com/", + }, + ) + + v_json = verify_resp.json() + total_elapsed = int((time.time() - start_time) * 1000) + + if str(v_json.get("errorCode")) == "0": + return { + "success": True, + "ticket": v_json.get("ticket", ""), + "randstr": v_json.get("randstr", ""), + "elapsed_ms": total_elapsed, + "points": pts, + "raw_response": v_json, + "error": "", + } + else: + err_code = str(v_json.get("errorCode")) + err_msg = v_json.get("errMessage", "") + if err_code == "12": + last_error = f"触发腾讯接口频控限制 (errorCode: 12, f_limit),建议稍候或更换IP" + elif err_code == "50": + last_error = f"验证坐标未命中或会话需刷新 (errorCode: 50)" + else: + last_error = f"验证失败 (errorCode: {err_code}, msg: {err_msg})" + + logger.warning(f"Attempt {attempt + 1} verify: {last_error}") + last_raw = v_json + # 如果接口返回了新的会话 sess,下一轮直接沿用该会话重试刷新 + new_sess = v_json.get("sess") + if new_sess: + active_sess = new_sess + else: + active_sess = None + except Exception as e: + last_error = f"请求异常: {e}" + logger.error(f"Attempt {attempt + 1} encountered exception: {e}") + + total_elapsed = int((time.time() - start_time) * 1000) + return { + "success": False, + "ticket": "", + "randstr": "", + "elapsed_ms": total_elapsed, + "points": [], + "raw_response": last_raw, + "error": last_error, + } + + +def main(): + import argparse + + parser = argparse.ArgumentParser(description="Tencent Captcha Pure Protocol Solver Client") + parser.add_argument("--aid", default="2039519451", help="Captcha AppID (default: 2039519451 for Rainyun)") + args = parser.parse_args() + + client = TencentCaptchaClient(aid=args.aid) + print(f"[*] 正在发起腾讯天御验证码纯协议识别 (AppID: {args.aid})...") + result = client.solve() + + if result["success"]: + print("\n" + "=" * 50) + print(" 🎉 验证码纯协议求解成功!") + print("=" * 50) + print(f"耗时: {result['elapsed_ms']} ms") + print(f"点击坐标: {result['points']}") + print(f"randstr: {result['randstr']}") + print(f"ticket: {result['ticket']}") + print("=" * 50) + else: + print(f"\n❌ 验证失败: {result['error']}") + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/dist/rainyun_qinglong/crypto.py b/dist/rainyun_qinglong/crypto.py new file mode 100644 index 0000000..4433e97 --- /dev/null +++ b/dist/rainyun_qinglong/crypto.py @@ -0,0 +1,312 @@ +""" +Pure Python implementation of Tencent TDC Chaos VM (JSVMP) Cryptographic +and Telemetry generation algorithm. + +Eliminates the Node.js sandbox dependency by replicating the 32-round custom +XTEA block cipher, human-like mouse movement trajectory generation, and +telemetry payload serialization. +""" + +import base64 +import math +import random +import re +import struct +import time +from typing import List, Tuple + +# TDC VM Cryptographic Constants +DELTA = 2654435769 # 0x9E3779B9 +DEFAULT_KEY = [1448563793, 1847730015, 1733184827, 1364613714] + +DEFAULT_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" +) +DEFAULT_RENDERER = "Apple M1" +DEFAULT_VENDOR = "Apple" +DEFAULT_CANVAS_FP = ( + "GgoAAAANSUhEUgAAAAEAAAABCAYAAAUhEUgAAAAEAAAABCAYAAAA" + "fFcSJAAAADhgGAWjR9awAAAABJRU5ErkJggg==" +) + + +def xtea_encrypt_block(v0: int, v1: int, key: List[int]) -> Tuple[int, int]: + """ + Tencent custom 32-round XTEA block cipher for a 64-bit block (v0, v1). + Includes specific additive offsets (2829068 and 328986) depending on round index. + """ + v0 &= 0xFFFFFFFF + v1 &= 0xFFFFFFFF + total_sum = 0 + + for _ in range(32): + # Step 1: Update v0 + k_idx1 = total_sum & 3 + shift1 = (((v1 << 4) ^ (v1 >> 5)) + v1) & 0xFFFFFFFF + if k_idx1 == 1: + k_term1 = (total_sum + key[1] + 2829068) & 0xFFFFFFFF + elif k_idx1 == 2: + k_term1 = (total_sum + key[2] + 328986) & 0xFFFFFFFF + else: + k_term1 = (total_sum + key[k_idx1]) & 0xFFFFFFFF + v0 = (v0 + (shift1 ^ k_term1)) & 0xFFFFFFFF + + # Step 2: Update total_sum (standard XTEA delta increment) + total_sum = (total_sum + DELTA) & 0xFFFFFFFFFFFFFFFF + + # Step 3: Update v1 + k_idx2 = (total_sum >> 11) & 3 + shift2 = (((v0 << 4) ^ (v0 >> 5)) + v0) & 0xFFFFFFFF + if k_idx2 == 1: + k_term2 = (total_sum + key[1] + 2829068) & 0xFFFFFFFF + elif k_idx2 == 2: + k_term2 = (total_sum + key[2] + 328986) & 0xFFFFFFFF + else: + k_term2 = (total_sum + key[k_idx2]) & 0xFFFFFFFF + v1 = (v1 + (shift2 ^ k_term2)) & 0xFFFFFFFF + + return v0, v1 + + +def encrypt_string(plaintext: str, key: List[int] = DEFAULT_KEY) -> str: + """ + Encrypts an ASCII/Latin1 plaintext string into Base64 using the custom XTEA cipher. + Input is padded with null bytes to an 8-byte boundary if needed. + """ + data = plaintext.encode("latin1") + pad_len = (8 - (len(data) % 8)) % 8 + data += b"\x00" * pad_len + + out = bytearray() + for i in range(0, len(data), 8): + v0, v1 = struct.unpack(" List[Tuple[int, int]]: + """Generates cubic Bézier interpolation curve points with natural tremor.""" + dx = p1[0] - p0[0] + dy = p1[1] - p0[1] + dist = math.hypot(dx, dy) or 1.0 + + perp_x = -dy / dist + perp_y = dx / dist + curvature = (random.random() - 0.5) * min(50.0, dist * 0.4) + + cp1 = ( + p0[0] + dx * 0.3 + perp_x * curvature, + p0[1] + dy * 0.3 + perp_y * curvature, + ) + cp2 = ( + p0[0] + dx * 0.7 - perp_x * curvature * 0.5, + p0[1] + dy * 0.7 - perp_y * curvature * 0.5, + ) + + pts = [] + for i in range(steps + 1): + t = i / float(steps) + ease_t = t * t * (3.0 - 2.0 * t) + inv_t = 1.0 - ease_t + + bx = ( + inv_t**3 * p0[0] + + 3 * inv_t**2 * ease_t * cp1[0] + + 3 * inv_t * ease_t**2 * cp2[0] + + ease_t**3 * p1[0] + ) + by = ( + inv_t**3 * p0[1] + + 3 * inv_t**2 * ease_t * cp1[1] + + 3 * inv_t * ease_t**2 * cp2[1] + + ease_t**3 * p1[1] + ) + tremor = 0.0 if i in (0, steps) else (random.random() - 0.5) * 1.5 + pts.append((int(round(bx + tremor)), int(round(by + tremor)))) + + return pts + + +def generate_trajectory_payload( + click_points: List[Tuple[int, int]], + target_len: int = 600, +) -> str: + """ + Generates realistic human mouse movement trajectory and serializes it + into the TDC relative-displacement JSON format: + [[x0, y0, 2], [dx1, dy1, dt1], [dx2, dy2, dt2], ...] + + The string is padded with trailing spaces to target_len (multiple of 8). + """ + # Scale from original 672x480 coordinate space to 360x360 captcha iframe display space + scale = 320.0 / 672.0 + mapped_targets = [ + (int(round(20 + p[0] * scale)), int(round(60 + p[1] * scale))) + for p in click_points + ] + + # Starting position near entry area + current_pt = ( + int(round(150 + (random.random() - 0.5) * 60)), + int(round(320 + (random.random() - 0.5) * 20)), + ) + + events: List[List[int]] = [] + events.append([current_pt[0], current_pt[1], 2]) # 2 represents mouse movement event + + last_x, last_y = current_pt[0], current_pt[1] + + for target in mapped_targets: + steps = random.randint(16, 24) + curve = _bezier_interp((last_x, last_y), target, steps) + + for pt in curve[1:]: + dx = pt[0] - last_x + dy = pt[1] - last_y + dt = random.choice([0, 0, 0, 1]) + events.append([dx, dy, dt]) + last_x, last_y = pt[0], pt[1] + + # Dwell hover on target icon + dwell_count = random.randint(3, 5) + for _ in range(dwell_count): + dx = random.randint(-1, 1) + dy = random.randint(-1, 1) + events.append([dx, dy, random.choice([0, 1])]) + last_x += dx + last_y += dy + + import json + raw_json = json.dumps(events, separators=(",", ":")) + + # Pad with spaces to multiple of 8 bytes (matching target_len, e.g. 600) + desired_len = max(len(raw_json), target_len) + pad_needed = (8 - (desired_len % 8)) % 8 + total_len = desired_len + pad_needed + + return raw_json.ljust(total_len, " ") + + +def extract_eks(tdc_js_code: str) -> str: + """Extracts session public key (`eks`) from TDC.js script header.""" + match = re.search(r"window\.KQnlRHY\w+\s*=\s*'([^']+)'", tdc_js_code) + if match: + return match.group(1) + # Fallback to any base64 string assignment + match_fallback = re.search(r"window\.\w+\s*=\s*'([A-Za-z0-9+/=]{100,})'", tdc_js_code) + if match_fallback: + return match_fallback.group(1) + return "" + + +def generate_collect( + click_points: List[Tuple[int, int]], + aid: str = "2039519451", + user_agent: str = DEFAULT_USER_AGENT, + renderer: str = DEFAULT_RENDERER, + vendor: str = DEFAULT_VENDOR, + canvas_fp: str = DEFAULT_CANVAS_FP, +) -> str: + """ + Generates the complete encrypted telemetry payload (`collect`) in 100% pure Python. + + Assembles: + 1. cd prefix (48 bytes, space padded) + 2. mouse trajectory (space padded to 8-byte multiple) + 3. cd post-body (504 bytes, space padded) + 4. sd suffix (16 bytes) + Encrypts each chunk with the 32-round custom XTEA algorithm, concatenates + the Base64 results, and applies TDC URL encoding (%2B, %2F, %3D). + """ + now_sec = int(time.time()) + rand_salt = random.randint(100000000, 999999999) + + # 1. cd prefix (48 bytes) + cd_pre_raw = '{"cd":[[],"1920-1080-1055-24-*-*-|-*",0,0,"",' + cd_pre = cd_pre_raw.ljust(48, " ") + + # 2. Mouse trajectory (600 bytes) + traj_payload = generate_trajectory_payload(click_points, target_len=600) + + # 3. cd post-body (504 bytes) + cd_post_inner = ( + f',"",' + f'"undefined?rand=1519713624347",' + f"1," + f'"{canvas_fp}",' + f"{now_sec}," + f"{rand_salt}," + f"{now_sec}," + f'"",' + f"0," + f'"{user_agent}",' + f'"top",' + f'"MacIntel",' + f'"98k",' + f"40," + f'"",' + f'"{renderer}",' + f"0," + f'"{vendor}",' + f"{now_sec}," + f'"+08",' + f'["zh-CN","zh","en"],' + f"1422008362," + f"2," + f"3," + f'"https://turing.captcha.qcloud.com/cap_union_new_show?aid={aid}",' + f"0," + f"24," + f"22," + f"1920," + f"1," + f"[360,360]," + f"1080]," + ) + cd_post = cd_post_inner.ljust(504, " ") + + # 4. sd suffix (16 bytes) + sd = '"sd":{"od":"C"}}' + + # Encrypt all 4 parts + enc_cd_pre = encrypt_string(cd_pre) + enc_traj = encrypt_string(traj_payload) + enc_cd_post = encrypt_string(cd_post) + enc_sd = encrypt_string(sd) + + combined_b64 = enc_cd_pre + enc_traj + enc_cd_post + enc_sd + + # TDC URL encoding + collect = ( + combined_b64 + .replace("+", "%2B") + .replace("/", "%2F") + .replace("=", "%3D") + ) + return collect + + +if __name__ == "__main__": + # Self-test unit verification + test4 = '"sd":{"od":"C"}}' + exp4 = "nshM2N4iPvX6JRNXi68v5w==" + act4 = encrypt_string(test4) + assert act4 == exp4, f"XTEA unit test failed: {act4} != {exp4}" + + test2 = '{"cd":[[],"1920-1080-1055-24-*-*-|-*",0,0,"", ' + exp2 = "9qJOPk+13StWMa2FQroCP6yba6nGw0BmJaarCCtGtLi0Wfg7WQyVXwMoFAjWcXAj" + act2 = encrypt_string(test2) + assert act2 == exp2, f"XTEA unit test failed: {act2} != {exp2}" + + sample_pts = [(626, 253), (120, 120), (438, 33)] + sample_collect = generate_collect(sample_pts) + print(f"XTEA Unit Tests: PASSED") + print(f"Sample collect length: {len(sample_collect)}") + print(f"Sample collect prefix: {sample_collect[:80]}...") diff --git a/dist/rainyun_qinglong/icon_classifier.pt b/dist/rainyun_qinglong/icon_classifier.pt new file mode 100644 index 0000000..fbee80f Binary files /dev/null and b/dist/rainyun_qinglong/icon_classifier.pt differ diff --git a/dist/rainyun_qinglong/rainyun_checkin.py b/dist/rainyun_qinglong/rainyun_checkin.py new file mode 100644 index 0000000..1793161 --- /dev/null +++ b/dist/rainyun_qinglong/rainyun_checkin.py @@ -0,0 +1,497 @@ +# -*- coding: utf-8 -*- +""" +cron: 30 8 * * * +new Env('雨云每日签到'); + +雨云 (Rainyun) 自动登录与每日签到脚本 (适用于青龙面板) +账号配置规范: + 环境变量 / 本地 .env 统一使用 RAINYUN_ACCOUNT="账号#密码" + 多账号分割: + - 青龙环境变量: 使用 & 分割 (如 RAINYUN_ACCOUNT="13754650804#pwd1&13800000000#pwd2") + - 本地 .env 文件: 支持 & 分割或换行分割 +""" + +import json +import logging +import os +import random +import re +import sys +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import httpx + +# 将项目根目录加入 sys.path 以便加载本地模块 +CURRENT_DIR = Path(__file__).resolve().parent +if str(CURRENT_DIR) not in sys.path: + sys.path.insert(0, str(CURRENT_DIR)) + +# 配置日志格式 +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger("rainyun_checkin") + +# 青龙通知模块导入 +try: + from sendNotify import send +except Exception: + try: + from notify import send + except Exception: + def send(title: str, content: str): + pass + + +# 默认配置 +RAINYUN_API_BASE = "https://api.v2.rainyun.com" +TENCENT_AID = "2039519451" +DEFAULT_UA = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" +) + +# 积分提现比例: 2000 积分 = 1 元现金 (支持直接提现) +POINTS_PER_CNY = 2000.0 + + +def format_points(points: int) -> str: + """格式化积分及等值提现金额 (2000积分 = 1元)""" + cny = points / POINTS_PER_CNY + return f"{points} 积分 (≈ {cny:.2f} 元)" + + +def mask_field(val: str) -> str: + """脱敏账号信息""" + if not val: + return "" + if len(val) <= 4: + return "***" + if len(val) == 11 and val.isdigit(): + return f"{val[:3]}****{val[7:]}" + return f"{val[:2]}****{val[-2:]}" + + +class CaptchaManager: + """验证码求解调度器:支持本地 tcaptcha 模块或远程 HTTP 接口""" + + def __init__(self, api_url: Optional[str] = None): + self.api_url = api_url or os.getenv("TCAPTCHA_API_URL", "").strip() + self.local_client = None + + # 尝试初始化本地客户端 (平铺目录优先 import client,兼容 tcaptcha.client) + if not self.api_url: + try: + from client import TencentCaptchaClient + self.local_client = TencentCaptchaClient(aid=TENCENT_AID) + logger.info("✓ 成功加载本地纯算法求解引擎") + except Exception: + try: + from tcaptcha.client import TencentCaptchaClient + self.local_client = TencentCaptchaClient(aid=TENCENT_AID) + logger.info("✓ 成功加载本地纯算法求解引擎") + except Exception as e: + logger.warning(f"未能加载本地求解引擎 ({e}),将尝试远程接口或环境变量配置") + + def get_ticket(self) -> Tuple[Optional[str], Optional[str]]: + """获取腾讯天御点选验证码票据 (ticket, randstr)""" + # 1. 优先使用远程 API (若已配置) + if self.api_url: + logger.info(f"正在调用远程验证码求解接口: {self.api_url} ...") + try: + with httpx.Client(timeout=30.0, trust_env=False) as client: + resp = client.post(self.api_url, json={"aid": TENCENT_AID}) + if resp.status_code == 200: + data = resp.json() + # 兼容直接返回或嵌套在 data 中 + res_data = data.get("data", data) + ticket = res_data.get("ticket") + randstr = res_data.get("randstr") + if ticket and randstr: + logger.info(f"✓ 远程验证码求解成功 (randstr: {randstr})") + return ticket, randstr + else: + logger.error(f"远程验证码接口未返回有效票据: {data}") + else: + logger.error(f"远程验证码接口响应异常: {resp.status_code} {resp.text}") + except Exception as e: + logger.error(f"调用远程验证码接口失败: {e}") + + # 2. 本地求解引擎 + if self.local_client: + logger.info("正在调用本地 tcaptcha AI 纯协议求解器...") + try: + res = self.local_client.solve() + if res.get("success"): + logger.info(f"✓ 本地求解成功 (耗时: {res.get('elapsed_ms')}ms, randstr: {res.get('randstr')})") + return res.get("ticket"), res.get("randstr") + else: + logger.error(f"本地求解失败: {res.get('error')}") + except Exception as e: + logger.error(f"本地求解器执行异常: {e}") + + logger.error("未找到可用的验证码求解方案!请确认部署了 tcaptcha 本地模型或配置了 TCAPTCHA_API_URL 环境变量") + return None, None + + +class RainyunClient: + """雨云 API 客户端""" + + def __init__( + self, + user: str, + password: str, + captcha_mgr: Optional[CaptchaManager] = None, + ): + self.user = user + self.password = password + self.captcha_mgr = captcha_mgr or CaptchaManager() + + self.account_name = mask_field(user) if user else "未知账号" + self.points_before = 0 + self.points_after = 0 + + # 初始化 HTTP 客户端 + self.session = httpx.Client( + base_url=RAINYUN_API_BASE, + timeout=15.0, + follow_redirects=True, + trust_env=False, # 屏蔽环境代理干扰以兼容各类青龙网络 + ) + self._update_common_headers() + + def _update_common_headers(self): + rys = f"{uuid.uuid4().hex}{time.time():.3f}" + self.session.headers.update({ + "User-Agent": DEFAULT_UA, + "Origin": "https://app.rainyun.com", + "Referer": "https://app.rainyun.com/", + "Accept": "application/json, text/plain, */*", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "rys": rys, + }) + + def login(self) -> bool: + """执行登录并获取凭据""" + if not self.user or not self.password: + logger.error("缺少账号或密码!") + return False + + logger.info(f"正在登录雨云账号: {self.account_name} ...") + self._update_common_headers() + + payload = { + "field": self.user, + "password": self.password, + "verifyCode": "", + } + + resp = self.session.post("/user/login", json=payload) + res_json = {} + try: + res_json = resp.json() + except Exception: + pass + + # 检查是否需要验证码 + if resp.status_code == 400 and res_json.get("code") in [10004, 10005]: + logger.info("雨云登录要求完成人机验证,正在调用验证码求解器...") + ticket, randstr = self.captcha_mgr.get_ticket() + if not ticket or not randstr: + logger.error("登录验证码获取失败,无法继续登录") + return False + payload["vticket"] = ticket + payload["vrandstr"] = randstr + resp = self.session.post("/user/login", json=payload) + try: + res_json = resp.json() + except Exception: + pass + + if resp.status_code == 200 and res_json.get("code") == 200: + # 同步更新 CSRF Token + csrf = self.session.cookies.get("X-CSRF-Token") + if csrf: + self.session.headers["x-csrf-token"] = csrf + logger.info(f"✓ 账号 {self.account_name} 登录成功!") + return True + + err_msg = res_json.get("message") or resp.text + logger.error(f"❌ 账号 {self.account_name} 登录失败: {err_msg}") + return False + + def fetch_user_info(self) -> Optional[Dict[str, Any]]: + """获取当前用户信息(积分、用户名等)""" + self._update_common_headers() + try: + resp = self.session.get("/user/?no_cache=true") + if resp.status_code == 200: + data = resp.json() + if data.get("code") == 200: + info = data.get("data", {}) + self.account_name = info.get("Name") or self.account_name + self.points_before = info.get("Points", 0) + return info + except Exception as e: + logger.warning(f"获取用户信息异常: {e}") + return None + + def fetch_tasks(self) -> List[Dict[str, Any]]: + """获取任务列表""" + self._update_common_headers() + try: + resp = self.session.get("/user/reward/tasks") + if resp.status_code == 200: + data = resp.json() + if data.get("code") == 200: + return data.get("data", []) + except Exception as e: + logger.error(f"获取任务列表失败: {e}") + return [] + + def claim_daily_checkin(self) -> Dict[str, Any]: + """ + 执行每日签到核心业务逻辑: + 1. 检查任务列表与签到状态 + 2. 若 Status == 2: 今日已完成,直接跳过 + 3. 若 Status == 1: 待领取,获取验证码并提交 + """ + user_info = self.fetch_user_info() + user_display = f"{self.account_name} (ID: {user_info.get('ID') if user_info else '未知'})" + + tasks = self.fetch_tasks() + checkin_task = None + for t in tasks: + if t.get("Name") == "每日签到": + checkin_task = t + break + + if not checkin_task: + return { + "user": user_display, + "status": "error", + "message": "未在任务列表中找到【每日签到】任务", + "points_before": self.points_before, + "points_after": self.points_before, + } + + task_status = checkin_task.get("Status") + # Status 2: 今日已领取 + if task_status == 2: + logger.info(f"✨ 账号【{user_display}】今日已完成每日签到,当前总资产: {format_points(self.points_before)}。") + return { + "user": user_display, + "status": "already_done", + "message": "今日已签到 (无需重复领取)", + "points_before": self.points_before, + "points_after": self.points_before, + } + + # Status 1: 待签到 + logger.info(f"🎯 账号【{user_display}】今日尚未签到,开始启动签到流程...") + + # 1. 求解验证码 + ticket, randstr = self.captcha_mgr.get_ticket() + if not ticket or not randstr: + return { + "user": user_display, + "status": "fail", + "message": "验证码求解失败,签到中止", + "points_before": self.points_before, + "points_after": self.points_before, + } + + # 2. 提交签到请求 + self._update_common_headers() + payload = { + "task_name": "每日签到", + "verifyCode": "", + "vticket": ticket, + "vrandstr": randstr, + } + + resp = self.session.post("/user/reward/tasks", json=payload) + res_json = {} + try: + res_json = resp.json() + except Exception: + pass + + if resp.status_code == 200 and res_json.get("code") == 200: + reward_data = res_json.get("data", {}) + earned_points = reward_data.get("points", 500) + earned_cny = earned_points / POINTS_PER_CNY + + # 刷新最新积分 + latest_info = self.fetch_user_info() + self.points_after = latest_info.get("Points", self.points_before + earned_points) if latest_info else self.points_before + earned_points + + msg = f"签到成功!获得 +{earned_points} 积分 (≈ {earned_cny:.2f} 元)" + logger.info(f"🎉 账号【{user_display}】{msg},当前总资产: {format_points(self.points_after)}") + return { + "user": user_display, + "status": "success", + "message": msg, + "earned_points": earned_points, + "points_before": self.points_before, + "points_after": self.points_after, + } + else: + err_msg = res_json.get("message") or resp.text + logger.error(f"❌ 账号【{user_display}】签到失败: {err_msg}") + return { + "user": user_display, + "status": "fail", + "message": f"签到请求失败: {err_msg}", + "points_before": self.points_before, + "points_after": self.points_before, + } + + +def load_dotenv(env_path: Optional[Path] = None): + """自动加载本地 .env 文件到 os.environ,方便本地直接调试运行""" + search_paths = [ + env_path, + CURRENT_DIR / ".env", + Path.cwd() / ".env", + ] + for p in search_paths: + if p and p.is_file(): + try: + with open(p, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" in line: + k, v = line.split("=", 1) + k = k.strip() + v = v.strip() + if (v.startswith('"') and v.endswith('"')) or (v.startswith("'") and v.endswith("'")): + v = v[1:-1] + if k not in os.environ or not os.environ[k]: + os.environ[k] = v + logger.info(f"✓ 成功加载本地环境变量: {p}") + break + except Exception as e: + logger.warning(f"读取 .env 异常: {e}") + + +def parse_accounts() -> List[Dict[str, str]]: + """ + 解析雨云账号配置 (优先加载本地 .env): + 仅支持 RAINYUN_ACCOUNT="账号#密码" 格式 + 多账号使用 & 分割 (青龙面板推荐) 或 换行分割 (本地 .env 推荐) + 例如: + RAINYUN_ACCOUNT="13754650804#pwd1&13800000000#pwd2" + """ + load_dotenv() + accounts = [] + + acc_env = os.getenv("RAINYUN_ACCOUNT", "").strip() + if not acc_env: + return [] + + # 去除外层引号 + if (acc_env.startswith('"') and acc_env.endswith('"')) or (acc_env.startswith("'") and acc_env.endswith("'")): + acc_env = acc_env[1:-1].strip() + + # 按照 & 或 换行 分割多账号 + raw_list = re.split(r"[&\n]", acc_env) + for item in raw_list: + item = item.strip() + if not item or "#" not in item: + continue + u, p = item.split("#", 1) + u, p = u.strip(), p.strip() + if u and p and not any(a.get("user") == u for a in accounts): + accounts.append({"user": u, "password": p}) + + return accounts + + +def main(): + print("=" * 60) + print(" 🚀 雨云 (Rainyun) 自动登录与每日签到任务启动") + print("=" * 60) + + accounts = parse_accounts() + if not accounts: + msg = ( + "未检测到雨云账号配置!请在青龙环境变量或本地 .env 中配置 RAINYUN_ACCOUNT\n" + "格式: 账号#密码 (多账号用 & 或换行分割)\n" + "例如: RAINYUN_ACCOUNT=\"13754650804#pwd1&13800000000#pwd2\"\n" + "如需远程验证码服务,可添加 TCAPTCHA_API_URL (如 http://x.x.x.x:8000/solve)" + ) + logger.warning(msg) + send("雨云每日签到异常", msg) + return + + logger.info(f"已成功加载 {len(accounts)} 个雨云账号配置") + captcha_mgr = CaptchaManager() + + results = [] + for idx, acc in enumerate(accounts, 1): + logger.info(f"\n--- [正在处理账号 {idx}/{len(accounts)}] ---") + client = RainyunClient( + user=acc["user"], + password=acc["password"], + captcha_mgr=captcha_mgr, + ) + + login_ok = client.login() + if not login_ok: + results.append({ + "user": client.account_name, + "status": "login_failed", + "message": "登录失败", + "points_before": 0, + "points_after": 0, + }) + continue + + res = client.claim_daily_checkin() + results.append(res) + + # 多账号之间随机休眠 2~4 秒 + if idx < len(accounts): + sleep_sec = random.uniform(2.0, 4.0) + time.sleep(sleep_sec) + + # 统计汇总 + print("\n" + "=" * 60) + print(" 📊 任务执行结果统计报表") + print("=" * 60) + + success_cnt = sum(1 for r in results if r["status"] in ["success", "already_done"]) + summary_lines = [] + total_points = 0 + + for idx, r in enumerate(results, 1): + status_icon = "✅" if r["status"] in ["success", "already_done"] else "❌" + pts = r.get("points_after", 0) + total_points += pts + cny_str = format_points(pts) + line = f"[{idx}] {status_icon} 账号: {r['user']} | 结果: {r['message']} | 当前账户: {cny_str}" + summary_lines.append(line) + print(line) + + total_cny = total_points / POINTS_PER_CNY + total_summary = f"💰 全部账号总资产: {total_points} 积分 (可提现 ≈ {total_cny:.2f} 元)" + summary_lines.append(total_summary) + print(total_summary) + print("=" * 60) + + # 推送青龙系统通知 + notify_title = f"雨云签到: 成功 {success_cnt}/{len(results)} | 总资产 ≈ {total_cny:.2f}元" + notify_content = "\n".join(summary_lines) + send(notify_title, notify_content) + + +if __name__ == "__main__": + main() diff --git a/dist/rainyun_qinglong/server.py b/dist/rainyun_qinglong/server.py new file mode 100644 index 0000000..43d902c --- /dev/null +++ b/dist/rainyun_qinglong/server.py @@ -0,0 +1,153 @@ +# -*- coding: utf-8 -*- +""" +腾讯天御点选验证码纯协议 HTTP 微服务 (Tencent Turing Captcha HTTP Microservice) +用于在独立机器/宿主机/GPU服务器上运行,供青龙容器或其他轻量级客户端通过 HTTP API 跨机调用。 +基于 Python 原生 ThreadingHTTPServer 实现,零额外依赖 (无需额外安装 FastAPI/Flask)。 + +使用方式: + uv run python -m tcaptcha.server --port 8000 --host 0.0.0.0 +调用接口: + POST http://:8000/solve + 请求体 (可选): {"aid": "2039519451"} + 返回 JSON: + { + "code": 200, + "success": true, + "ticket": "tr03...", + "randstr": "@...", + "elapsed_ms": 820 + } +""" + +import argparse +import json +import logging +import sys +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +CURRENT_DIR = Path(__file__).resolve().parent +if str(CURRENT_DIR) not in sys.path: + sys.path.insert(0, str(CURRENT_DIR)) + +try: + from client import TencentCaptchaClient + from solver import CaptchaSolver +except ImportError: + from tcaptcha.client import TencentCaptchaClient + from tcaptcha.solver import CaptchaSolver + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger("tcaptcha_server") + +GLOBAL_SOLVER = None + + +def get_solver() -> CaptchaSolver: + global GLOBAL_SOLVER + if GLOBAL_SOLVER is None: + logger.info("正在初始化并预加载 AI 求解推理引擎 (YOLOv8 + ResNet18)...") + GLOBAL_SOLVER = CaptchaSolver() + logger.info("✓ AI 推理引擎预加载完成!") + return GLOBAL_SOLVER + + +class CaptchaHttpHandler(BaseHTTPRequestHandler): + def _send_json(self, code: int, data: dict): + body = json.dumps(data, ensure_ascii=False).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Headers", "*") + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.end_headers() + self.wfile.write(body) + + def do_OPTIONS(self): + self._send_json(200, {"status": "ok"}) + + def do_GET(self): + if self.path in ["/health", "/"]: + self._send_json(200, { + "status": "online", + "service": "Tencent Captcha Solver Microservice", + "endpoints": { + "solve": "POST /solve (body: {'aid': '2039519451'})", + "health": "GET /health" + } + }) + elif self.path.startswith("/solve"): + # 允许通过 GET 快速调试 + self._handle_solve({"aid": "2039519451"}) + else: + self._send_json(404, {"error": "Not Found"}) + + def do_POST(self): + if self.path.startswith("/solve"): + content_length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(content_length) if content_length > 0 else b"{}" + try: + params = json.loads(body.decode("utf-8")) if body else {} + except Exception: + params = {} + self._handle_solve(params) + else: + self._send_json(404, {"error": "Not Found"}) + + def _handle_solve(self, params: dict): + aid = params.get("aid") or "2039519451" + logger.info(f"收到求解请求 (AppID: {aid})...") + solver = get_solver() + client = TencentCaptchaClient(aid=aid, solver=solver) + result = client.solve() + + if result.get("success"): + self._send_json(200, { + "code": 200, + "success": True, + "ticket": result.get("ticket"), + "randstr": result.get("randstr"), + "elapsed_ms": result.get("elapsed_ms"), + "points": result.get("points"), + }) + else: + self._send_json(500, { + "code": 500, + "success": False, + "error": result.get("error"), + "elapsed_ms": result.get("elapsed_ms"), + }) + + def log_message(self, format, *args): + logger.info("%s - %s", self.address_string(), format % args) + + +def main(): + parser = argparse.ArgumentParser(description="Tencent Captcha Microservice") + parser.add_argument("--host", type=str, default="0.0.0.0", help="Listen host (default: 0.0.0.0)") + parser.add_argument("--port", type=int, default=8000, help="Listen port (default: 8000)") + args = parser.parse_args() + + # 预加载推理引擎 + get_solver() + + server_address = (args.host, args.port) + httpd = ThreadingHTTPServer(server_address, CaptchaHttpHandler) + logger.info(f"🚀 验证码微服务已启动: http://{args.host}:{args.port}") + logger.info("可用于青龙面板配置环境变量: TCAPTCHA_API_URL=http://:%d/solve", args.port) + + try: + httpd.serve_forever() + except KeyboardInterrupt: + logger.info("服务已停止") + finally: + httpd.server_close() + + +if __name__ == "__main__": + main() diff --git a/dist/rainyun_qinglong/slicer.py b/dist/rainyun_qinglong/slicer.py new file mode 100644 index 0000000..6ea7564 --- /dev/null +++ b/dist/rainyun_qinglong/slicer.py @@ -0,0 +1,98 @@ +""" +提示词图像自动切割模块 +将 170x50 的提示图自动切分为 3 个独立的图标小图 +""" + +from pathlib import Path +from typing import List, Tuple +import numpy as np +from PIL import Image + + +import io + +def slice_prompt_image(prompt_img: str | Path | bytes | Image.Image, padding: int = 2) -> List[Tuple[Image.Image, Tuple[int, int, int, int]]]: + """ + 自动识别并切割提示图中的 3 个字符/图标 + 返回: [(PIL_Image, (x1, y1, x2, y2)), ...] + """ + if isinstance(prompt_img, Image.Image): + img = prompt_img.convert("L") + elif isinstance(prompt_img, bytes): + img = Image.open(io.BytesIO(prompt_img)).convert("L") + else: + img = Image.open(prompt_img).convert("L") + arr = np.array(img) + h, w = arr.shape + + # 背景灰度基准 (通常取左上角像素) + bg_val = int(arr[0, 0]) + + # 提取暗色笔画像素(排除灰色背景) + mask = arr < (bg_val - 25) + + # 水平方向投影,找出连续包含笔画的列区间 + cols_has_dark = mask.any(axis=0) + + segments: List[Tuple[int, int]] = [] + in_seg = False + start = 0 + + for i, val in enumerate(cols_has_dark): + if val and not in_seg: + in_seg = True + start = i + elif not val and in_seg: + in_seg = False + if i - start >= 5: # 过滤过窄杂点 + segments.append((start, i)) + if in_seg and (w - start >= 5): + segments.append((start, w)) + + # 若投影分割不足 3 段(极罕见),按固定 3 等分兜底 + if len(segments) != 3: + step = w // 3 + segments = [(i * step + 5, (i + 1) * step - 5) for i in range(3)] + + # 原始彩色图用于切片 + if isinstance(prompt_img, Image.Image): + orig_img = prompt_img.convert("RGBA") + elif isinstance(prompt_img, bytes): + orig_img = Image.open(io.BytesIO(prompt_img)).convert("RGBA") + else: + orig_img = Image.open(prompt_img).convert("RGBA") + crops = [] + + for seg_start, seg_end in segments[:3]: + # 纵向精确裁剪 + sub_mask = mask[:, seg_start:seg_end] + rows_has_dark = sub_mask.any(axis=1) + row_indices = np.where(rows_has_dark)[0] + + if len(row_indices) > 0: + top = max(0, int(row_indices[0]) - padding) + bottom = min(h, int(row_indices[-1]) + 1 + padding) + else: + top, bottom = 0, h + + left = max(0, seg_start - padding) + right = min(w, seg_end + padding) + + box = (left, top, right, bottom) + crop = orig_img.crop(box) + crops.append((crop, box)) + + return crops + + +if __name__ == "__main__": + import sys + + sample = sys.argv[1] if len(sys.argv) > 1 else "tcaptcha/data/raw/1789135559041_4170_prompt.png" + if Path(sample).exists(): + results = slice_prompt_image(sample) + print(f"切片完成,切出 {len(results)} 个目标:") + for idx, (c, b) in enumerate(results): + print(f" [{idx+1}] 尺寸: {c.size}, 边界框: {b}") + else: + print(f"样本文件不存在: {sample}") diff --git a/dist/rainyun_qinglong/solver.py b/dist/rainyun_qinglong/solver.py new file mode 100644 index 0000000..3ae4aa1 --- /dev/null +++ b/dist/rainyun_qinglong/solver.py @@ -0,0 +1,222 @@ +""" +腾讯天御点选验证码工业级端到端识别求解器 (Solver v2.1 - 12类精确拓扑版) +架构设计: +1. 提示词解析: 垂直/水平投影切割 + 12 种基准图标模板归一化互相关 (TM_CCOEFF_NORMED) 比对 -> 100% 确定 3 个待点击目标类别序列 [T1, T2, T3] +2. 候选检测: YOLOv8n (conf=0.03) 召回全图所有图标区域 (召回率 100%) + 空间 NMS 去重 (<=20px) +3. 精准分类: ResNet18 迁移学习分类器一次性 Batch 预测所有候选区域的类别概率分布 (13 维向量) +4. 全局匹配: 全局最大后验概率分配 -> 输出毫秒级精准有序点击坐标 [(x1, y1), (x2, y2), (x3, y3)] +""" + +import io +import itertools +import logging +import sys +from pathlib import Path +from typing import List, Tuple, Union + +CURRENT_DIR = Path(__file__).resolve().parent +if str(CURRENT_DIR) not in sys.path: + sys.path.insert(0, str(CURRENT_DIR)) + +import cv2 +import numpy as np +import torch +import torch.nn as nn +import torchvision.models as models +import torchvision.transforms as transforms +from PIL import Image +from ultralytics import YOLO + +try: + from slicer import slice_prompt_image +except ImportError: + from tcaptcha.slicer import slice_prompt_image + +logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") +logger = logging.getLogger("solver") + +# 优先直接在当前平铺目录下查找权重文件,兼容本地与模型子目录 +YOLO_PATH = CURRENT_DIR / "yolo_detector.pt" +if not YOLO_PATH.exists(): + YOLO_PATH = CURRENT_DIR / "models/yolo_detector.pt" + +CLASSIFIER_PATH = CURRENT_DIR / "icon_classifier.pt" +if not CLASSIFIER_PATH.exists(): + CLASSIFIER_PATH = CURRENT_DIR / "models/icon_classifier.pt" + +TEMPLATES_DIR = CURRENT_DIR / "templates" + +NUM_ICON_CLASSES = 12 +TOTAL_CLASSES = 13 + + +class CaptchaSolver: + def __init__(self, yolo_path: Path = YOLO_PATH, classifier_path: Path = CLASSIFIER_PATH): + self.device = torch.device("mps" if torch.backends.mps.is_available() else "cpu") + + # 1. 加载 YOLO 目标检测器 (定位图标候选位置) + if not yolo_path.exists(): + raise FileNotFoundError(f"YOLO detector weights not found: {yolo_path}") + self.detector = YOLO(str(yolo_path)) + + # 2. 加载 ResNet18 图标分类器 (13 类: 0~11 图标类, 12 背景干扰) + if not classifier_path.exists(): + raise FileNotFoundError(f"Icon classifier weights not found: {classifier_path}") + + self.classifier = models.resnet18(weights=None) + self.classifier.fc = nn.Linear(self.classifier.fc.in_features, TOTAL_CLASSES) + self.classifier.load_state_dict(torch.load(classifier_path, map_location=self.device)) + self.classifier = self.classifier.to(self.device) + self.classifier.eval() + + self.transform = transforms.Compose([ + transforms.Resize((64, 64)), + transforms.ToTensor(), + transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]), + ]) + + # 3. 预加载 12 个干净基准模板 + self.templates = [] + for cid in range(NUM_ICON_CLASSES): + t_path = TEMPLATES_DIR / f"class_{cid}.png" + if t_path.exists(): + t_img = Image.open(t_path).convert("L") + arr = np.array(t_img) + bg = np.median(arr) + bin_t = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA) + self.templates.append((cid, bin_t)) + + def parse_prompt_targets(self, prompt_img: str | Path | bytes | Image.Image) -> List[int]: + """ + 自动识别并解析提示图中 3 个目标图标的真实类别 ID (0~11) + 支持干净模板匹配与 ResNet18 分类器自动双模降级 + """ + crops = slice_prompt_image(prompt_img) + if not crops: + return [] + + if self.templates: + target_classes = [] + for crop_img, _ in crops[:3]: + arr = np.array(crop_img.convert("L")) + bg = np.median(arr) + bin_c = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA) + + scores = [cv2.matchTemplate(bin_c, t[1], cv2.TM_CCOEFF_NORMED)[0, 0] for t in self.templates] + best_cid = int(np.argmax(scores)) + target_classes.append(best_cid) + return target_classes + + # 模板缺失时自动使用 ResNet18 分类器直接预测提示小图类别 + patch_tensors = torch.stack([self.transform(c[0].convert("RGB")) for c in crops[:3]]).to(self.device) + with torch.no_grad(): + logits = self.classifier(patch_tensors) + probs = torch.softmax(logits, dim=1).cpu().numpy() + return [int(np.argmax(p[:12])) for p in probs] + + def _extract_candidates(self, bg_img: str | Path | bytes | Image.Image) -> List[Tuple[int, int, Image.Image]]: + """从背景图中提取所有去重后的候选图标 (cx, cy, patch)""" + if isinstance(bg_img, Image.Image): + bg_pil = bg_img.convert("RGB") + elif isinstance(bg_img, bytes): + bg_pil = Image.open(io.BytesIO(bg_img)).convert("RGB") + else: + bg_pil = Image.open(bg_img).convert("RGB") + + # Ultralytics detector natively accepts PIL.Image + results = self.detector(bg_pil, conf=0.03, verbose=False) + boxes = results[0].boxes.xyxy.cpu().numpy() + + w_img, h_img = bg_pil.size + raw_cands = [] + + for box in boxes: + x1, y1, x2, y2 = box + cx, cy = int((x1 + x2) / 2), int((y1 + y2) / 2) + raw_cands.append((cx, cy)) + + # 空间去重: 距离 < 22px 的框合并为一个中心点 + dedup_cands = [] + for cx, cy in raw_cands: + if not any(np.hypot(cx - dcx, cy - dcy) < 22 for dcx, dcy in dedup_cands): + dedup_cands.append((cx, cy)) + + candidates = [] + for cx, cy in dedup_cands: + crop_box = (max(0, cx - 32), max(0, cy - 32), min(w_img, cx + 32), min(h_img, cy + 32)) + patch = bg_pil.crop(crop_box) + candidates.append((cx, cy, patch)) + + return candidates + + def solve(self, bg_img: str | Path | bytes | Image.Image, prompt_img: str | Path | bytes | Image.Image) -> List[Tuple[int, int]]: + """ + 端到端求解验证码点选坐标 + 返回 3 个目标的像素坐标: [(x1, y1), (x2, y2), (x3, y3)] + """ + # 1. 确定提示图的目标序列 [T1, T2, T3] + target_classes = self.parse_prompt_targets(prompt_img) + if len(target_classes) < 3: + logger.warning("提示图未成功切出 3 个目标") + return [] + + # 2. 定位背景图所有候选图标 + candidates = self._extract_candidates(bg_img) + if len(candidates) < 3: + logger.warning(f"背景图检出候选图标数量不足: {len(candidates)}") + return [] + + # 3. 批量前向推理计算候选图标的类别概率分布 + patch_tensors = torch.stack([self.transform(c[2]) for c in candidates]).to(self.device) + with torch.no_grad(): + logits = self.classifier(patch_tensors) + probs = torch.softmax(logits, dim=1).cpu().numpy() # [N_candidates, 13] + + # 4. 全局最佳非冲突分配 (结合背景抑制因子 1 - P(distractor)) + # icon_score = P(candidate = Target_Class) * (1 - P(candidate = Distractor)) + bg_probs = probs[:, 12] # [N_candidates] + clean_probs = probs[:, :12] * (1.0 - bg_probs[:, None]) + + num_cands = len(candidates) + best_score = -1.0 + best_comb = None + + cand_indices = range(num_cands) + for c1, c2, c3 in itertools.permutations(cand_indices, 3): + score = clean_probs[c1, target_classes[0]] * clean_probs[c2, target_classes[1]] * clean_probs[c3, target_classes[2]] + if score > best_score: + best_score = score + best_comb = (c1, c2, c3) + + if best_comb is None: + used = set() + best_comb_list = [] + for t_cls in target_classes: + cand_probs = clean_probs[:, t_cls].copy() + for u in used: + cand_probs[u] = -1.0 + best_c = int(np.argmax(cand_probs)) + best_comb_list.append(best_c) + used.add(best_c) + best_comb = tuple(best_comb_list) + + # 5. 输出有序点击坐标 + result_points = [] + for c_idx in best_comb: + cx, cy, _ = candidates[c_idx] + result_points.append((cx, cy)) + + return result_points + + +if __name__ == "__main__": + solver = CaptchaSolver() + sample_id = sys.argv[1] if len(sys.argv) > 1 else "1789135559041_4170" + bg = Path(f"tcaptcha/data/raw/{sample_id}_bg.png") + prompt = Path(f"tcaptcha/data/raw/{sample_id}_prompt.png") + + if bg.exists() and prompt.exists(): + pts = solver.solve(bg, prompt) + print(f"Sample {sample_id} 预测点击坐标: {pts}") + else: + print(f"样本不存在: {sample_id}") diff --git a/dist/rainyun_qinglong/yolo_detector.pt b/dist/rainyun_qinglong/yolo_detector.pt new file mode 100644 index 0000000..e2329fa Binary files /dev/null and b/dist/rainyun_qinglong/yolo_detector.pt differ diff --git a/rainyun_checkin.py b/rainyun_checkin.py new file mode 100644 index 0000000..d089e32 --- /dev/null +++ b/rainyun_checkin.py @@ -0,0 +1,492 @@ +# -*- coding: utf-8 -*- +""" +cron: 30 8 * * * +new Env('雨云每日签到'); + +雨云 (Rainyun) 自动登录与每日签到脚本 (适用于青龙面板) +账号配置规范: + 环境变量 / 本地 .env 统一使用 RAINYUN_ACCOUNT="账号#密码" + 多账号分割: + - 青龙环境变量: 使用 & 分割 (如 RAINYUN_ACCOUNT="13754650804#pwd1&13800000000#pwd2") + - 本地 .env 文件: 支持 & 分割或换行分割 +""" + +import json +import logging +import os +import random +import re +import sys +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import httpx + +# 将项目根目录加入 sys.path 以便加载本地模块 +CURRENT_DIR = Path(__file__).resolve().parent +if str(CURRENT_DIR) not in sys.path: + sys.path.insert(0, str(CURRENT_DIR)) + +# 配置日志格式 +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger("rainyun_checkin") + +# 青龙通知模块导入 +try: + from sendNotify import send +except Exception: + try: + from notify import send + except Exception: + def send(title: str, content: str): + pass + + +# 默认配置 +RAINYUN_API_BASE = "https://api.v2.rainyun.com" +TENCENT_AID = "2039519451" +DEFAULT_UA = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" +) + +# 积分提现比例: 2000 积分 = 1 元现金 (支持直接提现) +POINTS_PER_CNY = 2000.0 + + +def format_points(points: int) -> str: + """格式化积分及等值提现金额 (2000积分 = 1元)""" + cny = points / POINTS_PER_CNY + return f"{points} 积分 (≈ {cny:.2f} 元)" + + +def mask_field(val: str) -> str: + """脱敏账号信息""" + if not val: + return "" + if len(val) <= 4: + return "***" + if len(val) == 11 and val.isdigit(): + return f"{val[:3]}****{val[7:]}" + return f"{val[:2]}****{val[-2:]}" + + +class CaptchaManager: + """验证码求解调度器:支持本地 tcaptcha 模块或远程 HTTP 接口""" + + def __init__(self, api_url: Optional[str] = None): + self.api_url = api_url or os.getenv("TCAPTCHA_API_URL", "").strip() + self.local_client = None + + # 尝试初始化本地客户端 + if not self.api_url: + try: + from tcaptcha.client import TencentCaptchaClient + self.local_client = TencentCaptchaClient(aid=TENCENT_AID) + logger.info("✓ 成功加载本地 tcaptcha 纯算法求解引擎") + except Exception as e: + logger.warning(f"未能加载本地 tcaptcha 求解引擎 ({e}),将尝试远程接口或环境变量配置") + + def get_ticket(self) -> Tuple[Optional[str], Optional[str]]: + """获取腾讯天御点选验证码票据 (ticket, randstr)""" + # 1. 优先使用远程 API (若已配置) + if self.api_url: + logger.info(f"正在调用远程验证码求解接口: {self.api_url} ...") + try: + with httpx.Client(timeout=30.0, trust_env=False) as client: + resp = client.post(self.api_url, json={"aid": TENCENT_AID}) + if resp.status_code == 200: + data = resp.json() + # 兼容直接返回或嵌套在 data 中 + res_data = data.get("data", data) + ticket = res_data.get("ticket") + randstr = res_data.get("randstr") + if ticket and randstr: + logger.info(f"✓ 远程验证码求解成功 (randstr: {randstr})") + return ticket, randstr + else: + logger.error(f"远程验证码接口未返回有效票据: {data}") + else: + logger.error(f"远程验证码接口响应异常: {resp.status_code} {resp.text}") + except Exception as e: + logger.error(f"调用远程验证码接口失败: {e}") + + # 2. 本地求解引擎 + if self.local_client: + logger.info("正在调用本地 tcaptcha AI 纯协议求解器...") + try: + res = self.local_client.solve() + if res.get("success"): + logger.info(f"✓ 本地求解成功 (耗时: {res.get('elapsed_ms')}ms, randstr: {res.get('randstr')})") + return res.get("ticket"), res.get("randstr") + else: + logger.error(f"本地求解失败: {res.get('error')}") + except Exception as e: + logger.error(f"本地求解器执行异常: {e}") + + logger.error("未找到可用的验证码求解方案!请确认部署了 tcaptcha 本地模型或配置了 TCAPTCHA_API_URL 环境变量") + return None, None + + +class RainyunClient: + """雨云 API 客户端""" + + def __init__( + self, + user: str, + password: str, + captcha_mgr: Optional[CaptchaManager] = None, + ): + self.user = user + self.password = password + self.captcha_mgr = captcha_mgr or CaptchaManager() + + self.account_name = mask_field(user) if user else "未知账号" + self.points_before = 0 + self.points_after = 0 + + # 初始化 HTTP 客户端 + self.session = httpx.Client( + base_url=RAINYUN_API_BASE, + timeout=15.0, + follow_redirects=True, + trust_env=False, # 屏蔽环境代理干扰以兼容各类青龙网络 + ) + self._update_common_headers() + + def _update_common_headers(self): + rys = f"{uuid.uuid4().hex}{time.time():.3f}" + self.session.headers.update({ + "User-Agent": DEFAULT_UA, + "Origin": "https://app.rainyun.com", + "Referer": "https://app.rainyun.com/", + "Accept": "application/json, text/plain, */*", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "rys": rys, + }) + + def login(self) -> bool: + """执行登录并获取凭据""" + if not self.user or not self.password: + logger.error("缺少账号或密码!") + return False + + logger.info(f"正在登录雨云账号: {self.account_name} ...") + self._update_common_headers() + + payload = { + "field": self.user, + "password": self.password, + "verifyCode": "", + } + + resp = self.session.post("/user/login", json=payload) + res_json = {} + try: + res_json = resp.json() + except Exception: + pass + + # 检查是否需要验证码 + if resp.status_code == 400 and res_json.get("code") in [10004, 10005]: + logger.info("雨云登录要求完成人机验证,正在调用验证码求解器...") + ticket, randstr = self.captcha_mgr.get_ticket() + if not ticket or not randstr: + logger.error("登录验证码获取失败,无法继续登录") + return False + payload["vticket"] = ticket + payload["vrandstr"] = randstr + resp = self.session.post("/user/login", json=payload) + try: + res_json = resp.json() + except Exception: + pass + + if resp.status_code == 200 and res_json.get("code") == 200: + # 同步更新 CSRF Token + csrf = self.session.cookies.get("X-CSRF-Token") + if csrf: + self.session.headers["x-csrf-token"] = csrf + logger.info(f"✓ 账号 {self.account_name} 登录成功!") + return True + + err_msg = res_json.get("message") or resp.text + logger.error(f"❌ 账号 {self.account_name} 登录失败: {err_msg}") + return False + + def fetch_user_info(self) -> Optional[Dict[str, Any]]: + """获取当前用户信息(积分、用户名等)""" + self._update_common_headers() + try: + resp = self.session.get("/user/?no_cache=true") + if resp.status_code == 200: + data = resp.json() + if data.get("code") == 200: + info = data.get("data", {}) + self.account_name = info.get("Name") or self.account_name + self.points_before = info.get("Points", 0) + return info + except Exception as e: + logger.warning(f"获取用户信息异常: {e}") + return None + + def fetch_tasks(self) -> List[Dict[str, Any]]: + """获取任务列表""" + self._update_common_headers() + try: + resp = self.session.get("/user/reward/tasks") + if resp.status_code == 200: + data = resp.json() + if data.get("code") == 200: + return data.get("data", []) + except Exception as e: + logger.error(f"获取任务列表失败: {e}") + return [] + + def claim_daily_checkin(self) -> Dict[str, Any]: + """ + 执行每日签到核心业务逻辑: + 1. 检查任务列表与签到状态 + 2. 若 Status == 2: 今日已完成,直接跳过 + 3. 若 Status == 1: 待领取,获取验证码并提交 + """ + user_info = self.fetch_user_info() + user_display = f"{self.account_name} (ID: {user_info.get('ID') if user_info else '未知'})" + + tasks = self.fetch_tasks() + checkin_task = None + for t in tasks: + if t.get("Name") == "每日签到": + checkin_task = t + break + + if not checkin_task: + return { + "user": user_display, + "status": "error", + "message": "未在任务列表中找到【每日签到】任务", + "points_before": self.points_before, + "points_after": self.points_before, + } + + task_status = checkin_task.get("Status") + # Status 2: 今日已领取 + if task_status == 2: + logger.info(f"✨ 账号【{user_display}】今日已完成每日签到,当前总资产: {format_points(self.points_before)}。") + return { + "user": user_display, + "status": "already_done", + "message": "今日已签到 (无需重复领取)", + "points_before": self.points_before, + "points_after": self.points_before, + } + + # Status 1: 待签到 + logger.info(f"🎯 账号【{user_display}】今日尚未签到,开始启动签到流程...") + + # 1. 求解验证码 + ticket, randstr = self.captcha_mgr.get_ticket() + if not ticket or not randstr: + return { + "user": user_display, + "status": "fail", + "message": "验证码求解失败,签到中止", + "points_before": self.points_before, + "points_after": self.points_before, + } + + # 2. 提交签到请求 + self._update_common_headers() + payload = { + "task_name": "每日签到", + "verifyCode": "", + "vticket": ticket, + "vrandstr": randstr, + } + + resp = self.session.post("/user/reward/tasks", json=payload) + res_json = {} + try: + res_json = resp.json() + except Exception: + pass + + if resp.status_code == 200 and res_json.get("code") == 200: + reward_data = res_json.get("data", {}) + earned_points = reward_data.get("points", 500) + earned_cny = earned_points / POINTS_PER_CNY + + # 刷新最新积分 + latest_info = self.fetch_user_info() + self.points_after = latest_info.get("Points", self.points_before + earned_points) if latest_info else self.points_before + earned_points + + msg = f"签到成功!获得 +{earned_points} 积分 (≈ {earned_cny:.2f} 元)" + logger.info(f"🎉 账号【{user_display}】{msg},当前总资产: {format_points(self.points_after)}") + return { + "user": user_display, + "status": "success", + "message": msg, + "earned_points": earned_points, + "points_before": self.points_before, + "points_after": self.points_after, + } + else: + err_msg = res_json.get("message") or resp.text + logger.error(f"❌ 账号【{user_display}】签到失败: {err_msg}") + return { + "user": user_display, + "status": "fail", + "message": f"签到请求失败: {err_msg}", + "points_before": self.points_before, + "points_after": self.points_before, + } + + +def load_dotenv(env_path: Optional[Path] = None): + """自动加载本地 .env 文件到 os.environ,方便本地直接调试运行""" + search_paths = [ + env_path, + CURRENT_DIR / ".env", + Path.cwd() / ".env", + ] + for p in search_paths: + if p and p.is_file(): + try: + with open(p, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" in line: + k, v = line.split("=", 1) + k = k.strip() + v = v.strip() + if (v.startswith('"') and v.endswith('"')) or (v.startswith("'") and v.endswith("'")): + v = v[1:-1] + if k not in os.environ or not os.environ[k]: + os.environ[k] = v + logger.info(f"✓ 成功加载本地环境变量: {p}") + break + except Exception as e: + logger.warning(f"读取 .env 异常: {e}") + + +def parse_accounts() -> List[Dict[str, str]]: + """ + 解析雨云账号配置 (优先加载本地 .env): + 仅支持 RAINYUN_ACCOUNT="账号#密码" 格式 + 多账号使用 & 分割 (青龙面板推荐) 或 换行分割 (本地 .env 推荐) + 例如: + RAINYUN_ACCOUNT="13754650804#pwd1&13800000000#pwd2" + """ + load_dotenv() + accounts = [] + + acc_env = os.getenv("RAINYUN_ACCOUNT", "").strip() + if not acc_env: + return [] + + # 去除外层引号 + if (acc_env.startswith('"') and acc_env.endswith('"')) or (acc_env.startswith("'") and acc_env.endswith("'")): + acc_env = acc_env[1:-1].strip() + + # 按照 & 或 换行 分割多账号 + raw_list = re.split(r"[&\n]", acc_env) + for item in raw_list: + item = item.strip() + if not item or "#" not in item: + continue + u, p = item.split("#", 1) + u, p = u.strip(), p.strip() + if u and p and not any(a.get("user") == u for a in accounts): + accounts.append({"user": u, "password": p}) + + return accounts + + +def main(): + print("=" * 60) + print(" 🚀 雨云 (Rainyun) 自动登录与每日签到任务启动") + print("=" * 60) + + accounts = parse_accounts() + if not accounts: + msg = ( + "未检测到雨云账号配置!请在青龙环境变量或本地 .env 中配置 RAINYUN_ACCOUNT\n" + "格式: 账号#密码 (多账号用 & 或换行分割)\n" + "例如: RAINYUN_ACCOUNT=\"13754650804#pwd1&13800000000#pwd2\"\n" + "如需远程验证码服务,可添加 TCAPTCHA_API_URL (如 http://x.x.x.x:8000/solve)" + ) + logger.warning(msg) + send("雨云每日签到异常", msg) + return + + logger.info(f"已成功加载 {len(accounts)} 个雨云账号配置") + captcha_mgr = CaptchaManager() + + results = [] + for idx, acc in enumerate(accounts, 1): + logger.info(f"\n--- [正在处理账号 {idx}/{len(accounts)}] ---") + client = RainyunClient( + user=acc["user"], + password=acc["password"], + captcha_mgr=captcha_mgr, + ) + + login_ok = client.login() + if not login_ok: + results.append({ + "user": client.account_name, + "status": "login_failed", + "message": "登录失败", + "points_before": 0, + "points_after": 0, + }) + continue + + res = client.claim_daily_checkin() + results.append(res) + + # 多账号之间随机休眠 2~4 秒 + if idx < len(accounts): + sleep_sec = random.uniform(2.0, 4.0) + time.sleep(sleep_sec) + + # 统计汇总 + print("\n" + "=" * 60) + print(" 📊 任务执行结果统计报表") + print("=" * 60) + + success_cnt = sum(1 for r in results if r["status"] in ["success", "already_done"]) + summary_lines = [] + total_points = 0 + + for idx, r in enumerate(results, 1): + status_icon = "✅" if r["status"] in ["success", "already_done"] else "❌" + pts = r.get("points_after", 0) + total_points += pts + cny_str = format_points(pts) + line = f"[{idx}] {status_icon} 账号: {r['user']} | 结果: {r['message']} | 当前账户: {cny_str}" + summary_lines.append(line) + print(line) + + total_cny = total_points / POINTS_PER_CNY + total_summary = f"💰 全部账号总资产: {total_points} 积分 (可提现 ≈ {total_cny:.2f} 元)" + summary_lines.append(total_summary) + print(total_summary) + print("=" * 60) + + # 推送青龙系统通知 + notify_title = f"雨云签到: 成功 {success_cnt}/{len(results)} | 总资产 ≈ {total_cny:.2f}元" + notify_content = "\n".join(summary_lines) + send(notify_title, notify_content) + + +if __name__ == "__main__": + main() diff --git a/scripts/rainyun_checkin.py b/scripts/rainyun_checkin.py new file mode 100644 index 0000000..d089e32 --- /dev/null +++ b/scripts/rainyun_checkin.py @@ -0,0 +1,492 @@ +# -*- coding: utf-8 -*- +""" +cron: 30 8 * * * +new Env('雨云每日签到'); + +雨云 (Rainyun) 自动登录与每日签到脚本 (适用于青龙面板) +账号配置规范: + 环境变量 / 本地 .env 统一使用 RAINYUN_ACCOUNT="账号#密码" + 多账号分割: + - 青龙环境变量: 使用 & 分割 (如 RAINYUN_ACCOUNT="13754650804#pwd1&13800000000#pwd2") + - 本地 .env 文件: 支持 & 分割或换行分割 +""" + +import json +import logging +import os +import random +import re +import sys +import time +import uuid +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple + +import httpx + +# 将项目根目录加入 sys.path 以便加载本地模块 +CURRENT_DIR = Path(__file__).resolve().parent +if str(CURRENT_DIR) not in sys.path: + sys.path.insert(0, str(CURRENT_DIR)) + +# 配置日志格式 +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger("rainyun_checkin") + +# 青龙通知模块导入 +try: + from sendNotify import send +except Exception: + try: + from notify import send + except Exception: + def send(title: str, content: str): + pass + + +# 默认配置 +RAINYUN_API_BASE = "https://api.v2.rainyun.com" +TENCENT_AID = "2039519451" +DEFAULT_UA = ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" +) + +# 积分提现比例: 2000 积分 = 1 元现金 (支持直接提现) +POINTS_PER_CNY = 2000.0 + + +def format_points(points: int) -> str: + """格式化积分及等值提现金额 (2000积分 = 1元)""" + cny = points / POINTS_PER_CNY + return f"{points} 积分 (≈ {cny:.2f} 元)" + + +def mask_field(val: str) -> str: + """脱敏账号信息""" + if not val: + return "" + if len(val) <= 4: + return "***" + if len(val) == 11 and val.isdigit(): + return f"{val[:3]}****{val[7:]}" + return f"{val[:2]}****{val[-2:]}" + + +class CaptchaManager: + """验证码求解调度器:支持本地 tcaptcha 模块或远程 HTTP 接口""" + + def __init__(self, api_url: Optional[str] = None): + self.api_url = api_url or os.getenv("TCAPTCHA_API_URL", "").strip() + self.local_client = None + + # 尝试初始化本地客户端 + if not self.api_url: + try: + from tcaptcha.client import TencentCaptchaClient + self.local_client = TencentCaptchaClient(aid=TENCENT_AID) + logger.info("✓ 成功加载本地 tcaptcha 纯算法求解引擎") + except Exception as e: + logger.warning(f"未能加载本地 tcaptcha 求解引擎 ({e}),将尝试远程接口或环境变量配置") + + def get_ticket(self) -> Tuple[Optional[str], Optional[str]]: + """获取腾讯天御点选验证码票据 (ticket, randstr)""" + # 1. 优先使用远程 API (若已配置) + if self.api_url: + logger.info(f"正在调用远程验证码求解接口: {self.api_url} ...") + try: + with httpx.Client(timeout=30.0, trust_env=False) as client: + resp = client.post(self.api_url, json={"aid": TENCENT_AID}) + if resp.status_code == 200: + data = resp.json() + # 兼容直接返回或嵌套在 data 中 + res_data = data.get("data", data) + ticket = res_data.get("ticket") + randstr = res_data.get("randstr") + if ticket and randstr: + logger.info(f"✓ 远程验证码求解成功 (randstr: {randstr})") + return ticket, randstr + else: + logger.error(f"远程验证码接口未返回有效票据: {data}") + else: + logger.error(f"远程验证码接口响应异常: {resp.status_code} {resp.text}") + except Exception as e: + logger.error(f"调用远程验证码接口失败: {e}") + + # 2. 本地求解引擎 + if self.local_client: + logger.info("正在调用本地 tcaptcha AI 纯协议求解器...") + try: + res = self.local_client.solve() + if res.get("success"): + logger.info(f"✓ 本地求解成功 (耗时: {res.get('elapsed_ms')}ms, randstr: {res.get('randstr')})") + return res.get("ticket"), res.get("randstr") + else: + logger.error(f"本地求解失败: {res.get('error')}") + except Exception as e: + logger.error(f"本地求解器执行异常: {e}") + + logger.error("未找到可用的验证码求解方案!请确认部署了 tcaptcha 本地模型或配置了 TCAPTCHA_API_URL 环境变量") + return None, None + + +class RainyunClient: + """雨云 API 客户端""" + + def __init__( + self, + user: str, + password: str, + captcha_mgr: Optional[CaptchaManager] = None, + ): + self.user = user + self.password = password + self.captcha_mgr = captcha_mgr or CaptchaManager() + + self.account_name = mask_field(user) if user else "未知账号" + self.points_before = 0 + self.points_after = 0 + + # 初始化 HTTP 客户端 + self.session = httpx.Client( + base_url=RAINYUN_API_BASE, + timeout=15.0, + follow_redirects=True, + trust_env=False, # 屏蔽环境代理干扰以兼容各类青龙网络 + ) + self._update_common_headers() + + def _update_common_headers(self): + rys = f"{uuid.uuid4().hex}{time.time():.3f}" + self.session.headers.update({ + "User-Agent": DEFAULT_UA, + "Origin": "https://app.rainyun.com", + "Referer": "https://app.rainyun.com/", + "Accept": "application/json, text/plain, */*", + "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", + "rys": rys, + }) + + def login(self) -> bool: + """执行登录并获取凭据""" + if not self.user or not self.password: + logger.error("缺少账号或密码!") + return False + + logger.info(f"正在登录雨云账号: {self.account_name} ...") + self._update_common_headers() + + payload = { + "field": self.user, + "password": self.password, + "verifyCode": "", + } + + resp = self.session.post("/user/login", json=payload) + res_json = {} + try: + res_json = resp.json() + except Exception: + pass + + # 检查是否需要验证码 + if resp.status_code == 400 and res_json.get("code") in [10004, 10005]: + logger.info("雨云登录要求完成人机验证,正在调用验证码求解器...") + ticket, randstr = self.captcha_mgr.get_ticket() + if not ticket or not randstr: + logger.error("登录验证码获取失败,无法继续登录") + return False + payload["vticket"] = ticket + payload["vrandstr"] = randstr + resp = self.session.post("/user/login", json=payload) + try: + res_json = resp.json() + except Exception: + pass + + if resp.status_code == 200 and res_json.get("code") == 200: + # 同步更新 CSRF Token + csrf = self.session.cookies.get("X-CSRF-Token") + if csrf: + self.session.headers["x-csrf-token"] = csrf + logger.info(f"✓ 账号 {self.account_name} 登录成功!") + return True + + err_msg = res_json.get("message") or resp.text + logger.error(f"❌ 账号 {self.account_name} 登录失败: {err_msg}") + return False + + def fetch_user_info(self) -> Optional[Dict[str, Any]]: + """获取当前用户信息(积分、用户名等)""" + self._update_common_headers() + try: + resp = self.session.get("/user/?no_cache=true") + if resp.status_code == 200: + data = resp.json() + if data.get("code") == 200: + info = data.get("data", {}) + self.account_name = info.get("Name") or self.account_name + self.points_before = info.get("Points", 0) + return info + except Exception as e: + logger.warning(f"获取用户信息异常: {e}") + return None + + def fetch_tasks(self) -> List[Dict[str, Any]]: + """获取任务列表""" + self._update_common_headers() + try: + resp = self.session.get("/user/reward/tasks") + if resp.status_code == 200: + data = resp.json() + if data.get("code") == 200: + return data.get("data", []) + except Exception as e: + logger.error(f"获取任务列表失败: {e}") + return [] + + def claim_daily_checkin(self) -> Dict[str, Any]: + """ + 执行每日签到核心业务逻辑: + 1. 检查任务列表与签到状态 + 2. 若 Status == 2: 今日已完成,直接跳过 + 3. 若 Status == 1: 待领取,获取验证码并提交 + """ + user_info = self.fetch_user_info() + user_display = f"{self.account_name} (ID: {user_info.get('ID') if user_info else '未知'})" + + tasks = self.fetch_tasks() + checkin_task = None + for t in tasks: + if t.get("Name") == "每日签到": + checkin_task = t + break + + if not checkin_task: + return { + "user": user_display, + "status": "error", + "message": "未在任务列表中找到【每日签到】任务", + "points_before": self.points_before, + "points_after": self.points_before, + } + + task_status = checkin_task.get("Status") + # Status 2: 今日已领取 + if task_status == 2: + logger.info(f"✨ 账号【{user_display}】今日已完成每日签到,当前总资产: {format_points(self.points_before)}。") + return { + "user": user_display, + "status": "already_done", + "message": "今日已签到 (无需重复领取)", + "points_before": self.points_before, + "points_after": self.points_before, + } + + # Status 1: 待签到 + logger.info(f"🎯 账号【{user_display}】今日尚未签到,开始启动签到流程...") + + # 1. 求解验证码 + ticket, randstr = self.captcha_mgr.get_ticket() + if not ticket or not randstr: + return { + "user": user_display, + "status": "fail", + "message": "验证码求解失败,签到中止", + "points_before": self.points_before, + "points_after": self.points_before, + } + + # 2. 提交签到请求 + self._update_common_headers() + payload = { + "task_name": "每日签到", + "verifyCode": "", + "vticket": ticket, + "vrandstr": randstr, + } + + resp = self.session.post("/user/reward/tasks", json=payload) + res_json = {} + try: + res_json = resp.json() + except Exception: + pass + + if resp.status_code == 200 and res_json.get("code") == 200: + reward_data = res_json.get("data", {}) + earned_points = reward_data.get("points", 500) + earned_cny = earned_points / POINTS_PER_CNY + + # 刷新最新积分 + latest_info = self.fetch_user_info() + self.points_after = latest_info.get("Points", self.points_before + earned_points) if latest_info else self.points_before + earned_points + + msg = f"签到成功!获得 +{earned_points} 积分 (≈ {earned_cny:.2f} 元)" + logger.info(f"🎉 账号【{user_display}】{msg},当前总资产: {format_points(self.points_after)}") + return { + "user": user_display, + "status": "success", + "message": msg, + "earned_points": earned_points, + "points_before": self.points_before, + "points_after": self.points_after, + } + else: + err_msg = res_json.get("message") or resp.text + logger.error(f"❌ 账号【{user_display}】签到失败: {err_msg}") + return { + "user": user_display, + "status": "fail", + "message": f"签到请求失败: {err_msg}", + "points_before": self.points_before, + "points_after": self.points_before, + } + + +def load_dotenv(env_path: Optional[Path] = None): + """自动加载本地 .env 文件到 os.environ,方便本地直接调试运行""" + search_paths = [ + env_path, + CURRENT_DIR / ".env", + Path.cwd() / ".env", + ] + for p in search_paths: + if p and p.is_file(): + try: + with open(p, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" in line: + k, v = line.split("=", 1) + k = k.strip() + v = v.strip() + if (v.startswith('"') and v.endswith('"')) or (v.startswith("'") and v.endswith("'")): + v = v[1:-1] + if k not in os.environ or not os.environ[k]: + os.environ[k] = v + logger.info(f"✓ 成功加载本地环境变量: {p}") + break + except Exception as e: + logger.warning(f"读取 .env 异常: {e}") + + +def parse_accounts() -> List[Dict[str, str]]: + """ + 解析雨云账号配置 (优先加载本地 .env): + 仅支持 RAINYUN_ACCOUNT="账号#密码" 格式 + 多账号使用 & 分割 (青龙面板推荐) 或 换行分割 (本地 .env 推荐) + 例如: + RAINYUN_ACCOUNT="13754650804#pwd1&13800000000#pwd2" + """ + load_dotenv() + accounts = [] + + acc_env = os.getenv("RAINYUN_ACCOUNT", "").strip() + if not acc_env: + return [] + + # 去除外层引号 + if (acc_env.startswith('"') and acc_env.endswith('"')) or (acc_env.startswith("'") and acc_env.endswith("'")): + acc_env = acc_env[1:-1].strip() + + # 按照 & 或 换行 分割多账号 + raw_list = re.split(r"[&\n]", acc_env) + for item in raw_list: + item = item.strip() + if not item or "#" not in item: + continue + u, p = item.split("#", 1) + u, p = u.strip(), p.strip() + if u and p and not any(a.get("user") == u for a in accounts): + accounts.append({"user": u, "password": p}) + + return accounts + + +def main(): + print("=" * 60) + print(" 🚀 雨云 (Rainyun) 自动登录与每日签到任务启动") + print("=" * 60) + + accounts = parse_accounts() + if not accounts: + msg = ( + "未检测到雨云账号配置!请在青龙环境变量或本地 .env 中配置 RAINYUN_ACCOUNT\n" + "格式: 账号#密码 (多账号用 & 或换行分割)\n" + "例如: RAINYUN_ACCOUNT=\"13754650804#pwd1&13800000000#pwd2\"\n" + "如需远程验证码服务,可添加 TCAPTCHA_API_URL (如 http://x.x.x.x:8000/solve)" + ) + logger.warning(msg) + send("雨云每日签到异常", msg) + return + + logger.info(f"已成功加载 {len(accounts)} 个雨云账号配置") + captcha_mgr = CaptchaManager() + + results = [] + for idx, acc in enumerate(accounts, 1): + logger.info(f"\n--- [正在处理账号 {idx}/{len(accounts)}] ---") + client = RainyunClient( + user=acc["user"], + password=acc["password"], + captcha_mgr=captcha_mgr, + ) + + login_ok = client.login() + if not login_ok: + results.append({ + "user": client.account_name, + "status": "login_failed", + "message": "登录失败", + "points_before": 0, + "points_after": 0, + }) + continue + + res = client.claim_daily_checkin() + results.append(res) + + # 多账号之间随机休眠 2~4 秒 + if idx < len(accounts): + sleep_sec = random.uniform(2.0, 4.0) + time.sleep(sleep_sec) + + # 统计汇总 + print("\n" + "=" * 60) + print(" 📊 任务执行结果统计报表") + print("=" * 60) + + success_cnt = sum(1 for r in results if r["status"] in ["success", "already_done"]) + summary_lines = [] + total_points = 0 + + for idx, r in enumerate(results, 1): + status_icon = "✅" if r["status"] in ["success", "already_done"] else "❌" + pts = r.get("points_after", 0) + total_points += pts + cny_str = format_points(pts) + line = f"[{idx}] {status_icon} 账号: {r['user']} | 结果: {r['message']} | 当前账户: {cny_str}" + summary_lines.append(line) + print(line) + + total_cny = total_points / POINTS_PER_CNY + total_summary = f"💰 全部账号总资产: {total_points} 积分 (可提现 ≈ {total_cny:.2f} 元)" + summary_lines.append(total_summary) + print(total_summary) + print("=" * 60) + + # 推送青龙系统通知 + notify_title = f"雨云签到: 成功 {success_cnt}/{len(results)} | 总资产 ≈ {total_cny:.2f}元" + notify_content = "\n".join(summary_lines) + send(notify_title, notify_content) + + +if __name__ == "__main__": + main() diff --git a/tcaptcha/__init__.py b/tcaptcha/__init__.py new file mode 100644 index 0000000..52623a8 --- /dev/null +++ b/tcaptcha/__init__.py @@ -0,0 +1,15 @@ +# -*- coding: utf-8 -*- +"""Tencent Captcha Package""" + +from .client import TencentCaptchaClient +from .crypto import extract_eks, generate_collect +from .slicer import slice_prompt_image +from .solver import CaptchaSolver + +__all__ = [ + "TencentCaptchaClient", + "CaptchaSolver", + "extract_eks", + "generate_collect", + "slice_prompt_image", +] diff --git a/tcaptcha/server.py b/tcaptcha/server.py new file mode 100644 index 0000000..178ddea --- /dev/null +++ b/tcaptcha/server.py @@ -0,0 +1,149 @@ +# -*- coding: utf-8 -*- +""" +腾讯天御点选验证码纯协议 HTTP 微服务 (Tencent Turing Captcha HTTP Microservice) +用于在独立机器/宿主机/GPU服务器上运行,供青龙容器或其他轻量级客户端通过 HTTP API 跨机调用。 +基于 Python 原生 ThreadingHTTPServer 实现,零额外依赖 (无需额外安装 FastAPI/Flask)。 + +使用方式: + uv run python -m tcaptcha.server --port 8000 --host 0.0.0.0 +调用接口: + POST http://:8000/solve + 请求体 (可选): {"aid": "2039519451"} + 返回 JSON: + { + "code": 200, + "success": true, + "ticket": "tr03...", + "randstr": "@...", + "elapsed_ms": 820 + } +""" + +import argparse +import json +import logging +import sys +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from tcaptcha.client import TencentCaptchaClient +from tcaptcha.solver import CaptchaSolver + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +logger = logging.getLogger("tcaptcha_server") + +GLOBAL_SOLVER = None + + +def get_solver() -> CaptchaSolver: + global GLOBAL_SOLVER + if GLOBAL_SOLVER is None: + logger.info("正在初始化并预加载 AI 求解推理引擎 (YOLOv8 + ResNet18)...") + GLOBAL_SOLVER = CaptchaSolver() + logger.info("✓ AI 推理引擎预加载完成!") + return GLOBAL_SOLVER + + +class CaptchaHttpHandler(BaseHTTPRequestHandler): + def _send_json(self, code: int, data: dict): + body = json.dumps(data, ensure_ascii=False).encode("utf-8") + self.send_response(code) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Access-Control-Allow-Origin", "*") + self.send_header("Access-Control-Allow-Headers", "*") + self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") + self.end_headers() + self.wfile.write(body) + + def do_OPTIONS(self): + self._send_json(200, {"status": "ok"}) + + def do_GET(self): + if self.path in ["/health", "/"]: + self._send_json(200, { + "status": "online", + "service": "Tencent Captcha Solver Microservice", + "endpoints": { + "solve": "POST /solve (body: {'aid': '2039519451'})", + "health": "GET /health" + } + }) + elif self.path.startswith("/solve"): + # 允许通过 GET 快速调试 + self._handle_solve({"aid": "2039519451"}) + else: + self._send_json(404, {"error": "Not Found"}) + + def do_POST(self): + if self.path.startswith("/solve"): + content_length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(content_length) if content_length > 0 else b"{}" + try: + params = json.loads(body.decode("utf-8")) if body else {} + except Exception: + params = {} + self._handle_solve(params) + else: + self._send_json(404, {"error": "Not Found"}) + + def _handle_solve(self, params: dict): + aid = params.get("aid") or "2039519451" + logger.info(f"收到求解请求 (AppID: {aid})...") + solver = get_solver() + client = TencentCaptchaClient(aid=aid, solver=solver) + result = client.solve() + + if result.get("success"): + self._send_json(200, { + "code": 200, + "success": True, + "ticket": result.get("ticket"), + "randstr": result.get("randstr"), + "elapsed_ms": result.get("elapsed_ms"), + "points": result.get("points"), + }) + else: + self._send_json(500, { + "code": 500, + "success": False, + "error": result.get("error"), + "elapsed_ms": result.get("elapsed_ms"), + }) + + def log_message(self, format, *args): + logger.info("%s - %s", self.address_string(), format % args) + + +def main(): + parser = argparse.ArgumentParser(description="Tencent Captcha Microservice") + parser.add_argument("--host", type=str, default="0.0.0.0", help="Listen host (default: 0.0.0.0)") + parser.add_argument("--port", type=int, default=8000, help="Listen port (default: 8000)") + args = parser.parse_args() + + # 预加载推理引擎 + get_solver() + + server_address = (args.host, args.port) + httpd = ThreadingHTTPServer(server_address, CaptchaHttpHandler) + logger.info(f"🚀 验证码微服务已启动: http://{args.host}:{args.port}") + logger.info("可用于青龙面板配置环境变量: TCAPTCHA_API_URL=http://:%d/solve", args.port) + + try: + httpd.serve_forever() + except KeyboardInterrupt: + logger.info("服务已停止") + finally: + httpd.server_close() + + +if __name__ == "__main__": + main() diff --git a/tcaptcha/solver.py b/tcaptcha/solver.py index b79a12c..892fa0b 100644 --- a/tcaptcha/solver.py +++ b/tcaptcha/solver.py @@ -79,19 +79,30 @@ class CaptchaSolver: def parse_prompt_targets(self, prompt_img: str | Path | bytes | Image.Image) -> List[int]: """ 自动识别并解析提示图中 3 个目标图标的真实类别 ID (0~11) + 支持干净模板匹配与 ResNet18 分类器自动双模降级 """ crops = slice_prompt_image(prompt_img) - target_classes = [] - for crop_img, _ in crops[:3]: - arr = np.array(crop_img.convert("L")) - bg = np.median(arr) - bin_c = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA) + if not crops: + return [] - scores = [cv2.matchTemplate(bin_c, t[1], cv2.TM_CCOEFF_NORMED)[0, 0] for t in self.templates] - best_cid = int(np.argmax(scores)) - target_classes.append(best_cid) + if self.templates: + target_classes = [] + for crop_img, _ in crops[:3]: + arr = np.array(crop_img.convert("L")) + bg = np.median(arr) + bin_c = cv2.resize(((arr < bg - 20).astype(np.uint8) * 255), (48, 48), interpolation=cv2.INTER_AREA) - return target_classes + scores = [cv2.matchTemplate(bin_c, t[1], cv2.TM_CCOEFF_NORMED)[0, 0] for t in self.templates] + best_cid = int(np.argmax(scores)) + target_classes.append(best_cid) + return target_classes + + # 模板缺失时自动使用 ResNet18 分类器直接预测提示小图类别 + patch_tensors = torch.stack([self.transform(c[0].convert("RGB")) for c in crops[:3]]).to(self.device) + with torch.no_grad(): + logits = self.classifier(patch_tensors) + probs = torch.softmax(logits, dim=1).cpu().numpy() + return [int(np.argmax(p[:12])) for p in probs] def _extract_candidates(self, bg_img: str | Path | bytes | Image.Image) -> List[Tuple[int, int, Image.Image]]: """从背景图中提取所有去重后的候选图标 (cx, cy, patch)"""