Upload paper_linking.py

backend/paper_linking.py

@@ -1,362 +1,199 @@
 from __future__ import annotations
 
-import json
-import os
+import hashlib
 import re
-import unicodedata
-from pathlib import Path
 
-from huggingface_hub import hf_hub_download
-from local_rag.config import LocalRagConfig
+from local_rag.summaries import load_manifest, summary_path
 
+_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)
 
-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",
+_NON_AUTHOR_PREFIXES = (
+    "not specified",
+    "not explicitly",
+    "not listed",
+    "not available",
+    "not provided",
+    "unknown",
+    "co-author",
 )
 
-HF_SUMMARY_PREFIX = os.getenv(
-    "WATERSCOPE_SUMMARY_PREFIX",
-    "Summary",
-)
 
-HF_SUMMARY_CACHE = Path(
-    os.getenv(
-        "WATERSCOPE_SUMMARY_CACHE",
-        "/tmp/waterscope-summary-cache",
-    )
-).expanduser().resolve()
+def _is_non_author_text(value: str) -> bool:
+    text = (value or "").strip().lower()
 
-HF_SUMMARY_ON_DEMAND = os.getenv(
-    "WATERSCOPE_SUMMARY_ON_DEMAND",
-    "1",
-).lower() in {
-    "1",
-    "true",
-    "yes",
-}
+    if not text:
+        return True
 
+    if text in {"n/a", "na", "none"}:
+        return True
 
-def _normalize_filename(file_name: str) -> str:
-    """
-    Normalize a filename so visually identical Unicode forms match.
+    return text.startswith(_NON_AUTHOR_PREFIXES)
 
-    Example:
-        Türkiye
-        Türkiye
 
-    NFKC is also used by the figure lookup code.
-    """
-    name = Path(file_name or "").name.strip()
-    return unicodedata.normalize("NFKC", name)
+def _surname_key(name: str) -> str:
+    """Normalize an author label for surname-level deduplication."""
+    text = (name or "").strip()
+    if not text:
+        return ""
 
+    # 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()
 
-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
+    # "Surname, Initials" format
+    if "," in text:
+        return text.split(",", 1)[0].strip().lower()
 
+    # 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.
 
-    The filename is Unicode-normalized before constructing the
-    Summary/<year>/<filename> path.
-    """
-    if not HF_SUMMARY_ON_DEMAND:
-        return None
+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 []
 
-    safe_name = _normalize_filename(file_name)
+    stem = name.rsplit(".", 1)[0]
+    match = _FILENAME_AUTHORS_RE.match(stem)
+    if not match:
+        return []
 
-    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
+    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()]
 
 
-def summary_search_dirs() -> list[Path]:
-    """Directories to search for summary .txt files (first match wins)."""
-    dirs: list[Path] = []
+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 ""
 
-    for raw in (os.getenv("PAPER_SUMMARIES_DIRS") or "").split(":"):
-        raw = raw.strip()
-        if raw:
-            dirs.append(
-                Path(raw).expanduser().resolve()
-            )
+    names: list[str] = []
+    by_surname: dict[str, int] = {}
 
-    override = os.getenv("PAPER_SUMMARIES_DIR")
-    if override:
-        dirs.append(
-            Path(override).expanduser().resolve()
-        )
+    def _prefer(current: str, candidate: str) -> str:
+        # Prefer the more specific/longer representation
+        if len(candidate) > len(current):
+            return candidate
+        return current
 
-    project_root = Path(__file__).resolve().parents[1]
-    dirs.append(
-        project_root / "paper_summaries"
-    )
-
-    unique: list[Path] = []
-    seen: set[str] = set()
+    for segment in _RP_SEGMENT_RE.split(text):
+        segment = segment.strip()
 
-    for path in dirs:
-        key = str(path)
-
-        if key in seen:
+        if not segment or _is_non_author_text(segment):
             continue
 
-        seen.add(key)
-        unique.append(path)
-
-    return unique
-
-
-def summaries_dir() -> Path:
-    for path in summary_search_dirs():
-        if path.exists():
-            return path
-
-    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
-
-
-def load_manifest(
-    config: LocalRagConfig | None = None,
-) -> dict:
-    path = _manifest_path(config)
-
-    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")
-    )
-
-
-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
+        name = segment.split("(", 1)[0].strip().rstrip(",")
 
+        if not name or _is_non_author_text(name):
+            continue
 
-def extract_title(
-    file_name: str,
-    summary: str | None = None,
-) -> str:
-    if summary:
-        match = TITLE_PATTERN.search(summary)
+        key = _surname_key(name)
+        if not key:
+            continue
 
-        if match:
-            return match.group(1).strip()
+        if key in by_surname:
+            idx = by_surname[key]
+            names[idx] = _prefer(names[idx], name)
+        else:
+            by_surname[key] = len(names)
+            names.append(name)
 
-    return title_from_filename(file_name)
+    return "; ".join(names)
 
 
-def _find_normalized_local_file(
-    directory: Path,
-    file_name: str,
-) -> Path | None:
+def resolve_authors(row: dict) -> str:
     """
-    Find a locally available/cached summary using exact
-    Unicode-normalized filename matching.
+    Resolve authors using the best available metadata source.
 
-    No fuzzy or prefix matching is used.
+    Priority:
+    1. AU  - full author list from publication metadata
+    2. RP  - author information parsed from full text
+    3. PDF/TXT filename - fallback only
     """
-    if not directory.is_dir():
-        return None
-
-    safe_name = _normalize_filename(file_name)
 
-    # Fast exact lookup first.
-    candidate = directory / safe_name
+    # 1. Prefer the full author list from metadata
+    au = str(row.get("AU") or "").strip()
+    if au:
+        return au
 
-    if candidate.is_file():
-        return candidate
+    # 2. Fall back to RP
+    rp_authors = parse_authors_from_rp(row.get("RP") or "").strip()
+    if rp_authors:
+        return rp_authors
 
-    # Unicode-normalized fallback.
-    target = safe_name.casefold()
+    # 3. Final fallback: filename
+    filename_authors = parse_authors_from_filename(
+        row.get("PDF_file") or row.get("TXT_file") or ""
+    )
 
-    try:
-        for path in directory.iterdir():
+    return "; ".join(filename_authors)
 
-            if not path.is_file():
-                continue
 
-            existing_name = _normalize_filename(
-                path.name
-            )
+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 existing_name.casefold() == target:
-                return path
 
-    except OSError:
+def document_id_for_txt_file(txt_file: str) -> str | None:
+    name = (txt_file or "").strip()
+    if not name:
         return None
 
-    return None
+    manifest = load_manifest()
+    files = manifest.get("files", [])
 
+    for entry in files:
+        if entry.get("file_name") == name:
+            return entry.get("document_id")
 
-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)
+    lowered = name.lower()
+    for entry in files:
+        file_name = entry.get("file_name", "")
+        if file_name.lower() == lowered:
+            return entry.get("document_id")
 
-    # Preserve existing local behavior first.
-    for root in summary_search_dirs():
+    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")
 
-        # 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
+    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]
 
+    return None
 
-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", [])
+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,
     }
-
-    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