更新 'spiders/yaoex/yaoex_snapshot_crawl.py'

spiders/yaoex/yaoex_snapshot_crawl.py

@@ -11,13 +11,23 @@ from urllib.parse import quote
 import requests
 from Crypto.Cipher import AES
 from DrissionPage import ChromiumPage, ChromiumOptions
-from commons.Logger import logger
+from commons.Logger import get_spider_logger
+from commons.scheduler import CrawlerScheduler
+from commons.config import YYC_ACCOUNT, YYC_DEVICE_ID
 from oss_upload.oss_upload import AliyunOSSUploader
 from pipelines.drug_pipelines import DrugPipeline
 from area_info.city_name_to_id import get_city
-from commons.config import YYC_ACCOUNT
 from Crypto.Util.Padding import unpad
 
+logger = get_spider_logger("yaoex_snapshot")
+
+# 平台常量（API 调度用）
+PLATFORM_ID = 6
+HEARTBEAT_INTERVAL_SECONDS = 60
+TASK_API_BASE = "http://192.168.2.246:8080/api/collect_task"
+TASK_API_TOKEN = "zhijiayun_crawler_2026"
+TASK_REPORT_URL = TASK_API_BASE + "/report"
+
 CAPTCHA_TOKEN = "zPzmt1mG1ouCU6GTzsZN2Lmm8pdZypapPcLJTBRETco"
 CAPTCHA_API_URL = "http://api.jfbym.com/api/YmServer/customApi"
 
@@ -25,7 +35,6 @@ chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe"
 # 项目根目录 → spiders/yaoex（与从哪执行脚本无关）
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 YAOEX_SPIDER_DIR = PROJECT_ROOT / "spiders" / "yaoex"
-BROWSER_PROFILE_SUBDIR = "chrome_profile"
 SLIDER_OFFSET_FIX = 10
 DETAIL_GET_TIMEOUT = 15
 DETAIL_URL_WAIT = 10
@@ -77,6 +86,7 @@ class YaoexSnapshotCrawl:
         self.is_not_product = 0
         self.user_id = YYC_ACCOUNT["user_id"]
         self.token = YYC_ACCOUNT["token"]
+        self.scheduler = CrawlerScheduler(YYC_DEVICE_ID, str(PLATFORM_ID), heartbeat_interval=HEARTBEAT_INTERVAL_SECONDS)
 
     def get_product_data(self):
         self.task_id = self.task_dict["id"]
@@ -94,10 +104,75 @@ class YaoexSnapshotCrawl:
         self.collect_region_id = self.task_dict.get("collect_region_id", "")
         self.collect_round = self.task_dict.get("collect_round", 1)
         self.start_page = self._parse_page(self.task_dict.get("start_page"), 1)
-        self.end_page = max(
-            self.start_page,
-            self._parse_page(self.task_dict.get("end_page"), self.start_page),
-        )
+        self.end_page = self._parse_page(self.task_dict.get("end_page"), 100)
+
+    # ---------- API report helpers ----------
+
+    def _send_report(self, params: dict, log_msg: str):
+        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", data.get('code'), msg)
+                            return "limit_reached"
+                        else:
+                            logger.warning("⚠️ 回告接口返回错误：code=%s, msg=%s", data.get('code'), msg)
+                            return "error"
+                except ValueError:
+                    logger.warning("⚠️ 回告响应非 JSON：%s", resp.text[:200])
+                    return "error"
+            else:
+                logger.warning("⚠️ 回告 HTTP %s", resp.status_code)
+                return "error"
+        except Exception as e:
+            logger.error("❌ 回告失败：%s，错误：%s", log_msg, str(e))
+            return "error"
+
+    def _report_start(self):
+        payload = {
+            "task_id": self.task_id,
+            "platform": str(self.platform),
+            "username": YYC_DEVICE_ID,
+            "current_page": 0,
+            "total_pages": 0,
+            "is_finished": 0,
+        }
+        logger.info("上报开始 task_id=%s", self.task_id)
+        return self._send_report(payload, "开始上报（is_finished=0）")
+
+    def _report_progress(self, current_page: int, total_pages: int = 0):
+        payload = {
+            "task_id": self.task_id,
+            "platform": str(self.platform),
+            "username": YYC_DEVICE_ID,
+            "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, total_pages: int = 0):
+        payload = {
+            "task_id": self.task_id,
+            "platform": str(self.platform),
+            "username": YYC_DEVICE_ID,
+            "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,
+        }
+        logger.info("上报结束 task_id=%s payload=%s", self.task_id, payload)
+        self._send_report(payload, "结束上报（is_finished=1）")
 
     @staticmethod
     def _parse_page(value, default=1):
@@ -136,22 +211,20 @@ class YaoexSnapshotCrawl:
             return s.getsockname()[1]
 
     def _resolve_browser_profile_dir(self):
-        """
-        浏览器数据固定落在 <项目根>/spiders/yaoex/ 下。
-        优先 chrome_profile/<账号>；若旧版直接在 yaoex/<账号> 已有登录态则继续沿用。
-        """
-        preferred = YAOEX_SPIDER_DIR / BROWSER_PROFILE_SUBDIR / self.account_name
-        legacy_flat = YAOEX_SPIDER_DIR / self.account_name
+        """浏览器数据目录: <项目根>/spiders/yaoex/<账号>"""
+        profile_dir = YAOEX_SPIDER_DIR / self.account_name
+        # 仅兼容历史误路径，新建不再使用 chrome_profile
         legacy_nested = YAOEX_SPIDER_DIR / "spiders" / "yaoex" / self.account_name
+        legacy_chrome_profile = YAOEX_SPIDER_DIR / "chrome_profile" / self.account_name
 
-        for candidate in (preferred, legacy_flat, legacy_nested):
+        for candidate in (profile_dir, legacy_nested, legacy_chrome_profile):
             if (candidate / "Default").is_dir() or (candidate / "Local State").is_file():
                 logger.info("使用已有浏览器配置目录: %s", candidate)
                 return candidate
 
-        preferred.parent.mkdir(parents=True, exist_ok=True)
-        logger.info("新建浏览器配置目录: %s", preferred)
-        return preferred
+        profile_dir.mkdir(parents=True, exist_ok=True)
+        logger.info("新建浏览器配置目录: %s", profile_dir)
+        return profile_dir
 
     def init_browser(self):
         co = ChromiumOptions().set_browser_path(chrome_path)
@@ -384,7 +457,24 @@ class YaoexSnapshotCrawl:
     def fetch_list_page(self, keyword, page):
         list_url = "https://gateway-b2b.fangkuaiyi.com/home/search/homeSearchList"
         resp = self._post_with_retry(list_url, self._list_payload(keyword, page))
-        return resp.json().get("data", {}).get("shopProducts", []) or []
+        data = resp.json()
+        recall_status = data.get("data", {}).get("recallStatus", 0)
+        data_block = data.get("data", {}) or {}
+        # 尝试提取总页数（常见字段名依次尝试）
+        total_pages = 0
+        for key in ("totalPage", "totalPages", "pageCount", "total", "totalCount", "totalNum", "totalSize"):
+            val = data_block.get(key)
+            if val is not None and int(val) > 0:
+                total_pages = int(val)
+                break
+        if total_pages:
+            logger.info("📊 列表 API 检测到总页数：%s (page=%s)", total_pages, page)
+        else:
+            logger.debug("列表 API 未返回总页数字段，data keys: %s", list(data_block.keys())[:20])
+        if int(recall_status) == 1:
+            return data_block.get("shopProducts", []) or [], total_pages
+        else:
+            return [], 0
 
     def fetch_shop(self, seller_code):
         detail_url = "https://gateway-b2b.fangkuaiyi.com/ycapp/shop/enterpriseQualification"
@@ -457,7 +547,11 @@ class YaoexSnapshotCrawl:
                 time.sleep(0.5)
                 self.driver.refresh()
                 time.sleep(2)
-                return True
+                ele = self.driver.ele("xpath=//div[@class='yaoex-product-detail__product-detail']")
+                if ele:
+                    return True
+                else:
+                    continue
             except Exception as e:
                 logger.warning(
                     "跳转详情异常 spu=%s seller=%s attempt=%s: %s",
@@ -582,15 +676,20 @@ class YaoexSnapshotCrawl:
             keyword = (keyword + " " + self.product_desc).strip()
 
         for page in range(self.start_page, self.end_page + 1):
+            if self.scheduler.end:
+                logger.warning("收到停止信号，终止采集")
+                break
             logger.info("正在爬取 %s %s，第%s页", self.brand, self.product, page)
-            page_items = self.fetch_list_page(keyword=keyword, page=page)
+            page_items, api_total_pages = self.fetch_list_page(keyword=keyword, page=page)
             if not page_items:
                 logger.info("第%s页无数据，停止", page)
                 break
 
             for item in page_items:
+                if self.scheduler.end:
+                    logger.warning("收到停止信号，终止当前页面采集")
+                    break
                 item, detail_url, spu_code, seller_code = self._build_detail_url(item)
-
                 name_part = (item.get("productName") or "").strip()
                 short_part = (item.get("shortName") or "").strip()
                 product_name = f"{name_part} {short_part}".strip()
@@ -628,15 +727,36 @@ class YaoexSnapshotCrawl:
             if self.is_not_product > NOT_PRODUCT_BREAK:
                 logger.info("连续不匹配商品过多，停止搜索")
                 break
+            # 每页结束上报进度，检查限额
+            self._page_no = page
+            prog_status = self._report_progress(page, total_pages=api_total_pages)
+            if prog_status == "limit_reached":
+                logger.warning("⛔ 进度上报检测到限额，通知停止")
+                self.scheduler.set_flag(True)
             time.sleep(random.uniform(1, 3))
 
         return True
 
     def run(self):
+        self._page_no = 0
         try:
+            self.scheduler.start()
+            self._report_start()
             self.init_browser()
             self.search()
         except Exception as e:
             logger.exception("运行异常: %s", e)
+            self.success = False
         finally:
+            self.scheduler.stop()
+            try:
+                self._report_end(
+                    success=self.success,
+                    real_count=self.pipeline.crawl_count,
+                    page_no=self._page_no,
+                )
+            except Exception:
+                pass
             self._quit_browser()
+        logger.info("爬取总数%s", self.pipeline.crawl_count)
+        return self.pipeline.crawl_count, self.success