Update app.py

app.py

@@ -32,15 +32,18 @@ from api_usage import (
 )
 
 # ─────────────────────────────────────────
-# Key-specific helper functions (원본 유지 - Gemini 포함)
+# Key-specific helper functions (OpenAI 함수 수정됨)
 # ─────────────────────────────────────────
 def get_key_oai_info(key: str) -> Dict[str, Any]:
     session = requests.Session()
-    status, org_data = check_key_availability(session, key)
+    # raw_status_code는 HTTP 응답 코드 또는 유사한 상태 표시자
+    # org_data_or_error는 성공 시 데이터, 실패 시 에러 정보 등
+    raw_status_code, org_data_or_error = check_key_availability(session, key)
 
     info_dict = {
         "key_type": "OpenAI",
-        "key_availability": True if status else False,
+        "key_availability": False,
+        "has_sufficient_quota": True, # 기본적으로 True로 가정, 문제 시 False로 변경
         "gpt4_availability": "",
         "gpt4_32k_availability": "",
         "default_org": "",
@@ -51,17 +54,63 @@ def get_key_oai_info(key: str) -> Dict[str, Any]:
         "tokens_per_minute": "",
         "quota": "",
         "all_models": "",
+        "status_message": ""
     }
 
-    if not status:
+    org_data_for_subscription = None
+
+    if raw_status_code == 200:
+        info_dict["key_availability"] = True
+        org_data_for_subscription = org_data_or_error
+    elif raw_status_code == 401: # Unauthorized
+        info_dict["status_message"] = "Unauthorized: Invalid API key."
+        info_dict["has_sufficient_quota"] = False
+        return info_dict
+    elif raw_status_code == 403: # Forbidden
+        status_me, orgs_me_data = get_orgs_me(session, key)
+        if status_me == 200:
+            info_dict["key_availability"] = True
+            org_data_for_subscription = orgs_me_data
+        else:
+            info_dict["status_message"] = f"Forbidden, and get_orgs_me failed (status: {status_me}). Key might be inactive or lack permissions."
+            info_dict["has_sufficient_quota"] = False
+            return info_dict
+    elif raw_status_code == 429: # Too Many Requests (Rate Limit or Quota)
+        info_dict["key_availability"] = True # 키 자체는 유효할 수 있음
+        info_dict["has_sufficient_quota"] = False # 쿼터 부족으로 간주
+        info_dict["status_message"] = "Rate limit or quota likely exceeded (initial check)."
+        if isinstance(org_data_or_error, dict) and "error" in org_data_or_error:
+            error_details = org_data_or_error["error"]
+            current_quota_message = error_details.get("message", "Quota details unavailable from initial check")
+            info_dict["quota"] = current_quota_message
+            if "insufficient_quota" not in current_quota_message.lower(): # 상세 메시지에 insufficient_quota가 없다면 상태 메시지에 추가
+                 info_dict["status_message"] += f" Error: {current_quota_message}"
+        # 여기서 반환하면 아래 구독 정보 확인을 건너뛰지만, has_sufficient_quota가 이미 False임
+        # return info_dict # 필요에 따라 주석 해제하여 더 빠른 반환 가능
+    else: # 기타 에러
+        info_dict["status_message"] = f"Key check failed (status: {raw_status_code})."
+        if isinstance(org_data_or_error, dict) and "error" in org_data_or_error:
+            info_dict["status_message"] += f" Error: {org_data_or_error['error'].get('message', str(org_data_or_error))}"
+        elif isinstance(org_data_or_error, str):
+             info_dict["status_message"] += f" Details: {org_data_or_error}"
+        info_dict["has_sufficient_quota"] = False
         return info_dict
 
-    if status == 403:
-        status_me, orgs_me = get_orgs_me(session, key)
+    if not info_dict["key_availability"]:
+        return info_dict # 키가 유효하지 않으면 더 이상 진행 안 함
+
+    # org_data_for_subscription이 준비되지 않은 경우 (예: 초기 200이었지만 org_data가 부적절)
+    if not org_data_for_subscription:
+        status_me, orgs_me_data = get_orgs_me(session, key)
         if status_me == 200:
-            org_data = orgs_me
+            org_data_for_subscription = orgs_me_data
+        else:
+            info_dict["status_message"] = (info_dict["status_message"] + " Could not identify organization for subscription.").strip()
+            info_dict["key_availability"] = False
+            info_dict["has_sufficient_quota"] = False
+            return info_dict
 
-    subscription_info = get_subscription(key, session, org_data)
+    subscription_info = get_subscription(key, session, org_data_for_subscription)
     if subscription_info:
         info_dict.update(
             {
@@ -73,10 +122,39 @@ def get_key_oai_info(key: str) -> Dict[str, Any]:
                 "models": subscription_info.get("models", ""),
                 "requests_per_minute": subscription_info.get("rpm", ""),
                 "tokens_per_minute": subscription_info.get("tpm", ""),
-                "quota": subscription_info.get("quota", ""),
+                "quota": subscription_info.get("quota", info_dict["quota"]),
                 "all_models": subscription_info.get("all_models", ""),
             }
         )
+
+        error_info = subscription_info.get("error")
+        if error_info and isinstance(error_info, dict):
+            err_type = error_info.get("type", "").lower()
+            err_code = error_info.get("code", "").lower()
+            err_msg = error_info.get("message", "").lower()
+
+            if "insufficient_quota" in err_type or \
+               "insufficient_quota" in err_code or \
+               "you exceeded your current quota" in err_msg or \
+               "payment required" in err_msg or \
+               ("billing" in err_msg and "issue" in err_msg):
+                info_dict["has_sufficient_quota"] = False
+                new_quota_message = f"Insufficient: {error_info.get('message', err_type)}"
+                info_dict["quota"] = new_quota_message # 쿼터 필드에 명시적 메시지
+                info_dict["status_message"] = (info_dict["status_message"] + f" Quota/Billing issue: {error_info.get('message', err_type)}").strip()
+
+        # 계정 비활성화 등 확인
+        if "account_deactivated" in str(subscription_info).lower() or \
+           "payment_failed" in str(subscription_info).lower():
+            info_dict["has_sufficient_quota"] = False
+            if "Account issue" not in info_dict["status_message"]: # 중복 메시지 방지
+                 info_dict["status_message"] = (info_dict["status_message"] + " Account issue (e.g., deactivated, payment failed).").strip()
+    else:
+        info_dict["status_message"] = (info_dict["status_message"] + " Failed to retrieve full subscription details.").strip()
+        info_dict["has_sufficient_quota"] = False
+        # 구독 정보 없으면 key_availability도 False로 간주할 수 있음.
+        # info_dict["key_availability"] = False # 주석 처리하여 키 자체는 유효할 수 있음을 남김
+
     return info_dict
 
 
@@ -367,10 +445,10 @@ def not_supported(key: str) -> Dict[str, Any]:
 
 
 # ─────────────────────────────────────────
-#  단일 키 비동기 처리 (업데이트됨 - OpenAI, Anthropic 작동 플래그 추가)
+#  단일 키 비동기 처리 (업데이트됨)
 # ─────────────────────────────────────────
 async def process_single_key(key: str, rate_limit: bool, claude_model: str) -> Dict[str, Any]:
-    """주어진 key 하나를 분석해 정보 dict 반환. 작동하는 키들의 플래그 포함."""
+    """주어진 key 하나를 분석해 정보 dict 반환. Gemini 키의 작동 여부 플래그 포함."""
     _key = key.strip()
 
     if not _key:
@@ -381,16 +459,11 @@ async def process_single_key(key: str, rate_limit: bool, claude_model: str) -> D
         result = get_key_openrouter_info(_key)
         return {"key": _key, **result}
 
-    # Anthropic Claude (작동 플래그 추가)
+    # Anthropic Claude
     if re.match(re.compile(r"sk-ant-api03-[a-zA-Z0-9\-_]{93}AA"), _key) or \
        (_key.startswith("sk-ant-") and len(_key) == 93) or \
        (len(_key) == 89 and re.match(re.compile(r"sk-[a-zA-Z0-9]{86}"), _key)):
         result = await get_key_ant_info(_key, rate_limit, claude_model)
-        # Anthropic 키 작동 여부 판단: key_availability가 True이고 status에 오류가 없으면 작동
-        is_working = result.get("key_availability") and \
-                    not any(error_word in str(result.get("status", "")).lower() 
-                           for error_word in ["error", "invalid", "expired", "forbidden", "unauthorized"])
-        result["is_anthropic_working"] = is_working
         return {"key": _key, **result}
 
     # Stability
@@ -403,15 +476,12 @@ async def process_single_key(key: str, rate_limit: bool, claude_model: str) -> D
          result = get_key_deepseek_info(_key)
          return {"key": _key, **result}
 
-    # OpenAI (다른 sk- 패턴보다 뒤에 와야 함) (작동 플래그 추가)
+    # OpenAI (다른 sk- 패턴보다 뒤에 와야 함)
     if _key.startswith("sk-"):
-        result = get_key_oai_info(_key)
-        # OpenAI 키 작동 여부 판단: key_availability가 True이면 작동
-        is_working = result.get("key_availability") == True
-        result["is_openai_working"] = is_working
+        result = get_key_oai_info(_key) # 수정된 함수 호출
         return {"key": _key, **result}
 
-    # Google Gemini 처리 (기존 유지)
+    # Google Gemini 처리
     if _key.startswith("AIzaSy"):
         gemini_info = get_key_gemini_info(_key)
         is_working = gemini_info.get("key_availability") and gemini_info.get("status") == "Working"
@@ -492,45 +562,51 @@ async def process_single_key(key: str, rate_limit: bool, claude_model: str) -> D
 
 
 # ─────────────────────────────────────────
-#  여러 key 비동기 처리 함수 (업데이트됨 - 모든 작동 키 반환)
+#  여러 key 비동기 처리 함수 (업데이트됨)
 # ─────────────────────────────────────────
-async def sort_keys(text: str, rate_limit: bool, claude_model: str) -> Tuple[List[Dict[str, Any]], str]:
-    """텍스트 박스에 입력된 여러 키를 줄 단위로 분석하고, 전체 결과와 작동하는 키 목록을 반환."""
+async def sort_keys(text: str, rate_limit: bool, claude_model: str) -> Tuple[List[Dict[str, Any]], str, str, str]:
+    """
+    텍스트 박스에 입력된 여러 키를 줄 단위로 분석하고,
+    전체 결과와 작동하는 OAI, Anthropic, Gemini 키 목록을 각각 반환합니다.
+    """
     keys = [k.strip() for k in text.splitlines() if k.strip()]
     if not keys:
-        return [], ""
+        return [], "", "", ""
 
     tasks = [process_single_key(k, rate_limit, claude_model) for k in keys]
     results = await asyncio.gather(*tasks)
 
-    # 작동하는 키들을 수집
-    working_keys = []
-    
+    working_gemini_keys = []
+    working_oai_keys = []
+    working_anthropic_keys = []
+
     for result in results:
-        key_value = result.get("key", "")
-        key_type = result.get("key_type", "")
-        
-        # OpenAI 키 확인
-        if result.get("is_openai_working"):
-            working_keys.append(f"[OpenAI] {key_value}")
-        
-        # Anthropic 키 확인
-        elif result.get("is_anthropic_working"):
-            working_keys.append(f"[Anthropic] {key_value}")
-        
-        # Gemini 키 확인 (기존)
-        elif result.get("is_gemini_working"):
-            working_keys.append(f"[Gemini] {key_value}")
-
-    # 결과를 JSON용 리스트와 작동하는 키 목록 문자열로 반환
-    return results, "\n".join(working_keys)
+        key_value = result.get("key")
+        if not key_value:
+            continue
+
+        key_type = result.get("key_type")
+
+        # Gemini 키 필터링
+        if result.get("is_gemini_working"):
+            working_gemini_keys.append(key_value)
+        # OpenAI 키 필터링 (수정됨: has_sufficient_quota 확인)
+        elif key_type == "OpenAI" and \
+             result.get("key_availability") is True and \
+             result.get("has_sufficient_quota") is True: # 이 필드가 True여야 함
+            working_oai_keys.append(key_value)
+        # Anthropic 키 필터링
+        elif key_type == "Anthropic Claude" and result.get("key_availability") is True:
+            working_anthropic_keys.append(key_value)
+
+    return results, "\n".join(working_oai_keys), "\n".join(working_anthropic_keys), "\n".join(working_gemini_keys)
 
 
 # ─────────────────────────────────────────
-#  UI util
+#  UI util (업데이트됨)
 # ─────────────────────────────────────────
 def clear_inputs():
-    return "", "", ""
+    return "", "", "", "", ""
 
 
 # ─────────────────────────────────────────
@@ -539,9 +615,10 @@ def clear_inputs():
 with gr.Blocks() as demo:
     gr.Markdown(
         """
-# Multi-API Key Status Checker (Enhanced Working Keys Display)
+# Multi-API Key Status Checker (OAI, Anthropic, Gemini Enhanced)
 *(Based on shaocongma, CncAnon1, su, Drago, kingbased key checkers)*
 Check the status and details of various API keys including OpenAI, Anthropic, Gemini, Azure, Mistral, Replicate, AWS Claude, OpenRouter, Vertex AI (GCP Anthropic), Groq, NovelAI, ElevenLabs, xAI, Stability AI, and DeepSeek.
+This version highlights working OpenAI (with sufficient quota), Anthropic, and Gemini keys in separate text boxes.
 
 **Key Formats:**
 *   **AWS:** `AWS_ACCESS_KEY_ID:AWS_SECRET_ACCESS_KEY` (root might not be accurate)
@@ -582,29 +659,40 @@ Check the status and details of various API keys including OpenAI, Anthropic, Ge
 
         with gr.Column(scale=2):
             info = gr.JSON(label=" API Key Information (All Results)", open=True)
-            # 작동하는 키들 출력 박스 (확장됨)
-            working_keys_output = gr.Textbox(
-                label="Working Keys (OpenAI, Anthropic, Gemini)",
-                info="Lists all keys confirmed as working for OpenAI, Anthropic Claude, and Gemini services.",
-                lines=5,
-                max_lines=15,
+            oai_keys_output = gr.Textbox(
+                label="Working OpenAI Keys (Sufficient Quota)",
+                info="Lists OpenAI keys confirmed as working and having sufficient quota.",
+                lines=3,
+                max_lines=10,
+                interactive=False,
+            )
+            anthropic_keys_output = gr.Textbox(
+                label="Working Anthropic Keys",
+                info="Lists Anthropic keys confirmed as working (key_availability is True).",
+                lines=3,
+                max_lines=10,
+                interactive=False,
+            )
+            gemini_keys_output = gr.Textbox(
+                label="Working Gemini Keys",
+                info="Lists Gemini keys confirmed as 'Working'.",
+                lines=3,
+                max_lines=10,
                 interactive=False,
             )
 
-    # Clear 버튼 클릭 시 입력 박스와 출력 박스 모두 초기화
     clear_button.click(
         fn=clear_inputs,
         inputs=None,
-        outputs=[key_box, info, working_keys_output]
+        outputs=[key_box, info, oai_keys_output, anthropic_keys_output, gemini_keys_output]
     )
 
-    # Submit 버튼 클릭 시 sort_keys 함수 호출 및 결과 분배
     submit_button.click(
         fn=sort_keys,
         inputs=[key_box, rate_limit, claude_model],
-        outputs=[info, working_keys_output],
+        outputs=[info, oai_keys_output, anthropic_keys_output, gemini_keys_output],
         api_name="sort_keys",
     )
 
-# demo.launch(share=True) # 필요시 공유 링크 활성화
+# demo.launch(share=True)
 demo.launch()