53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
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
|