From b17431a74f46d4140f30551dfb5a9eb706745bf6 Mon Sep 17 00:00:00 2001
From: huni <chenweize1202@gmail.com>
Date: Mon, 24 Aug 2026 11:12:07 +0800
Subject: [PATCH] add main.py

---
 main.py | 130 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 130 insertions(+)
 create mode 100644 main.py

diff --git a/main.py b/main.py
new file mode 100644
index 0000000..1104436
--- /dev/null
+++ b/main.py
@@ -0,0 +1,130 @@
+"""ModelHub XC 适配智能体 · 只读探针骨架 (probe-only)
+
+用途:
+  1) 验证平台「建仓 -> Kaniko 构建 -> 部署 -> /health -> 运行/停止」整条链路。
+  2) 零副作用观测平台注入的 STRATEGY_ID / EXTERNAL_SERVICE_TOKEN 是否为有效凭证
+     (启动时用注入 token 调一次只读 GET /api/adapt/task/page,看返回 code:
+      0 = 凭证有效; 40100 = 未登录/凭证无效)。
+
+严格约束:
+  - 只服务 GET /health 与 GET /,绝不提交任何验证任务 (不调 task/add, 不调 build-config)。
+  - token 只从环境读,绝不打印其值/落盘;探针只调只读 GET;只记录 http/code/计数,不打印返回体内容(防 PII)。
+  - 纯 stdlib 零依赖;正确处理 SIGTERM 优雅停机。
+"""
+import json
+import os
+import signal
+import threading
+import time
+import urllib.error
+import urllib.request
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+MAIN = os.getenv("MAIN_HOST", "https://modelhub.org.cn")
+TOKEN = os.getenv("EXTERNAL_SERVICE_TOKEN", "")     # 平台注入的服务 token
+STRATEGY_ID = os.getenv("STRATEGY_ID", "")          # 平台注入的自身策略 id
+PORT = int(os.getenv("PORT", "8080"))
+
+shutdown = threading.Event()
+PROBE = {"done": False, "results": []}              # 探针结果,暴露在 GET / (不含任何 token 值)
+
+
+def log(msg: str) -> None:
+    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)
+
+
+def _readonly_get(path: str, headers: dict, timeout: int = 20) -> dict:
+    """只读 GET,返回 {http, code, count};绝不返回/记录响应体内容。"""
+    req = urllib.request.Request(MAIN + path, method="GET", headers=headers)
+    try:
+        with urllib.request.urlopen(req, timeout=timeout) as r:
+            body = r.read().decode("utf-8", "ignore")
+            status = r.status
+    except urllib.error.HTTPError as e:
+        body = e.read().decode("utf-8", "ignore")
+        status = e.code
+    except Exception as e:  # noqa: BLE001
+        return {"http": -1, "err": str(e)[:120]}
+    code = count = None
+    try:
+        j = json.loads(body)
+        code = j.get("code")
+        d = j.get("data")
+        if isinstance(d, dict):
+            count = d.get("total")
+            if count is None and isinstance(d.get("records"), list):
+                count = len(d["records"])
+        elif isinstance(d, list):
+            count = len(d)
+    except Exception:  # noqa: BLE001
+        pass
+    return {"http": status, "code": code, "count": count}
+
+
+def probe() -> None:
+    """一次性只读探针:验证注入凭证有效性 + 记录 STRATEGY_ID。"""
+    log(f"probe start | strategy_id={STRATEGY_ID or 'MISSING'} | "
+        f"token_present={bool(TOKEN)} | token_len={len(TOKEN)}")
+    path = "/api/adapt/task/page?current=1&pageSize=1"
+    trials = [
+        ("Xc-Token", {"Xc-Token": TOKEN}),
+        ("Authorization-Bearer", {"Authorization": "Bearer " + TOKEN}),
+        ("Authorization-raw", {"Authorization": TOKEN}),
+    ]
+    for name, headers in trials:
+        res = {"skip": "no-token"} if not TOKEN else _readonly_get(path, headers)
+        PROBE["results"].append({"auth": name, **res})
+        log(f"probe {name} -> {json.dumps(res, ensure_ascii=False)}")
+    PROBE["done"] = True
+    log("probe done | 判读: code=0 => 该头凭证有效; code=40100 => 未登录/凭证无效")
+
+
+class Handler(BaseHTTPRequestHandler):
+    def _json(self, obj: dict, status: int = 200) -> None:
+        payload = json.dumps(obj, ensure_ascii=False).encode()
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def do_GET(self) -> None:
+        if self.path == "/health":
+            self._json({"status": "ok"})
+            return
+        if self.path == "/":
+            self._json({
+                "name": "huni-probe-agent",
+                "mode": "probe-only",
+                "strategy_id_present": bool(STRATEGY_ID),
+                "token_present": bool(TOKEN),
+                "probe": PROBE,
+            })
+            return
+        self._json({"error": "not found"}, 404)
+
+    def log_message(self, *_a) -> None:
+        pass
+
+
+def _handle_signal(signum: int, _frame) -> None:
+    shutdown.set()
+    log(f"received signal {signum}, shutting down")
+
+
+def main() -> None:
+    signal.signal(signal.SIGTERM, _handle_signal)
+    signal.signal(signal.SIGINT, _handle_signal)
+    threading.Thread(target=probe, daemon=True).start()
+
+    server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler)
+    server.timeout = 1
+    log(f"probe-only agent listening on 0.0.0.0:{PORT}")
+    while not shutdown.is_set():
+        server.handle_request()
+    server.server_close()
+    log("stopped")
+
+
+if __name__ == "__main__":
+    main()