Spaces:
Sleeping
Sleeping
| """ | |
| PDF Parser — Entry point for every paper analysis. | |
| Every analysis we do depends on clean text extraction. | |
| If this is wrong, everything downstream is wrong. | |
| SciPeerAI v2.3.1 — Built by Sameer Nadeem | |
| Table extraction + DOI harvesting added. | |
| """ | |
| import re | |
| import hashlib | |
| import fitz | |
| from dataclasses import dataclass | |
| from pathlib import Path | |
| MAX_FILE_SIZE_MB = 50 | |
| MAX_PAGES = 300 | |
| ALLOWED_MIME_HEADER = b"%PDF" | |
| class ParsedPaper: | |
| title: str | |
| full_text: str | |
| sections: dict | |
| page_count: int | |
| has_figures: bool | |
| figure_count: int | |
| metadata: dict | |
| tables_text: str = "" | |
| dois_found: list = None | |
| def __post_init__(self): | |
| if self.dois_found is None: | |
| self.dois_found = [] | |
| class PDFParser: | |
| """ | |
| Handles PDF ingestion and structured text extraction. | |
| v2.3.1 upgrades: | |
| - Table extraction via PyMuPDF blocks | |
| - Structured number extraction for GRIM/SPRITE/P-Curve | |
| - DOI harvesting from full document | |
| - Section-aware extraction | |
| """ | |
| def __init__(self): | |
| self._section_markers = [ | |
| "abstract", "introduction", "methods", "methodology", | |
| "results", "discussion", "conclusion", "references", | |
| "related work", "background", "experiments", | |
| "materials and methods", "statistical analysis", | |
| "data analysis", "findings", "participants", | |
| ] | |
| self._doi_pattern = re.compile( | |
| r'10\.\d{4,9}/[^\s\],;"\'<>\|\{\}\\]+', | |
| re.IGNORECASE | |
| ) | |
| def parse(self, pdf_path: str) -> ParsedPaper: | |
| pdf_path = Path(pdf_path) | |
| if not pdf_path.exists(): | |
| raise FileNotFoundError(f"Paper not found: {pdf_path}") | |
| if pdf_path.suffix.lower() != ".pdf": | |
| raise ValueError(f"Expected PDF file, got: {pdf_path.suffix}") | |
| raw_bytes = pdf_path.read_bytes() | |
| return self.parse_bytes(raw_bytes, filename=pdf_path.name) | |
| def parse_bytes(self, file_bytes: bytes, filename: str = "upload.pdf") -> ParsedPaper: | |
| filename = self._sanitize_filename(filename) | |
| self._validate_bytes(file_bytes, filename) | |
| doc = fitz.open(stream=file_bytes, filetype="pdf") | |
| if len(doc) > MAX_PAGES: | |
| doc.close() | |
| raise ValueError( | |
| f"Paper has {len(doc)} pages. Maximum allowed is {MAX_PAGES}." | |
| ) | |
| full_text = self._extract_text(doc) | |
| tables_text = self._extract_tables(doc) | |
| combined = full_text + "\n\n" + tables_text | |
| sections = self._split_into_sections(combined) | |
| figure_count = self._count_figures(doc) | |
| title = self._extract_title(doc, full_text) | |
| page_count = len(doc) | |
| dois_found = self._extract_dois(combined) | |
| doc.close() | |
| return ParsedPaper( | |
| title = title, | |
| full_text = combined, | |
| sections = sections, | |
| page_count = page_count, | |
| has_figures = figure_count > 0, | |
| figure_count = figure_count, | |
| dois_found = dois_found, | |
| tables_text = tables_text, | |
| metadata = { | |
| "filename": filename, | |
| "file_size_kb": round(len(file_bytes) / 1024, 2), | |
| "sha256": hashlib.sha256(file_bytes).hexdigest(), | |
| }, | |
| ) | |
| def _validate_bytes(self, file_bytes: bytes, filename: str) -> None: | |
| if len(file_bytes) == 0: | |
| raise ValueError("Uploaded file is empty.") | |
| max_bytes = MAX_FILE_SIZE_MB * 1024 * 1024 | |
| if len(file_bytes) > max_bytes: | |
| size_mb = round(len(file_bytes) / 1024 / 1024, 1) | |
| raise ValueError( | |
| f"File too large: {size_mb} MB. Maximum: {MAX_FILE_SIZE_MB} MB." | |
| ) | |
| if not file_bytes.startswith(ALLOWED_MIME_HEADER): | |
| raise ValueError("Invalid file. Only real PDF files are accepted.") | |
| def _sanitize_filename(filename: str) -> str: | |
| name = Path(filename).name | |
| if not name.lower().endswith(".pdf"): | |
| raise ValueError(f"Expected a PDF filename, got: {filename}") | |
| return name | |
| def _extract_text(self, doc: fitz.Document) -> str: | |
| pages = [] | |
| for page in doc: | |
| pages.append(page.get_text("text")) | |
| return "\n".join(pages) | |
| def _extract_tables(self, doc: fitz.Document) -> str: | |
| """ | |
| Extract structured table content from PDF using block-level parsing. | |
| This is critical for GRIM/SPRITE/P-Curve modules to find | |
| means, SDs, sample sizes, and p-values that live inside tables. | |
| """ | |
| table_parts = [] | |
| for page_num, page in enumerate(doc): | |
| blocks = page.get_text("blocks") | |
| rows = [] | |
| for block in blocks: | |
| if block[6] != 0: | |
| continue | |
| text = block[4].strip() | |
| if not text: | |
| continue | |
| rows.append((block[1], text)) | |
| rows.sort(key=lambda x: x[0]) | |
| for _, text in rows: | |
| if any(char.isdigit() for char in text): | |
| clean = " ".join(text.split()) | |
| table_parts.append(clean) | |
| if not table_parts: | |
| return "" | |
| return "\nTABLE_DATA:\n" + "\n".join(table_parts) | |
| def _split_into_sections(self, text: str) -> dict: | |
| sections = {} | |
| text_lower = text.lower() | |
| for i, marker in enumerate(self._section_markers): | |
| start_idx = text_lower.find(marker) | |
| if start_idx == -1: | |
| continue | |
| end_idx = len(text) | |
| for next_marker in self._section_markers[i + 1:]: | |
| next_idx = text_lower.find(next_marker, start_idx + 1) | |
| if next_idx != -1: | |
| end_idx = next_idx | |
| break | |
| sections[marker] = text[start_idx:end_idx].strip() | |
| return sections | |
| def _count_figures(self, doc: fitz.Document) -> int: | |
| total = 0 | |
| for page in doc: | |
| total += len(page.get_images()) | |
| return total | |
| def _extract_title(self, doc: fitz.Document, full_text: str) -> str: | |
| meta = doc.metadata | |
| if meta and meta.get("title"): | |
| return meta["title"].strip() | |
| for line in full_text.split("\n"): | |
| line = line.strip() | |
| if len(line) > 10: | |
| return line | |
| return "Unknown Title" | |
| def _extract_dois(self, text: str) -> list: | |
| """ | |
| Extract all DOIs from full document text including tables and headers. | |
| Uses broad pattern to catch DOIs in any format. | |
| """ | |
| dois = [] | |
| for m in self._doi_pattern.finditer(text): | |
| doi = m.group(0).rstrip('.,;)') | |
| doi_clean = doi.lower() | |
| if doi_clean not in dois: | |
| dois.append(doi_clean) | |
| return dois[:50] |