#!/usr/bin/env python3 """Show existing Codex reset credits. Never consume a credit or change settings.""" import datetime as dt import json import os import select import subprocess import sys import time from pathlib import Path CLI = Path('/Applications/ChatGPT.app/Contents/Resources/codex-cli/bin/codex') JST = dt.timezone(dt.timedelta(hours=9), 'JST') READ_METHODS = frozenset(('initialize', 'account/read', 'account/rateLimits/read')) class ReadError(Exception): pass class CodexReader: def __enter__(self): if not CLI.is_file(): raise ReadError(f'Mac付属のCodex CLIが見つかりません: {CLI}') self.process = subprocess.Popen( [str(CLI), 'app-server', '--stdio'], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, ) self.buffer = bytearray() self.request_id = 0 return self def send(self, message): try: self.process.stdin.write((json.dumps(message) + '\n').encode()) self.process.stdin.flush() except (BrokenPipeError, OSError): raise ReadError('Codexとの接続が終了しました。') from None def read(self, method, params=None): if method not in READ_METHODS: raise ReadError('このコマンドは読み取り専用です。') self.request_id += 1 request_id = self.request_id message = {'method': method, 'id': request_id} if params is not None: message['params'] = params self.send(message) deadline = time.monotonic() + 30 while True: while b'\n' in self.buffer: line, remainder = self.buffer.split(b'\n', 1) self.buffer[:] = remainder try: reply = json.loads(line) except (json.JSONDecodeError, UnicodeDecodeError): raise ReadError('Codexからの応答を読み取れませんでした。') from None if reply.get('id') != request_id or 'method' in reply: continue if 'error' in reply: code = reply['error'].get('code', '不明') raise ReadError(f'{method} の読み取りに失敗しました（コード: {code}）。') result = reply.get('result') if not isinstance(result, dict): raise ReadError(f'{method} の応答形式を確認できませんでした。') return result remaining = deadline - time.monotonic() if remaining <= 0 or not select.select([self.process.stdout], [], [], remaining)[0]: raise ReadError(f'{method} が30秒以内に応答しませんでした。') chunk = os.read(self.process.stdout.fileno(), 65536) if not chunk: raise ReadError('Codexとの接続が終了しました。') self.buffer.extend(chunk) def __exit__(self, *_): try: self.process.stdin.close() except (BrokenPipeError, OSError): pass try: self.process.wait(timeout=3) except subprocess.TimeoutExpired: self.process.terminate() try: self.process.wait(timeout=2) except subprocess.TimeoutExpired: self.process.kill() self.process.wait() self.process.stdout.close() def date_text(value, timezone=JST): if value is None: return '未提供' return dt.datetime.fromtimestamp(value, timezone).isoformat(sep=' ', timespec='seconds') def time_left(value, now): if value is None: return '期限情報なし' seconds = value - now.timestamp() if seconds <= 0: return '期限を過ぎています' minutes = int(seconds // 60) if minutes == 0: return 'あと1分未満' days, minutes = divmod(minutes, 1440) hours, minutes = divmod(minutes, 60) return 'あと' + ''.join( f'{amount}{unit}' for amount, unit in ((days, '日'), (hours, '時間'), (minutes, '分')) if amount ) def show(account, result): now = dt.datetime.now(JST) print('Codex リセット権の確認（読み取り専用）') print(f"アカウント: {account.get('email') or 'メール未提供'} / {account.get('planType') or 'プラン未提供'}") print(f'確認日時: {now.isoformat(sep=" ", timespec="seconds")} JST') resets = result.get('rateLimitResetCredits') count = resets.get('availableCount') if isinstance(resets, dict) else None print(f'\n使用可能なリセット権: {str(count) + "件" if count is not None else "未取得"}') rows = resets.get('credits') if isinstance(resets, dict) else None if rows is None: print('個別の権利情報: 未提供（残数ゼロとは限りません）') else: print(f'取得した明細: {len(rows)}件 / 表示は期限が近い順') for index, credit in enumerate(sorted( rows, key=lambda c: (c.get('expiresAt') is None, c.get('expiresAt') or 0) ), 1): expiry = credit.get('expiresAt') status = credit.get('status') status_text = '未使用（available）' if status == 'available' else status or '未提供' print(f'\n{index}. {credit.get("title") or "リセット権"} / {status_text}') print(f' 期限 JST: {date_text(expiry)} / {time_left(expiry, now)}') print(f' 期限 UTC: {date_text(expiry, dt.timezone.utc)}') print(f' expiresAt: {expiry if expiry is not None else "未提供"}') print(f' creditId: {credit.get("id") or "未提供"}') print(f' 対象: {credit.get("resetType") or "未提供"}') if count is not None and count != len(rows): print('\n総数は「使用可能なリセット権」の値を参照してください。明細は一部の場合があります。') buckets = result.get('rateLimitsByLimitId') if not isinstance(buckets, dict) or not buckets: bucket = result.get('rateLimits') buckets = {bucket.get('limitId') or 'codex': bucket} if isinstance(bucket, dict) else {} print('\n現在の利用枠:') shown = False for key, bucket in buckets.items(): for name in ('primary', 'secondary'): window = (bucket or {}).get(name) if not isinstance(window, dict): continue shown = True used = window.get('usedPercent') remaining = f'{max(0, min(100, 100-used)):g}%' if isinstance(used, (int, float)) else '未提供' duration = window.get('windowDurationMins') label = {300: '5時間', 10080: '週間'}.get(duration, f'{duration}分' if duration is not None else '期間未提供') print(f' {key} / {label}（{name}）: 残り{remaining}') print(f' 通常の利用枠更新: {date_text(window.get("resetsAt"))} JST') if not shown: print(' 未提供') print('\nこのコマンドは権利を消費しません。ID省略時の自動選択順は未確認です。') def main(): if len(sys.argv) > 1: if sys.argv[1:] in (['-h'], ['--help']): print('使い方: codex-resets\n既存のCodexアカウントから残数・期限・利用枠を取得し、日本時間で表示します。\nリセットの消費、購入、ログイン、設定変更は行いません。') return 0 print('使い方: codex-resets（引数は不要です）', file=sys.stderr) return 2 try: with CodexReader() as reader: reader.read('initialize', {'clientInfo': { 'name': 'codex_reset_readonly_check', 'version': '1.1.0', }}) reader.send({'method': 'initialized', 'params': {}}) account = reader.read('account/read', {'refreshToken': False}).get('account') if not isinstance(account, dict): raise ReadError('既存のCodexログインがありません。ログイン操作は行いません。') result = reader.read('account/rateLimits/read') show(account, result) return 0 except (ReadError, OSError) as error: print(f'確認できませんでした: {error}', file=sys.stderr) return 1 except KeyboardInterrupt: print('\n確認を中止しました。', file=sys.stderr) return 130 if __name__ == '__main__': sys.exit(main())