craw_spider/spiders/yaoex/login_yaoex.py

import base64
import random
import signal
import socket
import sys
import time

import requests
from DrissionPage import ChromiumPage, ChromiumOptions

from commons.Logger import logger
from oss_upload.oss_upload import AliyunOSSUploader

chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
CAPTCHA_TOKEN = "12445"
CAPTCHA_API_URL = "http://api.jfbym.com/api/YmServer/customApi"
SLIDER_OFFSET_FIX = 10


class YaoExCrawl:
    def __init__(self, product=None):
        self.product = product
        self.driver = None
        self.account_name = "yiyaocheng_1"
        self.ossuploader = AliyunOSSUploader()
        self._register_signal_handler()

    def _register_signal_handler(self):
        def handler(signum, frame):
            logger.info("收到退出信号，准备关闭浏览器")
            self._quit_browser()
            sys.exit(0)

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

    def _quit_browser(self):
        if self.driver:
            try:
                self.driver.quit()
            except Exception:
                pass
            self.driver = None

    @staticmethod
    def _get_free_port():
        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-dev-shm-usage")
        co.set_argument("--no-first-run")
        co.set_argument("--no-default-browser-check")
        co.set_user_agent(
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
            "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
        )
        co.headless(False)
        self.driver = ChromiumPage(co)

    def _is_logged_in(self):
        title = self.driver.ele(
            "xpath=//*[contains(text(),'云南靓桐医药有限公司蒙自益寿大药房二店')]",
            timeout=5,
        )
        return bool(title)

    def _call_captcha_api(self, image_bytes):
        try:
            b64 = base64.b64encode(image_bytes).decode()
            resp = requests.post(
                CAPTCHA_API_URL,
                json={"token": CAPTCHA_TOKEN, "type": "22222", "image": b64},
                headers={"Content-Type": "application/json"},
                timeout=15,
            ).json()
            logger.info("验证码 API 返回: %s", resp)
            return resp["data"]["data"]
        except Exception as e:
            logger.exception("验证码识别失败: %s", e)
            return None

    @staticmethod
    def _generate_human_track(distance):
        tracks = []
        current = 0
        mid = distance * 0.7
        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):
            y_offset = random.randint(-2, 2) if i % random.randint(2, 4) == 0 else 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((move, 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()
        for offset_x, offset_y, duration in self._generate_human_track(target_distance):
            self.driver.actions.move(offset_x, offset_y, duration=duration / 1000)
        self.driver.actions.release()

    def _solve_slider_if_present(self):
        modal = self.driver.ele("xpath://div[@class='yidun_modal']", timeout=3)
        if not modal:
            return True

        logger.info("检测到滑块验证码，开始处理")
        jpg_bytes = modal.get_screenshot(as_bytes="jpg")
        distance = self._call_captcha_api(jpg_bytes)
        if distance is None:
            return False

        slider = self.driver.ele(
            "xpath://div[contains(@class,'yidun_slider--hover')]", timeout=5
        )
        if not slider:
            logger.error("未找到滑块元素")
            return False
        self._simulate_slider_drag(slider, float(distance) + SLIDER_OFFSET_FIX)
        time.sleep(3)
        return True

    def login(self):
        self.driver.get("https://mall.yaoex.com/login", timeout=15)
        self.driver.wait.doc_loaded(timeout=10)

        input_name = self.driver.ele("xpath://input[@name='username']", timeout=5)
        if not input_name:
            logger.error("未找到用户名输入框")
            return False
        input_name.input("18687653982")
        time.sleep(random.uniform(1.2, 2.0))

        input_pass = self.driver.ele("xpath://input[@name='password']", timeout=5)
        if not input_pass:
            logger.error("未找到密码输入框")
            return False
        input_pass.input("liu198810060814")
        time.sleep(random.uniform(1.2, 2.0))

        geetest_click = self.driver.ele(
            "xpath=//div[contains(@class,'geetest_btn_click')]", timeout=3
        )
        if geetest_click:
            geetest_click.click()
            time.sleep(1.5)

        login_button = self.driver.ele("xpath://input[@id='login-btn']", timeout=5)
        if not login_button:
            logger.error("未找到登录按钮")
            return False
        login_button.click()
        self.driver.wait.doc_loaded(timeout=10)
        time.sleep(2)

        if not self._solve_slider_if_present():
            return False

        return self._is_logged_in()

    def get_snapshot(self):
        detail_url = "https://mall.yaoex.com/v2/product/#/spuCode/2918544090/sellerCode/8353"
        item_id = "2918544090"
        self.driver.get(detail_url, timeout=10)
        self.driver.wait.doc_loaded(timeout=10)
        time.sleep(3)
        ele = self.driver.ele("xpath=//div[@class='yaoex-product-detail__content']", timeout=8)
        if not ele:
            logger.warning("未找到详情截图区域")
            return ""

        jpg_bytes = ele.get_screenshot(as_bytes="jpg")
        img_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(item_id))
        logger.info("截图上传完成: %s", img_url)
        time.sleep(random.uniform(0.5, 1.2))
        return img_url

    def search(self):
        self.driver.get("https://mall.yaoex.com/", timeout=15)
        self.driver.wait.doc_loaded(timeout=10)
        if not self._is_logged_in():
            if not self.login():
                logger.error("登录失败")
                return False
        self.get_snapshot()
        return True

    def run(self):
        try:
            self.init_browser()
            self.search()
        except Exception as e:
            logger.exception("爬取异常: %s", e)
            time.sleep(3)
        finally:
            self._quit_browser()


if __name__ == "__main__":
    YaoExCrawl().run()