Spaces:
Runtime error
Runtime error
| import os | |
| import re | |
| import fitz # PyMuPDF | |
| import datetime | |
| from langchain_text_splitters import RecursiveCharacterTextSplitter | |
| from ..core.config import settings | |
| def clean_text_block(text: str) -> str: | |
| if not text: | |
| return "" | |
| # Normalize unicode spaces and control characters | |
| text = text.replace('\xa0', ' ') | |
| text = text.replace('\u200b', '') | |
| lines = [] | |
| for line in text.splitlines(): | |
| cleaned = line.strip() | |
| if not cleaned: | |
| continue | |
| # Replace 3 or more repeating divider symbols inside lines with a single one (e.g. ------ to -) | |
| cleaned = re.sub(r'([=\-_*#\.\~\+\|\\/\u2014])\1{2,}', r'\1', cleaned) | |
| cleaned = cleaned.strip() | |
| # Skip lines that collapse to a single divider symbol | |
| if cleaned in ['=', '-', '_', '*', '#', '.', '~', '+', '|', '\\', '/', '\u2014']: | |
| continue | |
| # Remove spaces to check if it's a spacer pattern like ". . . ." or "- - - -" | |
| no_spaces = re.sub(r'\s+', '', cleaned) | |
| if re.match(r'^[=\-_*#\.\~\+\|\\/\u2014]{2,}$', no_spaces): | |
| continue | |
| # Skip page reference headers and footers | |
| if re.match(r'^(page\s*\d+|\d+\s*of\s*\d+)$', cleaned, re.IGNORECASE): | |
| continue | |
| # Collapse tabs and consecutive spaces | |
| cleaned = re.sub(r'[ \t]+', ' ', cleaned) | |
| if cleaned: | |
| lines.append(cleaned) | |
| return "\n".join(lines) | |
| def process_pdf(pdf_path: str, user_id: str, original_filename: str, session_id: str = None): | |
| """ | |
| Extracts text from a PDF file, ignoring margins, and returns chunks with metadata. | |
| """ | |
| parent_splitter = RecursiveCharacterTextSplitter( | |
| chunk_size=settings.PARENT_CHUNK_SIZE, | |
| chunk_overlap=settings.PARENT_CHUNK_OVERLAP, | |
| separators=["\n\n", "\n"] | |
| ) | |
| child_splitter = RecursiveCharacterTextSplitter( | |
| chunk_size=settings.CHILD_CHUNK_SIZE, | |
| chunk_overlap=settings.CHILD_CHUNK_OVERLAP, | |
| separators=["\n", ". ", " ", ""] | |
| ) | |
| chunks_with_metadata = [] | |
| chunk_idx = 0 | |
| ingestion_time = datetime.datetime.now().isoformat() | |
| try: | |
| with fitz.open(pdf_path) as pdf: | |
| total_pages = len(pdf) | |
| for page_num, page in enumerate(pdf, start=1): | |
| page_rect = page.rect | |
| footer_threshold = page_rect.height * 0.97 | |
| header_threshold = page_rect.height * 0.03 | |
| blocks = page.get_text("blocks") | |
| blocks.sort(key=lambda b: (b[1], b[0])) | |
| page_text = "" | |
| for block in blocks: | |
| if block[1] > header_threshold and block[3] < footer_threshold: | |
| if block[6] == 0: # text block | |
| cleaned_block = clean_text_block(block[4]) | |
| if cleaned_block: | |
| page_text += cleaned_block + "\n\n" | |
| if not page_text.strip(): | |
| raw_text = page.get_text("text") | |
| page_text = clean_text_block(raw_text) | |
| if not page_text.strip(): | |
| continue | |
| parent_chunks = parent_splitter.split_text(page_text) | |
| for p_chunk in parent_chunks: | |
| child_chunks = child_splitter.split_text(p_chunk) | |
| for c_chunk in child_chunks: | |
| chunk_idx += 1 | |
| metadata = { | |
| "page": page_num, | |
| "total_pages": total_pages, | |
| "chunk_number": chunk_idx, | |
| "user_id": str(user_id), | |
| "filename": original_filename, | |
| "parent_text": p_chunk, | |
| "ingestion_timestamp": ingestion_time, | |
| "embedding_model": settings.EMBEDDING_MODEL_NAME, | |
| "chunk_size": settings.CHILD_CHUNK_SIZE, | |
| "chunk_overlap": settings.CHILD_CHUNK_OVERLAP, | |
| "file_type": "pdf" | |
| } | |
| if session_id: | |
| metadata["session_id"] = session_id | |
| chunks_with_metadata.append({ | |
| "text": c_chunk, | |
| "metadata": metadata | |
| }) | |
| return chunks_with_metadata | |
| except Exception as e: | |
| print(f"Error processing PDF {pdf_path}: {e}") | |
| return [] | |