jd-tb-yaofang-2/spiders/jd/jd_auto_crawl_snap2.py

import random
import re
import signal
import socket
import sys
import time
from decimal import Decimal, InvalidOperation
from urllib.parse import quote
from DrissionPage import ChromiumPage, ChromiumOptions
import json
import hashlib
import os
from base64 import b64decode
from commons.Logger import get_spider_logger
from commons.conn_mysql import MySQLPoolOn2
from pipelines.drug_pipelines import DrugPipeline
from commons.feishu_webhook import send_text
from spiders.jd.jd_captcha import handle_jd_slider_captcha
from oss_upload.oss_upload import AliyunOSSUploader
from commons.config import (
    JD_DEVICE_ID,
    CHROME_PATH,
    UA_JD,
    CRAWLER_TOKEN,
    JD_NO_MATCH_THRESHOLD,
    JD_COLLECT_MAX_STEPS,
)
import requests
logger = get_spider_logger("jd")

chrome_path = CHROME_PATH

FETCH_TIMEOUT_FIRST = 3
FETCH_TIMEOUT_SCROLL = 3
LISTEN_CLEAR_ROUNDS = 3
LISTEN_CLEAR_TIMEOUT = 0.45

# 验证码 URL 特征
CAPTCHA_URL_KEYS = ["risk_handler", "cfe.m.jd.com", "verifycode", "captcha"]
# 验证码页面文案特征
CAPTCHA_MARKERS = ["\u9a8c\u8bc1\u4e00\u4e0b\uff0c\u8d2d\u7269\u65e0\u5fe7", "\u524d\u65b9\u62e5\u6324", "\u62fc\u56fe\u9a8c\u8bc1"]

# 「下一页」是否在视口内（条件略宽）
_JS_NEXT_BTN_IN_VIEWPORT = """
var el = arguments[0];
if (!el) return false;
var r = el.getBoundingClientRect();
var h = window.innerHeight || document.documentElement.clientHeight || 800;
var w = window.innerWidth || document.documentElement.clientWidth || 1200;
return r.bottom > 80 && r.top < h - 40 && r.right > 0 && r.left < w;
"""


class _PageInterrupted(Exception):
    """采集过程中页面出现异常(验证码/被踢/渲染异常),中断本页采集。"""
    def __init__(self, kind):
        self.kind = kind  # "captcha" / "kicked" / "render"
        super().__init__(kind)


class JdCrawlerV2:
    def __init__(self, drug_dict=None, scheduler=None, driver=None, cumulative_pages=0, cumulative_items=0, cumulative_stored=0, cumulative_skipped=0):
        self.driver = driver
        self.cumulative_pages = cumulative_pages
        self.cumulative_items = cumulative_items
        self.cumulative_stored = cumulative_stored
        self.cumulative_skipped = cumulative_skipped
        self.page_stored = 0
        self.register_signal_handler()
        self.db = MySQLPoolOn2()
        self.ip = None
        self.account_name = None
        self.login_username = None
        self.login_password = None
        self.platform = 2
        self.pipeline = DrugPipeline("jd")
        self.task_dict = drug_dict or {}
        self.scheduler = scheduler
        self.report_data = {}
        self.heartbeat_interval = 30
        self.ossuploader = AliyunOSSUploader()
        self.start_page = 1
        self.end_page = 1
        if self.task_dict:
            self.get_product_data()
        self.success = True
        self.is_no_prodcut = 0
        self._mouse = (500, 400)  # 虚拟鼠标位置(视口坐标)
        self._snap_miss = 0      # 连续找不到商品元素的计数
        self._snap_local_count = 1  # 本地快照序号

    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", "15")
        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 = 100
        self.report_data = {'task_id': self.task_id, 'platform': self.platform, 'username': self.task_dict.get("username", JD_DEVICE_ID)}

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

    @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):
        if self.driver:
            logger.info("复用已有浏览器实例")
            self._listen_started = False
            return
            
        co = ChromiumOptions().set_browser_path(chrome_path)
        debug_port = self._get_free_port()
        co.set_user_data_path(f"./account_cache/jd/JD_101")

        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}")
        logger.info("启动浏览器: account=%s, debug_port=%s", self.account_name, debug_port)
        self.driver = ChromiumPage(co)
        self._listen_started = False

    def _start_listen(self):
        """登录完成后再开监听，避免干扰登录页/验证码拖动。"""
        if self._listen_started or not self.driver:
            return
        self.driver.listen.start("api?appid=search-pc-java")
        self._listen_started = True
        logger.info("已启动搜索接口监听")

    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 sleep(self, a, b):
        time.sleep(random.uniform(a, b))

    def _scroll_page_down(self, delta=900):
        """拟人滚轮:真实 CDP mouseWheel 事件,物理签名。"""
        try:
            self._human_wheel(int(delta))
        except Exception:
            self.driver.run_js(f"window.scrollBy(0, {int(delta)});")
            time.sleep(random.uniform(0.15, 0.35))

    def _scroll_next_into_view(self, el):
        if not el:
            return
        try:
            self.driver.run_js(
                "arguments[0].scrollIntoView({block:'center',behavior:'instant'});",
                el,
            )
            self.sleep(1, 2)
        except Exception as e:
            logger.warning("滚动到下一页按钮失败: %s", e)
            try:
                el.scroll.to_see()
            except Exception:
                pass

    def _get_scroll_info(self):
        return self.driver.run_js("""
            return {
                scrollY: window.scrollY || window.pageYOffset || 0,
                docH: Math.max(document.body.scrollHeight,
                               document.documentElement.scrollHeight,
                               document.body.offsetHeight),
                viewH: window.innerHeight || document.documentElement.clientHeight || 800
            };
        """)

    def _find_next_btn(self, timeout=0.3):
        try:
            return self.driver.ele("text=下一页", timeout=timeout)
        except Exception:
            return None

    def _is_next_btn_visible(self, btn):
        if not btn:
            return False
        try:
            return bool(self.driver.run_js(_JS_NEXT_BTN_IN_VIEWPORT, btn))
        except Exception:
            return False

    def _human_click(self, element):
        """拟人点击:贝塞尔移动+悬停+遮挡检测+真实 CDP mousePressed/Released。"""
        if not element:
            return False
        try:
            return self._human_click_el(element)
        except Exception as e:
            logger.warning("拟人点击失败,退回 JS click: %s", e)
            try:
                self.driver.run_js(
                    "arguments[0].scrollIntoView({block:'center',behavior:'instant'});",
                    element,
                )
                self.driver.run_js("arguments[0].click();", element)
                return True
            except Exception:
                try:
                    element.click()
                    return True
                except Exception:
                    return False

    @staticmethod
    def _estimated_price(json_data):
        fp = json_data.get("finalPrice")
        if isinstance(fp, dict):
            return fp.get("estimatedPrice", "") or ""
        return ""

    def get_heshu(self, full_title):
        last_box = None
        last_bottle = None
        for match in re.finditer(r"(\d+)(盒|瓶)", full_title):
            if match.group(2) == '盒':
                last_box = match
            else:  # 瓶
                last_bottle = match
        if last_box:
            return int(last_box.group(1))
        elif last_bottle:
            return int(last_bottle.group(1))
        else:
            return 1

    def _take_snapshot(self, upload_key, ele):
        """在指定标签页截图并上传。"""
        # 拟人:滚轮把元素带进视口 → 鼠标移到商品上(像人在看) → 随机停顿
        try:
            self._human_scroll_el_into_view(ele)
            pt = self._el_click_point(ele)
            if pt:
                self._human_move_to(*pt)
            if random.random() < 0.08:
                self.sleep(0.4, 0.9)  # 偶尔停下来细看
            else:
                self.sleep(0.05, 0.2)
        except Exception:
            pass
        time.sleep(0.05)  # 快照前固定等待(过短可能截到未加载完的图片)
        # CDP clip 截图:避免 ele.get_screenshot 触发 Chrome 视口重排导致页面跳动
        try:
            rect = self.driver.run_js(
                "var r=arguments[0].getBoundingClientRect();"
                "return [r.left + window.scrollX, r.top + window.scrollY, r.width, r.height,"
                "r.top, r.bottom, r.left, r.right, window.innerHeight, window.innerWidth, window.scrollY];",
                ele,
            )
            if not rect or rect[2] <= 0 or rect[3] <= 0:
                logger.warning("元素区域异常 upload_key=%s", upload_key)
                return ""
            x, y, w, h = rect[0], rect[1], rect[2], rect[3]
            clip = {"x": x, "y": y, "width": w, "height": h, "scale": 1}
            # 元素完整在视口内(垂直+水平)才不开 captureBeyondViewport
            in_view = (rect[4] >= 0 and rect[5] <= rect[8] and rect[6] >= 0 and rect[7] <= rect[9])
            if in_view:
                data = self.driver.run_cdp(
                    "Page.captureScreenshot", format="jpeg", quality=100, clip=clip, _timeout=12,
                )["data"]
            else:
                data = self.driver.run_cdp(
                    "Page.captureScreenshot", format="jpeg", quality=100,
                    captureBeyondViewport=True, clip=clip, _timeout=12,
                )["data"]
                self.driver.run_js(
                    "window.scrollTo({top: %d, behavior: 'instant'});" % int(rect[10])
                )
            jpg_bytes = b64decode(data)
            if not jpg_bytes:
                logger.warning("截图为空 upload_key=%s", upload_key)
                return ""
            self._last_snap_bytes = jpg_bytes  # 暂存,供 parse 内本地保存
            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(0.05, 0.15))
        return img_url

    def parse(self, ware_list, force=False):
        unprocessed = []
        for w in ware_list:
            # 每5条做一次页面级异常巡检,命中立刻中断本页采集
            if len(unprocessed) % 5 == 0:
                intr = self._interruption()
                if intr:
                    raise _PageInterrupted(intr)
            if not isinstance(w, dict):
                continue
            
            sku_id = str(w.get("skuId", ""))
            if not sku_id:
                continue

            ele_xpath = "//div[@id='main_search_conter']//div[contains(@class,'_goodsContainer_')]/div[@data-sku=" + "'" + sku_id + "'" + "]"
            ele_screen = self.driver.ele("xpath=" + ele_xpath, timeout=0.2)
            if not ele_screen and not force:
                unprocessed.append(w)
                continue

            title = w.get("wareName", "")

            title = re.sub(r"<[^>]*>", "", title).strip()
            color = w.get("color", "")
            full_title = title + " " + color

            logger.info(full_title)

            if self.product not in full_title:
                self.is_no_prodcut += 1
                continue
            if self.brand not in full_title:
                self.is_no_prodcut += 1
                continue
            if self.product_desc:
                if self.product_desc in full_title:
                    crawl_product_desc = self.product_desc
                else:
                    crawl_product_desc = ""
                    title = full_title
            else:
                crawl_product_desc = ""
                title = full_title

            if "+[" in title:
                continue

            self.is_no_prodcut = 0
            status = 1
            if self.product_keyword:
                search_keyword_list = self.product_keyword.split(",")
                for search_keyword in search_keyword_list:
                    if search_keyword.strip() not in title:
                        status = 0
            if status == 0:
                continue

            logger.info(f"商品名：{title}")
            sku_id = w.get("skuId", "")
            sales = w.get("totalSales", "")
            shop_id = w.get("shopId", "")
            shop_name = w.get("shopName", "")
            heshu_count = self.get_heshu(full_title)
            final_price = self._estimated_price(w)
            jd_price = w.get("jdPrice", "")
            item_url = f"https://item.jd.com/{sku_id}.html"
            low_price = final_price if final_price else jd_price

            # 获取列表页快照
            upload_key = hashlib.md5(item_url.encode("utf-8")).hexdigest()
            snap_url = ""
            if ele_screen:
                for i in range(3):
                    snap_url = self._take_snapshot(upload_key, ele_screen)
                    if snap_url != "":
                        break
            else:
                logger.warning(f"未找到商品元素无法截图: {sku_id}")

            try:
                price = Decimal(str(low_price)).quantize(Decimal("0.00"))
            except (InvalidOperation, ValueError):
                price = Decimal("0.00")

            # 本地留存快照用于人工检查质量: img/001_价格.jpg
            if snap_url and getattr(self, '_last_snap_bytes', None):
                try:
                    local_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "img")
                    os.makedirs(local_dir, exist_ok=True)
                    local_name = f"{self._snap_local_count:03d}_{price}.jpg"
                    with open(os.path.join(local_dir, local_name), "wb") as lf:
                        lf.write(self._last_snap_bytes)
                    self._snap_local_count += 1
                    self._last_snap_bytes = None
                except Exception:
                    pass

            item_url = f"https://item.jd.com/{sku_id}.html"
            mall_url = f"https://mall.jd.com/index-{shop_id}.html?from=pc"

            # 字段与 yaofangwang_crawl 对齐；键顺序须与 commons.sql_data.RETRIEVE_SCRAPE_INSERT_COLUMNS 一致
            now_ts = time.strftime("%Y-%m-%d %H:%M:%S")
            product = {
                "platform": self.platform,
                "item_id": sku_id,
                "enterprise_id": self.company_id,
                "product_name": title,
                "spec": crawl_product_desc,
                "one_price": "",
                "detail_url": item_url,
                "shop_name": shop_name,
                "anonymous_store_name": "",
                "shop_url": mall_url,
                "city_name": "",
                "city_id": "",
                "province_name": "",
                "province_id": "",
                "shipment_city_name": "",
                "shipment_city_id": "",
                "shipment_province_name": "",
                "shipment_province_id": "",
                "area_info": "",
                "factory_name": "",
                "scrape_date": time.strftime("%Y-%m-%d"),
                "price": price,
                "sales": sales,
                "stock_count": "",
                "snapshot_url": snap_url,
                "approval_num": "",
                "produced_time": "",
                "deadline": "",
                "update_time": now_ts,
                "insert_time": now_ts,
                "number": heshu_count,
                "product_brand": self.brand or "",
                "collect_task_id": self.collect_task_id,
                "task_id": self.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
            }

            try:
                affected_rows = self.pipeline.storge_data(product)
                if affected_rows and affected_rows > 0:
                    self.page_stored += 1
                logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
            except Exception as e:
                logger.exception("写入数据库失败: %s", e)
        return unprocessed

    @staticmethod
    def _response_has_ware_list(data):
        if not isinstance(data, dict):
            return False
        inner_data = data.get("data")
        if not isinstance(inner_data, dict):
            return False
        return bool(inner_data.get("wareList"))

    def fetch_items_once(self, timeout=FETCH_TIMEOUT_FIRST):
        wares = []
        for resp in self.driver.listen.steps(timeout=timeout):
            try:
                data = resp.response.body
                if not self._response_has_ware_list(data):
                    continue
                ware_list = data["data"]["wareList"]
                wares.extend(ware_list)
            except Exception as e:
                logger.warning("解析监听响应失败: %s", e)
        return wares

    def clear_listen_buffer(self, rounds=LISTEN_CLEAR_ROUNDS, timeout=LISTEN_CLEAR_TIMEOUT):
        try:
            for _ in range(rounds):
                resps = list(self.driver.listen.steps(timeout=timeout))
                if not resps:
                    break
            logger.debug("监听缓冲已清空")
        except Exception as e:
            logger.debug("清空监听缓冲失败: %s", e)

    def collect_full_page_items(self, max_steps=JD_COLLECT_MAX_STEPS):
        """单次循环：边滑动边收数据，到底 / 看见「下一页」即停。"""
        # 页首巡检 + 清理误开标签页
        self._close_unexpected_tabs()
        intr = self._interruption()
        if intr:
            raise _PageInterrupted(intr)
        pending_wares = []
        total_n = 0
        new_wares = self.fetch_items_once(timeout=FETCH_TIMEOUT_FIRST)
        total_n += len(new_wares)
        pending_wares.extend(new_wares)
        pending_wares = self.parse(pending_wares)

        stagnant = 0
        last_scroll_y = None

        for step in range(max_steps):
            # 每步巡检:中途出验证码/被踢立刻中断本页
            intr = self._interruption()
            if intr:
                raise _PageInterrupted(intr)
            next_btn = self._find_next_btn(timeout=0.3)
            if self._is_next_btn_visible(next_btn):
                new_wares = self.fetch_items_once(timeout=FETCH_TIMEOUT_SCROLL)
                total_n += len(new_wares)
                pending_wares.extend(new_wares)
                pending_wares = self.parse(pending_wares)
                self.parse(pending_wares, force=True)
                return total_n, next_btn

            info = self._get_scroll_info()
            scroll_y = info["scrollY"]
            doc_h = info["docH"]
            view_h = info["viewH"]

            at_bottom = (scroll_y + view_h >= doc_h - 20)
            if last_scroll_y is not None and abs(scroll_y - last_scroll_y) < 8:
                stagnant += 1
            else:
                stagnant = 0
            last_scroll_y = scroll_y

            if at_bottom and stagnant >= 2:
                new_wares = self.fetch_items_once(timeout=FETCH_TIMEOUT_SCROLL)
                total_n += len(new_wares)
                pending_wares.extend(new_wares)
                pending_wares = self.parse(pending_wares)
                self.parse(pending_wares, force=True)
                next_btn = self._find_next_btn(timeout=2)
                if next_btn:
                    self._scroll_next_into_view(next_btn)
                    return total_n, next_btn
                logger.info("已到页面底部且未发现下一页，停止滑动")
                return total_n, None


            # 滚动前约1/4概率先把鼠标移近滚动区域(真人边读边滚,鼠标不是死的)
            if random.random() < 0.25:
                try:
                    vw, vh = self.driver.run_js("return [window.innerWidth, window.innerHeight];")
                    self._human_move_to(
                        vw * random.uniform(0.25, 0.75), vh * random.uniform(0.3, 0.7)
                    )
                except Exception:
                    pass

            self._scroll_page_down(random.randint(600, 1100))

            if random.random() < 0.15:
                self._human_wheel(-random.randint(60, 140))

            self.sleep(0.1, 0.35)
            
            pending_wares = self.parse(pending_wares)

            if step % 3 == 2:
                new_wares = self.fetch_items_once(timeout=FETCH_TIMEOUT_SCROLL)
                total_n += len(new_wares)
                pending_wares.extend(new_wares)
                pending_wares = self.parse(pending_wares)

        new_wares = self.fetch_items_once(timeout=FETCH_TIMEOUT_SCROLL)
        total_n += len(new_wares)
        pending_wares.extend(new_wares)
        pending_wares = self.parse(pending_wares)
        self.parse(pending_wares, force=True)

        next_btn = self._find_next_btn(timeout=3)
        if next_btn and not self._is_next_btn_visible(next_btn):
            self._scroll_next_into_view(next_btn)

        return total_n, next_btn

    def get_account(self):
        sql_account = """
                     SELECT *
                     FROM `retrieve_collect_equipment_account`
                     WHERE `id` = %s
                       and `status` = 0
                     """
        account_list = self.db.select_data(sql_account, self.account_id)
        if not account_list:
            return False

        account_dict = account_list[0]
        print(account_dict)
        self.ip = account_dict.get("ip")
        self.account_name = account_dict.get("username")
        self.login_username = account_dict.get("phone") or account_dict.get("username", "")
        self.login_password = account_dict.get("password", "")
        logger.info("获取到账号: %s, ip: %s", self.account_name, self.ip)
        return True

    def disable_account(self):
        update_sql = f""" UPDATE `retrieve_collect_equipment_account` SET `status`= %s WHERE `name` = %s; """
        self.db.execute(update_sql, (1, self.account_name))

    def _build_search_keyword(self):
        parts = [p for p in (self.brand, self.product, self.product_desc) if p]
        return " ".join(parts).strip() or self.product

    def _is_logged_out(self):
        return bool(self.driver.ele("xpath=//*[@class='link-login']", timeout=2))

    # ==================== 拟人化:鼠标层 ====================

    def _human_move_to(self, x, y, duration=None):
        """贝塞尔曲线移动鼠标(ease-out),沿途产生真实 CDP mousemove 事件。"""
        x0, y0 = self._mouse
        dist = ((x - x0) ** 2 + (y - y0) ** 2) ** 0.5
        if dist < 2:
            self._mouse = (x, y)
            return
        c1 = (
            x0 + (x - x0) * random.uniform(0.15, 0.4) + random.uniform(-0.15, 0.15) * dist,
            y0 + (y - y0) * random.uniform(0.15, 0.4) + random.uniform(-0.15, 0.15) * dist,
        )
        c2 = (
            x0 + (x - x0) * random.uniform(0.6, 0.9) + random.uniform(-0.15, 0.15) * dist,
            y0 + (y - y0) * random.uniform(0.6, 0.9) + random.uniform(-0.15, 0.15) * dist,
        )
        steps = max(10, min(35, int(dist / 25)))
        if duration is None:
            duration = max(0.08, min(0.45, dist / random.uniform(2200, 3800)))
        for i in range(1, steps + 1):
            t = 1 - (1 - i / steps) ** 2
            mt = 1 - t
            px = mt ** 3 * x0 + 3 * mt ** 2 * t * c1[0] + 3 * mt * t ** 2 * c2[0] + t ** 3 * x
            py = mt ** 3 * y0 + 3 * mt ** 2 * t * c1[1] + 3 * mt * t ** 2 * c2[1] + t ** 3 * y
            self.driver.run_cdp("Input.dispatchMouseEvent", type="mouseMoved", x=px, y=py)
            time.sleep(duration / steps * random.uniform(0.6, 1.4))
        self._mouse = (x, y)

    def _el_click_point(self, el):
        """元素可视区域内的随机点击点(视口坐标),不在视口返回 None。"""
        try:
            r = self.driver.run_js(
                "var r=arguments[0].getBoundingClientRect();"
                "return [r.left, r.top, r.width, r.height, window.innerWidth, window.innerHeight];",
                el,
            )
        except Exception:
            return None
        if not r:
            return None
        left, top, w, h, vw, vh = r
        x = left + w * random.uniform(0.25, 0.75)
        y = top + h * random.uniform(0.3, 0.7)
        if not (2 <= x <= vw - 2 and 2 <= y <= vh - 2):
            return None
        return (x, y)

    def _hit_check(self, el, x, y):
        """确认 (x,y) 处顶层元素是目标本身/子元素;否则返回遮挡物描述。"""
        try:
            return self.driver.run_js(
                "var t=arguments[2];"
                "var h=document.elementFromPoint(arguments[0],arguments[1]);"
                "if(!h) return 'none';"
                "if(h===t||t.contains(h)||h.contains(t)) return 'ok';"
                "var d=(h.tagName||'').toLowerCase();"
                "if(h.id) d+='#'+h.id;"
                "if(h.className&&typeof h.className==='string') d+='.'+h.className.trim().split(' ')[0];"
                "return d;",
                x, y, el,
            )
        except Exception:
            return "err"

    def _human_click_el(self, el):
        """真人点击:自然滚到元素→贝塞尔移动→悬停→遮挡检测→真实 mousePressed/Released。"""
        for attempt in range(3):
            if not self._human_scroll_el_into_view(el):
                try:
                    self.driver.run_js(
                        "arguments[0].scrollIntoView({block:'center',behavior:'instant'});", el
                    )
                except Exception:
                    pass
            self.sleep(0.15, 0.35)
            self._wait_scroll_settle()
            pt = self._el_click_point(el)
            if not pt:
                return False
            self._human_move_to(*pt)
            self.sleep(0.2, 0.6)
            hit = self._hit_check(el, *pt)
            if hit == "ok":
                x, y = pt
                self.driver.run_cdp(
                    "Input.dispatchMouseEvent", type="mousePressed",
                    x=x, y=y, button="left", clickCount=1,
                )
                time.sleep(random.uniform(0.05, 0.12))
                self.driver.run_cdp(
                    "Input.dispatchMouseEvent", type="mouseReleased",
                    x=x, y=y, button="left", clickCount=1,
                )
                self._close_unexpected_tabs()
                return True
            logger.info("点击被遮挡(%s),按 Esc 关弹层重试(%d/3)", hit, attempt + 1)
            self._press_escape()
            self.sleep(0.5, 1.2)
            if attempt >= 1:
                self._hide_floater_at(*pt, el)
        logger.warning("多次重试仍被遮挡,放弃本次点击")
        return False

    def _mouse_wander(self):
        """采集间隙随机移动鼠标,模拟真人浏览时的手部小动作。"""
        try:
            loc = self.driver.run_js(
                "return [Math.round(window.scrollX + innerWidth * Math.random())," +
                " Math.round(window.scrollY + innerHeight * Math.random())];"
            )
            self.driver.actions.move_to(loc, duration=random.uniform(0.4, 1.0))
        except Exception:
            pass

    # ==================== 拟人化:滚轮层 ====================

    def _human_wheel(self, delta):
        """真实滚轮事件:120/格物理签名,1~4格一串,格间隔30~70ms。"""
        try:
            vw, vh = self.driver.run_js("return [window.innerWidth, window.innerHeight];")
        except Exception:
            vw, vh = 1200, 800
        x, y = self._mouse
        if not (0 <= x <= vw and 0 <= y <= vh):
            self._human_move_to(vw * random.uniform(0.3, 0.7), vh * random.uniform(0.3, 0.6))
            x, y = self._mouse
        sign = 1 if delta >= 0 else -1
        notches = max(1, round(abs(delta) / 120))
        while notches > 0:
            burst = min(notches, random.randint(1, 4))
            for _ in range(burst):
                self.driver.run_cdp(
                    "Input.dispatchMouseEvent", type="mouseWheel",
                    x=x, y=y, deltaX=0, deltaY=sign * 120,
                )
                time.sleep(random.uniform(0.03, 0.07))
            notches -= burst
            if notches > 0:
                time.sleep(random.uniform(0.04, 0.10))
        time.sleep(random.uniform(0.05, 0.12))

    def _human_scroll_el_into_view(self, el, max_tries=6):
        """用自然滚轮把元素带进视口(不用 scrollIntoView 瞬移)。"""
        for _ in range(max_tries):
            r = self.driver.run_js(
                "var r=arguments[0].getBoundingClientRect();"
                "return [r.top, r.bottom, r.left, r.right, window.innerHeight, window.innerWidth];",
                el,
            )
            if not r:
                return False
            top, bottom, left, right, vh, vw = r
            # ---- 垂直方向 ----
            el_h = bottom - top
            if el_h > vh - 60:
                # 高元素:视口放不下,顶部对齐即可
                v_ok = (8 <= top <= 120)
                v_delta = 0 if v_ok else int(top - 40)
            else:
                # 完整可见才算到位
                v_ok = (top >= 8 and bottom <= vh - 8)
                v_delta = 0 if v_ok else int((top + bottom) / 2 - vh / 2)
            # ---- 水平方向 ----
            el_w = right - left
            if el_w > vw - 16:
                # 宽元素:视口放不下,左边缘对齐即可
                h_ok = (0 <= left <= 30)
                h_delta = 0 if h_ok else int(left - 10)
            else:
                # 完整可见才算到位
                h_ok = (left >= 0 and right <= vw - 4)
                h_delta = 0 if h_ok else int((left + right) / 2 - vw / 2)
            if v_ok and h_ok:
                return True
            # 先调垂直(自然滚轮),再调水平(window.scrollBy:左右滚动没有滚轮物理事件)
            if v_delta != 0:
                self._human_wheel(max(-600, min(600, v_delta)))
            if h_delta != 0:
                # 水平滚动:使用 deltaX(触控板/高端鼠标本身就有水平滚轮,不做 Shift 修饰)
                sign = 1 if h_delta >= 0 else -1
                notches = max(1, round(abs(h_delta) / 120))
                while notches > 0:
                    burst = min(notches, random.randint(1, 4))
                    for _ in range(burst):
                        self.driver.run_cdp(
                            "Input.dispatchMouseEvent", type="mouseWheel",
                            x=self._mouse[0], y=self._mouse[1], deltaX=sign * 120, deltaY=0,
                        )
                        time.sleep(random.uniform(0.03, 0.07))
                    notches -= burst
                    if notches > 0:
                        time.sleep(random.uniform(0.04, 0.10))
                time.sleep(random.uniform(0.05, 0.12))
            self._wait_scroll_settle()  # 等页面滚动惯性完全停下再读坐标,避免振荡
        return False

    def _wait_scroll_settle(self, timeout=3):
        """等滚动惯性停下再取坐标。"""
        t0 = time.time()
        last = None
        while time.time() - t0 < timeout:
            try:
                y = self.driver.run_js("return Math.round(window.scrollY);")
            except Exception:
                return
            if last is not None and abs(y - last) < 2:
                return
            last = y
            time.sleep(0.15)

    # ==================== 拟人化:键盘/浮层/标签页 ====================

    def _human_type_digits(self, el, text):
        """拟人输入数字:点击聚焦→Ctrl+A全选→逐键 CDP rawKeyDown+char+keyUp。"""
        if not self._human_click_el(el):
            el.click()
        time.sleep(random.uniform(0.15, 0.4))
        self.driver.run_cdp(
            "Input.dispatchKeyEvent", type="rawKeyDown",
            key="a", code="KeyA", modifiers=2, windowsVirtualKeyCode=65,
        )
        self.driver.run_cdp(
            "Input.dispatchKeyEvent", type="keyUp",
            key="a", code="KeyA", modifiers=2, windowsVirtualKeyCode=65,
        )
        time.sleep(random.uniform(0.05, 0.15))
        for ch in text:
            vk = ord(ch)
            code = f"Digit{ch}"
            self.driver.run_cdp(
                "Input.dispatchKeyEvent", type="rawKeyDown",
                key=ch, code=code, windowsVirtualKeyCode=vk,
            )
            self.driver.run_cdp(
                "Input.dispatchKeyEvent", type="char",
                text=ch, key=ch, code=code, windowsVirtualKeyCode=vk,
            )
            self.driver.run_cdp(
                "Input.dispatchKeyEvent", type="keyUp",
                key=ch, code=code, windowsVirtualKeyCode=vk,
            )
            time.sleep(random.uniform(0.06, 0.18))

    def _search_via_box(self, keyword, kw):
        """拟人搜索路径:搜索框点击→全选清空→insertText注入中文→回车提交。"""
        search_url = f"https://search.jd.com/Search?keyword={kw}&enc=utf-8&wq={kw}"
        self._close_unexpected_tabs()
        url_now = self.driver.url or ""
        if "www.jd.com" not in url_now and "search.jd.com" not in url_now:
            self.driver.get("https://www.jd.com/", timeout=15)
            self.sleep(3, 5)
        box = None
        for loc in ("css=#key", "css=input[placeholder*='\u641c\u7d22']",
                    "css=.search-m input[type='text']", "css=input[name='keyword']"):
            box = self.driver.ele(loc, timeout=2)
            if box:
                break
        if not box:
            box = self.driver.run_js(
                "var ins=document.querySelectorAll(\"input[type='text'],input:not([type])\");"
                "for(var i=0;i<ins.length;i++){var r=ins[i].getBoundingClientRect();"
                "if(r.width>100&&r.height>15&&r.top>=-50&&r.top<window.innerHeight){return ins[i];}}"
                "return null;"
            )
        if not box:
            logger.info("未找到搜索框(url=%s),退回 URL 直达", self.driver.url)
            self.driver.get(search_url, timeout=15)
            return search_url
        self._human_click_el(box)
        self.sleep(0.3, 0.8)
        self.driver.run_cdp(
            "Input.dispatchKeyEvent", type="rawKeyDown",
            key="a", code="KeyA", modifiers=2, windowsVirtualKeyCode=65,
        )
        self.driver.run_cdp(
            "Input.dispatchKeyEvent", type="keyUp",
            key="a", code="KeyA", modifiers=2, windowsVirtualKeyCode=65,
        )
        self.sleep(0.1, 0.3)
        # 拟人输入中文:CDP insertText 是模拟 IME 输入的唯一可靠方式
        # Chrome 对 CDP 输入事件标记 isTrusted=true,页面无法区分真人和 CDP
        # 中文逐键模拟不可行(需要用 IME composition 事件,CDP 不支持),
        # 剪贴板方案反而不自然(真人不会凭空有这段文字在剪贴板里)
        self.driver.run_cdp("Input.insertText", text=keyword)
        self.sleep(0.5, 1.2)
        self.driver.run_cdp("Input.dispatchKeyEvent", type="rawKeyDown", key="Enter", code="Enter", windowsVirtualKeyCode=13)
        self.driver.run_cdp("Input.dispatchKeyEvent", type="char", text="\r", key="Enter", code="Enter", windowsVirtualKeyCode=13)
        self.driver.run_cdp("Input.dispatchKeyEvent", type="keyUp", key="Enter", code="Enter", windowsVirtualKeyCode=13)
        self.sleep(4, 7)
        if kw in (self.driver.url or ""):
            logger.info("已通过搜索框搜索「%s」", keyword)
            return search_url
        logger.info("搜索框提交后未跳转(url=%s),退回 URL 直达", self.driver.url)
        self.driver.get(search_url, timeout=15)
        return search_url

    def _press_escape(self):
        """真实 CDP Esc 按键。"""
        try:
            self.driver.run_cdp(
                "Input.dispatchKeyEvent", type="rawKeyDown",
                key="Escape", code="Escape", windowsVirtualKeyCode=27,
            )
            self.driver.run_cdp(
                "Input.dispatchKeyEvent", type="keyUp",
                key="Escape", code="Escape", windowsVirtualKeyCode=27,
            )
        except Exception:
            pass

    def _hide_floater_at(self, x, y, el):
        """把 (x,y) 处挡住目标的浮层 DOM 隐藏(Esc 关不掉的浮层专用)。"""
        try:
            res = self.driver.run_js(
                "var t=arguments[2];"
                "var h=document.elementFromPoint(arguments[0],arguments[1]);"
                "if(!h||h===t||t.contains(h)||h.contains(t)) return 'ok';"
                "var n=h;"
                "while(n&&n!==document.body){"
                "var p=getComputedStyle(n).position;"
                "if(p==='fixed'||p==='absolute'){n.style.display='none';return 'hidden';}"
                "n=n.parentElement;}"
                "h.style.display='none';return 'hidden-el';",
                x, y, el,
            )
            logger.info("浮层已隐藏(%s)", res)
        except Exception:
            pass

    def _close_unexpected_tabs(self):
        """点击后误开新标签页(调查问卷等),关掉并回到原标签。"""
        try:
            ids = list(self.driver.tab_ids or [])
            mine = self.driver.tab_id
            extra = [t for t in ids if t != mine]
            if not extra:
                return
            for t in extra:
                logger.info("检测到误开新标签页,关闭: %s", t)
                try:
                    self.driver.close_tabs(t)
                except Exception:
                    pass
            try:
                self.driver.activate_tab(mine)
            except Exception:
                pass
        except Exception:
            pass

    # ==================== 异常检测 ====================

    def _captcha_marker(self):
        """命中验证码特征返回描述,否则 None。"""
        url = (self.driver.url or "").lower()
        if "passport.jd.com" in url:
            return None
        for key in CAPTCHA_URL_KEYS:
            if key in url:
                return f"url:{key}"
        html = self.driver.html or ""
        for marker in CAPTCHA_MARKERS:
            if marker in html:
                return f"html:{marker}"
        return None

    def _has_captcha(self):
        return self._captcha_marker() is not None

    def _interruption(self):
        """页面级异常判断:返回 'captcha'/'kicked'/None。"""
        url = (self.driver.url or "").lower()
        if "passport.jd.com" in url:
            return "kicked"
        for key in CAPTCHA_URL_KEYS:
            if key in url:
                return "captcha"
        html = self.driver.html or ""
        for marker in CAPTCHA_MARKERS:
            if marker in html:
                return "captcha"
        if self.driver.ele("xpath=//*[@class='link-login']", timeout=0.25):
            return "kicked"
        return None

    def _is_empty_result(self):
        """空结果页(软性限流):搜索页框架在但主商品区为空。"""
        url_now = (self.driver.url or "").lower()
        if "passport.jd.com" in url_now or any(k in url_now for k in CAPTCHA_URL_KEYS):
            return False
        try:
            html = self.driver.html or ""
        except Exception:
            return False
        return "\u6682\u672a\u627e\u5230\u76f8\u5173\u5546\u54c1" in html

    # ==================== 恢复流程 ====================

    def _reposition_to_page(self, page_no, kw):
        """导航回指定页码。"""
        if page_no <= 1:
            self.driver.refresh()
            self.sleep(4, 6)
            return True
        search_url = f"https://search.jd.com/Search?keyword={kw}&enc=utf-8&wq={kw}"
        self.driver.get(search_url, timeout=15)
        self.sleep(4, 6)
        return self._jump_to_page(page_no)

    def _wait_captcha_solved(self, timeout=600):
        """验证码等待:不操作页面,只轮询,等人工完成。"""
        logger.info("[验证码] 页面不做任何操作,一直等人工完成验证……")
        t0 = time.time()
        last_tip = 0
        while time.time() - t0 < timeout:
            if self._is_logged_out():
                self.sleep(3, 5)
                if self._is_logged_out():
                    logger.info("[验证码] 等待期间变成未登录,转登录流程")
                    return "kicked"
                continue
            if not self._has_captcha():
                self.sleep(2, 3)
                if not self._has_captcha():
                    logger.info("[验证码] 人工验证已通过")
                    return True
            if time.time() - last_tip > 20:
                last_tip = time.time()
                logger.info("[验证码] 等待人工验证中…… 已等待 %ds", int(time.time() - t0))
            time.sleep(3)
        logger.warning("[验证码] 等待人工验证超时")
        return False

    def _recover_from_captcha(self, page_no, kw):
        """验证码恢复:等人工验证通过后跳回原页续采。"""
        result = self._wait_captcha_solved()
        if result == "kicked":
            return self._recover_from_kick(page_no, kw)
        if not result:
            return False
        return self._reposition_to_page(page_no, kw)

    def _recover_from_kick(self, page_no, kw):
        """被踢恢复:先试自动登录,失败则等人工扫码,成功后跳回原页。"""
        logger.info("[被踢] 第 %d 页检测到被踢出登录", page_no)
        # 先试自动登录
        if self.login_password and self.login_username:
            if self._ensure_logged_in():
                logger.info("[被踢] 自动登录成功,跳回第 %d 页", page_no)
                return self._reposition_to_page(page_no, kw)
        # 自动登录失败/无凭据:等人工扫码
        logger.info("[被踢] 请在浏览器中扫码登录,脚本将持续等待……")
        t0 = time.time()
        while time.time() - t0 < 3600:
            url = self.driver.url or ""
            if "passport.jd.com" not in url:
                self.sleep(2, 4)
                if not self._is_logged_out():
                    logger.info("[被踢] 人工登录成功,跳回第 %d 页", page_no)
                    return self._reposition_to_page(page_no, kw)
            time.sleep(5)
        logger.warning("[被踢] 等待人工登录超时")
        return False

    def _recover_from_empty(self, page_no, kw):
        """空页恢复:像人一样刷新重试,访问频繁时切长间隔。"""
        logger.info("[空页] 第 %d 页搜索返回空结果,像人一样刷新重试", page_no)
        for attempt in range(1, 4):
            if "\u8bbf\u95ee\u9891\u7e41" in (self.driver.html or ""):
                wait = {1: 90, 2: 180, 3: 300}.get(attempt, 300)
                logger.info("[空页] 检测到访问频繁,等 %ds 后重试", wait)
                time.sleep(wait)
            else:
                self.sleep(18, 35)
            self.driver.refresh()
            self.sleep(6, 10)
            if not self._is_empty_result() and not self._has_captcha():
                logger.info("[空页] 第 %d 次刷新后数据恢复", attempt)
                return True
            logger.info("[空页] 第 %d 次刷新后仍为空", attempt)
        logger.warning("[空页] 多次刷新仍无数据,停止(今日额度可能已尽)")
        return False

    def _error_report(self, data):
        self.report_data.update(data)
        if self.scheduler:
            self.scheduler.stop()
            self.scheduler.post_report(self.report_data)

    def _success_report(self, data):
        self.report_data.update(data)
        if self.scheduler:
            self.scheduler.post_report(self.report_data)

    def post_report(self, data):
        url = "http://192.168.2.246:8080/api/collect_task/report"
        print('传给返回接口的数据', data)
        headers = {'X-Crawler-Token': CRAWLER_TOKEN}
        response = requests.post(url, json=data, headers=headers)
        result = response.json()
        if result.get('code') != 'success':
            logger.info(f'翻页回传：{result}')
            if self.scheduler:
                self.scheduler.stop()
        print(f'任务进度上传 {result}')

    def perform_jd_login(self):
        """
        使用已有浏览器实例执行京东账号密码登录（含滑块验证码）。
        成功返回 True，失败返回 False。
        """
        username = self.login_username
        password = self.login_password
        login_url = "https://passport.jd.com/new/login.aspx"
        self.driver.get(login_url)
        input_name = self.driver.ele("xpath=//input[@id='loginname']", timeout=15)
        if not input_name:
            print("未找到用户名输入框")
            return False

        input_name.input(username)
        time.sleep(random.uniform(1.5, 2.5))

        input_pass = self.driver.ele("xpath://input[@name='nloginpwd']", timeout=5)
        if not input_pass:
            print("未找到密码输入框")
            return False

        input_pass.input(password)
        time.sleep(random.uniform(1.5, 2.5))

        login_btn = self.driver.ele("xpath://a[@id='loginsubmit']", timeout=5)
        if not login_btn:
            print("未找到登录按钮")
            return False
        login_btn.click()

        time.sleep(random.uniform(3, 5))

        if not handle_jd_slider_captcha(self.driver):
            print("滑块验证码未通过")
            return False

        return True

    def _ensure_logged_in(self):
        """未登录时自动走登录流程（账号密码 + 滑块）。"""
        if not self._is_logged_out():
            return True

        logger.info("检测到未登录，开始自动登录: %s", self.account_name)
        ok = self.perform_jd_login()
        if ok and not self._is_logged_out():
            logger.info("自动登录成功: %s", self.account_name)
            return True

        logger.error("自动登录失败: %s", self.account_name)
        return False

    def _check_page_blocked(self):
        html = self.driver.html or ""
        if "抱歉由于访问频繁导致无法搜索" in html:
            logger.error("账号无法搜索（访问频繁）")
            self.success = False
            return True
        return False

    def _jump_to_page(self, target_page):
        """跳转到指定页码，并清空跳转前的监听残留。"""
        to_page_input = self.driver.ele(
            "xpath=//input[contains(@class, 'pagination-input')] | //div[contains(@class,'_pagination_toPageNum_')]//input[@type='text']",
            timeout=3,
        )
        if not to_page_input:
            logger.warning("未找到跳页输入框，无法跳转到第 %s 页", target_page)
            return False

        self.clear_listen_buffer()
        # 拟人输入:逐键 CDP 打字(失败自动退回直接赋值)
        try:
            self._human_type_digits(to_page_input, str(target_page))
        except Exception:
            to_page_input.clear()
            to_page_input.input(str(target_page))
        self.sleep(0.4, 0.8)
        
        confirm_btn = self.driver.ele("xpath=//button[contains(@class, 'pagination-jump-btn')]", timeout=1)
        if confirm_btn:
            self.driver.run_js("arguments[0].click();", confirm_btn)
        else:
            self.driver.actions.key_down("enter").key_up("enter")
            
        self.sleep(1.5, 2.5)
        self.clear_listen_buffer()
        logger.info("已跳转到第 %s 页", target_page)
        return True

    def _go_next_page(self, next_btn):
        self.clear_listen_buffer()
        if not self._human_click(next_btn):
            logger.warning("点击下一页失败")
            return False
        self._close_unexpected_tabs()
        self.sleep(0.6, 1.2)
        return True

    def crawl(self):
        total = 0
        keyword = self._build_search_keyword()

        # 如果已在京东页面(搜索页/首页)直接用当前页的搜索框,不必每次都回首页
        url_now = (self.driver.url or "").lower()
        if "jd.com" not in url_now:
            self.driver.get("https://www.jd.com/", timeout=15)
            self.sleep(3, 5)

        if self._is_logged_out():
            # 打开扫码登录页,等人工扫码(实时轮询,登录即继续,最多等5分钟)
            logger.info("未登录,打开扫码登录页,请用京东APP扫码……")
            self.driver.get("https://passport.jd.com/new/login.aspx", timeout=15)
            t0 = time.time()
            logged_in = False
            while time.time() - t0 < 300:  # 最多等5分钟
                if "passport.jd.com" not in (self.driver.url or ""):
                    self.sleep(2, 4)
                    if not self._is_logged_out():
                        logged_in = True
                        break
                time.sleep(3)
            if logged_in:
                logger.info("人工登录成功")
            else:
                logger.error("等待人工登录超时(5分钟)")
                if self.login_password and self.login_username:
                    self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': self.start_page, 'exception_type': 1})
                else:
                    self.disable_account()
                    send_text(f"京东：{self.account_name}账号登录失败")
                    self.success = False
                return

        kw = quote(str(keyword or ""), safe="")
        self._search_kw = kw
        # 必须先监听再打开搜索页，否则首屏 wareList（前约 30 条）在监听开启前就返回了
        self._start_listen()
        # 拟人搜索:搜索框输入(失败自动退回 URL 直达)
        self._search_via_box(keyword, kw)
        self.sleep(2, 3)

        if self._check_page_blocked():
            self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': self.start_page, 'exception_type': 2})
            return

        if not handle_jd_slider_captcha(self.driver, pause_listen=False):
            logger.warning("进入搜索页后滑块验证码处理失败")
            self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': self.start_page, 'exception_type': 2})
            self.success = False
            return

        if self.start_page > 1:
            if not self._jump_to_page(self.start_page):
                logger.warning("跳页失败，将从第 1 页开始采集")
                self.start_page = 1

        logger.info(
            "采集页码范围: %s ~ %s（共 %s 页）",
            self.start_page,
            self.end_page,
            self.end_page - self.start_page + 1,
        )

        page_no = self.start_page
        while page_no <= self.end_page:
            # ---- 页首异常检测 + 恢复 ----
            intr = self._interruption()
            if intr == "kicked":
                if self._recover_from_kick(page_no, kw):
                    continue
                self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': page_no, 'exception_type': 1})
                break
            elif intr == "captcha":
                if self._recover_from_captcha(page_no, kw):
                    continue
                self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': page_no, 'exception_type': 2})
                break
            elif self._is_empty_result():
                if self._recover_from_empty(page_no, kw):
                    continue
                self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': page_no, 'exception_type': 2})
                break

            if self._is_logged_out():
                if not self._ensure_logged_in():
                    self.success = False
                    break
                self._search_via_box(keyword, kw)
                self.sleep(3, 5)
                if page_no > 1:
                    self._jump_to_page(page_no)

            if not handle_jd_slider_captcha(self.driver, pause_listen=True):
                logger.warning("滑块验证码处理失败，停止采集")
                self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': page_no, 'exception_type': 2})
                self.success = False
                break

            if self._check_page_blocked():
                break

            logger.info("===== 正在爬取第 %s 页 =====", page_no)
            self.page_stored = 0
            search_ele = self.driver.ele("xpath=//div[@id='search-condition']", timeout=10)
            if not search_ele:
                logger.warning("未找到搜索结果区域，停止采集")
                self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': page_no, 'exception_type': 2})
                break
            # ---- 采集 ----
            try:
                page_n, _ = self.collect_full_page_items()
            except _PageInterrupted as pi:
                if pi.kind == "captcha":
                    if not self._recover_from_captcha(page_no, kw):
                        self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': page_no, 'exception_type': 2})
                        break
                    continue
                if pi.kind == "render":
                    logger.warning("第 %d 页渲染异常,重新加载本页重采", page_no)
                    if not self._reposition_to_page(page_no, kw):
                        break
                    continue
                # kicked
                if not self._recover_from_kick(page_no, kw):
                    self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': page_no, 'exception_type': 1})
                    break
                continue

            logger.info("本页监听商品条数（含可能重复）: %s", page_n)
            total += page_n
            logger.info("累计监听条数: %s", total)

            if self.scheduler and self.scheduler.end:
                logger.info('心跳失败')
                self._error_report({'is_finished': 0, 'need_reassign': 0, 'current_page': page_no})
                break

            if self.is_no_prodcut > JD_NO_MATCH_THRESHOLD:
                logger.info("连续无匹配商品过多，停止采集")
                self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page_no})
                break

            self.cumulative_pages += 1
            self.cumulative_items += page_n
            page_skipped = page_n - self.page_stored
            self.cumulative_stored += self.page_stored
            self.cumulative_skipped += page_skipped

            logger.info(
                "关键字 %s 第 %s 页获取完成, 本页获取量: %d, 本页已入库: %d, 本页未入库: %d | 账号测试总页数: %d, 账号测试总数据量: %d, 总已入库: %d, 总未入库: %d",
                keyword,
                page_no,
                page_n,
                self.page_stored,
                page_skipped,
                self.cumulative_pages,
                self.cumulative_items,
                self.cumulative_stored,
                self.cumulative_skipped
            )
            print(f"[{keyword}] 当前页数: {page_no}, 本页获取数据量: {page_n}, 本页已入库: {self.page_stored}, 本页未入库: {page_skipped} | 账号测试总页数: {self.cumulative_pages}, 账号测试总数据量: {self.cumulative_items}, 总已入库: {self.cumulative_stored}, 总未入库: {self.cumulative_skipped}")

            if page_no >= self.end_page:
                self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page_no})
                break

            next_btn = self.driver.ele("text=下一页", timeout=2)
            if not next_btn:
                logger.info("没有下一页（未找到）")
                self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page_no})
                break
            cls_str = next_btn.attr("class") or ""
            if "disabled" in cls_str:
                logger.info("没有下一页（已禁用）")
                self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page_no})
                break

            if not self._go_next_page(next_btn):
                break

            page_no += 1

        if self.scheduler:
            self.scheduler.stop()

    def run(self):
        # 检测账号
        if not self.get_account():
            logger.info("==================当前无账号可用==================")
            self.success = False
            return self.pipeline.crawl_count, self.success, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped
        logger.info("获取到账号:%s,代理ip:%s", self.account_name, self.ip)

        # # # 每次选取账号，立马账号使用时间
        update_sql = f""" UPDATE `retrieve_collect_equipment_account` SET `status`= %s, `update_time`= %s WHERE `username` = %s; """
        self.db.execute(update_sql, (0, int(time.time()), self.account_name))

        try:
            self.init_browser()
            self.crawl()
        except Exception as e:
            self.success = False
            logger.exception("爬取异常: %s", e)
            self._error_report({'is_finished': 0, 'need_reassign': 1, 'current_page': self.start_page, 'exception_type': 3})
            self.sleep(3, 5)

        finally:
            pass
            # if self.driver:
            #     self.driver.quit()
            #     self.driver = None
        return self.pipeline.crawl_count, self.success, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped