"""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