Upload summaries.py

local_rag/summaries.py

@@ -5,10 +5,50 @@ import os
 import re
 from pathlib import Path
 
+from huggingface_hub import hf_hub_download
 from local_rag.config import LocalRagConfig
 
 TITLE_PATTERN = re.compile(r"\*\*Title:\*\*\s*(.+)", re.MULTILINE)
-_FILENAME_TITLE = re.compile(r"^.+?\s-\s\d{4}\s-\s(.+)$")
+_FILENAME_TITLE = re.compile(r"^.+?\\s-\\s\\d{4}\\s-\\s(.+)$")
+_FILENAME_YEAR = re.compile(r"^.+?\\s-\\s(\\d{4})\\s-\\s")
+
+HF_DATASET_REPO = os.getenv("WATERSCOPE_DATASET_REPO", "EnvironmentalAI/WaterScopeAI-Data")
+HF_SUMMARY_PREFIX = os.getenv("WATERSCOPE_SUMMARY_PREFIX", "Summary")
+HF_SUMMARY_CACHE = Path(
+    os.getenv("WATERSCOPE_SUMMARY_CACHE", "/tmp/waterscope-summary-cache")
+).expanduser().resolve()
+HF_SUMMARY_ON_DEMAND = os.getenv("WATERSCOPE_SUMMARY_ON_DEMAND", "1").lower() in {
+    "1", "true", "yes"
+}
+
+
+def _year_from_filename(file_name: str) -> str | None:
+    match = _FILENAME_YEAR.match(Path(file_name).name)
+    return match.group(1) if match else None
+
+
+def _download_cloud_summary(file_name: str) -> Path | None:
+    """Download one summary from the HF Dataset and cache it locally."""
+    if not HF_SUMMARY_ON_DEMAND:
+        return None
+
+    year = _year_from_filename(file_name)
+    if not year:
+        return None
+
+    repo_file = f"{HF_SUMMARY_PREFIX}/{year}/{Path(file_name).name}"
+    try:
+        downloaded = hf_hub_download(
+            repo_id=HF_DATASET_REPO,
+            repo_type="dataset",
+            filename=repo_file,
+            local_dir=str(HF_SUMMARY_CACHE),
+        )
+    except Exception:
+        return None
+
+    path = Path(downloaded)
+    return path if path.is_file() else None
 
 
 def summary_search_dirs() -> list[Path]:
@@ -73,11 +113,14 @@ def extract_title(file_name: str, summary: str | None = None) -> str:
 
 
 def summary_path(file_name: str) -> Path | None:
+    # Preserve existing local behavior first.
     for root in summary_search_dirs():
         candidate = root / file_name
         if candidate.exists():
             return candidate
-    return None
+
+    # Cloud production fallback: fetch only the requested summary.
+    return _download_cloud_summary(file_name)
 
 
 def load_summary(file_name: str) -> str | None: