From 75e85887b5eab3a53fe2fe45d07d3eed3e6315a5 Mon Sep 17 00:00:00 2001 From: y2keui Date: Wed, 10 Jun 2026 10:10:37 +0900 Subject: [PATCH] Add Telegram Codex usage command --- README.md | 3 + app/bot.py | 158 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 161 insertions(+) diff --git a/README.md b/README.md index fc2a7f6..998e975 100644 --- a/README.md +++ b/README.md @@ -14,6 +14,7 @@ only keeps connection-management commands. /repo /ask /work +/usage /new /status /settings @@ -51,6 +52,8 @@ explicitly. `/settings`, `/models`, `/schedules`, `/records`, and `/sessions` return inline buttons, so you can change speed/model/reasoning settings, delete schedules, manage personal records, and switch or archive sessions without copying ids. +`/usage` asks the local Codex app-server for account, plan, rate-limit, and +credit information when the installed Codex version exposes it. ## Runtime settings diff --git a/app/bot.py b/app/bot.py index 8368f30..fdb76c9 100644 --- a/app/bot.py +++ b/app/bot.py @@ -35,6 +35,7 @@ HELP_TEXT = """Telegram-Codex 연결 브리지 /ask <내용> - 내용을 Codex로 전달 /work <내용> - 파일 수정, 커밋, 푸시 같은 쓰기 작업 실행 /cancel - 현재 진행 중인 작업 취소 +/usage - Codex 사용량/제한 확인 /new - 이 채팅의 Codex 세션 초기화 /status - 브리지 상태 확인 /settings - 모델, 추론강도, 속도 설정 확인 @@ -130,6 +131,9 @@ class TelegramCodexBot: app.add_handler(CommandHandler("work", self.work)) app.add_handler(CommandHandler("write", self.work)) app.add_handler(CommandHandler("cancel", self.cancel)) + app.add_handler(CommandHandler("usage", self.usage)) + app.add_handler(CommandHandler("limits", self.usage)) + app.add_handler(CommandHandler("quota", self.usage)) app.add_handler(CommandHandler("new", self.new)) app.add_handler(CommandHandler("status", self.status)) app.add_handler(CommandHandler("settings", self.settings)) @@ -289,6 +293,24 @@ class TelegramCodexBot: await self._close_client_for_chat(chat_id) await self._reply(update, f"작업 취소를 요청했습니다.\n{self._format_job_status(chat_id)}") + async def usage(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: + if not await self._guard(update): + return + await self._reply_usage(update) + + async def _reply_usage(self, update: Update) -> None: + try: + account = await self.management_runner.call("account/read", {}) + except Exception as exc: + await self._reply( + update, + "Codex 사용량 정보를 가져오지 못했습니다.\n" + f"오류: {exc}\n\n" + "NAS의 Codex app-server가 구버전이거나 계정 정보 조회 표면을 제공하지 않을 수 있습니다.", + ) + return + await self._reply(update, _format_codex_usage(account)) + async def status(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await self._guard(update): return @@ -925,6 +947,9 @@ class TelegramCodexBot: text = text.strip() if not text: return + if _looks_like_usage_question(text): + await self._reply_usage(update) + return await self._run_codex_from_update(update, text) async def _run_codex_from_update( @@ -1432,6 +1457,139 @@ def _repo_name_tokens(prompt: str) -> list[str]: return _dedupe(tokens) +def _looks_like_usage_question(text: str) -> bool: + normalized = text.casefold() + usage_terms = ("사용량", "남은량", "남은 사용량", "한도", "쿼터", "quota", "limit", "usage", "credits", "크레딧") + if not any(term in normalized for term in usage_terms): + return False + + explicit_codex_terms = ("codex", "코덱스", "openai", "gpt", "모델") + if any(term in normalized for term in explicit_codex_terms): + return True + + non_codex_terms = ("디스크", "용량", "저장소", "나스", "nas", "gitea", "repo", "cpu", "메모리") + if any(term in normalized for term in non_codex_terms): + return False + + implicit_codex_terms = ("주간 사용량", "월간 사용량", "남은 사용량", "남은 한도", "쿼터", "quota", "credits", "크레딧") + return any(term in normalized for term in implicit_codex_terms) + + +def _format_codex_usage(payload: Any) -> str: + if not isinstance(payload, dict): + return "Codex 사용량 응답 형식을 해석하지 못했습니다." + + lines = ["Codex 사용량/제한:"] + account = payload.get("account") if isinstance(payload.get("account"), dict) else payload + plan = _first_present(account, "planType", "plan_type", "plan", "tier") + email = _first_present(account, "email", "userEmail", "user_email") + auth_mode = _first_present(account, "authMode", "auth_mode") + if email: + lines.append(f"계정={email}") + if plan: + lines.append(f"플랜={plan}") + if auth_mode: + lines.append(f"인증={auth_mode}") + + rate_limits = payload.get("rateLimitsByLimitId") + if isinstance(rate_limits, dict) and rate_limits: + lines.append("제한:") + for limit_id, bucket in rate_limits.items(): + if isinstance(bucket, dict): + lines.append(_format_rate_limit_bucket(str(limit_id), bucket)) + else: + legacy = payload.get("rateLimits") + if isinstance(legacy, dict) and legacy: + lines.append("제한:") + lines.append(_format_rate_limit_bucket("codex", legacy)) + elif isinstance(legacy, list) and legacy: + lines.append("제한:") + for index, bucket in enumerate(legacy, start=1): + if isinstance(bucket, dict): + lines.append(_format_rate_limit_bucket(str(bucket.get("limitId") or index), bucket)) + + credits = payload.get("credits") + if isinstance(credits, dict) and credits: + lines.append("크레딧:") + for key, value in credits.items(): + if _is_safe_scalar(value): + lines.append(f"- {key}={value}") + + if len(lines) == 1: + lines.append("현재 Codex app-server가 사용량/제한 값을 반환하지 않았습니다.") + lines.append("Codex UI의 usage dashboard 또는 CLI 활성 세션의 /status에서도 확인해 보세요.") + return "\n".join(lines) + + +def _format_rate_limit_bucket(limit_id: str, bucket: dict[str, Any]) -> str: + name = bucket.get("limitName") or bucket.get("name") or limit_id + used_percent = _number_or_none(_value_for_any_key(bucket, "usedPercent", "used_percent")) + remaining = None if used_percent is None else max(0.0, 100.0 - used_percent) + window = _value_for_any_key(bucket, "windowDurationMins", "window_duration_mins") + resets_at = _value_for_any_key(bucket, "resetsAt", "resets_at") + plan = _value_for_any_key(bucket, "planType", "plan_type") + reached = _value_for_any_key(bucket, "rateLimitReachedType", "rate_limit_reached_type") + + parts = [f"- {name}"] + if remaining is not None: + parts.append(f"남음≈{remaining:.1f}%") + parts.append(f"사용={used_percent:.1f}%") + if window: + parts.append(f"창={window}분") + reset_text = _format_unix_time(resets_at) + if reset_text: + parts.append(f"리셋={reset_text}") + if plan: + parts.append(f"플랜={plan}") + if reached: + parts.append(f"상태={reached}") + return " / ".join(parts) + + +def _first_present(mapping: dict[str, Any], *keys: str) -> Any: + for key in keys: + value = mapping.get(key) + if value: + return value + return None + + +def _value_for_any_key(mapping: dict[str, Any], *keys: str) -> Any: + for key in keys: + if key in mapping and mapping[key] is not None: + return mapping[key] + return None + + +def _number_or_none(value: Any) -> float | None: + if value is None: + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def _format_unix_time(value: Any) -> str | None: + if isinstance(value, str) and value.strip(): + return value.strip() + number = _number_or_none(value) + if number is None: + return None + if number > 10_000_000_000: + number = number / 1000 + try: + from datetime import datetime + + return datetime.fromtimestamp(number).astimezone().strftime("%Y-%m-%d %H:%M:%S %Z") + except Exception: + return str(value) + + +def _is_safe_scalar(value: Any) -> bool: + return isinstance(value, (str, int, float, bool)) or value is None + + def _checked_label(label: str, checked: bool) -> str: return f"✓ {label}" if checked else label