buff1.0/云码验证码识别_综合版_Playwright.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
BUFF验证码自动识别器 - 使用Playwright
支持两种验证码类型：
1. 顺序点击验证码（type=317）
2. 文字特征点选验证码（type=94）

使用方法：
1. 安装依赖: pip install playwright && playwright install chromium
2. 运行脚本: python 云码验证码识别_综合版.py
"""

import base64
import json
import time
import sys
from playwright.sync_api import sync_playwright
from playwright_stealth import stealth
import requests


class BuffCaptchaSolver:
    """BUFF验证码综合识别器"""

    def __init__(self, token: str):
        """
        初始化
        
        Args:
            token: 云码平台API Token
        """
        self.token = token
        self.api_url = "http://api.jfbym.com/api/YmServer/customApi"

    def detect_captcha_type(self, page) -> str:
        """
        检测验证码类型
        
        Args:
            page: Playwright Page对象
            
        Returns:
            str: 'icon_point' - 顺序点击验证码
                 'space' - 文字特征点选验证码
                 None - 未检测到验证码
        """
        try:
            # 等待验证码容器出现
            page.wait_for_selector(".yidun", timeout=10000)

            # 方法1：检查CSS类名（最可靠）
            class_name = page.evaluate("""
                () => {
                    const el = document.querySelector('.yidun');
                    return el ? el.className : '';
                }
            """)
            
            if "yidun--icon_point" in class_name:
                return "icon_point"
            elif "yidun--space" in class_name:
                return "space"

            # 方法2：检查提示文字
            try:
                tips_text = page.evaluate("""
                    () => {
                        const el = document.querySelector('.yidun_tips__text');
                        return el ? el.textContent.trim() : '';
                    }
                """)
                
                if tips_text.startswith("请依次点击"):
                    return "icon_point"
                elif tips_text.startswith("请点击"):
                    return "space"
            except:
                pass

            return None

        except Exception as e:
            print(f"验证码检测失败: {e}")
            return None

    def extract_captcha_image(self, page) -> str:
        """
        提取验证码图片并转为Base64
        
        Args:
            page: Playwright Page对象
            
        Returns:
            str: Base64编码的图片数据
        """
        try:
            # 获取图片src
            img_src = page.evaluate("""
                () => {
                    const img = document.querySelector('.yidun_bg-img');
                    return img ? img.src : null;
                }
            """)
            
            if not img_src:
                print("❌ 未找到验证码图片")
                return None
            
            print(f"验证码图片地址: {img_src}")

            if img_src.startswith("data:image"):
                # 提取base64数据部分
                base64_data = img_src.split(",")[1]
                return base64_data
            elif img_src.startswith("http"):
                # 下载图片
                response = requests.get(img_src)
                if response.status_code == 200:
                    return base64.b64encode(response.content).decode()

            return None

        except Exception as e:
            print(f"提取验证码图片失败: {e}")
            return None

    def extract_tip_text(self, page) -> str:
        """
        提取验证码提示文字
        
        Args:
            page: Playwright Page对象
            
        Returns:
            str: 提示文字
        """
        try:
            tip_text = page.evaluate("""
                () => {
                    const el = document.querySelector('.yidun_tips__text');
                    return el ? el.textContent.trim() : '';
                }
            """)
            print(f"验证码提示文字: {tip_text}")
            return tip_text
        except:
            return ""

    def call_api(self, image_base64: str, captcha_type: str, tip_text: str = "") -> dict:
        """
        调用云码API
        
        Args:
            image_base64: Base64编码的图片数据
            captcha_type: 验证码类型
            tip_text: 提示文字
            
        Returns:
            dict: API响应结果
        """
        # 根据类型选择接口ID
        type_ids = {
            "icon_point": "317",
            "space": "94"
        }

        type_id = type_ids.get(captcha_type)
        if not type_id:
            return {"code": -1, "msg": "未知的验证码类型"}

        payload = {
            "token": self.token,
            "type": type_id,
            "image": image_base64
        }

        # 文字点选类型可能需要传递提示文字
        if captcha_type == "space" and tip_text:
            payload["extra"] = tip_text

        headers = {"Content-Type": "application/json"}

        try:
            response = requests.post(
                self.api_url,
                json=payload,
                headers=headers,
                timeout=30
            )
            return response.json()

        except Exception as e:
            print(f"API调用失败: {e}")
            return {"code": -1, "msg": str(e)}

    def parse_result(self, api_result: dict) -> list:
        """
        解析API返回结果
        
        Args:
            api_result: API响应结果
            
        Returns:
            list: 坐标列表
        """
        try:
            if api_result.get("code") == 10000:
                data = api_result.get("data", [])
                if data:
                    result_str = data[0].get("data", "")
                    print(f"云码返回坐标: {result_str}")
                    
                    if result_str.startswith("[") and result_str.endswith("]"):
                        return json.loads(result_str)
                    else:
                        try:
                            return eval(result_str)
                        except:
                            print(f"无法解析返回数据: {result_str}")
                            return []
            else:
                print(f"识别失败: {api_result.get('msg')}")
                return []

        except Exception as e:
            print(f"解析结果失败: {e}")
            return []

    def click_coordinates(self, page, coordinates: list):
        """
        执行点击操作
        
        Args:
            page: Playwright Page对象
            coordinates: 坐标列表
        """
        try:
            # 获取验证码图片元素
            img_element = page.query_selector(".yidun_bg-img")
            box = img_element.bounding_box()
            
            if not box:
                print("❌ 无法获取验证码图片位置")
                return False

            # 执行点击
            for i, coord in enumerate(coordinates):
                x, y = coord
                
                # 计算相对于图片的绝对坐标
                abs_x = box["x"] + x
                abs_y = box["y"] + y
                
                # 移动并点击
                page.mouse.click(abs_x, abs_y)
                print(f"点击坐标: ({abs_x}, {abs_y})")
                
                # 每次点击后等待
                if i < len(coordinates) - 1:
                    time.sleep(0.3)

            print(f"✅ 完成点击，共点击 {len(coordinates)} 次")
            return True

        except Exception as e:
            print(f"点击操作失败: {e}")
            return False

    def solve(self, page) -> bool:
        """
        完整的验证码解决流程
        
        Args:
            page: Playwright Page对象
            
        Returns:
            bool: 解决成功返回True
        """
        # 1. 检测验证码类型
        captcha_type = self.detect_captcha_type(page)

        if captcha_type == "icon_point":
            type_name = "顺序点击验证码（第一种）"
            type_id = "317"
        elif captcha_type == "space":
            type_name = "文字特征点选验证码（第二种）"
            type_id = "94"
        else:
            print("❌ 无法识别验证码类型或未检测到验证码")
            return False

        print("=" * 60)
        print(f"✅ 检测到 {type_name}")
        print(f"   云码接口ID: {type_id}")
        print("=" * 60)

        # 2. 提取验证码图片
        image_base64 = self.extract_captcha_image(page)
        if not image_base64:
            print("❌ 提取验证码图片失败")
            return False

        print("✅ 验证码图片提取成功")

        # 3. 提取提示文字（仅第二种需要）
        tip_text = ""
        if captcha_type == "space":
            tip_text = self.extract_tip_text(page)

        # 4. 调用云码API
        print("🔄 正在调用云码API识别...")
        api_result = self.call_api(image_base64, captcha_type, tip_text)

        if not api_result or api_result.get("code") != 10000:
            print(f"❌ API识别失败: {api_result}")
            return False

        print("✅ API识别成功")

        # 5. 解析结果
        coordinates = self.parse_result(api_result)
        if not coordinates:
            print("❌ 解析坐标失败")
            return False

        print(f"📍 获取到 {len(coordinates)} 个点击坐标")

        # 6. 执行点击
        if self.click_coordinates(page, coordinates):
            print("✅ 点击完成")
            return True
        else:
            print("❌ 点击失败")
            return False


def main():
    """主函数"""
    # 配置
    TOKEN = "wVRyAIjI3hhJBicXndcs1ro9BN5UNMXIjW6SDWBFFyQ"  # 云码Token
    TARGET_URL = "https://buff.163.com/goods/900482?from=market#tab=selling"

    # 创建识别器
    solver = BuffCaptchaSolver(TOKEN)

    with sync_playwright() as p:
        # 启动浏览器（无头模式=False，可以看到浏览器窗口）
        print("🚀 启动浏览器...")
        browser = p.chromium.launch(
            headless=False,  # False=显示浏览器窗口
            args=[
                '--no-sandbox',
                '--disable-setuid-sandbox',
                '--disable-dev-shm-usage',
                '--disable-gpu',
                '--window-size=1920,1080'
            ]
        )
        
        # 创建上下文
        context = browser.new_context(
            viewport={'width': 1920, 'height': 1080},
            user_agent='Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36'
        )
        
        # 创建页面
        page = context.new_page()
        
        # 应用stealth防止被检测
        stealth(page)
        
        try:
            # 访问目标URL
            print(f"📍 正在访问: {TARGET_URL}")
            page.goto(TARGET_URL)
            
            # 等待页面加载
            page.wait_for_load_state("networkidle")
            
            print("✅ 页面加载完成")
            
            # 等待验证码出现（最多30秒）
            print("🔍 等待验证码出现...")
            
            # 尝试等待验证码
            try:
                page.wait_for_selector(".yidun", timeout=30000)
                print("✅ 检测到验证码！")
            except:
                print("❌ 30秒内未检测到验证码")
                input("\n按Enter键关闭浏览器...")
                browser.close()
                return
            
            # 等待验证码完全加载
            time.sleep(2)
            
            # 自动识别并解决验证码
            if solver.solve(page):
                print("\n🎉 验证码解决成功！")
            else:
                print("\n❌ 验证码解决失败")

        except Exception as e:
            print(f"错误: {e}")
            import traceback
            traceback.print_exc()

        finally:
            input("\n按Enter键关闭浏览器...")
            browser.close()


if __name__ == "__main__":
    print("=" * 60)
    print("BUFF验证码自动识别器（Playwright版本）")
    print("=" * 60)
    main()