| 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() | |