"""Stop waiting ModelHub XC validation tasks for the configured accounts. 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" 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-26. Only records # whose status was exactly "waiting" are included. TARGETS = [ { "account": "fanyi", "xc_token": "f2d501c9ae6543a589cd6cb789108c41", "task_ids": [ 3469928, 3469927, 3469926, 3469925, 3469924, 3469923, 3469922, 3469921, 3469920, 3469919, 3469918, 3469917, 3469916, 3469915, 3469914, 3469913, 3469911, 3469910, 3469909, 3469908, 3469907, 3469905, 3469904, 3469903, 3469902, 3469901, 3469900, 3469899, 3469898, 3469897, 3469896, 3469895, 3469894, 3469893, 3469892, 3469891, 3469890, 3469889, 3469888, 3469887, 3469886, 3469885, 3469884, 3469883, 3469882, 3469881, 3469879, 3469878, 3469877, 3469875, 3469874, 3469873, 3469872, 3469871, 3469870, 3469869, 3469868, 3469867, 3469866, 3469865, 3469864, 3469863, 3469862, 3469861, 3469860, 3469859, 3469858, 3469857, 3469856, 3469855, 3469854, 3469853, 3469852, 3469851, 3469850, 3469849, 3469848, 3469847, 3469846, 3469845, 3469844, 3469843, 3469842, 3469841, 3469840, 3469837, 3469836, 3469835, 3469834, 3469833, 3469832, 3469831, 3469830, 3469829, 3469828, 3469827, 3469826, 3469825, 3469822, 3469821, ], }, { "account": "jiajing", "xc_token": "5e051e0ff8384a81af53bea780deb28a", "task_ids": [ 3469551, 3469550, 3469549, 3469548, 3469547, 3469546, 3469545, 3469544, 3469543, 3469542, 3469541, 3469540, 3469539, 3469538, 3469537, 3469536, 3469535, 3469534, 3469533, 3469532, 3469531, 3469530, 3469529, 3469528, 3469526, 3469525, 3469524, 3469523, 3469522, 3469521, 3469520, 3469519, 3469518, 3469517, 3469516, 3469515, 3469514, 3469513, 3469512, 3469511, 3469510, 3469508, 3469507, 3469503, 3469501, 3469500, 3469499, 3469497, 3469495, 3469493, 3469492, 3469491, 3469490, 3469488, 3469486, 3469485, 3469484, 3469483, 3469482, 3469480, 3469477, 3469476, 3469475, 3469474, 3469472, 3469471, 3469470, 3469469, 3469468, 3469467, 3469466, 3469465, 3469464, 3469463, 3469462, 3469461, 3469460, 3469459, 3469458, 3469457, 3469456, 3469455, 3469454, 3469453, 3469452, 3469451, 3469450, 3469449, ], }, { "account": "i-zhouyuanxi@4paradigm.com", "xc_token": "62b9b487eff2488fb9f1da0b963f0b93", "task_ids": [ 3450582, 3450581, 3450580, 3450579, 3450575, 3450573, 3450572, 3450570, 3450569, 3450568, 3450567, 3450565, 3450560, 3450557, 3450556, 3450555, 3450554, 3450553, 3450552, 3450551, 3450550, 3450549, 3450548, 3450544, 3450541, 3450538, 3450537, 3450536, 3450535, 3450534, 3450533, 3450532, 3450531, 3450530, 3450529, 3450528, 3450527, 3450526, 3450525, 3450524, 3450523, 3450522, 3450521, 3450520, 3450519, 3450518, 3450517, 3450514, 3450513, 3450512, 3450511, 3450510, 3450509, 3450508, 3450507, 3450506, 3450505, 3450503, 3450501, 3450498, 3450496, 3450493, 3450483, 3450481, 3450480, 3450479, 3450478, 3450477, 3450476, 3450475, 3450474, 3450473, 3450472, 3450471, 3450470, 3450469, 3450463, 3450458, 3450455, 3450451, 3450441, 3450440, 3450439, 3450438, 3450437, 3450436, 3450435, 3450434, 3450433, 3450432, 3450431, 3450430, 3450429, 3450428, ], }, ] _shutdown = threading.Event() _state: dict[str, Any] = { "strategy_id": STRATEGY_ID, "accounts": { target["account"]: { "total": len(target["task_ids"]), "stopped": 0, "failed": 0, "failed_task_ids": [], } for target in TARGETS }, "phase": "starting", # starting | stopping | done | partial_failure | error "total": sum(len(target["task_ids"]) for target in TARGETS), "stopped": 0, "failed": 0, "failed_task_ids": [], "started_at": None, "finished_at": None, "error": None, } def _now() -> str: return datetime.now(timezone.utc).isoformat() 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(account: str, xc_token: str, 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] {account}: stopped task IDs: {task_ids}", flush=True) return True print( f"[stop] {account}: attempt {attempt}/{MAX_RETRIES} failed for {task_ids}: " f"HTTP {response.status_code}, {result}", flush=True, ) except requests.RequestException as exc: print( f"[stop] {account}: attempt {attempt}/{MAX_RETRIES} " f"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: for target in TARGETS: account = target["account"] account_state = _state["accounts"][account] task_ids = target["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(account, target["xc_token"], batch): _state["stopped"] += len(batch) account_state["stopped"] += len(batch) else: _state["failed"] += len(batch) _state["failed_task_ids"].extend(batch) account_state["failed"] += len(batch) account_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()