3
0

start_run_yaoex.py 3.6 KB

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