75 lines
2.1 KiB
Python
75 lines
2.1 KiB
Python
"""Shared HTTP session for SFDA / ADE scraping."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Any
|
|
|
|
import requests
|
|
|
|
DEFAULT_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
|
|
"Accept-Language": "en-US,en;q=0.9",
|
|
}
|
|
|
|
SFDA_BASE = "https://www.sfda.gov.sa"
|
|
ADE_BASE = "https://ade.sfda.gov.sa"
|
|
|
|
# Rate-limit protection for ADE (DownloadDocument throttles after ~10 rapid calls).
|
|
ADE_REQUEST_DELAY_SECONDS = 3
|
|
ADE_MAX_RETRIES = 3
|
|
ADE_RETRY_BACKOFF_SECONDS = 30
|
|
ADE_RETRY_STATUS_CODES = {429, 503}
|
|
|
|
|
|
def create_session() -> requests.Session:
|
|
session = requests.Session()
|
|
session.headers.update(DEFAULT_HEADERS)
|
|
return session
|
|
|
|
|
|
def _retry_wait_seconds(response: requests.Response, attempt: int) -> float:
|
|
retry_after = response.headers.get("Retry-After")
|
|
if retry_after:
|
|
try:
|
|
return max(float(retry_after), 1.0)
|
|
except ValueError:
|
|
pass
|
|
return ADE_RETRY_BACKOFF_SECONDS * (2**attempt)
|
|
|
|
|
|
def request_with_retry(
|
|
session: requests.Session,
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
max_retries: int = ADE_MAX_RETRIES,
|
|
retry_status_codes: set[int] | None = None,
|
|
**kwargs: Any,
|
|
) -> requests.Response:
|
|
"""Perform an HTTP request with exponential backoff on rate-limit responses."""
|
|
retry_status_codes = retry_status_codes or ADE_RETRY_STATUS_CODES
|
|
last_response: requests.Response | None = None
|
|
|
|
for attempt in range(max_retries + 1):
|
|
response = session.request(method, url, **kwargs)
|
|
last_response = response
|
|
|
|
if response.status_code not in retry_status_codes:
|
|
response.raise_for_status()
|
|
return response
|
|
|
|
if attempt >= max_retries:
|
|
response.raise_for_status()
|
|
|
|
wait_seconds = _retry_wait_seconds(response, attempt)
|
|
time.sleep(wait_seconds)
|
|
|
|
if last_response is not None:
|
|
last_response.raise_for_status()
|
|
raise requests.HTTPError(f"Request failed for {url}")
|