更新 'spiders/yaoex/yaoex_crawl.py'

spiders/yaoex/yaoex_crawl.py

@@ -3,6 +3,7 @@ import random
 import time
 import requests
 from commons.Logger import get_spider_logger
+from commons.scheduler import CrawlerScheduler
 import base64
 from Crypto.Cipher import AES
 from Crypto.Util.Padding import unpad
@@ -13,6 +14,16 @@ logger = get_spider_logger("yaoex")
 
 TOKEN = "Sm45MzRmREtiaStVTnJORXEySHhYYzNwUmQ2RUprWXlwelRDem4wV2RZUCtUUU5jMGVCVTRYYjNLVjdNSnFWSjg1YStxWllGQ2RQSExjaEVqU0dOaDFJczl4bTB1V09CZHZzVml2dU0xazd3UDdla3FTUzZBZlZkMHFSVHlaaDhDcFp3SWNDb3JNSDhuNC9vUzI1RVdEaU01YjcxQW5TS21Sdy90ZDRENi9VR2E0SW5wOWF4UE1VZ0poTDhhVkJtP2FwcElkPTEyNTAma2V5SWQ9MTI1MA=="
 
+# 平台常量（API 调度用）
+PLATFORM_ID = 6
+YYC_DEVICE_ID = "18193030281"
+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"
+HEARTBEAT_INTERVAL_SECONDS = 60
+
 headers = {
     "Accept": "application/json, text/plain, */*",
     "Accept-Language": "zh-CN,zh;q=0.9",
@@ -47,6 +58,7 @@ class YaoexCrawler:
         self.collect_task_id = None
         self.account_name = None
         self.pipeline = DrugPipeline("yaoex")
+        self.scheduler = CrawlerScheduler(YYC_DEVICE_ID, str(PLATFORM_ID), heartbeat_interval=HEARTBEAT_INTERVAL_SECONDS)
         if self.task_dict:
             self.get_product_data()
 
@@ -90,6 +102,74 @@ class YaoexCrawler:
         self.collect_region_id = self.task_dict.get("collect_region_id", "")
         self.collect_round = self.task_dict.get("collect_round", 1)
 
+    # ---------- 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 _timestamp_ms() -> str:
         return str(int(time.time() * 1000))
@@ -164,7 +244,23 @@ class YaoexCrawler:
         resp = self._post_with_retry(list_url, payload)
         data = resp.json()
 
-        return data.get("data", {}).get("shopProducts", []) or []
+        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_detail(self, spu_code, seller_code):
         payload = self._detail_payload(spu_code, seller_code)
@@ -300,8 +396,11 @@ class YaoexCrawler:
             keyword = keyword + " " + self.product_desc
 
         for page in range(1, 100):
+            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:
                 break
             for item in page_items:
@@ -315,9 +414,9 @@ class YaoexCrawler:
                 if self.product not in product_name:
                     self.is_not_product += 1
                     continue
-                # if self.brand not in product_name:
-                #     self.is_not_product += 1
-                #     continue
+                if self.brand not in product_name:
+                    self.is_not_product += 1
+                    continue
                 self.is_not_product = 0
                 product = self.parse_product(item)
 
@@ -332,14 +431,33 @@ class YaoexCrawler:
                 time.sleep(random.randint(1, 3))
             if self.is_not_product > 15:
                 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.randint(2, 5))
 
     def run(self):
+        self._page_no = 0
         try:
+            self.scheduler.start()
+            self._report_start()
             self.search_data()
         except Exception as e:
             logger.error(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=self._page_no,
+                )
+            except Exception:
+                pass
         logger.info(f"爬取总数{self.pipeline.crawl_count}")
         return self.pipeline.crawl_count, self.is_success