Discover current zhoukaile validation tasks at runtime

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):