import os from typing import List from langchain_text_splitters import RecursiveCharacterTextSplitter import shutil class DocumentProcessor: def __init__(self, upload_dir: str = "uploads"): self.upload_dir = upload_dir if not os.path.exists(self.upload_dir): os.makedirs(self.upload_dir) self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=100, length_function=len, ) async def save_upload(self, file) -> str: file_path = os.path.join(self.upload_dir, file.filename) with open(file_path, "wb") as buffer: shutil.copyfileobj(file.file, buffer) return file_path def extract_text(self, file_path: str) -> str: if file_path.endswith('.txt'): with open(file_path, 'r', encoding='utf-8', errors='ignore') as f: return f.read() return "Unsupported format." def create_chunks(self, text: str) -> List[str]: return self.text_splitter.split_text(text) processor = DocumentProcessor()