Spaces:
Configuration error
Configuration error
| import os | |
| import PyPDF2 | |
| import re | |
| from langchain_core.documents import Document | |
| from langchain_text_splitters import RecursiveCharacterTextSplitter | |
| def detect_file_type(file_path: str) -> str: | |
| """ | |
| Detect file type based on extension. | |
| Returns 'pdf' or 'text'. | |
| """ | |
| _, ext = os.path.splitext(file_path.lower()) | |
| if ext == '.pdf': | |
| return 'pdf' | |
| elif ext in ['.txt', '.text']: | |
| return 'text' | |
| else: | |
| # Default to text for unknown extensions | |
| return 'text' | |
| def get_text_title(file_path: str) -> str: | |
| """ | |
| Extract title from text file that follows the format "Title: XXXXX Abstract: XXXXX" | |
| """ | |
| try: | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| content = f.read() | |
| # Look for "Title:" pattern (case insensitive) | |
| title_match = re.search(r'title:\s*([^\n\r]+?)(?:\s+abstract:|$)', content, re.IGNORECASE) | |
| if title_match: | |
| title = title_match.group(1).strip() | |
| # Clean up the title | |
| title = re.sub(r'[^\w\s-]', '', title)[:100] | |
| if title: | |
| return title | |
| # If no "Title:" found, try to get the first non-empty line | |
| lines = content.strip().split('\n') | |
| for line in lines: | |
| line = line.strip() | |
| if line and not line.lower().startswith('abstract:'): | |
| # Remove common trailing punctuation from titles | |
| title = line.rstrip('.,;:!?') | |
| return title[:100] | |
| except Exception as e: | |
| print(f"❌ Error reading text file {file_path}: {e}") | |
| return os.path.basename(file_path) | |
| def get_pdf_title(file_path: str) -> str: | |
| """ | |
| Existing code that attempts to get a PDF title. | |
| """ | |
| try: | |
| with open(file_path, "rb") as f: | |
| reader = PyPDF2.PdfReader(f) | |
| meta = reader.metadata | |
| if meta and meta.title: | |
| title = re.sub(r'[^\w\s-]', '', meta.title.strip())[:100] | |
| return title | |
| first_page = reader.pages[0].extract_text() | |
| if first_page: | |
| return first_page.strip().split('\n')[0] | |
| except: | |
| pass | |
| return os.path.basename(file_path) | |
| def get_file_title(file_path: str) -> str: | |
| """ | |
| Get title from either PDF or text file based on file type. | |
| """ | |
| file_type = detect_file_type(file_path) | |
| if file_type == 'pdf': | |
| return get_pdf_title(file_path) | |
| else: | |
| return get_text_title(file_path) | |
| def remove_duplicate_chunks(docs): | |
| """ | |
| Your existing deduplication logic. | |
| """ | |
| seen = set() | |
| unique_docs = [] | |
| for doc in docs: | |
| key = (doc.page_content.strip(), doc.metadata.get("title", "")) | |
| if key not in seen: | |
| seen.add(key) | |
| unique_docs.append(doc) | |
| return unique_docs | |
| def extract_text_from_file(file_path: str) -> list: | |
| """ | |
| Extract text from either PDF or text file. | |
| Returns list of Document objects (typically length=1). | |
| """ | |
| docs = [] | |
| try: | |
| file_type = detect_file_type(file_path) | |
| title = get_file_title(file_path) | |
| if file_type == 'pdf': | |
| # Handle PDF files | |
| with open(file_path, "rb") as f: | |
| reader = PyPDF2.PdfReader(f) | |
| text = "\n".join([p.extract_text() or "" for p in reader.pages]) | |
| else: | |
| # Handle text files | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| text = f.read() | |
| if text.strip(): | |
| docs = [Document(page_content=text, metadata={"source": os.path.basename(file_path), "title": title, "file_type": file_type})] | |
| except Exception as e: | |
| print(f"❌ Error processing {file_path}: {e}") | |
| return docs | |
| def extract_text_from_pdf(file_path: str) -> list: | |
| """ | |
| Legacy function name for backward compatibility. | |
| Now handles both PDF and text files. | |
| """ | |
| return extract_text_from_file(file_path) | |
| def extract_and_split(file_path: str): | |
| """ | |
| Extract text from file (PDF or text) and split into chunks. | |
| """ | |
| docs = [] | |
| try: | |
| file_type = detect_file_type(file_path) | |
| title = get_file_title(file_path) | |
| if file_type == 'pdf': | |
| # Handle PDF files | |
| with open(file_path, "rb") as f: | |
| reader = PyPDF2.PdfReader(f) | |
| text = "\n".join([p.extract_text() or "" for p in reader.pages]) | |
| else: | |
| # Handle text files | |
| with open(file_path, "r", encoding="utf-8") as f: | |
| text = f.read() | |
| if not text.strip(): | |
| return [] | |
| splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) | |
| docs = splitter.create_documents([text], metadatas=[{"source": os.path.basename(file_path), "title": title, "file_type": file_type}]) | |
| docs = remove_duplicate_chunks(docs) | |
| print(f"📄 {title}: {len(docs)} unique chunks after deduplication") | |
| except Exception as e: | |
| print(f"❌ Error processing {file_path}: {e}") | |
| return docs | |
| def entity_match(dictionary: dict, target_phrase: str, window_size: int = 70) -> dict: | |
| """ | |
| Given one entity (target_phrase), return a dict: | |
| {paper_name: [list_of_snippets]} | |
| Each snippet is a window of ±`window_size` words around the matched term. | |
| """ | |
| output = {} | |
| phrase_tokens = target_phrase.lower().split() | |
| phrase_len = len(phrase_tokens) | |
| for paper_name, text in dictionary.items(): | |
| words = text.split() | |
| n = len(words) | |
| intervals = [] | |
| i = 0 | |
| while i < n: | |
| # Enough room for entire phrase | |
| if i + phrase_len <= n: | |
| slice_candidate = words[i : i + phrase_len] | |
| if [w.lower() for w in slice_candidate] == phrase_tokens: | |
| start_idx = max(0, i - window_size) | |
| end_idx = min(n, i + phrase_len + window_size) | |
| intervals.append((start_idx, end_idx)) | |
| # Advance i by 1 to allow overlapping matches | |
| i += 1 | |
| continue | |
| i += 1 | |
| # If no intervals found, just store empty list and move on | |
| if not intervals: | |
| output[paper_name] = [] | |
| continue | |
| # Sort intervals by their start index | |
| intervals.sort(key=lambda x: x[0]) | |
| # Merge overlapping intervals | |
| merged_intervals = [] | |
| current_start, current_end = intervals[0] | |
| for idx in range(1, len(intervals)): | |
| start, end = intervals[idx] | |
| if start <= current_end: # overlap | |
| current_end = max(current_end, end) | |
| else: | |
| merged_intervals.append((current_start, current_end)) | |
| current_start, current_end = start, end | |
| # Add the final interval | |
| merged_intervals.append((current_start, current_end)) | |
| # Convert merged intervals to text snippets | |
| snippets = [] | |
| for (start_i, end_i) in merged_intervals: | |
| snippet_words = words[start_i:end_i] | |
| snippet_text = " ".join(snippet_words) | |
| snippets.append(snippet_text) | |
| output[paper_name] = snippets | |
| return output |