修改ybm和yyc

spiders/yaoex/yaoex_crawl.py

@@ -19,11 +19,6 @@ 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, */*",
@@ -66,74 +61,6 @@ 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):
@@ -171,6 +98,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 +228,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

spiders/yaoex/yaoex_snapshot_crawl.py

@@ -24,9 +24,6 @@ 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"
@@ -106,74 +103,6 @@ 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:

spiders/ybm/ybm_crawl.py

@@ -42,12 +42,6 @@ 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
@@ -793,81 +787,6 @@ 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):

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,21 @@ 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()
+            # 先同步上报心跳，确保服务器登记后再拉任务
+            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)

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)
@@ -115,8 +83,6 @@ 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()