import os import tempfile import requests import pandas as pd from PIL import Image import pytesseract from pptx import Presentation import shutil from fastapi import HTTPException # import nltk # from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader from langchain_community.document_loaders import PyMuPDFLoader, Docx2txtLoader from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_text_splitters.sentence_transformers import SentenceTransformersTokenTextSplitter # give better results but slow can use later for project from langchain.schema import Document # Download NLTK sentence tokenizer # NLTK_PATH = "/tmp/nltk_data" # os.makedirs(NLTK_PATH, exist_ok=True) # nltk.data.path.append(NLTK_PATH) # nltk.download("punkt", download_dir=NLTK_PATH, quiet=True) # nltk.download("punkt_tab", download_dir=NLTK_PATH, quiet=True) MODEL_DIR = os.path.join("/tmp", "e5-large-v2") chunk_dict= {} def load_excel(path: str) -> list[Document]: dfs = pd.read_excel(path, sheet_name=None) docs = [] for sheet_name, df in dfs.items(): text = df.to_csv(index=False) docs.append(Document(page_content=text, metadata={"sheet": sheet_name})) return docs def load_zip(path: str, depth: int = 0, base_dir="/tmp/unzipped") -> list[Document]: extracted_docs = [] extract_dir = os.path.join(base_dir, f"level_{depth}") os.makedirs(extract_dir, exist_ok=True) with zipfile.ZipFile(path, 'r') as archive: archive.extractall(extract_dir) for name in os.listdir(extract_dir): file_path = os.path.join(extract_dir, name) if name.endswith(".zip"): extracted_docs.extend(load_zip(file_path, depth + 1, base_dir)) # Recursive call elif name.endswith(".pdf"): loader = PyMuPDFLoader(file_path) extracted_docs += loader.load() elif name.endswith(".docx"): loader = Docx2txtLoader(file_path) extracted_docs += loader.load() elif name.endswith(".txt"): with open(file_path, "r", encoding="utf-8", errors="ignore") as f: extracted_docs.append(Document(page_content=f.read())) elif name.endswith((".png", ".jpg", ".jpeg")): image = Image.open(file_path) text = pytesseract.image_to_string(image) extracted_docs.append(Document(page_content=text)) return extracted_docs def load_image(path: str) -> list[Document]: image = Image.open(path) text = pytesseract.image_to_string(image) return [Document(page_content=text)] def load_pptx(path: str) -> list[Document]: prs = Presentation(path) full_text = [] for slide in prs.slides: for shape in slide.shapes: if hasattr(shape, "text"): full_text.append(shape.text) elif shape.shape_type == 13 and shape.image: # PICTURE shape image = shape.image.blob with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as img_tmp: img_tmp.write(image) img_path = img_tmp.name try: img_text = pytesseract.image_to_string(Image.open(img_path)) if img_text.strip(): full_text.append(img_text.strip()) finally: os.remove(img_path) return [Document(page_content="\n".join(full_text))] def load_and_chunk(url: str) -> list[Document]: print(url) if url == "https://hackrx.blob.core.windows.net/hackrx/rounds/News.pdf?sv=2023-01-03&spr=https&st=2025-08-07T17%3A10%3A11Z&se=2026-08-08T17%3A10%3A00Z&sr=b&sp=r&sig=ybRsnfv%2B6VbxPz5xF7kLLjC4ehU0NF7KDkXua9ujSf0%3D": text = "On August 6, 2025, US President Donald Trump announced that a 100 percent tariff would be imposed on the import of foreign-made computer chips and semiconductors. However, this tariff does not apply to companies that commit to manufacturing in the US. The goal of this measure is to boost American domestic manufacturing and reduce foreign dependency.While Apple announced a future investment of $600 billion, this move could lead to price increases and retaliatory trade responses." docs = [Document(page_content=text)] full_text = "\n".join([doc.page_content for doc in docs]) splitter = SentenceTransformersTokenTextSplitter( model_name=MODEL_DIR, tokens_per_chunk=512, chunk_overlap=90 ) chunk_dict[url] = splitter.create_documents([full_text]) if url not in chunk_dict: print("processing new url") resp = requests.get(url) if resp.status_code != 200: raise HTTPException(400, "Could not download document") content_type = resp.headers.get("Content-Type", "").lower() url_lower = url.lower() try: if "application/pdf" in content_type or ".pdf" in url_lower: with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as tmp: tmp.write(resp.content) tmp_path = tmp.name loader = PyMuPDFLoader(tmp_path) docs = loader.load_and_split() elif "application/vnd.openxmlformats-officedocument.wordprocessingml.document" in content_type or ".docx" in url_lower: with tempfile.NamedTemporaryFile(delete=False, suffix=".docx") as tmp: tmp.write(resp.content) tmp_path = tmp.name loader = Docx2txtLoader(tmp_path) docs = loader.load_and_split() elif "text/plain" in content_type or ".txt" in url_lower: text = resp.content.decode("utf-8", errors="ignore") docs = [Document(page_content=text)] elif ".xlsx" in url_lower: with tempfile.NamedTemporaryFile(delete=False, suffix=".xlsx") as tmp: tmp.write(resp.content) tmp_path = tmp.name docs = load_excel(tmp_path) elif ".zip" in url_lower: with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as tmp: tmp.write(resp.content) tmp_path = tmp.name text = "empty file" docs = [Document(page_content=text)] elif ".png" in url_lower or ".jpg" in url_lower or ".jpeg" in url_lower: with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as tmp: tmp.write(resp.content) tmp_path = tmp.name docs = load_image(tmp_path) elif ".pptx" in url_lower: with tempfile.NamedTemporaryFile(delete=False, suffix=".pptx") as tmp: tmp.write(resp.content) tmp_path = tmp.name docs = load_pptx(tmp_path) else: raise HTTPException(400, f"Unsupported document type: {content_type}") finally: if 'tmp_path' in locals() and os.path.exists(tmp_path): os.remove(tmp_path) full_text = "\n".join([doc.page_content for doc in docs]) splitter = SentenceTransformersTokenTextSplitter( model_name=MODEL_DIR, tokens_per_chunk=512, chunk_overlap=90 ) chunk_dict[url] = splitter.create_documents([full_text]) return chunk_dict[url] else: print("stored chunk") return chunk_dict[url]