""" 壹药城 (Yaoex) 平台采集入口 使用 commons1 调度框架 —— 从内部 API 拉取任务,由 YaoexCrawler / YaoexSnapshotCrawl 执行采集, 拉取任务、心跳、回告全部走 commons。 """ import random import re import time from datetime import datetime 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, ) 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 _report_progress(scheduler, task_id, is_finished, crawled_count=0, current_page=0, total_pages=0, exception_type=None, remark=None): """通过 commons 调度器 API 回告任务进度""" data = { "task_id": task_id, "platform": str(PLATFORM_ID), "username": scheduler.username, "current_page": current_page, "is_finished": is_finished, "total_pages": total_pages, } if crawled_count: data["crawled_count"] = crawled_count if exception_type is not None: data["exception_type"] = exception_type if remark: data["remark"] = remark try: scheduler.post_report(data) logger.info("任务回告成功: is_finished=%s, crawled_count=%s", is_finished, crawled_count) except Exception as e: logger.warning("任务回告失败: %s", e) def notify_result(task, crawl_count, success): """发送飞书通知(也走 commons.feishu_webhook)""" 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(): """循环拉取任务并执行采集""" 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: task_id = None try: scheduler.start() # 启动心跳守护线程(commons 内部管理) task = scheduler.get_task() or {} if not task: logger.info("%s 暂无任务,等待 %s 秒后重试", PLATFORM_NAME, idle_seconds) time.sleep(idle_seconds) continue task_id = task.get("id") # 根据 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_id) crawler = spider_cls(task) crawl_count, is_success = crawler.run() is_finished = 1 if is_success else 0 logger.info("%s爬虫任务执行完成 task_id=%s count=%s success=%s", PLATFORM_NAME, task_id, crawl_count, is_success) # 回告:通过 commons scheduler 上报结果 _report_progress( scheduler, task_id, is_finished=is_finished, crawled_count=crawl_count, ) # 飞书通知(也走 commons) notify_result(task, crawl_count, is_success) except Exception as e: logger.error("%s 爬虫任务执行失败: %s", PLATFORM_NAME, e, exc_info=True) # 异常也通过 commons 回告 if task_id is not None: _report_progress( scheduler, task_id, is_finished=0, exception_type=type(e).__name__, remark=str(e)[:500], ) time.sleep(idle_seconds) if __name__ == '__main__': main_loop()