94 lines
2.3 KiB
Python
94 lines
2.3 KiB
Python
import hashlib
|
|
import json
|
|
from json import JSONDecodeError
|
|
import os
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import requests
|
|
from requests import RequestException
|
|
|
|
REQUEST_TIMEOUT = 15.0
|
|
|
|
|
|
def as_int(value: object, default: int) -> int:
|
|
if isinstance(value, bool):
|
|
return default
|
|
if isinstance(value, int):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
return int(value.strip())
|
|
except ValueError:
|
|
return default
|
|
return default
|
|
|
|
|
|
def as_float(value: object, default: float | None = None) -> float | None:
|
|
return float(value) if isinstance(value, (int, float)) else default
|
|
|
|
|
|
def as_str(value: object, default: str = "") -> str:
|
|
return str(value) if isinstance(value, (str, int, float)) else default
|
|
|
|
|
|
def as_dict(value: object) -> dict[str, object]:
|
|
return cast(dict[str, object], value) if isinstance(value, dict) else {}
|
|
|
|
|
|
def as_list(value: object) -> list[object]:
|
|
return cast(list[object], value) if isinstance(value, list) else []
|
|
|
|
|
|
class HttpError(Exception):
|
|
NO_CODE = 0
|
|
|
|
def __init__(self, status_code: int, body: object) -> None:
|
|
self.status_code = status_code
|
|
self.body = body
|
|
super().__init__(f"HTTP {status_code}")
|
|
|
|
|
|
def _cache_path(url: str) -> Path:
|
|
cdir = Path(os.getenv("CACHE_DIR", ".cached"))
|
|
cdir.mkdir(parents=True, exist_ok=True)
|
|
url_hash = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16]
|
|
path = cdir / f"{url_hash}.json"
|
|
return path
|
|
|
|
|
|
def fetch_text(url: str) -> str:
|
|
path = _cache_path(url)
|
|
try:
|
|
if path.exists():
|
|
return path.read_text(encoding="utf-8")
|
|
except OSError:
|
|
pass
|
|
|
|
try:
|
|
resp = requests.get(url, timeout=REQUEST_TIMEOUT)
|
|
except RequestException as e:
|
|
raise HttpError(HttpError.NO_CODE, str(e)) from e
|
|
|
|
if resp.status_code != 200:
|
|
try:
|
|
body: object = resp.json()
|
|
except ValueError:
|
|
body = resp.text
|
|
raise HttpError(resp.status_code, body)
|
|
|
|
content = resp.text
|
|
try:
|
|
path.write_text(content, encoding="utf-8")
|
|
except OSError:
|
|
pass
|
|
return content
|
|
|
|
|
|
def fetch_json(url: str) -> object:
|
|
raw = fetch_text(url)
|
|
try:
|
|
return json.loads(raw)
|
|
except JSONDecodeError as exc:
|
|
raise HttpError(HttpError.NO_CODE, f"invalid JSON: {raw[:200]!r}") from exc
|