app/ingestion/html_parser.py

from typing import List
from bs4 import BeautifulSoup

from app.ingestion.base_parser import BaseParser
from app.schemas.rich_content_block import RichContentBlock


class HtmlParser(BaseParser):

    def parse(
        self,
        file_path: str,
        document_id: str,
        source_file_name: str
    ) -> List[RichContentBlock]:

        with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
            html = f.read()

        soup = BeautifulSoup(html, "lxml")

        for tag in soup(["script", "style", "noscript", "svg"]):
            tag.decompose()

        blocks = []
        block_counter = 1
        current_section_title = None

        for element in soup.find_all(["h1", "h2", "h3", "h4", "p", "li", "blockquote", "pre", "code", "table"]):
            tag_name = element.name.lower()

            if tag_name in ["h1", "h2", "h3", "h4"]:
                text = clean_text(element.get_text(" ", strip=True))
                current_section_title = text
                content_type = "text"

            elif tag_name == "table":
                rows = []
                for tr in element.find_all("tr"):
                    cells = [
                        clean_text(cell.get_text(" ", strip=True))
                        for cell in tr.find_all(["th", "td"])
                    ]
                    if cells:
                        rows.append(cells)

                text = rows_to_markdown(rows)
                content_type = "table"

            elif tag_name in ["pre", "code"]:
                text = element.get_text("\n", strip=True)
                content_type = "code"

            else:
                text = clean_text(element.get_text(" ", strip=True))
                content_type = "text"

            if not text:
                continue

            blocks.append(
                RichContentBlock(
                    block_id=f"{document_id}_html_block_{block_counter}",
                    document_id=document_id,
                    content_type=content_type,
                    content=text,
                    page_number=None,
                    section_title=current_section_title,
                    source_file_name=source_file_name,
                    metadata={
                        "parser": "HtmlParser",
                        "original_format": "html",
                        "html_tag": tag_name
                    }
                )
            )
            block_counter += 1

        return blocks


def clean_text(text: str) -> str:
    return " ".join(text.split())


def rows_to_markdown(rows):
    if not rows:
        return ""

    header = rows[0]
    body = rows[1:]

    lines = []
    lines.append("| " + " | ".join(header) + " |")
    lines.append("| " + " | ".join(["---"] * len(header)) + " |")

    for row in body:
        if len(row) < len(header):
            row = row + [""] * (len(header) - len(row))
        if len(row) > len(header):
            row = row[:len(header)]
        lines.append("| " + " | ".join(row) + " |")

    return "\n".join(lines)