craw_spider/ceshi/jd_shop_info.py

import random
import re
import signal
import socket
import sys
import time
import base64
from DrissionPage import ChromiumPage, ChromiumOptions
import json

import requests

from PIL import Image

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/index-10305746.html?from=pc"
        #
        # self.driver.get(url, timeout=10)
        # time.sleep(3)
        # hover_ele = self.driver.ele("xpath=//div[@class='j-shopHeader']//div[@class='jLogo']")
        # if not hover_ele:
        #     logger.error("未找到店铺 Logo，无法执行悬浮操作")
        #     return
        # hover_ele.hover()
        # time.sleep(1.5)
        #
        # # 先在主文档中找“营业执照”
        # target_ele = self.driver.ele(
        #     "xpath=//a[contains(@title,'营业执照') or contains(normalize-space(text()),'营业执照')]",
        #     timeout=5,
        # )
        #
        #
        # if not target_ele:
        #     logger.error("悬浮后仍未找到“营业执照”链接")
        #     return
        #
        # try:
        #     target_ele.scroll.to_see()
        # except Exception:
        #     pass
        #
        # try:
        #     target_ele.click()
        # except Exception:
        #     # 回退到 JS 点击，避免被遮挡导致常规点击失败
        #     target_ele.click(by_js=True)
        # logger.info("已点击“营业执照”链接")
        # time.sleep(10)
        url = "https://mall.jd.com/showLicence-4fc010bb739186871c97fe8159fdb58e68030b5168522fc2aa8be6dedfec0d63.html"

        self.driver.get(url, timeout=10)
        time.sleep(2)
        print("为滑块验证码")
        for i in range(3):
            capt_ele = self.driver.ele('xpath://img[@id="main_img"]', timeout=2)
            capt_ele.get_screenshot('./element_screenshot.png')

            distance = self.verify(2)
            print(f"滑块距离：{distance}")

            slider_element = self.driver.ele(
                "xpath://img[@class='move-img']")

            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).json()
        print(response)
        return response["data"]["data"]

    def generate_human_track(self, distance):
        """
        生成人类拖动的轨迹
        :param distance: 需要拖动的距离（像素）
        :return: 轨迹点列表，每个点包含(x偏移, y偏移, 延迟时间)
        """
        tracks = []
        current = 0
        mid = distance * 0.7  # 70%处开始减速
        t = 0.2
        v = 0
        move_points = []

        # 第一阶段：加速
        while current < mid:
            a = random.uniform(2, 4)
            v0 = v
            v = v0 + a * t
            move = v0 * t + 0.5 * a * t * t
            current += move
            move_points.append(move)

        # 第二阶段：减速
        while current < distance:
            a = -random.uniform(0.5, 1.5)
            v0 = v
            v = v0 + a * t
            if v < 0.5:  # 防止速度过小
                v = 0.5
            move = v0 * t + 0.5 * a * t * t
            current += move
            move_points.append(move)

        # 添加随机性并生成最终轨迹
        total_points = len(move_points)
        for i, move in enumerate(move_points):
            x_offset = move

            # 添加垂直抖动（模拟手抖）
            if i % random.randint(2, 4) == 0:
                y_offset = random.randint(-2, 2)
            else:
                y_offset = 0

            # 时间间隔（模拟人类反应）
            if i < total_points * 0.3:  # 开始阶段较快
                duration = random.uniform(0.01, 0.03)
            elif i > total_points * 0.7:  # 结束阶段较慢
                duration = random.uniform(0.03, 0.08)
            else:  # 中间阶段
                duration = random.uniform(0.02, 0.05)

            # 随机添加微小停顿
            if random.random() < 0.05:
                duration += random.uniform(0.05, 0.1)

            tracks.append((x_offset, y_offset, duration))

        # 最终微调：到达终点后轻微回拉
        if random.random() < 0.7:
            tracks.append((-random.randint(1, 3), 0, 0.05))

        return tracks

    def simulate_slider_drag(self, slider_element, target_distance):
        """
        模拟人类拖动滑块
        """
        # 移动到滑块并按住
        self.driver.actions.move_to(slider_element).hold()
        # 生成轨迹
        tracks = self.generate_human_track(target_distance)
        # 按轨迹拖动
        for track in tracks:
            offset_x, offset_y, duration = track
            self.driver.actions.move(offset_x, offset_y, duration=duration / 1000)

        time.sleep(0.8)
        # 释放鼠标
        self.driver.actions.release()

    def run(self):
        try:
            self.init_browser()
            self.get_shop()
        except Exception as e:
            self.success = False


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


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