|
|
@@ -1,13 +1,13 @@
|
|
|
"""
|
|
|
壹药城 (Yaoex) 平台采集入口
|
|
|
-使用 commons1 调度框架 —— 从内部 API 拉取任务，由 YaoexCrawler / YaoexSnapshotCrawl 执行采集。
|
|
|
+使用 commons1 调度框架 —— 从内部 API 拉取任务，由 YaoexCrawler / YaoexSnapshotCrawl 执行采集，
|
|
|
+拉取任务、心跳、回告全部走 commons。
|
|
|
"""
|
|
|
import random
|
|
|
import re
|
|
|
import time
|
|
|
from datetime import datetime
|
|
|
|
|
|
-import requests
|
|
|
from commons.scheduler import CrawlerScheduler
|
|
|
from commons.Logger import get_spider_logger
|
|
|
from commons.feishu_webhook import send_text, send_error_card
|
|
|
@@ -25,8 +25,34 @@ 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 ""
|
|
|
@@ -56,22 +82,17 @@ def main_loop():
|
|
|
logger.info("壹药城采集启动，每轮间隔 %s 秒", idle_seconds)
|
|
|
|
|
|
while True:
|
|
|
+ task_id = None
|
|
|
try:
|
|
|
- # 先同步心跳确保登记，再启守护线程拉任务
|
|
|
- import requests as _r
|
|
|
- _r.post(
|
|
|
- scheduler.heartbeat_url,
|
|
|
- json={"platform": scheduler.platform, "username": scheduler.username},
|
|
|
- headers={'X-Crawler-Token': 'zhijiayun_crawler_2026'},
|
|
|
- timeout=5
|
|
|
- )
|
|
|
- scheduler.start()
|
|
|
+ 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:
|
|
|
@@ -81,16 +102,33 @@ def main_loop():
|
|
|
spider_cls = YaoexCrawler
|
|
|
logger.info("snapshot_collect_status=%s → 启用接口模式 YaoexCrawler", snapshot_status)
|
|
|
|
|
|
- logger.info("开始执行%s爬虫任务 task_id=%s", PLATFORM_NAME, task.get("id"))
|
|
|
+ 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.get("id"), crawl_count, is_success)
|
|
|
+ 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)
|
|
|
|
shanxiaohong