update

spiders/yaojiujiu/login_yaojiujiu.py

@@ -0,0 +1,196 @@
+"""药九九浏览器自动化登录 — Playwright 填表登录，成功后 cookie 写入 accounts_platform"""
+import json
+import time
+import random
+from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError
+
+from commons.Logger import get_spider_logger
+from commons.conn_mysql import MySQLPoolOn2
+
+logger = get_spider_logger("yaojiujiu_login")
+
+LOGIN_URL = "https://www.yyjzt.com/login"
+PLATFORM_ID = 7
+
+
+class YaoJiuJiuLogin:
+    """Playwright 浏览器自动化登录药九九"""
+
+    def __init__(self):
+        self.db = MySQLPoolOn2()
+
+    # ==================== 账号获取 ====================
+    def get_account(self):
+        sql = """SELECT `id`, `phone`, `password` FROM `accounts_platform`
+                 WHERE `platform`=%s AND `status`=1
+                 ORDER BY `cookie_timestamp` ASC LIMIT 1"""
+        rows = self.db.select_data(sql, (PLATFORM_ID,))
+        return rows[0] if rows else None
+
+    # ==================== 浏览器登录 ====================
+    def login(self, phone, password, headless=True):
+        """
+        打开浏览器 → 填表 → 点击登录 → 提取 cookie → 关闭浏览器
+        成功返回 yjj-token cookie 值，失败返回 None
+        """
+        with sync_playwright() as p:
+            browser = p.chromium.launch(
+                headless=headless,
+                channel="chrome",
+                args=[
+                    "--disable-blink-features=AutomationControlled",
+                    "--no-sandbox",
+                ],
+            )
+            context = browser.new_context(
+                locale="zh-CN",
+                timezone_id="Asia/Shanghai",
+                viewport={"width": 1600, "height": 900},
+            )
+            page = context.new_page()
+
+            try:
+                # 1. 打开登录页
+                page.goto(LOGIN_URL, timeout=30000)
+                page.wait_for_load_state("networkidle")
+
+                # 2. 确保在"账户密码登录"tab（id="tab-first"）
+                pwd_tab = page.locator("#tab-first")
+                if pwd_tab.count() > 0:
+                    pwd_tab.click()
+                    page.wait_for_timeout(500)
+
+                # 3. 填账号
+                username_input = page.locator("input[name='username']")
+                username_input.wait_for(state="visible", timeout=10000)
+                username_input.click()
+                username_input.fill("")
+                # 模拟逐字输入
+                for ch in str(phone):
+                    username_input.type(ch, delay=random.randint(60, 180))
+                    page.wait_for_timeout(random.randint(30, 80))
+                logger.info("已输入账号")
+
+                # 4. 填密码
+                password_input = page.locator("input[name='password']")
+                password_input.click()
+                password_input.fill("")
+                for ch in str(password):
+                    password_input.type(ch, delay=random.randint(60, 180))
+                    page.wait_for_timeout(random.randint(30, 80))
+                logger.info("已输入密码")
+
+                # 5. 点登录
+                login_btn = page.locator("button:has-text('登录')").first
+                if login_btn.locator("..").first:
+                    # 等待按钮可用（Element UI 的验证需要输入完整后才解除 disabled）
+                    page.wait_for_timeout(500)
+                login_btn.click()
+                logger.info("已点击登录")
+
+                # 6. 等待登录结果 — 页面跳转或报错
+                page.wait_for_timeout(3000)
+                try:
+                    page.wait_for_load_state("networkidle", timeout=15000)
+                except Exception:
+                    pass
+
+                # 7. 检查登录状态
+                if "login" in page.url.lower():
+                    # 检查是否有错误提示
+                    err = page.locator(".el-message--error, .el-form-item__error, [class*='error']").first
+                    if err.count() > 0:
+                        logger.error("登录失败: %s", err.inner_text(timeout=3000))
+                    else:
+                        logger.error("登录失败，仍在登录页")
+                    return None
+
+                # 8. 提取 cookie
+                cookies = context.cookies()
+                browser.close()
+
+                # 找 yjj-token
+                token = ""
+                for c in cookies:
+                    if c["name"] == "yjj-token":
+                        token = c["value"].replace("ssr_", "")
+                        break
+
+                if not token:
+                    logger.error("登录后未找到 yjj-token cookie")
+                    return None
+
+                logger.info("登录成功, token=%s...", token[:20])
+                return {"token": token, "cookies": cookies}
+
+            except PlaywrightTimeoutError as e:
+                logger.error("登录超时: %s", e)
+                return None
+            except Exception as e:
+                logger.exception("登录异常: %s", e)
+                return None
+            finally:
+                try:
+                    browser.close()
+                except Exception:
+                    pass
+
+    # ==================== Cookie 入库 ====================
+    def save_cookies(self, phone, cookies):
+        """保存全套 cookie 到 accounts_platform，供快照模式注入"""
+        # 提取关键 cookie
+        cookie_map = {}
+        for c in cookies:
+            cookie_map[c["name"]] = c["value"]
+        # Token 单独存，兼容 API 模式
+        token = cookie_map.get("yjj-token", "").replace("ssr_", "")
+        cookie_map["Token"] = token
+        cookie_str = json.dumps(cookie_map, ensure_ascii=False)
+        now_ts = int(time.time())
+        sql = """UPDATE `accounts_platform`
+                 SET `cookie_str`=%s, `cookie_timestamp`=%s
+                 WHERE `phone`=%s AND `platform`=%s"""
+        affected = self.db.execute(sql, (cookie_str, now_ts, str(phone), PLATFORM_ID))
+        if affected:
+            logger.info("Cookie 已写入 accounts_platform (含 %s 个 key), phone=%s", len(cookie_map), phone)
+        else:
+            logger.warning("Cookie 写入失败, phone=%s", phone)
+
+    def save_token(self, phone, token):
+        """兼容旧调用"""
+        self.save_cookies(phone, [{"name": "yjj-token", "value": "ssr_" + str(token)}])
+
+    # ==================== 入口 ====================
+    def run(self, headless=True):
+        """执行登录：读数据库账号 → 浏览器登录 → 入库，返回 token"""
+        acc = self.get_account()
+        if not acc:
+            logger.error("无可用药九九账号")
+            return None
+
+        phone = acc["phone"]
+        password = acc["password"]
+
+        # 先看缓存
+        cached = acc.get("cookie_str") or ""
+        if cached:
+            try:
+                cd = json.loads(cached)
+                t = cd.get("Token") or cd.get("token")
+                if t:
+                    logger.info("使用缓存 token")
+                    return t
+            except json.JSONDecodeError:
+                pass
+
+        # 浏览器登录
+        result = self.login(phone, password, headless=headless)
+        if result:
+            self.save_cookies(phone, result["cookies"])
+            return result["token"]
+        return None
+
+
+if __name__ == "__main__":
+    t = YaoJiuJiuLogin().run(headless=False)
+    print("token:", t)

spiders/yaojiujiu/yyjzt_crawl.py

@@ -0,0 +1,452 @@
+"""药九九搜索爬虫 — 通过 API 接口采集商品价格，缺店铺信息时自动补齐"""
+import hashlib
+import json
+import random
+import time
+
+import requests
+
+from commons.Logger import get_spider_logger
+from commons.conn_mysql import MySQLPoolOn2
+from pipelines.drug_pipelines import DrugPipeline
+from pipelines.shop_pipelines import ShopPipeline
+from spiders.yaojiujiu.login_yaojiujiu import YaoJiuJiuLogin
+
+logger = get_spider_logger("yaojiujiu")
+
+# ---------- 常量 ----------
+PLATFORM_ID = 7
+SEARCH_API = "https://web-api.yyjzt.com/search/api/search/item/list"
+STORE_API = "https://web-api.yyjzt.com/sale/api/Sale/storeinfo/findById"
+PRODUCT_LINK_TPL = (
+    "https://www.yyjzt.com/goodDetail?ladderNum"
+    "&itemStoreId={item_store_id}"
+    "&sourceProdetail=%2Fsearch"
+    "&is_store=0"
+)
+PAGE_SIZE = 60
+MAX_PAGES = 30
+
+
+def compute_one_price(item):
+    """
+    完全复刻页面 gc-l2-price 展示逻辑:
+      有活动(特价/团购) → sortPrice（活动价）
+      无活动(普通)     → itemPrice（原价）
+    """
+    discount = item.get("marketItemStoreDiscountPriceToTrade") or {}
+    if discount.get("specialPrice") is not None:
+        return float(discount["specialPrice"])
+    if discount.get("joinGroupPrice") is not None:
+        return float(discount["joinGroupPrice"])
+    return float(item.get("itemPrice") or 0)
+
+
+def generate_sign(token, body_dict, ts):
+    """_s 签名: MD5(token + JSON.stringify(body) + timestamp)"""
+    body_json = json.dumps(body_dict, separators=(",", ":"), ensure_ascii=False)
+    raw = token + body_json + str(ts)
+    return hashlib.md5(raw.encode("utf-8")).hexdigest()
+
+
+class YyjztSpider:
+    def __init__(self, task_dict=None):
+        self.task_dict = task_dict or {}
+        self.pipeline = DrugPipeline("yaojiujiu")
+        self.shop_pipeline = ShopPipeline("yaojiujiu")
+        self.db = MySQLPoolOn2()
+        self.page = 0
+        self.is_success = True
+        self.collect_task_id = None
+        self._seen_stores = set()  # 已处理的 storeId，避免重复查库/调接口
+
+        # ---- 从 task 提取参数 ----
+        self.task_id = ""
+        self.company_id = ""
+        self.product_name = ""
+        self.product_specs = ""
+        self.product_brand = ""
+        self.keyword = ""
+        self.sampling_cycle = ""
+        self.sampling_start_time = ""
+        self.sampling_end_time = ""
+        self.account_id = ""
+        self.collect_region_id = ""
+        self.collect_round = "1"
+
+        if self.task_dict:
+            self._parse_task()
+
+    # ==================== 任务解析 ====================
+    def _parse_task(self):
+        self.task_id = self.task_dict.get("id", "")
+        self.company_id = self.task_dict.get("company_id", "")
+        self.product_name = self.task_dict.get("product_name", "")
+        self.product_specs = self.task_dict.get("product_specs", "")
+        self.product_brand = self.task_dict.get("product_brand", "")
+        self.keyword = (
+            self.task_dict.get("product_keyword", "")
+            or (self.task_dict.get("product_brand", "") + " " + self.product_name + " " + self.product_specs).strip()
+        ).strip()
+        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.account_id = self.task_dict.get("collect_equipment_account_id", "")
+        self.collect_region_id = self.task_dict.get("collect_region_id", "")
+        self.collect_round = self.task_dict.get("collect_round", "1")
+        self.collect_task_id = self.task_dict.get("collect_task_id", "")
+
+        # 登录账号
+        self._account_phone = self.task_dict.get("account_phone", "")
+        self._account_password = self.task_dict.get("account_password", "")
+
+        # deviceId 用账号 phone 或固定值
+        self._device_id = self._account_phone or "yyjzt_device"
+
+        # token 从 accounts_platform 缓存读取，过期后自动重登
+        self._token = None
+        self._ensure_token()
+
+    # ==================== 进度回告 ====================
+    def _report_exception(self, exception_type, remark=""):
+        scheduler = self.task_dict.get("scheduler")
+        if not scheduler:
+            return
+        data = {
+            "task_id": self.task_id,
+            "platform": str(PLATFORM_ID),
+            "username": self.task_dict.get("_device_id", ""),
+            "current_page": self.page,
+            "is_finished": 0,
+            "exception_type": exception_type,
+            "remark": remark,
+        }
+        if self.pipeline.crawl_count:
+            data["crawled_count"] = self.pipeline.crawl_count
+        scheduler.post_report(data)
+
+    def _report_page_progress(self, page, crawled_count, is_finished=0, total_pages=0):
+        scheduler = self.task_dict.get("scheduler")
+        if not scheduler:
+            return False
+        data = {
+            "task_id": self.task_id,
+            "platform": str(PLATFORM_ID),
+            "username": self.task_dict.get("_device_id", ""),
+            "current_page": page,
+            "crawled_count": crawled_count,
+            "is_finished": is_finished,
+        }
+        if total_pages:
+            data["total_pages"] = total_pages
+        scheduler.post_report(data)
+        return scheduler.end
+
+    # ==================== 登录 & Token 管理 ====================
+    # 参考 main.py is_login() — 主动检测，不等到 API 报错
+    def _is_token_valid(self):
+        """用轻量 API 主动检测 token 是否有效（类似 main.py 检测'欢迎登录'文本）"""
+        if not self._token:
+            return False
+        body = {"pageIndex": 1, "pageSize": 1, "platformType": 1,
+                "keyword": "999感冒灵", "searchNum": 0, "showSearchTips": 0, "isNeedAggregation": 0}
+        try:
+            headers = self._signed_headers(body)
+            resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=10)
+            data = resp.json()
+            return data.get("success") and data.get("code") == 200
+        except Exception:
+            return False
+
+    def _get_token_from_db(self):
+        """直接从 DB 读缓存 token"""
+        sql = """SELECT `cookie_str` FROM `accounts_platform`
+                 WHERE `platform`=%s AND `status`=1 AND `phone`=%s
+                 ORDER BY `cookie_timestamp` DESC LIMIT 1"""
+        rows = self.db.select_data(sql, (PLATFORM_ID, self._account_phone))
+        if not rows:
+            return None
+        try:
+            cd = json.loads(rows[0].get("cookie_str") or "{}")
+            return cd.get("Token") or cd.get("token")
+        except json.JSONDecodeError:
+            return None
+
+    def _ensure_token(self):
+        """先读 DB 缓存 → 主动验证是否过期 → 过期则浏览器重新登录"""
+        # 1. 读缓存（类似 main.py load_cookies）
+        token = self._get_token_from_db()
+        if token:
+            self._token = token
+            # 2. 主动检测是否有效（类似 main.py is_login）
+            if self._is_token_valid():
+                logger.info("缓存 token 有效: %s...", token[:16])
+                return
+            else:
+                logger.info("缓存 token 已过期")
+
+        # 3. 过期或不存在 → 浏览器登录（类似 main.py login_operation + save_cookies）
+        logger.info("启动浏览器登录...")
+        login = YaoJiuJiuLogin()
+        token = login.run(headless=True)
+        if token:
+            self._token = token
+        else:
+            logger.error("登录失败")
+            self._token = ""
+
+    def _re_login(self):
+        """Token 失效，强制重新登录"""
+        logger.info("Token 失效，重新登录...")
+        login = YaoJiuJiuLogin()
+        token = login.run(headless=True)
+        if token:
+            self._token = token
+            return True
+        return False
+
+    # ==================== API 请求头 ====================
+    def _base_headers(self):
+        return {
+            "Origin": "https://www.yyjzt.com",
+            "Referer": "https://www.yyjzt.com/",
+            "sy_plat": "B2B",
+            "token_platform_client_type": "USER",
+            "zhcaiToken": self._token,
+            "deviceId": self._device_id,
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
+        }
+
+    def _signed_headers(self, body):
+        ts = int(time.time() * 1000)
+        sign = generate_sign(self._token, body, ts)
+        headers = self._base_headers()
+        headers["Content-Type"] = "application/json"
+        headers["_s"] = sign
+        headers["_t"] = str(ts)
+        return headers
+
+    # ==================== 搜索 API ====================
+    def _search_page(self, page_index, retry=True):
+        if not self._token:
+            raise Exception("无登录 token")
+        body = {
+            "pageIndex": page_index,
+            "pageSize": PAGE_SIZE,
+            "platformType": 1,
+            "keyword": self.keyword,
+            "searchNum": 0,
+            "showSearchTips": 0,
+            "isNeedAggregation": 0,
+        }
+        headers = self._signed_headers(body)
+        resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=30)
+        resp.raise_for_status()
+        data = resp.json()
+
+        # 登录失效则重登后重试一次
+        if not (data.get("success") and data.get("code") == 200):
+            msg = data.get("msg", "")
+            if retry and ("登录" in msg or "token" in msg.lower() or "未登录" in msg):
+                logger.warning("Token 失效(%s)，尝试重新登录", msg)
+                if self._re_login():
+                    return self._search_page(page_index, retry=False)
+            raise Exception(f"API error: {msg}")
+
+        return data["data"]["itemList"]
+
+    # ==================== 店铺信息补齐 ====================
+    def _ensure_shop_info(self, store_id, store_name):
+        """确保店铺信息已入库。先查 retrieve_scrape_shop_info，没有则调 API 抓取。"""
+        if not store_id or store_id in self._seen_stores:
+            return
+        self._seen_stores.add(store_id)
+
+        # 查库
+        exist = self.db.select_data(
+            "SELECT 1 FROM retrieve_scrape_shop_info WHERE platform=%s AND shop=%s LIMIT 1",
+            (str(PLATFORM_ID), store_name),
+        )
+        if exist:
+            return
+
+        # 调店铺信息接口
+        logger.info("店铺表无 [%s]，抓取店铺信息 storeId=%s", store_name, store_id)
+        try:
+            resp = requests.get(
+                STORE_API,
+                params={"id": store_id},
+                headers=self._base_headers(),
+                timeout=15,
+            )
+            resp.raise_for_status()
+            data = resp.json()
+            if not (data.get("success") and data.get("code") == 200):
+                logger.warning("店铺接口异常 storeId=%s: %s", store_id, data.get("msg"))
+                return
+            info = data["data"]
+        except Exception as e:
+            logger.error("店铺接口请求失败 storeId=%s: %s", store_id, e)
+            return
+
+        # 映射到 retrieve_scrape_shop_info 字段
+        today = time.strftime("%Y-%m-%d")
+        shop_row = {
+            "id": PLATFORM_ID,
+            "platform": str(PLATFORM_ID),
+            "shop": (info.get("storeShortName") or info.get("storeTitle") or store_name),
+            "province": info.get("provinceName") or "",
+            "city": info.get("cityName") or "",
+            "contact_address": info.get("storeAddress") or "",
+            "qualification_number": info.get("organizationCode") or "",
+            "business_license_company": info.get("storeName") or "",
+            "business_license_address": info.get("storeAddress") or "",
+            "scrape_date": today,
+            "ai_generate": 0,
+        }
+        try:
+            self.shop_pipeline.storge_data(shop_row)
+        except Exception as e:
+            logger.exception("店铺入库失败 storeId=%s: %s", store_id, e)
+        # 店铺接口休息 1-2 秒
+        time.sleep(random.uniform(1, 2))
+
+    # ==================== 数据映射 ====================
+    def _to_product(self, item):
+        now = time.strftime("%Y-%m-%d %H:%M:%S")
+        today = time.strftime("%Y-%m-%d")
+
+        store_name = item.get("storeName") or ""
+        store_full_name = item.get("storeFullName") or ""
+        item_store_id = str(item.get("itemStoreId") or "")
+
+        # 省份/城市/area_info 由 DrugPipeline.get_shop_city() 从 retrieve_scrape_shop_info 表补齐
+        return {
+            "platform": PLATFORM_ID,
+            "item_id": item_store_id,
+            "enterprise_id": self.company_id,
+            "product_name": item.get("itemStoreName") or "",
+            "spec": item.get("specs") or "",
+            "one_price": compute_one_price(item),
+            "detail_url": PRODUCT_LINK_TPL.format(item_store_id=item_store_id) if item_store_id else "",
+            "shop_name": store_name,
+            "anonymous_store_name": store_name,
+            "shop_url": "",
+            "city_name": "",
+            "city_id": "",
+            "province_name": "",
+            "province_id": "",
+            "area_info": "",
+            "factory_name": item.get("manufacturer") or item.get("manufactureName") or "",
+            "scrape_date": today,
+            "price": float(item.get("sortPrice") or 0),
+            "sales": item.get("totalCustCount") or "",
+            "stock_count": item.get("itemStorage") or item.get("ownStorageNumber") or 0,
+            "snapshot_url": "",  # API 模式无快照，快照模式后续用 Playwright 截图
+            "approval_num": item.get("approvalNo") or "",
+            "produced_time": "",
+            "deadline": str(item.get("validateTimeEnd") or ""),
+            "update_time": now,
+            "insert_time": now,
+            "number": int(item.get("middlePackageAmount") or 1),
+            "product_brand": item.get("brandName") or self.product_brand,
+            "collect_task_id": self.collect_task_id,
+            "search_name": self.product_name,
+            "company_name": store_full_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,
+            }, ensure_ascii=False),
+            "account_id": self.account_id,
+            "collect_region_id": self.collect_region_id,
+            "collect_round": self.collect_round,
+            "is_sold_out": 1 if item.get("isSoldOut") else 0,
+        }
+
+    # ==================== 主流程 ====================
+    def _match_item(self, product):
+        """检查商品是否匹配品牌+品名，返回 True=匹配"""
+        product_brand = product.get("product_brand", "")
+        product_name = product.get("product_name", "")
+
+        brand_ok = True
+        name_ok = True
+        if self.product_brand:
+            brand_ok = self.product_brand in str(product_brand)
+        if self.product_name:
+            name_ok = self.product_name in str(product_name)
+        return brand_ok and name_ok
+
+    def search_data(self):
+        if not self.task_dict:
+            return
+
+        no_match_count = 0       # 连续不匹配计数
+        MAX_NO_MATCH = 20        # 连续超过此数则停止（参考 main.py）
+
+        for page in range(1, MAX_PAGES + 1):
+            logger.info("药九九 爬取第 %s 页, 关键词=%s", page, self.keyword)
+
+            try:
+                items = self._search_page(page)
+            except Exception as e:
+                logger.error("第 %s 页请求失败: %s", page, e)
+                self.is_success = False
+                self._report_exception(5, f"第{page}页请求失败: {e}")
+                return
+
+            if not items:
+                logger.info("第 %s 页无数据，停止翻页", page)
+                break
+
+            # 先补齐所有店铺信息
+            for item in items:
+                store_id = item.get("storeId")
+                store_name = item.get("storeName") or ""
+                self._ensure_shop_info(store_id, store_name)
+
+            # 逐条入库
+            for item in items:
+                product = self._to_product(item)
+                if not product.get("item_id"):
+                    continue
+
+                # 品牌/品名过滤
+                if not self._match_item(product):
+                    no_match_count += 1
+                    logger.warning("不匹配(%s/%s): brand=%s name=%s",
+                                   no_match_count, MAX_NO_MATCH,
+                                   product.get("product_brand"), product.get("product_name"))
+                    if no_match_count >= MAX_NO_MATCH:
+                        logger.error("连续 %s 条不匹配，停止翻页", MAX_NO_MATCH)
+                        return
+                    continue
+                no_match_count = 0  # 匹配成功则重置
+
+                try:
+                    self.pipeline.storge_data(product)
+                except Exception as e:
+                    logger.exception("入库失败: %s", e)
+
+            self.page = page
+
+            if self._report_page_progress(page, self.pipeline.crawl_count):
+                break
+
+            if len(items) < PAGE_SIZE:
+                break
+
+            # 翻页休息 2-3 秒，模拟真人
+            time.sleep(random.uniform(2, 3))
+
+    def run(self):
+        try:
+            self.search_data()
+        except Exception as e:
+            self.is_success = False
+            logger.error("药九九爬虫异常: %s", e, exc_info=True)
+            self._report_exception(5, f"未知异常: {e}")
+
+        logger.info("药九九 爬取完成: %s 条, %s 页", self.pipeline.crawl_count, self.page)
+        return self.pipeline.crawl_count, self.is_success, self.page

spiders/yaojiujiu/yyjzt_snapshot.py

@@ -0,0 +1,647 @@
+"""药九九快照模式 — 一个浏览器从头到尾：检查登录 → 搜索 → 监听API → 匹配DOM → 截卡片"""
+import json
+import random
+import time
+
+import requests
+from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError
+
+from commons.Logger import get_spider_logger
+from commons.conn_mysql import MySQLPoolOn2
+from pipelines.drug_pipelines import DrugPipeline
+from oss_upload.oss_upload import AliyunOSSUploader
+from spiders.yaojiujiu.yyjzt_crawl import (
+    PLATFORM_ID, PAGE_SIZE, MAX_PAGES, SEARCH_API,
+    PRODUCT_LINK_TPL, compute_one_price, generate_sign,
+)
+
+logger = get_spider_logger("yaojiujiu_snapshot")
+
+import urllib.parse
+SEARCH_PAGE_URL = "https://www.yyjzt.com/search?keyword={keyword}"
+LOGIN_URL = "https://www.yyjzt.com/login"
+HOME_URL = "https://www.yyjzt.com"
+
+# 参考 main.py 的反爬/弹窗处理
+USERNAME_SELECTOR = "input[name='username']"
+PASSWORD_SELECTOR = "input[name='password']"
+LOGIN_BTN_SELECTOR = "button:has-text('登录')"
+
+
+class YyjztSnapshotSpider:
+    def __init__(self, task_dict=None):
+        self.task_dict = task_dict or {}
+        self.pipeline = DrugPipeline("yaojiujiu_snapshot")
+        self.db = MySQLPoolOn2()
+        self.is_success = True
+
+        # 任务参数
+        self.task_id = ""
+        self.company_id = ""
+        self.product_name = ""
+        self.product_specs = ""
+        self.product_brand = ""
+        self.keyword = ""
+        self.collect_task_id = ""
+        self.collect_round = "1"
+        self.account_id = ""
+        self.collect_region_id = ""
+        self.sampling_cycle = ""
+        self.sampling_start_time = ""
+        self.sampling_end_time = ""
+
+        # 账号
+        self._phone = ""
+        self._password = ""
+
+        if task_dict:
+            self._parse_task()
+
+    def _parse_task(self):
+        self.task_id = self.task_dict.get("id", "")
+        self.company_id = self.task_dict.get("company_id", "")
+        self.product_name = self.task_dict.get("product_name", "")
+        self.product_specs = self.task_dict.get("product_specs", "")
+        self.product_brand = self.task_dict.get("product_brand", "")
+        self.keyword = (
+            self.task_dict.get("product_keyword", "")
+            or (self.task_dict.get("product_brand", "") + " " + self.product_name + " " + self.product_specs).strip()
+        ).strip()
+        self.collect_task_id = self.task_dict.get("collect_task_id", "")
+        self.collect_round = self.task_dict.get("collect_round", "1")
+        self.account_id = self.task_dict.get("collect_equipment_account_id", "")
+        self.collect_region_id = self.task_dict.get("collect_region_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._phone = self.task_dict.get("account_phone", "")
+        self._password = self.task_dict.get("account_password", "")
+
+    # ============ 浏览器启动 + 登录 ============
+    def _get_account(self):
+        sql = """SELECT `phone`, `password`, `cookie_str` FROM `accounts_platform`
+                 WHERE `platform`=%s AND `status`=1
+                 ORDER BY `cookie_timestamp` ASC LIMIT 1"""
+        rows = self.db.select_data(sql, (PLATFORM_ID,))
+        return rows[0] if rows else None
+
+    def _is_login(self, page):
+        """参考 main.py is_login()"""
+        try:
+            page.goto(HOME_URL, timeout=15000)
+            page.wait_for_load_state("networkidle")
+            time.sleep(2)
+            if "login" in page.url.lower():
+                return False
+            if page.locator(":has-text('欢迎登录')").count() > 0:
+                return False
+            return True
+        except Exception:
+            return False
+
+    def _login_operation(self, page, phone, password):
+        """参考 main.py login_operation()"""
+        try:
+            page.goto(LOGIN_URL, timeout=15000)
+            page.wait_for_load_state("networkidle")
+            time.sleep(2)
+
+            # 切到密码登录 tab
+            pwd_tab = page.locator("#tab-first")
+            if pwd_tab.count() > 0:
+                pwd_tab.click()
+                page.wait_for_timeout(500)
+
+            # 填手机号
+            page.fill(USERNAME_SELECTOR, "")
+            for ch in str(phone):
+                page.locator(USERNAME_SELECTOR).type(ch, delay=random.randint(60, 180))
+            logger.info("已输入账号")
+
+            # 填密码
+            page.fill(PASSWORD_SELECTOR, "")
+            for ch in str(password):
+                page.locator(PASSWORD_SELECTOR).type(ch, delay=random.randint(60, 180))
+            logger.info("已输入密码")
+
+            # 点登录
+            page.click(LOGIN_BTN_SELECTOR)
+            logger.info("已点击登录")
+            time.sleep(3)
+
+            try:
+                page.wait_for_load_state("networkidle", timeout=15000)
+            except Exception:
+                pass
+
+            # 检查是否登录成功
+            if "login" in page.url.lower():
+                err = page.locator(".el-message--error, [class*='error']").first
+                if err.count() > 0:
+                    logger.error("登录失败: %s", err.inner_text(timeout=3000))
+                return False
+            return True
+        except PlaywrightTimeoutError as e:
+            logger.error("登录超时: %s", e)
+            return False
+
+    def _kill_popups(self, page):
+        """参考 main.py: 先 JS 暴力清遮罩，再点按钮，三层防线不崩"""
+        # 第一层：JS 先暴力移除遮罩 + 滑块验证 + 恢复滚动（main.py kill_masks）
+        page.evaluate("""
+        () => {
+          // 精准删已知遮罩 + 滑块验证
+          const selectors = [
+            '.v-modal', '.el-overlay', '.el-overlay-dialog',
+            '.el-dialog__wrapper', '.el-message-box__wrapper', '.el-loading-mask',
+            '.safety-slider-box', '[class*="safety"]', '[class*="slider"]',
+            '[class*="captcha"]', '.el-popup-parent--hidden'
+          ];
+          selectors.forEach(s => {
+            document.querySelectorAll(s).forEach(e => {
+              try { e.remove(); } catch(_) {}
+            });
+          });
+
+          // 泛化兜底：近似全屏 + 高 z-index 的层强制隐藏
+          const all = Array.from(document.querySelectorAll('body *'));
+          for (const el of all) {
+            try {
+              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.7 && r.height >= innerHeight * 0.7;
+                if (nearFull) {
+                  el.style.pointerEvents = 'none';
+                  el.style.display = 'none';
+                }
+              }
+            } catch(_) {}
+          }
+
+          // 恢复滚动
+          document.documentElement.style.overflow = 'auto';
+          document.body.style.overflow = 'auto';
+          document.body.style.position = 'static';
+          document.body.style.width = 'auto';
+          document.body.classList.remove('el-popup-parent--hidden');
+        }
+        """)
+        page.wait_for_timeout(300)
+
+        # 第二层：连续点引导/关闭按钮（兼容多步引导），try 包住不崩
+        for _ in range(6):
+            try:
+                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(300)
+                    continue
+            except Exception:
+                pass
+            try:
+                close_icon = page.locator(
+                    ".el-dialog__headerbtn, .el-message-box__headerbtn, "
+                    ".close, .icon-close, .el-icon-close"
+                ).first
+                if close_icon.count() > 0 and close_icon.is_visible():
+                    close_icon.click(timeout=1200, force=True)
+                    page.wait_for_timeout(300)
+                    continue
+            except Exception:
+                pass
+            break
+
+    # ============ API 搜索 ============
+    def _api_search_all(self):
+        """API 搜全量，返回 [(item_dict, page), ...]"""
+        all_items = []
+        for p in range(1, MAX_PAGES + 1):
+            body = {"pageIndex": p, "pageSize": PAGE_SIZE, "platformType": 1,
+                    "keyword": self.keyword, "searchNum": 0,
+                    "showSearchTips": 0, "isNeedAggregation": 0}
+            ts = int(time.time() * 1000)
+            # 用 browser 当前 cookie 里的 token
+            headers = {
+                "Content-Type": "application/json", "Origin": "https://www.yyjzt.com",
+                "sy_plat": "B2B", "token_platform_client_type": "USER",
+                "zhcaiToken": self._token, "deviceId": self._phone,
+                "_s": generate_sign(self._token, body, ts), "_t": str(ts),
+            }
+            try:
+                resp = requests.post(SEARCH_API, json=body, headers=headers, timeout=30)
+                data = resp.json()
+                if data.get("success") and data.get("code") == 200:
+                    items = data["data"]["itemList"]
+                    for item in items:
+                        all_items.append((item, p))
+                    total_count = int(data["data"].get("totalCount") or 0)
+                    if total_count > 0 and len(all_items) >= total_count:
+                        break
+                    if len(items) == 0:
+                        break
+                else:
+                    logger.error("API page %s: %s", p, data.get("msg"))
+                    break
+            except Exception as e:
+                logger.error("API page %s: %s", p, e)
+                break
+            time.sleep(random.uniform(1, 2))
+        logger.info("API 共 %s 条", len(all_items))
+        return all_items
+
+    # ============ DOM 匹配 ============
+    def _extract_dom_cards(self, page):
+        """逐张卡片滚到可视区，触发懒加载后再取 innerText"""
+        cards = []
+        card_els = page.locator(".goods_card")
+        cnt = card_els.count()
+        if cnt == 0:
+            logger.warning("页面无 .goods_card 元素")
+            return cards
+
+        # 先滚回顶部
+        page.evaluate("window.scrollTo(0, 0)")
+        page.wait_for_timeout(300)
+
+        # 预估每张卡高度，用于判断是否需要继续滚
+        first_card_height = 0
+        try:
+            box = card_els.nth(0).bounding_box()
+            if box:
+                first_card_height = box["height"]
+        except Exception:
+            pass
+        # 每屏大约能显示多少张卡（假设卡高约 260px，视口 900px → ~3排 = 随实际布局调整）
+        cards_per_screen = max(2, int(900 / max(first_card_height, 200))) if first_card_height else 3
+
+        for i in range(cnt):
+            try:
+                card = card_els.nth(i)
+                # 滚到这张卡中间位置，确保它进入可视区
+                card.scroll_into_view_if_needed()
+                page.wait_for_timeout(300)  # 给懒加载时间
+
+                all_text = card.inner_text(timeout=2000).strip()
+                if all_text:
+                    cards.append({"all_text": all_text, "locator": card, "index": i})
+                else:
+                    # 偶尔 inner_text 第一次拿不到（渲染慢），重试一次
+                    page.wait_for_timeout(500)
+                    all_text = card.inner_text(timeout=2000).strip()
+                    if all_text:
+                        cards.append({"all_text": all_text, "locator": card, "index": i})
+                    else:
+                        logger.warning("card[%s] 文本为空，可能的结构异常", i)
+            except Exception as e:
+                logger.warning("card[%s] 提取失败: %s", i, e)
+
+            # 每屏结束时额外等一下，防止滚太快
+            if (i + 1) % cards_per_screen == 0:
+                page.wait_for_timeout(400)
+
+        logger.info("逐卡提取: %s/%s 个卡片有文本", len(cards), cnt)
+        return cards
+
+    @staticmethod
+    def _price_variants(value):
+        """同一个价格的不同字符串形态：15.0 → {"15.00","15.0","15"}"""
+        vs = set()
+        vs.add(f"{value:.2f}")
+        vs.add(str(value))
+        vs.add(f"{value:.2f}".rstrip("0").rstrip("."))
+        vs.add(f"{value:.1f}".rstrip("0").rstrip("."))
+        return vs
+
+    def _match(self, api_item, dom_cards, used):
+        api_name = api_item.get("itemStoreName", "")
+        api_store = api_item.get("storeName", "")
+        api_specs = api_item.get("specs", "")
+        api_price = compute_one_price(api_item)
+        # sortPrice 是页面实际展示价（折后/拼团后），compute_one_price 是计算价，
+        # DOM 可能显示其中任意一个，两者都纳入匹配池
+        api_sort_price = float(api_item.get("sortPrice") or 0)
+
+        # 价格变体：compute_one_price + sortPrice 的所有字符串形态
+        price_pool = self._price_variants(api_price)
+        if api_sort_price and abs(api_sort_price - api_price) > 0.001:
+            price_pool |= self._price_variants(api_sort_price)
+
+        fail_reasons = []
+        for c in dom_cards:
+            if c["index"] in used:
+                continue
+            txt = c["all_text"]
+            name_ok = bool(api_name) and (api_name in txt)
+            store_ok = bool(api_store) and (api_store in txt)
+            price_ok = any(p in txt for p in price_pool)
+            # specs 辅助消歧：同名+同店+同价时，用规格区分
+            specs_ok = (not api_specs) or (api_specs in txt)
+            if name_ok and store_ok and price_ok and specs_ok:
+                return c, ""
+            fail_reasons.append(
+                f"  card[{c['index']}]: name={name_ok} store={store_ok} price={price_ok} specs={specs_ok} "
+                f"card_text={txt[:80]}"
+            )
+        # 没匹配上，返回原因
+        detail = (
+            f"API: name={api_name!r} store={api_store!r} price={api_price}"
+            f" sortPrice={api_sort_price} specs={api_specs!r} pool={price_pool}\n"
+            + "\n".join(fail_reasons[:5])
+        )
+        return None, detail
+
+    # ============ 翻页 ============
+    def _goto_next_page(self, page):
+        """点击下一页按钮（参考 main.py goto_next_page），成功返回 True"""
+        candidates = [
+            ".el-pagination button.btn-next:not(.is-disabled)",
+            ".el-pagination__next:not(.is-disabled)",
+            "button:has-text('下一页'):not([disabled])",
+            "a:has-text('下一页')",
+        ]
+        for sel in candidates:
+            btn = page.locator(sel).first
+            if btn.count() > 0:
+                try:
+                    page.evaluate("window.scrollTo(0, 0);")
+                    btn.click(timeout=5000)
+                    page.wait_for_load_state("networkidle")
+                    return True
+                except Exception:
+                    continue
+        return False
+
+    # ============ 截图+入库 ============
+    def _process_page(self, page, items, page_no):
+        self._kill_popups(page)
+        if page_no == 1:
+            # 第一页：URL 打开搜索
+            url = SEARCH_PAGE_URL.format(keyword=urllib.parse.quote(self.keyword))
+            page.goto(url, timeout=20000)
+        else:
+            # 后续页：点击"下一页"按钮（URL 翻页不生效，Vue 前端必须点按钮）
+            if not self._goto_next_page(page):
+                logger.warning("第 %s 页翻页失败，停止", page_no)
+                return False
+        page.wait_for_load_state("networkidle")
+        time.sleep(random.uniform(1, 2))
+        self._kill_popups(page)  # 页面加载完再清一次，防止延迟弹窗
+
+        # 截个整页图方便排查
+        page.screenshot(path=f"./snapshots/full_page_{page_no}.png", full_page=True)
+
+        dom_cards = self._extract_dom_cards(page)
+        logger.info("页%s: DOM=%s API=%s", page_no, len(dom_cards), len(items))
+
+        import os
+        os.makedirs("./snapshots", exist_ok=True)
+        os.makedirs("./snapshots/unmatched", exist_ok=True)
+        os.makedirs("./snapshots/debug", exist_ok=True)
+
+        # 先把所有 DOM 卡片截小图存 debug，方便排查
+        for c in dom_cards[:5]:  # 只截前5个看结构
+            try:
+                jpg = c["locator"].screenshot(type="jpeg", quality=60)
+                fname = f"debug_card_{page_no}_{c['index']}.jpg"
+                with open(os.path.join("./snapshots/debug", fname), "wb") as f:
+                    f.write(jpg)
+            except Exception:
+                pass
+
+        used = set()
+        matched = 0
+        unmatched_log = []
+        for item, _ in items:
+            card, fail_detail = self._match(item, dom_cards, used)
+            if not card:
+                unmatched_log.append(fail_detail)
+                continue
+            used.add(card["index"])
+            try:
+                card["locator"].scroll_into_view_if_needed()
+            except Exception:
+                pass
+            time.sleep(0.3)
+
+            # 截图
+            try:
+                jpg = card["locator"].screenshot(type="jpeg", quality=80)
+            except Exception as e:
+                logger.warning("截图失败 item=%s: %s", item.get("itemStoreId"), e)
+                continue
+
+            # 上传 OSS
+            fname = f"yjj_{item.get('itemStoreId','?')}_{page_no}_{card['index']}"
+            try:
+                snapshot_url = self.uploader.upload_from_bytes(jpg, fname)
+            except Exception as e:
+                logger.warning("OSS上传失败: %s, 存本地", e)
+                local_path = os.path.join("./snapshots", fname + ".jpg")
+                with open(local_path, "wb") as f:
+                    f.write(jpg)
+                snapshot_url = local_path
+
+            # 入库
+            product = self._to_product(item, snapshot_url)
+            try:
+                self.pipeline.storge_data(product)
+                matched += 1
+            except Exception as e:
+                logger.exception("入库失败: %s", e)
+
+            # 同时写文件
+            self._all_results.append(product)
+
+        # 未匹配的截全卡片图存 unmatched，同时写日志
+        skipped = len(items) - matched
+        if unmatched_log:
+            log_path = os.path.join("./snapshots", f"unmatched_page{page_no}.log")
+            with open(log_path, "w", encoding="utf-8") as lf:
+                lf.write(f"=== 页{page_no} 未匹配 {skipped} 条 ===\n\n")
+                lf.write("\n---\n".join(unmatched_log))
+            logger.info("未匹配日志: %s", log_path)
+            # 截几个未匹配的 DOM 卡片
+            for i, c in enumerate(dom_cards):
+                if c["index"] not in used and i < 10:
+                    try:
+                        jpg = c["locator"].screenshot(type="jpeg", quality=70)
+                        fname = f"unmatched_{page_no}_{c['index']}.jpg"
+                        with open(os.path.join("./snapshots/unmatched", fname), "wb") as f:
+                            f.write(jpg)
+                    except Exception:
+                        pass
+
+        if skipped:
+            logger.info("页%s: 匹配入库 %s 条, 未匹配跳过 %s 条", page_no, matched, skipped)
+
+    def _to_product(self, item, snapshot_url=""):
+        now = time.strftime("%Y-%m-%d %H:%M:%S")
+        today = time.strftime("%Y-%m-%d")
+        sid = str(item.get("itemStoreId") or "")
+        # min_price = gc-l2-price = 页面实际显示价
+        min_price = compute_one_price(item)
+        return {
+            "platform": PLATFORM_ID, "item_id": sid,
+            "enterprise_id": self.company_id,
+            "product_name": item.get("itemStoreName") or "",
+            "spec": item.get("specs") or "",
+            "one_price": 0,   # 不存此字段，填0
+            "detail_url": PRODUCT_LINK_TPL.format(item_store_id=sid) if sid else "",
+            "shop_name": item.get("storeName") or "",
+            "anonymous_store_name": "",   # 不存此字段
+            "shop_url": "", "city_name": "", "city_id": "",
+            "province_name": "", "province_id": "", "area_info": "",
+            "factory_name": item.get("manufacturer") or item.get("manufactureName") or "",
+            "scrape_date": today,
+            "price": min_price,   # → DB min_price = gc-l2-price
+            "sales": item.get("totalCustCount") or "",
+            "stock_count": item.get("itemStorage") or item.get("ownStorageNumber") or 0,
+            "snapshot_url": snapshot_url,
+            "approval_num": item.get("approvalNo") or "",
+            "produced_time": "",
+            "deadline": str(item.get("validateTimeEnd") or ""),
+            "update_time": now, "insert_time": now,
+            "number": 1,   # 固定为1
+            "product_brand": item.get("brandName") or self.product_brand,
+            "collect_task_id": self.collect_task_id,
+            "search_name": self.product_name,
+            "company_name": item.get("storeFullName") or "",
+            "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),
+            "account_id": self.account_id,
+            "collect_region_id": self.collect_region_id,
+            "collect_round": self.collect_round,
+            "is_sold_out": 1 if item.get("isSoldOut") else 0,
+        }
+
+    # ============ 主流程 ============
+    def run(self):
+        self.uploader = AliyunOSSUploader()
+        self._all_results = []
+
+        # 1. 拿账号
+        acc = self._get_account()
+        if not acc:
+            logger.error("无可用账号")
+            return 0, False
+        phone = acc["phone"]
+        password = acc["password"]
+
+        # 2. 一个浏览器干到底（headless + Linux 兼容）
+        with sync_playwright() as p:
+            browser = p.chromium.launch(
+                headless=True,
+                args=[
+                    "--disable-blink-features=AutomationControlled",
+                    "--no-sandbox",
+                    "--disable-setuid-sandbox",
+                    "--disable-dev-shm-usage",
+                    "--disable-gpu",
+                ],
+            )
+            context = browser.new_context(
+                locale="zh-CN", timezone_id="Asia/Shanghai",
+                viewport={"width": 1600, "height": 900},
+            )
+            page = context.new_page()
+
+            # 3. 加载 DB 缓存的 cookie（类似 main.py load_cookies）
+            db_cookie_str = acc.get("cookie_str") or ""
+            if db_cookie_str:
+                try:
+                    db_cookies = json.loads(db_cookie_str)
+                    cookie_list = [
+                        {"name": k, "value": str(v), "domain": ".yyjzt.com", "path": "/"}
+                        for k, v in db_cookies.items() if v and k not in ("Token",)
+                    ]
+                    if cookie_list:
+                        page.goto(HOME_URL, timeout=10000)  # 先访问域名
+                        context.add_cookies(cookie_list)
+                        logger.info("已加载 DB 缓存 %s 个 cookie", len(cookie_list))
+                except (json.JSONDecodeError, Exception) as e:
+                    logger.warning("加载缓存 cookie 失败: %s", e)
+
+            # 4. 检查登录态 → 没登录当场登（同一个浏览器）
+            if self._is_login(page):
+                logger.info("Cookie 有效，已登录")
+            else:
+                logger.info("未登录，开始登录...")
+                if not self._login_operation(page, phone, password):
+                    logger.error("登录失败")
+                    browser.close()
+                    return 0, False
+                logger.info("登录成功")
+                # 登录后清弹窗
+                self._kill_popups(page)
+                # 保存 cookie 到 DB
+                cookies = context.cookies()
+                cookie_map = {c["name"]: c["value"] for c in cookies}
+                token = cookie_map.get("yjj-token", "").replace("ssr_", "")
+                cookie_map["Token"] = token
+                now_ts = int(time.time())
+                self.db.execute(
+                    "UPDATE accounts_platform SET cookie_str=%s, cookie_timestamp=%s WHERE phone=%s AND platform=%s",
+                    (json.dumps(cookie_map, ensure_ascii=False), now_ts, str(phone), PLATFORM_ID),
+                )
+                logger.info("已保存 %s 个 cookie 到 DB", len(cookie_map))
+
+            # 4. 从 browser cookie 提取 token 给 API 用
+            cookies = context.cookies()
+            for c in cookies:
+                if c["name"] == "yjj-token":
+                    self._token = c["value"].replace("ssr_", "")
+                    break
+
+            # 5. API 搜全量数据
+            all_items = self._api_search_all()
+            if not all_items:
+                browser.close()
+                return 0, False
+
+            # 6. 按页码分组，逐页截图
+            from collections import defaultdict
+            pages = defaultdict(list)
+            for item, page_no in all_items:
+                pages[page_no].append((item, page_no))
+
+            for page_no in sorted(pages.keys()):
+                if self._process_page(page, pages[page_no], page_no) is False:
+                    break
+                time.sleep(random.uniform(2, 3))
+
+            # 写结果文件
+            out_path = f"./snapshots/result_{int(time.time())}.json"
+            with open(out_path, "w", encoding="utf-8") as f:
+                json.dump(self._all_results, f, ensure_ascii=False, indent=2, default=str)
+            logger.info("结果文件: %s (%s 条)", out_path, len(self._all_results))
+
+            browser.close()
+
+        logger.info("快照完成: %s 条", self.pipeline.crawl_count)
+        return self.pipeline.crawl_count, self.is_success, len(pages)
+
+
+if __name__ == "__main__":
+    task = {
+        "id": 999, "company_id": "1",
+        "product_name": "蒲地蓝消炎片", "product_specs": "",
+        "product_brand": "999", "product_keyword": "999 蒲地蓝消炎片",
+        "current_page": 0, "start_page": 1, "end_page": 0,
+        "scheduler": None, "_device_id": "test", "_platform": "7",
+        "collect_task_id": "999", "snapshot_collect_status": 0,
+        "collect_round": 1,
+        "account_phone": "", "account_password": "", "account_name": "",
+        "sampling_cycle": "", "sampling_start_time": "", "sampling_end_time": "",
+        "collect_equipment_account_id": "", "collect_region_id": "",
+    }
+    s = YyjztSnapshotSpider(task)
+    print(s.run())

start_run_yaojiujiu.py

@@ -0,0 +1,158 @@
+"""药九九统一入口 — 根据任务 snapshot_collect_status 字段自动切换普通/快照模式。
+
+对接 CrawlerScheduler API（拉取任务/心跳/进度上报），
+snapshot_collect_status=0 → 快照模式（Playwright截图），非0 → 普通API模式。
+"""
+import random
+import time
+import json
+
+from spiders.yaojiujiu.yyjzt_crawl import YyjztSpider
+from spiders.yaojiujiu.yyjzt_snapshot import YyjztSnapshotSpider
+from commons.scheduler import CrawlerScheduler
+from commons.conn_mysql import MySQLPoolOn2
+from commons.Logger import logger
+
+PLATFORM_NAME = "药九九"
+PLATFORM_ID = 7
+
+
+def get_device_config():
+    """从 accounts_platform 获取药九九账号（platform=7, status=1）。"""
+    db = MySQLPoolOn2()
+    sql = """SELECT `name`, `phone`, `password` FROM `accounts_platform`
+             WHERE `platform`=7 AND `status`=1
+             ORDER BY `cookie_timestamp` ASC LIMIT 1"""
+    rows = db.select_data(sql)
+    if not rows:
+        logger.error("无可用药九九账号（platform=7, status=1）")
+        return None
+    account = rows[0]
+    logger.info("获取到药九九账号: name=%s, phone=%s", account["name"], account["phone"])
+    return account["phone"], account["password"], account["name"]
+
+
+def _build_task_dict(api_task, device_id, account_password, account_name, scheduler):
+    """组装 spider 需要的 task_dict。"""
+    current_page = api_task.get("current_page", 0)
+
+    task_dict = {
+        "id": api_task.get("id"),
+        "company_id": api_task.get("company_id"),
+        "product_name": api_task.get("product_name", ""),
+        "product_specs": api_task.get("product_specs", ""),
+        "product_brand": api_task.get("product_brand", ""),
+        "product_keyword": api_task.get("product_keyword", ""),
+        "current_page": current_page,
+        "start_page": current_page if current_page > 0 else 1,
+        "end_page": 0,
+        "scheduler": scheduler,
+        "_device_id": device_id,
+        "_platform": str(PLATFORM_ID),
+        "collect_task_id": str(api_task.get("id", "")),
+        "snapshot_collect_status": api_task.get("snapshot_collect_status"),
+        "sampling_cycle": "",
+        "sampling_start_time": "",
+        "sampling_end_time": "",
+        "collect_equipment_account_id": "",
+        "collect_region_id": "",
+        "collect_equipment_id": "",
+        "collect_round": 1,
+        # 登录账号信息（spider 内检查登录状态用）
+        "account_phone": device_id,
+        "account_password": account_password,
+        "account_name": account_name,
+    }
+    return task_dict
+
+
+def _report_progress(scheduler, task_id, is_finished, crawled_count=0,
+                     current_page=0, total_pages=0,
+                     exception_type=None, remark=None):
+    data = {
+        "task_id": task_id,
+        "platform": str(PLATFORM_ID),
+        "username": scheduler.username,
+        "current_page": current_page,
+        "is_finished": is_finished,
+        "total_pages": total_pages,
+    }
+    if crawled_count:
+        data["crawled_count"] = crawled_count
+    if exception_type is not None:
+        data["exception_type"] = exception_type
+    if remark:
+        data["remark"] = remark
+    try:
+        scheduler.post_report(data)
+        logger.info("任务上报成功: is_finished=%s, crawled_count=%s", is_finished, crawled_count)
+    except Exception as e:
+        logger.warning("任务上报失败: %s", e)
+
+
+def run_one_task(scheduler, device_id, account_password, account_name):
+    """拉取一个任务，根据 snapshot_collect_status 选 spider 执行并上报。"""
+    api_task = scheduler.get_task()
+    if not api_task:
+        logger.info("%s 暂无任务", PLATFORM_NAME)
+        return
+
+    task_id = api_task.get("id", "")
+    task_dict = _build_task_dict(api_task, device_id, account_password,
+                                 account_name, scheduler)
+
+    log_dict = {k: v for k, v in task_dict.items() if k != "scheduler"}
+    logger.info("拉取到任务: %s", json.dumps(log_dict, ensure_ascii=False))
+
+    # 根据 snapshot_collect_status 决定走普通爬虫还是快照爬虫
+    snapshot_status = task_dict.get("snapshot_collect_status", 0)
+    if snapshot_status == 0:
+        spider_cls = YyjztSnapshotSpider
+        logger.info("任务 %s 启用快照模式", task_id)
+    else:
+        spider_cls = YyjztSpider
+        logger.info("任务 %s 使用普通模式", task_id)
+
+    result = spider_cls(task_dict).run()
+    if len(result) == 3:
+        crawl_count, is_success, total_pages = result
+    else:
+        crawl_count, is_success = result
+        total_pages = 0
+
+    if is_success:
+        _report_progress(scheduler, task_id, is_finished=1,
+                         crawled_count=crawl_count,
+                         current_page=total_pages, total_pages=total_pages)
+    else:
+        _report_progress(scheduler, task_id, is_finished=0,
+                         crawled_count=crawl_count,
+                         current_page=total_pages, total_pages=total_pages)
+
+
+if __name__ == "__main__":
+    # 1. 获取账号
+    result = get_device_config()
+    if result:
+        device_id, account_password, account_name = result
+    else:
+        device_id, account_password, account_name = "YYJZT", None, None
+
+    # 2. 启动调度器
+    scheduler = CrawlerScheduler(DEVICE_ID=device_id, platform=str(PLATFORM_ID))
+    scheduler.start()
+    time.sleep(2)
+    logger.info("药九九调度器已启动, device_id=%s", device_id)
+
+    # 3. 循环执行
+    idle_seconds = random.randint(180, 300)
+    logger.info("药九九循环任务已启动, 间隔 %s 秒", idle_seconds)
+
+    while True:
+        try:
+            logger.info("开始执行药九九爬虫")
+            run_one_task(scheduler, device_id, account_password, account_name)
+            logger.info("药九九爬虫执行完成")
+        except Exception as e:
+            logger.error("药九九爬虫执行失败: %s", e, exc_info=True)
+        time.sleep(idle_seconds)

start_run_yaoshibang_v2.py

@@ -57,7 +57,7 @@ def _build_task_dict(api_task, device_id, account_password, account_name, schedu
         "_device_id": device_id,
         "_platform": str(PLATFORM_ID),
         "collect_task_id": str(api_task.get("id", "")),
-        "snapshot_collect_status": api_task.get("snapshot_collect_status", 1),
+        "snapshot_collect_status": api_task.get("snapshot_collect_status"),
         "sampling_cycle": "",
         "sampling_start_time": "",
         "sampling_end_time": "",