start_run_yaoex.py 3.6 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798
  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. requests.post(
  54. scheduler.heartbeat_url,
  55. json={"platform": scheduler.platform, "username": scheduler.username},
  56. headers={'X-Crawler-Token': 'zhijiayun_crawler_2026'},
  57. timeout=5
  58. )
  59. scheduler.start()
  60. task = scheduler.get_task() or {}
  61. if not task:
  62. logger.info("%s 暂无任务,等待 %s 秒后重试", PLATFORM_NAME, idle_seconds)
  63. time.sleep(idle_seconds)
  64. continue
  65. # 根据 snapshot_collect_status 选择爬虫类型
  66. snapshot_status = task.get("snapshot_collect_status", 1)
  67. if snapshot_status == 0:
  68. spider_cls = YaoexSnapshotCrawl
  69. logger.info("snapshot_collect_status=%s → 启用快照模式 YaoexSnapshotCrawl", snapshot_status)
  70. else:
  71. spider_cls = YaoexCrawler
  72. logger.info("snapshot_collect_status=%s → 启用接口模式 YaoexCrawler", snapshot_status)
  73. logger.info("开始执行%s爬虫任务 task_id=%s", PLATFORM_NAME, task.get("id"))
  74. crawler = spider_cls(task)
  75. crawl_count, is_success = crawler.run()
  76. logger.info("%s爬虫任务执行完成 task_id=%s count=%s success=%s",
  77. PLATFORM_NAME, task.get("id"), crawl_count, is_success)
  78. notify_result(task, crawl_count, is_success)
  79. except Exception as e:
  80. logger.error("%s 爬虫任务执行失败: %s", PLATFORM_NAME, e, exc_info=True)
  81. time.sleep(idle_seconds)
  82. if __name__ == '__main__':
  83. main_loop()