上传文件至 ''

start_run_ybm.py

@@ -0,0 +1,121 @@
+"""
+药帮忙 (YBM) 平台采集入口
+使用 commons1 调度框架 —— 从内部 API 拉取任务，由 YbmCrawler 执行采集。
+"""
+import random
+import time
+import re
+from datetime import datetime
+
+import requests
+
+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,
+)
+
+logger = get_spider_logger("ybm_runner")
+
+PLATFORM_NAME = "药帮忙"
+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:
+        drug_name = task.get("product_name", "")
+        spec = task.get("product_specs", "") or ""
+        spec_items = [item.strip() for item in re.split(r"[|、，,\n\r]+", spec) if item.strip()]
+        display_spec = "、".join(spec_items)
+        notice_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        text_msg = (
+            f"{notice_time} 通知:\n"
+            f"平台: {PLATFORM_NAME}, 药品: {drug_name}, 品规: {display_spec}, 爬取数据: {crawl_count}条"
+        )
+        if success:
+            send_text(text_msg)
+        else:
+            send_error_card(
+                task_name=drug_name,
+                err_msg=f"task_id={task.get('id')}, company_id={task.get('company_id')}, crawl_count={crawl_count}",
+                mention_all=False,
+            )
+    except Exception as e:
+        logger.warning("飞书通知发送失败：%s", e)
+
+
+def main_loop():
+    """循环拉取任务并执行采集"""
+    load_city_mapping()
+    idle_seconds = random.randint(IDLE_SECONDS_MIN, IDLE_SECONDS_MAX)
+    logger.info("药帮忙采集启动，每轮间隔 %s 秒", idle_seconds)
+
+    while True:
+        try:
+            # 先上报心跳，告诉服务器爬虫在线
+            send_heartbeat()
+            task = pull_task()
+            if not task:
+                logger.info("%s 暂无任务，等待 %s 秒后重试", PLATFORM_NAME, idle_seconds)
+                time.sleep(idle_seconds)
+                continue
+
+            logger.info("开始执行%s爬虫任务 task_id=%s", PLATFORM_NAME, task.get("id"))
+            crawler = YbmCrawler(task)
+            crawl_count, is_success = crawler.run()
+            logger.info("%s爬虫任务执行完成 task_id=%s count=%s success=%s",
+                        PLATFORM_NAME, task.get("id"), crawl_count, is_success)
+
+            notify_result(task, crawl_count, is_success)
+
+        except Exception as e:
+            logger.error("%s 爬虫任务执行失败: %s", PLATFORM_NAME, e, exc_info=True)
+
+        time.sleep(idle_seconds)
+
+
+if __name__ == '__main__':
+    main_loop()