project/mt_V2/yzm.py

import base64
import io
import math
import os
import random
import re
import threading
import time

import cv2
import numpy as np
import requests
from PIL import Image, ImageDraw
import uiautomator2 as u2

API_URL = "http://api.jfbym.com/api/YmServer/customApi"
# 验证码平台 token
API_TOKEN = "1nDVocTE2mJ0yLEYb2sZJ5uUY2VIEoGTkIpW44X7Kgk"
# 原始截图保存路径
SCREENSHOT_PATH = "./a.jpg"
# 裁剪后图片保存路径z
CROP_PATH = "./b.jpg"
# 滑动轨迹图保存目录
TRACK_DIR = "./slider_tracks"
# 验证码截图日志根目录
CAPTCHA_LOG_ROOT = "./captcha_logs"
UNKNOWN_DEVICE_ID = "unknown_device"
_RUNTIME_CONTEXT = threading.local()
SLIDER_METHODS = ("track", "bezier")

# 图标点选验证码
CAPTCHA_ICON_CLICK = "icon_click"
# 空间推理验证码
CAPTCHA_SPACE_REASON = "space_reason"
# 文字点选验证码
CAPTCHA_TEXT_CLICK = "text_click"
# 滑块验证码
CAPTCHA_SLIDER = "slider"
# 图片内容输入验证码
CAPTCHA_TEXT_INPUT = "text_input"

# 输入型验证码的裁剪区域
INPUT_CROP = {
    "x_start": 210,
    "x_end": 510,
    "y_start": 666,
    "y_end": 788,
}

# 图标点选验证码的裁剪区域
ICON_CLICK_CROP = {
    "x_start": 40,
    "x_end": 680,
    "y_start": 471,
    "y_end": 1143,
}

# 不同验证码类型对应的平台参数配置
VERIFY_CONFIG = {
    CAPTCHA_ICON_CLICK: {"type": "88888", "direction": None, "extra": None, "y_offset": 0, "result": "points"},
    CAPTCHA_SPACE_REASON: {"type": "88888", "direction": None, "extra": "请点击", "y_offset": 0, "result": "points"},
    CAPTCHA_TEXT_CLICK: {"type": "30114", "direction": None, "extra": "phrase", "y_offset": 0, "result": "points"},
    CAPTCHA_SLIDER: {"type": "22222", "direction": None, "extra": None, "y_offset": 0, "result": "text"},
    CAPTCHA_TEXT_INPUT: {"type": "10103", "direction": None, "extra": None, "y_offset": 0, "result": "text"},
    # lianxian : {"type": "10114", "direction": None, "extra": None, "y_offset": 0, "result": "text"}
}


def _safe_name(value, default):
    if value is None:
        return default
    text = str(value).strip()
    if not text:
        return default
    text = re.sub(r"[^0-9A-Za-z_\-.]+", "_", text)
    return text or default


def _resolve_device_id(d=None, device_id=None):
    if device_id:
        return _safe_name(device_id, UNKNOWN_DEVICE_ID)

    candidates = []
    if d is not None:
        for attr in ("serial", "_serial", "device_id"):
            value = getattr(d, attr, None)
            if value:
                candidates.append(value)
        try:
            info = d.device_info
            if isinstance(info, dict):
                for key in ("serial", "serialno", "udid", "deviceId"):
                    value = info.get(key)
                    if value:
                        candidates.append(value)
        except Exception:
            pass

    for value in candidates:
        normalized = _safe_name(value, "")
        if normalized:
            return normalized
    return UNKNOWN_DEVICE_ID


def _set_runtime_device_id(d=None, device_id=None):
    _RUNTIME_CONTEXT.device_id = _resolve_device_id(d=d, device_id=device_id)


def _get_runtime_device_id(d=None):
    current = getattr(_RUNTIME_CONTEXT, "device_id", None)
    if current:
        return current
    resolved = _resolve_device_id(d=d)
    _RUNTIME_CONTEXT.device_id = resolved
    return resolved


def _next_slider_method_order():
    idx = getattr(_RUNTIME_CONTEXT, "slider_method_idx", 0)
    first = SLIDER_METHODS[idx % len(SLIDER_METHODS)]
    second = SLIDER_METHODS[(idx + 1) % len(SLIDER_METHODS)]
    _RUNTIME_CONTEXT.slider_method_idx = (idx + 1) % len(SLIDER_METHODS)
    return [first, second]


def _ensure_captcha_dir(captcha_type):
    safe_type = _safe_name(captcha_type, "unknown_captcha")
    folder = os.path.join(CAPTCHA_LOG_ROOT, safe_type)
    os.makedirs(folder, exist_ok=True)
    return folder


def _build_captcha_image_path(captcha_type, d=None, ext=".png", tag=None):
    device_id = _get_runtime_device_id(d=d)
    random_part = random.randint(10000000, 99999999)
    safe_tag = _safe_name(tag, "") if tag else ""
    suffix = f"_{safe_tag}" if safe_tag else ""
    filename = f"{device_id}_{random_part}{suffix}{ext}"
    return os.path.join(_ensure_captcha_dir(captcha_type), filename)


def _save_debug_screenshot(d, captcha_type, tag="full"):
    path = _build_captcha_image_path(captcha_type, d=d, ext=".png", tag=tag)
    try:
        d.screenshot(path)
        print(f"[captcha-shot] saved: {path}")
        return path
    except Exception as e:
        print(f"[captcha-shot] save failed: {e}")
        return None


def post_api(image_path, captcha_type, extra=None, direction=None, label_image_path=None, timeout=20):
    with open(image_path, 'rb') as f:
        image_base64 = base64.b64encode(f.read()).decode()

    data = {
        "token": API_TOKEN,
        "type": captcha_type,
        "image": image_base64,
    }
    if label_image_path:
        with open(label_image_path, 'rb') as f:
            data["label_image"] = base64.b64encode(f.read()).decode()
    if extra is not None:
        data["extra"] = extra
    if direction is not None:
        data["direction"] = direction

    headers = {
        "Content-Type": "application/json"
    }
    response = requests.post(API_URL, headers=headers, json=data, timeout=timeout).json()
    print(response)
    return response


def parse_points(response, y_offset=0):
    tuple_points = []
    data = response.get("data", {}).get("data", "")
    if not data:
        return tuple_points

    for s in data.split('|'):
        x, y = s.split(',')
        tuple_points.append((int(x), int(y) + y_offset))
    return tuple_points


def verify(image_path, captcha_type):
    config = VERIFY_CONFIG.get(captcha_type)
    if not config:
        raise ValueError(f"不支持的验证码类型: {captcha_type}")

    response = post_api(
        image_path,
        config["type"],
        extra=config["extra"],
        direction=config["direction"]
    )

    if config["result"] == "points":
        return parse_points(response, y_offset=config["y_offset"])

    return response.get("data", {}).get("data")


def crop_image_xy(
        image_path,
        output_path=None,
        x_start=None,
        x_end=None,
        y_start=471,
        y_end=1143
):
    if output_path is None:
        dir_name, file_name = os.path.split(image_path)
        name, ext = os.path.splitext(file_name)
        output_path = os.path.join(dir_name, f"{name}_cropped{ext}")

    with Image.open(image_path) as img:
        width, height = img.size

        if x_start is None:
            x_start = 0
        if x_end is None:
            x_end = width - 1

        x_start = max(0, min(x_start, width - 1))
        x_end = max(x_start, min(x_end, width - 1))
        y_start = max(0, min(y_start, height - 1))
        y_end = max(y_start, min(y_end, height - 1))

        cropped = img.crop((x_start, y_start, x_end + 1, y_end + 1))
        cropped.save(output_path)

    return output_path


def _capture_by_bounds(d, xpath_candidates, output_path=None, screenshot_path=None, captcha_type="generic"):
    """
    按元素 bounds 截图并裁剪。
    return: (cropped_path, bounds_dict) or (None, None)
    """
    if isinstance(xpath_candidates, str):
        xpath_candidates = [xpath_candidates]

    if screenshot_path is None:
        screenshot_path = _build_captcha_image_path(captcha_type, d=d, ext=".png", tag="full")
    if output_path is None:
        output_path = _build_captcha_image_path(captcha_type, d=d, ext=".png", tag="crop")

    for xpath in xpath_candidates:
        try:
            node = d.xpath(xpath)
            if not node.exists:
                continue
            bounds = node.info.get("bounds", {})
            if not bounds:
                continue

            d.screenshot(screenshot_path)
            cropped = crop_image_xy(
                screenshot_path,
                output_path=output_path,
                x_start=bounds["left"],
                x_end=bounds["right"],
                y_start=bounds["top"],
                y_end=bounds["bottom"],
            )
            return cropped, bounds
        except Exception:
            continue

    return None, None


def srwz(d):
    captcha_image_xpaths = [
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]/android.widget.Image[1]',
        '//*[@resource-id="captchaImg"]',
    ]
    image_path, _ = _capture_by_bounds(d, captcha_image_xpaths, captcha_type=CAPTCHA_TEXT_INPUT)
    if not image_path:
        return False

    data = verify(image_path, CAPTCHA_TEXT_INPUT)
    input_box = d.xpath('//*[@hint="请输入验证码"]')
    if input_box.exists:
        input_box.click()
        time.sleep(0.5)
        input_box.set_text(data)
        d.xpath('//*[@text="验证"]').click()
        return True
    else:
        print("未找到输入框")
        return False


def _clamp(value, min_value, max_value):
    return max(min_value, min(value, max_value))


def _slider_duration(distance):
    if distance <= 90:
        return round(random.uniform(0.18, 0.27), 3)
    if distance <= 160:
        return round(random.uniform(0.23, 0.34), 3)
    return round(random.uniform(0.28, 0.42), 3)


def _build_human_slider_track(start_x, start_y, distance):
    # Keep the reference shape: mostly flat first, then a single smooth downward bend.
    overshoot = random.randint(1, 2) if distance > 140 and random.random() < 0.18 else 0
    move_distance = distance + overshoot
    steps = int(_clamp(move_distance / random.uniform(6.0, 8.0), 16, 32))

    base_y = start_y + random.randint(-1, 1)
    flat_ratio = random.uniform(0.30, 0.42)  # front section almost horizontal
    drop = distance * random.uniform(0.08, 0.14)
    if distance > 260:
        drop *= random.uniform(1.35, 1.75)
    drop = _clamp(drop, 12.0, 58.0)  # obvious tail drop
    noise_amp = random.uniform(0.03, 0.22)

    points = [(start_x, base_y)]
    last_x = start_x

    for i in range(1, steps + 1):
        t = i / steps

        progress = 1 - (1 - t) ** 2.0
        progress += random.uniform(-0.0018, 0.0018)
        progress = _clamp(progress, 0.0, 1.0)

        x = start_x + int(move_distance * progress)
        if x <= last_x:
            x = last_x + 1
        last_x = x

        if t < flat_ratio:
            # slight tiny rise then back, still near flat
            u = t / max(flat_ratio, 1e-6)
            y_curve = -0.9 * math.sin(math.pi * u)
        else:
            u = (t - flat_ratio) / max(1 - flat_ratio, 1e-6)
            y_curve = drop * (u ** 1.7)

        y_noise = random.uniform(-noise_amp, noise_amp)
        y = int(round(base_y + y_curve + y_noise))
        points.append((x, y))

    final_x = start_x + distance
    final_y = int(round(base_y + drop + random.uniform(-0.6, 0.6)))
    if overshoot > 0:
        points.append((start_x + distance + overshoot, final_y))
    points.append((final_x, final_y + random.choice([0, 0, 1])))
    return points


def _save_slider_track_image(
        points,
        distance=None,
        duration=None,
        drag_ok=True,
        screenshot_path=SCREENSHOT_PATH,
        captcha_type=CAPTCHA_SLIDER
):
    if not points:
        return None

    output_path = _build_captcha_image_path(captcha_type, ext=".png", tag="track")

    try:
        if os.path.exists(screenshot_path):
            with Image.open(screenshot_path) as img:
                canvas = img.convert("RGB")
        else:
            canvas = Image.new("RGB", (720, 1280), "white")

        draw = ImageDraw.Draw(canvas)

        if len(points) >= 2:
            draw.line(points, fill=(245, 20, 30), width=9)

        sx, sy = points[0]
        ex, ey = points[-1]
        draw.ellipse((sx - 5, sy - 5, sx + 5, sy + 5), fill=(40, 200, 80))
        draw.ellipse((ex - 5, ey - 5, ex + 5, ey + 5), fill=(50, 120, 255))

        info = f"ok={drag_ok} dist={distance} dur={duration}s points={len(points)}"
        draw.rectangle((8, 8, min(canvas.size[0] - 8, 520), 42), fill=(0, 0, 0))
        draw.text((14, 14), info, fill=(255, 255, 255))

        canvas.save(output_path)
        print(f"[slider-track] saved: {output_path}")
        return output_path
    except Exception as e:
        print(f"[slider-track] save failed: {e}")
        return None


def _downsample_track_points(points, target_count):
    if not points or len(points) <= target_count:
        return points[:]
    if target_count < 2:
        return [points[0], points[-1]]

    sampled = []
    last_index = len(points) - 1
    for i in range(target_count):
        idx = int(round(i * last_index / (target_count - 1)))
        sampled.append(points[idx])
    return sampled


def _execute_track(d, points, total_duration):
    if not points or len(points) < 2:
        return False, points

    duration = max(0.18, float(total_duration))

    # Use fewer points to avoid step explosion, keep curve shape.
    max_points = random.randint(14, 22)
    exec_points = _downsample_track_points(points, max_points)

    try:
        if hasattr(d, "swipe_points"):
            # u2: duration here means time-per-step; steps = duration/0.005.
            # To approximate total duration:
            # total ~= (duration/0.005) * (len(exec_points)-1) * 0.005 = duration * segments
            seg_count = max(1, len(exec_points) - 1)
            per_segment = max(0.01, duration / seg_count)
            d.swipe_points(exec_points, duration=per_segment)
            return True, exec_points
    except Exception:
        pass

    # fallback
    try:
        sx, sy = exec_points[0]
        ex, ey = exec_points[-1]
        d.swipe(sx, sy, ex, ey, duration=duration)
        return False, exec_points
    except Exception:
        return False, exec_points


def _slider_still_exists(d):
    xpath_candidates = [
        '//*[@text="请拖动下方滑块完成拼图"]',
        '//*[contains(@text, "拖动下方滑块")]',
        '//*[@resource-id="puzzleSliderBox"]',
        '//*[@resource-id="puzzleImageMain"]',
    ]
    for xpath in xpath_candidates:
        try:
            if d.xpath(xpath).exists:
                return True
        except Exception:
            continue
    return False


def _cubic_bezier(t, p0, p1, p2, p3):
    one_minus_t = 1 - t
    x = (
            one_minus_t ** 3 * p0[0]
            + 3 * one_minus_t ** 2 * t * p1[0]
            + 3 * one_minus_t * t ** 2 * p2[0]
            + t ** 3 * p3[0]
    )
    y = (
            one_minus_t ** 3 * p0[1]
            + 3 * one_minus_t ** 2 * t * p1[1]
            + 3 * one_minus_t * t ** 2 * p2[1]
            + t ** 3 * p3[1]
    )
    return x, y


def _generate_bezier_slider_points(start, end, deviation=30, steps=50):
    sx, sy = start
    ex, ey = end
    mid_x = (sx + ex) / 2
    mid_y = (sy + ey) / 2

    p1 = (
        mid_x - (ex - sx) / 4 + random.uniform(-deviation, deviation),
        mid_y - (ey - sy) / 4 + random.uniform(-deviation / 2, deviation / 2),
    )
    p2 = (
        mid_x + (ex - sx) / 4 + random.uniform(-deviation, deviation),
        mid_y + (ey - sy) / 4 + random.uniform(-deviation / 2, deviation / 2),
    )

    points = []
    for i in range(steps + 1):
        t = i / steps
        x, y = _cubic_bezier(t, start, p1, p2, end)
        if 0 < i < steps:
            x += random.gauss(0, 1.5)
            y += random.gauss(0, 1.5)
        points.append((int(round(x)), int(round(y))))
    return points


def _execute_bezier_slider(d, start_x, start_y, end_x, end_y):
    points = _generate_bezier_slider_points(
        (int(round(start_x)), int(round(start_y))),
        (int(round(end_x)), int(round(end_y))),
        deviation=random.randint(20, 40),
        steps=50,
    )
    if len(points) < 2:
        return False, points

    try:
        d.touch.down(points[0][0], points[0][1])
        time.sleep(random.uniform(0.1, 0.2))
        total = max(1, len(points) - 1)
        for i, (x, y) in enumerate(points[1:], 1):
            d.touch.move(x, y)
            t = i / total
            if 0.2 < t < 0.8:
                interval = random.uniform(0.02, 0.04)
            else:
                interval = random.uniform(0.04, 0.08)
            time.sleep(interval)
        time.sleep(random.uniform(0.05, 0.15))
        d.touch.up(points[-1][0], points[-1][1])
        return True, points
    except Exception:
        try:
            d.touch.up(points[-1][0], points[-1][1])
        except Exception:
            pass
        return False, points


def _slider_knob_center(d):
    slider_xpath = (
        '//*[@resource-id="puzzleSliderBox"] | '
        '//*[@resource-id="yodaBox"] | '
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[2]/android.view.View[1]'
    )
    try:
        slider_node = d.xpath(slider_xpath)
        if slider_node.exists:
            bounds = slider_node.info.get("bounds", {})
            if bounds:
                return (
                    (bounds["left"] + bounds["right"]) / 2 + random.uniform(-4, 4),
                    (bounds["top"] + bounds["bottom"]) / 2 + random.uniform(-3, 3),
                )
    except Exception:
        pass
    return None


def _first_existing_bounds(d, xpath_candidates):
    if isinstance(xpath_candidates, str):
        xpath_candidates = [xpath_candidates]
    for xpath in xpath_candidates:
        try:
            node = d.xpath(xpath)
            if not node.exists:
                continue
            bounds = node.info.get("bounds", {})
            if bounds:
                return xpath, bounds
        except Exception:
            continue
    return None, None


def _click_captcha_close(d, captcha_xpath=None):
    """点击验证码右上角关闭按钮；优先点显式关闭控件，失败后按容器右上角估算点位。"""
    close_xpaths = [
        '//*[@resource-id="com.sankuai.meituan:id/btn_close_verify"]',
        '//*[@resource-id="btn_close_verify"]',
        '//*[@content-desc="关闭"]',
        '//*[@text="关闭"]',
    ]
    _, close_bounds = _first_existing_bounds(d, close_xpaths)
    if close_bounds:
        cx = int((close_bounds["left"] + close_bounds["right"]) / 2) + random.randint(-2, 2)
        cy = int((close_bounds["top"] + close_bounds["bottom"]) / 2) + random.randint(-2, 2)
        d.click(cx, cy)
        print(f"[captcha-close] click explicit close at ({cx}, {cy})")
        return True

    popup_xpaths = [
        '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]',
        '//*[@resource-id="com.sankuai.meituan:id/h5_container"]',
        '//*[@resource-id="root"]',
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]',
        '//*[@text="身份核实"]/android.view.View[1]',
    ]
    if captcha_xpath:
        popup_xpaths.append(captcha_xpath)
    _, popup_bounds = _first_existing_bounds(d, popup_xpaths)
    if not popup_bounds:
        return False

    left = popup_bounds["left"]
    right = popup_bounds["right"]
    top = popup_bounds["top"]
    bottom = popup_bounds["bottom"]
    width = max(1, right - left)
    height = max(1, bottom - top)

    # 参考示例：[40,391][680,1223] -> 右上角叉号中心约(640, 431)。
    offset_x = int(_clamp(width * 0.06, 20, 56))
    offset_y = int(_clamp(height * 0.05, 20, 56))
    click_x = int(right - offset_x) + random.randint(-3, 3)
    click_y = int(top + offset_y) + random.randint(-3, 3)
    d.click(click_x, click_y)
    print(f"[captcha-close] click inferred close at ({click_x}, {click_y})")
    return True


def _build_directional_track(start_x, start_y, end_x, end_y):
    distance_x = end_x - start_x
    distance_y = end_y - start_y
    if abs(distance_x) < 2 and abs(distance_y) < 2:
        return [(int(start_x), int(start_y)), (int(end_x), int(end_y))]

    steps = int(_clamp(abs(distance_x) / random.uniform(7.0, 10.0), 22, 48))
    points = [(int(start_x), int(start_y))]
    last_x = float(start_x)

    for i in range(1, steps + 1):
        t = i / steps
        progress = 1 - (1 - t) ** random.uniform(1.8, 2.25)

        x = start_x + distance_x * progress + random.uniform(-0.9, 0.9)
        y = start_y + distance_y * progress + random.uniform(-0.8, 0.8)

        if distance_x >= 0:
            if x < last_x:
                x = last_x + random.uniform(0.2, 1.2)
        else:
            if x > last_x:
                x = last_x - random.uniform(0.2, 1.2)

        last_x = x
        points.append((int(round(x)), int(round(y))))

    points.append((int(round(end_x)), int(round(end_y))))
    return points


def _move_with_pressed_touch(d, points):
    if not points:
        return
    for x, y in points:
        d.touch.move(x, y)
        time.sleep(random.uniform(0.0015, 0.0045))


def _screenshot_to_image(d):
    shot = d.screenshot()
    if isinstance(shot, Image.Image):
        return shot.convert("RGB")
    if isinstance(shot, bytes):
        return Image.open(io.BytesIO(shot)).convert("RGB")
    if isinstance(shot, str) and os.path.exists(shot):
        return Image.open(shot).convert("RGB")
    if hasattr(shot, "convert"):
        return shot.convert("RGB")
    fallback_path = _build_captcha_image_path("generic", d=d, ext=".png", tag="fallback")
    d.screenshot(fallback_path)
    return Image.open(fallback_path).convert("RGB")


def hk(d):
    screenshot_path = _build_captcha_image_path(CAPTCHA_SLIDER, d=d, ext=".png", tag="full")
    d.screenshot(screenshot_path)
    data = verify(screenshot_path, CAPTCHA_SLIDER)
    if not data:
        return False

    try:
        raw_distance = float(data)
    except (TypeError, ValueError):
        return False

    if raw_distance <= 0:
        return False

    image_width = 720
    try:
        with Image.open(screenshot_path) as img:
            image_width = img.size[0] or 720
    except Exception:
        pass

    try:
        screen_width = int(d.info.get("displayWidth", image_width))
    except Exception:
        screen_width = image_width

    scale = screen_width / image_width if image_width else 1.0
    slide_distance = int(raw_distance * scale)
    if slide_distance < 80:
        slide_distance += random.randint(3, 6)
    elif slide_distance < 160:
        slide_distance += random.randint(2, 5)
    else:
        slide_distance += random.randint(1, 4)

    start_x = 84 + random.randint(-1, 1)
    start_y = 1034 + random.randint(-2, 2)
    knob_center = _slider_knob_center(d)
    if knob_center:
        start_x, start_y = knob_center
    max_target_x = screen_width - random.randint(26, 42)
    target_x = _clamp(start_x + slide_distance, start_x + 18, max_target_x)
    distance = target_x - start_x
    if distance < 18:
        return False

    method_order = _next_slider_method_order()
    for method in method_order:
        if method == "track":
            points = _build_human_slider_track(int(round(start_x)), int(round(start_y)), int(round(distance)))
            duration = _slider_duration(distance)
            time.sleep(random.uniform(0.015, 0.05))
            drag_ok, exec_points = _execute_track(d, points, duration)
            _save_slider_track_image(
                exec_points,
                distance=distance,
                duration=duration,
                drag_ok=drag_ok,
                screenshot_path=screenshot_path,
                captcha_type=CAPTCHA_SLIDER
            )
        else:
            end_x = target_x + random.uniform(-3, 3)
            end_y = start_y + random.uniform(-1, 1)
            drag_ok, exec_points = _execute_bezier_slider(d, start_x, start_y, end_x, end_y)
            _save_slider_track_image(
                exec_points,
                distance=distance,
                duration=None,
                drag_ok=drag_ok,
                screenshot_path=screenshot_path,
                captcha_type=CAPTCHA_SLIDER
            )

        time.sleep(random.uniform(1.0, 1.8))
        if not _slider_still_exists(d):
            return True
        print(f"[slider] method {method} failed, switch to next")

    return False


def dianxuan(d):
    click_area_xpaths = [
        '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]',
        '//*[@resource-id="com.sankuai.meituan:id/h5_container"]',
        '//*[@resource-id="root"]',
    ]
    image_path, bounds = _capture_by_bounds(d, click_area_xpaths, captcha_type=CAPTCHA_ICON_CLICK)
    if not image_path or not bounds:
        return False

    left = bounds["left"]
    top = bounds["top"]
    data = verify(image_path, CAPTCHA_ICON_CLICK)
    if not data:
        return False

    for x, y in data:
        time.sleep(random.randint(1, 2))
        d.click(left + x + random.randint(-7, 7), top + y + random.randint(-7, 7))
    return True


def wenzidianxuan(d):
    # 文字点选：按验证码容器 bounds 裁剪后，调用 88888，按返回坐标依次点击
    click_area_xpaths = [
        '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]',
        '//*[@resource-id="com.sankuai.meituan:id/h5_container"]',
        '//*[@resource-id="root"]',
    ]
    image_path, bounds = _capture_by_bounds(d, click_area_xpaths, captcha_type=CAPTCHA_TEXT_CLICK)
    if not image_path or not bounds:
        return False

    image_left = bounds["left"]
    image_top = bounds["top"]

    result = post_api(image_path, "88888")
    verify_data = result.get("data", {})
    if not (result.get("code") == 10000 and isinstance(verify_data, dict) and verify_data.get("code") == 0):
        return False

    coords_str = verify_data.get("data", "")
    if not coords_str:
        return False

    clicked = 0
    for coord in coords_str.split("|"):
        try:
            x_img_str, y_img_str = coord.split(",")
            x_img = int(x_img_str.strip())
            y_img = int(y_img_str.strip())
            x_screen = image_left + x_img + random.randint(-5, 5)
            y_screen = image_top + y_img + random.randint(-5, 5)
            d.click(x_screen, y_screen)
            clicked += 1
            time.sleep(random.uniform(0.8, 1.6))
        except Exception:
            continue

    return clicked > 0


def click_side(d):
    """空间推理验证码（请点击数字）。"""
    click_area_xpaths = [
        '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]',
        '//*[@resource-id="com.sankuai.meituan:id/h5_container"]',
        '//*[@resource-id="root"]',
    ]
    image_path, bounds = _capture_by_bounds(d, click_area_xpaths, output_path=CROP_PATH)
    if not image_path or not bounds:
        return False

    left = bounds["left"]
    top = bounds["top"]
    points = verify(image_path, CAPTCHA_SPACE_REASON)
    if not points:
        return False

    x, y = points[0]
    d.click(left + x + random.randint(-2, 2), top + y + random.randint(-2, 2))

    # d_list = [
    #         '//*[@resource-id="com.sankuai.meituan:id/btn_close_verify"]',
    #         '//*[@resource-id="com.sankuai.meituan:id/yoda_toolbar_title"]',
    #         '//*[@resource-id="com.sankuai.meituan:id/btn_close_verify"]'
    #           ]
    # for i in d_list:
    #     d.xpath(i).click()
    return True


def Swipe_right(d):
    """向右拖动到最右侧（非拼图滑块）。"""
    track_xpath = (
        '//*[@resource-id="yodaBoxWrapper"] | '
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]'
    )
    slider_xpath = (
        '//*[@resource-id="yodaBox"] | '
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]/android.view.View[1]'
    )

    if not d.xpath(track_xpath).exists or not d.xpath(slider_xpath).exists:
        return False

    track_bounds = d.xpath(track_xpath).info.get("bounds", {})
    slider_bounds = d.xpath(slider_xpath).info.get("bounds", {})
    if not track_bounds or not slider_bounds:
        return False

    start_x = int((slider_bounds["left"] + slider_bounds["right"]) / 2) + random.randint(-2, 2)
    start_y = int((slider_bounds["top"] + slider_bounds["bottom"]) / 2) + random.randint(-2, 2)
    right_limit = int(track_bounds["right"]) - random.randint(4, 10)
    distance = right_limit - start_x
    if distance <= 10:
        return False

    screenshot_path = _save_debug_screenshot(d, "Swipe_right", tag="full")
    points = _build_human_slider_track(start_x, start_y, distance)
    duration = _slider_duration(distance)
    drag_ok, exec_points = _execute_track(d, points, duration)
    _save_slider_track_image(
        exec_points,
        distance=distance,
        duration=duration,
        drag_ok=drag_ok,
        screenshot_path=screenshot_path or SCREENSHOT_PATH,
        captcha_type="Swipe_right"
    )
    return True


def complexs(d):
    """Complex slider flow: move to far-right, OCR with label image, then drag back to target."""
    slider_xpath_candidates = [
        '//*[@resource-id="yodaBox"]',
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[2]/android.view.View[1]',
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]/android.view.View[1]',
    ]
    track_xpath_candidates = [
        '//*[@resource-id="yodaBoxWrapper"]',
        '//*[contains(@text, "请按照说明拖动滑块")]',
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]',
    ]
    label_xpath_candidates = [
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.widget.TextView[1]',
    ]
    image_xpath_candidates = [
        '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]',
    ]

    _, slider_bounds = _first_existing_bounds(d, slider_xpath_candidates)
    _, track_bounds = _first_existing_bounds(d, track_xpath_candidates)
    if not slider_bounds or not track_bounds:
        return False

    slider_left = slider_bounds["left"]
    slider_top = slider_bounds["top"]
    slider_right = slider_bounds["right"]
    slider_bottom = slider_bounds["bottom"]
    slider_width = slider_right - slider_left
    slider_center_x = (slider_left + slider_right) / 2
    slider_center_y = (slider_top + slider_bottom) / 2

    track_left = track_bounds["left"]
    track_right = track_bounds["right"]
    right_end_center_x = track_right - slider_width / 2
    right_end_center_y = slider_center_y

    print(f"滑块中心: ({slider_center_x}, {slider_center_y})")
    print(f"最右端滑块中心坐标: ({right_end_center_x}, {right_end_center_y})")

    touch_down = False
    try:
        d.touch.down(slider_center_x, slider_center_y)
        touch_down = True
        time.sleep(random.uniform(0.08, 0.16))

        move_right_points = _build_directional_track(
            slider_center_x,
            slider_center_y,
            right_end_center_x,
            right_end_center_y,
        )
        _move_with_pressed_touch(d, move_right_points[1:])
        print("滑块已到达最右端")

        _, label_bounds = _first_existing_bounds(d, label_xpath_candidates)
        _, image_bounds = _first_existing_bounds(d, image_xpath_candidates)
        if not label_bounds or not image_bounds:
            return False

        capture_label_left = label_bounds["left"]
        capture_label_top = label_bounds["top"]
        capture_label_right = label_bounds["right"]
        capture_label_bottom = label_bounds["bottom"]

        capture_left = image_bounds["left"]
        capture_top = image_bounds["top"]
        capture_right = image_bounds["right"]
        capture_bottom = image_bounds["bottom"]

        print(
            "截图区域1(提示文本): "
            f"left={capture_label_left}, top={capture_label_top}, "
            f"width={capture_label_right - capture_label_left}, "
            f"height={capture_label_bottom - capture_label_top}"
        )
        print(
            "截图区域2(图片): "
            f"left={capture_left}, top={capture_top}, "
            f"width={capture_right - capture_left}, "
            f"height={capture_bottom - capture_top}"
        )

        screenshot_label_path = _build_captcha_image_path("complexs", d=d, ext=".png", tag="label")
        screenshot_image_path = _build_captcha_image_path("complexs", d=d, ext=".png", tag="image")
        image = _screenshot_to_image(d)
        image.crop(
            (capture_label_left, capture_label_top, capture_label_right, capture_label_bottom)
        ).save(screenshot_label_path)
        image.crop(
            (capture_left, capture_top, capture_right, capture_bottom)
        ).save(screenshot_image_path)
        print(f"截图1已保存: {screenshot_label_path}")
        print(f"截图2已保存: {screenshot_image_path}")

        result = post_api(
            screenshot_image_path,
            "29013",
            label_image_path=screenshot_label_path,
            timeout=30,
        )
        print(f"API返回结果: {result}")
        verify_data = result.get("data", {})
        print(f"verify_data={verify_data}")

        if not (result.get("code") == 10000 and isinstance(verify_data, dict) and verify_data.get("code") == 0):
            return False

        data_str = verify_data.get("data", "")
        if not data_str:
            return False
        data_value = int(data_str)
        print(f"云码返回的像素距离: {data_value}")

        slider_target_center_x = track_left + data_value
        min_x = track_left + slider_width / 2
        max_x = track_right - slider_width / 2
        slider_target_center_x = _clamp(slider_target_center_x, min_x, max_x)
        print(f"滑块中心目标X坐标: {slider_target_center_x}")

        _, current_slider_bounds = _first_existing_bounds(d, slider_xpath_candidates)
        if current_slider_bounds:
            current_slider_center_x = (current_slider_bounds["left"] + current_slider_bounds["right"]) / 2
        else:
            current_slider_center_x = right_end_center_x
        actual_distance = slider_target_center_x - current_slider_center_x
        print(f"实际需要滑动的距离: {actual_distance}")

        back_points = _build_directional_track(
            current_slider_center_x,
            right_end_center_y,
            slider_target_center_x,
            right_end_center_y,
        )
        _move_with_pressed_touch(d, back_points[1:])
        time.sleep(random.uniform(0.2, 0.4))
        d.touch.up(slider_target_center_x, right_end_center_y)
        touch_down = False
        time.sleep(random.uniform(1.8, 3.2))
        return True
    except Exception as e:
        print(f"complex captcha failed: {e}")
        return False
    finally:
        if touch_down:
            try:
                d.touch.up(right_end_center_x, right_end_center_y)
            except Exception:
                pass


def Numbers_English_verify(d):
    return srwz(d)


def slider_verify(d):
    return hk(d)


def Click_images(d):
    # 两种点选入口统一处理
    if d.xpath('//*[@text="请按语序依次点击下图文字"]').exists:
        return wenzidianxuan(d)
    return dianxuan(d)


def Shortest_connection(d):
    return lianxian(d)


def _handle_generic_captcha(d, xpath, timeout=60):
    """通用验证码处理：等待人工处理完成。"""
    start = time.time()
    while time.time() - start < timeout:
        if xpath and not d.xpath(xpath).exists:
            return True
        time.sleep(1)
    return False


def handle_captcha(d, captcha_type, xpath=None, device_id=None):
    _set_runtime_device_id(d=d, device_id=device_id)
    handlers = {
        "Numbers_English": Numbers_English_verify,
        "Swipe_right": Swipe_right,
        "Click_images": Click_images,
        "slider": slider_verify,
        "complexs": complexs,
        "Shortest_connection": Shortest_connection,
        "click_side": click_side,
    }
    func = handlers.get(captcha_type)
    if func is None:
        return _handle_generic_captcha(d, xpath)
    return func(d)


def _extract_color_name(api_result):
    if not isinstance(api_result, dict):
        return ""

    if api_result.get("code") == 0 and isinstance(api_result.get("data"), str):
        return api_result.get("data", "").strip()

    if api_result.get("code") == 10000:
        inner = api_result.get("data")
        if isinstance(inner, dict) and inner.get("code") == 0:
            return str(inner.get("data", "")).strip()
        if isinstance(inner, str):
            return inner.strip()

    return ""


def _normalize_color_name(color_name):
    if not color_name:
        return ""

    alias = {
        "红": "红色",
        "红的": "红色",
        "绿": "绿色",
        "蓝": "蓝色",
        "黄": "黄色",
        "橙": "橙色",
        "紫": "紫色",
        "黑": "黑色",
        "白": "白色",
        "棕": "棕色",
        "褐": "褐色",
    }
    if color_name in alias:
        return alias[color_name]
    for k, v in alias.items():
        if k in color_name:
            return v
    return color_name


def _find_color_coordinates(image_path, color_name):
    color_name = _normalize_color_name(color_name)
    color_ranges = {
        "红色": (([0, 120, 70], [10, 255, 255]), ([170, 120, 70], [180, 255, 255])),
        "绿色": (([35, 50, 50], [85, 255, 255]),),
        "蓝色": (([90, 50, 50], [130, 255, 255]),),
        "黄色": (([20, 100, 100], [30, 255, 255]),),
        "橙色": (([5, 100, 100], [18, 255, 255]),),
        "紫色": (([130, 50, 50], [165, 255, 255]),),
        "黑色": (([0, 0, 0], [180, 255, 50]),),
        "白色": (([0, 0, 200], [180, 35, 255]),),
        "棕色": (([8, 60, 20], [20, 255, 180]),),
        "褐色": (([8, 60, 20], [20, 255, 180]),),
    }
    if color_name not in color_ranges:
        return []

    image = cv2.imread(image_path)
    if image is None:
        return []

    hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV)
    ranges = color_ranges[color_name]
    if len(ranges) == 2:
        lower1 = np.array(ranges[0][0])
        upper1 = np.array(ranges[0][1])
        lower2 = np.array(ranges[1][0])
        upper2 = np.array(ranges[1][1])
        mask = cv2.bitwise_or(cv2.inRange(hsv, lower1, upper1), cv2.inRange(hsv, lower2, upper2))
    else:
        lower = np.array(ranges[0][0])
        upper = np.array(ranges[0][1])
        mask = cv2.inRange(hsv, lower, upper)

    kernel = np.ones((3, 3), np.uint8)
    mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel)
    mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel)

    contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    coordinates = []
    for contour in contours:
        area = cv2.contourArea(contour)
        if area < 30:
            continue
        m = cv2.moments(contour)
        if m["m00"] == 0:
            continue
        cx = int(m["m10"] / m["m00"])
        cy = int(m["m01"] / m["m00"])
        coordinates.append((cx, cy))
    return coordinates


def _nearest_neighbor_path(points):
    if not points:
        return []
    if len(points) <= 2:
        return points[:]

    unvisited = points[:]
    path = [unvisited.pop(0)]
    while unvisited:
        last_x, last_y = path[-1]
        idx = min(
            range(len(unvisited)),
            key=lambda i: math.hypot(last_x - unvisited[i][0], last_y - unvisited[i][1])
        )
        path.append(unvisited.pop(idx))
    return path


def _human_like_path(points):
    if len(points) < 2:
        return points[:]

    curved = []
    for i in range(len(points) - 1):
        start = points[i]
        end = points[i + 1]
        mid_x = (start[0] + end[0]) / 2
        mid_y = (start[1] + end[1]) / 2

        if abs(end[0] - start[0]) > abs(end[1] - start[1]):
            offset_x = 0
            offset_y = random.uniform(-15, 15)
        else:
            offset_x = random.uniform(-15, 15)
            offset_y = 0

        control_x = mid_x + offset_x
        control_y = mid_y + offset_y
        curved.append(start)
        for t in np.arange(0.1, 1.0, 0.1):
            x = (1 - t) ** 2 * start[0] + 2 * (1 - t) * t * control_x + t ** 2 * end[0]
            y = (1 - t) ** 2 * start[1] + 2 * (1 - t) * t * control_y + t ** 2 * end[1]
            curved.append((int(x), int(y)))
    curved.append(points[-1])
    return curved


def _simulate_human_drawing(d, path):
    if len(path) < 2:
        return False
    try:
        sx, sy = path[0]
        d.touch.down(sx, sy)
        time.sleep(random.uniform(0.05, 0.1))
        for i in range(1, len(path)):
            x, y = path[i]
            d.touch.move(x + random.randint(-2, 2), y + random.randint(-2, 2))
            time.sleep(random.uniform(0.01, 0.03))
        time.sleep(random.uniform(0.1, 0.2))
        d.touch.up(path[-1][0], path[-1][1])
        return True
    except Exception:
        return False


def retry_captcha(
        d,
        xpath_text,
        handle_func,
        retry_count=5,
        captcha_name=None,
        allow_close_on_third_fail=True,
        fail_limit_before_close=3
):
    # 如果当前页面存在对应验证码，就循环重试处理
    if d.xpath(xpath_text).exists:
        current_name = captcha_name or getattr(handle_func, "__name__", "captcha")
        fail_streak = 0
        for _ in range(retry_count):
            if not d.xpath(xpath_text).exists:
                break
            _save_debug_screenshot(d, current_name, tag="full")
            try:
                handle_func(d)
            except Exception as e:
                print(f"[captcha] {current_name} handler error: {e}")
            time.sleep(3)
            # 验证码消失了，说明处理成功，直接退出
            if not d.xpath(xpath_text).exists:
                break
            fail_streak += 1
            if allow_close_on_third_fail and fail_streak >= fail_limit_before_close:
                closed = _click_captcha_close(d, captcha_xpath=xpath_text)
                print(f"[captcha] {current_name} failed {fail_streak} times, switch captcha: {closed}")
                fail_streak = 0
                time.sleep(1.2)


def yzm(d=None, device_id=None):
    # 如果没有传设备对象，就默认连接当前设备
    if d is None:
        d = u2.connect()
    _set_runtime_device_id(d=d, device_id=device_id)

    # 向右滑动验证码
    retry_captcha(
        d,
        '//*[contains(@text, "请向右滑动滑块")]',
        Swipe_right,
        captcha_name="Swipe_right",
        allow_close_on_third_fail=False
    )

    # 滑块验证码
    retry_captcha(
        d,
        '//*[@text="请拖动下方滑块完成拼图"]',
        hk,
        captcha_name="slider",

    )

    # 空间推理验证码
    retry_captcha(
        d,
        '//*[contains(@text, "请点击")]',
        click_side,
        captcha_name="click_side",
        allow_close_on_third_fail = False
    )

    # 复杂拖动滑块验证码
    retry_captcha(
        d,
        '//*[contains(@text, "拖动滑块")]',
        complexs,
        captcha_name="complexs",
        allow_close_on_third_fail=False
    )

    # 输入型验证码
    retry_captcha(d, '//*[@text="请输入图片中的内容"]', srwz, captcha_name="text_input")

    # 图标点选验证码
    retry_captcha(d, '//*[@text="请依次点击下图图标"]', dianxuan, captcha_name="icon_click")

    # 文字点选验证码
    retry_captcha(d, '//*[@text="请按语序依次点击下图文字"]', wenzidianxuan, captcha_name="text_click")

    # 最短线连接验证码
    retry_captcha(d, '//*[contains(@text, "用最短线连接")]', lianxian, captcha_name="Shortest_connection")


def lianxian(d):
    art_text_xpath = '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]'
    color_points_xpath = '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[2]/android.view.View[1]/android.widget.Image[1]'

    art_text_img_path = _build_captcha_image_path("Shortest_connection", d=d, ext=".png", tag="art_text")
    color_points_img_path = _build_captcha_image_path("Shortest_connection", d=d, ext=".png", tag="color_points")

    art_text_img_path, _ = _capture_by_bounds(
        d,
        art_text_xpath,
        output_path=art_text_img_path,
        captcha_type="Shortest_connection"
    )
    color_points_img_path, color_bounds = _capture_by_bounds(
        d,
        color_points_xpath,
        output_path=color_points_img_path,
        captcha_type="Shortest_connection"
    )
    if not art_text_img_path or not color_points_img_path or not color_bounds:
        return False

    element_left = color_bounds["left"]
    element_top = color_bounds["top"]
    element_width = color_bounds["right"] - color_bounds["left"]
    element_height = color_bounds["bottom"] - color_bounds["top"]

    api_result = post_api(art_text_img_path, "10118")
    color_name = _extract_color_name(api_result)
    if not color_name:
        return False

    relative_points = _find_color_coordinates(color_points_img_path, color_name)
    if len(relative_points) < 2:
        return False

    color_img = cv2.imread(color_points_img_path)
    if color_img is None:
        return False
    img_h, img_w = color_img.shape[:2]
    if img_w <= 0 or img_h <= 0:
        return False

    screen_points = []
    for rx, ry in relative_points:
        sx = element_left + int(rx * (element_width / img_w))
        sy = element_top + int(ry * (element_height / img_h))
        screen_points.append((sx, sy))

    path = _nearest_neighbor_path(screen_points)
    curved = _human_like_path(path)
    return _simulate_human_drawing(d, curved)


if __name__ == '__main__':
    d = u2.connect("GQIRKB7LVOONM7VW")
    yzm(d)