"""Reload a persisted paper module (papers//) back into typed objects. Lets per-stage CLI commands (enrich/discover/study/review/export) operate on an already-generated module without re-ingesting the source. Parsing is best-effort and tolerant of missing files. """ from __future__ import annotations import json import re from pathlib import Path from researchlink.schemas.paper import PaperExtraction, PaperMetadata from researchlink.services.paper_module import spec_filename def _field(fields: dict, name: str): entry = fields.get(name) or {} return entry.get("value") if isinstance(entry, dict) else None def load_metadata(module_dir: Path) -> PaperMetadata: """Reconstruct PaperMetadata from metadata.json (resolved field values).""" meta_path = module_dir / "metadata.json" meta = PaperMetadata(slug=module_dir.name, id=module_dir.name) if not meta_path.exists(): return meta data = json.loads(meta_path.read_text(encoding="utf-8")) fields = data.get("fields", {}) meta.slug = data.get("slug", module_dir.name) meta.id = meta.slug meta.title = _field(fields, "title") or "Unknown Title" meta.year = _field(fields, "year") meta.venue = _field(fields, "venue") meta.doi = _field(fields, "doi") meta.arxiv_id = _field(fields, "arxiv_id") meta.paper_url = _field(fields, "url") meta.code_url = _field(fields, "code_url") meta.authors_provisional = _field(fields, "authors") or [] return meta def load_references(module_dir: Path) -> list[str]: """Parse reference strings back from references.md (``N. text`` lines).""" refs_path = module_dir / "references.md" if not refs_path.exists(): return [] refs: list[str] = [] for line in refs_path.read_text(encoding="utf-8", errors="replace").splitlines(): m = re.match(r"\s*\d+\.\s+(.*)", line) if m: # strip trailing status markers like " ⚠️ `needs-verification`" refs.append(re.sub(r"\s*⚠️.*$", "", m.group(1)).strip()) return refs def load_extraction(module_dir: Path) -> PaperExtraction: """Best-effort PaperExtraction from paper.md + references.md.""" ext = PaperExtraction(references_raw=load_references(module_dir)) paper_path = module_dir / "paper.md" if not paper_path.exists(): return ext text = paper_path.read_text(encoding="utf-8", errors="replace") abstract = _section(text, "Abstract") if abstract and "no abstract extracted" not in abstract: ext.abstract = abstract.strip() headings_block = _section(text, "Section Headings") if headings_block: ext.section_headings = [ m.group(1).strip() for line in headings_block.splitlines() if (m := re.match(r"-\s+(.*)", line)) and "no section headings" not in line ] fence = re.search(r"## Full Extracted Text\s*\n```text\n(.*?)\n```", text, re.DOTALL) if fence: ext.full_text = fence.group(1).strip() return ext def _section(text: str, heading: str) -> str | None: """Return the body under ``## `` up to the next ``## ``.""" m = re.search(rf"##\s+{re.escape(heading)}\s*\n(.*?)(?:\n##\s|\Z)", text, re.DOTALL) return m.group(1).strip() if m else None def write_module_file(module_dir: Path, name: str, content: str) -> Path: """Write content into the module under its spec filename. Returns the path.""" dest = module_dir / spec_filename(name) dest.parent.mkdir(parents=True, exist_ok=True) dest.write_text(content, encoding="utf-8") return dest