|
|
@@ -0,0 +1,556 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+"""
|
|
|
+BUFF验证码自动识别器 - 使用Playwright(简化版)
|
|
|
+支持两种验证码类型:
|
|
|
+1. 顺序点击验证码(type=317)
|
|
|
+2. 文字特征点选验证码(type=94)
|
|
|
+
|
|
|
+使用方法:
|
|
|
+1. 安装依赖: pip install playwright && playwright install chromium
|
|
|
+2. 运行脚本: python3 云码验证码识别_综合版_简化版.py
|
|
|
+"""
|
|
|
+
|
|
|
+import base64
|
|
|
+import json
|
|
|
+import time
|
|
|
+import sys
|
|
|
+from playwright.sync_api import sync_playwright
|
|
|
+import requests
|
|
|
+
|
|
|
+
|
|
|
+def apply_stealth(page):
|
|
|
+ """
|
|
|
+ 手动应用 stealth 脚本,防止被检测为自动化工具
|
|
|
+ """
|
|
|
+ page.add_init_script("""
|
|
|
+ Object.defineProperty(navigator, 'webdriver', {
|
|
|
+ get: () => undefined
|
|
|
+ });
|
|
|
+ Object.defineProperty(navigator, 'languages', {
|
|
|
+ get: () => ['zh-CN', 'zh', 'en-US', 'en']
|
|
|
+ });
|
|
|
+ Object.defineProperty(navigator, 'plugins', {
|
|
|
+ get: () => [1, 2, 3, 4, 5]
|
|
|
+ });
|
|
|
+ window.navigator.chrome = {
|
|
|
+ runtime: {},
|
|
|
+ loadTimes: function() {},
|
|
|
+ csi: function() {},
|
|
|
+ app: {}
|
|
|
+ };
|
|
|
+ Object.defineProperty(window, 'chrome', {
|
|
|
+ get: () => window.navigator.chrome
|
|
|
+ });
|
|
|
+ Object.defineProperty(navigator, 'platform', {
|
|
|
+ get: () => 'MacIntel'
|
|
|
+ });
|
|
|
+ Object.defineProperty(navigator, 'deviceMemory', {
|
|
|
+ get: () => 8
|
|
|
+ });
|
|
|
+ Object.defineProperty(navigator, 'hardwareConcurrency', {
|
|
|
+ get: () => 8
|
|
|
+ });
|
|
|
+ """)
|
|
|
+
|
|
|
+
|
|
|
+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 : '';
|
|
|
+ }
|
|
|
+ """)
|
|
|
+
|
|
|
+ print(f"验证码容器class: {class_name}")
|
|
|
+
|
|
|
+ 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() : '';
|
|
|
+ }
|
|
|
+ """)
|
|
|
+
|
|
|
+ print(f"验证码提示文字: {tips_text}")
|
|
|
+
|
|
|
+ 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[:100]}...")
|
|
|
+
|
|
|
+ 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": "30332", # 顺序点击验证码(第一种)- 根据云码文档
|
|
|
+ "space": "30109" # 文字特征点选验证码(第二种)- 根据云码文档
|
|
|
+ }
|
|
|
+
|
|
|
+ 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
|
|
|
+ }
|
|
|
+
|
|
|
+ # 第一种验证码(通用图标点选)- 需要 direction 参数
|
|
|
+ if captcha_type == "icon_point":
|
|
|
+ # 根据提示文字判断提示信息位置
|
|
|
+ # "按顺序点击下图中的" - 通常在图片下方显示提示
|
|
|
+ if tip_text:
|
|
|
+ if "上方" in tip_text or "上面" in tip_text:
|
|
|
+ payload["direction"] = "top"
|
|
|
+ else:
|
|
|
+ payload["direction"] = "bottom"
|
|
|
+ else:
|
|
|
+ payload["direction"] = "bottom" # 默认
|
|
|
+ payload["click_num"] = "4" # 默认4个坐标
|
|
|
+
|
|
|
+ # 第二种验证码(文字特征点选)- 需要 extra 参数
|
|
|
+ elif 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: 坐标列表 [[x1, y1], [x2, y2], ...]
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ print(f"API完整返回: {api_result}")
|
|
|
+
|
|
|
+ if api_result.get("code") == 10000:
|
|
|
+ data = api_result.get("data", {})
|
|
|
+ print(f"data字段: {data}, 类型: {type(data)}")
|
|
|
+
|
|
|
+ # 新的返回格式:data 是字典
|
|
|
+ if isinstance(data, dict):
|
|
|
+ inner_code = data.get("code", 0)
|
|
|
+ result_str = data.get("data", "")
|
|
|
+ print(f"inner_code: {inner_code}, result_str: {result_str}")
|
|
|
+
|
|
|
+ if inner_code == 0 and result_str:
|
|
|
+ # 格式1: JSON数组 "[x1,y1,x2,y2,x3,y3,x4,y4]"
|
|
|
+ if isinstance(result_str, str) and result_str.startswith("["):
|
|
|
+ try:
|
|
|
+ coords = json.loads(result_str)
|
|
|
+ # 转换为一维数组为二维坐标数组
|
|
|
+ if isinstance(coords, list) and len(coords) >= 2:
|
|
|
+ if all(isinstance(c, (int, float)) for c in coords):
|
|
|
+ # 一维数组 [x1,y1,x2,y2,...] 转为 [[x1,y1],[x2,y2],...]
|
|
|
+ result = []
|
|
|
+ for i in range(0, len(coords) - 1, 2):
|
|
|
+ if i + 1 < len(coords):
|
|
|
+ result.append([int(coords[i]), int(coords[i + 1])])
|
|
|
+ if result:
|
|
|
+ print(f"解析到 {len(result)} 个坐标: {result}")
|
|
|
+ return result
|
|
|
+ except json.JSONDecodeError:
|
|
|
+ pass
|
|
|
+
|
|
|
+ # 格式2: 逗号分隔 "x1,y1,x2,y2,x3,y3,x4,y4"
|
|
|
+ if isinstance(result_str, str) and "," in result_str:
|
|
|
+ parts = result_str.replace(" ", "").split(",")
|
|
|
+ if len(parts) >= 2:
|
|
|
+ try:
|
|
|
+ coords = [int(p) for p in parts if p.strip().isdigit() or (p.strip().lstrip('-').isdigit())]
|
|
|
+ if len(coords) >= 2:
|
|
|
+ result = []
|
|
|
+ for i in range(0, len(coords) - 1, 2):
|
|
|
+ if i + 1 < len(coords):
|
|
|
+ result.append([coords[i], coords[i + 1]])
|
|
|
+ if result:
|
|
|
+ print(f"解析到 {len(result)} 个坐标: {result}")
|
|
|
+ return result
|
|
|
+ except ValueError:
|
|
|
+ pass
|
|
|
+
|
|
|
+ # 格式3: 单个坐标 "156,61"
|
|
|
+ if isinstance(result_str, str) and result_str.count(",") == 1:
|
|
|
+ parts = result_str.split(",")
|
|
|
+ if len(parts) == 2:
|
|
|
+ try:
|
|
|
+ x, y = int(parts[0].strip()), int(parts[1].strip())
|
|
|
+ print(f"解析到单个坐标: [{x}, {y}]")
|
|
|
+ return [[x, y]]
|
|
|
+ except ValueError:
|
|
|
+ pass
|
|
|
+
|
|
|
+ # 旧的返回格式:data 是列表
|
|
|
+ elif isinstance(data, list) and data:
|
|
|
+ result_str = data[0].get("data", "")
|
|
|
+ print(f"result_str: {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}")
|
|
|
+ import traceback
|
|
|
+ traceback.print_exc()
|
|
|
+ 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
|
|
|
+
|
|
|
+ print(f"验证码图片位置: x={box['x']}, y={box['y']}, width={box['width']}, height={box['height']}")
|
|
|
+
|
|
|
+ # 执行点击
|
|
|
+ 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"点击坐标 ({i+1}): ({abs_x}, {abs_y})")
|
|
|
+
|
|
|
+ # 每次点击后等待
|
|
|
+ if i < len(coordinates) - 1:
|
|
|
+ time.sleep(0.5)
|
|
|
+
|
|
|
+ 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 = "30332" # 根据云码文档
|
|
|
+ elif captcha_type == "space":
|
|
|
+ type_name = "文字特征点选验证码(第二种)"
|
|
|
+ type_id = "30109" # 根据云码文档
|
|
|
+ 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)
|
|
|
+
|
|
|
+ print("=" * 60)
|
|
|
+ print("BUFF验证码自动识别器(Playwright简化版)")
|
|
|
+ print("=" * 60)
|
|
|
+
|
|
|
+ with sync_playwright() as p:
|
|
|
+ # 启动浏览器(无痕模式)
|
|
|
+ print("启动浏览器(无痕模式)...")
|
|
|
+
|
|
|
+ # 使用 launch_persistent_context 启动真正的无痕模式
|
|
|
+ context = p.chromium.launch_persistent_context(
|
|
|
+ user_data_dir="", # 空目录 = 临时无痕模式
|
|
|
+ headless=False, # False=显示浏览器窗口,方便调试
|
|
|
+ viewport={'width': 1920, 'height': 1080},
|
|
|
+ user_agent='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',
|
|
|
+ locale='zh-CN',
|
|
|
+ args=[
|
|
|
+ '--no-sandbox',
|
|
|
+ '--disable-setuid-sandbox',
|
|
|
+ '--disable-dev-shm-usage',
|
|
|
+ '--disable-gpu',
|
|
|
+ '--window-size=1920,1080',
|
|
|
+ '--start-maximized'
|
|
|
+ ]
|
|
|
+ )
|
|
|
+
|
|
|
+ # 获取页面
|
|
|
+ page = context.pages[0] if context.pages else context.new_page()
|
|
|
+
|
|
|
+ # 应用stealth防止被检测
|
|
|
+ apply_stealth(page)
|
|
|
+
|
|
|
+ try:
|
|
|
+ # 访问目标URL
|
|
|
+ print(f"正在访问: {TARGET_URL}")
|
|
|
+ page.goto(TARGET_URL)
|
|
|
+
|
|
|
+ # 等待页面加载
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+
|
|
|
+ print("页面加载完成")
|
|
|
+
|
|
|
+ # 设置session cookie防止验证码
|
|
|
+ print("设置session cookie...")
|
|
|
+ context.add_cookies([
|
|
|
+ {
|
|
|
+ "name": "session",
|
|
|
+ "value": "1-29ene-taFfBv6_EK2btogRrVsqsQsuuJooUGnxBDUwLI2043282888",
|
|
|
+ "domain": ".buff.163.com",
|
|
|
+ "path": "/"
|
|
|
+ }
|
|
|
+ ])
|
|
|
+ print("session cookie设置成功")
|
|
|
+
|
|
|
+ # 刷新页面使cookie生效
|
|
|
+ page.reload()
|
|
|
+ page.wait_for_load_state("networkidle")
|
|
|
+ print("页面已刷新")
|
|
|
+
|
|
|
+ # 等待验证码出现(最多30秒)
|
|
|
+ print("等待验证码出现...")
|
|
|
+
|
|
|
+ # 尝试等待验证码
|
|
|
+ try:
|
|
|
+ page.wait_for_selector(".yidun", timeout=30000)
|
|
|
+ print("检测到验证码!")
|
|
|
+ except:
|
|
|
+ print("30秒内未检测到验证码")
|
|
|
+ print("可能的原因:")
|
|
|
+ print("1. 页面没有触发验证码")
|
|
|
+ print("2. 验证码已经被手动处理")
|
|
|
+ print("3. 页面结构已变化")
|
|
|
+ input("\n按Enter键关闭浏览器...")
|
|
|
+ context.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__":
|
|
|
+ main()
|