"""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 = "l112233" XC_TOKEN = "40cb6910dc9a442a816298a228da65ac" 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 # 99 records whose status was exactly "waiting"; completed/failed tasks are # deliberately excluded. DEFAULT_TASK_IDS = [ 3471434, 3471433, 3471431, 3471430, 3471429, 3471428, 3471427, 3471426, 3471425, 3471424, 3471423, 3471422, 3471421, 3471420, 3471419, 3471418, 3471416, 3471415, 3471414, 3471413, 3471412, 3471411, 3471410, 3471409, 3471408, 3471407, 3471406, 3471405, 3471404, 3471403, 3471402, 3471401, 3471400, 3471399, 3471398, 3471397, 3471396, 3471395, 3471394, 3471393, 3471392, 3471391, 3471390, 3471389, 3471388, 3471387, 3471386, 3471385, 3471384, 3471383, 3471382, 3471381, 3471380, 3471379, 3471378, 3471377, 3471376, 3471375, 3471374, 3471373, 3471372, 3471371, 3471370, 3471369, 3471368, 3471367, 3471366, 3471365, 3471364, 3471363, 3471362, 3471361, 3471360, 3471359, 3471358, 3471357, 3471356, 3471355, 3471354, 3471353, 3471352, 3471351, 3471350, 3471348, 3471347, 3471346, 3471345, 3471344, 3471343, 3471342, 3471341, 3471340, 3471339, 3471338, 3471337, 3471336, 3007636, 3007635, 3007634, ] _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()