""" 壹药城 (Yaoex) 平台采集入口 使用 commons1 调度框架 —— 从内部 API 拉取任务,由 YaoexCrawler / YaoexSnapshotCrawl 执行采集。 """ import random import re import time 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.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 logger = get_spider_logger("yaoex_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': 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: 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(): """循环拉取任务并执行采集""" 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 # 根据 snapshot_collect_status 选择爬虫类型 snapshot_status = task.get("snapshot_collect_status", 1) if snapshot_status == 0: spider_cls = YaoexSnapshotCrawl logger.info("snapshot_collect_status=%s → 启用快照模式 YaoexSnapshotCrawl", snapshot_status) else: spider_cls = YaoexCrawler logger.info("snapshot_collect_status=%s → 启用接口模式 YaoexCrawler", snapshot_status) logger.info("开始执行%s爬虫任务 task_id=%s", PLATFORM_NAME, task.get("id")) crawler = spider_cls(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()