long_articles_agentic_supply/supply/server/xxl_endpoints.py

"""XXL-JOB Admin → Executor 通信端点 —— 对标 Java XxlJobController.java

提供给 XXL-JOB Admin 调用的 HTTP API：
    POST /beat       — 心跳检测
    POST /idleBeat   — 空闲检测
    POST /run        — 触发任务执行
    POST /kill       — 强制终止任务
    POST /log        — 读取执行日志
"""

import logging

from quart import Blueprint, request, jsonify

logger = logging.getLogger(__name__)

# 全局引用，由 create_xxl_bp 注入
_executor_ref = None


def set_xxl_executor(executor):
    """注入 XxlJobExecutor 实例（在 app 启动时调用）"""
    global _executor_ref
    _executor_ref = executor


async def _parse_body():
    """解析请求体：优先 JSON，兜底 form-data

    XXL-JOB Admin 不同版本发送方式不同（JSON 或 form），统一处理。
    """
    data = await request.get_json(silent=True)
    if data:
        return data
    return await request.form


def create_xxl_bp() -> Blueprint:
    """创建 XXL-JOB API Blueprint

    XXL-JOB Admin 通过此 Blueprint 与执行器通信。
    端点路径与 XXL-JOB 标准一致：/beat, /idleBeat, /run, /kill, /log
    """
    bp = Blueprint("xxl_job", __name__)

    @bp.route("/beat", methods=["POST"])
    async def beat():
        """心跳检测"""
        result = _executor_ref.beat()
        return jsonify(result)

    @bp.route("/idleBeat", methods=["POST"])
    async def idle_beat():
        """空闲检测：检查指定 jobId 是否正在运行"""
        body = await _parse_body()
        job_id = int(body.get("jobId", 0))
        result = _executor_ref.idle_beat(job_id)
        return jsonify(result)

    @bp.route("/run", methods=["POST"])
    async def run():
        """触发任务执行"""
        body = await _parse_body()
        trigger_params = {
            "jobId": int(body.get("jobId", 0)),
            "executorHandler": body.get("executorHandler", "").strip(),
            "executorParams": body.get("executorParams", ""),
            "executorBlockStrategy": body.get("executorBlockStrategy", ""),
            "executorTimeout": int(body.get("executorTimeout", 0)),
            "logId": int(body.get("logId", 0)),
            "logDateTime": int(body.get("logDateTime", 0)),
            "glueType": body.get("glueType", ""),
            "glueSource": body.get("glueSource", ""),
            "glueUpdatetime": int(body.get("glueUpdatetime", 0)),
            "broadcastIndex": int(body.get("broadcastIndex", 0)),
            "broadcastTotal": int(body.get("broadcastTotal", 1)),
        }
        logger.info(
            "XXL-JOB run: handler=%s, jobId=%s, params=%s",
            trigger_params["executorHandler"],
            trigger_params["jobId"],
            trigger_params["executorParams"],
        )
        result = await _executor_ref.run(trigger_params)
        return jsonify(result)

    @bp.route("/kill", methods=["POST"])
    async def kill():
        """强制终止任务"""
        body = await _parse_body()
        job_id = int(body.get("jobId", 0))
        logger.info("XXL-JOB kill: jobId=%s", job_id)
        result = await _executor_ref.kill(job_id)
        return jsonify(result)

    @bp.route("/log", methods=["POST"])
    async def log():
        """读取执行日志"""
        body = await _parse_body()
        log_date_time = int(body.get("logDateTim", 0))
        log_id = int(body.get("logId", 0))
        from_line_num = int(body.get("fromLineNum", 1))
        result = _executor_ref.log(log_date_time, log_id, from_line_num)
        return jsonify(result)

    return bp