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

@@ -171,6 +171,7 @@ class YaoexCrawler:
         self.collect_region_id = self.task_dict.get("collect_region_id", "")
         self.collect_round = self.task_dict.get("collect_round", 1)
 
+    @staticmethod
     def _timestamp_ms() -> str:
         return str(int(time.time() * 1000))
 
@@ -300,7 +301,7 @@ class YaoexCrawler:
         raw_price = item.get("price")
         price = self.decrypt_price(raw_price)
 
-        hash_text = str(seller_code)+str(price)
+        hash_text = f"{seller_code}_{spu_code}_{price}"
         item_id = hashlib.md5(hash_text.encode('utf-8')).hexdigest()
 
         is_sold_out = 0
@@ -433,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)
@@ -443,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)
@@ -451,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

@@ -729,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)
@@ -741,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:
@@ -750,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

@@ -810,12 +810,10 @@ class YbmCrawler:
                     else:
                         msg = data.get("msg", "")
                         if "已达平台限额" in msg or "限额" in msg:
-                            logger.warning("回告接口返回限额错误：code=%s, msg=%s",
-                                         data.get('code'), 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)
+                            logger.warning("回告接口返回错误：code=%s, msg=%s", data.get('code'), msg)
                             return "error"
                 except ValueError:
                     logger.warning("回告响应非 JSON：%s", resp.text[:200])
@@ -1152,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)
@@ -1470,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)
@@ -1597,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
@@ -1618,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

@@ -8,16 +8,13 @@ import time
 from datetime import datetime
 
 import requests
-
+from commons.scheduler import CrawlerScheduler
 from commons.Logger import get_spider_logger
 from commons.feishu_webhook import send_text, send_error_card
 from spiders.yaoex.yaoex_crawl import (
     YaoexCrawler,
     PLATFORM_ID,
     YYC_DEVICE_ID,
-    TASK_PULL_URL,
-    TASK_HEARTBEAT_URL,
-    TASK_API_TOKEN,
 )
 from spiders.yaoex.yaoex_snapshot_crawl import YaoexSnapshotCrawl
 
@@ -28,42 +25,6 @@ IDLE_SECONDS_MIN = 180
 IDLE_SECONDS_MAX = 300
 
 
-def pull_task():
-    """从内部调度 API 拉取任务"""
-    headers = {'X-Crawler-Token': TASK_API_TOKEN}
-    params = {'platform': PLATFORM_ID, 'username': YYC_DEVICE_ID}
-    try:
-        response = requests.get(TASK_PULL_URL, params=params, headers=headers, timeout=15)
-        result = response.json()
-        logger.info("拉取任务 API 返回: %s", result)
-        if result.get('code') == 'success':
-            task = result.get('data', {}).get('task') or {}
-            if task and task.get("id"):
-                logger.info("拉取任务成功 platform=%s task_id=%s", PLATFORM_ID, task.get("id"))
-                return task
-            else:
-                logger.info("拉取任务成功但无有效任务（task=%s），视为暂无任务", task)
-                return {}
-        else:
-            logger.warning("拉取任务 API 返回非 success: code=%s, msg=%s",
-                           result.get('code'), result.get('msg'))
-            return {}
-    except Exception as e:
-        logger.error("拉取任务 API 请求失败: %s", e)
-        return {}
-
-
-def send_heartbeat():
-    """向调度 API 上报心跳，告知服务器爬虫在线"""
-    headers = {'X-Crawler-Token': TASK_API_TOKEN}
-    payload = {"platform": str(PLATFORM_ID), "username": YYC_DEVICE_ID}
-    try:
-        resp = requests.post(TASK_HEARTBEAT_URL, json=payload, headers=headers, timeout=10)
-        logger.debug("心跳上报: HTTP %s, %s", resp.status_code, resp.text[:200])
-    except Exception as e:
-        logger.warning("心跳上报失败: %s", e)
-
-
 def notify_result(task, crawl_count, success):
     """发送飞书通知"""
     try:
@@ -90,14 +51,22 @@ def notify_result(task, crawl_count, success):
 
 def main_loop():
     """循环拉取任务并执行采集"""
+    scheduler = CrawlerScheduler(YYC_DEVICE_ID, str(PLATFORM_ID))
     idle_seconds = random.randint(IDLE_SECONDS_MIN, IDLE_SECONDS_MAX)
     logger.info("壹药城采集启动，每轮间隔 %s 秒", idle_seconds)
 
     while True:
         try:
-            # 先上报心跳，告诉服务器爬虫在线
-            send_heartbeat()
-            task = pull_task()
+            # 先同步心跳确保登记，再启守护线程拉任务
+            import requests as _r
+            _r.post(
+                scheduler.heartbeat_url,
+                json={"platform": scheduler.platform, "username": scheduler.username},
+                headers={'X-Crawler-Token': 'zhijiayun_crawler_2026'},
+                timeout=5
+            )
+            scheduler.start()
+            task = scheduler.get_task() or {}
             if not task:
                 logger.info("%s 暂无任务，等待 %s 秒后重试", PLATFORM_NAME, idle_seconds)
                 time.sleep(idle_seconds)

start_run_ybm.py

@@ -8,16 +8,13 @@ import re
 from datetime import datetime
 
 import requests
-
+from commons.scheduler import CrawlerScheduler
 from commons.Logger import get_spider_logger
 from commons.feishu_webhook import send_text, send_error_card
 from spiders.ybm.ybm_crawl import (
     YbmCrawler,
     PLATFORM_ID,
     YBM_DEVICE_ID,
-    TASK_PULL_URL,
-    TASK_HEARTBEAT_URL,
-    TASK_API_TOKEN,
     load_city_mapping,
 )
 
@@ -28,42 +25,6 @@ IDLE_SECONDS_MIN = 180
 IDLE_SECONDS_MAX = 300
 
 
-def pull_task():
-    """从内部调度 API 拉取任务"""
-    headers = {'X-Crawler-Token': TASK_API_TOKEN}
-    params = {'platform': PLATFORM_ID, 'username': YBM_DEVICE_ID}
-    try:
-        response = requests.get(TASK_PULL_URL, params=params, headers=headers, timeout=15)
-        result = response.json()
-        logger.info("拉取任务 API 返回: %s", result)
-        if result.get('code') == 'success':
-            task = result.get('data', {}).get('task') or {}
-            if task and task.get("id"):
-                logger.info("拉取任务成功 platform=%s task_id=%s", PLATFORM_ID, task.get("id"))
-                return task
-            else:
-                logger.info("拉取任务成功但无有效任务（task=%s），视为暂无任务", task)
-                return {}
-        else:
-            logger.warning("拉取任务 API 返回非 success: code=%s, msg=%s",
-                           result.get('code'), result.get('msg'))
-            return {}
-    except Exception as e:
-        logger.error("拉取任务 API 请求失败: %s", e)
-        return {}
-
-
-def send_heartbeat():
-    """向调度 API 上报心跳，告知服务器爬虫在线"""
-    headers = {'X-Crawler-Token': TASK_API_TOKEN}
-    payload = {"platform": str(PLATFORM_ID), "username": YBM_DEVICE_ID}
-    try:
-        resp = requests.post(TASK_HEARTBEAT_URL, json=payload, headers=headers, timeout=10)
-        logger.debug("心跳上报: HTTP %s, %s", resp.status_code, resp.text[:200])
-    except Exception as e:
-        logger.warning("心跳上报失败: %s", e)
-
-
 def notify_result(task, crawl_count, success):
     """发送飞书通知"""
     try:
@@ -91,14 +52,21 @@ def notify_result(task, crawl_count, success):
 def main_loop():
     """循环拉取任务并执行采集"""
     load_city_mapping()
+    scheduler = CrawlerScheduler(YBM_DEVICE_ID, str(PLATFORM_ID))
     idle_seconds = random.randint(IDLE_SECONDS_MIN, IDLE_SECONDS_MAX)
     logger.info("药帮忙采集启动，每轮间隔 %s 秒", idle_seconds)
 
     while True:
         try:
-            # 先上报心跳，告诉服务器爬虫在线
-            send_heartbeat()
-            task = pull_task()
+            # 先同步上报心跳，确保服务器登记后再拉任务
+            requests.post(
+                scheduler.heartbeat_url,
+                json={"platform": scheduler.platform, "username": scheduler.username},
+                headers={'X-Crawler-Token': 'zhijiayun_crawler_2026'},
+                timeout=5
+            )
+            scheduler.start()
+            task = scheduler.get_task() or {}
             if not task:
                 logger.info("%s 暂无任务，等待 %s 秒后重试", PLATFORM_NAME, idle_seconds)
                 time.sleep(idle_seconds)