openclaw/scripts/codex-limit-monitor.sh

#!/usr/bin/env bash
set -euo pipefail

SKILL_DIR="/home/openclaw/.openclaw/workspace/skills/codex-account-switcher-skill"
cd "$SKILL_DIR"

before_json="$(./codex-accounts.py list --json 2>/dev/null || echo '{}')"
auto_json="$(./codex-accounts.py auto --json 2>/dev/null || echo '{}')"
after_json="$(./codex-accounts.py list --json 2>/dev/null || echo '{}')"

python3 - "$before_json" "$auto_json" "$after_json" <<'PY'
import json
import subprocess
import sys
from datetime import datetime, timezone, timedelta

before_s, auto_s, after_s = sys.argv[1:4]


def j(s):
    try:
        return json.loads(s)
    except Exception:
        return {}


before = j(before_s)
auto = j(auto_s)
after = j(after_s)

MSK = timezone(timedelta(hours=3))
now = datetime.now(MSK).strftime('%Y-%m-%d %H:%M:%S MSK')

accounts = [a.get('name') for a in after.get('accounts', []) if a.get('name') and not str(a.get('name')).startswith('.')]
old_name = before.get('active')
new_name = auto.get('switched_to') or after.get('active')
all_acc = auto.get('all_accounts', {}) if isinstance(auto.get('all_accounts', {}), dict) else {}


def account_ttl(data, name):
    for acc in data.get('accounts', []) if isinstance(data.get('accounts'), list) else []:
        if acc.get('name') == name:
            ttl = acc.get('token_ttl_seconds')
            if isinstance(ttl, (int, float)):
                return int(ttl)
    return None


def fmt_ttl(seconds):
    if seconds is None:
        return 'ttl н/д'
    if seconds <= 0:
        return 'ttl expired'
    m = seconds // 60
    if m < 60:
        return f'ttl {m}m'
    h = m // 60
    rm = m % 60
    return f'ttl {h}h{rm:02d}m'


def auth_probe(name):
    try:
        subprocess.run(['./codex-accounts.py', 'use', name], check=False, capture_output=True, text=True, timeout=15)
        p = subprocess.run(['codex', 'exec', 'PING', '--skip-git-repo-check', '--json'], check=False, capture_output=True, text=True, timeout=60)
        return 'ok' if p.returncode == 0 else 'fail'
    except Exception:
        return 'fail'


probe = {}
for n in accounts:
    probe[n] = auth_probe(n)

# Restore active account after probes
restore = new_name or old_name
if restore:
    subprocess.run(['./codex-accounts.py', 'use', restore], check=False, capture_output=True, text=True, timeout=15)


def fmt_acc(name):
    d = all_acc.get(name, {}) if name else {}
    auth = probe.get(name, 'н/д') if name else 'н/д'

    if d and 'error' not in d:
        w = d.get('weekly_used')
        a = d.get('available')
        dd = d.get('daily_used')
        if w is not None and a is not None:
            return f"auth={auth}, week used={w:.0f}%, week left={a:.0f}%, day used={dd:.0f}%"

    ttl_after = account_ttl(after, name)
    ttl_before = account_ttl(before, name)
    ttl = ttl_after if ttl_after is not None else ttl_before
    if d and 'error' in d:
        return f"auth={auth}, лимит: {d['error']}, {fmt_ttl(ttl)}"
    return f"auth={auth}, {fmt_ttl(ttl)}"


if auto.get('error'):
    status = f"quota-check failed: {auto.get('error')}"
else:
    status = 'ok'

switch_note = 'без переключения'
if old_name and new_name and old_name != new_name:
    switch_note = f"переключение: {old_name} -> {new_name}"
elif new_name:
    switch_note = f"активный аккаунт: {new_name}"

accounts_summary = ', '.join([f"{n}:{probe.get(n, 'н/д')}" for n in accounts]) if accounts else 'н/д'

def reset_eta(epoch):
    try:
        if not epoch:
            return 'н/д'
        dt = datetime.fromtimestamp(int(epoch), tz=MSK)
        return dt.strftime('%Y-%m-%d %H:%M MSK')
    except Exception:
        return 'н/д'

lines = []
lines.append(f"📊 Codex quota report — {now}")
lines.append(f"Статус: {status}")
lines.append(f"Маршрут: {switch_note}")
lines.append("")
lines.append("Аккаунты:")

for name in accounts:
    d = all_acc.get(name, {})
    auth = probe.get(name, 'н/д')
    ttl = account_ttl(after, name)
    ttl_text = fmt_ttl(ttl)
    if d and 'error' not in d and d.get('weekly_used') is not None:
        lines.append(
            f"- {name}: auth={auth}, за неделю использовано {d['weekly_used']:.0f}% / осталось {d['available']:.0f}%, "
            f"за день использовано {d.get('daily_used', 0):.0f}%, сброс лимита {reset_eta(d.get('weekly_resets_at'))}, {ttl_text}"
        )
    elif d and 'error' in d:
        lines.append(f"- {name}: auth={auth}, лимит: {d['error']}, {ttl_text}")
    else:
        lines.append(f"- {name}: auth={auth}, {ttl_text}")

if new_name:
    lines.append("")
    lines.append(f"✅ Рекомендуемый/активный: {new_name}")

print("\n".join(lines))
PY