yml/hello_js_reverse_skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(rainyun): 完成雨云纯协议自动登录签到脚本及青龙单目录部署包

Browse Source 
- 逆向分析雨云登录与签到接口，打通腾讯天御点选验证码全协议过码闭环
- 支持 RAINYUN_ACCOUNT='账号#密码' 极简配置 (多账号支持 & 与换行分割)
- 增加积分提现金额换算展示 (2000积分 = 1元现金) 及青龙系统通知推送
- 新增 dist/rainyun_qinglong 单层平铺部署目录与专属 README
- 增强 solver.py 在模板缺失时的 ResNet18 分类器双模降级能力
This commit is contained in:
yml2213
2026-09-12 10:56:25 +08:00
parent da62cb7e09
commit 5835d7cb15
15 changed files with 2837 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dist/rainyun_qinglong/README.md
Vendored
+53
View File
@@ -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 自动执行）
dist/rainyun_qinglong/client.py
Vendored
+323
View File
@@ -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()
dist/rainyun_qinglong/crypto.py
Vendored
+312
View File
@@ -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("<II", data[i : i + 8])
ev0, ev1 = xtea_encrypt_block(v0, v1, key)
out.extend(struct.pack("<II", ev0, ev1))
return base64.b64encode(out).decode("ascii")
def _bezier_interp(
p0: Tuple[float, float],
p1: Tuple[float, float],
steps: int,
) -> 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]}...")
dist/rainyun_qinglong/icon_classifier.pt
Vendored
BIN
View File
Binary file not shown.
dist/rainyun_qinglong/rainyun_checkin.py
Vendored
+497
View File
@@ -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()
dist/rainyun_qinglong/server.py
Vendored
+153
View File
@@ -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://<IP>: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://<IP>:%d/solve", args.port)
try:
httpd.serve_forever()
except KeyboardInterrupt:
logger.info("服务已停止")
finally:
httpd.server_close()
if __name__ == "__main__":
main()
dist/rainyun_qinglong/slicer.py
Vendored
+98
View File
@@ -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}")
dist/rainyun_qinglong/solver.py
Vendored
+222
View File
@@ -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}")
dist/rainyun_qinglong/yolo_detector.pt
Vendored
BIN
View File
Binary file not shown.