From de7c34034296ec10cb2799706fd2a16b755e4694 Mon Sep 17 00:00:00 2001
From: zlt <zlt@zlt.com>
Date: Mon, 3 Aug 2026 16:33:23 +0800
Subject: [PATCH] =?UTF-8?q?=E9=80=9A=E8=BF=87=E8=B0=83=E5=BA=A6=E7=B3=BB?=
 =?UTF-8?q?=E7=BB=9F=E6=8E=A5=E5=85=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 api.py              |   5 +-
 config.py           |   6 ++
 test_xxl_trigger.py | 208 ++++++++++++++++++++++++++++++++++++++++++++
 xxl.py              | 206 +++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 424 insertions(+), 1 deletion(-)
 create mode 100644 test_xxl_trigger.py
 create mode 100644 xxl.py

diff --git a/api.py b/api.py
index 0782413..6177670 100644
--- a/api.py
+++ b/api.py
@@ -10,7 +10,8 @@ cron 的 docker exec 构建/出计划照旧，互不影响。局域网内部服
   GET  /plan/dates                 可用日期列表
   POST /plan/refresh?date=...      重新生成该日计划文件（data/plan/*.md）
 
-将来接 XXL-JOB / 事件回调，触发器打这层即可，不必进容器。
+统一任务调度平台（XXL-JOB）触发入口挂在 /api/v1/xxl/*（见 xxl.py，2026-08-03）：
+盘前链（build → plan → push-pool）可由平台拉起并回调结案，.env 配 XXL_TRIGGER_KEY 才启用。
 """
 import pandas as pd
 from fastapi import FastAPI, HTTPException
@@ -18,8 +19,10 @@ from fastapi.responses import PlainTextResponse
 
 import db
 import plan
+from xxl import router as xxl_router
 
 app = FastAPI(title="akg-factor-bridge · 每日选股计划", version="0.1")
+app.include_router(xxl_router)
 
 
 @app.get("/health")
diff --git a/config.py b/config.py
index 4269a15..4dbe9bb 100644
--- a/config.py
+++ b/config.py
@@ -147,6 +147,12 @@ ENABLE_TRACK_GATE = os.environ.get("ENABLE_TRACK_GATE", "0") == "1"
 # 因此不进主榜、观察档、计划与升降档。研究口径想看全貌时置 0 关闭。
 EXCLUDE_RISK_NAMES = os.environ.get("EXCLUDE_RISK_NAMES", "1") == "1"
 
+# --- 统一任务调度平台触发（08-03；键名与决策系统保持一致，便于平台侧统一配置）----
+# 空串 = /api/v1/xxl/* 整组端点禁用（安全默认）。回调契约见 xxl.py 模块说明。
+XXL_TRIGGER_KEY = os.environ.get("XXL_TRIGGER_KEY", "")
+XXL_CALLBACK_READ_TIMEOUT = float(os.environ.get("XXL_CALLBACK_READ_TIMEOUT", "10"))
+XXL_CALLBACK_MAX_RETRIES = int(os.environ.get("XXL_CALLBACK_MAX_RETRIES", "3"))
+
 # --- 选股计划入池（08-03 定稿；规则与流程见 docs/选股计划入池_对接说明.md）------
 # 写 Mongo stock_groups 的独立分组，决策系统每晚扫描按分组并集覆盖 → 候选票自动
 # 获得夜间推理。入池范围与 PMS 候选同口径：强传导主榜前 POOL_TOP 只 + 当前持仓。
diff --git a/test_xxl_trigger.py b/test_xxl_trigger.py
new file mode 100644
index 0000000..8a5fc20
--- /dev/null
+++ b/test_xxl_trigger.py
@@ -0,0 +1,208 @@
+# -*- coding: utf-8 -*-
+"""平台触发入口（xxl.py）的单测——子进程与回调都打桩，不跑真链、不发真请求。
+
+运行:  docker compose exec -T akg-factor-bridge python test_xxl_trigger.py
+全过输出 "ALL PASS (n cases)"，任一失败退出码 1。
+"""
+import sys
+import time
+import traceback
+
+from fastapi import HTTPException
+
+import config
+import xxl
+
+RESULTS = []
+
+
+def case(name):
+    def deco(fn):
+        RESULTS.append((name, fn))
+        return fn
+    return deco
+
+
+def _reset(key="test-key"):
+    """每例开跑前清场：配置好 key、清任务表、装默认桩。"""
+    config.XXL_TRIGGER_KEY = key
+    config.XXL_CALLBACK_MAX_RETRIES = 0
+    xxl._jobs.clear()
+    xxl._running_id = None
+    ran, sent = [], []
+    xxl._run_step = lambda cmd: (ran.append(list(cmd)), (0, "ok"))[1]
+    xxl._post_callback = lambda url, data: (sent.append((url, data)), "delivered http=200")[1]
+    return ran, sent
+
+
+def _wait_done(task_id, timeout=5.0):
+    """等到任务连同结案回调都处理完（closed=True），避免断言撞上后台线程的时间差。"""
+    t0 = time.time()
+    while time.time() - t0 < timeout:
+        if xxl._jobs[task_id].get("closed"):
+            return xxl._jobs[task_id]
+        time.sleep(0.02)
+    raise AssertionError("任务超时未结束")
+
+
+def _trigger(**kw):
+    args = dict(callbackUrl=None, steps="build,plan,push-pool", date=None,
+                key="test-key", x_job_key=None)
+    args.update(kw)
+    return xxl.trigger_daily_build(**args)
+
+
+@case("没配 XXL_TRIGGER_KEY → 整组端点禁用 (403)，与决策系统同款安全默认")
+def _():
+    _reset(key="")
+    try:
+        _trigger(key="whatever")
+        raise AssertionError("该 403 没 403")
+    except HTTPException as e:
+        assert e.status_code == 403, e
+
+
+@case("key 不对 → 401；对了 → 受理并返回 task_id")
+def _():
+    ran, _ = _reset()
+    try:
+        _trigger(key="wrong")
+        raise AssertionError("该 401 没 401")
+    except HTTPException as e:
+        assert e.status_code == 401, e
+    r = _trigger()
+    assert r["status"] == "accepted" and r["task_id"], r
+    _wait_done(r["task_id"])
+
+
+@case("三步按固定顺序跑完 → success，成功回调是空包")
+def _():
+    ran, sent = _reset()
+    r = _trigger(callbackUrl="http://xxl/callback?logId=1")
+    job = _wait_done(r["task_id"])
+    assert job["state"] == "success", job
+    assert [c[1] for c in ran] == ["build", "plan", "push-pool"], ran
+    assert sent and sent[0][0].startswith("http://xxl/callback"), sent
+    assert sent[0][1] == b"", sent          # 成功=空包，对端按成功结案
+
+
+@case("中间一步失败 → 后面的步骤不跑，失败回调带 handleCode=500 与原因")
+def _():
+    ran, sent = _reset()
+    xxl._run_step = lambda cmd: (ran.append(list(cmd)),
+                                 (1, "boom") if cmd[1] == "plan" else (0, "ok"))[1]
+    r = _trigger(callbackUrl="http://xxl/cb")
+    job = _wait_done(r["task_id"])
+    assert job["state"] == "failed" and "plan" in job["note"], job
+    assert [c[1] for c in ran] == ["build", "plan"], ran      # push-pool 没跑
+    body = sent[0][1].decode()
+    assert "handleCode=500" in body and "plan" in body, body
+
+
+@case("同一时刻只允许一条链: 在跑时再触发 → 409")
+def _():
+    _reset()
+    gate = {"open": False}
+
+    def slow(cmd):
+        while not gate["open"]:
+            time.sleep(0.01)
+        return 0, "ok"
+    xxl._run_step = slow
+    r1 = _trigger(steps="build")
+    try:
+        _trigger(steps="build")
+        raise AssertionError("该 409 没 409")
+    except HTTPException as e:
+        assert e.status_code == 409, e
+    finally:
+        gate["open"] = True
+    _wait_done(r1["task_id"])
+    r2 = _trigger(steps="build")                 # 跑完后可以再触发
+    _wait_done(r2["task_id"])
+
+
+@case("steps 白名单与顺序: 乱序请求被摆正, 未知步骤 400")
+def _():
+    ran, _ = _reset()
+    r = _trigger(steps="push-pool,build")
+    _wait_done(r["task_id"])
+    assert [c[1] for c in ran] == ["build", "push-pool"], ran   # 顺序固定
+    try:
+        _trigger(steps="build,发大财")
+        raise AssertionError("该 400 没 400")
+    except HTTPException as e:
+        assert e.status_code == 400, e
+
+
+@case("date 参数透传到每一步命令")
+def _():
+    ran, _ = _reset()
+    r = _trigger(steps="build,plan", date="2026-08-01")
+    _wait_done(r["task_id"])
+    for cmd in ran:
+        assert cmd[-2:] == ["--date", "2026-08-01"], cmd
+
+
+@case("无 callbackUrl 也能正常结案 (只记日志, 平台走状态查询)")
+def _():
+    _, sent = _reset()
+    r = _trigger(callbackUrl=None, steps="build")
+    job = _wait_done(r["task_id"])
+    assert job["state"] == "success" and not sent
+
+
+@case("状态接口: 能查到 state/steps/tail; 不认识的 task_id 404; 也要带 key")
+def _():
+    _reset()
+    r = _trigger(steps="build")
+    _wait_done(r["task_id"])
+    st = xxl.job_status(r["task_id"], key="test-key", x_job_key=None)
+    assert st["state"] == "success" and st["steps"] == ["build"] and st["tail"]
+    try:
+        xxl.job_status("no-such-id", key="test-key", x_job_key=None)
+        raise AssertionError("该 404 没 404")
+    except HTTPException as e:
+        assert e.status_code == 404, e
+    try:
+        xxl.job_status(r["task_id"], key="wrong", x_job_key=None)
+        raise AssertionError("该 401 没 401")
+    except HTTPException as e:
+        assert e.status_code == 401, e
+
+
+@case("回调连接层失败: 重试次数用尽后放弃, 任务状态不受影响")
+def _():
+    _reset()
+    config.XXL_CALLBACK_MAX_RETRIES = 0          # 失败一次即放弃, 免得测试睡 30 秒
+
+    def dead(url, data):
+        raise OSError("connect refused")
+    xxl._post_callback = dead
+    r = _trigger(callbackUrl="http://xxl/cb", steps="build")
+    job = _wait_done(r["task_id"])
+    assert job["state"] == "success", job        # 回调失败不改任务结论
+    assert any("放弃" in ln for ln in job["tail"]), job["tail"]
+
+
+# ---------------------------------------------------------------- runner
+def main():
+    passed, failed = 0, 0
+    for name, fn in RESULTS:
+        try:
+            fn()
+            print(f"  PASS  {name}")
+            passed += 1
+        except Exception:
+            print(f"  FAIL  {name}")
+            traceback.print_exc()
+            failed += 1
+    print("-" * 60)
+    if failed:
+        print(f"FAILED: {failed} / {passed + failed}")
+        sys.exit(1)
+    print(f"ALL PASS ({passed} cases)")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/xxl.py b/xxl.py
new file mode 100644
index 0000000..ee77376
--- /dev/null
+++ b/xxl.py
@@ -0,0 +1,206 @@
+"""统一任务调度平台（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