_PARA_BREAKERS = { "address", "article", "aside", "blockquote", "div", "dl", "fieldset", "figcaption", "figure", "footer", "form", "h1", "h2", "h3", "h4", "h5", "h6", "header", "hr", "main", "nav", "ol", "ul", "li", "p", #
inside
also forces closing "pre", "section", "table", "thead", "tbody", "tfoot", "tr", "td", } _CODE_TAG_SET: Final = {"code", "kbd", "samp"} _FORMAT_TAG_MAP: Final = { "b": {"bold": True}, "strong": {"bold": True}, "i": {"italic": True}, "em": {"italic": True}, "var": {"italic": True}, # "mark", # "small", "s": {"strikethrough": True}, "del": {"strikethrough": True}, "u": {"underline": True}, "ins": {"underline": True}, "sub": {"script": Script.SUB}, "sup": {"script": Script.SUPER}, **{k: {} for k in _CODE_TAG_SET}, } class _Context(BaseModel): list_ordered_flag_by_ref: dict[str, bool] = {} list_start_by_ref: dict[str, int] = {} class AnnotatedText(BaseModel): text: str hyperlink: Union[AnyUrl, Path, None] = None formatting: Union[Formatting, None] = None code: bool = False class AnnotatedTextList(list): def to_single_text_element(self) -> AnnotatedText: current_h = None current_text = "" current_f = None current_code = False for at in self: t = at.text h = at.hyperlink f = at.formatting c = at.code current_text += t.strip() + " " if f is not None and current_f is None: current_f = f elif f is not None and current_f is not None and f != current_f: _log.warning( f"Clashing formatting: '{f}' and '{current_f}'! Chose '{current_f}'" ) if h is not None and current_h is None: current_h = h elif h is not None and current_h is not None and h != current_h: _log.warning( f"Clashing hyperlinks: '{h}' and '{current_h}'! Chose '{current_h}'" ) current_code = c if c else current_code return AnnotatedText( text=current_text.strip(), hyperlink=current_h, formatting=current_f, code=current_code, ) def simplify_text_elements(self) -> "AnnotatedTextList": simplified = AnnotatedTextList() if not self: return self text = self[0].text hyperlink = self[0].hyperlink formatting = self[0].formatting code = self[0].code last_elm = text for i in range(1, len(self)): if ( hyperlink == self[i].hyperlink and formatting == self[i].formatting and code == self[i].code ): sep = " " if not self[i].text.strip() or not last_elm.strip(): sep = "" text += sep + self[i].text last_elm = self[i].text else: simplified.append( AnnotatedText( text=text, hyperlink=hyperlink, formatting=formatting, code=code ) ) text = self[i].text last_elm = text hyperlink = self[i].hyperlink formatting = self[i].formatting code = self[i].code if text: simplified.append( AnnotatedText( text=text, hyperlink=hyperlink, formatting=formatting, code=code ) ) return simplified def split_by_newline(self): super_list = [] active_annotated_text_list = AnnotatedTextList() for el in self: sub_texts = el.text.split("\n") if len(sub_texts) == 1: active_annotated_text_list.append(el) else: for text in sub_texts: sub_el = deepcopy(el) sub_el.text = text active_annotated_text_list.append(sub_el) super_list.append(active_annotated_text_list) active_annotated_text_list = AnnotatedTextList() if active_annotated_text_list: super_list.append(active_annotated_text_list) return super_list class HTMLDocumentBackend(DeclarativeDocumentBackend): @override def __init__( self, in_doc: InputDocument, path_or_stream: Union[BytesIO, Path], options: HTMLBackendOptions = HTMLBackendOptions(), ): super().__init__(in_doc, path_or_stream, options) self.options: HTMLBackendOptions self.soup: Optional[BeautifulSoup] = None self.path_or_stream: Union[BytesIO, Path] = path_or_stream self.base_path: Optional[str] = str(options.source_uri) # Initialize the parents for the hierarchy self.max_levels = 10 self.level = 0 self.parents: dict[int, Optional[Union[DocItem, GroupItem]]] = {} self.ctx = _Context() for i in range(self.max_levels): self.parents[i] = None self.hyperlink: Union[AnyUrl, Path, None] = None self.format_tags: list[str] = [] try: raw = ( path_or_stream.getvalue() if isinstance(path_or_stream, BytesIO) else Path(path_or_stream).read_bytes() ) self.soup = BeautifulSoup(raw, "html.parser") except Exception as e: raise RuntimeError( "Could not initialize HTML backend for file with " f"hash {self.document_hash}." ) from e @override def is_valid(self) -> bool: return self.soup is not None @classmethod @override def supports_pagination(cls) -> bool: return False @override def unload(self): if isinstance(self.path_or_stream, BytesIO): self.path_or_stream.close() self.path_or_stream = None @classmethod @override def supported_formats(cls) -> set[InputFormat]: return {InputFormat.HTML} @override def convert(self) -> DoclingDocument: _log.debug("Starting HTML conversion...") if not self.is_valid(): raise RuntimeError("Invalid HTML document.") origin = DocumentOrigin( filename=self.file.name or "file", mimetype="text/html", binary_hash=self.document_hash, ) doc = DoclingDocument(name=self.file.stem or "file", origin=origin) assert self.soup is not None # set the title as furniture, since it is part of the document metadata title = self.soup.title if title and self.options.add_title: title_text = title.get_text(separator=" ", strip=True) title_clean = HTMLDocumentBackend._clean_unicode(title_text) doc.add_title( text=title_clean, orig=title_text, content_layer=ContentLayer.FURNITURE, ) # remove script and style tags for tag in self.soup(["script", "noscript", "style"]): tag.decompose() # remove any hidden tag for tag in self.soup(hidden=True): tag.decompose() # fix flow content that is not permitted inside
HTMLDocumentBackend._fix_invalid_paragraph_structure(self.soup)
content = self.soup.body or self.soup
# normalize
tags
for br in content("br"):
br.replace_with(NavigableString("\n"))
# set default content layer
# Furniture before the first heading rule, except for headers in tables
header = None
# Find all headers first
all_headers = content.find_all(["h1", "h2", "h3", "h4", "h5", "h6"])
# Keep only those that do NOT have a
| and | )
# We don't want this recursive to support nested tables
cells = row(["td", "th"], recursive=False)
# Check if cell is in a column header or row header
col_header = True
row_header = True
for html_cell in cells:
if isinstance(html_cell, Tag):
_, row_span = HTMLDocumentBackend._get_cell_spans(html_cell)
if html_cell.name == "td":
col_header = False
row_header = False
elif row_span == 1:
row_header = False
if not row_header:
row_idx += 1
start_row_span = 0
else:
start_row_span += 1
# Extract the text content of each cell
col_idx = 0
for html_cell in cells:
if not isinstance(html_cell, Tag):
continue
# extract inline formulas
for formula in html_cell("inline-formula"):
math_parts = formula.text.split("$$")
if len(math_parts) == 3:
math_formula = f"$${math_parts[1]}$$"
formula.replace_with(NavigableString(math_formula))
provs_in_cell: list[RefItem] = []
rich_table_cell = self._is_rich_table_cell(html_cell)
if rich_table_cell:
# Parse table cell sub-tree for Rich Cells content:
with self._use_table_cell_context():
provs_in_cell = self._walk(html_cell, doc)
group_name = f"rich_cell_group_{len(doc.tables)}_{col_idx}_{start_row_span + row_idx}"
rich_table_cell, ref_for_rich_cell = (
HTMLDocumentBackend.process_rich_table_cells(
provs_in_cell, group_name, doc, docling_table
)
)
# Extracting text
text = HTMLDocumentBackend._clean_unicode(
self.get_text(html_cell).strip()
)
col_span, row_span = self._get_cell_spans(html_cell)
if row_header:
row_span -= 1
while (
col_idx < num_cols
and grid[row_idx + start_row_span][col_idx] is not None
):
col_idx += 1
for r in range(start_row_span, start_row_span + row_span):
for c in range(col_span):
if row_idx + r < num_rows and col_idx + c < num_cols:
grid[row_idx + r][col_idx + c] = text
if rich_table_cell:
rich_cell = RichTableCell(
text=text,
row_span=row_span,
col_span=col_span,
start_row_offset_idx=start_row_span + row_idx,
end_row_offset_idx=start_row_span + row_idx + row_span,
start_col_offset_idx=col_idx,
end_col_offset_idx=col_idx + col_span,
column_header=col_header,
row_header=((not col_header) and html_cell.name == "th"),
ref=ref_for_rich_cell, # points to an artificial group around children
)
doc.add_table_cell(table_item=docling_table, cell=rich_cell)
else:
simple_cell = TableCell(
text=text,
row_span=row_span,
col_span=col_span,
start_row_offset_idx=start_row_span + row_idx,
end_row_offset_idx=start_row_span + row_idx + row_span,
start_col_offset_idx=col_idx,
end_col_offset_idx=col_idx + col_span,
column_header=col_header,
row_header=((not col_header) and html_cell.name == "th"),
)
doc.add_table_cell(table_item=docling_table, cell=simple_cell)
return data
def _walk(self, element: Tag, doc: DoclingDocument) -> list[RefItem]:
"""Parse an XML tag by recursively walking its content.
While walking, the method buffers inline text across tags like or ,
emitting text nodes only at block boundaries.
Args:
element: The XML tag to parse.
doc: The Docling document to be updated with the parsed content.
"""
added_refs: list[RefItem] = []
buffer: AnnotatedTextList = AnnotatedTextList()
def _flush_buffer() -> None:
if not buffer:
return
annotated_text_list: AnnotatedTextList = buffer.simplify_text_elements()
parts = annotated_text_list.split_by_newline()
buffer.clear()
if not "".join([el.text for el in annotated_text_list]):
return
for annotated_text_list in parts:
with self._use_inline_group(annotated_text_list, doc) as inline_ref:
for annotated_text in annotated_text_list:
if annotated_text.text.strip():
seg_clean = HTMLDocumentBackend._clean_unicode(
annotated_text.text.strip()
)
if annotated_text.code:
docling_code2 = doc.add_code(
parent=self.parents[self.level],
text=seg_clean,
content_layer=self.content_layer,
formatting=annotated_text.formatting,
hyperlink=annotated_text.hyperlink,
)
if inline_ref is None:
added_refs.append(docling_code2.get_ref())
else:
docling_text2 = doc.add_text(
parent=self.parents[self.level],
label=DocItemLabel.TEXT,
text=seg_clean,
content_layer=self.content_layer,
formatting=annotated_text.formatting,
hyperlink=annotated_text.hyperlink,
)
if inline_ref is None:
added_refs.append(docling_text2.get_ref())
if inline_ref is not None:
added_refs.append(inline_ref)
for node in element.contents:
if isinstance(node, Tag):
name = node.name.lower()
if name == "img":
_flush_buffer()
im_ref3 = self._emit_image(node, doc)
if im_ref3:
added_refs.append(im_ref3)
elif name in _FORMAT_TAG_MAP:
_flush_buffer()
with self._use_format([name]):
wk = self._walk(node, doc)
added_refs.extend(wk)
elif name == "a":
with self._use_hyperlink(node):
wk2 = self._walk(node, doc)
added_refs.extend(wk2)
elif name in _BLOCK_TAGS:
_flush_buffer()
blk = self._handle_block(node, doc)
added_refs.extend(blk)
elif node.find(_BLOCK_TAGS):
_flush_buffer()
wk3 = self._walk(node, doc)
added_refs.extend(wk3)
else:
buffer.extend(
self._extract_text_and_hyperlink_recursively(
node, find_parent_annotation=True, keep_newlines=True
)
)
elif isinstance(node, NavigableString) and not isinstance(
node, PreformattedString
):
if str(node).strip("\n\r") == "":
_flush_buffer()
else:
buffer.extend(
self._extract_text_and_hyperlink_recursively(
node, find_parent_annotation=True, keep_newlines=True
)
)
_flush_buffer()
return added_refs
@staticmethod
def _collect_parent_format_tags(item: PageElement) -> list[str]:
tags = []
for format_tag in _FORMAT_TAG_MAP:
this_parent = item.parent
while this_parent is not None:
if this_parent.name == format_tag:
tags.append(format_tag)
break
this_parent = this_parent.parent
return tags
@property
def _formatting(self):
kwargs = {}
for t in self.format_tags:
kwargs.update(_FORMAT_TAG_MAP[t])
if not kwargs:
return None
return Formatting(**kwargs)
def _extract_text_and_hyperlink_recursively(
self,
item: PageElement,
ignore_list=False,
find_parent_annotation=False,
keep_newlines=False,
) -> AnnotatedTextList:
result: AnnotatedTextList = AnnotatedTextList()
# If find_parent_annotation, make sure that we keep track of
# any a- or formatting-tag that has been present in the
# DOM-parents already.
if find_parent_annotation:
format_tags = self._collect_parent_format_tags(item)
this_parent = item.parent
while this_parent is not None:
if this_parent.name == "a" and this_parent.get("href"):
with self._use_format(format_tags):
with self._use_hyperlink(this_parent):
return self._extract_text_and_hyperlink_recursively(
item, ignore_list
)
this_parent = this_parent.parent
if isinstance(item, PreformattedString):
return AnnotatedTextList()
if isinstance(item, NavigableString):
text = item.strip()
code = any(code_tag in self.format_tags for code_tag in _CODE_TAG_SET)
if text:
return AnnotatedTextList(
[
AnnotatedText(
text=text,
hyperlink=self.hyperlink,
formatting=self._formatting,
code=code,
)
]
)
if keep_newlines and item.strip("\n\r") == "":
return AnnotatedTextList(
[
AnnotatedText(
text="\n",
hyperlink=self.hyperlink,
formatting=self._formatting,
code=code,
)
]
)
return AnnotatedTextList()
tag = cast(Tag, item)
if not ignore_list or (tag.name not in ["ul", "ol"]):
for child in tag:
if isinstance(child, Tag) and child.name in _FORMAT_TAG_MAP:
with self._use_format([child.name]):
result.extend(
self._extract_text_and_hyperlink_recursively(
child, ignore_list, keep_newlines=keep_newlines
)
)
elif isinstance(child, Tag) and child.name == "a":
with self._use_hyperlink(child):
result.extend(
self._extract_text_and_hyperlink_recursively(
child, ignore_list, keep_newlines=keep_newlines
)
)
else:
# Recursively get the child's text content
result.extend(
self._extract_text_and_hyperlink_recursively(
child, ignore_list, keep_newlines=keep_newlines
)
)
return result
@contextmanager
def _use_hyperlink(self, tag: Tag):
old_hyperlink: Union[AnyUrl, Path, None] = None
new_hyperlink: Union[AnyUrl, Path, None] = None
this_href = tag.get("href")
if this_href is None:
yield None
else:
if isinstance(this_href, str) and this_href:
old_hyperlink = self.hyperlink
this_href = self._resolve_relative_path(this_href)
# ugly fix for relative links since pydantic does not support them.
try:
new_hyperlink = AnyUrl(this_href)
except ValidationError:
new_hyperlink = Path(this_href)
self.hyperlink = new_hyperlink
try:
yield None
finally:
if new_hyperlink:
self.hyperlink = old_hyperlink
@contextmanager
def _use_format(self, tags: list[str]):
if not tags:
yield None
else:
self.format_tags.extend(tags)
try:
yield None
finally:
self.format_tags = self.format_tags[: -len(tags)]
@contextmanager
def _use_inline_group(
self, annotated_text_list: AnnotatedTextList, doc: DoclingDocument
) -> Iterator[RefItem | None]:
"""Create an inline group for annotated texts.
Checks if annotated_text_list has more than one item and if so creates an inline
group in which the text elements can then be generated. While the context manager
is active the inline group is set as the current parent.
Args:
annotated_text_list (AnnotatedTextList): Annotated text
doc (DoclingDocument): Currently used document
Yields:
The RefItem of the created InlineGroup, or None when the list has only one
element and no group is created.
"""
if len(annotated_text_list) > 1:
inline_fmt = doc.add_group(
label=GroupLabel.INLINE,
parent=self.parents[self.level],
content_layer=self.content_layer,
)
self.parents[self.level + 1] = inline_fmt
self.level += 1
try:
yield inline_fmt.get_ref()
finally:
self.parents[self.level] = None
self.level -= 1
else:
yield None
@contextmanager
def _use_details(self, tag: Tag, doc: DoclingDocument):
"""Create a group with the content of a details tag.
While the context manager is active, the hierarchy level is set one
level higher as the cuurent parent.
Args:
tag: The details tag.
doc: Currently used document.
"""
self.parents[self.level + 1] = doc.add_group(
name=tag.name,
label=GroupLabel.SECTION,
parent=self.parents[self.level],
content_layer=self.content_layer,
)
self.level += 1
try:
yield None
finally:
self.parents[self.level + 1] = None
self.level -= 1
@contextmanager
def _use_footer(self, tag: Tag, doc: DoclingDocument):
"""Create a group with a footer.
Create a group with the content of a footer tag. While the context manager
is active, the hierarchy level is set one level higher as the cuurent parent.
Args:
tag: The footer tag.
doc: Currently used document.
"""
current_layer = self.content_layer
self.content_layer = ContentLayer.FURNITURE
self.parents[self.level + 1] = doc.add_group(
name=tag.name,
label=GroupLabel.SECTION,
parent=self.parents[self.level],
content_layer=self.content_layer,
)
self.level += 1
try:
yield None
finally:
self.parents[self.level + 1] = None
self.level -= 1
self.content_layer = current_layer
@contextmanager
def _use_table_cell_context(self):
"""Preserve the hierarchy level and parents during table cell processing.
While the context manager is active, the hierarchy level and parents can be modified.
When exiting, the original level and parents are restored.
"""
original_level = self.level
original_parents = self.parents.copy()
try:
yield
finally:
self.level = original_level
self.parents = original_parents
def _handle_heading(self, tag: Tag, doc: DoclingDocument) -> list[RefItem]:
added_ref = []
tag_name = tag.name.lower()
# set default content layer to BODY as soon as we encounter a heading
self.content_layer = ContentLayer.BODY
level = int(tag_name[1])
annotated_text_list = self._extract_text_and_hyperlink_recursively(
tag, find_parent_annotation=True
)
annotated_text = annotated_text_list.to_single_text_element()
text_clean = HTMLDocumentBackend._clean_unicode(annotated_text.text)
# the first level is for the title item
if level == 1:
for key in self.parents.keys():
self.parents[key] = None
self.level = 0
self.parents[self.level + 1] = doc.add_title(
text_clean,
content_layer=self.content_layer,
formatting=annotated_text.formatting,
hyperlink=annotated_text.hyperlink,
)
p1 = self.parents[self.level + 1]
if p1 is not None:
added_ref = [p1.get_ref()]
# the other levels need to be lowered by 1 if a title was set
else:
level -= 1
if level > self.level:
# add invisible group
for i in range(self.level, level):
_log.debug(f"Adding invisible group to level {i}")
self.parents[i + 1] = doc.add_group(
name=f"header-{i + 1}",
label=GroupLabel.SECTION,
parent=self.parents[i],
content_layer=self.content_layer,
)
self.level = level
elif level < self.level:
# remove the tail
for key in self.parents.keys():
if key > level + 1:
_log.debug(f"Remove the tail of level {key}")
self.parents[key] = None
self.level = level
self.parents[self.level + 1] = doc.add_heading(
parent=self.parents[self.level],
text=text_clean,
orig=annotated_text.text,
level=self.level,
content_layer=self.content_layer,
formatting=annotated_text.formatting,
hyperlink=annotated_text.hyperlink,
)
p2 = self.parents[self.level + 1]
if p2 is not None:
added_ref = [p2.get_ref()]
self.level += 1
for img_tag in tag("img"):
if isinstance(img_tag, Tag):
im_ref = self._emit_image(img_tag, doc)
if im_ref:
added_ref.append(im_ref)
return added_ref
def _handle_list(self, tag: Tag, doc: DoclingDocument) -> RefItem:
tag_name = tag.name.lower()
start: Optional[int] = None
name: str = ""
is_ordered = tag_name == "ol"
if is_ordered:
start_attr = tag.get("start")
if isinstance(start_attr, str) and start_attr.isnumeric():
start = int(start_attr)
name = "ordered list" + (f" start {start}" if start is not None else "")
else:
name = "list"
# Create the list container
list_group = doc.add_list_group(
name=name,
parent=self.parents[self.level],
content_layer=self.content_layer,
)
self.parents[self.level + 1] = list_group
self.ctx.list_ordered_flag_by_ref[list_group.self_ref] = is_ordered
if is_ordered and start is not None:
self.ctx.list_start_by_ref[list_group.self_ref] = start
self.level += 1
# For each top-level or |
|---|