修改ybm和yyc

commons/config.py

@@ -17,4 +17,7 @@ YYC_ACCOUNT = {
     "password":"a666666",
     "token":"Sm45MzRmREtiaThGc1gvLzZaVnRFWXFjTlg4bVBZaytxZXoyRjR5OXB1NmFJMk84VDhadzI4NUdNUmhFZjFtZCszSnFGV1JUdThVTWtwUi9SVG9jUk01T0JNZ1RyUVZCNUsyNms0Z0F3OUhRc3JKRzZ1S0dFOWNLUVAwNFdweWRtaDkyS25iOUVwU3dnUC9FcVpnZWdSbS9QQm1qSHhsOS84RHlOc1czUWgxRTZkSEh4d2NUOCtkNzA1dTlLWjkzP2FwcElkPTEyNTAma2V5SWQ9MTI1MA==",
     "user_id":387715
-}
+}
+
+# 壹药城设备ID（用于内部调度 API）
+YYC_DEVICE_ID = "18193030281"

spiders/yaoex/yaoex_crawl.py

@@ -19,6 +19,11 @@ PLATFORM_ID = 6
 YYC_DEVICE_ID = "18193030281"
 
 HEARTBEAT_INTERVAL_SECONDS = 60
+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"
 
 headers = {
     "Accept": "application/json, text/plain, */*",
@@ -61,6 +66,74 @@ class YaoexCrawler:
         self.is_success = True
         self.is_not_product = 0
 
+    # ---------- 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）")
+
     def _post_with_retry(self, url, payload, retries=REQUEST_RETRY_COUNT, timeout=REQUEST_TIMEOUT_SEC):
         last_err = None
         for attempt in range(1, retries + 1):
@@ -361,7 +434,7 @@ class YaoexCrawler:
                 break
             # 每页结束上报进度，检查限额
             self._page_no = page
-            prog_status = self.scheduler.report_progress(self.task_id, page, total_pages=api_total_pages)
+            prog_status = self._report_progress(page, total_pages=api_total_pages)
             if prog_status == "limit_reached":
                 logger.warning("⛔ 进度上报检测到限额，通知停止")
                 self.scheduler.set_flag(True)
@@ -371,7 +444,7 @@ class YaoexCrawler:
         self._page_no = 0
         try:
             self.scheduler.start()
-            self.scheduler.report_start(self.task_id)
+            self._report_start()
             self.search_data()
         except Exception as e:
             logger.error(e)
@@ -379,8 +452,7 @@ class YaoexCrawler:
         finally:
             self.scheduler.stop()
             try:
-                self.scheduler.report_end(
-                    self.task_id,
+                self._report_end(
                     success=self.is_success,
                     real_count=self.pipeline.crawl_count,
                     page_no=self._page_no,

spiders/yaoex/yaoex_snapshot_crawl.py

@@ -24,6 +24,9 @@ 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"
@@ -103,6 +106,74 @@ class YaoexSnapshotCrawl:
         self.start_page = self._parse_page(self.task_dict.get("start_page"), 1)
         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):
         try:
@@ -658,7 +729,7 @@ class YaoexSnapshotCrawl:
                 break
             # 每页结束上报进度，检查限额
             self._page_no = page
-            prog_status = self.scheduler.report_progress(self.task_id, page, total_pages=api_total_pages)
+            prog_status = self._report_progress(page, total_pages=api_total_pages)
             if prog_status == "limit_reached":
                 logger.warning("⛔ 进度上报检测到限额，通知停止")
                 self.scheduler.set_flag(True)
@@ -670,7 +741,7 @@ class YaoexSnapshotCrawl:
         self._page_no = 0
         try:
             self.scheduler.start()
-            self.scheduler.report_start(self.task_id)
+            self._report_start()
             self.init_browser()
             self.search()
         except Exception as e:
@@ -679,8 +750,7 @@ class YaoexSnapshotCrawl:
         finally:
             self.scheduler.stop()
             try:
-                self.scheduler.report_end(
-                    self.task_id,
+                self._report_end(
                     success=self.success,
                     real_count=self.pipeline.crawl_count,
                     page_no=self._page_no,

spiders/ybm/ybm_crawl.py

@@ -42,6 +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 +793,79 @@ class YbmCrawler:
         }, 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):
@@ -1071,7 +1150,7 @@ class YbmCrawler:
                 if extracted_total is not None:
                     total_pages = extracted_total
                     # 读到总页数立刻回报进度
-                    prog_status = self.scheduler.report_progress(self.task_id, page_no, total_pages=total_pages)
+                    prog_status = self._report_progress(page_no, total_pages=total_pages)
                     if prog_status == "limit_reached":
                         logger.warning("⛔ 进度上报检测到限额，通知停止")
                         self.scheduler.set_flag(True)
@@ -1389,7 +1468,7 @@ class YbmCrawler:
             logger.info(f"⏳ 翻页前随机等待 {delay:.2f}s（反爬）")
 
             # 每页结束上报进度
-            prog_status = self.scheduler.report_progress(self.task_id, page_no, total_pages=total_pages)
+            prog_status = self._report_progress(page_no, total_pages=total_pages)
             if prog_status == "limit_reached":
                 logger.warning("⛔ 进度上报检测到限额，通知停止")
                 self.scheduler.set_flag(True)
@@ -1516,7 +1595,7 @@ class YbmCrawler:
                 store_page.wait_for_load_state('networkidle')
 
                 # 上报采集开始，检查是否限额
-                report_status = self.scheduler.report_start(self.task_id)
+                report_status = self._report_start()
                 if report_status == "limit_reached":
                     logger.warning("⛔ 任务 %s 开始上报时账号已达限额，退出当前任务", self.task_id)
                     self.is_success = False
@@ -1537,8 +1616,7 @@ class YbmCrawler:
                 self.scheduler.stop()
                 # 上报采集结束
                 try:
-                    self.scheduler.report_end(
-                        self.task_id,
+                    self._report_end(
                         success=self.is_success,
                         real_count=self.pipeline.crawl_count,
                         page_no=getattr(self, '_page_no', 0),

start_run_yaoex.py

@@ -57,8 +57,9 @@ def main_loop():
 
     while True:
         try:
-            # 先同步上报心跳，确保服务器登记后再拉任务
-            requests.post(
+            # 先同步心跳确保登记，再启守护线程拉任务
+            import requests as _r
+            _r.post(
                 scheduler.heartbeat_url,
                 json={"platform": scheduler.platform, "username": scheduler.username},
                 headers={'X-Crawler-Token': 'zhijiayun_crawler_2026'},

start_run_ybm.py

@@ -83,6 +83,8 @@ def main_loop():
         except Exception as e:
             logger.error("%s 爬虫任务执行失败: %s", PLATFORM_NAME, e, exc_info=True)
 
-        if __name__ == '__main__':
-            main_loop()
         time.sleep(idle_seconds)
+
+
+if __name__ == '__main__':
+    main_loop()