project/xhs/commons/scheduler.py

import time
import requests
from commons.Logger import get_spider_logger
import threading

logger = get_spider_logger('scheduler')


class CrawlerScheduler:
    """爬虫任务调度器"""

    def __init__(self, DEVICE_ID, platform, heartbeat_interval=30):
        """
        初始化调度器

        Args:
            platform: 平台名称
            heartbeat_url: 心跳上报URL
            heartbeat_interval: 心跳间隔时间（秒）
        """

        self.username = DEVICE_ID
        self.platform = platform

        self._lock = threading.Lock()

        self.heartbeat_url = 'http://pricesys.kailin.com.cn:8083/api/collect_task/heartbeat'
        self.heartbeat_interval = heartbeat_interval
        self.end = False
        self.heartbeat_thread = None

    def _heartbeat_reporter(self):
        """守护线程：只负责上报心跳"""
        headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
        while True:
            try:
                json_data = {
                    "platform": self.platform,
                    "username": self.username
                }
                logger.info(f'上传心跳数据：{json_data},链接:{self.heartbeat_url}')
                response = requests.post(
                    self.heartbeat_url,
                    json=json_data,
                    headers=headers,
                    timeout=3
                )

                print(f"[心跳] 发送成功: {response.status_code}")
                result = response.json()

                if self.end:
                    break

                if result.get('code') != 'success':
                    logger.error(f'心跳回传异常：{result}')
                else:
                    logger.info(f'心跳回传：{result}')

                time.sleep(self.heartbeat_interval)

            except Exception as e:
                print(e)
                logger.error(e)
                time.sleep(5)

    def start(self):
        """启动调度器"""
        self.set_flag(False)
        if self.heartbeat_thread and self.heartbeat_thread.is_alive():
            return
        self.heartbeat_thread = threading.Thread(
            target=self._heartbeat_reporter,
            daemon=True
        )
        self.heartbeat_thread.start()

    def stop(self):
        """停止调度器"""
        self.set_flag(True)
        logger.info('心跳停止')

    def get_task(self):
        try:
            task_api = "http://pricesys.kailin.com.cn:8083/api/collect_task/pull"

            headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}

            params = {
                'platform': self.platform,
                'username': self.username
            }
            logger.info(f'拉取任务参数：{params},链接:{task_api}')

            response = requests.get(task_api, params=params, headers=headers, timeout=10)
            result = response.json()
            logger.info(f'拉取任务：{result}')
            if result.get('code') == 'success':
                return result.get('data').get('task')

            print('拉取任务返回', result)

        except Exception as e:
            logger.error('获取任务报错', e)

    def post_report(self, data):
        try:
            url = "http://pricesys.kailin.com.cn:8083/api/collect_task/report"
            print('传给返回接口的数据', data)
            logger.info(f'report上传数据：{data},链接:{url}')
            headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
            response = requests.post(url, json=data, headers=headers, timeout=5)
            # 记录日志
            result = response.json()

            logger.info(f"report回传数据{result}")

            if (result.get('code') != 'success'):
                logger.error(f'翻页回传结果不成功：{result}')
                self.stop()
            print(f'任务进度上传 {result}')
        except Exception as e:
            logger.error(e)

    def set_flag(self, value):
        with self._lock:
            self.end = value