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