project/tbsg/commons/scheduler.py

import time
import requests
import threading
import urllib3

from commons import err_log

# 禁用SSL警告（https 自签名证书）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)


def _log(msg):
    print(f"[scheduler] {msg}")


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

        # 回告接口返回 code=error（如：该账号今日爬取量已达平台限额，任务已释放）
        # → 置位标志，采集主循环(step3)每批检查，立即停止采集
        self.limit_reached = False
        self.limit_msg = ""

        self._lock = threading.Lock()

        # ⭐ 必须用 https：http 会被 nginx 301 重定向，POST 会被降级成 GET 导致 405（美团同款坑）
        self.heartbeat_url = 'https://pricesys.kailin.com.cn:8082/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
                }
                _log(f'上传心跳数据：{json_data},链接:{self.heartbeat_url}')
                response = requests.post(
                    self.heartbeat_url,
                    json=json_data,
                    headers=headers,
                    timeout=3,
                    verify=False   # 忽略SSL证书验证
                )

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

                if self.end:
                    break

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

                time.sleep(self.heartbeat_interval)

            except Exception as e:
                print(e)
                _log(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)
        _log('心跳停止')

    def get_task(self):
        try:
            # ⭐ 必须用 https（原因同上：http 会被 301 重定向降级）
            task_api = "https://pricesys.kailin.com.cn:8082/api/collect_task/pull"

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

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

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

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

        except Exception as e:
            _log('获取任务报错', e)
            err_log.log_error(self.username, "pull_task_error", exc=e)

    def post_report(self, data):
        try:
            # ⭐ 必须用 https（原因同上：http 会被 301 重定向降级）
            url = "https://pricesys.kailin.com.cn:8082/api/collect_task/report"
            print('传给返回接口的数据', data)
            _log(f'report上传数据：{data},链接:{url}')
            headers = {'X-Crawler-Token': 'zhijiayun_crawler_2026'}
            response = requests.post(url, json=data, headers=headers, timeout=5, verify=False)
            # 记录日志
            result = response.json()

            _log(f"report回传数据{result}")

            if (result.get('code') != 'success'):
                _log(f'翻页回传结果不成功：{result}')
                self.stop()
            if result.get('code') == 'error':
                # 平台限额/任务已释放等错误 → 置位标志，采集主循环立即停止
                self.limit_msg = str(result.get('msg', ''))
                self.limit_reached = True
                _log(f'回告返回error，置位停止采集标志：{self.limit_msg}')
            print(f'任务进度上传 {result}')
        except Exception as e:
            _log(e)
            err_log.log_error(self.username, "report_error", exc=e,
                              extra={"report": {k: data.get(k) for k in ("task_id", "is_finished", "need_reassign", "current_page", "crawled_count") if isinstance(data, dict)}})

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