"""Stop zhoukaile's waiting 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``. """ import json import os import signal import threading import time from datetime import datetime, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any import requests BASE_URL = os.environ.get("BASE_URL", "https://modelhub.org.cn").rstrip("/") STOP_TASK_ENDPOINT = "/api/async/task/stop-create-contest-task" USER_ACCOUNT = "zhoukaile" XC_TOKEN = "bd7c52f3b9604ef48a14dd6174513935" STRATEGY_ID = os.environ.get("STRATEGY_ID", "") HTTP_HOST = "0.0.0.0" HTTP_PORT = int(os.environ.get("PORT", "8080")) 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")) # Snapshot queried from /api/adapt/task/page on 2026-07-25. These are the # 100 records whose status was exactly "waiting"; completed/failed tasks are # deliberately excluded. DEFAULT_TASK_IDS = [ 3467793, 3467787, 3467785, 3467784, 3467782, 3467781, 3467779, 3467777, 3467776, 3467775, 3467773, 3467772, 3467771, 3467768, 3467761, 3467759, 3467757, 3467754, 3467753, 3467752, 3467751, 3467750, 3467749, 3467748, 3467747, 3467746, 3467745, 3467744, 3467743, 3467742, 3467741, 3467740, 3467739, 3467738, 3467736, 3467735, 3467734, 3467733, 3467732, 3467731, 3467730, 3467729, 3467728, 3467727, 3467726, 3467725, 3467724, 3467723, 3467721, 3467711, 3467705, 3467704, 3467703, 3467700, 3467699, 3467698, 3467697, 3467696, 3467695, 3467694, 3467691, 3467683, 3467679, 3467674, 3467673, 3467672, 3467671, 3467670, 3467669, 3467666, 3467663, 3467662, 3467660, 3467659, 3467658, 3467657, 3467656, 3467654, 3467653, 3467651, 3467650, 3467649, 3467648, 3467647, 3467646, 3467645, 3467644, 3467642, 3467641, 3467639, 3467638, 3467637, 3467636, 3467635, 3467634, 3467633, 3467632, 3467631, 3467630, 3467629, ] _shutdown = threading.Event() _state: dict[str, Any] = { "strategy_id": STRATEGY_ID, "account": USER_ACCOUNT, "phase": "starting", # starting | stopping | done | partial_failure | error "total": 0, "stopped": 0, "failed": 0, "failed_task_ids": [], "started_at": None, "finished_at": None, "error": None, } def _now() -> str: return datetime.now(timezone.utc).isoformat() 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 None task_ids: list[int] = [] for value in raw_task_ids.split(","): value = value.strip() if not value: continue try: task_ids.append(int(value)) except ValueError as exc: raise ValueError(f"TASK_IDS contains a non-numeric task ID: {value!r}") from exc return list(dict.fromkeys(task_ids)) class Handler(BaseHTTPRequestHandler): def do_GET(self) -> None: # noqa: N802 - BaseHTTPRequestHandler API if self.path == "/health": self._json({"status": "ok"}) elif self.path == "/status": self._json(_state) else: self._json({"error": "not found"}, 404) def _json(self, body: dict[str, Any], status: int = 200) -> None: payload = json.dumps(body, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) def log_message(self, fmt: str, *args: object) -> None: print(f"[http] {self.address_string()} {fmt % args}", flush=True) def _run_http() -> None: server = ThreadingHTTPServer((HTTP_HOST, HTTP_PORT), Handler) server.timeout = 1 print(f"[http] listening on {HTTP_HOST}:{HTTP_PORT}", flush=True) while not _shutdown.is_set(): server.handle_request() server.server_close() def _stop_batch(task_ids: list[int]) -> bool: headers = { "Content-Type": "application/json", "xc-Token": XC_TOKEN, } url = f"{BASE_URL}{STOP_TASK_ENDPOINT}" for attempt in range(1, MAX_RETRIES + 1): try: response = requests.put( url, headers=headers, json={"taskIds": task_ids}, timeout=REQUEST_TIMEOUT, ) try: result = response.json() except ValueError: result = {"message": response.text[:500]} if response.ok and result.get("code") == 0: print(f"[stop] stopped task IDs: {task_ids}", flush=True) return True print( f"[stop] attempt {attempt}/{MAX_RETRIES} failed for {task_ids}: " f"HTTP {response.status_code}, {result}", flush=True, ) except requests.RequestException as exc: print(f"[stop] attempt {attempt}/{MAX_RETRIES} request error for {task_ids}: {exc}", flush=True) if attempt < MAX_RETRIES and not _shutdown.wait(attempt): continue if _shutdown.is_set(): break return False def _run_worker() -> None: _state["started_at"] = _now() _state["phase"] = "stopping" try: task_ids = _task_ids_from_environment() or DEFAULT_TASK_IDS if not task_ids: 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): if _shutdown.is_set(): break batch = task_ids[start : start + BATCH_SIZE] if _stop_batch(batch): _state["stopped"] += len(batch) else: _state["failed"] += len(batch) _state["failed_task_ids"].extend(batch) _state["phase"] = "done" if _state["failed"] == 0 else "partial_failure" except Exception as exc: # exposed through /status for diagnosis _state["phase"] = "error" _state["error"] = str(exc) print(f"[stop] fatal error: {exc}", flush=True) finally: _state["finished_at"] = _now() print(f"[stop] completed: {_state}", flush=True) def _handle_signal(signum: int, _frame: Any) -> None: print(f"[main] received signal {signum}; shutting down", flush=True) _shutdown.set() def main() -> None: signal.signal(signal.SIGTERM, _handle_signal) signal.signal(signal.SIGINT, _handle_signal) http_thread = threading.Thread(target=_run_http, daemon=False) http_thread.start() threading.Thread(target=_run_worker, daemon=True).start() _shutdown.wait() http_thread.join(timeout=5) if __name__ == "__main__": main()