init

cnet.py

@@ -0,0 +1,93 @@
+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