BetterToken API 错误排查：状态码、限流与重试

import json
import os
import random
import time
import urllib.error
import urllib.request
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime

def retry_after_seconds(value):
    try:
        return max(0.0, float(value))
    except ValueError:
        return max(0.0, (
            parsedate_to_datetime(value) - datetime.now(timezone.utc)
        ).total_seconds())

retry_temporary_429 = False  # Enable only after diagnosing the returned limit.
payload = json.dumps({
    "model": os.environ["BETTERTOKEN_MODEL_ID"],
    "messages": [{"role": "user", "content": "Reply with one word: OK"}],
}).encode()
for attempt in range(4):
    request = urllib.request.Request(
        "https://bettertoken.ai/v1/chat/completions", data=payload,
        headers={
            "Authorization": "Bearer " + os.environ["BETTERTOKEN_API_KEY"],
            "Content-Type": "application/json",
        },
    )
    try:
        with urllib.request.urlopen(request, timeout=120) as response:
            print(json.load(response)["choices"][0]["message"]["content"])
        break
    except urllib.error.HTTPError as error:
        body = error.read().decode("utf-8", errors="replace")
        retryable = error.code in (500, 502, 503) or (
            error.code == 429 and retry_temporary_429
        )
        if not retryable or attempt == 3:
            raise RuntimeError(f"HTTP {error.code}: {body}") from error
        delay = random.uniform(0, min(8.0, 2.0 ** attempt))
        retry_after = error.headers.get("Retry-After")
        if retry_after:
            delay = max(delay, retry_after_seconds(retry_after))
        if delay > 30:
            raise RuntimeError("Retry-After exceeds this retry budget") from error
        time.sleep(delay)