Spaces:
Sleeping
Sleeping
| """ | |
| PDF Text Extractor - Extracts text from PDF documents. | |
| Handles scanned PDFs, multi-column layouts, and tables. | |
| """ | |
| import os | |
| import json | |
| import time | |
| import logging | |
| from typing import Optional | |
| import pdfplumber | |
| from scraper.config import RAW_SUBDIRS | |
| logger = logging.getLogger("PDFExtractor") | |
| class PDFExtractor: | |
| """Extracts text from PDF files using pdfplumber.""" | |
| def __init__(self, output_dir: Optional[str] = None): | |
| self.pdf_dir = RAW_SUBDIRS["pdfs"] | |
| self.output_dir = output_dir or RAW_SUBDIRS["pdfs"] | |
| self.results = [] | |
| os.makedirs(self.output_dir, exist_ok=True) | |
| def extract_from_file(self, filepath: str) -> dict: | |
| """Extract text from a single PDF file.""" | |
| logger.info(f"📄 Extracting: {os.path.basename(filepath)}") | |
| try: | |
| text_parts = [] | |
| metadata = {} | |
| with pdfplumber.open(filepath) as pdf: | |
| metadata = { | |
| "num_pages": len(pdf.pages), | |
| "pdf_info": pdf.metadata or {}, | |
| } | |
| for i, page in enumerate(pdf.pages): | |
| page_text = page.extract_text() | |
| if page_text: | |
| text_parts.append(page_text) | |
| # Also try to extract tables | |
| tables = page.extract_tables() | |
| for table in tables: | |
| if table: | |
| table_text = "\n".join( | |
| " | ".join(str(cell or "") for cell in row) | |
| for row in table | |
| ) | |
| text_parts.append(f"\n[Tabela]:\n{table_text}\n") | |
| full_text = "\n\n".join(text_parts) | |
| result = { | |
| "url": f"file://{filepath}", | |
| "title": os.path.basename(filepath).replace(".pdf", ""), | |
| "text": full_text, | |
| "text_length": len(full_text), | |
| "source": "PDF Document", | |
| "num_pages": metadata.get("num_pages", 0), | |
| "pdf_metadata": metadata.get("pdf_info", {}), | |
| "crawled_at": time.strftime("%Y-%m-%dT%H:%M:%S"), | |
| } | |
| return result | |
| except Exception as e: | |
| logger.error(f"Error extracting {filepath}: {e}") | |
| return {} | |
| def extract_all(self) -> list: | |
| """Extract text from all PDFs in the pdfs directory.""" | |
| logger.info(f"🚀 Extracting all PDFs from: {self.pdf_dir}") | |
| pdf_files = [ | |
| os.path.join(self.pdf_dir, f) | |
| for f in os.listdir(self.pdf_dir) | |
| if f.lower().endswith(".pdf") | |
| ] | |
| if not pdf_files: | |
| logger.info("No PDF files found.") | |
| return [] | |
| for filepath in pdf_files: | |
| result = self.extract_from_file(filepath) | |
| if result and result.get("text_length", 0) > 50: | |
| # Save JSON alongside PDF | |
| json_path = filepath.replace(".pdf", ".json") | |
| with open(json_path, "w", encoding="utf-8") as f: | |
| json.dump(result, f, ensure_ascii=False, indent=2) | |
| self.results.append(result) | |
| logger.info(f" ✅ {result['title']} ({result['num_pages']} pages, {result['text_length']} chars)") | |
| else: | |
| logger.info(f" ⏭️ Skipped (no text): {os.path.basename(filepath)}") | |
| logger.info(f"✅ PDF extraction complete: {len(self.results)} documents") | |
| return self.results | |
| def get_stats(self) -> dict: | |
| return { | |
| "source": "PDFs", | |
| "documents": len(self.results), | |
| "total_pages": sum(r.get("num_pages", 0) for r in self.results), | |
| "total_chars": sum(r.get("text_length", 0) for r in self.results), | |
| } | |