Blame for backend/paper_linking.py

from __future__ import annotations

import hashlib
import re
import unicodedata

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)

_NON_AUTHOR_PREFIXES = (
    "not specified",
    "not explicitly",
    "not listed",
    "not available",
    "not provided",
    "unknown",
    "co-author",
)


def _is_non_author_text(value: str) -> bool:
    text = (value or "").strip().lower()

    if not text:
        return True

    if text in {"n/a", "na", "none"}:
        return True

    return text.startswith(_NON_AUTHOR_PREFIXES)


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()

    # "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 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 []

    stem = name.rsplit(".", 1)[0]
    match = _FILENAME_AUTHORS_RE.match(stem)
    if not match:
        return []

    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 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 _prefer(current: str, candidate: str) -> str:
        # Prefer the more specific/longer representation
        if len(candidate) > len(current):
            return candidate
        return current

    for segment in _RP_SEGMENT_RE.split(text):
        segment = segment.strip()

        if not segment or _is_non_author_text(segment):
            continue

        name = segment.split("(", 1)[0].strip().rstrip(",")

        if not name or _is_non_author_text(name):
            continue

        key = _surname_key(name)
        if not key:
            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)

    return "; ".join(names)


def resolve_authors(row: dict) -> str:
    """

    Resolve authors using the best available metadata source.



    Priority:

    1. AU  - full author list from publication metadata

    2. RP  - author information parsed from full text

    3. PDF/TXT filename - fallback only

    """

    # 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

    # 3. Final fallback: filename
    filename_authors = parse_authors_from_filename(
        row.get("PDF_file") or row.get("TXT_file") or ""
    )

    return "; ".join(filename_authors)


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."


def _normalize_filename(name: str) -> str:
    """

    Normalize filenames for reliable matching across Unicode forms.



    Example:

        Türkiye

        Türkiye



    Both normalize to the same value.



    NFKC is used to stay consistent with paper_figures.py.

    """
    return unicodedata.normalize(
        "NFKC",
        (name or "").strip(),
    ).casefold()


def document_id_for_txt_file(txt_file: str) -> str | None:
    """

    Resolve a metadata TXT filename to the document_id in manifest.json.



    Matching priority:

    1. Exact filename match

    2. Unicode-normalized filename match

    3. Unicode-normalized match after adding .txt

    4. Existing summary/hash fallback

    """
    name = (txt_file or "").strip()

    if not name:
        return None

    manifest = load_manifest()
    files = manifest.get("files", [])

    # 1. Exact filename match
    for entry in files:
        if entry.get("file_name") == name:
            return entry.get("document_id")

    # 2. Unicode-normalized filename match
    normalized_name = _normalize_filename(name)

    for entry in files:
        file_name = entry.get("file_name", "")

        if _normalize_filename(file_name) == normalized_name:
            return entry.get("document_id")

    # 3. If metadata omitted ".txt", try again with the extension
    if not name.lower().endswith(".txt"):
        with_txt = f"{name}.txt"
        normalized_with_txt = _normalize_filename(with_txt)

        for entry in files:
            file_name = entry.get("file_name", "")

            if _normalize_filename(file_name) == normalized_with_txt:
                return entry.get("document_id")

    # 4. Preserve the original fallback behavior
    #
    # Only reached when no corresponding manifest entry was found.
    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 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,
    }