) stay processable
for carried in sub.iter("*"):
carried.tag = "done"
def handle_lists(element: _Element, options: Extractor) -> _Element | None:
"Process lists elements including their descendants."
processed_element = Element(element.tag)
if text_chars_test(element.text):
new_child_elem = SubElement(processed_element, "item")
new_child_elem.text = element.text
for child in element.iterdescendants("item"):
# items of a nested list are processed by the recursive call and renamed "done",
# but lxml fetches the first of them before that happens: skip it here
if child.tag == "done":
continue
new_child_elem = Element("item")
if len(child) == 0:
processed_child = process_node(child, options)
if processed_child is not None:
new_child_elem.text = " ".join(filter(None, (processed_child.text, processed_child.tail)))
else:
process_nested_elements(child, new_child_elem, options)
if text_chars_test(child.tail) and len(new_child_elem) > 0:
last = new_child_elem[-1]
last.tail = f"{last.tail} {child.tail}" if text_chars_test(last.tail) else child.tail
if is_text_element(new_child_elem) or next(new_child_elem.iter("graphic"), None) is not None:
update_elem_rendition(child, new_child_elem)
processed_element.append(new_child_elem)
child.tag = "done"
element.tag = "done"
if is_text_element(processed_element):
update_elem_rendition(element, processed_element)
return processed_element
return None
def is_code_block_element(element: _Element) -> bool:
"Check if it is a code element according to common structural markers."
# pip
if element.get("lang") or element.tag == "code":
return True
# GitHub
parent = element.getparent()
if parent is not None and "highlight" in parent.get("class", ""):
return True
# highlightjs
code = element.find("code")
return code is not None and len(element) == 1 and not (element.text or "").strip() and not (code.tail or "").strip()
def handle_code_blocks(element: _Element) -> _Element:
"Turn element into a properly tagged code block."
processed_element = deepcopy(element)
for child in element.iter("*"):
child.tag = "done"
processed_element.tag = "code"
return processed_element
def handle_quotes(element: _Element, options: Extractor) -> _Element | None:
"Process quotes elements."
if is_code_block_element(element):
return handle_code_blocks(element)
processed_element = Element(element.tag)
process_nested_elements(element, processed_element, options)
if is_text_element(processed_element):
# avoid double/nested tags
strip_tags(processed_element, "quote")
return processed_element
return None
def handle_other_elements(element: _Element, potential_tags: set[str], options: Extractor) -> _Element | None:
"Handle diverse or unknown elements in the scope of relevant tags."
# handle w3schools code
if element.tag == "div" and "w3-code" in element.get("class", ""):
return handle_code_blocks(element)
if element.tag != "div" or "div" not in potential_tags:
if element.tag != "done":
_log_event("discarding element", element.tag, element.text)
return None
processed_element = handle_textnode(element, options, comments_fix=False, preserve_spaces=True)
if processed_element is None or not text_chars_test(processed_element.text):
return None
processed_element.attrib.clear()
processed_element.tag = "p"
return processed_element
def handle_paragraphs(element: _Element, potential_tags: set[str], options: Extractor) -> _Element | None:
"Process paragraphs along with their children, trim and clean the content."
# attrib.clear() verified unnecessary here (output_diff 0/1501, 2026-08)
# no children
if len(element) == 0:
return process_node(element, options)
# children
processed_element = Element(element.tag)
# unexpected children: keep their text in place
strip_tags(element, *{str(c.tag) for c in element.iterdescendants("*") if c.tag not in {*potential_tags, "done"}})
for child in element.iter("*"):
# todo: act on spacing here?
processed_child = handle_textnode(child, options, comments_fix=False, preserve_spaces=True)
if processed_child is not None:
# todo: needing attention!
if processed_child.tag == "p":
_log_event("extra in p", "p", processed_child.text)
processed_element.text = " ".join(filter(None, (processed_element.text, processed_child.text)))
child.tag = "done"
continue
# handle formatting
newsub = Element(child.tag)
if processed_child.tag in INLINE_CONSUMING:
# carry inline children verbatim (ref/hi/del wrapping formatting)
if _wraps_inline(processed_child):
define_newelem(processed_child, processed_element, keep_children=True)
child.tag = "done"
continue
# check depth and clean
for item in processed_child:
if item.tag == "lb" and item.tail:
item.tail = " " + item.tail.lstrip()
elif text_chars_test(item.text):
item.text = f" {item.text}"
strip_tags(processed_child, *{str(item.tag) for item in processed_child})
_copy_attrs(child, newsub)
newsub.text, newsub.tail = processed_child.text, processed_child.tail
if processed_child.tag == "graphic":
image_elem = handle_image(processed_child, options)
if image_elem is not None:
newsub = image_elem
processed_element.append(newsub)
child.tag = "done"
# finish
if len(processed_element) > 0:
last_elem = processed_element[-1]
# clean trailing lb-elements
if last_elem.tag == "lb" and last_elem.tail is None:
delete_element(last_elem)
return processed_element
# text_chars_test, not truthiness: layout wrappers (clearfix/spacer divs, cleared-out
# ad slots) leave indentation-only text, which used to be emitted as blank paragraphs
# and, sitting between two copies of a paragraph, also defeated the adjacent dedup
if text_chars_test(processed_element.text):
return processed_element
_log_event("discarding element:", "p", tostring(processed_element))
return None
def define_cell_type(is_header: bool) -> _Element:
"Determine cell element type and mint new element."
cell_element = Element("cell")
if is_header:
cell_element.set("role", "head")
return cell_element
_MAX_SPAN = 100
def _span(cell: _Element, attr: str) -> int:
"Parse a cell's col/rowspan, defaulting to 1, capped at _MAX_SPAN."
# isdecimal, not isdigit: int() rejects the superscripts isdigit() admits
value = cell.get(attr, "1")
return min(int(value), _MAX_SPAN) if value.isdecimal() else 1
def _flush_rowspan_phantoms(rowspan_map: dict[int, int], newrow: _Element) -> None:
"Insert empty placeholder cells for rowspan-occupied columns at the current row position."
while (col := len(newrow)) in rowspan_map:
newrow.append(define_cell_type(False))
rowspan_map[col] -= 1
if rowspan_map[col] == 0:
del rowspan_map[col]
def _finalize_row(newtable: _Element, newrow: _Element, rowspan_map: dict[int, int], max_cols: int) -> None:
"Close a row: insert trailing rowspan placeholders, pad to width, append if non-empty."
_flush_rowspan_phantoms(rowspan_map, newrow)
while len(newrow) < max_cols:
newrow.append(define_cell_type(False))
if any(cell.text or len(cell) > 0 for cell in newrow):
newtable.append(newrow)
def _fill_cell(
new_child_elem: _Element,
cell: _Element,
nested_elems: set[_Element],
ptags_with_div: set[str],
options: Extractor,
) -> None:
"Extract a source td/th cell's content into the new
for child in cell.iterdescendants():
if not isinstance(child.tag, str) or child.tag == "done":
continue
if child in nested_elems:
# nested tables are left to the main loop, their tail moves to the cell holding them
if child.tag == "table" and next(child.iterancestors("table")) not in nested_elems:
_append_block(new_child_elem, None, child)
child.tail = None
continue
if separates_inline(child) and not text_chars_test(child.tail):
new_child_elem.append(Element("lb"))
child.tag = "done"
continue
if child.tag in TABLE_ELEMS: # stray cell from malformed HTML
child.tag = "cell"
processed_subchild = handle_textnode(child, options, preserve_spaces=True)
elif child.tag in INLINE_CONSUMING:
processed_subchild = handle_textnode(child, options, preserve_spaces=True)
# handle_textnode drops inline wrappers (ref/hi/del) with children but no direct
# text (e.g. folded into its elements if nothing found or text too short
# todo: test precision and recall settings here
if len(result_body) == 0 or len(temp_text) < options.min_extracted_size:
# copies of what the main pass consumed are not recovered again
consumed: set[_Element] = {twin for elem, twin in zip(order, backup_tree.iter(), strict=True) if elem.tag == "done"}
result_body = recover_wild_text(backup_tree, result_body, options, potential_tags, consumed)
temp_text = " ".join(result_body.itertext()).strip()
# drop substantial elements repeating the previous one (overlapping-candidate / recovery artifact);
# length-gated so short genuine repeats stay for the dedup (#778) and tree-size guards
previous = None
for el in list(result_body):
current = _elem_text(el)
if current and current == previous and len(current) > MIN_DUPLICATE_LENGTH:
delete_element(el, keep_tail=False)
else:
previous = current
# filter output
strip_elements(result_body, "done")
strip_tags(result_body, "div")
# stripping the divs above merges their indentation into the preceding link text); carry the subtree directly instead
if processed_subchild is None and len(child) > 0:
processed_subchild = child
# lists in cells only in recall mode: keeping them otherwise is noise (measured precision loss)
elif child.tag == "list":
processed_list = handle_lists(child, options)
_append_block(new_child_elem, processed_list if options.focus == "recall" else None, child)
continue
else:
processed_subchild = handle_textelem(child, ptags_with_div, options)
define_newelem(processed_subchild, new_child_elem, keep_children=True)
child.tag = "done"
def handle_table(table_elem: _Element, potential_tags: set[str], options: Extractor) -> _Element | None:
"Process single table element."
newtable = Element("table")
ptags_with_div = potential_tags | {"div"}
# strip these structural elements
strip_tags(table_elem, "thead", "tbody", "tfoot")
# Collect elements inside nested descendants. Used only in the inner cell
# walk: skip without "done"-marking so the main extraction loop can call handle_table()
# on each nested table separately. Must hold the elements (not id()) — lxml proxies are
# weakly referenced; id() values become stale as soon as the proxy is freed.
nested_elems: set[_Element] = set()
for nested_table in table_elem.iterdescendants("table"):
nested_elems.update(nested_table.iter())
# direct children only, nested tables are left to the main loop.
# Orphan cells join the current row but do not count for max_cols.
rows: list[list[_Element]] = [[]]
captions: list[str] = []
max_cols = 0
for elem in table_elem:
if elem.tag == "tr":
rows.append([cell for cell in elem if cell.tag in TABLE_ELEMS])
max_cols = max(max_cols, sum(_span(cell, "colspan") for cell in rows[-1]))
elif elem.tag in TABLE_ELEMS:
rows[-1].append(elem)
continue
elif elem.tag == "caption":
captions.append(" ".join(elem.itertext()).strip())
# text consumed above, not to be emitted again, images are left to the main loop
for sub in elem.iterdescendants("*"):
if sub.tag == "graphic":
sub.tail = None
else:
sub.tag = "done"
elif not isinstance(elem.tag, str) or elem.tag == "table":
continue
elem.tag = "done"
max_cols = min(max_cols, _MAX_SPAN)
for caption_text in filter(None, captions):
caption_cell = define_cell_type(True)
caption_cell.text = caption_text
caption_row = Element("row")
caption_row.append(caption_cell)
_finalize_row(newtable, caption_row, {}, max_cols)
header_row_emitted = False
rowspan_map: dict[int, int] = {} # col_idx → rows still spanned from a rowspan cell
for cells in rows:
newrow = Element("row")
row_has_th = False
for cell in cells:
is_header = cell.tag == "th" and not header_row_emitted
row_has_th = row_has_th or is_header
_flush_rowspan_phantoms(rowspan_map, newrow)
new_child_elem = define_cell_type(is_header)
colspan = _span(cell, "colspan")
# Track rowspan: mark all spanned columns as occupied for subsequent rows
rowspan = _span(cell, "rowspan")
if rowspan > 1:
for c in range(len(newrow), len(newrow) + colspan):
rowspan_map[c] = rowspan - 1
_fill_cell(new_child_elem, cell, nested_elems, ptags_with_div, options)
# add to tree (keep empty cells so column positions stay aligned)
newrow.append(new_child_elem)
# inline colspan: pad with empty cells so subsequent rows align
for _ in range(colspan - 1):
newrow.append(define_cell_type(is_header))
cell.tag = "done"
_finalize_row(newtable, newrow, rowspan_map, max_cols)
header_row_emitted = header_row_emitted or row_has_th
return newtable if len(newtable) > 0 else None
def handle_image(element: _Element, options: Extractor | None = None) -> _Element | None:
"Process image elements and their relevant attributes."
link = image_src(element)
if link is None:
return None
if not link.startswith("http"):
if options is not None and options.url is not None:
link = urljoin(options.url, link)
else:
link = re.sub(r"^//", "http://", link)
processed_element = Element(element.tag)
processed_element.set("src", link)
if alt_attr := element.get("alt"):
processed_element.set("alt", alt_attr)
if title_attr := element.get("title"):
processed_element.set("title", title_attr)
processed_element.tail = element.tail
return processed_element
def handle_textelem(element: _Element, potential_tags: set[str], options: Extractor) -> _Element | None:
"""Process text element and determine how to deal with its content"""
new_element = None
# bypass: nested elements
if element.tag == "list":
new_element = handle_lists(element, options)
elif element.tag in CODES_QUOTES:
new_element = handle_quotes(element, options)
elif element.tag == "head":
new_element = handle_titles(element, options)
elif element.tag == "p":
new_element = handle_paragraphs(element, potential_tags, options)
elif element.tag == "lb":
if text_chars_test(element.tail) is True:
this_element = process_node(element, options)
if this_element is not None:
new_element = Element("p")
new_element.text = this_element.tail
elif element.tag in INLINE_CONSUMING:
new_element = handle_formatting(element, options) # process_node(element, options)
elif element.tag == "table" and "table" in potential_tags:
new_element = handle_table(element, potential_tags, options)
elif element.tag == "graphic" and "graphic" in potential_tags:
new_element = handle_image(element, options)
else:
# other elements (div, ??, ??)
new_element = handle_other_elements(element, potential_tags, options)
return new_element
def recover_wild_text(
tree: HtmlElement,
result_body: _Element,
options: Extractor,
potential_tags: set[str] | None = None,
consumed: set[_Element] | None = None,
) -> _Element:
"""Look for all previously unconsidered wild elements, including outside of the determined
frame and throughout the document to recover potentially missing text parts.
Do not widen `search_expr` (e.g. headings, more div shapes) without benchmarking the
full suite: extra recovered text increases the extracted length, which can suppress
the stronger rescues that run after this one (compare_extraction, the baseline rescue,
the recall escalation).
"""
LOGGER.debug("Recovering wild text elements")
# copy: the recall branch below mutates potential_tags, must not leak back to the caller
potential_tags = set(TAG_CATALOG if potential_tags is None else potential_tags)
# blockquote/pre/q are already renamed to quote/code by convert_tags before this tree is built
search_expr = ".//code|.//p|.//quote|.//table|.//div[contains(@class, 'w3-code')]"
if options.focus == "recall":
potential_tags.update(["div", "lb"])
search_expr += "|.//div|.//lb|.//list"
# prune; in fast mode (no external comparator to defer to) keep teaser-class blocks, some of
# which are real content — this is the last-resort path after the confident extractor failed
search_tree = prune_unwanted_sections(tree, potential_tags, options, keep_teasers=options.fast)
subelems = search_tree.xpath(search_expr)
# dedup against the pre-main-pass snapshot: skip what the main pass already took -- exact
# match (not length-gated, #634; accepted cost: identical-text elements collapse) or a
# length-gated substring (a
container)
elem_texts = [_elem_text(el) for el in result_body]
# newline-joined (trimmed element text has no newline) so no substring match spans two elements
existing = "\n".join(filter(None, elem_texts))
existing_elems = set(elem_texts)
# elements emitted or deduped whole: their descendants are covered
handled: set[_Element] = set()
for subelem in subelems:
if (consumed and subelem in consumed) or any(a in handled for a in subelem.iterancestors()):
continue
processed = handle_textelem(subelem, potential_tags, options)
if processed is None:
continue
if processed is subelem:
handled.add(subelem)
text = _elem_text(processed)
# image-only blocks have no text to compare, check their sources against the live body
images = {img.get("src") for img in processed.iter("graphic")}
if not text and images and images <= {g.get("src") for g in result_body.iter("graphic")}:
continue
# past the cap, the substring scan is skipped and `existing` stops growing
under_cap = len(existing) <= DEDUPE_SCAN_CAP
# a copy carrying its source tail is a fragment of a text run: no length gate
fragment = trim(text + (processed.tail or "")) if text_chars_test(processed.tail) else ""
if text and (
text in existing_elems
or (under_cap and ((len(text) > MIN_DUPLICATE_LENGTH and text in existing) or (fragment and fragment in existing)))
):
continue
result_body.append(processed)
if under_cap:
existing += "\n" + (fragment or text)
existing_elems.add(text)
return result_body
def prune_unwanted_sections(
tree: HtmlElement,
potential_tags: set[str],
options: Extractor,
keep_teasers: bool = False,
) -> HtmlElement:
"Rule-based deletion of targeted document sections"
favor_precision = options.focus == "precision"
# prune the rest
tree = prune_unwanted_nodes(tree, OVERALL_DISCARD_XPATH, with_backup=True)
# decide if images are preserved
if "graphic" not in potential_tags:
tree = prune_unwanted_nodes(tree, DISCARD_IMAGE_ELEMENTS)
# balance precision/recall
if options.focus != "recall":
# teaser-class blocks are sometimes real content; keep them on the recovery path,
# which only runs once the confident extractor has already come up short
if not keep_teasers:
tree = prune_unwanted_nodes(tree, TEASER_DISCARD_XPATH)
if favor_precision:
tree = prune_unwanted_nodes(tree, PRECISION_DISCARD_XPATH)
# remove elements by link density, several passes
for _ in range(2):
tree = delete_by_link_density(tree, "div", backtracking=True, favor_precision=favor_precision)
tree = delete_by_link_density(tree, "list", backtracking=False, favor_precision=favor_precision)
tree = delete_by_link_density(tree, "p", backtracking=False, favor_precision=favor_precision)
# tables
if "table" in potential_tags or favor_precision:
# collect before deleting: removing a table mid-iteration can make tree.iter() skip a table
# that follows a deleted one containing a nested table (iterator descends into the detached subtree)
boilerplate_tables = [elem for elem in tree.iter("table") if link_density_test_tables(elem) is True]
for elem in boilerplate_tables:
delete_element(elem, keep_tail=False)
if favor_precision:
# delete trailing titles
while len(tree) > 0 and (tree[-1].tag == "head"):
delete_element(tree[-1], keep_tail=False)
tree = delete_by_link_density(tree, "head", backtracking=False, favor_precision=True)
tree = delete_by_link_density(tree, "quote", backtracking=False, favor_precision=True)
# after the link density tests, which need the refs
strip_tags(tree, "span", *(() if "ref" in potential_tags else ("ref",)))
return tree
# A sibling-article container is the body of a listing page, but on an article page the same
# shape is a teaser strip ("recommended stories") sitting beside the body. Only the listing
# container carries the bulk of the page, so that share is what tells the two apart.
LISTING_TEXT_SHARE = 0.6
def _holds_most_of_the_text(tree: HtmlElement, expr: XPath) -> bool:
"Whether the first element matched by expr carries most of the text of the tree."
match = next((s for s in expr(tree) if s is not None), None)
if match is None:
return False
page_length = len(trim(" ".join(tree.itertext())))
if not page_length:
return False
return len(trim(" ".join(match.itertext()))) / page_length >= LISTING_TEXT_SHARE
def _entries_link_out(container: _Element) -> bool:
"""Whether every entry of the container is a teaser: its first heading links to a page of its
own, which no other entry links to. Forum posts share the shape but open with the poster's
name, which comes back along the thread or carries no link at all, and they hold user cards
(description lists) and quotes, which teasers do not."""
targets: set[str] = set()
for entry in container.iterchildren("article"):
if entry.find(".//list[@rend='dl']") is not None or entry.find(".//quote") is not None:
return False
heading = entry.find(".//head")
link = heading.find(".//ref") if heading is not None else None
if link is None:
return False
# convert_tags keeps the href when links are off and moves it to target when they are on
target = link.get("href") or link.get("target")
if not target or target in targets:
return False
targets.add(target)
return True
def _is_listing_page(tree: HtmlElement, options: Extractor) -> bool:
"""Whether the sibling-article container is the body of a listing page. The container only
widens what BODY_XPATH would take on its own: that element has to be one of its entries and a
short teaser. A body outside the container means a teaser strip beside the article, and an
entry that already clears the recall "sure thing" length is a full article, as on pages that
preload the next stories below the current one. Every entry also has to be a teaser that
links out from its heading."""
listing_expr = LISTING_BODY_XPATH[0]
container = next((s for s in listing_expr(tree) if s is not None), None)
if container is None:
return False
entry = next((s for expr in BODY_XPATH for s in expr(tree) if s is not None), None)
if entry is None or entry.getparent() is not container:
return False
if len(trim(" ".join(entry.itertext()))) > options.min_extracted_size * 10:
return False
if not _entries_link_out(container):
return False
return _holds_most_of_the_text(tree, listing_expr)
def _extract(tree: HtmlElement, options: Extractor, is_forum: bool = False) -> tuple[_Element, str, set[str]]:
# init
potential_tags = set(TAG_CATALOG)
if options.tables is True:
potential_tags.update(["table", "td", "th", "tr"])
if options.images is True:
potential_tags.add("graphic")
if options.links is True:
potential_tags.add("ref")
result_body = Element("body")
expressions = BODY_XPATH
# the posts of a thread forum line up the same way, each with its user card and quotes
if options.focus == "recall" and not is_forum and _is_listing_page(tree, options):
expressions = LISTING_BODY_XPATH + BODY_XPATH
# iterate
for expr in expressions:
# select tree if the expression has been found
subtree = next((s for s in expr(tree) if s is not None), None)
if subtree is None:
continue
# prune the subtree
subtree = prune_unwanted_sections(subtree, potential_tags, options)
# skip if empty tree
if len(subtree) == 0:
continue
# no paragraphs containing text, or not enough
ptest = subtree.xpath("//p//text()")
factor = 1 if options.focus == "precision" else 3
if not ptest or len("".join(ptest)) < options.min_extracted_size * factor:
potential_tags.add("div")
LOGGER.debug(sorted(potential_tags))
subelems = subtree.xpath(".//*")
# a single text run (only lb or inline elements): process the frame itself
# lb-only frames go there unconditionally, a failed main pass then hands them to the recovery
tags = {e.tag for e in subelems}
if tags == {"lb"} or (subtree.tag == "div" and "div" in potential_tags and tags <= INLINE_FORMATTABLE | {"lb"}):
subelems = [subtree]
# the frame's own text before its first child, kept if the frame is accepted
lead, start = None, len(result_body)
if subelems != [subtree] and text_chars_test(subtree.text):
lead = Element("p")
lead.text = subtree.text
lead = process_node(lead, options)
# extract content
for elem in subelems:
# handle_other_elements() emits a text-bearing div as-is, children and all, and
# appending it MOVES it into result_body. Its descendants are still in the list
# captured above, so revisiting one would emit its text twice -- an
whose tail
# carries a paragraph turned into a duplicate of that paragraph. handle_paragraphs()
# marks the children it consumes "done"; this covers the elements it cannot retag.
if elem.getroottree().getroot() is result_body:
continue
# handlers may rename the element to "done", so read these first
tag, tail = elem.tag, elem.tail
processed_elem = handle_textelem(elem, potential_tags, options)
if processed_elem is not None:
result_body.append(processed_elem)
# a copy leaves its source behind for a later, wider subtree
if elem.getroottree().getroot() is not result_body:
elem.tag = "done"
# text right after a rebuilt block is a paragraph of its own, like an
tail,
# unless the handler already carried it over (a code block copied as a whole)
if tag in REBUILT_BLOCKS and text_chars_test(tail) and (processed_elem is None or processed_elem.tail is None):
tail_elem = Element("p")
tail_elem.text = tail
if process_node(tail_elem, options) is not None:
result_body.append(tail_elem)
# remove trailing titles
while len(result_body) > 0 and (result_body[-1].tag in NOT_AT_THE_END):
delete_element(result_body[-1], keep_tail=False)
# exit once there is real content, not just a lone image
if sum(e.tag != "graphic" for e in result_body) > 1:
LOGGER.debug(trim(str(expr)))
if lead is not None:
result_body.insert(start, lead)
break
temp_text = " ".join(result_body.itertext()).strip()
return result_body, temp_text, potential_tags
def extract_content(cleaned_tree: HtmlElement, options: Extractor, is_forum: bool = False) -> tuple[_Element, str]:
"""Find the main content of a page using a set of XPath expressions,
then extract relevant elements, strip them of unwanted subparts and
convert them. is_forum marks a thread forum, whose posts are never
read as the entries of a listing page."""
# backup
backup_tree = deepcopy(cleaned_tree)
# source order before the main pass moves elements, the backup keeps it
order = list(cleaned_tree.iter())
result_body, temp_text, potential_tags = _extract(cleaned_tree, options, is_forum)
# try parsing wild
's tail, so an
#