3
0

start_run_yaoex.py 3.4 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192
  1. """
  2. 壹药城 (Yaoex) 平台采集入口
  3. 使用 commons1 调度框架 —— 从内部 API 拉取任务,由 YaoexCrawler / YaoexSnapshotCrawl 执行采集。
  4. """
  5. import random
  6. import re
  7. import time
  8. from datetime import datetime
  9. from commons.Logger import get_spider_logger
  10. from commons.scheduler import CrawlerScheduler
  11. from commons.feishu_webhook import send_text, send_error_card
  12. from spiders.yaoex.yaoex_crawl import (
  13. YaoexCrawler,
  14. PLATFORM_ID,
  15. YYC_DEVICE_ID,
  16. )
  17. from spiders.yaoex.yaoex_snapshot_crawl import YaoexSnapshotCrawl
  18. logger = get_spider_logger("yaoex_runner")
  19. PLATFORM_NAME = "壹药城"
  20. IDLE_SECONDS_MIN = 180
  21. IDLE_SECONDS_MAX = 300
  22. def notify_result(task, crawl_count, success):
  23. """发送飞书通知"""
  24. try:
  25. drug_name = task.get("product_name", "")
  26. spec = task.get("product_specs", "") or ""
  27. spec_items = [item.strip() for item in re.split(r"[|、,,\n\r]+", spec) if item.strip()]
  28. display_spec = "、".join(spec_items)
  29. notice_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
  30. text_msg = (
  31. f"{notice_time} 通知:\n"
  32. f"平台: {PLATFORM_NAME}, 药品: {drug_name}, 品规: {display_spec}, 爬取数据: {crawl_count}条"
  33. )
  34. if success:
  35. send_text(text_msg)
  36. else:
  37. send_error_card(
  38. task_name=drug_name,
  39. err_msg=f"task_id={task.get('id')}, company_id={task.get('company_id')}, crawl_count={crawl_count}",
  40. mention_all=False,
  41. )
  42. except Exception as e:
  43. logger.warning("飞书通知发送失败: %s", e)
  44. def main_loop():
  45. """循环拉取任务并执行采集"""
  46. scheduler = CrawlerScheduler(YYC_DEVICE_ID, str(PLATFORM_ID))
  47. idle_seconds = random.randint(IDLE_SECONDS_MIN, IDLE_SECONDS_MAX)
  48. logger.info("壹药城采集启动,每轮间隔 %s 秒", idle_seconds)
  49. while True:
  50. try:
  51. # 先同步心跳,确保服务器登记后再拉任务
  52. scheduler.heartbeat_once()
  53. scheduler.start()
  54. task = scheduler.get_task() or {}
  55. if not task:
  56. logger.info("%s 暂无任务,等待 %s 秒后重试", PLATFORM_NAME, idle_seconds)
  57. time.sleep(idle_seconds)
  58. continue
  59. # 根据 snapshot_collect_status 选择爬虫类型
  60. snapshot_status = task.get("snapshot_collect_status", 1)
  61. if snapshot_status == 0:
  62. spider_cls = YaoexSnapshotCrawl
  63. logger.info("snapshot_collect_status=%s → 启用快照模式 YaoexSnapshotCrawl", snapshot_status)
  64. else:
  65. spider_cls = YaoexCrawler
  66. logger.info("snapshot_collect_status=%s → 启用接口模式 YaoexCrawler", snapshot_status)
  67. logger.info("开始执行%s爬虫任务 task_id=%s", PLATFORM_NAME, task.get("id"))
  68. crawler = spider_cls(task)
  69. crawl_count, is_success = crawler.run()
  70. logger.info("%s爬虫任务执行完成 task_id=%s count=%s success=%s",
  71. PLATFORM_NAME, task.get("id"), crawl_count, is_success)
  72. notify_result(task, crawl_count, is_success)
  73. except Exception as e:
  74. logger.error("%s 爬虫任务执行失败: %s", PLATFORM_NAME, e, exc_info=True)
  75. time.sleep(idle_seconds)
  76. if __name__ == '__main__':
  77. main_loop()