Reddit請求頻率限制：429錯誤排除與重試設定

from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
import math
import os
import random
import time
from typing import Optional
from urllib.parse import urlparse

import requests


OAUTH_TOKEN = os.environ["REDDIT_OAUTH_TOKEN"]
USER_AGENT = os.environ["REDDIT_USER_AGENT"]
AUTHORIZED_PATH_PREFIX = os.environ["AUTHORIZED_REDDIT_PATH_PREFIX"].rstrip("/") or "/"
REQUIRED_JSON_FIELDS = tuple(
    field.strip()
    for field in os.environ["REQUIRED_JSON_FIELDS"].split(",")
    if field.strip()
)

MAX_ATTEMPTS = 3
MAX_INLINE_WAIT_SECONDS = 300
TOTAL_ELAPSED_BUDGET_SECONDS = 360
ALLOWED_REDDIT_HOST = "oauth.reddit.com"


def validate_authorized_url(value: str) -> str:
    parsed = urlparse(value)
    prefix_matches = (
        parsed.path.startswith("/")
        if AUTHORIZED_PATH_PREFIX == "/"
        else parsed.path == AUTHORIZED_PATH_PREFIX
        or parsed.path.startswith(f"{AUTHORIZED_PATH_PREFIX}/")
    )
    if (
        parsed.scheme != "https"
        or parsed.hostname != ALLOWED_REDDIT_HOST
        or parsed.port not in (None, 443)
        or parsed.username is not None
        or parsed.password is not None
        or parsed.fragment
        or not prefix_matches
    ):
        raise ValueError("Refusing a Reddit URL outside the approved HTTPS host and path")
    return value


if not REQUIRED_JSON_FIELDS:
    raise ValueError("REQUIRED_JSON_FIELDS must name the endpoint's required fields")

AUTHORIZED_URL = validate_authorized_url(os.environ["AUTHORIZED_REDDIT_URL"])


def parse_retry_after(value: Optional[str]) -> Optional[int]:
    if value is None:
        return None
    text = value.strip()
    if text.isdigit():
        return int(text)
    try:
        retry_at = parsedate_to_datetime(text)
    except (TypeError, ValueError, OverflowError):
        return None
    if retry_at.tzinfo is None:
        return None
    seconds = (retry_at - datetime.now(timezone.utc)).total_seconds()
    return max(0, math.ceil(seconds))


def parse_reset_seconds(value: Optional[str]) -> Optional[int]:
    if value is None:
        return None
    try:
        seconds = float(value.strip())
    except (TypeError, ValueError):
        return None
    if not math.isfinite(seconds) or seconds < 0:
        return None
    return math.ceil(seconds)


def parse_wait_seconds(response: requests.Response) -> int:
    candidates = [
        parse_retry_after(response.headers.get("Retry-After")),
        parse_reset_seconds(response.headers.get("X-Ratelimit-Reset")),
    ]
    valid = [value for value in candidates if value is not None]
    if not valid:
        raise RuntimeError("429 had no valid Retry-After or reset value; defer the job")
    return max(1, max(valid))


def validate_success_payload(response: requests.Response) -> dict:
    media_type = response.headers.get("Content-Type", "").split(";", 1)[0].lower()
    if media_type != "application/json" and not media_type.endswith("+json"):
        raise RuntimeError(f"Unexpected success content type: {media_type or 'missing'}")
    try:
        payload = response.json()
    except ValueError as exc:
        raise RuntimeError("Success response did not contain valid JSON") from exc
    if not isinstance(payload, dict):
        raise RuntimeError("Success JSON did not match the expected object shape")
    missing = [field for field in REQUIRED_JSON_FIELDS if field not in payload]
    if missing:
        raise RuntimeError(f"Success JSON is missing required fields: {', '.join(missing)}")
    return payload


session = requests.Session()
session.headers.update(
    {
        "Authorization": f"Bearer {OAUTH_TOKEN}",
        "User-Agent": USER_AGENT,
        "Accept": "application/json",
    }
)

started_at = time.monotonic()

for attempt in range(1, MAX_ATTEMPTS + 1):
    elapsed = time.monotonic() - started_at
    if elapsed >= TOTAL_ELAPSED_BUDGET_SECONDS:
        raise RuntimeError("Elapsed-time budget exhausted before the next request")

    try:
        response = session.get(
            AUTHORIZED_URL,
            timeout=(5, 20),
            allow_redirects=False,
        )
    except requests.Timeout as exc:
        raise RuntimeError("Reddit request timed out; no retry was scheduled") from exc
    except requests.RequestException as exc:
        raise RuntimeError("Reddit request failed before a valid response arrived") from exc

    if 300 <= response.status_code < 400:
        raise RuntimeError("Redirect refused so the Bearer token stays on the approved host")

    if response.status_code == 429:
        wait_seconds = parse_wait_seconds(response)
        if attempt == MAX_ATTEMPTS:
            raise RuntimeError("Reddit rate limit persisted after bounded attempts")
        if wait_seconds > MAX_INLINE_WAIT_SECONDS:
            raise RuntimeError(
                f"Defer this job for at least {wait_seconds} seconds instead of sleeping inline"
            )

        sleep_seconds = wait_seconds + random.uniform(0.25, 1.0)
        if time.monotonic() - started_at + sleep_seconds > TOTAL_ELAPSED_BUDGET_SECONDS:
            raise RuntimeError("Defer the job because the next wait exceeds its elapsed-time budget")
        time.sleep(sleep_seconds)
        continue

    if not 200 <= response.status_code < 300:
        try:
            response.raise_for_status()
        except requests.HTTPError as exc:
            raise RuntimeError(f"Reddit returned HTTP {response.status_code}") from exc
        raise RuntimeError(f"Unexpected non-success HTTP status: {response.status_code}")

    validate_success_payload(response)
    print(
        "Validated Reddit JSON response",
        response.status_code,
        "required_fields=",
        ",".join(REQUIRED_JSON_FIELDS),
    )
    break
else:
    raise RuntimeError("Request loop ended without a response")