Upload paper_linking.py

backend/paper_linking.py

@@ -1,199 +1,362 @@
 from __future__ import annotations
 
-import hashlib
+import json
+import os
 import re
+import unicodedata
+from pathlib import Path
 
-from local_rag.summaries import load_manifest, summary_path
+from huggingface_hub import hf_hub_download
+from local_rag.config import LocalRagConfig
 
-_RP_SEGMENT_RE = re.compile(r";\s*")
-_FILENAME_AUTHORS_RE = re.compile(r"^(.+?)\s*-\s*\d{4}\s*-\s*")
-_ET_AL_RE = re.compile(r"\bet\s+al\b\.?", re.I)
 
-_NON_AUTHOR_PREFIXES = (
-    "not specified",
-    "not explicitly",
-    "not listed",
-    "not available",
-    "not provided",
-    "unknown",
-    "co-author",
+TITLE_PATTERN = re.compile(r"\*\*Title:\*\*\s*(.+)", re.MULTILINE)
+_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",
+)
 
-def _is_non_author_text(value: str) -> bool:
-    text = (value or "").strip().lower()
+HF_SUMMARY_CACHE = Path(
+    os.getenv(
+        "WATERSCOPE_SUMMARY_CACHE",
+        "/tmp/waterscope-summary-cache",
+    )
+).expanduser().resolve()
 
-    if not text:
-        return True
+HF_SUMMARY_ON_DEMAND = os.getenv(
+    "WATERSCOPE_SUMMARY_ON_DEMAND",
+    "1",
+).lower() in {
+    "1",
+    "true",
+    "yes",
+}
 
-    if text in {"n/a", "na", "none"}:
-        return True
 
-    return text.startswith(_NON_AUTHOR_PREFIXES)
+def _normalize_filename(file_name: str) -> str:
+    """
+    Normalize a filename so visually identical Unicode forms match.
 
+    Example:
+        Türkiye
+        Türkiye
 
-def _surname_key(name: str) -> str:
-    """Normalize an author label for surname-level deduplication."""
-    text = (name or "").strip()
-    if not text:
-        return ""
+    NFKC is also used by the figure lookup code.
+    """
+    name = Path(file_name or "").name.strip()
+    return unicodedata.normalize("NFKC", name)
 
-    # For "... Surname et al.", always use the word immediately before "et al."
-    et_al_match = re.search(r"\bet\s+al\b\.?", text, flags=re.I)
-    if et_al_match:
-        before_et_al = text[:et_al_match.start()].strip()
-        parts = before_et_al.split()
-        if parts:
-            return parts[-1].strip(".,;:").lower()
 
-    # "Surname, Initials" format
-    if "," in text:
-        return text.split(",", 1)[0].strip().lower()
+def _year_from_filename(file_name: str) -> str | None:
+    safe_name = _normalize_filename(file_name)
+    match = _FILENAME_YEAR.match(safe_name)
+    return match.group(1) if match else None
 
-    # Fallback
-    parts = text.split()
-    return parts[0].strip(".,;:").lower() if parts else ""
 
+def _download_cloud_summary(file_name: str) -> Path | None:
+    """
+    Download one summary from the HF Dataset and cache it locally.
 
-def parse_authors_from_filename(file_name: str) -> list[str]:
-    """Extract author names from 'Authors - YYYY - Title.ext' filenames."""
-    name = (file_name or "").strip()
-    if not name:
-        return []
+    The filename is Unicode-normalized before constructing the
+    Summary/<year>/<filename> path.
+    """
+    if not HF_SUMMARY_ON_DEMAND:
+        return None
 
-    stem = name.rsplit(".", 1)[0]
-    match = _FILENAME_AUTHORS_RE.match(stem)
-    if not match:
-        return []
+    safe_name = _normalize_filename(file_name)
 
-    author_part = match.group(1).strip()
-    if _ET_AL_RE.search(author_part):
-        return [author_part]
-    return [part.strip() for part in re.split(r"\s+and\s+", author_part, flags=re.I) if part.strip()]
+    year = _year_from_filename(safe_name)
+    if not year:
+        return None
+
+    repo_file = f"{HF_SUMMARY_PREFIX}/{year}/{safe_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 as exc:
+        print(
+            f"[SUMMARY DOWNLOAD ERROR] {repo_file}: {repr(exc)}",
+            flush=True,
+        )
+        return None
 
+    path = Path(downloaded)
+    return path if path.is_file() else None
 
-def parse_authors_from_rp(rp: str) -> str:
-    """Extract usable author names from RP and keep the more specific duplicate."""
-    text = (rp or "").strip()
-    if not text or _is_non_author_text(text):
-        return ""
 
-    names: list[str] = []
-    by_surname: dict[str, int] = {}
+def summary_search_dirs() -> list[Path]:
+    """Directories to search for summary .txt files (first match wins)."""
+    dirs: list[Path] = []
 
-    def _prefer(current: str, candidate: str) -> str:
-        # Prefer the more specific/longer representation
-        if len(candidate) > len(current):
-            return candidate
-        return current
+    for raw in (os.getenv("PAPER_SUMMARIES_DIRS") or "").split(":"):
+        raw = raw.strip()
+        if raw:
+            dirs.append(
+                Path(raw).expanduser().resolve()
+            )
 
-    for segment in _RP_SEGMENT_RE.split(text):
-        segment = segment.strip()
+    override = os.getenv("PAPER_SUMMARIES_DIR")
+    if override:
+        dirs.append(
+            Path(override).expanduser().resolve()
+        )
 
-        if not segment or _is_non_author_text(segment):
-            continue
+    project_root = Path(__file__).resolve().parents[1]
+    dirs.append(
+        project_root / "paper_summaries"
+    )
 
-        name = segment.split("(", 1)[0].strip().rstrip(",")
+    unique: list[Path] = []
+    seen: set[str] = set()
 
-        if not name or _is_non_author_text(name):
-            continue
+    for path in dirs:
+        key = str(path)
 
-        key = _surname_key(name)
-        if not key:
+        if key in seen:
             continue
 
-        if key in by_surname:
-            idx = by_surname[key]
-            names[idx] = _prefer(names[idx], name)
-        else:
-            by_surname[key] = len(names)
-            names.append(name)
+        seen.add(key)
+        unique.append(path)
 
-    return "; ".join(names)
+    return unique
 
 
-def resolve_authors(row: dict) -> str:
-    """
-    Resolve authors using the best available metadata source.
+def summaries_dir() -> Path:
+    for path in summary_search_dirs():
+        if path.exists():
+            return path
 
-    Priority:
-    1. AU  - full author list from publication metadata
-    2. RP  - author information parsed from full text
-    3. PDF/TXT filename - fallback only
-    """
+    return (
+        Path(__file__).resolve().parents[1]
+        / "paper_summaries"
+    )
+
+
+def _manifest_path(
+    config: LocalRagConfig | None = None,
+) -> Path:
+    cfg = config or LocalRagConfig.from_env()
+    return cfg.manifest_path
 
-    # 1. Prefer the full author list from metadata
-    au = str(row.get("AU") or "").strip()
-    if au:
-        return au
 
-    # 2. Fall back to RP
-    rp_authors = parse_authors_from_rp(row.get("RP") or "").strip()
-    if rp_authors:
-        return rp_authors
+def load_manifest(
+    config: LocalRagConfig | None = None,
+) -> dict:
+    path = _manifest_path(config)
 
-    # 3. Final fallback: filename
-    filename_authors = parse_authors_from_filename(
-        row.get("PDF_file") or row.get("TXT_file") or ""
+    if not path.exists():
+        raise FileNotFoundError(
+            f"Manifest not found: {path}. "
+            "Rebuild local_rag_data/manifest.json first."
+        )
+
+    return json.loads(
+        path.read_text(encoding="utf-8")
     )
 
-    return "; ".join(filename_authors)
 
+def title_from_filename(file_name: str) -> str:
+    stem = Path(file_name).stem
+
+    match = _FILENAME_TITLE.match(stem)
+
+    if match:
+        return match.group(1).strip()
+
+    return stem
+
+
+def extract_title(
+    file_name: str,
+    summary: str | None = None,
+) -> str:
+    if summary:
+        match = TITLE_PATTERN.search(summary)
 
-def format_authors(authors: str, *, max_names: int | None = None) -> str:
-    """Display string for UI cards. Default shows every parsed author."""
-    if not authors:
-        return ""
-    parts = [part.strip() for part in authors.split(";") if part.strip()]
-    if max_names is None or len(parts) <= max_names:
-        return "; ".join(parts)
-    return "; ".join(parts[:max_names]) + "; et al."
+        if match:
+            return match.group(1).strip()
 
+    return title_from_filename(file_name)
 
-def document_id_for_txt_file(txt_file: str) -> str | None:
-    name = (txt_file or "").strip()
-    if not name:
+
+def _find_normalized_local_file(
+    directory: Path,
+    file_name: str,
+) -> Path | None:
+    """
+    Find a locally available/cached summary using exact
+    Unicode-normalized filename matching.
+
+    No fuzzy or prefix matching is used.
+    """
+    if not directory.is_dir():
         return None
 
-    manifest = load_manifest()
-    files = manifest.get("files", [])
+    safe_name = _normalize_filename(file_name)
 
-    for entry in files:
-        if entry.get("file_name") == name:
-            return entry.get("document_id")
+    # Fast exact lookup first.
+    candidate = directory / safe_name
 
-    lowered = name.lower()
-    for entry in files:
-        file_name = entry.get("file_name", "")
-        if file_name.lower() == lowered:
-            return entry.get("document_id")
+    if candidate.is_file():
+        return candidate
 
-    if not name.endswith(".txt"):
-        with_txt = f"{name}.txt"
-        for entry in files:
-            if entry.get("file_name") == with_txt:
-                return entry.get("document_id")
+    # Unicode-normalized fallback.
+    target = safe_name.casefold()
 
-    if summary_path(name) is not None:
-        stem = name[:-4] if name.lower().endswith(".txt") else name
-        return hashlib.sha1(stem.encode("utf-8")).hexdigest()[:16]
+    try:
+        for path in directory.iterdir():
+
+            if not path.is_file():
+                continue
+
+            existing_name = _normalize_filename(
+                path.name
+            )
+
+            if existing_name.casefold() == target:
+                return path
+
+    except OSError:
+        return None
 
     return None
 
 
-def enrich_metadata_row(row: dict, row_index: int) -> dict:
-    txt_file = (row.get("TXT_file") or row.get("txt_file") or "").strip()
-    document_id = document_id_for_txt_file(txt_file)
-    authors = resolve_authors(row)
-    return {
-        "metadata_row_index": row_index,
-        "title": (row.get("TI") or "").strip(),
-        "abstract": (row.get("AB") or "").strip(),
-        "keywords": (row.get("DE") or "").strip(),
-        "authors": authors,
-        "doi": (row.get("DI") or "").strip(),
-        "year": (row.get("PY") or "").strip(),
-        "txt_file": txt_file,
-        "document_id": document_id,
-        "has_chat": document_id is not None,
+def summary_path(file_name: str) -> Path | None:
+    """
+    Resolve the summary locally first, then fetch it from HF.
+
+    Unicode-normalized exact filename matching is used.
+    """
+    safe_name = _normalize_filename(file_name)
+
+    year = _year_from_filename(safe_name)
+
+    # Preserve existing local behavior first.
+    for root in summary_search_dirs():
+
+        # Flat layout:
+        # paper_summaries/<file>.txt
+        found = _find_normalized_local_file(
+            root,
+            safe_name,
+        )
+
+        if found is not None:
+            return found
+
+        # Also support year-organized local folders.
+        if year:
+            found = _find_normalized_local_file(
+                root / year,
+                safe_name,
+            )
+
+            if found is not None:
+                return found
+
+    # Cloud production fallback:
+    # Summary/<year>/<normalized filename>
+    return _download_cloud_summary(safe_name)
+
+
+def load_summary(file_name: str) -> str | None:
+    path = summary_path(file_name)
+
+    if path is None:
+        return None
+
+    return path.read_text(
+        encoding="utf-8"
+    )
+
+
+def list_papers(
+    config: LocalRagConfig | None = None,
+) -> list[dict]:
+    """List indexed papers without reading every summary file into memory."""
+    manifest = load_manifest(config)
+
+    papers: list[dict] = []
+
+    for entry in manifest.get("files", []):
+        file_name = entry["file_name"]
+
+        papers.append(
+            {
+                "document_id": entry["document_id"],
+                "file_name": file_name,
+                "title": title_from_filename(
+                    file_name
+                ),
+                "chunk_count": entry.get(
+                    "chunk_count",
+                    0,
+                ),
+                "has_summary": (
+                    summary_path(file_name)
+                    is not None
+                ),
+            }
+        )
+
+    papers.sort(
+        key=lambda item: item["title"].lower()
+    )
+
+    return papers
+
+
+def summaries_for_documents(
+    document_ids: list[str],
+    config: LocalRagConfig | None = None,
+) -> list[dict]:
+    manifest = load_manifest(config)
+
+    by_id = {
+        entry["document_id"]: entry
+        for entry in manifest.get("files", [])
     }
+
+    selected: list[dict] = []
+
+    for document_id in document_ids:
+        entry = by_id.get(document_id)
+
+        if not entry:
+            continue
+
+        file_name = entry["file_name"]
+
+        summary = load_summary(file_name)
+
+        if not summary:
+            continue
+
+        selected.append(
+            {
+                "document_id": document_id,
+                "title": extract_title(
+                    file_name,
+                    summary,
+                ),
+                "file_name": file_name,
+                "text": summary,
+            }
+        )
+
+    return selected