From 4f0725332e252483cad3f6acbe182552f66aaf5b Mon Sep 17 00:00:00 2001 From: zhouyuanxi Date: Sat, 25 Jul 2026 15:18:52 +0800 Subject: [PATCH] Discover current zhoukaile validation tasks at runtime --- README.md | 12 ++++---- main.py | 87 ++++++++++++++++++++++++++++++++++++++++++++++--------- 2 files changed, 80 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index fac764d..87a5c52 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,6 @@ # xc_validation_strategy_vllm_stop -用于停止 `zhoukaile` 账号验证任务的 ModelHub XC 策略服务。 +用于停止 `zhoukaile` 账号当前全部活跃验证任务的 ModelHub XC 策略服务。 服务启动后会调用： @@ -13,17 +13,18 @@ PUT /api/async/task/stop-create-contest-task ## 必填配置 -在策略构建/运行环境中配置机密环境变量： +在策略构建/运行环境中配置以下机密环境变量： ```text XC_TOKEN= +USER_PASSWORD= ``` -令牌不写入仓库。可选配置如下： +`USER_PASSWORD` 仅用于获得短期 Bearer Token，以分页读取该账号实时提交的验证任务；`XC_TOKEN` 用于停止接口。两者均不写入仓库。可选配置如下： | 变量 | 默认值 | 说明 | | --- | --- | --- | -| `TASK_IDS` | 内置的 21 个历史任务 ID | 逗号分隔的任务 ID；设置后会完全替换内置列表。 | +| `TASK_IDS` | 无 | 可选的逗号分隔任务 ID；设置后跳过实时查询，仅停止指定 ID。 | | `BASE_URL` | `https://modelhub.org.cn` | ModelHub 服务地址。 | | `BATCH_SIZE` | `50` | 每个停止请求包含的任务数。 | | `MAX_RETRIES` | `3` | 单批任务的最大请求次数。 | @@ -37,12 +38,13 @@ XC_TOKEN= ## 运行与构建 -仓库根目录的 `Dockerfile` 使用平台 Python 基础镜像，暴露 `8080` 端口并以 `python main.py` 启动。策略构建时请设置 `XC_TOKEN` 为机密变量；构建成功后启动实例即可执行停止操作。 +仓库根目录的 `Dockerfile` 使用平台 Python 基础镜像，暴露 `8080` 端口并以 `python main.py` 启动。策略构建时请将 `XC_TOKEN` 与 `USER_PASSWORD` 都设置为机密变量；构建成功后，策略会先分页读取当前账号的全部活跃任务，再分批停止。 本地验证： ```bash export XC_TOKEN='***' +export USER_PASSWORD='***' python main.py curl http://127.0.0.1:8080/health curl http://127.0.0.1:8080/status diff --git a/main.py b/main.py index 33b2ed5..a228415 100644 --- a/main.py +++ b/main.py @@ -1,4 +1,4 @@ -"""Stop the listed ModelHub XC validation tasks for the zhoukaile account. +"""Stop zhoukaile's active ModelHub XC validation tasks. The service performs the stop requests once at startup and then stays alive so the strategy platform can probe it through ``/health`` and inspect ``/status``. @@ -16,8 +16,11 @@ from typing import Any import requests BASE_URL = os.environ.get("BASE_URL", "https://modelhub.org.cn").rstrip("/") +LOGIN_ENDPOINT = "/adminApi/user/login" +TASK_PAGE_ENDPOINT = "/api/adapt/task/page" STOP_TASK_ENDPOINT = "/api/async/task/stop-create-contest-task" -USER_ACCOUNT = "zhoukaile" +USER_ACCOUNT = os.environ.get("USER_ACCOUNT", "zhoukaile") +USER_PASSWORD = os.environ.get("USER_PASSWORD", "") XC_TOKEN = os.environ.get("XC_TOKEN", "") STRATEGY_ID = os.environ.get("STRATEGY_ID", "") @@ -27,13 +30,8 @@ BATCH_SIZE = int(os.environ.get("BATCH_SIZE", "50")) MAX_RETRIES = int(os.environ.get("MAX_RETRIES", "3")) REQUEST_TIMEOUT = int(os.environ.get("REQUEST_TIMEOUT", "30")) -# The complete task list from stop_create_contest_task_zhoukaile.py. TASK_IDS -# may be supplied by the strategy runtime to replace this list without rebuild. -DEFAULT_TASK_IDS = [ - 2224725, 2224726, 2224729, 2224730, 2224732, 2224733, 2224734, - 2224736, 2224738, 2224739, 2224741, 2224743, 2224744, 2224746, - 2224747, 2224748, 2224750, 2224751, 2224752, 2224753, 2224754, -] +PAGE_SIZE = 100 +ACTIVE_STATUSES = {"waiting", "running", "pending", "processing", "queued"} _shutdown = threading.Event() _state: dict[str, Any] = { @@ -54,11 +52,11 @@ def _now() -> str: return datetime.now(timezone.utc).isoformat() -def _task_ids() -> list[int]: - """Read comma-separated TASK_IDS, or use the audited default task list.""" +def _task_ids_from_environment() -> list[int] | None: + """Return an explicit TASK_IDS override, if one was supplied.""" raw_task_ids = os.environ.get("TASK_IDS", "").strip() if not raw_task_ids: - return DEFAULT_TASK_IDS.copy() + return None task_ids: list[int] = [] for value in raw_task_ids.split(","): @@ -72,6 +70,65 @@ def _task_ids() -> list[int]: return list(dict.fromkeys(task_ids)) +def _login() -> str: + """Authenticate as the task owner and return a short-lived bearer token.""" + if not USER_PASSWORD: + raise RuntimeError("USER_PASSWORD is required to query zhoukaile's current task IDs") + response = requests.post( + f"{BASE_URL}{LOGIN_ENDPOINT}", + headers={"Content-Type": "application/json"}, + json={"userAccount": USER_ACCOUNT, "userPassword": USER_PASSWORD}, + timeout=REQUEST_TIMEOUT, + ) + response.raise_for_status() + result = response.json() + token = result.get("data", {}).get("token") + if result.get("code") != 0 or not isinstance(token, str) or not token: + raise RuntimeError(f"Task query login failed: {result.get('message', 'missing token')}") + return token + + +def _current_task_ids() -> list[int]: + """Fetch every active task currently submitted by USER_ACCOUNT. + + This intentionally queries the user-facing task page at run time. A static + list becomes obsolete as soon as zhoukaile submits more validation tasks. + """ + bearer_token = _login() + headers = {"Authorization": f"Bearer {bearer_token}"} + task_ids: list[int] = [] + page = 1 + + while True: + response = requests.get( + f"{BASE_URL}{TASK_PAGE_ENDPOINT}", + headers=headers, + params={"current": page, "pageSize": PAGE_SIZE}, + timeout=REQUEST_TIMEOUT, + ) + response.raise_for_status() + result = response.json() + if result.get("code") != 0: + raise RuntimeError(f"Task query failed: {result.get('message', 'unknown error')}") + + data = result.get("data") or {} + records = data.get("records") or [] + for record in records: + if str(record.get("status", "")).lower() not in ACTIVE_STATUSES: + continue + try: + task_ids.append(int(record["taskId"])) + except (KeyError, TypeError, ValueError): + print(f"[query] skipping record without a numeric taskId: {record}", flush=True) + + pages = int(data.get("pages") or 0) + if page >= pages: + break + page += 1 + + return list(dict.fromkeys(task_ids)) + + class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API if self.path == "/health": @@ -143,9 +200,11 @@ def _run_worker() -> None: if not XC_TOKEN: raise RuntimeError("XC_TOKEN is required and must be configured in the strategy environment") - task_ids = _task_ids() + task_ids = _task_ids_from_environment() or _current_task_ids() if not task_ids: - raise RuntimeError("No task IDs were configured") + print("[query] no active validation tasks found", flush=True) + _state["phase"] = "done" + return _state["total"] = len(task_ids) for start in range(0, len(task_ids), BATCH_SIZE):