2.0/spiders/pdd/test_captcha.py

# coding=utf-8
"""PDD 滑块验证码测试 — 云码平台"""

import uiautomator2 as u2
import requests
import base64
import time
import subprocess
import logging
import random
import io
from PIL import Image

# ============================================================
# ★ 填参数
# ============================================================
DEVICE_ID = "ZDQWUSSWBEDI896T"
YUNMA_TOKEN = "1nDVocTE2mJ0yLEYb2sZJ5uUY2VIEoGTkIpW44X7Kgk"
MAX_ROUNDS = 1

# ============================================================
# PDD 验证码 XPath
# ============================================================
SAFE_VERIFY = '//android.widget.TextView[@text="安全验证"]'  # 入口按钮（可能没有）

# 验证码入口容器 → 一定存在，检测用
CAPTCHA_ENTRY = '//*[@resource-id="intelVerify"]'

# 真正的验证码弹窗容器 → 截图裁剪用
CAPTCHA_DIALOG = '//*[@resource-id="captcha-dialog"]'

# 弹窗内图标 + 标题 → 两个都有即判定在验证码弹窗内
CAPTCHA_ICON = '//*[@text=""]'
CAPTCHA_TITLE = '//*[@text="拼多多安全验证"]'

# 中间验证页面：同时存在「前往验证」和「安全验证」
GO_TO_VERIFY = '//*[@text="前往验证"]'

# 安全验证文本 XPath 候选列表（intelVerify 找不到时的兜底检测，≥2 命中即判定验证码存在）
SAFE_VERIFY_TEXT_XPATHS = [
    '//*[@text="点击下方"安全验证"按钮，完成拼多多官方验证"]',
    '//*[@text="b04804a5-893c-4af7-9076-6fc5f223be44"]',
    '//android.widget.TextView[@text="安全验证"]',
]

# 滑块定位候选列表（按优先级依次尝试）
SLIDER_CANDIDATES = [
    '//*[@resource-id="slide-button"]',
    '(//android.widget.TextView)[6]',

]

# 距离缩放系数（如果实际拖动距离与云码返回距离有偏差，调整此值）
DIST_SCALE = 1.0

# 计算题验证码
CALC_PROMPT = '//*[@text="请输入图中的计算结果"]'
SUBMIT_BTN_CANDIDATES = [
    '//*[@text="确定"]',
    '//*[@text="提交"]',
    '//*[@text="确认"]',
]

# 云码
YUNMA_URL       = "http://api.jfbym.com/api/YmServer/customApi"
YUNMA_TYPE      = "22222"   # 滑块
YUNMA_CALC_TYPE = "50100"   # 数字计算

logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
log = logging.getLogger(__name__)


def connect(dev):
    d = u2.connect_usb(dev)
    subprocess.run(f'adb -s {dev} shell /data/local/tmp/atx-agent server -d --stop',
                   capture_output=True, text=True, shell=True)
    time.sleep(2)
    subprocess.run(f'adb -s {dev} shell /data/local/tmp/atx-agent server -d',
                   capture_output=True, text=True, shell=True)
    time.sleep(2)
    return d


def _find_slider(d):
    """在候选列表中定位滑块元素，返回其 bounds，找不到返回 None"""
    for xpath in SLIDER_CANDIDATES:
        try:
            el = d.xpath(xpath)
            if el.exists:
                b = el.info.get('bounds', {})
                # 校验 bounds 合理性：宽高至少 10px
                w = b.get('right', 0) - b.get('left', 0)
                h = b.get('bottom', 0) - b.get('top', 0)
                if w >= 10 and h >= 10:
                    log.info(f"滑块定位成功: {xpath} → bounds={b}")
                    return b
        except Exception:
            continue
    return None


def _get_captcha_bounds(d):
    """获取验证码弹窗 bounds（captcha-dialog 容器，用于截图裁剪）"""
    el = d.xpath(CAPTCHA_DIALOG)
    if el.exists:
        b = el.info.get('bounds', {})
        if b:
            log.info(f"captcha-dialog bounds: {b}")
            return b
    return None


def _is_captcha_entry_visible(d):
    """验证码入口容器 intelVerify 是否可见"""
    return d.xpath(CAPTCHA_ENTRY).exists


def _is_captcha_dialog_visible(d):
    """验证码弹窗是否可见（图标 + 标题两个都有）"""
    return d.xpath(CAPTCHA_ICON).exists and d.xpath(CAPTCHA_TITLE).exists


def _is_captcha_detected_by_safe_text(d):
    """intelVerify 找不到时的回退检测：统计安全验证文本 XPath 命中数，≥2 即判定验证码存在"""
    hits = 0
    for xpath in SAFE_VERIFY_TEXT_XPATHS:
        try:
            if d.xpath(xpath).exists:
                hits += 1
                log.info(f"安全验证文本命中: {xpath}")
        except Exception:
            continue
    log.info(f"安全验证文本命中数: {hits}/{len(SAFE_VERIFY_TEXT_XPATHS)}")
    return hits >= 2


def _is_on_intermediate_verify_page(d):
    """检测是否在中间验证页面（同时存在「前往验证」+「安全验证」）"""
    has_go = d.xpath(GO_TO_VERIFY).exists
    has_safe = d.xpath(SAFE_VERIFY).exists
    log.info(f"中间页面检测: 前往验证={has_go}, 安全验证={has_safe}")
    return has_go and has_safe


def _wait_for_captcha_dialog(d, timeout=5):
    """等待 captcha-dialog + 标题出现，返回 True/False"""
    for _ in range(timeout):
        has_icon = d.xpath(CAPTCHA_ICON).exists
        has_title = d.xpath(CAPTCHA_TITLE).exists
        if has_icon and has_title:
            log.info("验证码弹窗确认: 图标 + 拼多多安全验证 ✓")
            return True
        log.info(f"等待弹窗... icon={has_icon}, title={has_title}")
        time.sleep(1)
    return False


def _enter_captcha_dialog(d):
    """等待弹窗出现，等不到则检测中间页面并点击「前往验证」后再等，成功返回 True"""
    if _wait_for_captcha_dialog(d):
        return True
    # 等了5次都没有 → 检测是否跳到中间页面
    if _is_on_intermediate_verify_page(d):
        log.info("检测到中间页面（前往验证 + 安全验证），点击「前往验证」...")
        d.xpath(GO_TO_VERIFY).click()
        time.sleep(1.5)
        if _wait_for_captcha_dialog(d):
            return True
        log.warning("中间页面导航后仍未出现弹窗")
        return False
    log.warning("等待超时，验证码弹窗未出现")
    return False


def _screenshot_to_b64(d, crop=False):
    """截图并返回 base64，可选裁剪到 captcha-dialog 区域"""
    im = d.screenshot()
    if crop:
        cb = _get_captcha_bounds(d)
        if cb:
            im = im.crop((
                max(0, cb['left']), max(0, cb['top']),
                min(im.width, cb['right']), min(im.height, cb['bottom']),
            ))
    buf = io.BytesIO()
    im.save(buf, format='PNG')
    return base64.b64encode(buf.getvalue()).decode()


def solve_calc(d):
    """解决计算题验证码，成功返回 True（50100 直接返回计算结果）"""
    log.info("检测到计算题验证码")

    # 截图 → 云码 50100
    b64 = _screenshot_to_b64(d, crop=True)

    resp = requests.post(YUNMA_URL, json={
        "token": YUNMA_TOKEN, "type": YUNMA_CALC_TYPE, "image": b64
    }, headers={"Content-Type": "application/json"}, timeout=30).json()
    log.info(f"云码(50100): {resp}")

    if resp.get("msg") != "识别成功":
        log.warning(f"云码 50100 识别失败: {resp.get('msg')}")
        return False

    data = resp.get("data", {})
    result_str = data.get("data", "") if isinstance(data, dict) else str(data or "")
    try:
        result = int(str(result_str).strip())
    except (ValueError, TypeError):
        log.warning(f"50100 返回非数字: {result_str}")
        return False
    log.info(f"计算结果: {result}")

    # 找到输入框并输入结果
    edit_text = d(className='android.widget.EditText')
    if not edit_text.exists:
        log.warning("未找到输入框")
        return False

    edit_text.click()
    time.sleep(0.3)
    edit_text.set_text(str(result))
    time.sleep(0.5)

    # 点击确定/提交
    clicked = False
    for xpath in SUBMIT_BTN_CANDIDATES:
        btn = d.xpath(xpath)
        if btn.exists:
            btn.click()
            clicked = True
            log.info(f"点击: {xpath}")
            break
    if not clicked:
        d.press('enter')

    time.sleep(2)

    # 验证是否解开
    if not _is_captcha_dialog_visible(d) and not _is_captcha_entry_visible(d):
        log.info("计算题验证码已解开 ✓")
        return True
    else:
        log.info("计算题验证码未解开")
        return False


def solve(d):
    """一次验证码处理，自动识别类型（滑块/计算题），成功返回 True"""
    # ── 第一步：检测验证码 ──
    # 优先级1: captcha-dialog 弹窗已存在（重入场景，跳过入口）
    if _is_captcha_dialog_visible(d):
        log.info("已在验证码弹窗内，跳过入口")
    # 优先级2: intelVerify 入口容器 → 点按钮进入
    elif d.xpath(CAPTCHA_ENTRY).exists:
        log.info("检测到验证码入口 intelVerify")

        if not d.xpath(SAFE_VERIFY).exists:
            log.error("入口容器存在但无「安全验证」按钮，无法触发！")
            return False

        log.info("点击「安全验证」...")
        d.xpath(SAFE_VERIFY).click()

        # ── 等待弹窗（等不到则检测中间页面）──
        if not _enter_captcha_dialog(d):
            return False
    else:
        # 兜底：intelVerify 找不到时，通过安全验证文本检测（≥2 命中即判定验证码存在）
        if _is_captcha_detected_by_safe_text(d):
            log.info("兜底检测：安全验证文本命中，确认验证码存在")
            # 先点击「安全验证」再等弹窗
            if d.xpath(SAFE_VERIFY).exists:
                log.info("点击「安全验证」...")
                d.xpath(SAFE_VERIFY).click()
            if not _enter_captcha_dialog(d):
                return False
        else:
            return False

    # ── 第四步：弹窗内循环尝试（失败刷新仍在弹窗内，类型可能变）──
    for attempt in range(5):
        # 弹窗不在 → 检查是否跳到了中间页面/安全验证入口
        if not _is_captcha_dialog_visible(d):
            if _is_on_intermediate_verify_page(d):
                log.info(f"第 {attempt+1} 次重试: 检测到中间页面，点击「前往验证」...")
                d.xpath(GO_TO_VERIFY).click()
                time.sleep(1.5)
                if not _wait_for_captcha_dialog(d):
                    log.warning("重新导航失败")
                    return False
            elif d.xpath(SAFE_VERIFY).exists:
                log.info(f"第 {attempt+1} 次重试: 检测到安全验证入口，重新等待弹窗...")
                if not _wait_for_captcha_dialog(d):
                    log.warning("重新导航失败")
                    return False
            else:
                log.info("弹窗消失，验证码已解开 ✓")
                return True

        # 每次重试重新判断类型（刷新后类型可能变）
        if d.xpath(CALC_PROMPT).exists:
            log.info(f"第 {attempt+1} 次尝试 → 计算题")
            if solve_calc(d):
                return True
        else:
            log.info(f"第 {attempt+1} 次尝试 → 滑块")
            if _solve_slider(d):
                return True

        log.info(f"第 {attempt+1} 次失败，等待验证码刷新...")
        time.sleep(2)

    log.warning("重试 5 次仍未解开")
    return False


def _solve_slider(d):
    """滑块验证码核心逻辑，成功返回 True"""
    slider_bounds = None
    for _ in range(5):
        slider_bounds = _find_slider(d)
        if slider_bounds:
            break
        time.sleep(1)

    if not slider_bounds:
        log.warning("未找到滑块元素")
        return False

    sx = (slider_bounds['left'] + slider_bounds['right']) // 2
    sy = (slider_bounds['top'] + slider_bounds['bottom']) // 2

    # ── 关键：先按住滑块2秒，滑块图会变 ──
    log.info(f"按住滑块 ({sx},{sy})")
    d.touch.down(sx, sy)
    time.sleep(random.uniform(1.5, 2))

    # ── 按住状态下截图 → 云码 ──
    b64 = _screenshot_to_b64(d, crop=False)

    resp = requests.post(YUNMA_URL, json={
        "token": YUNMA_TOKEN, "type": YUNMA_TYPE, "image": b64
    }, headers={"Content-Type": "application/json"}, timeout=30).json()
    log.info(f"云码: {resp}")

    if resp.get("msg") != "识别成功":
        log.warning(f"云码识别失败: {resp.get('msg')}")
        d.touch.up(sx, sy)
        return False

    dist = resp.get("data", {}).get("data", 0)
    dist = float(dist or 0) * DIST_SCALE +random.uniform(-3.1, 3.1)
    if not dist:
        log.warning("距离为0")
        d.touch.up(sx, sy)
        return False

    target_x = sx + int(dist)
    log.info(f"拖动 {dist:.0f}px 从 ({sx},{sy}) 到 ({target_x},{sy})")

    # ── 模拟真人轨迹：多点、变速、微小抖动 ──
    steps = [
        (sx + int(dist * 0.05),  sy + random.randint(-2, 2), random.uniform(0.05, 0.1)),
        (sx + int(dist * 0.20),  sy + random.randint(-2, 2), random.uniform(0.08, 0.15)),
        (sx + int(dist * 0.40),  sy + random.randint(-1, 1), random.uniform(0.1, 0.2)),
        (sx + int(dist * 0.60),  sy + random.randint(-1, 1), random.uniform(0.1, 0.2)),
        (sx + int(dist * 0.80),  sy + random.randint(-1, 1), random.uniform(0.08, 0.15)),
        (sx + int(dist * 0.95),  sy + random.randint(-1, 1), random.uniform(0.05, 0.1)),
        (target_x,               sy,                         0.3),
    ]
    for step_x, step_y, delay in steps:
        d.touch.move(step_x, step_y)
        time.sleep(delay)
    d.touch.up(target_x, sy)
    time.sleep(2)

    # 等待弹窗消失（最多等5秒，弹窗关闭有延迟）
    for _ in range(5):
        if not _is_captcha_dialog_visible(d):
            log.info("滑块验证通过 ✓")
            return True
        log.info("等待弹窗消失...")
        time.sleep(1)
    log.info("滑块验证未通过")
    return False


if __name__ == "__main__":
    if not DEVICE_ID or not YUNMA_TOKEN:
        log.error("先填 DEVICE_ID 和 YUNMA_TOKEN")
        exit(1)

    dev = connect(DEVICE_ID)
    log.info(f"已连接 {DEVICE_ID}")

    for i in range(MAX_ROUNDS):
        try:
            if solve(dev):
                log.info(f"第 {i+1} 轮完成")
            else:
                log.info(f"第 {i+1} 轮未检测到")
            time.sleep(3)
        except KeyboardInterrupt:
            break
        except Exception as e:
            log.error(f"异常: {e}")
            time.sleep(3)

    log.info("结束")