72 lines
3.2 KiB
Python
72 lines
3.2 KiB
Python
"""Run after interactive Grok login; enable the provider only after a real reply."""
|
|
import json
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
|
|
import requests
|
|
|
|
|
|
def activate():
|
|
root = Path(__file__).resolve().parent
|
|
config = json.loads((root / "grok-client.json").read_text(encoding="utf-8"))
|
|
session = requests.Session()
|
|
session.trust_env = False
|
|
endpoint = config["public_url"]
|
|
try:
|
|
session.get(endpoint + "/health", timeout=3).raise_for_status()
|
|
except requests.RequestException:
|
|
pythonw = str(Path(sys.executable).with_name("pythonw.exe"))
|
|
subprocess.Popen([pythonw, str(root / "launch_local.py")], cwd=root,
|
|
creationflags=0x08000000 | 0x00000008)
|
|
for _ in range(20):
|
|
try:
|
|
session.get(endpoint + "/health", timeout=1).raise_for_status()
|
|
break
|
|
except requests.RequestException:
|
|
time.sleep(.5)
|
|
else:
|
|
raise RuntimeError("本地 Grok 兼容服务未能启动，请查看 grok-bridge.log")
|
|
print("正在验证 Grok 登录状态和文本回复……")
|
|
response = session.post(endpoint + "/v1/chat/completions",
|
|
headers={"Authorization": "Bearer " + config["token"]},
|
|
json={"model": "grok-local", "messages": [{"role": "user", "content": "Reply with OK only."}]}, timeout=260)
|
|
if not response.ok or not response.json().get("choices", [{}])[0].get("message", {}).get("content"):
|
|
raise RuntimeError("Grok 仍无法生成回复，请完成 Grok CLI 登录后再试；火宝中的 Grok 保持停用。")
|
|
huobao = Path.home() / "AppData/Roaming/HuobaoDrama"
|
|
def app_url():
|
|
ports = re.findall(r"port=(\d+) userData=", (huobao / "logs/startup.log").read_text(encoding="utf-8"))
|
|
return "http://127.0.0.1:" + ports[-1] + "/api/v1"
|
|
try:
|
|
base = app_url()
|
|
session.get(base + "/health", timeout=3).raise_for_status()
|
|
except (requests.RequestException, OSError, IndexError):
|
|
executable = root.parent / "HuobaoDrama/HuobaoDrama.exe"
|
|
subprocess.Popen([str(executable)], cwd=executable.parent)
|
|
for _ in range(30):
|
|
time.sleep(1)
|
|
try:
|
|
base = app_url()
|
|
session.get(base + "/health", timeout=1).raise_for_status()
|
|
break
|
|
except (requests.RequestException, OSError, IndexError):
|
|
pass
|
|
else:
|
|
raise RuntimeError("Grok 已登录，但火宝未启动，请打开火宝后重新运行此脚本")
|
|
rows = session.get(base + "/ai-configs", timeout=10).json()["data"]
|
|
row = next((r for r in rows if "grok-local" in r.get("model", []) and r["base_url"].startswith(endpoint)), None)
|
|
if not row:
|
|
raise RuntimeError("未找到已准备的 Grok 配置")
|
|
session.put(base + "/ai-configs/" + str(row["id"]), json={"name": "Grok（本机）", "is_active": True}, timeout=10).raise_for_status()
|
|
print("Grok 已通过实测并在火宝中启用。DeepSeek 仍是默认模型。")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
activate()
|
|
except Exception as error:
|
|
print(str(error))
|
|
sys.exit(1)
|