207 lines
9.3 KiB
Python
207 lines
9.3 KiB
Python
"""统一任务调度平台（XXL-JOB）触发入口 —— 盘前链的外部拉起与结案回调。
|
||
|
||
参照决策系统 bionic_trader 的同款接入（app/api/xxl_jobs.py 的 trigger_daily_scan +
|
||
tasks_periodic 的 notify_xxl），对平台表现完全一致：
|
||
|
||
触发: GET|POST /api/v1/xxl/daily-build?key=...&steps=...&date=...
|
||
立即返回 200 + task_id；平台会自动在 URL 末尾追加 &callbackUrl=...
|
||
状态: GET /api/v1/xxl/status/{task_id} 回调未送达时的兜底查询
|
||
结案: 任务真正跑完后 POST callbackUrl ——
|
||
成功 = 空包（对端默认按成功结案）；
|
||
失败 = 表单 {"handleCode": 500, "handleMsg": 原因}。
|
||
有 HTTP 响应（含 4xx/5xx）即算送达、不重试；只有连接层失败（不可达/超时）
|
||
才隔 30 秒重试，最多 XXL_CALLBACK_MAX_RETRIES 次，放弃后靠状态查询兜底。
|
||
|
||
与决策系统那套只有一处实现差别：桥没有 celery，任务由 API 进程起一条后台线程、
|
||
逐步以**子进程**方式执行 `python run.py <步骤>`。用子进程而不是进程内调用，是为了
|
||
保住「每次执行都重读 .env、重载代码」的既有性质——常驻 API 自身不会自动重载，
|
||
但它派生的子进程会，和宿主 cron 的 docker exec 语义完全一致。
|
||
|
||
安全默认与决策系统相同：.env 不配 XXL_TRIGGER_KEY，这一组端点整体禁用。
|
||
同一时刻只允许一条链在跑：已有任务未结束时再触发回 409，由平台按失败重试策略处理。
|
||
|
||
步骤白名单（steps 参数，逗号分隔，默认三步全跑，执行顺序固定）：
|
||
build 因子构建（run.py build all --mode daily）
|
||
plan 生成当日选股计划文件
|
||
push-pool 计划写入股票池 + 触发决策系统增量补扫
|
||
"""
|
||
import datetime as dt
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import threading
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid
|
||
|
||
from fastapi import APIRouter, Header, HTTPException, Query
|
||
|
||
import config
|
||
|
||
router = APIRouter(prefix="/api/v1/xxl", tags=["xxl"])
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
LOG_PATH = os.path.join(HERE, "data", "xxl_build.log")
|
||
|
||
# 步骤 → 命令（顺序即执行顺序；--date 由触发参数统一追加）
|
||
STEP_ORDER = ("build", "plan", "push-pool")
|
||
STEP_CMDS = {
|
||
"build": ["run.py", "build", "all", "--mode", "daily"],
|
||
"plan": ["run.py", "plan"],
|
||
"push-pool": ["run.py", "push-pool"],
|
||
}
|
||
STEP_TIMEOUT_SEC = 3600 # 单步上限一小时，防呆死（正常盘前链全程分钟级）
|
||
|
||
_lock = threading.Lock()
|
||
_jobs: dict = {} # task_id -> 状态字典（常驻进程内存，重启即清空）
|
||
_running_id = None
|
||
_KEEP_JOBS = 20 # 只留最近 20 条状态，防内存慢涨
|
||
|
||
|
||
def _require_key(x_job_key, key):
|
||
cfg = config.XXL_TRIGGER_KEY
|
||
if not cfg:
|
||
raise HTTPException(status_code=403,
|
||
detail="XXL trigger disabled: XXL_TRIGGER_KEY not set")
|
||
if (x_job_key or key) != cfg:
|
||
raise HTTPException(status_code=401, detail="invalid key")
|
||
|
||
|
||
def _log(job, line: str):
|
||
stamp = dt.datetime.now().strftime("%H:%M:%S")
|
||
job["tail"].append(f"[{stamp}] {line}")
|
||
job["tail"] = job["tail"][-60:]
|
||
try:
|
||
os.makedirs(os.path.dirname(LOG_PATH), exist_ok=True)
|
||
with open(LOG_PATH, "a", encoding="utf-8") as f:
|
||
f.write(f"{dt.datetime.now().isoformat(timespec='seconds')} "
|
||
f"[{job['task_id'][:8]}] {line}\n")
|
||
except Exception: # noqa: BLE001 —— 落盘日志失败不影响任务本体
|
||
pass
|
||
|
||
|
||
def _run_step(cmd: list) -> tuple:
|
||
"""跑一步（子进程）。单测在这里打桩。返回 (退出码, 合并输出)。"""
|
||
p = subprocess.run([sys.executable] + cmd, cwd=HERE, capture_output=True,
|
||
text=True, timeout=STEP_TIMEOUT_SEC)
|
||
return p.returncode, (p.stdout or "") + (p.stderr or "")
|
||
|
||
|
||
def _post_callback(url: str, data: bytes) -> str:
|
||
"""回调一跳（单测在这里打桩）。返回送达说明；连接层失败抛异常。"""
|
||
req = urllib.request.Request(url, data=data, method="POST")
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=config.XXL_CALLBACK_READ_TIMEOUT) as resp:
|
||
return f"delivered http={resp.status}"
|
||
except urllib.error.HTTPError as e:
|
||
return f"delivered http={e.code}" # 对端有响应即算送达，不重试
|
||
|
||
|
||
def _notify(job, callback_url: str, ok: bool, note: str):
|
||
"""结案回调，契约与决策系统 notify_xxl 一致。"""
|
||
if not callback_url:
|
||
_log(job, "无 callbackUrl，跳过结案回调（平台可查状态接口）")
|
||
return
|
||
data = (b"" if ok else urllib.parse.urlencode(
|
||
{"handleCode": 500, "handleMsg": (note or "job failed")[:500]}).encode())
|
||
for attempt in range(config.XXL_CALLBACK_MAX_RETRIES + 1):
|
||
try:
|
||
_log(job, f"结案回调 ok={ok}: {_post_callback(callback_url, data)}")
|
||
return
|
||
except Exception as e: # noqa: BLE001 —— 连接层失败才会走到这
|
||
if attempt >= config.XXL_CALLBACK_MAX_RETRIES:
|
||
_log(job, f"结案回调最终未达，放弃（靠状态查询兜底）: {e!r}")
|
||
return
|
||
_log(job, f"结案回调连接失败，30 秒后第 {attempt + 1} 次重试: {e!r}")
|
||
time.sleep(30)
|
||
|
||
|
||
def _run_job(task_id: str, steps: list, date, callback_url):
|
||
global _running_id
|
||
job = _jobs[task_id]
|
||
ok, note = True, ""
|
||
try:
|
||
for step in steps:
|
||
job["step"] = step
|
||
cmd = list(STEP_CMDS[step]) + (["--date", date] if date else [])
|
||
_log(job, f"开始 {step}: python {' '.join(cmd)}")
|
||
try:
|
||
code, out = _run_step(cmd)
|
||
except subprocess.TimeoutExpired:
|
||
ok, note = False, f"{step} 超过 {STEP_TIMEOUT_SEC} 秒未结束，判失败"
|
||
_log(job, note)
|
||
break
|
||
for line in (out or "").strip().splitlines()[-15:]:
|
||
_log(job, f" {line}")
|
||
if code != 0:
|
||
ok, note = False, f"{step} 退出码 {code}"
|
||
_log(job, note)
|
||
break
|
||
_log(job, f"完成 {step}")
|
||
except Exception as e: # noqa: BLE001 —— 线程里绝不让异常无声消失
|
||
ok, note = False, f"意外异常: {type(e).__name__}: {e}"
|
||
_log(job, note)
|
||
job.update(state="success" if ok else "failed", note=note,
|
||
ended_at=dt.datetime.now().isoformat(timespec="seconds"), step=None)
|
||
with _lock:
|
||
_running_id = None
|
||
_notify(job, callback_url, ok, note)
|
||
job["closed"] = True # 结案回调也处理完了（送达或放弃），状态接口可据此判断
|
||
|
||
|
||
@router.api_route("/daily-build", methods=["GET", "POST"])
|
||
def trigger_daily_build(
|
||
callbackUrl: str = Query(None, description="平台下发的结案回调地址（自动追加）"),
|
||
steps: str = Query("build,plan,push-pool",
|
||
description="要跑哪几步，逗号分隔；顺序固定 build→plan→push-pool"),
|
||
date: str = Query(None, description="补跑指定数据日 YYYY-MM-DD；空=最新数据日"),
|
||
key: str = Query(None),
|
||
x_job_key: str = Header(None),
|
||
):
|
||
"""触发盘前链。立即返回 task_id，跑完由结案回调通知平台。
|
||
|
||
平台执行参数示例（方式 GET，key 写进 URL，平台会自己追加 callbackUrl）：
|
||
http://<桥机IP>:8300/api/v1/xxl/daily-build?key=<KEY>
|
||
"""
|
||
global _running_id
|
||
_require_key(x_job_key, key)
|
||
wanted = [s.strip() for s in (steps or "").split(",") if s.strip()]
|
||
bad = [s for s in wanted if s not in STEP_CMDS]
|
||
if bad or not wanted:
|
||
raise HTTPException(status_code=400,
|
||
detail=f"steps 只认 {list(STEP_ORDER)}，收到 {wanted}")
|
||
ordered = [s for s in STEP_ORDER if s in wanted]
|
||
|
||
with _lock:
|
||
if _running_id and _jobs.get(_running_id, {}).get("state") == "running":
|
||
raise HTTPException(status_code=409,
|
||
detail=f"已有任务在跑（task_id={_running_id}），本次拒绝")
|
||
task_id = str(uuid.uuid4())
|
||
_jobs[task_id] = {"task_id": task_id, "state": "running", "steps": ordered,
|
||
"step": None, "date": date, "note": "", "tail": [],
|
||
"started_at": dt.datetime.now().isoformat(timespec="seconds"),
|
||
"ended_at": None, "closed": False}
|
||
while len(_jobs) > _KEEP_JOBS: # 只留最近 N 条
|
||
oldest = next(iter(_jobs))
|
||
if oldest == task_id:
|
||
break
|
||
_jobs.pop(oldest)
|
||
_running_id = task_id
|
||
|
||
threading.Thread(target=_run_job, args=(task_id, ordered, date, callbackUrl),
|
||
daemon=True).start()
|
||
return {"status": "accepted", "task_id": task_id, "steps": ordered, "date": date}
|
||
|
||
|
||
@router.get("/status/{task_id}")
|
||
def job_status(task_id: str, key: str = Query(None), x_job_key: str = Header(None)):
|
||
"""状态兜底：running / success / failed + 当前步骤 + 最近输出。"""
|
||
_require_key(x_job_key, key)
|
||
job = _jobs.get(task_id)
|
||
if not job:
|
||
raise HTTPException(status_code=404,
|
||
detail="task_id 不存在（API 重启会清空历史状态）")
|
||
return job
|