Spaces:
Sleeping
Sleeping
File size: 3,504 Bytes
338036b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | from .BaseController import BaseController
from .ProjectController import ProjectController
import os
from langchain_community.document_loaders import TextLoader
from langchain_community.document_loaders import PyMuPDFLoader
from models import ProcessingEnum
from typing import List
from dataclasses import dataclass
@dataclass
class Documents:
page_content: str
metadata: dict
class ProcessController(BaseController):
def __init__(self, project_id: str):
super().__init__()
self.project_id = project_id
self.project_path = ProjectController().get_project_path(project_id = project_id)
def get_file_extention(self, file_id:str): # file_id => file name
return os.path.splitext(file_id)[-1]
def get_file_loader(self,file_id:str):
file_ext = self.get_file_extention(file_id = file_id)
file_path = os.path.join(
self.project_path ,
file_id
)
# Bug: processing used to blow up with a deep loader traceback when the file id
# did not exist under the given project folder.
# Fix: return `None` early so the route can handle the missing file more clearly.
if not os.path.exists(file_path):
return None
if file_ext == ProcessingEnum.TXT.value:
return TextLoader(file_path, encoding = "utf-8")
elif file_ext == ProcessingEnum.PDF.value:
return PyMuPDFLoader(file_path)
return None
def get_file_content(self, file_id: str):
file_path = os.path.join(self.project_path, file_id)
if not os.path.exists(file_path):
raise FileNotFoundError(file_path)
loader = self.get_file_loader(file_id=file_id)
if loader is None:
# Bug: unsupported extensions used to fail later in less obvious ways.
# Fix: raise a direct validation error at the controller boundary.
raise ValueError(f"Unsupported file type for file_id: {file_id}")
return loader.load()
def process_file_content(self, file_content: list, file_id:str,
chunk_size: int = 100 , overlap_size: int=20):
file_content_texts = [
rec.page_content
for rec in file_content
]
file_content_metadata = [
rec.metadata
for rec in file_content
]
chunks = self.prosecc_simpler_splitter(
texts = file_content_texts,
metadatas=file_content_metadata,
chunk_size=chunk_size
)
return chunks
def prosecc_simpler_splitter(self, texts: List[str], metadatas: List[dict], chunk_size: int, splitter_tag : str="\n"):
full_text= " ".join(texts)
lines = [ doc.strip() for doc in full_text.split(splitter_tag) if len(doc.strip())> 1]
chunks = []
current_chunk = ""
for line in lines:
current_chunk += splitter_tag + line
if len(current_chunk) >= chunk_size:
chunks.append(Documents(
page_content=current_chunk.strip(),
metadata={}
))
current_chunk = ""
if len(current_chunk) > 0:
chunks.append(Documents(
page_content=current_chunk.strip(),
metadata={}
))
return chunks |