158 lines
5.1 KiB
Python
158 lines
5.1 KiB
Python
#!/usr/bin/env python3
|
||
"""修复 Cockpit Tools 无法启动 Codex：路径、当前账号、auth 同步。"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
COCKPIT = Path.home() / ".antigravity_cockpit"
|
||
CODEX_HOME = Path.home() / ".codex"
|
||
CODEX_APP = Path("/Applications/Codex.app")
|
||
HEALTHY_OAUTH_ID = "codex_4d0c0e374407d42d21202d01cd05eaf3" # sled · 无 401
|
||
APIKEY_ID = "codex_apikey_5df7aea89c1abb33b2043a4d3cad9071" # 龙虾U盘AI
|
||
USE_APIKEY = "--apikey" in sys.argv
|
||
|
||
|
||
def ts() -> str:
|
||
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
||
|
||
def backup(path: Path) -> None:
|
||
if path.exists():
|
||
shutil.copy2(path, f"{path}.bak.{datetime.now().strftime('%Y%m%d%H%M%S')}")
|
||
|
||
|
||
def load_json(path: Path) -> dict:
|
||
return json.loads(path.read_text(encoding="utf-8"))
|
||
|
||
|
||
def save_json(path: Path, data: dict) -> None:
|
||
backup(path)
|
||
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def write_auth_from_account(account: dict) -> None:
|
||
tokens = account.get("tokens") or {}
|
||
if not tokens.get("access_token"):
|
||
raise SystemExit(f"账号 {account.get('id')} 无 access_token")
|
||
auth = {
|
||
"OPENAI_API_KEY": account.get("openai_api_key"),
|
||
"last_refresh": ts(),
|
||
"tokens": {
|
||
"access_token": tokens.get("access_token", ""),
|
||
"account_id": account.get("account_id"),
|
||
"id_token": tokens.get("id_token", ""),
|
||
"refresh_token": tokens.get("refresh_token", ""),
|
||
},
|
||
}
|
||
auth_path = CODEX_HOME / "auth.json"
|
||
backup(auth_path)
|
||
auth_path.write_text(json.dumps(auth, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def fix_cockpit_path() -> str:
|
||
cfg_path = COCKPIT / "config.json"
|
||
cfg = load_json(cfg_path)
|
||
old = cfg.get("codex_app_path", "")
|
||
cfg["codex_app_path"] = str(CODEX_APP)
|
||
save_json(cfg_path, cfg)
|
||
return old
|
||
|
||
|
||
def set_current_account(target_id: str) -> None:
|
||
idx_path = COCKPIT / "codex_accounts.json"
|
||
idx = load_json(idx_path)
|
||
idx["current_account_id"] = target_id
|
||
save_json(idx_path, idx)
|
||
|
||
inst_path = COCKPIT / "codex_instances.json"
|
||
inst = load_json(inst_path)
|
||
ds = inst.setdefault("defaultSettings", {})
|
||
ds["bindAccountId"] = target_id
|
||
save_json(inst_path, inst)
|
||
|
||
|
||
def sanitize_config_toml(text: str) -> str:
|
||
"""去掉 Codex 无法解析的孤立 ensure-hub 伪表头。"""
|
||
out = []
|
||
for line in text.splitlines():
|
||
if line.strip().startswith('["/Users/karuo/.persistent-chat-local/ensure-hub.sh"'):
|
||
continue
|
||
out.append(line)
|
||
while out and not out[-1].strip():
|
||
out.pop()
|
||
return "\n".join(out) + "\n"
|
||
|
||
|
||
def apply_api_provider(account: dict) -> None:
|
||
"""为龙虾U盘AI 写入 provider_secrets；config.toml 仅做 sanitize，避免破坏 TOML。"""
|
||
provider_id = "longxia_udisk_ai"
|
||
api_key = account.get("openai_api_key") or ""
|
||
base_url = (account.get("api_base_url") or "").rstrip("/")
|
||
if not api_key or not base_url:
|
||
raise SystemExit("API Key 账号缺少 openai_api_key 或 api_base_url")
|
||
|
||
secrets = CODEX_HOME / "provider_secrets.toml"
|
||
lines = secrets.read_text(encoding="utf-8").splitlines() if secrets.exists() else []
|
||
out, skip = [], False
|
||
for line in lines:
|
||
if line.strip() == f"[{provider_id}]":
|
||
skip = True
|
||
continue
|
||
if skip and line.startswith("["):
|
||
skip = False
|
||
if skip:
|
||
continue
|
||
out.append(line)
|
||
while out and not out[-1].strip():
|
||
out.pop()
|
||
out.extend([f"[{provider_id}]", f'api_key = "{api_key}"', ""])
|
||
backup(secrets)
|
||
secrets.write_text("\n".join(out) + "\n", encoding="utf-8")
|
||
|
||
cfg = CODEX_HOME / "config.toml"
|
||
if cfg.exists():
|
||
backup(cfg)
|
||
cfg.write_text(sanitize_config_toml(cfg.read_text(encoding="utf-8")), encoding="utf-8")
|
||
print("[WARN] API Key 线路请用 Cockpit 切换；勿手改 config.toml 追加 provider 块")
|
||
|
||
|
||
def restart_codex() -> None:
|
||
subprocess.run(["killall", "Codex"], check=False)
|
||
time.sleep(1.5)
|
||
subprocess.run(["/usr/bin/open", "-a", "Codex"], check=True)
|
||
|
||
|
||
def main() -> None:
|
||
if not CODEX_APP.exists():
|
||
raise SystemExit(f"未找到 {CODEX_APP}")
|
||
|
||
old_path = fix_cockpit_path()
|
||
print(f"[OK] codex_app_path: {old_path!r} -> {CODEX_APP}")
|
||
|
||
api_acct = load_json(COCKPIT / "codex_accounts" / f"{APIKEY_ID}.json")
|
||
oauth_acct = load_json(COCKPIT / "codex_accounts" / f"{HEALTHY_OAUTH_ID}.json")
|
||
|
||
if USE_APIKEY:
|
||
set_current_account(APIKEY_ID)
|
||
write_auth_from_account(oauth_acct)
|
||
apply_api_provider(api_acct)
|
||
print(f"[OK] 当前账号=龙虾U盘AI，OAuth 壳={oauth_acct.get('email')}")
|
||
else:
|
||
set_current_account(HEALTHY_OAUTH_ID)
|
||
write_auth_from_account(oauth_acct)
|
||
print(f"[OK] 当前账号={oauth_acct.get('email')}")
|
||
|
||
restart_codex()
|
||
print("[OK] 已重启 /Applications/Codex.app")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|