更新 'spiders/ybm/ybm_crawl.py'

spiders/ybm/ybm_crawl.py

@@ -42,7 +42,12 @@ 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
@@ -787,6 +792,82 @@ class YbmCrawler:
             "sampling_end_time": self.sampling_end_time,
         }, ensure_ascii=False, default=str)
 
+
+    # ---------- 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_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"开始上报 task_id={self.task_id}")
+
+    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):
+        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"结束上报 task_id={self.task_id}")
+
     # ---------- login ----------
 
     def _is_login(self, page):