""" 药帮忙 (YBM) 平台采集入口 使用 commons1 调度框架 —— 从内部 API 拉取任务,由 YbmCrawler 执行采集, 拉取任务、心跳、回告全部走 commons。 """ import random import time import re 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.ybm.ybm_crawl import ( YbmCrawler, PLATFORM_ID, YBM_DEVICE_ID, load_city_mapping, ) logger = get_spider_logger("ybm_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(): """循环拉取任务并执行采集""" 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: 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") logger.info("开始执行%s爬虫任务 task_id=%s", PLATFORM_NAME, task_id) crawler = YbmCrawler(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()