chaos
6f7c949681
Overwrite previous grok-keygen contents with the local registration/CPA protocol codebase (secrets and runtime data excluded).
131 lines
3.9 KiB
Python
131 lines
3.9 KiB
Python
"""Probe free Grok 4.5 via cli-chat-proxy with a CPA access_token."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import ssl
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
from .proxyutil import resolve_proxy
|
|
from .schema import DEFAULT_BASE_URL, DEFAULT_CLIENT_HEADERS
|
|
|
|
|
|
def _ssl_context() -> ssl.SSLContext | None:
|
|
try:
|
|
import certifi # type: ignore
|
|
|
|
return ssl.create_default_context(cafile=certifi.where())
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _opener(proxy: str | None = None) -> urllib.request.OpenerDirector:
|
|
p = resolve_proxy(proxy)
|
|
handlers: list[Any] = []
|
|
ctx = _ssl_context()
|
|
if ctx is not None:
|
|
handlers.append(urllib.request.HTTPSHandler(context=ctx))
|
|
if p:
|
|
handlers.append(urllib.request.ProxyHandler({"http": p, "https": p}))
|
|
return urllib.request.build_opener(*handlers) if handlers else urllib.request.build_opener()
|
|
|
|
|
|
def probe_models(
|
|
access_token: str,
|
|
*,
|
|
base_url: str = DEFAULT_BASE_URL,
|
|
timeout: float = 30.0,
|
|
proxy: str | None = None,
|
|
) -> dict[str, Any]:
|
|
base = base_url.rstrip("/")
|
|
url = f"{base}/models"
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Accept": "application/json",
|
|
**DEFAULT_CLIENT_HEADERS,
|
|
}
|
|
opener = _opener(proxy)
|
|
req = urllib.request.Request(url, headers=headers, method="GET")
|
|
try:
|
|
with opener.open(req, timeout=timeout) as resp:
|
|
body = json.loads(resp.read().decode("utf-8"))
|
|
ids = [x.get("id") for x in body.get("data") or [] if isinstance(x, dict)]
|
|
return {
|
|
"ok": True,
|
|
"status": getattr(resp, "status", 200),
|
|
"model_ids": ids,
|
|
"has_grok_45": any(i == "grok-4.5" for i in ids),
|
|
}
|
|
except urllib.error.HTTPError as e:
|
|
return {
|
|
"ok": False,
|
|
"status": e.code,
|
|
"error": e.read().decode("utf-8", errors="replace")[:500],
|
|
"model_ids": [],
|
|
"has_grok_45": False,
|
|
}
|
|
except Exception as e: # noqa: BLE001
|
|
return {
|
|
"ok": False,
|
|
"status": 0,
|
|
"error": str(e),
|
|
"model_ids": [],
|
|
"has_grok_45": False,
|
|
}
|
|
|
|
|
|
def probe_mini_response(
|
|
access_token: str,
|
|
*,
|
|
base_url: str = DEFAULT_BASE_URL,
|
|
timeout: float = 60.0,
|
|
proxy: str | None = None,
|
|
) -> dict[str, Any]:
|
|
base = base_url.rstrip("/")
|
|
url = f"{base}/responses"
|
|
payload = {
|
|
"model": "grok-4.5",
|
|
"stream": False,
|
|
"input": "Reply with exactly MINT_OK",
|
|
"reasoning": {"effort": "low"},
|
|
}
|
|
headers = {
|
|
"Authorization": f"Bearer {access_token}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
**DEFAULT_CLIENT_HEADERS,
|
|
}
|
|
opener = _opener(proxy)
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(payload).encode("utf-8"),
|
|
headers=headers,
|
|
method="POST",
|
|
)
|
|
try:
|
|
with opener.open(req, timeout=timeout) as resp:
|
|
body = json.loads(resp.read().decode("utf-8"))
|
|
texts: list[str] = []
|
|
for item in body.get("output") or []:
|
|
if item.get("type") == "message":
|
|
for c in item.get("content") or []:
|
|
if c.get("type") == "output_text":
|
|
texts.append(c.get("text") or "")
|
|
return {
|
|
"ok": True,
|
|
"status": getattr(resp, "status", 200),
|
|
"model": body.get("model"),
|
|
"text": "\n".join(texts),
|
|
"usage": body.get("usage"),
|
|
}
|
|
except urllib.error.HTTPError as e:
|
|
return {
|
|
"ok": False,
|
|
"status": e.code,
|
|
"error": e.read().decode("utf-8", errors="replace")[:800],
|
|
}
|
|
except Exception as e: # noqa: BLE001
|
|
return {"ok": False, "status": 0, "error": str(e)}
|