import os import tempfile import requests import pandas as pd from PIL import Image import pytesseract from pptx import Presentation import shutil from langdetect import detect from fastapi import HTTPException 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 from langchain_google_genai import ChatGoogleGenerativeAI # directory for saved model for SentenceTransformersTokenTextSplitter MODEL_DIR = os.path.join("/tmp", "e5-large-v2") # for storing chunk (saving timeeeeeeeeeeee) chunk_dict= {} GOOGLE_API_KEY2 = os.getenv("gemini_api_key2") llm = ChatGoogleGenerativeAI( model="gemini-2.0-flash", api_key=GOOGLE_API_KEY2, ) 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 """ I am not using zip further as no questions were asked and also the nested zip has no data . i did not able to submit this coded approaches for differnet types of files due to sudden change of level. do not delete it , may be all levels will reopened on last day""" 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 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) min_len = 100 if len(docs[0].page_content) >= 100 else len(docs[0].page_content) lang = detect(docs[0].page_content[:min_len]) text = "" if lang != "en": for doc in docs: translated = llm.invoke(f"Translate this into English:\n{doc.page_content}") text = text + translated.content else: full_text = "\n".join([doc.page_content for doc in docs]) splitter = SentenceTransformersTokenTextSplitter( model_name=MODEL_DIR, tokens_per_chunk=512, chunk_overlap=90 ) if lang !="en": chunk_dict[url] = splitter.create_documents([text]) else: chunk_dict[url] = splitter.create_documents([full_text]) return chunk_dict[url] else: print("stored chunk") return chunk_dict[url]