langchain: Replace lxml and XSLT with BeautifulSoup in HTMLHeaderTextSplitter for Improved Large HTML File Processing

libs/text-splitters/langchain_text_splitters/html.py

@@ -6,7 +6,9 @@
 from typing import Any, Dict, Iterable, List, Optional, Tuple, TypedDict, cast
 
 import requests
-from langchain_core.documents import Document
+from bs4 import BeautifulSoup
+from bs4.element import Tag
+from langchain.docstore.document import Document
 
 from langchain_text_splitters.character import RecursiveCharacterTextSplitter
 
@@ -89,80 +91,100 @@ def split_text(self, text: str) -> List[Document]:
             text: HTML text
         """
         return self.split_text_from_file(StringIO(text))
-
+        
     def split_text_from_file(self, file: Any) -> List[Document]:
         """Split HTML file.
 
         Args:
-            file: HTML file
+            file: HTML file path or file-like object.
+
+        Returns:
+            List of Document objects with page_content and metadata.
         """
-        try:
-            from lxml import etree
-        except ImportError as e:
-            raise ImportError(
-                "Unable to import lxml, please install with `pip install lxml`."
-            ) from e
-        # use lxml library to parse html document and return xml ElementTree
-        # Explicitly encoding in utf-8 allows non-English
-        # html files to be processed without garbled characters
-        parser = etree.HTMLParser(encoding="utf-8")
-        tree = etree.parse(file, parser)
-
-        # document transformation for "structure-aware" chunking is handled with xsl.
-        # see comments in html_chunks_with_headers.xslt for more detailed information.
-        xslt_path = pathlib.Path(__file__).parent / "xsl/html_chunks_with_headers.xslt"
-        xslt_tree = etree.parse(xslt_path)
-        transform = etree.XSLT(xslt_tree)
-        result = transform(tree)
-        result_dom = etree.fromstring(str(result))
+        # Read the HTML content from the file or file-like object
+        if isinstance(file, str):
+            with open(file, 'r', encoding='utf-8') as f:
+                html_content = f.read()
+        else:
+            # Assuming file is a file-like object
+            html_content = file.read()
 
-        # create filter and mapping for header metadata
-        header_filter = [header[0] for header in self.headers_to_split_on]
-        header_mapping = dict(self.headers_to_split_on)
+        # Parse the HTML content using BeautifulSoup
+        soup = BeautifulSoup(html_content, 'html.parser')
 
-        # map xhtml namespace prefix
-        ns_map = {"h": "http://www.w3.org/1999/xhtml"}
+        # Extract the header tags and their corresponding metadata keys
+        headers_to_split_on = [tag[0] for tag in self.headers_to_split_on]
+        header_mapping = dict(self.headers_to_split_on)
 
-        # build list of elements from DOM
-        elements = []
-        for element in result_dom.findall("*//*", ns_map):
-            if element.findall("*[@class='headers']") or element.findall(
-                "*[@class='chunk']"
-            ):
-                elements.append(
-                    ElementType(
-                        url=file,
-                        xpath="".join(
-                            [
-                                node.text or ""
-                                for node in element.findall("*[@class='xpath']", ns_map)
-                            ]
-                        ),
-                        content="".join(
-                            [
-                                node.text or ""
-                                for node in element.findall("*[@class='chunk']", ns_map)
-                            ]
-                        ),
-                        metadata={
-                            # Add text of specified headers to metadata using header
-                            # mapping.
-                            header_mapping[node.tag]: node.text or ""
-                            for node in filter(
-                                lambda x: x.tag in header_filter,
-                                element.findall("*[@class='headers']/*", ns_map),
-                            )
-                        },
-                    )
-                )
+        documents = []
 
-        if not self.return_each_element:
-            return self.aggregate_elements_to_chunks(elements)
+        # Find the body of the document
+        body = soup.body if soup.body else soup
+
+        # Find all header tags in the order they appear
+        all_headers = body.find_all(headers_to_split_on)
+
+        # If there's content before the first header, collect it
+        first_header = all_headers[0] if all_headers else None
+        if first_header:
+            pre_header_content = ''
+            for elem in first_header.find_all_previous():
+                if isinstance(elem, bs4.Tag):
+                    text = elem.get_text(separator=' ', strip=True)
+                    if text:
+                        pre_header_content = text + ' ' + pre_header_content
+            if pre_header_content.strip():
+                documents.append(Document(
+                    page_content=pre_header_content.strip(),
+                    metadata={}  # No metadata since there's no header
+                ))
         else:
-            return [
-                Document(page_content=chunk["content"], metadata=chunk["metadata"])
-                for chunk in elements
-            ]
+            # If no headers are found, return the whole content
+            full_text = body.get_text(separator=' ', strip=True)
+            if full_text.strip():
+                documents.append(Document(
+                    page_content=full_text.strip(),
+                    metadata={}
+                ))
+            return documents
+
+        # Process each header and its associated content
+        for header in all_headers:
+            current_metadata = {}
+            header_name = header.name
+            header_text = header.get_text(separator=' ', strip=True)
+            current_metadata[header_mapping[header_name]] = header_text
+
+            # Collect all sibling elements until the next header of the same or higher level
+            content_elements = []
+            for sibling in header.find_next_siblings():
+                if sibling.name in headers_to_split_on:
+                    # Stop at the next header
+                    break
+                if isinstance(sibling, bs4.Tag):
+                    content_elements.append(sibling)
+
+            # Get the text content of the collected elements
+            current_content = ''
+            for elem in content_elements:
+                text = elem.get_text(separator=' ', strip=True)
+                if text:
+                    current_content += text + ' '
+
+            # Create a Document if there is content
+            if current_content.strip():
+                documents.append(Document(
+                    page_content=current_content.strip(),
+                    metadata=current_metadata.copy()
+                ))
+            else:
+                # If there's no content, but we have metadata, still create a Document
+                documents.append(Document(
+                    page_content='',
+                    metadata=current_metadata.copy()
+                ))
+
+        return documents
 
 
 class HTMLSectionSplitter: