tob/spiders/yaoshibang/ysb_snapshot_crawl.py

import base64
import hashlib
import json
import math
import random
import re
import signal
import socket
import sys
import time
import zlib
from pathlib import Path
import requests
import secrets
import string
from Crypto.Cipher import AES
from commons.conn_mysql import MySQLPoolOn2
from DrissionPage import ChromiumPage, ChromiumOptions
from commons.Logger import logger
from oss_upload.oss_upload import AliyunOSSUploader
from pipelines.drug_pipelines import DrugPipeline
from datetime import datetime, timedelta
from area_info.city_name_to_id import get_city
import platform



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

SLIDER_OFFSET_FIX = 10
DETAIL_GET_TIMEOUT = 15
DETAIL_URL_WAIT = 10
DETAIL_DOM_WAIT = 8
DETAIL_NAV_RETRIES = 3
DETAIL_APPROVAL_XPATH = (
    'xpath://div[@class="drug-info"]//span[contains(text(),"批准文号")]'
)

chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
PROJECT_ROOT = Path(__file__).resolve().parents[2]
YSB_SPIDER_DIR = PROJECT_ROOT / "spiders" / "yaoshibang"

headers = {
    "Accept": "*/*",
    "Accept-Language": "zh-CN,zh;q=0.9",
    "Connection": "keep-alive",
    "Content-Type": "application/json",
    "Origin": "https://dian.ysbang.cn",
    "Referer": "https://dian.ysbang.cn/",
    "Sec-Fetch-Dest": "empty",
    "Sec-Fetch-Mode": "cors",
    "Sec-Fetch-Site": "same-origin",
    "User-Agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36"
    ),
    "sec-ch-ua": '"Chromium";v="146", "Not-A.Brand";v="24", "Google Chrome";v="146"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": '"Windows"',
}


def pkcs7_unpad(data):
    if not data:
        raise ValueError("Empty data for PKCS7 unpad")
    pad_len = data[-1]
    if pad_len < 1 or pad_len > 16:
        raise ValueError("Invalid PKCS7 padding length")
    if data[-pad_len:] != bytes([pad_len]) * pad_len:
        raise ValueError("Invalid PKCS7 padding bytes")
    return data[:-pad_len]


def derive_ysb_key():
    base = "BhCLxFfFhd12K4qRGPfy"
    md5_hex = hashlib.md5(base.encode("utf-8")).hexdigest()
    return md5_hex[:16].upper().encode("utf-8")


def decrypt_ysb_payload(cipher_text_b64):
    """解密药师帮列表接口 data.o 字段，返回 JSON 对象。"""
    key = derive_ysb_key()
    cipher_bytes = base64.b64decode(cipher_text_b64)
    cipher = AES.new(key, AES.MODE_ECB)
    decrypted = cipher.decrypt(cipher_bytes)
    unpadded = pkcs7_unpad(decrypted)
    json_bytes = zlib.decompress(unpadded, zlib.MAX_WBITS | 16)
    return json.loads(json_bytes.decode("utf-8"))


class YaoShiBangSnapshot:
    def __init__(self, drug_dict=None):
        self.driver = None
        self.db = MySQLPoolOn2()
        self.ip = None
        self.login_username = None
        self.login_password = None
        self.platform = 5
        self.page = 0
        self.pipeline = DrugPipeline("ysb")
        self.task_dict = drug_dict or {}
        self.ossuploader = AliyunOSSUploader()
        self.start_page = 1
        self.end_page = 1
        self.account_name = None
        self.phone = None
        self.password = None
        self._register_signal_handler()
        if self.task_dict:
            self.get_product_data()
        self.success = True
        self.is_no_prodcut = 0
        self.is_product_count = 0
        self.token = ""
        self._state_value = ""
        self.start_date = (datetime.now() - timedelta(minutes=500)).strftime("%Y-%m-%d %H:%M")

    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", "")
        self.sampling_cycle = self.task_dict.get("sampling_cycle", "")
        self.sampling_start_time = self.task_dict.get("sampling_start_time", "")
        self.sampling_end_time = self.task_dict.get("sampling_end_time", "")
        self.collect_equipment_id = self.task_dict.get("collect_equipment_id", "")
        self.account_id = self.task_dict.get("collect_equipment_account_id", "")
        self.collect_region_id = self.task_dict.get("collect_region_id", "")
        self.collect_round = self.task_dict.get("collect_round", 1)
        self.start_page = self._parse_page(self.task_dict.get("start_page"), 1)
        self.end_page = max(
            self.start_page,
            self._parse_page(self.task_dict.get("end_page"), self.start_page),
        )

    def _report_exception(self, exception_type, remark=""):
        scheduler = self.task_dict.get("scheduler")
        if scheduler:
            data = {
                "task_id": self.task_id,
                "platform": self.task_dict.get("_platform", str(self.platform)),
                "username": self.task_dict.get("_device_id", ""),
                "current_page": self.page,
                "is_finished": 0,
                "exception_type": exception_type,
                "remark": remark,
            }
            if self.page:
                data["total_pages"] = self.page
            if self.pipeline.crawl_count:
                data["crawled_count"] = self.pipeline.crawl_count
            scheduler.post_report(data)


    def _report_page_progress(self, page, crawled_count, is_finished=0, total_pages=0):
        """逐页回告进度，返回 True=code=error 需停止"""
        scheduler = self.task_dict.get("scheduler")
        if scheduler:
            data = {
                "task_id": self.task_id,
                "platform": self.task_dict.get("_platform", str(self.platform)),
                "username": self.task_dict.get("_device_id", ""),
                "current_page": page,
                "crawled_count": crawled_count,
                "is_finished": is_finished,
            }
            if total_pages:
                data["total_pages"] = total_pages
            scheduler.post_report(data)
            return scheduler.end
        return False

    @staticmethod
    def _parse_page(value, default=1):
        try:
            page = int(value)
            return page if page >= 1 else default
        except (TypeError, ValueError):
            return default

    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 _resolve_browser_profile_dir(self):
        """浏览器数据目录: <项目根>/spiders/yaoshibang/<账号>"""
        profile_dir = YSB_SPIDER_DIR / self.account_name
        # 仅兼容历史误路径，新建不再使用 chrome_profile
        legacy_nested = YSB_SPIDER_DIR / "spiders" / "yaoshibang" / self.account_name
        legacy_chrome_profile = YSB_SPIDER_DIR / "chrome_profile" / self.account_name

        for candidate in (profile_dir, legacy_nested, legacy_chrome_profile):
            if (candidate / "Default").is_dir() or (candidate / "Local State").is_file():
                logger.info("使用已有浏览器配置目录: %s", candidate)
                return candidate

        profile_dir.mkdir(parents=True, exist_ok=True)
        logger.info("新建浏览器配置目录: %s", profile_dir)
        return profile_dir

    def get_account(self):
        # 优先用入口传入的账号密码，避免重复查 DB
        phone = self.task_dict.get("account_phone")
        password = self.task_dict.get("account_password")
        name = self.task_dict.get("account_name")
        if phone and password:
            self.account_name = name or phone
            self.phone = phone
            self.password = password
            logger.info("使用入口传入的账号: name=%s, phone=%s", self.account_name, phone)
            return {"name": self.account_name, "phone": phone, "password": password}
        # 兜底：查 DB
        sql_account = f""" select `id`,`name`,`ip`,`phone`,`password`,`cookie_timestamp`,`cookie_str` from `accounts_platform` where `platform`=5 and `status`=1 and `equipment_id`=3 order by `cookie_timestamp` asc limit 1 """
        account_list = self.db.select_data(sql_account)
        if not account_list:
            return {}
        account_dict = account_list[0]
        self.account_name = account_dict["name"]
        self.phone = account_dict["phone"]
        self.password = account_dict["password"]
        return account_dict

    def init_browser(self):
        co = ChromiumOptions()

        # ---------- 1. 根据系统选择 Chrome 路径 ----------
        system = platform.system()
        if system == "Windows":
            chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
            # 如果装在非默认路径，可再追加备选
        elif system == "Linux":
            chrome_path = "/usr/bin/google-chrome"  # 或 "/usr/bin/chromium-browser"
        else:  # macOS 或其他
            chrome_path = None  # 让 DrissionPage 自动查找

        if chrome_path:
            co.set_browser_path(chrome_path)

        # ---------- 2. 统一使用 set_argument 设置（避免 set_headless 报错） ----------
        co.set_argument('--headless')  # 旧版无头（更兼容）
        co.set_argument('--window-size=1920,1080')
        co.set_argument('--no-sandbox')  # Linux 必需，Windows 下被忽略，无害
        co.set_argument('--disable-dev-shm-usage')  # Linux 解决 /dev/shm 问题，Windows 无害
        co.set_argument('--disable-gpu')  # 无头模式下可减少 GPU 问题，兼容所有系统
        co.set_argument('--disable-blink-features=AutomationControlled')
        co.set_argument('--start-maximized')
        co.set_argument('--no-first-run')
        co.set_argument('--no-default-browser-check')

        # ---------- 3. 用户数据目录（跨平台） ----------
        # 例如放在项目目录下的 profiles/ 中，按账号区分
        profile_dir = Path("./browser_profiles") / self.account_name
        profile_dir.mkdir(parents=True, exist_ok=True)
        co.set_user_data_path(str(profile_dir))

        co.set_argument("--remote-debugging-port=0")
        co.set_argument("--remote-debugging-address=127.0.0.1")

        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("验证码识别失败")
            self._report_exception(3, "滑块验证码识别失败")
            return False

        logger.info("滑块距离: %s", distance)
        slider = self.driver.ele(
            "xpath://div[contains(@class,'yidun_slider--hover')]", timeout=5
        )
        if not slider:
            logger.error("未找到滑块元素")
            self._report_exception(3, "未找到滑块元素")
            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 - 5)
        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 _current_url(self):
        try:
            return self.driver.url or ""
        except Exception:
            return ""

    def _goto_detail_page(self, item_id, detail_url):
        """get 后 refresh 一次，让 SPA 按当前 URL 重新渲染详情。"""
        for attempt in range(1, DETAIL_NAV_RETRIES + 1):
            try:
                self.driver.get(detail_url, timeout=5)
                time.sleep(1.5)
                eles = self.driver.eles("xpath=//div[@class='y-dialog']//button[contains(text(),'确认')]", timeout=3)
                if len(eles) == 2:
                    eles[1].click()
                    time.sleep(1)
                    self.driver.refresh()

                time.sleep(1.5)
                ele = self.driver.ele("xpath=//div[@class='drug-pic-viewer']")
                if not ele:
                    continue
                else:
                    return True

            except Exception as e:
                logger.warning(
                    "跳转详情异常 item_id=%s attempt=%s: %s",
                    item_id, attempt, e,
                )
                time.sleep(random.uniform(0.8, 1.5))
        return False

    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("未找到账号输入框")
            self._report_exception(4, "登录页未找到账号输入框")
            return False
        input_name.input(self.phone)
        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("未找到密码输入框")
            self._report_exception(4, "登录页未找到密码输入框")
            return False
        input_pass.input(self.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("未找到登录按钮")
            self._report_exception(4, "登录页未找到登录按钮")
            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 save_cookies(self, cookies_list):
        cookies_dict = {c['name']: c['value'] for c in cookies_list}

        timestamp = int(time.time())
        next_update_time = timestamp + random.randint(3600, 7200)
        update_sql = f""" UPDATE `accounts_platform` SET `cookie_timestamp` = %s, `cookie_str`= %s,`cookie_update_time` = %s, `status`= %s WHERE `name` = %s; """
        self.db.execute(update_sql,
                        (timestamp, json.dumps(cookies_dict), next_update_time, 1, self.account_name))
        logger.info("cookie已保存成功")

    def _take_snapshot(self, upload_key):
        """在当前页面截图并上传，不再重复跳转。"""
        time.sleep(1)
        self._dismiss_popup_before_screenshot()
        try:
            jpg_bytes = self.driver.get_screenshot(as_bytes="jpg", full_page=True)
            if not jpg_bytes:
                logger.warning("截图为空 upload_key=%s", upload_key)
                return ""
            img_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(upload_key))
        except Exception:
            logger.exception("截图或 OSS 上传失败 upload_key=%s", upload_key)
            return ""
        if not img_url:
            logger.warning("OSS 未返回有效地址 upload_key=%s", upload_key)
            return ""
        logger.info("截图上传完成 upload_key=%s url=%s", upload_key, img_url)
        time.sleep(random.uniform(1, 2))
        return img_url

    def gen_pair(self, ex1_len=9, o_raw_len=16):
        alphabet = string.ascii_lowercase + string.digits
        ex1 = "".join(secrets.choice(alphabet) for _ in range(ex1_len))
        o = base64.b64encode(secrets.token_bytes(o_raw_len)).decode("ascii")
        return {"ex1": ex1, "o": o}

    def build_base_payload(self, keyword, page, first_search):
        date_str = time.strftime("%Y-%m-%d %H:%M:%S")
        return {
            "platform": "pc",
            "version": "6.0.0",
            "ua": "Chrome146",
            'ex': '{} drugInfo {} {}'.format(self.start_date, date_str, date_str),
            "trafficType": 1,
            "ex1": "",
            "o": "",
            "lastClick": -1,
            "page": page,
            "pagesize": "60",
            "classify_id": "",
            "searchkey": keyword,
            "onlyTcm": 0,
            "operationtype": 1,
            "qualifiedLoanee": 0,
            "drugId": -1,
            "tagId": "",
            "showRecentlyPurchasedFlag": True,
            "onlySimpleLoan": 0,
            "sn": "",
            "buttons": [],
            "buttonList": [],
            "synonymId": 0,
            "activityTypes": [],
            "provider_filter": "",
            "factoryNames": "",
            "tcmGradeNames": [],
            "tcmExeStandardIds": [],
            "specs": "",
            "deliverFloor": 0,
            "purchaseLimitFloor": 0,
            "nextRequestKey": "",
            "adConfigId": 0,
            "stateValue": self._state_value,
            "firstSearch": first_search,
            "token": self.token,
        }

    @staticmethod
    def _extract_state_value(json_data, data_block):
        for src in (json_data, data_block):
            if not isinstance(src, dict):
                continue
            val = src.get("stateValue") or src.get("state_value")
            if val:
                return str(val)
        return None

    def _dismiss_popup_before_screenshot(self):
        """截图前关闭或隐藏营销弹窗，避免遮挡。"""
        self.driver.run_js("""
        // 暴力清除所有高 z-index 遮罩和弹窗
        document.querySelectorAll('*').forEach(el => {
          const s = getComputedStyle(el);
          const z = parseInt(s.zIndex) || 0;
          if (z > 100 && (s.position === 'fixed' || s.position === 'absolute') &&
              (el.clientWidth > 100 || el.clientHeight > 100)) {
            el.remove();
          }
        });
        """)
        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')]",
            "xpath=//*[contains(text(),'智能采购')]/ancestor::div[3]//*[contains(@class,'close')]",
            "xpath=//*[contains(text(),'智能采购')]/ancestor::div[3]//button[last()]",
        ]
        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 to_product(self, item):

        now = time.strftime("%Y-%m-%d %H:%M:%S")
        item_id = item.get("wholesaleid", "")
        provider_id = item.get("providerId", "")

        city_str = item.get("warehouseCity", "")
        city_id = province_id = city = province = ""
        price = item.get("disPrice", "")

        if not price:
            price = item.get("minprice", "")
        if not price:
            price = item.get("price", "")

        shop_name = item.get("provider_name", "")
        if not shop_name:
            shop_name = item.get("abbreviation", "")

        # 销量: 优先 order_amount > orderAmountPlus > alreadysales
        sales = (
            item.get("order_amount")
            or item.get("orderAmountPlus")
            or item.get("alreadysales")
            or ""
        )

        product = {
            "platform": self.platform,
            "item_id": item_id,
            "enterprise_id": self.company_id,
            "product_name": item.get("drugname", ""),
            "spec": item.get("specification", ""),
            "one_price": '',
            "detail_url": f"https://dian.ysbang.cn/#/drugInfo?wholesaleid={item_id}&trafficType=1",
            "shop_name": shop_name,
            "anonymous_store_name": "",
            "shop_url": f"https://dian.ysbang.cn/#/supplierstore?providerId={provider_id}&trafficType=4",
            "city_name": city,
            "city_id": city_id,
            "province_name": province,
            "province_id": province_id,
            "area_info": "",
            "factory_name": item.get("manufacturer", ""),
            "scrape_date": time.strftime("%Y-%m-%d"),
            "price": price,
            "sales": sales,
            "stock_count": item.get("stockAvailable", ""),
            "snapshot_url": "",
            "approval_num": "",
            "produced_time": item.get("prodDate", ""),
            "deadline": item.get("valid_date", ""),
            "update_time": now,
            "insert_time": now,
            "number": 1,
            "product_brand": self.brand or "",
            "collect_task_id": self.collect_task_id,
            "search_name": self.product,
            "company_name": "",
            "collect_config_info": json.dumps(
                {"sampling_cycle": self.sampling_cycle, "sampling_start_time": self.sampling_start_time,
                 "sampling_end_time": self.sampling_end_time}),
            "account_id": self.account_id,
            "collect_region_id": self.collect_region_id,
            "collect_round": self.collect_round,
            "is_sold_out": 0
        }
        return product

    def parse_detail(self, product):
        appvolnum_ele = self.driver.ele(
            'xpath://div[@class="drug-info"]//span[contains(text(),"批准文号")]/following-sibling::span[1]')
        appvolnum_value = appvolnum_ele.text if appvolnum_ele else ""
        price = ""
        discount_ele = self.driver.ele(
            'xpath://div[@class="sale-info-wrap"]//div[@class="tooltip-content"]',
            timeout=2,
        )
        discount_value = discount_ele.text if discount_ele else ""
        if not price and discount_value:
            price_re = re.search(r"¥([0-9.]+)", discount_value)
            if price_re:
                price = price_re.group(1).strip()

        current_ele = self.driver.ele(
            'xpath://div[@class="sale-info-wrap"]//span[contains(@class,"current-price")]',
            timeout=3,
        )

        if current_ele and not price:
            price = (current_ele.text or "").replace("¥", "").strip()

        list_price = product.get("price", "")
        if price:
            product["price"] = price

        if appvolnum_value:
            product["approval_num"] = appvolnum_value
        logger.info(
            "详情解析 wholesaleid=%s list_price=%s dom_price=%s url=%s",
            product.get("item_id"),
            list_price,
            product.get("price"),
            self._current_url(),
        )
        return product

    def search(self):
        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
        time.sleep(3)
        cookies_list = self.driver.cookies()
        cookies_dict = {c['name']: c['value'] for c in cookies_list}
        self.save_cookies(cookies_list)
        self.token = cookies_dict.get("Token") or cookies_dict.get("token")

        keyword = self.product
        if self.brand:
            keyword = (self.brand + " " + self.product).strip()
        if self.product_desc:
            keyword = (keyword + " " + self.product_desc).strip()

        self._state_value = ""
        # 支持续传：从 current_page（>0）或 start_page 开始
        page_start = int(self.task_dict.get("current_page", 0) or 0)
        if page_start <= 0:
            page_start = self.start_page
        self.page = page_start  # 兜底：避免0页时回告 total_pages=0
        for page in range(page_start, 100):
            first_search = page == 1
            logger.info("药师帮爬取第%s页 firstSearch=%s stateValue=%s", page, first_search,
                        self._state_value or "(空)")
            pair = self.gen_pair()
            payload = self.build_base_payload(keyword, page=page, first_search=first_search)
            payload["ex1"] = pair["ex1"]
            payload["o"] = pair["o"]

            response = None
            for attempt in range(3):
                try:
                    response = requests.post(
                        "https://dian.ysbang.cn/wholesale-drug/sales/getWholesaleList/v4270", headers=headers,
                        json=payload, timeout=30
                    )
                    if response.status_code == 200:
                        break
                except Exception as e:
                    logger.error("第%s页请求失败 (%s/3): %s", page, attempt + 1, e)
                    response = None
                    time.sleep(10)
            if not response or response.status_code != 200:
                logger.error("第%s页请求失败，停止爬取", page)
                self._report_exception(5, f"第{page}页请求失败(status={response.status_code if response else 'None'})")
                return False

            try:
                data_json = response.json()
            except json.JSONDecodeError:
                logger.exception("第%s页响应不是合法 JSON", page)
                self._report_exception(5, f"第{page}页响应非合法JSON")
                return False
            data_block = data_json.get("data") or {}
            if str(data_json.get("message", "")) == "该操作需要登录":
                logger.warning("第%s页需要登录，请检查浏览器登录态", page)
                self._report_exception(1, "账号登录态失效")
                return False

            encrypted_o = data_block.get("o")
            if not encrypted_o:
                logger.warning("第%s页返回无加密 data.o: %s", page, data_json)
                break

            try:
                json_data = decrypt_ysb_payload(encrypted_o)
            except Exception as e:
                logger.exception("第%s页解密失败: %s", page, e)
                self._report_exception(6, f"第{page}页数据解密失败")
                continue

            state_val = self._extract_state_value(json_data, data_block)
            logger.info("第%s页 解密后 keys: %s, stateValue=%s, has_wholesales=%s", page,
                        list(json_data.keys())[:20], state_val, bool(json_data.get("wholesales")))
            if state_val:
                self._state_value = state_val

            wholesales = json_data.get("wholesales", [])
            if not wholesales:
                logger.info(f"第{page}页无数据，停止")
                break

            for item in wholesales:
                item_id = item.get("wholesaleid", "")
                if not item_id:
                    continue

                detail_url = (
                    f"https://dian.ysbang.cn/#/drugInfo?wholesaleid={item_id}&trafficType=1"
                )
                product = self.to_product(item)
                title = product.get("product_name", "")
                if self.brand not in title:
                    self.is_product_count += 1
                    continue
                if self.product not in title:
                    self.is_product_count += 1
                    continue
                if self.product in title and self.brand in title:
                    self.is_product_count = 0
                if self.is_product_count >= 20:
                    return False

                if not self._goto_detail_page(item_id, detail_url):
                    logger.warning(
                        "详情页跳转失败，跳过 item_id=%s url=%s",
                        item_id, detail_url,
                    )
                    continue

                product = self.parse_detail(product)
                upload_key = hashlib.md5(detail_url.encode("utf-8")).hexdigest()
                product["snapshot_url"] = self._take_snapshot(upload_key)

                try:
                    self.pipeline.storge_data(product)
                    logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
                except Exception as e:
                    logger.exception("写入数据库失败: %s", e)

            # 逐页回告进度
            self.page = page
            if self._report_page_progress(page, self.pipeline.crawl_count) == True:
                break

    def run(self):
        account_dict = self.get_account()
        if not account_dict:
            logger.error("ysb快照模式无账号可用")
            self._report_exception(1, "无可用账号")
            return 0, False
        try:
            self.init_browser()
            self.search()
        except Exception as e:
            logger.exception("运行异常: %s", e)
            self.success = False
            self._report_exception(5, f"未知异常: {e}")
        finally:
            self._quit_browser()
        return self.pipeline.crawl_count, self.success, self.page