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, }