craw_spider/spiders/yaoshibang/ysb_snapshot_crawl_bak.py

import random
import signal
import socket
import sys
import time
import base64
import math
import requests
from commons.conn_mysql import MySQLPoolOnline
from DrissionPage import ChromiumPage, ChromiumOptions
from commons.Logger import logger
from oss_upload.oss_upload import AliyunOSSUploader
from commons.config import YSB_ACCOUNT

CAPTCHA_TOKEN = "zPzmt1mG1ouCU6GTzsZN2Lmm8pdZypapPcLJTBRETco"
CAPTCHA_API_URL = "http://api.jfbym.com/api/YmServer/customApi"

SLIDER_OFFSET_FIX = 10

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


class YaoShiBangSnapshot:
    def __init__(self, product=None):
        self.product = product
        self.driver = None
        self.account_name = "ysbang_1"
        self.platform = 5
        self.db_online = MySQLPoolOnline()
        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():
        """获取一个当前可用的本地端口，供 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"./spiders/yaoshibang/{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("--start-maximized")
        co.set_argument("--no-first-run")  # 避免首次运行弹窗
        co.set_argument("--no-default-browser-check")  # 避免默认浏览器检查
        self.driver = ChromiumPage(co)

    def _solve_slider_captcha(self):
        """检测并处理易盾滑块验证码，成功返回 True。"""
        self.driver.wait.doc_loaded()
        time.sleep(2)

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

        logger.info("检测到滑块验证码，开始处理")
        jpg_bytes = yidun.get_screenshot(as_bytes="jpg")

        distance = self._call_captcha_api(jpg_bytes)
        if distance is None:
            logger.error("验证码识别失败")
            return False

        logger.info("滑块距离: %s", distance)
        slider = self.driver.ele(
            "xpath://div[contains(@class,'yidun_slider--hover')]", timeout=5
        )
        if not slider:
            logger.error("未找到滑块元素")
            return False

        try:
            drag_distance = float(distance) + SLIDER_OFFSET_FIX
        except (TypeError, ValueError):
            logger.error("滑块距离非数字: %r", distance)
            return False

        if not math.isfinite(drag_distance) or drag_distance <= 0:
            logger.error("滑块距离无效: %s", drag_distance)
            return False
        self._simulate_slider_drag(slider, drag_distance)
        time.sleep(3)
        return True

    def _call_captcha_api(self, image_bytes):
        """调用云码平台识别滑块距离，失败返回 None。"""
        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)
            if not isinstance(resp, dict):
                return None
            data = resp.get("data")
            if isinstance(data, dict):
                dist = data.get("data")
            else:
                dist = data
            if dist is None:
                logger.error("验证码 API 未返回距离字段: %s", resp)
                return None
            try:
                d = float(dist)
            except (TypeError, ValueError):
                logger.error("验证码距离无法解析为数字: %r", dist)
                return None
            if not math.isfinite(d):
                logger.error("验证码距离非有限数值: %r", dist)
                return None
            return d
        except Exception as e:
            logger.exception("验证码 API 调用失败: %s", e)
            return None

    @staticmethod
    def _generate_human_track(distance):
        try:
            distance = float(distance)
        except (TypeError, ValueError):
            return []
        if distance <= 0 or not math.isfinite(distance):
            return []
        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):
        if target_distance <= 0:
            logger.warning("滑块目标距离无效: %s", target_distance)
            return
        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 _is_logged_in(self):
        # 与当前账号店铺展示文案一致；换店后需同步修改或改为配置项
        title = self.driver.ele(
            "xpath=//span[@class='logout']",
            timeout=5,
        )
        return bool(title)

    def login(self):
        logger.info("开始登录药师帮")
        self.driver.get("https://dian.ysbang.cn/#/login", timeout=15)
        self.driver.wait.doc_loaded(timeout=10)
        time.sleep(2)

        input_name = self.driver.ele("xpath://input[@name='userAccount']", timeout=5)
        if not input_name:
            logger.error("未找到账号输入框")
            return False
        input_name.input(YSB_ACCOUNT["account"])
        time.sleep(random.uniform(1.5, 2.5))

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

        login_btn = self.driver.ele("xpath://button[text()='登录']", timeout=5)
        if not login_btn:
            logger.error("未找到登录按钮")
            return False

        login_btn.click()
        time.sleep(3)
        for i in range(3):
            self._solve_slider_captcha()
            time.sleep(3)

            if self._is_logged_in():
                logger.info("登录成功")
                return True

        logger.error("登录后未检测到目标店铺名，登录可能失败")
        return False

    def get_snapshot(self, detail_url, row_id):
        self.driver.get(detail_url, timeout=15)
        self.driver.wait.doc_loaded(timeout=10)
        time.sleep(2)
        self._dismiss_popup_before_screenshot()

        try:
            # jpg_bytes = ele.get_screenshot(as_bytes="jpg")
            jpg_bytes = self.driver.get_screenshot(as_bytes="jpg")
            if not jpg_bytes:
                logger.warning("截图为空 row_id=%s", row_id)
                return ""
            img_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(row_id))
        except Exception:
            logger.exception("截图或 OSS 上传失败 row_id=%s url=%s", row_id, detail_url)
            return ""
        if not img_url:
            logger.warning("OSS 未返回有效地址 row_id=%s", row_id)
            return ""
        logger.info("截图上传完成 row_id=%s url=%s", row_id, img_url)
        time.sleep(random.uniform(2, 3))
        return img_url

    def _dismiss_popup_before_screenshot(self):
        """截图前关闭或隐藏营销弹窗，避免遮挡。"""
        close_locs = [
            "xpath=//div[contains(@class,'dialog')]//i[contains(@class,'close')]",
            "xpath=//div[contains(@class,'popup')]//i[contains(@class,'close')]",
            "xpath=//div[contains(@class,'modal')]//i[contains(@class,'close')]",
            "xpath=//button[contains(@class,'close')]",
            "xpath=//span[text()='×']",
            "xpath=//*[contains(text(),'智能采购')]/ancestor::div[1]//*[contains(@class,'close')]",
        ]
        for loc in close_locs:
            try:
                btn = self.driver.ele(loc, timeout=0.5)
                if btn:
                    btn.click()
                    time.sleep(0.2)
            except Exception:
                pass

        try:
            # 兜底：隐藏常见高层弹窗和遮罩
            self.driver.run_js(
                """
                const sels = [
                  '[class*="modal"]',
                  '[class*="popup"]',
                  '[class*="dialog"]',
                  '[class*="mask"]',
                  '[class*="overlay"]'
                ];
                for (const s of sels) {
                  document.querySelectorAll(s).forEach(el => {
                    const style = getComputedStyle(el);
                    const z = parseInt(style.zIndex || '0', 10);
                    if (z >= 999 && style.display !== 'none') {
                      el.style.display = 'none';
                    }
                  });
                }
                document.body.style.overflow = 'auto';
                """
            )
            time.sleep(0.2)
        except Exception:
            pass

    def _save_snapshot_url(self, row_id, img_url):
        """上传成功后回写库，避免下次任务重复拉取同一批。"""
        if row_id is None or not img_url:
            return
        sql = (
            "UPDATE `retrieve_process_lowprice_product` "
            "SET `snapshot_url` = %s WHERE `id` = %s AND `platform` = %s"
        )
        n = self.db_online.execute(sql, (img_url, row_id, self.platform))
        if n <= 0:
            logger.warning("snapshot_url 回写未影响行数 id=%s platform=%s", row_id, self.platform)

    def search(self, data_list):
        self.driver.get("https://dian.ysbang.cn/#/home", timeout=15)
        self.driver.wait.doc_loaded(timeout=10)
        time.sleep(2)

        if not self._is_logged_in():
            if not self.login():
                return False

        ok, fail = 0, 0
        for data in data_list:
            row_id = data.get("id")
            link_url = data.get("link_url")
            if not link_url:
                logger.warning("缺少 link_url，跳过 id=%s", row_id)
                fail += 1
                continue
            img_url = self.get_snapshot(link_url, row_id)
            if img_url:
                # self._save_snapshot_url(row_id, img_url)
                ok += 1
            else:
                fail += 1
            time.sleep(2)
        logger.info("快照任务结束 成功=%s 失败=%s 总计=%s", ok, fail, len(data_list))
        return ok > 0

    def run(self):
        date_str = time.strftime("%Y-%m-%d")
        sql = """
              SELECT `id`, `link_url`
              FROM `retrieve_process_lowprice_product`
              WHERE `platform` = %s
                AND `snapshot_url` = ""
                AND `scrape_date` = %s LIMIT 100 """

        data_list = self.db_online.select_data(sql, (self.platform, date_str))
        if not data_list:
            logger.info("当前不需要更新快照")
            return

        try:
            self.init_browser()
            self.search(data_list)
        except Exception as e:
            logger.exception("运行异常: %s", e)
        finally:
            self._quit_browser()


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