74 lines
2.3 KiB
Python
74 lines
2.3 KiB
Python
"""透過 codex app-server 的 JSON-RPC account/read 取得目前登入帳號的 email。
|
||
|
||
這是 TUI `/status` Account 欄位的程式化來源，不解析本地 OAuth token，
|
||
而是由 codex 自身回報登入帳號。將 email 印到 stdout（取不到時印空字串）。
|
||
"""
|
||
|
||
import json
|
||
import subprocess
|
||
import time
|
||
|
||
|
||
def read_account_email(timeout_seconds: float = 25.0) -> str:
|
||
"""透過 codex app-server 的 JSON-RPC account/read 取得目前登入帳號的 email。
|
||
|
||
啟動 ``codex app-server`` 子行程，依序送出 initialize / initialized /
|
||
account/read 三筆 JSON-RPC 訊息，並讀取其回報的登入帳號 email。
|
||
|
||
Args:
|
||
timeout_seconds: 等待 app-server 回應的秒數上限，預設 25.0。
|
||
|
||
Returns:
|
||
登入帳號的 email；取不到或逾時時回傳空字串 ""。
|
||
|
||
使用情境：
|
||
在 CI 中驗證登入身分時呼叫。前置條件為 codex CLI 已安裝，
|
||
且 $HOME/.codex/auth.json 已寫入有效的 OAuth token。
|
||
"""
|
||
process = subprocess.Popen(
|
||
["codex", "app-server"],
|
||
stdin=subprocess.PIPE,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.DEVNULL,
|
||
text=True,
|
||
bufsize=1,
|
||
)
|
||
|
||
def send(obj):
|
||
"""將 dict 物件序列化成 JSON 後寫入 app-server 的 stdin 並 flush。"""
|
||
process.stdin.write(json.dumps(obj) + "\n")
|
||
process.stdin.flush()
|
||
|
||
send({
|
||
"jsonrpc": "2.0",
|
||
"id": 1,
|
||
"method": "initialize",
|
||
"params": {"clientInfo": {"name": "ci", "version": "1.0"}},
|
||
})
|
||
send({"jsonrpc": "2.0", "method": "initialized", "params": {}})
|
||
send({"jsonrpc": "2.0", "id": 2, "method": "account/read", "params": {}})
|
||
|
||
email = ""
|
||
deadline = time.time() + timeout_seconds
|
||
try:
|
||
while time.time() < deadline:
|
||
line = process.stdout.readline()
|
||
if not line:
|
||
break
|
||
try:
|
||
message = json.loads(line)
|
||
except ValueError:
|
||
continue
|
||
if message.get("id") == 2:
|
||
account = (message.get("result") or {}).get("account") or {}
|
||
email = account.get("email") or ""
|
||
break
|
||
finally:
|
||
process.terminate()
|
||
|
||
return email
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print(read_account_email())
|