refactor(codex): 將 app-server 取帳號邏輯抽到 app/codex_account.py
CI / Release Tag Version (pull_request) Successful in 4s
CI / Codex (pull_request) Successful in 21s

把原本內嵌在 action.yml heredoc 的 python 抽成獨立檔 app/codex_account.py,
action 改以 $GITHUB_ACTION_PATH/app/codex_account.py 呼叫,便於閱讀與維護。
This commit is contained in:
Jeffery
2026-06-29 12:52:01 +08:00
parent 6e771e2e52
commit 672900762e
2 changed files with 58 additions and 37 deletions
+1 -37
View File
@@ -52,42 +52,6 @@ runs:
- name: 取得登入帳號 - name: 取得登入帳號
id: account id: account
run: | run: |
cat > "$RUNNER_TEMP/codex_account.py" <<'PY' email="$(python3 "$GITHUB_ACTION_PATH/app/codex_account.py")"
import json, subprocess, time
p = subprocess.Popen(
["codex", "app-server"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, text=True, bufsize=1,
)
def send(obj):
p.stdin.write(json.dumps(obj) + "\n")
p.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() + 25
while time.time() < deadline:
line = p.stdout.readline()
if not line:
break
try:
msg = json.loads(line)
except Exception:
continue
if msg.get("id") == 2:
account = (msg.get("result") or {}).get("account") or {}
email = account.get("email") or ""
break
p.terminate()
print(email)
PY
email="$(python3 "$RUNNER_TEMP/codex_account.py")"
printf 'email=%s\n' "$email" >> "$GITHUB_OUTPUT" printf 'email=%s\n' "$email" >> "$GITHUB_OUTPUT"
shell: bash shell: bash
+57
View File
@@ -0,0 +1,57 @@
"""透過 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:
process = subprocess.Popen(
["codex", "app-server"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
bufsize=1,
)
def send(obj):
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())