tg_factory/geetest_crack.py

import ddddocr


from PIL import Image
import io
import numpy as np
import time
import json
import base64
import random
import math
import requests
import urllib.parse

class TDCSimulator:
    def __init__(self):
        self.start_time = int(time.time() * 1000)
        self.data = {
            "tdf": 1,
            "ts": [],
            "points": [],
            "events": [],
            "ua": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "tm": {
                "st": self.start_time,
                "ed": 0
            }
        }
    
    def add_point(self, x, y, t):
        self.data["points"].append([x, y])
        self.data["ts"].append(t - self.start_time)
    
    def add_event(self, type_, x, y):
        t = int(time.time() * 1000) - self.start_time
        self.data["events"].append({
            "type": type_,
            "t": t,
            "x": x,
            "y": y
        })
    
    def get_data(self):
        self.data["tm"]["ed"] = int(time.time() * 1000)
        return base64.b64encode(json.dumps(self.data).encode()).decode()

def generate_tracks(start_pos, target_pos):
    tracks = []
    distance = target_pos[0] - start_pos[0]
    current = 0
    t = int(time.time() * 1000)
    v = 0  # 初始速度
    
    x = start_pos[0]
    y = start_pos[1]
    
    while current < distance:
        delta_t = 10  # 10ms
        t += delta_t
        
        # 加速阶段
        if current < distance * 0.6:
            v += (random.random() * 2 + 2)
        # 减速阶段
        else:
            v *= 0.98
        
        # 位移
        delta_x = v * (delta_t / 1000)
        current += delta_x
        x += delta_x
        
        # y轴抖动
        y = start_pos[1] + (random.random() * 2 - 1)
        
        tracks.append({
            "x": round(x),
            "y": round(y),
            "t": t
        })
    
    # 确保最后到达目标位置
    tracks.append({
        "x": target_pos[0],
        "y": target_pos[1],
        "t": t + 10
    })
    
    return tracks

def get_collect(start_pos, target_pos):
    tdc = TDCSimulator()
    
    # 开始事件
    tdc.add_event("start", start_pos[0], start_pos[1])
    
    # 按下事件
    tdc.add_event("down", start_pos[0], start_pos[1])
    
    # 移动轨迹
    tracks = generate_tracks(start_pos, target_pos)
    for track in tracks:
        tdc.add_point(track["x"], track["y"], track["t"])
    
    # 松开事件
    tdc.add_event("up", target_pos[0], target_pos[1])
    
    # 结束事件
    tdc.add_event("end", target_pos[0], target_pos[1])
    
    return tdc.get_data()

def get_gap(target_img:bytes, background_img:bytes):
    det = ddddocr.DdddOcr(det=False, ocr=False, show_ad=False)
    res = det.slide_match(target_img, background_img,simple_target=True)
    print(res)
    return res["target"][0]
def parse_positions(data):
    """
    解析验证码位置信息
    
    参数：
    data: _aq_88765 返回的数据
    
    返回：
    {
        "start_pos": [x, y],
        "move_limit": [min_x, max_x],
        "sess": session_id,
        "image_urls": {
            "bg": background_image_url,
            "sprite": sprite_image_url
        }
    }
    """
    try:
        # 获取滑块信息
        fg_elem_list = data["data"]["dyn_show_info"]["fg_elem_list"]
        slider_info = None
        for elem in fg_elem_list:
            if elem["id"] == 1:  # id为1的元素是滑块
                slider_info = elem
                break
                
        if not slider_info:
            raise ValueError("未找到滑块信息")
            
        # 解析移动限制
        track_limit = slider_info["move_cfg"]["track_limit"]  # "x>=50&&x<=552"
        limits = track_limit.replace("x>=", "").replace("&&x<=", ",").split(",")
        min_x = int(limits[0])
        max_x = int(limits[1])
        
        # 获取图片URL
        bg_img_url = data["data"]["dyn_show_info"]["bg_elem_cfg"]["img_url"]
        sprite_url = data["data"]["dyn_show_info"]["sprite_url"]
        
        # 获取session
        sess = data["sess"]
        
        return {
            "start_pos": slider_info["init_pos"],  # [50, 226]
            "move_limit": [min_x, max_x],          # [50, 552]
            "sess": sess,
            "image_urls": {
                "bg": bg_img_url,
                "sprite": sprite_url
            }
        }
        
    except Exception as e:
        print(f"解析位置信息失败: {str(e)}")
        return None

def get_captcha_prehandle():
    # 基础URL
    base_url = "https://turing.captcha.qcloud.com/cap_union_prehandle"
    
    # 构造请求参数
    params = {
        "aid": "2026859470",
        "protocol": "https",
        "accver": "1",
        "showtype": "popup",
        "ua": "TW96aWxsYS81LjAgKFdpbmRvd3MgTlQgMTAuMDsgV2luNjQ7IHg2NCkgQXBwbGVXZWJLaXQvNTM3LjM2IChLSFRNTCwgbGlrZSBHZWNrbykgQ2hyb21lLzEzMi4wLjAuMCBTYWZhcmkvNTM3",
        "noheader": "1",
        "fb": "1",
        "aged": "0",
        "enableAged": "0",
        "enableDarkMode": "0",
        "grayscale": "1",
        "clientype": "2",
        "cap_cd": "",
        "uid": "",
        "lang": "zh-cn",
        "entry_url": "https://open.yuewen.com/",
        "elder_captcha": "0",
        "js": "/tcaptcha-frame.c055d939.js",
        "login_appid": "",
        "wb": "1",
        "subsid": "5",
        "callback": "_aq_88765",
        "sess": ""
    }
    
    # 设置请求头
    headers = {
        "accept": "*/*",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
        "sec-ch-ua": '"Not A(Brand";v="8", "Chromium";v="132", "Microsoft Edge";v="132"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "script",
        "sec-fetch-mode": "no-cors",
        "sec-fetch-site": "cross-site",
        "Referer": "https://open.yuewen.com/",
        "Referrer-Policy": "strict-origin-when-cross-origin"
    }
    
    try:
        # 发送GET请求
        response = requests.get(
            base_url,
            params=params,
            headers=headers
        )
        
        # 检查响应状态
        response.raise_for_status()
        
        # 获取响应文本
        return response.text
        
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {str(e)}")
        return None

# start_pos = [50, 226]    # 滑块初始位置
# target_pos = [342, 226]  # 目标位置

# collect = get_collect(start_pos, target_pos)
# print(collect)