B

app/services/fetch_cache.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import hashlib
+import json
+import time
+from pathlib import Path
+
+from pydantic import BaseModel, Field
+
+
+class FetchCacheEntry(BaseModel):
+    text: str
+    headers: dict[str, str] = Field(default_factory=dict)
+
+
+def build_fetch_cache_key(*, name: str, url: str) -> str:
+    raw = f"{name}|{url}"
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+
+def load_fetch_cache(*, cache_dir: Path, cache_key: str, ttl_seconds: int) -> FetchCacheEntry | None:
+    text_path = cache_dir / f"{cache_key}.txt"
+    meta_path = cache_dir / f"{cache_key}.json"
+    if not text_path.is_file() or not meta_path.is_file():
+        return None
+
+    expires_at = meta_path.stat().st_mtime + ttl_seconds
+    if expires_at < time.time():
+        return None
+
+    try:
+        metadata = json.loads(meta_path.read_text(encoding="utf-8"))
+    except json.JSONDecodeError:
+        return None
+
+    headers = metadata.get("headers")
+    if not isinstance(headers, dict):
+        headers = {}
+
+    return FetchCacheEntry(
+        text=text_path.read_text(encoding="utf-8"),
+        headers={str(k): str(v) for k, v in headers.items()},
+    )
+
+
+def save_fetch_cache(*, cache_dir: Path, cache_key: str, text: str, headers: dict[str, str]) -> Path:
+    cache_dir.mkdir(parents=True, exist_ok=True)
+    text_path = cache_dir / f"{cache_key}.txt"
+    meta_path = cache_dir / f"{cache_key}.json"
+    text_path.write_text(text, encoding="utf-8")
+    meta_path.write_text(json.dumps({"headers": headers}, ensure_ascii=False, indent=2), encoding="utf-8")
+    return text_path