Spaces:
Sleeping
Sleeping
| 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, | |
| } |