脚本内容,可以看下,我是GLM5.3给我写的。
#!/usr/bin/env python3
# codexsa — codex auth.json 配置切换器(交互菜单,零依赖,Python 3 标准库)
#
# 用法:
# codexsa 交互菜单
# codexsa --selftest 内置自检(使用临时目录与假 codex,不触碰真实 ~/.codex)
#
# 存储布局(权限 600):
# $CODEX_HOME/auth-switcher/profiles/<配置名>.json 每个配置 = auth.json 原样字节拷贝
# $CODEX_HOME/auth-switcher/current 一行文本:当前配置名
# CODEX_HOME 默认 ~/.codex,与 codex CLI 的同名环境变量语义一致。
import json
import os
import subprocess
import sys
CODEX_HOME = os.environ.get("CODEX_HOME") or os.path.expanduser("~/.codex")
AUTH_JSON = os.path.join(CODEX_HOME, "auth.json")
STORE_DIR = os.path.join(CODEX_HOME, "auth-switcher")
PROFILES_DIR = os.path.join(STORE_DIR, "profiles")
CURRENT_FILE = os.path.join(STORE_DIR, "current")
NAME_CHARS = set(
"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_"
)
STOP_TIMEOUT = 30 # daemon stop 超时(秒),超时按失败处理
# ── 纯函数 / 底层工具 ────────────────────────────────────────────
def valid_name(name: str) -> bool:
return bool(name) and all(c in NAME_CHARS for c in name)
def is_stop_success(status: str) -> bool:
"""stopped / notRunning / not_running 等写法均视为可继续切换。"""
s = status.strip().lower().replace("_", "").replace("-", "")
return s in {"stopped", "notrunning"}
def atomic_write(path: str, data: bytes) -> None:
"""临时文件 + rename,权限 600;凭证文件不允许写一半被看到。"""
tmp = path + ".tmp"
with open(tmp, "wb") as f:
f.write(data)
f.flush()
os.fsync(f.fileno())
os.chmod(tmp, 0o600)
os.replace(tmp, path)
def ask(prompt: str = "") -> str | None:
"""input() 的安全包装:EOF / Ctrl-C 返回 None(视为取消)。"""
try:
return input(prompt)
except (EOFError, KeyboardInterrupt):
print()
return None
def wait() -> None:
ask("\n按回车返回菜单...")
# ── 存取 ─────────────────────────────────────────────────────────
def ensure_dirs() -> None:
os.makedirs(PROFILES_DIR, exist_ok=True)
def profile_path(name: str) -> str:
return os.path.join(PROFILES_DIR, name + ".json")
def list_profiles() -> list[str]:
try:
return sorted(
p[:-5] for p in os.listdir(PROFILES_DIR)
if p.endswith(".json") and os.path.isfile(
os.path.join(PROFILES_DIR, p))
)
except FileNotFoundError:
return []
def read_current() -> str | None:
try:
with open(CURRENT_FILE, "r", encoding="utf-8") as f:
v = f.read().strip()
return v or None
except FileNotFoundError:
return None
def write_current(name: str | None) -> None:
if name is None:
try:
os.remove(CURRENT_FILE)
except FileNotFoundError:
pass
return
atomic_write(CURRENT_FILE, (name + "\n").encode("utf-8"))
def stop_daemon() -> tuple[bool, str]:
"""执行 codex app-server daemon stop,返回 (是否成功, 展示信息)。"""
try:
r = subprocess.run(
["codex", "app-server", "daemon", "stop"],
capture_output=True, text=True, timeout=STOP_TIMEOUT,
)
except FileNotFoundError:
return False, "未找到 codex 命令,请确认已安装且在 PATH 中"
except subprocess.TimeoutExpired:
return False, f"命令超时({STOP_TIMEOUT} 秒),视为失败"
out = (r.stdout or "").strip()
err = (r.stderr or "").strip()
if r.returncode != 0:
return False, f"退出码 {r.returncode}\n{out or err or '(无输出)'}"
try:
obj = json.loads(out)
except json.JSONDecodeError:
return False, f"输出不是合法 JSON:\n{out or '(无输出)'}"
status = str(obj.get("status", ""))
if is_stop_success(status):
return True, out
return False, f"status = {status!r}(预期 stopped / notRunning):\n{out}"
# ── 交互组件 ─────────────────────────────────────────────────────
def choose_profile(title: str, current: str | None = None) -> str | None:
names = list_profiles()
if not names:
print("暂无已保存配置,请先用菜单「3 添加配置」创建。")
wait()
return None
print(title)
for i, n in enumerate(names, 1):
mark = " ★当前" if current is not None and n == current else ""
print(f" {i}) {n}{mark}")
print(" 0) 取消")
while True:
s = ask("请选择编号: ")
if s is None or s.strip() == "0":
return None
if s.strip().isdigit() and 1 <= int(s.strip()) <= len(names):
return names[int(s.strip()) - 1]
print("无效编号,请重新输入。")
def paste_json() -> str | None:
"""多行粘贴,单独一行 . 或 Ctrl-D 结束;返回原文(未校验)。"""
print("粘贴 JSON 内容(支持多行),完成后单独一行输入 . 回车,"
"或按 Ctrl-D 结束:")
lines = []
while True:
line = ask()
if line is None or line.strip() == ".":
break
lines.append(line)
if not lines:
return None
return "\n".join(lines)
def pretty_print_file(path: str) -> None:
with open(path, "rb") as f:
raw = f.read()
try:
obj = json.loads(raw.decode("utf-8"))
print(json.dumps(obj, indent=2, ensure_ascii=False))
except (UnicodeDecodeError, json.JSONDecodeError):
# 文件可能被手动改坏:原样展示,不做格式化
print(raw.decode("utf-8", errors="replace"), end="")
# ── 菜单动作 ─────────────────────────────────────────────────────
def action_add() -> None:
name = ask("配置名(仅字母/数字/-/_): ")
if name is None:
return
name = name.strip()
if not name:
return
if not valid_name(name):
print("✗ 配置名只能包含字母、数字、- 和 _。")
wait()
return
if os.path.exists(profile_path(name)):
print(f"✗ 配置 {name!r} 已存在(配置名唯一)。"
"如需更新请先删除再添加。")
wait()
return
while True:
text = paste_json()
if text is None:
print("已取消添加。")
wait()
return
try:
json.loads(text)
break
except json.JSONDecodeError as e:
print(f"✗ JSON 校验失败: {e}")
again = ask("重新粘贴? [Y/n]: ")
if again is None or again.strip().lower() == "n":
print("已取消添加。")
wait()
return
atomic_write(profile_path(name), text.encode("utf-8"))
print(f"✓ 已保存配置 {name!r}。")
wait()
def action_set_current() -> None:
name = choose_profile("选择要设为当前使用的配置:")
if name is None:
return
write_current(name)
print(f"✓ 当前配置已设为 {name!r}。")
wait()
def action_delete() -> None:
cur = read_current()
name = choose_profile("选择要删除的配置:", current=cur)
if name is None:
return
r = ask(f"确认删除配置 {name!r}? [y/N]: ")
if r is None or r.strip().lower() != "y":
print("已取消删除。")
wait()
return
os.remove(profile_path(name))
if cur == name:
write_current(None)
print(f"✓ 已删除 {name!r},并已清空当前配置记录。")
else:
print(f"✓ 已删除 {name!r}。")
wait()
def action_view() -> None:
print(" 1) 实际生效的 auth.json")
print(" 2) 已保存的配置(★ 为当前配置)")
print(" 0) 返回")
while True:
s = ask("请选择: ")
if s is None or s.strip() == "0":
return
if s.strip() == "1":
if os.path.isfile(AUTH_JSON):
print(f"── {AUTH_JSON} ──")
pretty_print_file(AUTH_JSON)
wait()
else:
print(f"✗ 未找到 {AUTH_JSON}")
wait()
return
if s.strip() == "2":
name = choose_profile("选择要查看的配置:", current=read_current())
if name:
print(f"── {profile_path(name)} ──")
pretty_print_file(profile_path(name))
wait()
return
print("无效选项。")
def action_switch() -> None:
# 1) 前置检查(只读、无副作用):不满足时不触碰 daemon
cur = read_current()
if not cur:
print("✗ 未设置当前配置,无法切换。请先用菜单「2 设置当前配置」。")
wait()
return
if not os.path.isfile(AUTH_JSON):
print(f"✗ 未找到 {AUTH_JSON},无法切换。")
wait()
return
if not os.path.isfile(profile_path(cur)):
print(f"✗ 当前配置 {cur!r} 的文件已不存在(可能被删除)。"
"请先用菜单「2」重新设置当前配置。")
wait()
return
# 2) 停止 daemon
ok, msg = stop_daemon()
print(("✓ " if ok else "✗ ") + "停止 daemon:")
print(" " + msg.replace("\n", "\n "))
if not ok:
print("✗ 已中止切换,未修改任何文件。")
wait()
return
# 3) 回写:当前 auth.json 原样覆盖到当前配置
with open(AUTH_JSON, "rb") as f:
live = f.read()
atomic_write(profile_path(cur), live)
print(f"· 已将当前 auth.json 回写到配置 {cur!r}。")
# 4) 选择目标并覆盖
target = choose_profile("选择要切换到的配置:", current=cur)
if target is None:
print("已取消切换。注意:daemon 已停止,当前配置已完成回写。")
wait()
return
with open(profile_path(target), "rb") as f:
data = f.read()
atomic_write(AUTH_JSON, data)
write_current(target)
print(f"✓ 已切换到配置 {target!r},auth.json 已更新。")
print("如需启动 daemon: codex app-server daemon start")
wait()
# ── 主循环 ───────────────────────────────────────────────────────
ACTIONS = {
"1": ("切换配置", action_switch),
"2": ("设置当前配置", action_set_current),
"3": ("添加配置", action_add),
"4": ("删除配置", action_delete),
"5": ("查看配置", action_view),
}
def main() -> None:
ensure_dirs()
while True:
cur = read_current()
print()
print("════════ codexsa · codex auth 配置切换器 ════════")
print(f"当前配置: {cur if cur else '未设置'}")
for key, (title, _) in ACTIONS.items():
print(f" {key}) {title}")
print(" q) 退出")
s = ask("请选择: ")
if s is None or s.strip().lower() in ("q", "quit", "exit"):
return
entry = ACTIONS.get(s.strip())
if entry is None:
print("无效选项。")
continue
entry[1]()
# ── 内置自检 ─────────────────────────────────────────────────────
def selftest() -> None:
import shutil
import tempfile
# 纯函数
assert is_stop_success("stopped")
assert is_stop_success("notRunning")
assert is_stop_success("not_running")
assert is_stop_success(" Not-Running ")
assert not is_stop_success("running")
assert not is_stop_success("")
assert valid_name("a-b_2") and valid_name("A9")
assert not valid_name("") and not valid_name("a b")
assert not valid_name("a/b") and not valid_name("配置")
# 原子写:内容保真 + 权限 600
tmp = tempfile.mkdtemp(prefix="codexsa-selftest-")
try:
p = os.path.join(tmp, "f.json")
atomic_write(p, b'{"x":1}')
with open(p, "rb") as f:
assert f.read() == b'{"x":1}'
assert (os.stat(p).st_mode & 0o777) == 0o600
finally:
shutil.rmtree(tmp)
# 端到端:假 CODEX_HOME + 假 codex,跑完整菜单流
tmp = tempfile.mkdtemp(prefix="codexsa-e2e-")
try:
home = os.path.join(tmp, "home")
fakebin = os.path.join(tmp, "bin")
os.makedirs(os.path.join(home, "auth-switcher", "profiles"))
os.makedirs(fakebin)
fake = os.path.join(fakebin, "codex")
with open(fake, "w", encoding="utf-8") as f:
f.write('#!/bin/sh\necho \'{"status":"stopped","backend":'
'"pid","socketPath":"/tmp/s.sock","cliVersion":"0"}\'\n')
os.chmod(fake, 0o755)
live_before = b'{"OPENAI_API_KEY":"live-key-1"}'
with open(os.path.join(home, "auth.json"), "wb") as f:
f.write(live_before)
env = dict(
os.environ, CODEX_HOME=home,
PATH=fakebin + os.pathsep + os.environ["PATH"],
PYTHONIOENCODING="utf-8",
)
# 添加 p1/p2 → 设当前 p1 → 查看 live → 取消切换 → 真切换到 p2
# → 删除 p2(current 指向项)→ 退出
stdin = (
"3\np1\n{\"A\":1}\n.\n\n"
"3\np2\n{\"B\":2}\n.\n\n"
"2\n1\n\n"
"5\n1\n\n"
"1\n0\n\n"
"1\n2\n\n"
"4\n2\ny\n\n"
"q\n"
)
r = subprocess.run(
[sys.executable, os.path.abspath(__file__)],
input=stdin, capture_output=True, text=True,
env=env, timeout=60,
)
out = r.stdout
assert r.returncode == 0, out + "\nSTDERR:\n" + r.stderr
assert "live-key-1" in out, "查看 live auth.json 未显示内容"
assert "已切换到配置 'p2'" in out, "切换未报告成功"
# 回写:切换前 live auth.json 已原样存回 p1
with open(os.path.join(home, "auth-switcher", "profiles",
"p1.json"), "rb") as f:
assert f.read() == live_before, "回写内容不符"
# 覆盖:auth.json 现在是 p2 的粘贴原文
with open(os.path.join(home, "auth.json"), "rb") as f:
assert f.read() == b'{"B":2}', "auth.json 覆盖失败"
# 删除 current 指向的配置 → 记录清空、文件删除
assert not os.path.exists(
os.path.join(home, "auth-switcher", "current")), "current 未清空"
assert not os.path.exists(
os.path.join(home, "auth-switcher", "profiles",
"p2.json")), "p2 未删除"
print("selftest: 全部通过 ✓")
finally:
shutil.rmtree(tmp)
if __name__ == "__main__":
if "--selftest" in sys.argv:
selftest()
else:
main()