2.0/spiders/ybm/ybm_crawl.py

# ====================================================================================
# ybm_crawl.py — 药帮忙平台爬虫
# 遵循 commons1 调度框架，平台专属逻辑保留在类内
# 由 CollectScheduleRunner / run_scheduled_loop 调度执行
# ====================================================================================

# ==================== SECTION 1: IMPORTS ====================
import base64
import json
import os
import random
import re
import threading
import time
import traceback
from datetime import datetime
from io import BytesIO

import oss2
import requests
from PIL import Image
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError

# ---- commons1 公共能力 ----
from commons.Logger import get_spider_logger
from commons.conn_mysql import MySQLPoolOn2
from commons.feishu_webhook import send_text as feishu_send_text, send_error_card as feishu_send_error_card
from commons.scheduler import CrawlerScheduler

# ---- pipelines (基于 commons) ----
from pipelines.drug_pipelines import DrugPipeline

logger = get_spider_logger("ybm")

# ==================== SECTION 2: PLATFORM CONFIG ====================

# MySQL（供店铺信息直接读写；商品数据走 DrugPipeline）
mysql_pool = MySQLPoolOn2()

# 平台标识
PLATFORM_ID = 9
YBM_DEVICE_ID = "18008650300"

# API（内部调度系统）
TASK_API_BASE = "http://192.168.2.246:8080/api/collect_task"
TASK_API_TOKEN = "zhijiayun_crawler_2026"

TASK_PULL_URL = TASK_API_BASE + "/pull"
TASK_HEARTBEAT_URL = TASK_API_BASE + "/heartbeat"
TASK_REPORT_URL = TASK_API_BASE + "/report"

# 手动任务（调试用）
USE_MANUAL_TASK = False
MANUAL_TASK = {
    "id": 2001,
    "collect_task_id": 5000,
    "company_id": 1,
    "product_name": "三九胃泰颗粒",
    "product_specs": "10",
    "product_keyword": "",
    "product_brand": "999",
    "sampling_cycle": 1,
    "sampling_start_time": 1778083200,
    "sampling_end_time": 1778342399,
    "collect_equipment_account_id": 0,
    "collect_region_id": 0,
    "collect_equipment_id": 2,
    "collect_round": 1,
}

# 反爬参数
MIN_CLICK_DELAY = 1.5
MAX_CLICK_DELAY = 3.5
MIN_INPUT_DELAY = 0.1
MAX_INPUT_DELAY = 0.3
MIN_PAGE_DELAY = 2.0
MAX_PAGE_DELAY = 4.0
MIN_KEYWORD_DELAY = 8.0
MAX_KEYWORD_DELAY = 15.0

SCROLL_TARGET_DISTANCE = 400
SCROLL_OFFSET_RANGE = 50
SCROLL_STEP = 50
SCROLL_INTERVAL = 0.05

# Cookie & 登录
COOKIE_FILE_PATH = "ybm_cookies.json"
LOGIN_VALIDATE_URL = "https://www.ybm100.com/new/"
USERNAME = "18008650300"
PASSWORD = "12345678"
TARGET_LOGIN_URL = "https://www.ybm100.com/new/login"

# 选择器
USERNAME_SELECTOR = "input[placeholder*=请输入账号]"
PASSWORD_SELECTOR = "input[placeholder*=请输入密码]"
LOGIN_BTN_SELECTOR = "button:has(span:text('登录'))"
SEARCH_INPUT_SELECTOR = "input[placeholder*='药品名称/厂家名称']"
SEARCH_INPUT_SELECTOR2 = "div.home-search-container-search-head"
SEARCH_BTN_SELECTOR = "div.home-search-container-search-head-btn[data-scmd=\"text-搜索\"]"

PRODUCT_ITEM_SELECTOR = "div.product-list-item"
PRODUCT_TITLE_SELECTOR = "div.product-name"
PRODUCT_PRICE_SELECTOR = "div.main-price"
PRODUCT_STORE_SELECTOR = 'div.prduct-shop-name div.shop-name'
PRODUCT_COMPANY_SELECTOR = "div.product-manufacturer"
PRODUCT_VALIDITY_SELECTOR = "div.product-period"

# 等待时间
ELEMENT_TIMEOUT = 10000
LOGIN_AFTER_CLICK = 5000
SEARCH_BTN_TIMEOUT = 5000
COLLECT_DELAY = 3000
DETAIL_LOAD_TIMEOUT = 5000

# 浏览器
BROWSER_HEADLESS = False
BROWSER_CHANNEL = "chrome"
SLOW_MO_MIN = 50
SLOW_MO_MAX = 100

# 百度 OCR
request_url_config = "https://aip.baidubce.com/rest/2.0/ocr/v1/business_license"
AppKey_config = "tRK2RhyItCSh6BzyT4CNVXQa"
AppSecret_config = "TDgKiPo94i2mOM1sDqOuDnlcK1bG66jh"
token_url_config = 'https://aip.baidubce.com/oauth/2.0/token'

# OSS
OSS_ACCESS_KEY_ID = 'LTAI5tDwjfteBvivYN41r8sJ'
OSS_ACCESS_KEY_SECRET = 'yowuOGi2nYYnrqGpO3qcz94C4brcPp'
OSS_ENDPOINT = "oss-cn-shenzhen.aliyuncs.com"
OSS_BUCKET_NAME = "zhijiayun-jiansuo"
OSS_PREFIX = "scrape_data/"
LOCAL_SCREENSHOT_DIR = "local_screenshots"
LOCAL_CROPPED_DIR = "./local_cropped_screenshots"
IMAGE_COMPRESS_ENABLE = True
IMAGE_COMPRESS_QUALITY = 30
IMAGE_COMPRESS_PNG_LEVEL = 9

# 心跳上报间隔（秒）
HEARTBEAT_INTERVAL_SECONDS = 60


# ==================== SECTION 3: OSS / SCREENSHOT HELPERS ====================

def init_local_screenshot_dir():
    if not os.path.exists(LOCAL_SCREENSHOT_DIR):
        os.makedirs(LOCAL_SCREENSHOT_DIR)
        logger.info(f"本地截图目录已创建: {LOCAL_SCREENSHOT_DIR}")
    else:
        logger.debug(f"本地截图目录已存在: {LOCAL_SCREENSHOT_DIR}")


def init_oss_bucket():
    try:
        auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        bucket = oss2.Bucket(auth, OSS_ENDPOINT, OSS_BUCKET_NAME)
        bucket.get_bucket_info()
        logger.info("OSS Bucket 初始化成功")
        return bucket
    except Exception as e:
        logger.error(f"OSS Bucket 初始化失败: {str(e)}")
        raise


def upload_local_screenshot_to_oss(bucket, local_file_path, oss_file_path=None):
    if not os.path.exists(local_file_path):
        raise FileNotFoundError(f"本地截图文件不存在: {local_file_path}")
    if not oss_file_path:
        local_file_name = os.path.basename(local_file_path)
        oss_file_path = f"screenshots/{local_file_name}"
    try:
        bucket.put_object_from_file(oss_file_path, local_file_path)
        oss_file_url = f"https://{OSS_BUCKET_NAME}.{OSS_ENDPOINT}/{oss_file_path}"
        logger.info(f"截图上传 OSS 成功: {oss_file_url}")
        return oss_file_url
    except Exception as e:
        logger.error(f"截图上传 OSS 失败: {str(e)}")
        raise


def crop_local_screenshot(local_file_path, cropped_file_path=None, crop_region=None):
    if not os.path.exists(local_file_path):
        raise FileNotFoundError(f"原始截图文件不存在: {local_file_path}")
    os.makedirs(LOCAL_CROPPED_DIR, exist_ok=True)
    if not cropped_file_path:
        file_name = os.path.basename(local_file_path)
        file_name_no_ext, file_ext = os.path.splitext(file_name)
        cropped_file_name = f"{file_name_no_ext}_cropped{file_ext}"
        cropped_file_path = os.path.join(LOCAL_CROPPED_DIR, cropped_file_name)
    with Image.open(local_file_path) as img:
        img_width, img_height = img.size
        if not crop_region:
            crop_region = (0, 0, int(img_width), int(img_height * 0.3))
        c_left, c_upper, c_right, c_lower = crop_region
        if c_right > img_width or c_lower > img_height or c_left < 0 or c_upper < 0:
            raise ValueError(f"裁剪区域超出图片范围，图片尺寸=({img_width}, {img_height})，裁剪区域={crop_region}")
        cropped_img = img.crop(crop_region)
        file_ext = os.path.splitext(cropped_file_path)[1].lower()
        try:
            if IMAGE_COMPRESS_ENABLE:
                if file_ext in ['.jpg', '.jpeg']:
                    cropped_img.save(cropped_file_path, format='JPEG', quality=IMAGE_COMPRESS_QUALITY, optimize=True, progressive=True)
                else:
                    cropped_img.save(cropped_file_path)
            else:
                cropped_img.save(cropped_file_path, format='JPEG')
        except Exception:
            cropped_img.save(cropped_file_path, format='JPEG')
    try:
        if os.path.exists(cropped_file_path):
            os.remove(local_file_path)
    except OSError:
        pass
    return cropped_file_path


def screenshot_target_page_to_local_then_oss(target_page, local_file_path=None, oss_file_path=None, full_page=True, crop_region=None):
    os.makedirs(LOCAL_SCREENSHOT_DIR, exist_ok=True)
    if not local_file_path:
        current_time = datetime.now().strftime("%Y%m%d_%H%M%S")
        local_file_name = f"{current_time}_p{PLATFORM_ID}_target_page.jpg"
        local_file_path = os.path.join(LOCAL_SCREENSHOT_DIR, local_file_name)
    logger.info(f"开始页面截图: {local_file_path}")
    target_page.screenshot(path=local_file_path, full_page=full_page, omit_background=False, timeout=10000)
    cropped_file_path = crop_local_screenshot(local_file_path=local_file_path, crop_region=crop_region)
    bucket = init_oss_bucket()
    oss_file_url = upload_local_screenshot_to_oss(bucket, cropped_file_path, oss_file_path)
    return cropped_file_path, oss_file_url


# ==================== SECTION 4: CITY / PROVINCE MAPPING ====================

CITY_JSON_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "..", "commons", "city.json")

PROVINCE_ID_MAP = {}
CITY_ID_MAP = {}
CITY_TO_PROVINCES_MAP = {}
DIRECT_MUNICIPALITIES = {"北京市", "上海市", "天津市", "重庆市"}
DIRECT_MUNICIPALITY_BASE_NAMES = {"北京", "上海", "天津", "重庆"}
DIRECT_MUNICIPALITY_ALIAS = {
    "北京": "北京市", "上海": "上海市", "天津": "天津市", "重庆": "重庆市",
}


def load_city_mapping():
    global PROVINCE_ID_MAP, CITY_ID_MAP, CITY_TO_PROVINCES_MAP
    PROVINCE_ID_MAP.clear()
    CITY_ID_MAP.clear()
    CITY_TO_PROVINCES_MAP.clear()
    if not os.path.exists(CITY_JSON_PATH):
        logger.error(f"❌ 城市JSON文件不存在：{CITY_JSON_PATH}")
        return
    try:
        with open(CITY_JSON_PATH, "r", encoding="utf-8") as f:
            data = json.load(f)
        for province_item in data:
            p_name = province_item['name']
            PROVINCE_ID_MAP[p_name] = province_item['id']
            for city_item in province_item.get('sons', []):
                c_name = city_item['name']
                CITY_ID_MAP[(p_name, c_name)] = city_item['id']
                CITY_TO_PROVINCES_MAP.setdefault(c_name, set()).add(p_name)
        logger.info(f"✅ 城市映射加载完成，共 {len(PROVINCE_ID_MAP)} 个省份，{len(CITY_ID_MAP)} 个城市")
    except Exception as e:
        logger.error(f"❌ 加载城市JSON失败：{str(e)}")


def _clean_province_name(name: str) -> str:
    return (name or "").replace("省", "").replace("市", "").replace("自治区", "").replace("特别行政区", "").strip()


def _clean_city_name(name: str) -> str:
    return (name or "").replace("市", "").replace("自治州", "").replace("地区", "").replace("盟", "").strip()


def normalize_province_city_names(province_name: str, city_name: str):
    province = (province_name or "").strip()
    city = (city_name or "").strip()
    if province and province not in DIRECT_MUNICIPALITIES and province not in PROVINCE_ID_MAP:
        clean_p = _clean_province_name(province)
        for standard_name in PROVINCE_ID_MAP.keys():
            if clean_p and clean_p == _clean_province_name(standard_name):
                province = standard_name
                break
    if not province and city:
        matched_provinces = CITY_TO_PROVINCES_MAP.get(city, set())
        if not matched_provinces:
            clean_c = _clean_city_name(city)
            if clean_c:
                matched_provinces = {
                    p_name
                    for (p_name, c_name) in CITY_ID_MAP.keys()
                    if _clean_city_name(c_name) == clean_c
                }
        if len(matched_provinces) == 1:
            province = next(iter(matched_provinces))
        elif len(matched_provinces) > 1:
            logger.warning(f"⚠️ 城市名存在跨省重名，无法唯一反推省份: city={city}, candidates={sorted(matched_provinces)}")
    if province in DIRECT_MUNICIPALITY_BASE_NAMES:
        province = DIRECT_MUNICIPALITY_ALIAS[province]
    if province and city and (province, city) not in CITY_ID_MAP:
        clean_c = _clean_city_name(city)
        for (p_name, c_name), _ in CITY_ID_MAP.items():
            if _clean_province_name(p_name) == _clean_province_name(province) and clean_c and _clean_city_name(c_name) == clean_c:
                city = c_name
                break
    if province in DIRECT_MUNICIPALITIES and not city:
        city = province
    return province, city


def get_province_city_ids(province_name, city_name):
    province_name, city_name = normalize_province_city_names(province_name, city_name)
    province_id = PROVINCE_ID_MAP.get(province_name) if province_name else None
    if province_name and province_id is None:
        clean_p = _clean_province_name(province_name)
        for name, pid in PROVINCE_ID_MAP.items():
            if clean_p and clean_p == _clean_province_name(name):
                province_id = pid
                province_name = name
                break
        if province_id is None:
            logger.warning(f"⚠️ 未找到省份ID: {province_name}")
            province_id = 0
    elif province_id is None:
        province_id = 0
    if _clean_province_name(province_name) in DIRECT_MUNICIPALITY_BASE_NAMES and not city_name:
        city_name = f"{_clean_province_name(province_name)}市"
    city_id = CITY_ID_MAP.get((province_name, city_name)) if province_name and city_name else None
    if province_name and city_name and city_id is None:
        clean_c = _clean_city_name(city_name)
        for (p_name, c_name), cid in CITY_ID_MAP.items():
            if p_name == province_name:
                if clean_c and clean_c == _clean_city_name(c_name):
                    city_id = cid
                    city_name = c_name
                    break
        if city_id is None:
            if _clean_province_name(province_name) in DIRECT_MUNICIPALITY_BASE_NAMES and province_id:
                city_id = province_id
            else:
                logger.warning(f"⚠️ 未找到城市ID: {province_name} - {city_name}")
                city_id = 0
    elif city_id is None:
        city_id = 0
    return province_id, city_id


def extract_province_city(address):
    if not address:
        return "", ""
    province_pattern = re.compile(r'([^省]+省|.+自治区|北京市|上海市|天津市|重庆市|.+特别行政区)')
    province_match = province_pattern.search(address)
    province = province_match.group(1) if province_match else ""
    address_remain = address.replace(province, "").strip() if province else address.strip()
    city_pattern = re.compile(r'([^市]+市|.+自治州|.+地区|.+盟|^[^\d区县镇]+)')
    city_match = city_pattern.search(address_remain)
    city = city_match.group(1).strip() if city_match else ""
    if province in ["北京市", "上海市", "天津市", "重庆市"]:
        city = province
    if not province and not city:
        simple_pattern = re.compile(r'^([^\d区县镇]+)')
        simple_match = simple_pattern.search(address)
        if simple_match:
            city = simple_match.group(1).strip()
    if city and province and city != province and province in city:
        city = city.replace(province, "").strip()
    province, city = normalize_province_city_names(province, city)
    return province.strip(), city.strip()


# ==================== SECTION 5: ANTI-CRAWL UTILITIES ====================

def get_current_time():
    return datetime.now().strftime('%Y-%m-%d %H:%M:%S')


def random_delay(min_seconds, max_seconds):
    delay = random.uniform(min_seconds, max_seconds)
    time.sleep(delay)
    return delay


def save_cookies(context, cookie_path=COOKIE_FILE_PATH):
    try:
        cookies = context.cookies()
        with open(cookie_path, "w", encoding="utf-8") as f:
            json.dump(cookies, f, ensure_ascii=False, indent=2)
        logger.info(f"Cookie已保存到：{cookie_path}")
        return True
    except Exception as e:
        logger.error(f" 保存Cookie失败：{e}")
        return False


def load_cookies(context, cookie_path=COOKIE_FILE_PATH):
    if not os.path.exists(cookie_path):
        logger.warning(f" Cookie文件不存在：{cookie_path}")
        return False
    try:
        with open(cookie_path, "r", encoding="utf-8") as f:
            cookies = json.load(f)
        context.add_cookies(cookies)
        logger.info(f"✅ 已从{cookie_path}加载Cookie")
        return True
    except Exception as e:
        logger.error(f" 加载Cookie失败：{e}")
        return False


def slow_scroll_400px(page, scroll_distance1=400):
    try:
        scroll_distance = random.randint(scroll_distance1 - SCROLL_OFFSET_RANGE, scroll_distance1 + SCROLL_OFFSET_RANGE)
        remaining_distance = scroll_distance
        total_steps = int(scroll_distance / SCROLL_STEP)
        logger.info(f"📜 开始慢速滚动（目标距离：{scroll_distance}px，总步数：{total_steps}，总时长约{total_steps*SCROLL_INTERVAL:.2f}秒）")
        for _ in range(total_steps):
            step = min(SCROLL_STEP, remaining_distance)
            page.evaluate(f"window.scrollBy(0, {step});")
            remaining_distance -= step
            time.sleep(SCROLL_INTERVAL)
        if remaining_distance > 0:
            page.evaluate(f"window.scrollBy(0, {remaining_distance});")
            time.sleep(SCROLL_INTERVAL)
        page.wait_for_load_state("networkidle", timeout=8000)
        random_delay(2.0, 3.0)
        logger.info(f" 慢速滚动完成，实际滚动距离：{scroll_distance - remaining_distance}px")
        return True
    except Exception as e:
        logger.warning(f" 慢速滚动失败：{e}")
        return False


def type_slow(locator, text: str, min_delay=0.06, max_delay=0.18):
    for ch in text:
        locator.type(ch, delay=int(random.uniform(min_delay, max_delay) * 1000))


def kill_masks(page):
    page.evaluate(r"""
    () => {
      const knownSelectors = [
        '.v-modal', '.el-overlay', '.el-overlay-dialog', '.el-dialog__wrapper',
        '.el-message-box__wrapper', '.el-loading-mask', '.el-popup-parent--hidden'
      ];
      for (const sel of knownSelectors) {
        document.querySelectorAll(sel).forEach(el => el.remove());
      }
      const all = Array.from(document.querySelectorAll('body *'));
      for (const el of all) {
        const s = window.getComputedStyle(el);
        if (!s) continue;
        const z = parseInt(s.zIndex || '0', 10);
        if ((s.position === 'fixed' || s.position === 'absolute') && z >= 1000 && s.pointerEvents !== 'none') {
          const r = el.getBoundingClientRect();
          const nearFullScreen = r.width >= window.innerWidth * 0.8 && r.height >= window.innerHeight * 0.8;
          if (nearFullScreen) {
            el.style.pointerEvents = 'none';
            el.style.display = 'none';
          }
        }
      }
      document.documentElement.style.overflow = 'auto';
      document.body.style.overflow = 'auto';
      document.body.style.position = 'static';
      document.body.style.width = 'auto';
      document.body.style.paddingRight = '0px';
      document.body.classList.remove('el-popup-parent--hidden');
    }
    """)


def force_close_popup(page):
    try:
        for _ in range(5):
            btn = page.locator(
                "//button[normalize-space()='下一步' or normalize-space()='完成' or normalize-space()='我知道了' or normalize-space()='关闭']"
            ).first
            if btn.count() > 0 and btn.is_visible():
                btn.click(timeout=1500)
                page.wait_for_timeout(300)
                continue
            close_icon = page.locator(
                "xpath=//*[contains(@class,'close') or contains(@class,'el-icon-close') or name()='svg' or name()='i'][1]"
            ).first
            if close_icon.count() > 0 and close_icon.is_visible():
                close_icon.click(timeout=1000)
                page.wait_for_timeout(300)
                continue
            break
        page.evaluate("""
        const selectors = [
          '.v-modal', '.el-overlay', '.el-overlay-dialog', '.el-dialog__wrapper',
          '[class*="mask"]', '[class*="overlay"]', '[style*="z-index"]'
        ];
        for (const sel of selectors) {
          document.querySelectorAll(sel).forEach(el => {
            const s = window.getComputedStyle(el);
            if ((s.position === 'fixed' || s.position === 'absolute') && parseInt(s.zIndex || '0', 10) >= 1000) {
              el.remove();
            }
          });
        }
        """)
    except Exception:
        pass


def popup_guard(page, tag=""):
    try:
        page.wait_for_timeout(300)
        for _ in range(6):
            btn = page.locator(
                "xpath=//button[normalize-space()='下一步' or normalize-space()='完成' or normalize-space()='我知道了' or normalize-space()='关闭']"
            ).first
            if btn.count() > 0 and btn.is_visible():
                btn.click(timeout=1500)
                page.wait_for_timeout(250)
                continue
            close_btn = page.locator(
                "css=.el-dialog__headerbtn, .el-message-box__headerbtn, .close, .icon-close, .el-icon-close"
            ).first
            if close_btn.count() > 0 and close_btn.is_visible():
                close_btn.click(timeout=1200)
                page.wait_for_timeout(250)
                continue
            break
        page.evaluate(r"""
        () => {
          const selectors = [
            '.v-modal', '.el-overlay', '.el-overlay-dialog', '.el-dialog__wrapper',
            '.el-message-box__wrapper', '.el-loading-mask'
          ];
          selectors.forEach(sel => document.querySelectorAll(sel).forEach(e => e.remove()));
          const all = Array.from(document.querySelectorAll('body *'));
          for (const el of all) {
            const s = getComputedStyle(el);
            const z = parseInt(s.zIndex || '0', 10);
            if ((s.position === 'fixed' || s.position === 'absolute') && z >= 1000 && s.pointerEvents !== 'none') {
              const r = el.getBoundingClientRect();
              const nearFull = r.width >= innerWidth * 0.8 && r.height >= innerHeight * 0.8;
              if (nearFull) {
                el.style.pointerEvents = 'none';
                el.style.display = 'none';
              }
            }
          }
          document.documentElement.style.overflow = 'auto';
          document.body.style.overflow = 'auto';
          document.body.classList.remove('el-popup-parent--hidden');
        }
        """)
        logger.info("杀除弹窗成功")
    except Exception:
        pass


# ==================== SECTION 6: OCR HELPERS ====================

def compress_image(image_data, max_size=4*1024*1024):
    try:
        img = Image.open(BytesIO(image_data))
        if img.mode in ('RGBA', 'P'):
            bg_img = Image.new('RGB', img.size, (255, 255, 255))
            bg_img.paste(img, mask=img.split()[-1] if img.mode == 'RGBA' else None)
            img = bg_img
        if img.width > 1000:
            ratio = 1000 / img.width
            img = img.resize((int(img.width * ratio), int(img.height * ratio)), Image.Resampling.LANCZOS)
        output = BytesIO()
        img.save(output, format='JPEG', quality=80)
        compressed_data = output.getvalue()
        if len(compressed_data) > max_size:
            output2 = BytesIO()
            img.save(output2, format='JPEG', quality=60)
            compressed_data = output2.getvalue()
        return compressed_data
    except Exception as e:
        logger.debug(f"图片压缩失败：{e}")
        return image_data


def download_image_to_base64(image_url, save_dir="./download_images"):
    try:
        if not os.path.exists(save_dir):
            os.makedirs(save_dir)
    except Exception as e:
        print(f"创建保存目录失败：{str(e)}")
        return None
    try:
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        }
        response = requests.get(image_url, headers=headers, timeout=15)
        response.raise_for_status()
        compressed_data = compress_image(response.content)
        image_base64 = base64.b64encode(compressed_data).decode("utf-8")
        image_data = compressed_data
        file_name = image_url.split("/")[-1]
        file_name = file_name.replace("?", "").replace("&", "").replace("=", "")
        save_path = os.path.join(save_dir, file_name)
        with open(save_path, "wb") as f:
            f.write(image_data)
        print(f"图片已保存到本地：{save_path}")
        return image_base64
    except requests.exceptions.Timeout:
        print(f"下载图片超时：{image_url}")
        return None
    except requests.exceptions.HTTPError as e:
        code = e.response.status_code if e.response is not None else "?"
        logger.warning("下载图片 HTTP 错误 url=%s status=%s", image_url, code)
        return None
    except Exception as e:
        print(f"下载图片失败：{str(e)}")
        return None


def get_access_token():
    url = f"{token_url_config}?grant_type=client_credentials&client_id={AppKey_config}&client_secret={AppSecret_config}"
    headers = {'Content-Type': 'application/json', 'Accept': 'application/json'}
    try:
        response = requests.request("POST", url, headers=headers, data="")
        response.raise_for_status()
        return response.json()['access_token']
    except Exception as e:
        print(f"获取access_token失败：{str(e)}")
        return None


def get_ocr_res(img):
    try:
        print(f'开始识别图片:{img}')
        img_base64 = download_image_to_base64(img)
        if not img_base64:
            print("图片下载/转Base64失败，终止OCR识别")
            return None
        access_token = get_access_token()
        if not access_token:
            print("获取access_token失败，无法调用OCR接口")
            return None
        request_url = request_url_config + "?access_token=" + access_token
        headers = {'content-type': 'application/x-www-form-urlencoded'}
        response = requests.post(request_url, data={"image": img_base64}, headers=headers)
        if response:
            res = response.json()
            if "error_code" in res:
                print(f"百度OCR接口错误：{res['error_msg']}（错误码：{res['error_code']}）")
                return None
            new_dic = {}
            for ite in res['words_result'].keys():
                new_dic[ite] = res['words_result'][ite]['words']
            print('资质数据信息', new_dic)
            return new_dic
        else:
            print("OCR接口返回空响应")
            return None
    except requests.exceptions.RequestException as e:
        print(f"网络错误（图片下载/OCR请求失败）：{str(e)}")
        return None
    except KeyError as e:
        print(f"OCR响应格式异常，缺失字段：{str(e)}")
        return None
    except Exception as e:
        print(f"OCR识别未知错误：{str(e)}")
        return None


# ==================== SECTION 7: SHOP HELPERS ====================

def clean_shop_name(raw_shop_name):
    if not raw_shop_name:
        return ''
    pattern = r'【.*?】|\(.*?\)|\[.*?\]'
    cleaned = re.sub(pattern, '', raw_shop_name)
    cleaned = cleaned.strip().replace('\n', '').replace('\r', '')
    cleaned = re.sub(r'\s+', ' ', cleaned)
    return cleaned if cleaned else raw_shop_name


def shop_is_exists_database(shop):
    query_sql = """
        SELECT province, city, business_license_company, qualification_number, business_license_address
        FROM retrieve_scrape_shop_info
        WHERE shop = %s and platform = %s
    """
    try:
        rows = mysql_pool.select_data(query_sql, (shop, 9))
        result = rows[0] if rows else None
        logger.debug("店铺存在校验 shop=%r result=%s", shop, result)
        is_exists = bool(result)
        if is_exists:
            logger.info(f"【店铺存在校验】店铺已存在 | 店铺名：{repr(shop)} | 结果：存在（True）不要执行采集店铺")
        else:
            logger.info(f"【店铺存在校验】店铺不存在 | 店铺名：{repr(shop)} | 结果：不存在（False）")
        return is_exists, result
    except Exception as e:
        logger.error(f"查询店铺失败：{e}")
        return False, None


def insert_shop_info_to_db(shop, contact_address, qualification_number, business_license_company,
                           business_license_address, scrape_date, platform, province, city, create_time, update_time):
    sql = """
        INSERT INTO retrieve_scrape_shop_info (
            shop, contact_address, qualification_number, business_license_company,
            business_license_address, scrape_date, platform, province, city, create_time, update_time
        ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON DUPLICATE KEY UPDATE
        contact_address = VALUES(contact_address),
        qualification_number = VALUES(qualification_number),
        business_license_company = VALUES(business_license_company),
        business_license_address = VALUES(business_license_address),
        scrape_date = VALUES(scrape_date),
        platform = VALUES(platform),
        province = VALUES(province),
        city = VALUES(city),
        update_time = VALUES(update_time)
    """
    params = (shop, contact_address, qualification_number, business_license_company,
              business_license_address, scrape_date, platform, province, city, create_time, update_time)
    conn = None
    cursor = None
    try:
        conn = mysql_pool.get_conn()
        cursor = conn.cursor()
        cursor.execute(sql, params)
        conn.commit()
        logger.info("店铺信息写入成功 shop=%s company=%s", shop, business_license_company)
        return True
    except Exception as e:
        logger.error("插入店铺失败：%s\n%s", e, traceback.format_exc())
        if conn:
            conn.rollback()
        return False
    finally:
        if cursor:
            cursor.close()
        if conn:
            conn.close()


# ==================== SECTION 8: YbmCrawler CLASS ====================

class YbmCrawler:
    """药帮忙平台爬虫 —— 遵循 commons1 调度框架。

    由 CollectScheduleRunner 调用：
        crawler = YbmCrawler(task_dict)
        count, is_success = crawler.run()
    """

    def __init__(self, drug_dict=None):
        self.task_dict = drug_dict or {}
        self.platform_id = PLATFORM_ID
        self.username = YBM_DEVICE_ID
        self.platform_name = '药帮忙'

        # Pipeline（商品数据走 DrugPipeline，做去重入库）
        self.pipeline = DrugPipeline("ybm")
        self.is_success = True

        # 由 CrawlerScheduler 管理心跳和停止信号
        self.scheduler = CrawlerScheduler(YBM_DEVICE_ID, str(PLATFORM_ID), heartbeat_interval=HEARTBEAT_INTERVAL_SECONDS)

        if self.task_dict:
            self._extract_task_params()

    # ---------- task params ----------

    def _extract_task_params(self):
        t = self.task_dict
        self.task_id = t.get("id", 0)
        self.collect_task_id = t.get("collect_task_id", 0)
        self.company_id = t.get("company_id", 0)
        self.product_name = t.get("product_name", "")
        self.product_specs = t.get("product_specs", "")
        self.product_keyword = t.get("product_keyword", "")
        self.product_brand = t.get("product_brand", "")
        self.sampling_cycle = t.get("sampling_cycle", 1)
        self.sampling_start_time = t.get("sampling_start_time", 0)
        self.sampling_end_time = t.get("sampling_end_time", 0)
        self.collect_equipment_account_id = t.get("collect_equipment_account_id", 0)
        self.collect_region_id = t.get("collect_region_id", 0)
        self.collect_round = t.get("collect_round", 1)
        self.api_current_page = t.get("current_page", 0) or 0

        self.keyword = self.product_brand + self.product_name

        self.collect_config_info = json.dumps({
            "sampling_cycle": self.sampling_cycle,
            "sampling_start_time": self.sampling_start_time,
            "sampling_end_time": self.sampling_end_time,
        }, ensure_ascii=False, default=str)

    # ---------- API report helpers ----------

    def _send_report(self, params: dict, log_msg: str):
        """通用回告方法，POST 到内部调度 API"""
        try:
            resp = requests.post(
                TASK_REPORT_URL, json=params, timeout=10,
                headers={'X-Crawler-Token': TASK_API_TOKEN},
            )
            if resp.status_code == 200:
                try:
                    data = resp.json()
                    if data.get("code") == "success":
                        logger.info("✅ 回告成功：%s", log_msg)
                        return "success"
                    else:
                        msg = data.get("msg", "")
                        if "已达平台限额" in msg or "限额" in msg:
                            logger.warning("⛔ 回告接口返回限额错误：code=%s, msg=%s，参数：%s", data.get('code'), msg, params)
                            return "limit_reached"
                        else:
                            logger.warning("⚠️ 回告接口返回错误：code=%s, msg=%s，参数：%s", data.get('code'), msg, params)
                            return "error"
                except ValueError:
                    logger.warning("⚠️ 回告响应非 JSON：%s，参数：%s", resp.text[:200], params)
                    return "error"
            else:
                logger.warning("⚠️ 回告 HTTP %s，参数：%s", resp.status_code, params)
                return "error"
        except Exception as e:
            logger.error("❌ 回告失败：%s，错误：%s", log_msg, str(e))
            return "error"

    def _report_start(self):
        """上报采集开始（is_finished=0）"""
        payload = {
            "task_id": self.task_id,
            "platform": str(self.platform_id),
            "username": self.username,
            "current_page": 0,
            "total_pages": 0,
            "is_finished": 0,
        }
        logger.info("上报开始 task_id=%s keyword=%s", self.task_id, self.keyword)
        return self._send_report(payload, f"开始上报（is_finished=0）关键词：{self.keyword}")

    def _report_progress(self, current_page: int, total_pages: int = 0):
        """上报采集进度"""
        payload = {
            "task_id": self.task_id,
            "platform": str(self.platform_id),
            "username": self.username,
            "current_page": current_page,
            "total_pages": total_pages,
            "is_finished": 0,
        }
        return self._send_report(payload, f"进度上报 page={current_page}")

    def _report_end(self, success: bool, real_count: int, page_no: int = 0,
                    exception_type: int = None, need_reassign: int = None, total_pages: int = 0):
        """上报采集结束（is_finished=1 或 0）"""
        payload = {
            "task_id": self.task_id,
            "platform": str(self.platform_id),
            "username": self.username,
            "current_page": page_no,
            "total_pages": total_pages if total_pages > 0 else page_no,
            "crawled_count": real_count,
            "is_finished": 1 if success else 0,
        }
        if exception_type is not None:
            payload["exception_type"] = exception_type
        if need_reassign is not None:
            payload["need_reassign"] = need_reassign
        logger.info("上报结束 task_id=%s keyword=%s payload=%s", self.task_id, self.keyword, payload)
        self._send_report(payload, f"结束上报（is_finished=1）关键词：{self.keyword}")

    # ---------- login ----------

    def _is_login(self, page):
        try:
            page.goto(LOGIN_VALIDATE_URL, timeout=300000)
            page.wait_for_load_state("networkidle")
            if "login" in page.url.lower():
                logger.warning(" Cookie失效，需要重新登录")
                return False
            logger.info(" Cookie有效，已保持登录状态")
            return True
        except Exception as e:
            logger.error(f" 验证登录状态失败：{e}")
            return False

    def _login_operation(self, page):
        try:
            page.wait_for_selector(USERNAME_SELECTOR, timeout=ELEMENT_TIMEOUT, state="visible")
            page.wait_for_timeout(timeout=3000)
            page.fill(USERNAME_SELECTOR, USERNAME)
            logger.info(" 已输入登录账号")
            page.wait_for_selector(PASSWORD_SELECTOR, timeout=ELEMENT_TIMEOUT, state="visible")
            page.wait_for_timeout(timeout=3000)
            page.fill(PASSWORD_SELECTOR, PASSWORD)
            logger.info(" 已输入登录密码")
            random_delay(1, 2)
            agree_btn = page.locator('span.el-checkbox__inner')
            agree_btn.click()
            page.wait_for_selector(LOGIN_BTN_SELECTOR, timeout=ELEMENT_TIMEOUT)
            page.wait_for_timeout(timeout=3000)
            page.click(LOGIN_BTN_SELECTOR)
            logger.info(" 已点击登录按钮")
            page.wait_for_timeout(LOGIN_AFTER_CLICK)
            return True
        except PlaywrightTimeoutError as e:
            logger.error(f" 登录失败：元素定位超时 - {str(e)}")
            return False
        except Exception as e:
            logger.error(f" 登录异常：{str(e)}")
            return False

    # ---------- search ----------

    @staticmethod
    def _pick_search_input(page):
        inputs = page.locator(SEARCH_INPUT_SELECTOR)
        cnt = inputs.count()
        for i in range(min(cnt, 2)):
            candidate = inputs.nth(i)
            try:
                candidate.wait_for(state="visible", timeout=1500)
                if candidate.is_enabled():
                    return candidate
            except PlaywrightTimeoutError:
                continue
        candidate = page.locator(f"{SEARCH_INPUT_SELECTOR}:visible").first
        candidate.wait_for(state="visible", timeout=ELEMENT_TIMEOUT)
        return candidate

    def _search_operation(self, page, keyword, is_first_search=True):
        try:
            search_locator = self._pick_search_input(page)
            search_locator.wait_for(timeout=ELEMENT_TIMEOUT)
            search_locator.click(force=True)
            search_locator.fill("")
            page.keyboard.down("Control")
            page.keyboard.press("a")
            page.keyboard.up("Control")
            page.keyboard.press("Backspace")
            type_slow(search_locator, keyword, min_delay=0.06, max_delay=0.18)
            logger.info(f"📝 已输入搜索关键词：{keyword}")
            btn = page.locator(SEARCH_BTN_SELECTOR)
            btn.wait_for(state="visible", timeout=SEARCH_BTN_TIMEOUT)
            page.wait_for_timeout(3000)
            detail_page = page
            if is_first_search:
                try:
                    with page.context.expect_page(timeout=60000) as new_page_info:
                        btn.click()
                    detail_page = new_page_info.value
                    detail_page.wait_for_load_state("networkidle", timeout=20000)
                except PlaywrightTimeoutError:
                    logger.warning(f"{get_current_time()}   未检测到新标签页")
                    return None, False
                except Exception as e:
                    logger.warning(f"{get_current_time()}   等待新标签页异常：{e}")
                    return None, False
            else:
                btn.click()
                page.wait_for_load_state("networkidle", timeout=20000)
                detail_page = page
                logger.info("✅ 后续搜索：已在原页面完成跳转加载")
            test_btn = detail_page.locator("div[data-v-c65c36bc].first-time-highlight-message-btn button")
            if test_btn.count() > 0:
                test_btn.wait_for(state="attached", timeout=5000)
                test_btn.click()
            force_close_popup(detail_page)
            kill_masks(detail_page)
            logger.info("✅ 已触发搜索")
            return detail_page, True
        except PlaywrightTimeoutError as e:
            logger.error(f" 搜索失败：元素定位超时 - {str(e)}")
            return None, False
        except Exception as e:
            logger.error(f" 搜索异常：{str(e)}")
            return None, False

    # ---------- pagination ----------

    @staticmethod
    def _goto_next_page(page) -> bool:
        try:
            next_btn = page.locator("button.btn-next").first
            next_btn.wait_for(state="attached", timeout=3000)
            aria_disabled = next_btn.get_attribute("aria-disabled")
            logger.info(f"下一页按钮 aria-disabled 属性值：{aria_disabled}")
            if aria_disabled == "true":
                logger.warning("⚠️ 下一页按钮 aria-disabled=true，已无更多页面")
                return False
            page.wait_for_timeout(500)
            if next_btn.is_visible() and next_btn.is_enabled():
                next_btn.click(timeout=5000)
            else:
                next_btn.click(force=True, timeout=5000)
            logger.info("✅ 翻页成功，下一页按钮 aria-disabled=false")
            return True
        except PlaywrightTimeoutError:
            logger.warning("⚠️ 下一页按钮加载超时，判定无更多页面")
            return False
        except Exception as e:
            logger.warning(f"⚠️ 翻页操作异常：{e}，判定无更多页面")
            return False

    @staticmethod
    def _get_total_pages(page):
        """从分页组件提取总页数，失败返回 None"""
        try:
            total_elem = page.locator("div.filter-panel-tags-pagination-number-total").first
            total_elem.wait_for(state="attached", timeout=3000)
            if total_elem.count() > 0:
                text = total_elem.inner_text(timeout=2000).strip()
                total = int(text)
                logger.info(f"📊 检测到总页数：{total}")
                return total
        except PlaywrightTimeoutError:
            logger.warning("⚠️ 总页数元素加载超时，降级为盲翻")
        except (ValueError, TypeError) as e:
            logger.warning(f"⚠️ 总页数解析失败：{e}，降级为盲翻")
        except Exception as e:
            logger.warning(f"⚠️ 获取总页数异常：{e}，降级为盲翻")
        return None

    # ---------- shop license collection ----------

    @staticmethod
    def _collect_shop_license(target_page, shop):
        """采集店铺营业执照信息（OCR识别），返回 dict 或 None"""
        random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
        entershop_btn = target_page.locator('div[data-v-5485589c].shop-info-container-left-info')
        entershop_btn.wait_for(state="visible", timeout=10000)
        entershop_btn.scroll_into_view_if_needed()
        entershop_btn.hover()
        random_delay(0.2, 0.5)
        try:
            with target_page.expect_popup(timeout=15000) as pop:
                entershop_btn.click()
                random_delay(0.05, 0.15)
            shop_page = pop.value
        except PlaywrightTimeoutError:
            logger.warning("⚠️ 店铺弹窗未出现")
            return None
        shop_page.wait_for_load_state("domcontentloaded")
        store_url = shop_page.url
        logger.info(f"📌 获取到店铺链接：{store_url}")

        random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
        shop_license_page = shop_page.locator(
            '//div[contains(@class, "shop-info-container-right-btns-item") and contains(span, "资质/售后")]')
        shop_license_page.wait_for(state="attached", timeout=15000)
        shop_license_page.scroll_into_view_if_needed()
        shop_license_page.hover()
        random_delay(0.2, 0.5)
        shop_license_page.click()
        random_delay(0.05, 0.15)
        random_delay(0.05, 0.1)
        shop_page.wait_for_load_state("networkidle")
        shop_page.wait_for_load_state("load")

        shop_license_img = shop_page.locator(
            '//span[contains(text(), "企业营业执照") or contains(text(), "营业执照（正本）")]/ancestor::div[@class="shop-info-drawer-zz-tab1-list-item"]/img'
        ).first
        shop_license_img.wait_for(state="visible", timeout=60000)
        shop_license_src = None
        ocr_res = None
        try:
            if shop_license_img.count() > 0:
                shop_license_src = shop_license_img.get_attribute('src')
                shop_license_src = shop_license_src.strip() if shop_license_src else None
                ocr_res = get_ocr_res(shop_license_src)
        except Exception as e:
            logger.warning(f"提取营业执照图片src失败：{e}")
        logger.debug("营业执照图片链接：%s", shop_license_src)

        qualification_number = ocr_res.get('社会信用代码', '') if ocr_res else ''
        business_license_company = ocr_res.get('单位名称', '') if ocr_res else ''
        business_license_address = ocr_res.get('地址', '') if ocr_res else ''

        province, city = extract_province_city(business_license_address)
        logger.info(f"原始地址：{business_license_address}")
        logger.info(f"提取的省份：{province} | 城市：{city}")

        current_time = datetime.now().strftime("%Y-%m-%d")
        insert_shop_info_to_db(
            shop=shop, contact_address=store_url,
            qualification_number=qualification_number,
            business_license_company=business_license_company,
            business_license_address=business_license_address,
            scrape_date=current_time, platform='药帮忙',
            province=province, city=city,
            create_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            update_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        )

        return {
            "shop_page": shop_page,
            "store_url": store_url,
            "province": province,
            "city": city,
            "business_license_company": business_license_company,
            "qualification_number": qualification_number,
            "business_license_address": business_license_address,
        }

    # ---------- core collection ----------

    def _collect_data(self, store_page):
        """核心采集循环：遍历翻页 → 逐商品采集 → 走 DrugPipeline 入库"""
        brand = self.product_brand
        name = self.product_name
        keyword = self.keyword
        collect_result = []

        logger.info(f"📊 开始采集「{keyword}」的商品数据")
        store_page.wait_for_load_state("networkidle")

        page_no = 1
        start_page = self.api_current_page + 1 if self.api_current_page > 0 else 1

        # 任务续爬：翻到起始页
        if start_page > 1:
            logger.info(f"⏩ 任务续爬：从第 {start_page} 页开始，正在翻页...")
            while page_no < start_page:
                if self.scheduler.end:
                    logger.warning("⛔ 翻页过程中收到停止信号，终止采集")
                    return collect_result, page_no
                if self._goto_next_page(store_page):
                    page_no += 1
                    store_page.wait_for_load_state("networkidle")
                    logger.info(f"⏩ 已翻到第 {page_no} 页")
                else:
                    logger.warning(f"⚠️ 无法翻到第 {start_page} 页（当前仅 {page_no} 页），从当前页开始采集")
                    break

        total_pages = 0

        while True:
            if self.scheduler.end:
                logger.warning("⛔ 收到停止信号，终止采集")
                break

            logger.info(f"\n📄 「{keyword}」开始采集第 {page_no} 页")
            list_page_url = store_page.url
            logger.info(f"📌 已记录商品列表页URL：{list_page_url}")

            store_page.wait_for_load_state("domcontentloaded")
            store_page.wait_for_load_state("networkidle")
            store_page.wait_for_timeout(500)

            # 首次提取总页数
            if total_pages == 0:
                extracted_total = self._get_total_pages(store_page)
                if extracted_total is not None:
                    total_pages = extracted_total
                    # 读到总页数立刻回报进度
                    prog_status = self._report_progress(page_no, total_pages=total_pages)
                    if prog_status == "limit_reached":
                        logger.warning("⛔ 进度上报检测到限额，通知停止")
                        self.scheduler.set_flag(True)

            total_limit = store_page.locator(PRODUCT_ITEM_SELECTOR).count()
            logger.info(f"📌 「{keyword}」第{page_no}页 初始商品个数（count）：{total_limit}")
            collected_count = 0

            not_found_keywords = store_page.locator("div.filter-panel-container-empty-text")
            if not_found_keywords.count() > 0:
                logger.warning(f"⚠️ 关键词「{keyword}」无匹配商品，直接跳过整个关键词采集")
                return [], 0

            for idx in range(total_limit):
                if self.scheduler.end:
                    logger.warning("⛔ 收到停止信号，终止当前页采集")
                    break

                detail_page = None
                try:
                    item = store_page.locator(PRODUCT_ITEM_SELECTOR).nth(idx)
                    collected_count += 1
                    store_page.wait_for_load_state("networkidle")
                    delay = random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
                    logger.info(f"📌 「{keyword}」第{page_no}页 第{collected_count}/{total_limit}个商品 - 等待{delay:.2f}秒后采集（反爬）")

                    # ---- 初始化默认值 ----
                    title = ""
                    shop = ""
                    expiry_date = "无有效期"
                    manufacture_date = "无生产日期"
                    approval_number = "无批准文号"
                    manufacturer = "未知公司"
                    spec = "未知规格"
                    current_time = datetime.now().strftime("%Y-%m-%d")
                    is_sold_out = 0
                    business_license_address = ''

                    # 售罄
                    sold_locator = item.locator('div.product-status')
                    if sold_locator.count() > 0:
                        is_sold_out = 1
                        logger.warning(f" 「{keyword}」第{page_no}页 第{collected_count}个商品已售罄")

                    # 商品ID (platform_item_id)
                    product_id = ''
                    product_id_elem = item.locator('div.product-card[data-product-id]')
                    if product_id_elem.count() > 0:
                        product_id = product_id_elem.get_attribute("data-product-id")

                    # 标题
                    product_locator = item.locator(PRODUCT_TITLE_SELECTOR)
                    if product_locator.count() > 0:
                        title = product_locator.inner_text(timeout=3000).strip()
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页标题：{title}{'='*10}")
                    else:
                        logger.warning(f" 「{keyword}」第{collected_count}个商品 - 列表页标题元素未找到")

                    # 品牌/名称筛选
                    if not (brand in title and name in title):
                        logger.warning(f" 「{keyword}」第{collected_count}个商品 - 标题「{title}」不包含品牌「{brand}」、名称「{name}」，跳过")
                        continue

                    # 价格
                    price_int = item.locator('//span[@class="price-int"]').text_content().strip()
                    price_decimal_elem = item.locator('//span[@class="price-decimal"]')
                    price_decimal = price_decimal_elem.text_content().strip() if price_decimal_elem.count() > 0 else ''
                    full_price = f"{price_int}{price_decimal}".strip()
                    try:
                        full_price_num = float(full_price)
                    except ValueError:
                        full_price_num = 0.0
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页价格无法解析 raw={full_price!r}，按 0 处理")
                    logger.info(f"✅ 提取到价格：{full_price_num}")

                    # 公司名
                    manufacturer_locator = item.locator(PRODUCT_COMPANY_SELECTOR)
                    if manufacturer_locator.count() > 0:
                        manufacturer = manufacturer_locator.inner_text(timeout=3000).strip()
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页公司名：{manufacturer}{'='*10}")
                    else:
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页公司名称元素未找到")

                    # 店铺名
                    shop_locator = item.locator(PRODUCT_STORE_SELECTOR)
                    if shop_locator.count() > 0:
                        raw_shop = shop_locator.inner_text(timeout=3000).strip()
                        shop = clean_shop_name(raw_shop)
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 列表页店名：{shop}{'='*10}")
                    else:
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 列表页店铺名称元素未找到")

                    # 折扣价
                    discount_price_locator = item.locator('span[data-v-4cb6cc1f].discount-int').first
                    if discount_price_locator.count() > 0:
                        discount_price_val_origin = discount_price_locator.inner_text(timeout=3000).strip()
                        match = re.search(r'\d+\.?\d*', str(discount_price_val_origin))
                        discount_price_val = float(match.group()) if match else 0.00
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页折扣价：{discount_price_val}{'='*10}")
                    else:
                        discount_price_val = full_price_num
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 折扣价元素未找到，使用采购价兜底：{discount_price_val}")

                    # 有效期
                    expiry_date_locator = item.locator(PRODUCT_VALIDITY_SELECTOR)
                    if expiry_date_locator.count() > 0:
                        expiry_date = expiry_date_locator.inner_text(timeout=3000).strip().replace('-', '')
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页有效期：{expiry_date}{'='*10}")
                    else:
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 有效期元素未找到")

                    # ---- 点击进入详情页 ----
                    logger.info(f"📌 「{keyword}」第{page_no}页 第{collected_count}个商品「{title}」- 模拟鼠标移动并点击")
                    item.hover()
                    random_delay(0.2, 0.5)
                    item.dispatch_event("mousedown")
                    random_delay(0.05, 0.15)
                    item.dispatch_event("mouseup")
                    random_delay(0.05, 0.1)

                    try:
                        with store_page.context.expect_page(timeout=60000) as p:
                            item.click(delay=random.uniform(0.1, 0.3))
                        detail_page = p.value
                    except PlaywrightTimeoutError:
                        logger.warning(f" 「{keyword}」第{page_no}页 第{collected_count}个商品「{title}」- 未检测到新标签页，使用当前页采集详情")
                        detail_page = None

                    target_page = detail_page if detail_page else store_page
                    target_page.wait_for_load_state("networkidle", timeout=20000)
                    delay = random_delay(MIN_PAGE_DELAY, MAX_PAGE_DELAY)
                    logger.info(f"📌 「{keyword}」第{page_no}页 第{collected_count}个商品「{title}」- 详情页加载完成，等待{delay:.2f}秒（反爬）")

                    # 详情页链接
                    product_link = target_page.url
                    logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页链接：{product_link}{'='*10}")

                    # 生产日期
                    manufacture_date_locator = target_page.locator(
                        '//div[contains(@class, "spec-info-item") and .//div[contains(@class, "spec-info-item-label") and normalize-space(.)="生产日期"]]//div[contains(@class, "spec-info-item-value-text")]')
                    if manufacture_date_locator.count() > 0:
                        manufacture_date = manufacture_date_locator.inner_text(timeout=3000).strip()
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页生产日期：{manufacture_date}{'='*10}")
                    else:
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 生产日期元素未找到")

                    # 批准文号
                    approval_number_locator = target_page.locator(
                        '//div[contains(@class, "spec-info-item") and .//div[contains(@class, "spec-info-item-label") and normalize-space(.)="批准文号"]]//div[contains(@class, "spec-info-item-value-text")]')
                    if approval_number_locator.count() > 0:
                        approval_number = approval_number_locator.inner_text(timeout=3000).strip()
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页批准文号：{approval_number}{'='*10}")
                    else:
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 批准文号元素未找到")

                    # 规格
                    spec_locator = target_page.locator(
                        '//div[contains(@class, "spec-info-item") and .//div[contains(@class, "spec-info-item-label") and normalize-space(.)="规格"]]//div[contains(@class, "spec-info-item-value-text")]')
                    if spec_locator.count() > 0:
                        spec = spec_locator.inner_text(timeout=3000).strip()
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页规格：{spec}{'='*10}")
                    else:
                        logger.warning(f" 「{keyword}」第{collected_count}个商品「{title}」- 规格元素数量不足")

                    # 库存
                    storage = ''
                    storage_locator = target_page.locator('[data-v-51f0e85d].detail-input-num-right-title')
                    if storage_locator.count() > 0:
                        storage = storage_locator.inner_text(timeout=3000).strip()
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页库存：{storage}{'='*10}")

                    # 销量
                    sell = ''
                    sell_locator = target_page.locator('div.detail-info-content-item-value-price-top-right div[data-v-95163d4a]', has_text='已售')
                    if sell_locator.count() > 0:
                        sell = sell_locator.inner_text(timeout=3000).strip()
                        logger.info(f"{'='*10}「{keyword}」第{collected_count}个商品 - 详情页销量：{sell}{'='*10}")

                    # 快照
                    oss_url = ""
                    try:
                        local_path, oss_url = screenshot_target_page_to_local_then_oss(target_page=target_page, full_page=True)
                        logger.info("详情页快照已上传 local=%s oss=%s", local_path, oss_url)
                    except Exception as e:
                        logger.warning("详情页快照上传失败：%s", e)

                    # ---- 店铺资质采集 ----
                    province = ""
                    city = ""
                    business_license_company = ""
                    qualification_number = ''
                    shop_exists, shop_info = shop_is_exists_database(shop)
                    shop_page = None
                    store_url = ''

                    if not shop_exists:
                        logger.info("数据库没有该店铺的营业执照，开始采集店铺资质")
                        license_info = self._collect_shop_license(target_page, shop)
                        if license_info:
                            shop_page = license_info["shop_page"]
                            store_url = license_info["store_url"]
                            province = license_info["province"]
                            city = license_info["city"]
                            business_license_company = license_info["business_license_company"]
                            qualification_number = license_info["qualification_number"]
                            business_license_address = license_info["business_license_address"]
                    else:
                        logger.info("数据库有该店名，在数据库拿取对应字段填充")
                        if shop_info:
                            province = shop_info.get("province", "") or ""
                            city = shop_info.get("city", "") or ""
                            business_license_company = shop_info.get("business_license_company", "") or ""
                            qualification_number = shop_info.get("qualification_number", "") or ""
                            business_license_address = shop_info.get("business_license_address", "") or ""

                    # 关闭店铺页
                    try:
                        if shop_page and not shop_page.is_closed():
                            random_delay(4, 8)
                            shop_page.close()
                            logger.info(f"📌 「{keyword}」第{collected_count}个商品 - 已关闭店铺页标签 shop_page")
                    except Exception as e:
                        logger.warning(f"⚠️ 关闭 shop_page 失败：{e}")

                    random_delay(5, 8)

                    # 关闭详情页，切回列表页
                    if detail_page and not detail_page.is_closed():
                        detail_page.close()
                        logger.info(f"📌 「{keyword}」第{collected_count}个商品 - 已关闭详情页标签页")
                    store_page.bring_to_front()
                    store_page.mouse.move(random.randint(100, 300), random.randint(200, 400))
                    random_delay(0.5, 1.0)
                    store_page.wait_for_load_state("networkidle")
                    random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
                    logger.info(f" 「{keyword}」第{collected_count}个商品「{title}」- 已切回列表页")
                    random_delay(2, 4)

                    # 省市ID
                    province_id, city_id = get_province_city_ids(province, city)

                    # ---- 组装 product dict（兼容 DrugPipeline） ----
                    now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
                    product = {
                        "platform": self.platform_id,
                        "item_id": product_id,
                        "enterprise_id": self.company_id,
                        "product_name": title,
                        "spec": spec,
                        "one_price": discount_price_val,
                        "detail_url": product_link,
                        "shop_name": shop,
                        "anonymous_store_name": "",
                        "shop_url": store_url,
                        "city_name": city,
                        "city_id": city_id,
                        "province_name": province,
                        "province_id": province_id,
                        "shipment_city_name": "",
                        "shipment_city_id": 0,
                        "shipment_province_name": "",
                        "shipment_province_id": 0,
                        "area_info": business_license_address,
                        "factory_name": manufacturer,
                        "scrape_date": current_time,
                        "price": discount_price_val,
                        "sales": sell,
                        "stock_count": storage,
                        "snapshot_url": oss_url,
                        "approval_num": approval_number,
                        "produced_time": manufacture_date,
                        "deadline": expiry_date,
                        "update_time": now_str,
                        "insert_time": now_str,
                        "number": 1,
                        "product_brand": brand,
                        "collect_task_id": self.collect_task_id,
                        "search_name": keyword,
                        "company_name": business_license_company,
                        "collect_config_info": self.collect_config_info,
                        "account_id": self.collect_equipment_account_id,
                        "collect_region_id": self.collect_region_id,
                        "collect_round": self.collect_round,
                        "is_sold_out": is_sold_out,
                    }

                    # 走 DrugPipeline 入库（含去重 + 店铺城市补全）
                    self.pipeline.storge_data(product)
                    collect_result.append(product)
                    logger.info(f" 「{keyword}」第{collected_count}个商品「{title}」采集完成")

                except Exception as e:
                    logger.exception(f" 「{keyword}」第{collected_count}个商品采集核心异常：{str(e)}")
                    try:
                        if detail_page and not detail_page.is_closed():
                            detail_page.close()
                        if store_page and not store_page.is_closed():
                            store_page.bring_to_front()
                        store_page.wait_for_load_state("networkidle")
                        random_delay(MIN_CLICK_DELAY, MAX_CLICK_DELAY)
                    except Exception as e2:
                        logger.error(f" 「{keyword}」第{collected_count}个商品详情采集异常（处理时）：{str(e2)}，原异常：{str(e)}")
                        continue

                # 每采6个商品滚动一次
                if collected_count % 6 == 0 and collected_count > 0 and collected_count != total_limit:
                    logger.info("采满6个往下滑")
                    slow_scroll_400px(store_page)
                    store_page.wait_for_load_state("networkidle")

            if self.scheduler.end:
                break

            delay = random_delay(1.5, 3.0)
            logger.info(f"⏳ 翻页前随机等待 {delay:.2f}s（反爬）")

            # 每页结束上报进度
            prog_status = self._report_progress(page_no, total_pages=total_pages)
            if prog_status == "limit_reached":
                logger.warning("⛔ 进度上报检测到限额，通知停止")
                self.scheduler.set_flag(True)

            # 翻页
            if self._goto_next_page(store_page):
                logger.info(f"「{keyword}」还有下一页")
                page_no += 1
                store_page.wait_for_load_state("networkidle")
                total_limit = store_page.locator(PRODUCT_ITEM_SELECTOR).count()
                logger.info(f"📌 「{keyword}」第{page_no}页 商品个数更新为：{total_limit}")
                continue
            else:
                logger.info(f" 「{keyword}」已无下一页，关键词采集结束")
                break

        long_delay = random_delay(MIN_KEYWORD_DELAY, MAX_KEYWORD_DELAY)
        logger.info(f" 「{keyword}」采集完成，共{len(collect_result)}条数据（pipeline 实际入库 {self.pipeline.crawl_count} 条），等待{long_delay:.2f}秒后继续下一个关键词（反爬）")
        return collect_result, page_no

    # ---------- main entry ----------

    def run(self):
        """蜘蛛主入口，返回 (crawl_count, is_success)。

        由 CollectScheduleRunner 调用：
            count, is_success = YbmCrawler(task_dict).run()
        """
        load_city_mapping()
        logger.info("\n" + "=" * 50)
        logger.info("🚀 药帮忙采集程序启动")
        logger.info(f"⏰ 启动时间：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
        logger.info("=" * 50)

        with sync_playwright() as p:
            browser = p.chromium.launch(
                channel="chrome",
                headless=True,
                slow_mo=random.randint(100, 300),
                args=[
                    "--disable-blink-features=AutomationControlled",
                    "--enable-automation=false",
                    "--disable-infobars",
                    "--remote-debugging-port=0",
                    "--start-maximized",
                    "--disable-extensions",
                    "--disable-plugins-discovery",
                    "--no-sandbox",
                    "--disable-dev-shm-usage",
                    f"--user-agent=Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{random.randint(110, 120)}.0.0.0 Safari/537.36",
                ],
            )
            context = browser.new_context(
                locale="zh-CN",
                timezone_id="Asia/Shanghai",
                geolocation={"latitude": 31.230416, "longitude": 121.473701},
                permissions=["geolocation"],
                user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
                viewport={"width": 1800, "height": 1000},
                java_script_enabled=True,
                bypass_csp=True,
            )
            page = context.new_page()

            page.add_init_script("""
                Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
                Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3] });
                Object.defineProperty(navigator, 'mimeTypes', { get: () => [1, 2, 3] });
                window.chrome = { runtime: {}, loadTimes: () => ({}) };
                delete window.navigator.languages;
                window.navigator.languages = ['zh-CN', 'zh'];
                (() => {
                    const originalAddEventListener = EventTarget.prototype.addEventListener;
                    EventTarget.prototype.addEventListener = function(type, listener) {
                        if (type === 'mousemove') {
                            return originalAddEventListener.call(this, type, (e) => {
                                e._automation = undefined;
                                listener(e);
                            });
                        }
                        return originalAddEventListener.call(this, type, listener);
                    };
                })();
            """)

            try:
                # 登录
                load_cookies(context)
                if not self._is_login(page):
                    page.goto(TARGET_LOGIN_URL)
                    page.wait_for_load_state("networkidle")
                    logger.info("🔑 开始执行登录流程")
                    login_success = self._login_operation(page)
                    if not login_success:
                        logger.error(" 登录失败，程序终止")
                        self.is_success = False
                        self._page_no = 0
                        return 0, False
                    save_cookies(context)
                    logger.info(" 登录并保存Cookie成功！")

                # 启动心跳守护线程（CrawlerScheduler）
                self.scheduler.start()

                # 搜索
                popup_guard(page, "before_search")
                store_page, search_success = self._search_operation(page, self.keyword, is_first_search=True)
                if store_page:
                    popup_guard(store_page, "after_search")

                if store_page is None:
                    logger.warning(f" 「{self.keyword}」搜索页面为空，任务失败")
                    self.is_success = False
                    self._page_no = 0
                    return 0, False

                if not search_success:
                    logger.warning(f" 「{self.keyword}」搜索失败，跳过采集")
                    self.is_success = False
                    self._page_no = 0
                    return 0, False

                store_page.wait_for_load_state("domcontentloaded")
                store_page.wait_for_load_state('networkidle')

                # 上报采集开始，检查是否限额
                report_status = self._report_start()
                if report_status == "limit_reached":
                    logger.warning("⛔ 任务 %s 开始上报时账号已达限额，退出当前任务", self.task_id)
                    self.is_success = False
                    self._page_no = 0
                    return 0, False

                # 核心采集
                data_list, page_no = self._collect_data(store_page)
                self._page_no = page_no
                real_count = len(data_list)
                self.is_success = True
                logger.info(f"关键词「{self.keyword}」采集完成，共 {real_count} 条数据（pipeline 实际入库 {self.pipeline.crawl_count} 条），共 {page_no} 页")

            except Exception as e:
                logger.exception("程序异常：%s", e)
                self.is_success = False
            finally:
                self.scheduler.stop()
                # 上报采集结束
                try:
                    self._report_end(
                        success=self.is_success,
                        real_count=self.pipeline.crawl_count,
                        page_no=getattr(self, '_page_no', 0),
                        total_pages=0,
                    )
                except Exception:
                    pass
                browser.close()
                logger.info(" 浏览器已关闭，程序结束")

        return self.pipeline.crawl_count, self.is_success


# ==================== SECTION 9: ENTRY POINT (测试/直接运行) ====================

if __name__ == '__main__':
    # 直接运行：使用手动任务进行测试
    load_city_mapping()
    if USE_MANUAL_TASK:
        task = MANUAL_TASK
    else:
        # 从内部 API 拉取任务
        try:
            resp = requests.get(
                TASK_PULL_URL,
                params={'platform': PLATFORM_ID, 'username': YBM_DEVICE_ID},
                headers={'X-Crawler-Token': TASK_API_TOKEN},
                timeout=15,
            )
            result = resp.json()
            task = result.get('data', {}).get('task') or {}
        except Exception as e:
            logger.error(f"拉取任务失败：{e}")
            task = {}

    if not task:
        logger.error("未获取到任何任务，程序退出")
    else:
        crawler = YbmCrawler(task)
        count, success = crawler.run()
        logger.info(f"测试完成：crawl_count={count}, is_success={success}")

        # 发送飞书通知
        try:
            spec_items = [item.strip() for item in re.split(r"[|、，,\n\r]+", task.get("product_specs", "") or "") if item.strip()]
            display_spec = "、".join(spec_items)
            notice_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
            if success:
                feishu_send_text(
                    f"{notice_time} 通知:\n"
                    f"平台: 药帮忙, 药品: {task.get('product_name', '')}, 品规: {display_spec}, 爬取数据: {count}条"
                )
            else:
                feishu_send_error_card(
                    task_name=task.get("product_name", ""),
                    err_msg=f"task_id={task.get('id')}, company_id={task.get('company_id')}, crawl_count={count}",
                    mention_all=False,
                )
        except Exception as e:
            logger.warning(f"飞书通知发送失败：{e}")