marcocosta10000
723ca1506b
Adds `_fetch_page_content()`, which downloads a URL with a browser-like User-Agent and uses BeautifulSoup with the lxml parser to strip scripts, navigation, and other non-content markup, returning up to 3000 characters of readable text. The response validator now also compares generated replies against this scraped text, flagging named entities and claims that do not appear in the retrieved page content. Dependencies: adds beautifulsoup4==4.12.3 and pins lxml==5.3.0 for parser compatibility.
276 lines
9.1 KiB
Python
276 lines
9.1 KiB
Python
# cython: language_level=2
|
|
|
|
"""
|
|
SAX-based adapter to copy trees from/to the Python standard library.
|
|
|
|
Use the `ElementTreeContentHandler` class to build an ElementTree from
|
|
SAX events.
|
|
|
|
Use the `ElementTreeProducer` class or the `saxify()` function to fire
|
|
the SAX events of an ElementTree against a SAX ContentHandler.
|
|
|
|
See https://lxml.de/sax.html
|
|
"""
|
|
|
|
|
|
from xml.sax.handler import ContentHandler
|
|
from lxml import etree
|
|
from lxml.etree import ElementTree, SubElement
|
|
from lxml.etree import Comment, ProcessingInstruction
|
|
|
|
|
|
class SaxError(etree.LxmlError):
|
|
"""General SAX error.
|
|
"""
|
|
|
|
|
|
def _getNsTag(tag):
|
|
if tag[0] == '{':
|
|
return tuple(tag[1:].split('}', 1))
|
|
else:
|
|
return None, tag
|
|
|
|
|
|
class ElementTreeContentHandler(ContentHandler):
|
|
"""Build an lxml ElementTree from SAX events.
|
|
"""
|
|
def __init__(self, makeelement=None):
|
|
ContentHandler.__init__(self)
|
|
self._root = None
|
|
self._root_siblings = []
|
|
self._element_stack = []
|
|
self._default_ns = None
|
|
self._ns_mapping = { None : [None] }
|
|
self._new_mappings = {}
|
|
if makeelement is None:
|
|
makeelement = etree.Element
|
|
self._makeelement = makeelement
|
|
|
|
def _get_etree(self):
|
|
"Contains the generated ElementTree after parsing is finished."
|
|
return ElementTree(self._root)
|
|
|
|
etree = property(_get_etree, doc=_get_etree.__doc__)
|
|
|
|
def setDocumentLocator(self, locator):
|
|
pass
|
|
|
|
def startDocument(self):
|
|
pass
|
|
|
|
def endDocument(self):
|
|
pass
|
|
|
|
def startPrefixMapping(self, prefix, uri):
|
|
self._new_mappings[prefix] = uri
|
|
try:
|
|
self._ns_mapping[prefix].append(uri)
|
|
except KeyError:
|
|
self._ns_mapping[prefix] = [uri]
|
|
if prefix is None:
|
|
self._default_ns = uri
|
|
|
|
def endPrefixMapping(self, prefix):
|
|
ns_uri_list = self._ns_mapping[prefix]
|
|
ns_uri_list.pop()
|
|
if prefix is None:
|
|
self._default_ns = ns_uri_list[-1]
|
|
|
|
def _buildTag(self, ns_name_tuple):
|
|
ns_uri, local_name = ns_name_tuple
|
|
if ns_uri:
|
|
el_tag = "{%s}%s" % ns_name_tuple
|
|
elif self._default_ns:
|
|
el_tag = "{%s}%s" % (self._default_ns, local_name)
|
|
else:
|
|
el_tag = local_name
|
|
return el_tag
|
|
|
|
def startElementNS(self, ns_name, qname, attributes=None):
|
|
el_name = self._buildTag(ns_name)
|
|
if attributes:
|
|
attrs = {}
|
|
try:
|
|
iter_attributes = attributes.iteritems()
|
|
except AttributeError:
|
|
iter_attributes = attributes.items()
|
|
|
|
for name_tuple, value in iter_attributes:
|
|
if name_tuple[0]:
|
|
attr_name = "{%s}%s" % name_tuple
|
|
else:
|
|
attr_name = name_tuple[1]
|
|
attrs[attr_name] = value
|
|
else:
|
|
attrs = None
|
|
|
|
element_stack = self._element_stack
|
|
if self._root is None:
|
|
element = self._root = \
|
|
self._makeelement(el_name, attrs, self._new_mappings)
|
|
if self._root_siblings and hasattr(element, 'addprevious'):
|
|
for sibling in self._root_siblings:
|
|
element.addprevious(sibling)
|
|
del self._root_siblings[:]
|
|
else:
|
|
element = SubElement(element_stack[-1], el_name,
|
|
attrs, self._new_mappings)
|
|
element_stack.append(element)
|
|
|
|
self._new_mappings.clear()
|
|
|
|
def processingInstruction(self, target, data):
|
|
pi = ProcessingInstruction(target, data)
|
|
if self._root is None:
|
|
self._root_siblings.append(pi)
|
|
else:
|
|
self._element_stack[-1].append(pi)
|
|
|
|
def endElementNS(self, ns_name, qname):
|
|
element = self._element_stack.pop()
|
|
el_tag = self._buildTag(ns_name)
|
|
if el_tag != element.tag:
|
|
raise SaxError("Unexpected element closed: " + el_tag)
|
|
|
|
def startElement(self, name, attributes=None):
|
|
if attributes:
|
|
attributes = {(None, k): v for k, v in attributes.items()}
|
|
self.startElementNS((None, name), name, attributes)
|
|
|
|
def endElement(self, name):
|
|
self.endElementNS((None, name), name)
|
|
|
|
def characters(self, data):
|
|
last_element = self._element_stack[-1]
|
|
try:
|
|
# if there already is a child element, we must append to its tail
|
|
last_element = last_element[-1]
|
|
last_element.tail = (last_element.tail or '') + data
|
|
except IndexError:
|
|
# otherwise: append to the text
|
|
last_element.text = (last_element.text or '') + data
|
|
|
|
ignorableWhitespace = characters
|
|
|
|
|
|
class ElementTreeProducer:
|
|
"""Produces SAX events for an element and children.
|
|
"""
|
|
def __init__(self, element_or_tree, content_handler):
|
|
try:
|
|
element = element_or_tree.getroot()
|
|
except AttributeError:
|
|
element = element_or_tree
|
|
self._element = element
|
|
self._content_handler = content_handler
|
|
from xml.sax.xmlreader import AttributesNSImpl as attr_class
|
|
self._attr_class = attr_class
|
|
self._empty_attributes = attr_class({}, {})
|
|
|
|
def saxify(self):
|
|
self._content_handler.startDocument()
|
|
|
|
element = self._element
|
|
if hasattr(element, 'getprevious'):
|
|
siblings = []
|
|
sibling = element.getprevious()
|
|
while getattr(sibling, 'tag', None) is ProcessingInstruction:
|
|
siblings.append(sibling)
|
|
sibling = sibling.getprevious()
|
|
for sibling in siblings[::-1]:
|
|
self._recursive_saxify(sibling, {})
|
|
|
|
self._recursive_saxify(element, {})
|
|
|
|
if hasattr(element, 'getnext'):
|
|
sibling = element.getnext()
|
|
while getattr(sibling, 'tag', None) is ProcessingInstruction:
|
|
self._recursive_saxify(sibling, {})
|
|
sibling = sibling.getnext()
|
|
|
|
self._content_handler.endDocument()
|
|
|
|
def _recursive_saxify(self, element, parent_nsmap):
|
|
content_handler = self._content_handler
|
|
tag = element.tag
|
|
if tag is Comment or tag is ProcessingInstruction:
|
|
if tag is ProcessingInstruction:
|
|
content_handler.processingInstruction(
|
|
element.target, element.text)
|
|
tail = element.tail
|
|
if tail:
|
|
content_handler.characters(tail)
|
|
return
|
|
|
|
element_nsmap = element.nsmap
|
|
new_prefixes = []
|
|
if element_nsmap != parent_nsmap:
|
|
# There have been updates to the namespace
|
|
for prefix, ns_uri in element_nsmap.items():
|
|
if parent_nsmap.get(prefix) != ns_uri:
|
|
new_prefixes.append( (prefix, ns_uri) )
|
|
|
|
attribs = element.items()
|
|
if attribs:
|
|
attr_values = {}
|
|
attr_qnames = {}
|
|
for attr_ns_name, value in attribs:
|
|
attr_ns_tuple = _getNsTag(attr_ns_name)
|
|
attr_values[attr_ns_tuple] = value
|
|
attr_qnames[attr_ns_tuple] = self._build_qname(
|
|
attr_ns_tuple[0], attr_ns_tuple[1], element_nsmap,
|
|
preferred_prefix=None, is_attribute=True)
|
|
sax_attributes = self._attr_class(attr_values, attr_qnames)
|
|
else:
|
|
sax_attributes = self._empty_attributes
|
|
|
|
ns_uri, local_name = _getNsTag(tag)
|
|
qname = self._build_qname(
|
|
ns_uri, local_name, element_nsmap, element.prefix, is_attribute=False)
|
|
|
|
for prefix, uri in new_prefixes:
|
|
content_handler.startPrefixMapping(prefix, uri)
|
|
content_handler.startElementNS(
|
|
(ns_uri, local_name), qname, sax_attributes)
|
|
text = element.text
|
|
if text:
|
|
content_handler.characters(text)
|
|
for child in element:
|
|
self._recursive_saxify(child, element_nsmap)
|
|
content_handler.endElementNS((ns_uri, local_name), qname)
|
|
for prefix, uri in new_prefixes:
|
|
content_handler.endPrefixMapping(prefix)
|
|
tail = element.tail
|
|
if tail:
|
|
content_handler.characters(tail)
|
|
|
|
def _build_qname(self, ns_uri, local_name, nsmap, preferred_prefix, is_attribute):
|
|
if ns_uri is None:
|
|
return local_name
|
|
|
|
if not is_attribute and nsmap.get(preferred_prefix) == ns_uri:
|
|
prefix = preferred_prefix
|
|
else:
|
|
# Pick the first matching prefix, in alphabetical order.
|
|
candidates = [
|
|
pfx for (pfx, uri) in nsmap.items()
|
|
if pfx is not None and uri == ns_uri
|
|
]
|
|
prefix = (
|
|
candidates[0] if len(candidates) == 1
|
|
else min(candidates) if candidates
|
|
else None
|
|
)
|
|
|
|
if prefix is None:
|
|
# Default namespace
|
|
return local_name
|
|
return prefix + ':' + local_name
|
|
|
|
|
|
def saxify(element_or_tree, content_handler):
|
|
"""One-shot helper to generate SAX events from an XML tree and fire
|
|
them against a SAX ContentHandler.
|
|
"""
|
|
return ElementTreeProducer(element_or_tree, content_handler).saxify()
|