community: Huridocs pdf loader more functionality added

libs/community/langchain_community/document_loaders/pdf.py

@@ -945,5 +945,144 @@
         yield from self.parser.parse(blob)
 
 
+class HuridocsPDFLoader(BasePDFLoader):
+    """Load a PDF with Huridocs"""
+
+    def __init__(
+        self,
+        file_path: str,
+        server_url: str,
+        fast: Optional[bool] = False,
+    ) -> None:
+        """
+        Initialize the object for PDF file processing with
+        Huridocs pdf-document-layout-analysis.
+
+        This constructor initializes a HuridocsPDFLoader object to be used
+        for parsing files using the pdf-document-layout-analysis API.
+        Loader uses VGT layout model.
+        Parameters:
+        -----------
+        file_path : str
+            The path to the file that needs to be parsed.
+        server_url: str
+            The path to pdf-document-layout-analysis self-hosted API server.
+
+        Types of the Segments:
+        ---------
+        1: "Caption"
+        2: "Footnote"
+        3: "Formula"
+        4: "List item"
+        5: "Page footer"
+        6: "Page header"
+        7: "Picture"
+        8: "Section header"
+        9: "Table"
+        10: "Text"
+        11: "Title"
+
+
+        Examples:
+        ---------
+        >>> pdf_loader = HuridocsPDFLoader(
+        ...     file_path="path/to/file",
+        ...     server_url="path/to/sef-hosted/api"
+        ... )
+
+        pdf_analysis = pdf_loader.analyze_pdf()
+        table_of_contents = pdf_loader.get_table_of_contents()
+        pdf_loader.get_visualization(/path/to/output/pdf)
+        pdf_content = pdf_loader.get_text()
+        """
+        self.server_url = server_url
+        self.fast = fast
+
+        try:
+            response = requests.get(self.server_url)
+            response.raise_for_status()
+        except requests.exceptions.HTTPError as err:
+            raise err
+
+        super().__init__(file_path)
+
+    def analyze_pdf(self) -> str:
+        with open(self.file_path, "rb") as f:
+            files = {"file": f}
+            try:
+                data = {"fast": self.fast}
+                response = requests.post(f"{self.server_url}/", files=files, data=data)
+                response.raise_for_status()
+            except requests.exceptions.HTTPError as err:
+                raise err
+
+        response_data = response.json()
+
+        return response_data
+
+    def get_table_of_contents(self) -> str:
+        with open(self.file_path, "rb") as f:
+            files = {"file": f}
+            try:
+                data = {"fast": self.fast}
+                response = requests.post(f"{self.server_url}/toc", files=files, data=data)
+                response.raise_for_status()
+            except requests.exceptions.HTTPError as err:
+                raise err
+
+        response_data = response.json()
+
+        return response_data
+
+    def get_visualization(self, output_destination_path: str):
+        with open(self.file_path, "rb") as f:
+            files = {"file": f}
+            try:
+                data = {"fast": self.fast}
+                response = requests.post(f"{self.server_url}/visualize", files=files, data=data)
+                response.raise_for_status()
+                with open(output_destination_path, "wb") as file:
+                    for chunk in response.iter_content(chunk_size=8192):
+                        file.write(chunk)
+
+            except requests.exceptions.HTTPError as err:
+                raise err
+
+    def get_text(self, types: str = "all") -> str:
+        with open(self.file_path, "rb") as f:
+            files = {"file": f}
+            try:
+                data = {"fast": self.fast, "types": types}
+                response = requests.post(f"{self.server_url}/text", files=files, data=data)
+                response.raise_for_status()
+            except requests.exceptions.HTTPError as err:
+                raise err
+
+        response_data = response.json()
+
+        return response_data
+
+    def load(self) -> List[Document]:
+        """Load data into Document objects."""
+        return list(self.lazy_load())
+
+    def lazy_load(
+        self,
+    ) -> Iterator[Document]:
+        """Lazy load given path as pages."""
+        elements = self.analyze_pdf()
+
+        for el in elements:
+            yield Document(
+                page_content=el["text"],
+                metadata={
+                    "coordinates": (el["left"], el["top"], el["width"], el["height"]),
+                    "page_number": el["page_number"],
+                    "page_width": el["page_width"],
+                    "page_height": el["page_height"],
+                    "type": el["type"],
+                },
+            )
+
 # Legacy: only for backwards compatibility. Use PyPDFLoader instead
 PagedPDFSplitter = PyPDFLoader