更新 'spiders/ybm/ybm_crawl.py'

spiders/ybm/ybm_crawl.py

@@ -42,12 +42,7 @@ 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
@@ -792,83 +787,6 @@ 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):
-        """通用回告方法，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):
@@ -1153,7 +1071,7 @@ class YbmCrawler:
                 if extracted_total is not None:
                     total_pages = extracted_total
                     # 读到总页数立刻回报进度
-                    prog_status = self._report_progress(page_no, total_pages=total_pages)
+                    prog_status = self.scheduler.report_progress(self.task_id, page_no, total_pages=total_pages)
                     if prog_status == "limit_reached":
                         logger.warning("⛔ 进度上报检测到限额，通知停止")
                         self.scheduler.set_flag(True)
@@ -1471,7 +1389,7 @@ class YbmCrawler:
             logger.info(f"⏳ 翻页前随机等待 {delay:.2f}s（反爬）")
 
             # 每页结束上报进度
-            prog_status = self._report_progress(page_no, total_pages=total_pages)
+            prog_status = self.scheduler.report_progress(self.task_id, page_no, total_pages=total_pages)
             if prog_status == "limit_reached":
                 logger.warning("⛔ 进度上报检测到限额，通知停止")
                 self.scheduler.set_flag(True)
@@ -1598,7 +1516,7 @@ class YbmCrawler:
                 store_page.wait_for_load_state('networkidle')
 
                 # 上报采集开始，检查是否限额
-                report_status = self._report_start()
+                report_status = self.scheduler.report_start(self.task_id)
                 if report_status == "limit_reached":
                     logger.warning("⛔ 任务 %s 开始上报时账号已达限额，退出当前任务", self.task_id)
                     self.is_success = False
@@ -1619,7 +1537,8 @@ class YbmCrawler:
                 self.scheduler.stop()
                 # 上报采集结束
                 try:
-                    self._report_end(
+                    self.scheduler.report_end(
+                        self.task_id,
                         success=self.is_success,
                         real_count=self.pipeline.crawl_count,
                         page_no=getattr(self, '_page_no', 0),
@@ -1642,18 +1561,10 @@ if __name__ == '__main__':
         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 = {}
+        scheduler = CrawlerScheduler(YBM_DEVICE_ID, str(PLATFORM_ID))
+        task = scheduler.get_task() or {}
+        if not task:
+            logger.error("拉取任务失败")
 
     if not task:
         logger.error("未获取到任何任务，程序退出")