craw_spider/spiders/jd/jd_shop_info2.py

import random
import signal
import socket
import sys
import time
import base64
from DrissionPage import ChromiumPage, ChromiumOptions
import math
import requests
from DrissionPage.common import Actions

token = "zPzmt1mG1ouCU6GTzsZN2Lmm8pdZypapPcLJTBRETco"

chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"


class JdCrawlerV2:
    def __init__(self, drug_dict=None):
        self.driver = None
        self.register_signal_handler()

        self.ip = None
        self.account_name = None
        self.platform = 2

        self.task_dict = drug_dict or {}

        if self.task_dict:
            self.get_product_data()
        self.success = True
        self.is_no_prodcut = 0

    def get_product_data(self):
        self.task_id = self.task_dict["id"]
        self.company_id = self.task_dict["company_id"]
        self.product = self.task_dict["product_name"]
        self.product_desc = self.task_dict.get("product_specs", "")
        self.brand = self.task_dict.get("product_brand", "")
        self.product_keyword = self.task_dict.get("product_keyword", "")
        self.collect_task_id = self.task_dict.get("collect_task_id", "")

    @staticmethod
    def _get_free_port():
        """获取一个当前可用的本地端口，供 Chrome 调试使用。"""
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(("127.0.0.1", 0))
            return s.getsockname()[1]

    def init_browser(self):
        co = ChromiumOptions().set_browser_path(chrome_path)
        debug_port = self._get_free_port()
        co.set_user_data_path(f"./{self.account_name}")

        co.set_local_port(debug_port)
        co.set_argument(f"--remote-debugging-port={debug_port}")
        co.set_argument("--remote-debugging-address=127.0.0.1")
        # co.set_argument("--disable-blink-features=AutomationControlled")
        co.set_argument("--disable-dev-shm-usage")
        co.set_argument("--no-first-run")  # 避免首次运行弹窗
        co.set_argument("--no-default-browser-check")  # 避免默认浏览器检查

        if self.ip:
            proxy = self.ip.strip()
            if not proxy.startswith(("http://", "https://")):
                proxy = f"http://{proxy}"
            co.set_argument(f"--proxy-server={proxy}")
        self.driver = ChromiumPage(co)
        self.driver.listen.start("api?appid=search-pc-java")

    def register_signal_handler(self):
        def handler(signum, frame):
            print("\n⚠️ 程序退出")
            if self.driver:
                self.driver.quit()
            sys.exit(0)

        signal.signal(signal.SIGINT, handler)
        if hasattr(signal, "SIGTERM"):
            signal.signal(signal.SIGTERM, handler)

    def get_shop(self):
        url = "https://mall.jd.com/showLicence-4fc010bb739186871c97fe8159fdb58e68030b5168522fc2aa8be6dedfec0d63.html"

        self.driver.get(url, timeout=10)
        time.sleep(5)
        print("为滑块验证码")
        for i in range(2):
            capt_ele = self.driver.ele('xpath://img[@id="main_img"]', timeout=2)
            if not capt_ele:
                print("未找到验证码主图，可能已通过验证或页面未加载完成")
                break
            capt_ele.get_screenshot('./element_screenshot.png')

            distance = self.verify(2)
            try:
                distance = float(distance)
            except (TypeError, ValueError):
                print(f"滑块距离格式异常：{distance}")
                continue
            print(f"滑块距离：{distance}")

            # 获取滑块元素

            slider = self.driver.ele(
                "xpath://img[@class='move-img']",
                timeout=2
            )

            if not slider:
                print("未找到滑块")
                return

            start_x, start_y = slider.rect.midpoint

            start_x += random.uniform(-1, 1)
            start_y += random.uniform(-1, 1)

            end_x = (
                    start_x +
                    distance +

                    random.uniform(-3, 3)
            )

            end_y = start_y+ random.uniform(-1, 1)

            self.human_slide(
                start_x,
                start_y,
                end_x,
                end_y
            )
            # self.swipe(start_x, start_y, end_x, end_y,
            #                   duration=random.uniform(1.2, 2.0),
            #                   deviation=random.randint(20, 40))
            # self.human_slide(start_x, start_y, end_x, end_y)

        time.sleep(100)

            # self.simulate_slider_drag(slider_element, float(distance)-1.5)
            # # 滑块验证处理
            #
            # time.sleep(5)
            # capt_ele = self.driver.ele('xpath://*[@id="captcha_modal"]', timeout=2)
            # if not capt_ele:
            #     break
            # time.sleep(5)

    def verify(self, type_num):
        """调用云码平台服务"""
        with open('element_screenshot.png', 'rb') as f:
            b = base64.b64encode(f.read()).decode()

        url = "http://api.jfbym.com/api/YmServer/customApi"

        if type_num == 1:
            # 坐标类型
            data = {
                "token": token,
                "type": "30332",
                "direction": "top",
                "click_num": 3,
                "image": b,
            }
        else:
            # 滑块类型
            data = {
                "token": token,
                "type": "22222",
                "image": b,
            }

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

        response = requests.request("POST", url, headers=_headers, json=data, timeout=30).json()
        print(response)
        return response.get("data", {}).get("data")

    def human_slide(self, start_x, start_y, end_x, end_y):
        """
        更真实滑块拖动
        """
        actions = Actions(self.driver)

        points = []

        total_steps = random.randint(20, 30)

        distance_x = end_x - start_x
        distance_y = end_y - start_y
        total_distance = math.sqrt(distance_x ** 2 + distance_y ** 2)

        # 防止 randint 越界
        max_offset = max(2, min(
            5,
            int(total_distance * 0.01) + 1
        ))

        if random.random() < 0.7:
            offset_x = random.randint(1, max_offset)
        else:
            offset_x = -random.randint(1, 3)

        stop_x = end_x + offset_x
        stop_y = end_y

        accel_ratio = random.uniform(
            0.25,
            0.35
        )

        decel_ratio = random.uniform(
            0.25,
            0.35
        )

        points.append((start_x, start_y))

        for i in range(1, total_steps):

            t = i / (total_steps - 1)

            if t < accel_ratio:
                p = (t / accel_ratio) ** 3 * 0.3

            elif t < (1 - decel_ratio):

                mid_t = (
                                t - accel_ratio
                        ) / (1 - accel_ratio - decel_ratio)

                p = 0.3 + mid_t * 0.5

            else:
                end_t = (
                                t - (1 - decel_ratio)
                        ) / decel_ratio

                p = 0.8 + (1 - (1 - end_t) ** 3) * 0.2

            jitter_x = random.randint(-1, 1)
            jitter_y = random.randint(-1, 1)

            x = start_x + (stop_x - start_x) * p + jitter_x
            y = start_y + (stop_y - start_y) * p + jitter_y

            if x < points[-1][0]:
                x = points[-1][0]

            points.append((x, y))

        points[-1] = (stop_x, stop_y)

        print("开始拖动")

        # 按住滑块
        # 先移动到滑块元素
        slider = self.driver.ele(
            "xpath://img[@class='move-img']"
        )

        actions = Actions(self.driver)

        actions.move_to(slider).hold()

        last_x, last_y = points[0]

        for x, y in points[1:]:
            dx = x - last_x
            dy = y - last_y

            actions.move(
                dx,
                dy
            )

            time.sleep(
                random.uniform(
                    0.005,
                    0.02
                )
            )

            last_x, last_y = x, y

        actions.release()

        time.sleep(
            random.uniform(
                1,
                2
            )
        )

    def run(self):
        try:
            self.init_browser()
            self.get_shop()
        except Exception as e:
            self.success = False
            print(f"运行异常: {e}")


        finally:
            if self.driver:
                self.driver.quit()
                self.driver = None


if __name__ == '__main__':
    JdCrawlerV2().run()