element.""" title \= "" title\_element \= tree.find(".//head//title") if title\_element is not None: title \= trim(title\_element.text\_content()) if match := HTMLTITLE\_REGEX.match(title): return title, match\[1\], match\[2\] LOGGER.debug("no main title found") return title, None, None def extract\_title(tree: HtmlElement) \-> Optional\[str\]: """Extract the document title""" \# only one h1-element: take it h1\_results \= tree.findall(".//h1") if len(h1\_results) \== 1: title \= trim(h1\_results\[0\].text\_content()) if title: return title \# extract using x-paths title \= extract\_metainfo(tree, TITLE\_XPATHS) or "" if title: return title \# extract using title tag title, first, second \= examine\_title\_element(tree) for t in (first, second): if t and "." not in t: return t \# take first h1-title if h1\_results: return h1\_results\[0\].text\_content() \# take first h2-title try: title \= tree.xpath(".//h2")\[0\].text\_content() except IndexError: LOGGER.debug("no h2 title found") return title def extract\_author(tree: HtmlElement) \-> Optional\[str\]: """Extract the document author(s)""" subtree \= prune\_unwanted\_nodes(deepcopy(tree), AUTHOR\_DISCARD\_XPATHS) author \= extract\_metainfo(subtree, AUTHOR\_XPATHS, len\_limit\=120) if author: author \= normalize\_authors(None, author) \# copyright? return author def extract\_url(tree: HtmlElement, default\_url: Optional\[str\] \= None) \-> Optional\[str\]: """Extract the URL from the canonical link""" for selector in URL\_SELECTORS: element \= tree.find(selector) url \= element.attrib.get("href") if element is not None else None if url: break \# fix relative URLs if url and url.startswith("/"): for element in tree.iterfind(".//head//meta\[@content\]"): attrtype \= element.get("name") or element.get("property") or "" if attrtype.startswith("og:") or attrtype.startswith("twitter:"): base\_url \= get\_base\_url(element.attrib\["content"\]) if base\_url: \# prepend URL url \= base\_url + url break \# do not return invalid URLs if url: validation\_result, parsed\_url \= validate\_url(url) url \= normalize\_url(parsed\_url) if validation\_result else None return url or default\_url def extract\_sitename(tree: HtmlElement) \-> Optional\[str\]: """Extract the name of a site from the main title (if it exists)""" \_, \*parts \= examine\_title\_element(tree) return next((part for part in parts if part and "." in part), None) def extract\_catstags(metatype: str, tree: HtmlElement) \-> List\[str\]: """Find category and tag information""" results: List\[str\] \= \[\] regexpr \= "/" + metatype + "\[s|ies\]?/" xpath\_expression \= CATEGORIES\_XPATHS if metatype \== "category" else TAGS\_XPATHS \# search using custom expressions for catexpr in xpath\_expression: results.extend( elem.text\_content() for elem in catexpr(tree) if re.search(regexpr, elem.attrib\["href"\]) ) if results: break \# category fallback if metatype \== "category" and not results: for element in tree.xpath( './/head//meta\[@property="article:section" or contains(@name, "subject")\]\[@content\]' ): results.append(element.attrib\["content"\]) \# optional: search through links \# if not results: \# for elem in tree.xpath('.//a\[@href\]'): \# search for 'category' return \[r for r in dict.fromkeys(line\_processing(x) for x in results if x) if r\] def parse\_license\_element(element: HtmlElement, strict: bool \= False) \-> Optional\[str\]: """Probe a link for identifiable free license cues. Parse the href attribute first and then the link text.""" \# look for Creative Commons elements match \= LICENSE\_REGEX.search(element.get("href", "")) if match: return f"CC {match\[1\].upper()} {match\[2\]}" if element.text: \# check if it could be a CC license if strict: match \= TEXT\_LICENSE\_REGEX.search(element.text) return match\[0\] if match else None return trim(element.text) return None def extract\_license(tree: HtmlElement) \-> Optional\[str\]: """Search the HTML code for license information and parse it.""" \# look for links labeled as license for element in tree.findall('.//a\[@rel="license"\]\[@href\]'): result \= parse\_license\_element(element, strict\=False) if result is not None: return result \# probe footer elements for CC links for element in tree.xpath( './/footer//a\[@href\]|.//div\[contains(@class, "footer") or contains(@id, "footer")\]//a\[@href\]' ): result \= parse\_license\_element(element, strict\=True) if result is not None: return result return None [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.extract_metadata) def extract\_metadata( filecontent: Union\[HtmlElement, str\], default\_url: Optional\[str\] \= None, date\_config: Optional\[Any\] \= None, extensive: bool \= True, author\_blacklist: Optional\[Set\[str\]\] \= None, ) \-> Document: """Main process for metadata extraction. Args: filecontent: HTML code as string or parsed tree. default\_url: Previously known URL of the downloaded document. date\_config: Provide extraction parameters to htmldate as dict(). author\_blacklist: Provide a blacklist of Author Names as set() to filter out authors. Returns: A trafilatura.settings.Document containing the extracted metadata information or None. The Document class has .as\_dict() method that will return a copy as a dict. """ \# init author\_blacklist \= author\_blacklist or set() date\_config \= date\_config or set\_date\_params(extensive) \# load contents tree \= load\_html(filecontent) if tree is None: return Document() \# initialize dict and try to strip meta tags metadata \= examine\_meta(tree) \# to check: remove it and replace with author\_blacklist in test case if metadata.author and " " not in metadata.author: metadata.author \= None \# fix: try json-ld metadata and override try: metadata \= extract\_meta\_json(tree, metadata) except Exception as err: \# bugs in json\_metadata.py LOGGER.warning("error in JSON metadata extraction: %s", err) \# title if not metadata.title: metadata.title \= extract\_title(tree) \# check author in blacklist if metadata.author and author\_blacklist: metadata.author \= check\_authors(metadata.author, author\_blacklist) \# author if not metadata.author: metadata.author \= extract\_author(tree) \# recheck author in blacklist if metadata.author and author\_blacklist: metadata.author \= check\_authors(metadata.author, author\_blacklist) \# url if not metadata.url: metadata.url \= extract\_url(tree, default\_url) \# hostname if metadata.url: metadata.hostname \= extract\_domain(metadata.url, fast\=True) \# extract date with external module htmldate date\_config\["url"\] \= metadata.url metadata.date \= find\_date(tree, \*\*date\_config) \# sitename if not metadata.sitename: metadata.sitename \= extract\_sitename(tree) if metadata.sitename: \# fix: take 1st element (\['Westdeutscher Rundfunk'\]) if isinstance(metadata.sitename, list): metadata.sitename \= metadata.sitename\[0\] \# hotfix: probably an error coming from json\_metadata (#195) elif isinstance(metadata.sitename, dict): metadata.sitename \= str(metadata.sitename) \# scrap Twitter ID metadata.sitename \= metadata.sitename.lstrip("@") \# capitalize if ( metadata.sitename and "." not in metadata.sitename and not metadata.sitename\[0\].isupper() ): metadata.sitename \= metadata.sitename.title() \# use URL elif metadata.url: mymatch \= META\_URL.match(metadata.url) if mymatch: metadata.sitename \= mymatch\[1\] \# categories if not metadata.categories: metadata.categories \= extract\_catstags("category", tree) \# tags if not metadata.tags: metadata.tags \= extract\_catstags("tag", tree) \# license metadata.license \= extract\_license(tree) \# safety checks metadata.filedate \= date\_config\["max\_date"\] metadata.clean\_and\_trim() return metadata --- # trafilatura.spider — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/spider.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.spider ================================== \# pylint:disable-msg=E0611,E1101,I1101 """ Functions dedicated to website navigation and crawling/spidering. """ import logging from configparser import ConfigParser from time import sleep from typing import List, Optional, Tuple from urllib.robotparser import RobotFileParser from courlan import ( UrlStore, extract\_links, fix\_relative\_urls, get\_base\_url, is\_navigation\_page, is\_not\_crawlable, ) try: import py3langid \# type: ignore except ImportError: pass from lxml.etree import XPath, tostring from .core import baseline, prune\_unwanted\_nodes from .downloads import Response, fetch\_response, fetch\_url from .settings import DEFAULT\_CONFIG from .utils import LANGID\_FLAG, decode\_file, load\_html LOGGER \= logging.getLogger(\_\_name\_\_) URL\_STORE \= UrlStore(compressed\=False, strict\=False) ROBOTS\_TXT\_URL \= "/robots.txt" MAX\_SEEN\_URLS \= 10 MAX\_KNOWN\_URLS \= 100000 class CrawlParameters: "Store necessary information to manage a focused crawl." \_\_slots\_\_ \= \["start", "base", "lang", "rules", "ref", "i", "known\_num", "is\_on", "prune\_xpath"\] def \_\_init\_\_( self, start: str, lang: Optional\[str\] \= None, rules: Optional\[RobotFileParser\] \= None, prune\_xpath: Optional\[str\] \= None, ) \-> None: self.start: str \= start self.base: str \= self.\_get\_base\_url(start) self.ref: str \= self.\_get\_reference(start) self.lang: Optional\[str\] \= lang self.rules: Optional\[RobotFileParser\] \= rules or get\_rules(self.base) self.i: int \= 0 self.known\_num: int \= 0 self.is\_on: bool \= True self.prune\_xpath: Optional\[str\] \= prune\_xpath def \_get\_base\_url(self, start: str) \-> str: "Set reference domain for the crawl." base: str \= get\_base\_url(start) if not base: raise ValueError(f"cannot start crawl: {start}") return base def \_get\_reference(self, start: str) \-> str: "Determine the reference URL." return start.rsplit("/", 1)\[0\] if start.count("/") \>= 3 else start def update\_metadata(self, url\_store: UrlStore) \-> None: "Adjust crawl data based on URL store info." self.is\_on \= bool(url\_store.find\_unvisited\_urls(self.base)) self.known\_num \= len(url\_store.find\_known\_urls(self.base)) def filter\_list(self, todo: Optional\[List\[str\]\]) \-> List\[str\]: "Prepare the todo list, excluding invalid URLs." if not todo: return \[\] return \[u for u in todo if u != self.start and self.ref in u\] def is\_valid\_link(self, link: str) \-> bool: "Run checks: robots.txt rules, URL type and crawl breadth." return ( (not self.rules or self.rules.can\_fetch("\*", link)) and self.ref in link and not is\_not\_crawlable(link) ) def refresh\_detection( htmlstring: str, homepage: str ) \-> Tuple\[Optional\[str\], Optional\[str\]\]: "Check if there could be a redirection by meta-refresh tag." if '"refresh"' not in htmlstring and '"REFRESH"' not in htmlstring: return htmlstring, homepage html\_tree \= load\_html(htmlstring) if html\_tree is None: return htmlstring, homepage \# test meta-refresh redirection \# https://stackoverflow.com/questions/2318446/how-to-follow-meta-refreshes-in-python results \= html\_tree.xpath( './/meta\[@http-equiv="refresh" or @http-equiv="REFRESH"\]/@content' ) result \= results\[0\] if results else "" if not result or ";" not in result: logging.info("no redirect found: %s", homepage) return htmlstring, homepage url2 \= result.split(";")\[1\].strip().lower().replace("url=", "") if not url2.startswith("http"): \# Relative URL, adapt base\_url \= get\_base\_url(url2) url2 \= fix\_relative\_urls(base\_url, url2) \# second fetch newhtmlstring \= fetch\_url(url2) if newhtmlstring is None: logging.warning("failed redirect: %s", url2) return None, None \# else: logging.info("successful redirect: %s", url2) return newhtmlstring, url2 def probe\_alternative\_homepage( homepage: str, ) \-> Tuple\[Optional\[str\], Optional\[str\], Optional\[str\]\]: "Check if the homepage is redirected and return appropriate values." response \= fetch\_response(homepage, decode\=False) if not response or not response.data: return None, None, None \# get redirected URL here? if response.url not in (homepage, "/"): logging.info("followed homepage redirect: %s", response.url) homepage \= response.url \# decode response htmlstring \= decode\_file(response.data) \# is there a meta-refresh on the page? new\_htmlstring, new\_homepage \= refresh\_detection(htmlstring, homepage) if new\_homepage is None: \# malformed or malicious content return None, None, None logging.debug("fetching homepage OK: %s", new\_homepage) return new\_htmlstring, new\_homepage, get\_base\_url(new\_homepage) def parse\_robots(robots\_url: str, data: str) \-> Optional\[RobotFileParser\]: "Parse a robots.txt file with the standard library urllib.robotparser." \# https://github.com/python/cpython/blob/main/Lib/urllib/robotparser.py rules \= RobotFileParser() rules.set\_url(robots\_url) \# exceptions happening here try: rules.parse(data.splitlines()) except Exception as exc: LOGGER.error("cannot read robots.txt: %s", exc) return None return rules def get\_rules(base\_url: str) \-> Optional\[RobotFileParser\]: "Attempt to fetch and parse robots.txt file for a given website." robots\_url \= base\_url + ROBOTS\_TXT\_URL data \= fetch\_url(robots\_url) return parse\_robots(robots\_url, data) if data else None def is\_target\_language(htmlstring: str, language: Optional\[str\]) \-> bool: """Run a baseline extraction and use a language detector to check if the content matches the target language. Return True if language checks are bypassed.""" if htmlstring and language and LANGID\_FLAG: \_, text, \_ \= baseline(htmlstring) result, \_ \= py3langid.classify(text) return bool(result \== language) return True def is\_still\_navigation(todo: List\[str\]) \-> bool: """Probe if there are still navigation URLs in the queue.""" return any(is\_navigation\_page(url) for url in todo) def process\_links( htmlstring: str, params: CrawlParameters, url: Optional\[str\] \= "", ) \-> None: """Examine the HTML code and process the retrieved internal links. Extract and filter new internal links after an optional language check. Store the links in todo-list while prioritizing the navigation ones.""" if not is\_target\_language(htmlstring, params.lang): return if htmlstring and params.prune\_xpath is not None: if isinstance(params.prune\_xpath, str): params.prune\_xpath \= \[params.prune\_xpath\] \# type: ignore\[assignment\] tree \= load\_html(htmlstring) if tree is not None: tree \= prune\_unwanted\_nodes(tree, \[XPath(x) for x in params.prune\_xpath\]) htmlstring \= tostring(tree).decode() links, links\_priority \= \[\], \[\] for link in extract\_links( pagecontent\=htmlstring, url\=url or params.base, external\_bool\=False, language\=params.lang, with\_nav\=True, strict\=False, ): if not params.is\_valid\_link(link): continue if is\_navigation\_page(link): links\_priority.append(link) else: links.append(link) URL\_STORE.add\_urls(urls\=links, appendleft\=links\_priority) def process\_response( response: Optional\[Response\], params: CrawlParameters, ) \-> None: """Convert urllib3 response object and extract links.""" if response is None or not response.data: return \# add final document URL to known\_links URL\_STORE.add\_urls(\[response.url\], visited\=True) \# convert urllib3 response to string and proceed to link extraction process\_links(decode\_file(response.data), params, params.base) def init\_crawl( start: str, lang: Optional\[str\] \= None, rules: Optional\[RobotFileParser\] \= None, todo: Optional\[List\[str\]\] \= None, known: Optional\[List\[str\]\] \= None, prune\_xpath: Optional\[str\] \= None, ) \-> CrawlParameters: """Initialize crawl by setting variables, copying values to the URL store and retrieving the initial page if the crawl starts.""" params \= CrawlParameters(start, lang, rules, prune\_xpath) \# todo: just known or also visited? URL\_STORE.add\_urls(urls\=known or \[\], visited\=True) URL\_STORE.add\_urls(urls\=params.filter\_list(todo)) URL\_STORE.store\_rules(params.base, params.rules) \# visiting the start page if necessary if not todo: URL\_STORE.add\_urls(urls\=\[params.start\], visited\=False) params \= crawl\_page(params, initial\=True) else: params.update\_metadata(URL\_STORE) return params def crawl\_page( params: CrawlParameters, initial: bool \= False, ) \-> CrawlParameters: """Examine a webpage, extract navigation links and links.""" \# config=DEFAULT\_CONFIG url \= URL\_STORE.get\_url(params.base) if not url: params.is\_on \= False params.known\_num \= len(URL\_STORE.find\_known\_urls(params.base)) return params params.i += 1 if initial: \# probe and process homepage htmlstring, homepage, new\_base\_url \= probe\_alternative\_homepage(url) if htmlstring and homepage and new\_base\_url: \# register potentially new homepage URL\_STORE.add\_urls(\[homepage\]) \# extract links on homepage process\_links(htmlstring, params, url\=url) else: response \= fetch\_response(url, decode\=False) process\_response(response, params) \# optional backup of gathered pages without nav-pages ? ... params.update\_metadata(URL\_STORE) return params [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.spider.focused_crawler) def focused\_crawler( homepage: str, max\_seen\_urls: int \= MAX\_SEEN\_URLS, max\_known\_urls: int \= MAX\_KNOWN\_URLS, todo: Optional\[List\[str\]\] \= None, known\_links: Optional\[List\[str\]\] \= None, lang: Optional\[str\] \= None, config: ConfigParser \= DEFAULT\_CONFIG, rules: Optional\[RobotFileParser\] \= None, prune\_xpath: Optional\[str\] \= None, ) \-> Tuple\[List\[str\], List\[str\]\]: """Basic crawler targeting pages of interest within a website. Args: homepage: URL of the page to first page to fetch, preferably the homepage of a website. max\_seen\_urls: maximum number of pages to visit, stop iterations at this number or at the exhaustion of pages on the website, whichever comes first. max\_known\_urls: stop if the total number of pages "known" exceeds this number. todo: provide a previously generated list of pages to visit / crawl frontier. known\_links: provide a list of previously known pages. lang: try to target links according to language heuristics. config: use a different configuration (configparser format). rules: provide politeness rules (urllib.robotparser.RobotFileParser() format). prune\_xpath: remove unwanted elements from the HTML pages using XPath. Returns: List of pages to visit, deque format, possibly empty if there are no further pages to visit. Set of known links. """ params \= init\_crawl(homepage, lang, rules, todo, known\_links, prune\_xpath) sleep\_time \= URL\_STORE.get\_crawl\_delay( params.base, default\=config.getfloat("DEFAULT", "SLEEP\_TIME") ) \# visit pages until a limit is reached while ( params.is\_on and params.i < max\_seen\_urls and params.known\_num < max\_known\_urls ): params \= crawl\_page(params) sleep(sleep\_time) \# refocus todo-list on URLs without navigation? todo \= list(dict.fromkeys(URL\_STORE.find\_unvisited\_urls(params.base))) \# \[u for u in todo if not is\_navigation\_page(u)\] known\_links \= list(dict.fromkeys(URL\_STORE.find\_known\_urls(params.base))) return todo, known\_links --- # trafilatura.utils — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/utils.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.utils ================================= \# pylint:disable-msg=E0611,I1101 """ Module bundling functions related to HTML and text processing, content filtering and language detection. """ try: import gzip HAS\_GZIP \= True except ImportError: HAS\_GZIP \= False import logging import re try: import zlib HAS\_ZLIB \= True except ImportError: HAS\_ZLIB \= False from functools import lru\_cache from itertools import islice from typing import Any, List, Literal, Optional, Tuple, Union from unicodedata import normalize \# response compression try: import brotli \# type: ignore HAS\_BROTLI \= True except ImportError: HAS\_BROTLI \= False try: import zstandard HAS\_ZSTD \= True except ImportError: HAS\_ZSTD \= False \# language detection try: import py3langid \# type: ignore LANGID\_FLAG \= True except ImportError: LANGID\_FLAG \= False \# CChardet is faster and can be more accurate try: from cchardet import detect as cchardet\_detect \# type: ignore except ImportError: cchardet\_detect \= None from charset\_normalizer import from\_bytes from lxml.etree import \_Element from lxml.html import HtmlElement, HTMLParser, fromstring \# response types from urllib3.response import HTTPResponse LOGGER \= logging.getLogger(\_\_name\_\_) UNICODE\_ALIASES \= {'utf-8', 'utf\_8'} DOCTYPE\_TAG \= re.compile("^< ?! ?DOCTYPE.+?/ ?>", re.I) FAULTY\_HTML \= re.compile(r"(<html.\*?)\\s\*/>", re.I) HTML\_STRIP\_TAGS \= re.compile(r'(|<\[^>\]\*>)') \# note: htmldate could use HTML comments \# huge\_tree=True, remove\_blank\_text=True HTML\_PARSER \= HTMLParser(collect\_ids\=False, default\_doctype\=False, encoding\='utf-8', remove\_comments\=True, remove\_pis\=True) LINES\_TRIMMING \= re.compile(r'(?<!\[p{P}\>\])\\n', flags\=re.UNICODE|re.MULTILINE) URL\_BLACKLIST\_REGEX \= re.compile(r'^https?://|/+$') \# Regex to check image file extensions IMAGE\_EXTENSION \= re.compile(r'\[^\\s\]+\\.(avif|bmp|gif|hei\[cf\]|jpe?g|png|webp)(\\b|$)') FORMATTING\_PROTECTED \= {'cell', 'head', 'hi', 'item', 'p', 'quote', 'ref', 'td'} SPACING\_PROTECTED \= {'code', 'pre'} \# https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Language TARGET\_LANG\_ATTRS \= ('http-equiv="content-language"', 'property="og:locale"') RE\_HTML\_LANG \= re.compile(r'(\[a-z\]{2})') \# Mostly filters for social media RE\_FILTER \= re.compile(r'\\W\*(Drucken|E-?Mail|Facebook|Flipboard|Google|Instagram|' 'Linkedin|Mail|PDF|Pinterest|Pocket|Print|QQ|Reddit|Twitter|' 'WeChat|WeiBo|Whatsapp|Xing|Mehr zum Thema:?|More on this.{,8}$)$', flags\=re.IGNORECASE) \# COMMENTS\_BLACKLIST = ('( Abmelden / Ändern )') # Fill in your details below|Trage deine Daten unten|Kommentar verfassen|Bitte logge dich|Hinterlasse einen Kommentar| to %s| mit %s) def handle\_compressed\_file(filecontent: bytes) \-> bytes: """ Don't trust response headers and try to decompress a binary string with a cascade of installed packages. Use magic numbers when available. """ if not isinstance(filecontent, bytes): return filecontent \# source: https://stackoverflow.com/questions/3703276/how-to-tell-if-a-file-is-gzip-compressed if HAS\_GZIP and filecontent\[:3\] \== b"\\x1f\\x8b\\x08": try: return gzip.decompress(filecontent) except Exception: \# EOFError, OSError, gzip.BadGzipFile LOGGER.warning("invalid GZ file") \# try zstandard if HAS\_ZSTD and filecontent\[:4\] \== b"\\x28\\xb5\\x2f\\xfd": try: return zstandard.decompress(filecontent) \# max\_output\_size=??? except zstandard.ZstdError: LOGGER.warning("invalid ZSTD file") \# try brotli if HAS\_BROTLI: try: return brotli.decompress(filecontent) \# type: ignore\[no-any-return\] except brotli.error: pass \# logging.debug('invalid Brotli file') \# try zlib/deflate if HAS\_ZLIB: try: return zlib.decompress(filecontent) except zlib.error: pass \# return content unchanged if decompression failed return filecontent def isutf8(data: bytes) \-> bool: """Simple heuristic to determine if a bytestring uses standard unicode encoding""" try: data.decode('UTF-8') except UnicodeDecodeError: return False return True def detect\_encoding(bytesobject: bytes) \-> List\[str\]: """"Read all input or first chunk and return a list of encodings""" \# alternatives: https://github.com/scrapy/w3lib/blob/master/w3lib/encoding.py \# unicode-test if isutf8(bytesobject): return \['utf-8'\] guesses \= \[\] \# additional module if cchardet\_detect is not None: cchardet\_guess \= cchardet\_detect(bytesobject)\['encoding'\] if cchardet\_guess is not None: guesses.append(cchardet\_guess.lower()) \# try charset\_normalizer on first part, fallback on full document if len(bytesobject) < 10000: detection\_results \= from\_bytes(bytesobject) else: detection\_results \= from\_bytes(bytesobject\[:5000\] + bytesobject\[\-5000:\]) or \\ from\_bytes(bytesobject) \# return alternatives if len(detection\_results) \> 0: guesses.extend(\[r.encoding for r in detection\_results\]) \# it cannot be utf-8 (tested above) return \[g for g in guesses if g not in UNICODE\_ALIASES\] [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.decode_file) def decode\_file(filecontent: Union\[bytes, str\]) \-> str: """Check if the bytestring could be GZip and eventually decompress it, guess bytestring encoding and try to decode to Unicode string. Resort to destructive conversion otherwise.""" if isinstance(filecontent, str): return filecontent htmltext \= None \# GZip and Brotli test filecontent \= handle\_compressed\_file(filecontent) \# encoding for guessed\_encoding in detect\_encoding(filecontent): try: htmltext \= filecontent.decode(guessed\_encoding) except (LookupError, UnicodeDecodeError): \# VISCII: lookup LOGGER.warning('wrong encoding detected: %s', guessed\_encoding) htmltext \= None else: break \# return original content if nothing else succeeded return htmltext or str(filecontent, encoding\='utf-8', errors\='replace') def is\_dubious\_html(beginning: str) \-> bool: "Assess if the object is proper HTML (awith a corresponding tag or declaration)." return "html" not in beginning def repair\_faulty\_html(htmlstring: str, beginning: str) \-> str: "Repair faulty HTML strings to make then palatable for libxml2." \# libxml2/LXML issue: https://bugs.launchpad.net/lxml/+bug/1955915 if "doctype" in beginning: firstline, \_, rest \= htmlstring.partition("\\n") htmlstring \= DOCTYPE\_TAG.sub("", firstline, count\=1) + "\\n" + rest \# other issue with malformed documents: check first three lines for i, line in enumerate(iter(htmlstring.splitlines())): if "<html" in line and line.endswith("/>"): htmlstring \= FAULTY\_HTML.sub(r"\\1>", htmlstring, count\=1) break if i \> 2: break return htmlstring def fromstring\_bytes(htmlobject: str) \-> Optional\[HtmlElement\]: "Try to pass bytes to LXML parser." tree \= None try: tree \= fromstring(htmlobject.encode("utf8", "surrogatepass"), parser\=HTML\_PARSER) except Exception as err: LOGGER.error("lxml parser bytestring %s", err) return tree [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.load_html) def load\_html(htmlobject: Any) \-> Optional\[HtmlElement\]: """Load object given as input and validate its type (accepted: lxml.html tree, trafilatura/urllib3 response, bytestring and string) """ \# use tree directly if isinstance(htmlobject, HtmlElement): return htmlobject \# use trafilatura or urllib3 responses directly if isinstance(htmlobject, HTTPResponse) or hasattr(htmlobject, "data"): htmlobject \= htmlobject.data \# do not accept any other type after this point if not isinstance(htmlobject, (bytes, str)): raise TypeError("incompatible input type", type(htmlobject)) \# start processing tree \= None \# try to guess encoding and decode file: if None then keep original htmlobject \= decode\_file(htmlobject) \# sanity checks beginning \= htmlobject\[:50\].lower() check\_flag \= is\_dubious\_html(beginning) \# repair first htmlobject \= repair\_faulty\_html(htmlobject, beginning) \# first pass: use Unicode string fallback\_parse \= False try: tree \= fromstring(htmlobject, parser\=HTML\_PARSER) except ValueError: \# "Unicode strings with encoding declaration are not supported." tree \= fromstring\_bytes(htmlobject) fallback\_parse \= True except Exception as err: \# pragma: no cover LOGGER.error("lxml parsing failed: %s", err) \# second pass: try passing bytes to LXML if (tree is None or len(tree) < 1) and not fallback\_parse: tree \= fromstring\_bytes(htmlobject) \# rejection test: is it (well-formed) HTML at all? \# log parsing errors if tree is not None and check\_flag is True and len(tree) < 2: LOGGER.error( "parsed tree length: %s, wrong data type or not valid HTML", len(tree) ) tree \= None return tree @lru\_cache(maxsize\=2\*\*14) \# sys.maxunicode = 1114111 def return\_printables\_and\_spaces(char: str) \-> str: 'Return a character if it belongs to certain classes' return char if char.isprintable() or char.isspace() else '' def remove\_control\_characters(string: str) \-> str: '''Prevent non-printable and XML invalid character errors''' return ''.join(map(return\_printables\_and\_spaces, string)) def normalize\_unicode(string: str, unicodeform: Literal\['NFC', 'NFD', 'NFKC', 'NFKD'\] \= 'NFC') \-> str: 'Normalize the given string to the specified unicode format.' return normalize(unicodeform, string) @lru\_cache(maxsize\=1024) def line\_processing(line: str, preserve\_space: bool \= False, trailing\_space: bool \= False) \-> Optional\[str\]: '''Remove HTML space entities, then discard incompatible unicode and invalid XML characters on line level''' \# spacing HTML entities: https://www.w3.org/MarkUp/html-spec/html-spec\_13.html \# unique code spaces new\_line \= remove\_control\_characters(line.replace(' ', '\\r').replace(' ', '\\n').replace(' ', '\\u00A0')) if not preserve\_space: \# remove newlines that are not related to punctuation or markup \# remove non-printable chars and normalize space characters (including Unicode spaces) new\_line \= trim(LINES\_TRIMMING.sub(r" ", new\_line)) \# prune empty lines if all(map(str.isspace, new\_line)): new\_line \= None \# type: ignore\[assignment\] elif trailing\_space: space\_before \= " " if line\[0\].isspace() else "" space\_after \= " " if line\[\-1\].isspace() else "" new\_line \= "".join(\[space\_before, new\_line, space\_after\]) return new\_line [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.sanitize) def sanitize(text: str, preserve\_space: bool \= False, trailing\_space: bool \= False) \-> Optional\[str\]: '''Convert text and discard incompatible and invalid characters''' \# consider all text as a single line if trailing\_space: return line\_processing(text, preserve\_space, True) \# process line by line try: return '\\n'.join(filter(None, (line\_processing(l, preserve\_space) for l in text.splitlines()))).replace('\\u2424', '') except AttributeError: return None def sanitize\_tree(tree: \_Element) \-> \_Element: '''Trims spaces, removes control characters and normalizes unicode''' for elem in tree.iter(): parent \= elem.getparent() parent\_tag \= parent.tag if parent is not None else "" \# preserve space if the element or its parent is a specific tag, or if the element has text and children \# the last part is relevant for item elements with ref inside for example preserve\_space \= elem.tag in SPACING\_PROTECTED or parent\_tag in SPACING\_PROTECTED trailing\_space \= elem.tag in FORMATTING\_PROTECTED or parent\_tag in FORMATTING\_PROTECTED or preserve\_space \# remove invalid attributes for attribute in elem.attrib: if ':' in attribute: \# colon is reserved for namespaces in XML if not elem.attrib\[attribute\] or attribute.split(':', 1)\[0\] not in tree.nsmap: elem.attrib.pop(attribute) if elem.text: elem.text \= sanitize(elem.text, preserve\_space, trailing\_space) if elem.tail: elem.tail \= sanitize(elem.tail, preserve\_space, trailing\_space) return tree [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.trim) @lru\_cache(maxsize\=1024) def trim(string: str) \-> str: "Remove unnecessary spaces within a text string." try: \# remove newlines that are not related to punctuation or markup + proper trimming return " ".join(string.split()).strip() except (AttributeError, TypeError): return "" def is\_image\_element(element: \_Element) \-> bool: '''Check if an element is a valid img element''' for attr in ("data-src", "src"): src \= element.get(attr, "") if is\_image\_file(src): return True else: \# take the first corresponding attribute for attr, value in element.attrib.items(): if attr.startswith("data-src") and is\_image\_file(value): return True return False def is\_image\_file(imagesrc: Optional\[str\]) \-> bool: '''Check if the observed string corresponds to a valid image extension. Use a length threshold and apply a regex on the content.''' if imagesrc is None or len(imagesrc) \> 8192: return False return bool(IMAGE\_EXTENSION.search(imagesrc)) def make\_chunks(iterable: Any, n: int) \-> Any: "Chunk data into smaller pieces." \# 3.12+: https://docs.python.org/3/library/itertools.html#itertools.batched iterator \= iter(iterable) while batch := tuple(islice(iterator, n)): yield batch def is\_acceptable\_length(my\_len: int, options: Any) \-> bool: "Check if the document length is within acceptable boundaries." if my\_len < options.min\_file\_size: LOGGER.error("too small/incorrect for URL %s", options.url) return False if my\_len \> options.max\_file\_size: LOGGER.error("too large: length %s for URL %s", my\_len, options.url) return False return True def check\_html\_lang(tree: HtmlElement, target\_language: str, strict: bool \= False) \-> bool: """Check HTML meta-elements for language information and split the result in case there are several languages.""" for attr in TARGET\_LANG\_ATTRS: elems \= tree.findall(f'.//meta\[@{attr}\]\[@content\]') if elems: if any(target\_language in RE\_HTML\_LANG.split(elem.get("content", "").lower()) for elem in elems): return True LOGGER.debug("%s lang attr failed", attr) return False \# HTML lang attribute: sometimes a wrong indication if strict: elems \= tree.xpath("//html\[@lang\]") if elems: if any(target\_language in RE\_HTML\_LANG.split(elem.get("lang", "").lower()) for elem in elems): return True LOGGER.debug("HTML lang failed") return False LOGGER.debug("No relevant lang elements found") return True def language\_classifier(temp\_text: str, temp\_comments: str) \-> Optional\[str\]: '''Run external component (if installed) for language identification''' if LANGID\_FLAG is True: result, \_ \= ( py3langid.classify(temp\_text) if len(temp\_text) \> len(temp\_comments) else py3langid.classify(temp\_comments) ) else: \# pragma: no cover LOGGER.warning('Language detector not installed, skipping detection') result \= None return result \# type: ignore\[no-any-return\] def language\_filter(temp\_text: str, temp\_comments: str, target\_language: str, docmeta: Any) \-> Tuple\[bool, Any\]: '''Filter text based on language detection and store relevant information''' \# todo: run and pass info along anyway? if target\_language is not None: \# more thorough: detection on actual text content docmeta.language \= language\_classifier(temp\_text, temp\_comments) \# HTML lang check? sometimes contradicted by detection above #if docmeta.language is None: \# if check\_html\_lang(tree, target\_language) is False: \# LOGGER.error('wrong HTML meta language for URL %s', url) \# raise ValueError if docmeta.language is not None and docmeta.language != target\_language: LOGGER.warning('wrong language: %s %s', docmeta.language, docmeta.url) return True, docmeta return False, docmeta def textfilter(element: \_Element) \-> bool: '''Filter out unwanted text''' testtext \= element.tail if element.text is None else element.text \# to check: line len → continue if len(line) <= 5 return not testtext or testtext.isspace() or any(map(RE\_FILTER.match, testtext.splitlines())) def text\_chars\_test(string: Optional\[str\]) \-> bool: '''Determine if a string is only composed of spaces and/or control characters''' \# or not re.search(r'\\w', string) \# return string is not None and len(string) != 0 and not string.isspace() return bool(string) and not string.isspace() \# type: ignore\[union-attr\] --- # trafilatura.xml — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/xml.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.xml =============================== \# pylint:disable-msg=E0611,I1101 """ All functions related to XML generation, processing and validation. """ import csv import logging from html import unescape from importlib.metadata import version from io import StringIO from json import dumps as json\_dumps from pathlib import Path from typing import List, Optional from lxml.etree import (\_Element, Element, SubElement, XMLParser, fromstring, tostring, DTD) from .settings import Document, Extractor from .utils import sanitize, sanitize\_tree, text\_chars\_test LOGGER \= logging.getLogger(\_\_name\_\_) PKG\_VERSION \= version("trafilatura") \# validation TEI\_SCHEMA \= str(Path(\_\_file\_\_).parent / "data" / "tei\_corpus.dtd") TEI\_VALID\_TAGS \= {'ab', 'body', 'cell', 'code', 'del', 'div', 'graphic', 'head', 'hi', \\ 'item', 'lb', 'list', 'p', 'quote', 'ref', 'row', 'table'} TEI\_VALID\_ATTRS \= {'rend', 'rendition', 'role', 'target', 'type'} TEI\_DTD \= None \# to be downloaded later if necessary TEI\_REMOVE\_TAIL \= {"ab", "p"} TEI\_DIV\_SIBLINGS \= {"p", "list", "table", "quote", "ab"} CONTROL\_PARSER \= XMLParser(remove\_blank\_text\=True) NEWLINE\_ELEMS \= {'code', 'graphic', 'head', 'lb', 'list', 'p', 'quote', 'row', 'table'} SPECIAL\_FORMATTING \= {'del', 'head', 'hi', 'ref'} WITH\_ATTRIBUTES \= {'cell', 'row', 'del', 'graphic', 'head', 'hi', 'item', 'list', 'ref'} NESTING\_WHITELIST \= {"cell", "figure", "item", "note", "quote"} META\_ATTRIBUTES \= \[\ 'sitename', 'title', 'author', 'date', 'url', 'hostname',\ 'description', 'categories', 'tags', 'license', 'id',\ 'fingerprint', 'language'\ \] HI\_FORMATTING \= {'#b': '\*\*', '#i': '\*', '#u': '\_\_', '#t': '\`'} MAX\_TABLE\_WIDTH \= 1000 \# https://github.com/lxml/lxml/blob/master/src/lxml/html/\_\_init\_\_.py def delete\_element(element: \_Element, keep\_tail: bool \= True) \-> None: """ Removes this element from the tree, including its children and text. The tail text is joined to the previous element or parent. """ parent \= element.getparent() if parent is None: return if keep\_tail and element.tail: previous \= element.getprevious() if previous is None: parent.text \= (parent.text or "") + element.tail else: previous.tail \= (previous.tail or "") + element.tail parent.remove(element) def merge\_with\_parent(element: \_Element, include\_formatting: bool \= False) \-> None: '''Merge element with its parent and convert formatting to markdown.''' parent \= element.getparent() if parent is None: return full\_text \= replace\_element\_text(element, include\_formatting) if element.tail is not None: full\_text += element.tail previous \= element.getprevious() if previous is not None: \# There is a previous node, append text to its tail previous.tail \= f'{previous.tail} {full\_text}' if previous.tail else full\_text elif parent.text is not None: parent.text \= f'{parent.text} {full\_text}' else: parent.text \= full\_text parent.remove(element) def remove\_empty\_elements(tree: \_Element) \-> \_Element: '''Remove text elements without text.''' for element in tree.iter('\*'): \# 'head', 'hi', 'item', 'p' if len(element) \== 0 and text\_chars\_test(element.text) is False and text\_chars\_test(element.tail) is False: parent \= element.getparent() \# not root element or element which is naturally empty \# do not remove elements inside <code> to preserve formatting if parent is not None and element.tag != "graphic" and parent.tag != 'code': parent.remove(element) return tree def strip\_double\_tags(tree: \_Element) \-> \_Element: "Prevent nested tags among a fixed list of tags." for elem in reversed(tree.xpath(".//head | .//code | .//p")): for subelem in elem.iterdescendants("code", "head", "p"): if subelem.tag \== elem.tag and subelem.getparent().tag not in NESTING\_WHITELIST: merge\_with\_parent(subelem) return tree def build\_json\_output(docmeta: Document, with\_metadata: bool \= True) \-> str: '''Build JSON output based on extracted information''' if with\_metadata: outputdict \= {slot: getattr(docmeta, slot, None) for slot in docmeta.\_\_slots\_\_} outputdict.update({ 'source': outputdict.pop('url'), 'source-hostname': outputdict.pop('sitename'), 'excerpt': outputdict.pop('description'), 'categories': ';'.join(outputdict.pop('categories') or \[\]), 'tags': ';'.join(outputdict.pop('tags') or \[\]), 'text': xmltotxt(outputdict.pop('body'), include\_formatting\=False), }) commentsbody \= outputdict.pop('commentsbody') else: outputdict \= {'text': xmltotxt(docmeta.body, include\_formatting\=False)} commentsbody \= docmeta.commentsbody outputdict\['comments'\] \= xmltotxt(commentsbody, include\_formatting\=False) return json\_dumps(outputdict, ensure\_ascii\=False) def clean\_attributes(tree: \_Element) \-> \_Element: '''Remove unnecessary attributes.''' for elem in tree.iter('\*'): if elem.tag not in WITH\_ATTRIBUTES: elem.attrib.clear() return tree def build\_xml\_output(docmeta: Document) \-> \_Element: '''Build XML output tree based on extracted information''' output \= Element('doc') add\_xml\_meta(output, docmeta) docmeta.body.tag \= 'main' \# clean XML tree output.append(clean\_attributes(docmeta.body)) docmeta.commentsbody.tag \= 'comments' output.append(clean\_attributes(docmeta.commentsbody)) return output def control\_xml\_output(document: Document, options: Extractor) \-> str: '''Make sure the XML output is conform and valid if required''' strip\_double\_tags(document.body) remove\_empty\_elements(document.body) func \= build\_xml\_output if options.format \== "xml" else build\_tei\_output output\_tree \= func(document) output\_tree \= sanitize\_tree(output\_tree) \# necessary for cleaning output\_tree \= fromstring(tostring(output\_tree, encoding\='unicode'), CONTROL\_PARSER) \# validate if options.format \== 'xmltei' and options.tei\_validation: LOGGER.debug('TEI validation result: %s %s', validate\_tei(output\_tree), options.source) return tostring(output\_tree, pretty\_print\=True, encoding\='unicode').strip() def add\_xml\_meta(output: \_Element, docmeta: Document) \-> None: '''Add extracted metadata to the XML output tree''' for attribute in META\_ATTRIBUTES: value \= getattr(docmeta, attribute, None) if value: output.set(attribute, value if isinstance(value, str) else ';'.join(value)) def build\_tei\_output(docmeta: Document) \-> \_Element: '''Build TEI-XML output tree based on extracted information''' \# build TEI tree output \= write\_teitree(docmeta) \# filter output (strip unwanted elements), just in case \# check and repair output \= check\_tei(output, docmeta.url) return output def check\_tei(xmldoc: \_Element, url: Optional\[str\]) \-> \_Element: '''Check if the resulting XML file is conform and scrub remaining tags''' \# convert head tags for elem in xmldoc.iter('head'): elem.tag \= 'ab' elem.set('type', 'header') parent \= elem.getparent() if parent is None: continue if len(elem) \> 0: new\_elem \= \_tei\_handle\_complex\_head(elem) parent.replace(elem, new\_elem) elem \= new\_elem if parent.tag \== "p": \_move\_element\_one\_level\_up(elem) \# convert <lb/> when child of <div> to <p> for elem in xmldoc.findall(".//text/body//div/lb"): if elem.tail and elem.tail.strip(): elem.tag, elem.text, elem.tail \= 'p', elem.tail, None \# look for elements that are not valid for elem in xmldoc.findall('.//text/body//\*'): \# check elements if elem.tag not in TEI\_VALID\_TAGS: \# disable warnings for chosen categories \# if element.tag not in ('div', 'span'): LOGGER.warning('not a TEI element, removing: %s %s', elem.tag, url) merge\_with\_parent(elem) continue if elem.tag in TEI\_REMOVE\_TAIL: \_handle\_unwanted\_tails(elem) elif elem.tag \== "div": \_handle\_text\_content\_of\_div\_nodes(elem) \_wrap\_unwanted\_siblings\_of\_div(elem) #if len(elem) == 0: \# elem.getparent().remove(elem) \# check attributes for attribute in \[a for a in elem.attrib if a not in TEI\_VALID\_ATTRS\]: LOGGER.warning('not a valid TEI attribute, removing: %s in %s %s', attribute, elem.tag, url) elem.attrib.pop(attribute) return xmldoc [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.xml.validate_tei) def validate\_tei(xmldoc: \_Element) \-> bool: '''Check if an XML document is conform to the guidelines of the Text Encoding Initiative''' global TEI\_DTD if TEI\_DTD is None: \# https://tei-c.org/release/xml/tei/custom/schema/dtd/tei\_corpus.dtd TEI\_DTD \= DTD(TEI\_SCHEMA) result \= TEI\_DTD.validate(xmldoc) if result is False: LOGGER.warning('not a valid TEI document: %s', TEI\_DTD.error\_log.last\_error) return result def replace\_element\_text(element: \_Element, include\_formatting: bool) \-> str: "Determine element text based on just the text of the element. One must deal with the tail separately." elem\_text \= element.text or "" \# handle formatting: convert to markdown if include\_formatting and element.text: if element.tag \== "head": try: number \= int(element.get("rend")\[1\]) \# type: ignore\[index\] except (TypeError, ValueError): number \= 2 elem\_text \= f'{"#" \* number} {elem\_text}' elif element.tag \== "del": elem\_text \= f"~~{elem\_text}~~" elif element.tag \== "hi": rend \= element.get("rend") if rend in HI\_FORMATTING: elem\_text \= f"{HI\_FORMATTING\[rend\]}{elem\_text}{HI\_FORMATTING\[rend\]}" elif element.tag \== "code": if "\\n" in element.text: elem\_text \= f"\`\`\`\\n{elem\_text}\\n\`\`\`" else: elem\_text \= f"\`{elem\_text}\`" \# handle links if element.tag \== "ref": if elem\_text: link\_text \= f"\[{elem\_text}\]" target \= element.get("target") if target: elem\_text \= f"{link\_text}({target})" else: LOGGER.warning("missing link attribute: %s %s'", elem\_text, element.attrib) elem\_text \= link\_text else: LOGGER.warning("empty link: %s %s", elem\_text, element.attrib) \# cells if element.tag \== "cell" and elem\_text and len(element) \> 0: if element\[0\].tag \== 'p': elem\_text \= f"{elem\_text} " if element.getprevious() is not None else f"| {elem\_text} " elif element.tag \== 'cell' and elem\_text: \# add | before first cell elem\_text \= f"{elem\_text}" if element.getprevious() is not None else f"| {elem\_text}" \# lists elif element.tag \== "item" and elem\_text: elem\_text \= f"- {elem\_text}\\n" return elem\_text def process\_element(element: \_Element, returnlist: List\[str\], include\_formatting: bool) \-> None: "Recursively convert a LXML element and its children to a flattened string representation." if element.text: \# this is the text that comes before the first child returnlist.append(replace\_element\_text(element, include\_formatting)) for child in element: process\_element(child, returnlist, include\_formatting) if not element.text and not element.tail: if element.tag \== "graphic": \# add source, default to '' text \= f'{element.get("title", "")} {element.get("alt", "")}' returnlist.append(f'!\[{text.strip()}\]({element.get("src", "")})') \# newlines for textless elements elif element.tag in NEWLINE\_ELEMS: \# add line after table head if element.tag \== "row": cell\_count \= len(element.xpath(".//cell")) \# restrict columns to a maximum of 1000 span\_info \= element.get("colspan") or element.get("span") if not span\_info or not span\_info.isdigit(): max\_span \= 1 else: max\_span \= min(int(span\_info), MAX\_TABLE\_WIDTH) \# row ended so draw extra empty cells to match max\_span if cell\_count < max\_span: returnlist.append(f'{"|" \* (max\_span \- cell\_count)}\\n') \# if this is a head row, draw the separator below if element.xpath("./cell\[@role='head'\]"): returnlist.append(f'\\n|{"---|" \* max\_span}\\n') else: returnlist.append("\\n") elif element.tag != "cell": \# cells still need to append vertical bars \# but nothing more to do with other textless elements return \# Process text \# Common elements (Now processes end-tag logic correctly) if element.tag in NEWLINE\_ELEMS and not element.xpath("ancestor::cell"): \# spacing hack returnlist.append("\\n\\u2424\\n" if include\_formatting and element.tag != 'row' else "\\n") elif element.tag \== "cell": returnlist.append(" | ") elif element.tag not in SPECIAL\_FORMATTING: returnlist.append(" ") \# this is text that comes after the closing tag, so it should be after any NEWLINE\_ELEMS if element.tail: returnlist.append(element.tail) [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.xml.xmltotxt) def xmltotxt(xmloutput: Optional\[\_Element\], include\_formatting: bool) \-> str: "Convert to plain text format and optionally preserve formatting as markdown." if xmloutput is None: return "" returnlist: List\[str\] \= \[\] process\_element(xmloutput, returnlist, include\_formatting) return unescape(sanitize("".join(returnlist)) or "") def xmltocsv(document: Document, include\_formatting: bool, \*, delim: str \= "\\t", null: str \= "null") \-> str: "Convert the internal XML document representation to a CSV string." \# preprocessing posttext \= xmltotxt(document.body, include\_formatting) or null commentstext \= xmltotxt(document.commentsbody, include\_formatting) or null \# output config output \= StringIO() outputwriter \= csv.writer(output, delimiter\=delim, quoting\=csv.QUOTE\_MINIMAL) \# organize fields outputwriter.writerow(\[d if d else null for d in (\ document.url,\ document.id,\ document.fingerprint,\ document.hostname,\ document.title,\ document.image,\ document.date,\ posttext,\ commentstext,\ document.license,\ document.pagetype\ )\]) return output.getvalue() def write\_teitree(docmeta: Document) \-> \_Element: '''Bundle the extracted post and comments into a TEI tree''' teidoc \= Element('TEI', xmlns\='http://www.tei-c.org/ns/1.0') write\_fullheader(teidoc, docmeta) textelem \= SubElement(teidoc, 'text') textbody \= SubElement(textelem, 'body') \# post postbody \= clean\_attributes(docmeta.body) postbody.tag \= 'div' postbody.set('type', 'entry') textbody.append(postbody) \# comments commentsbody \= clean\_attributes(docmeta.commentsbody) commentsbody.tag \= 'div' commentsbody.set('type', 'comments') textbody.append(commentsbody) return teidoc def \_define\_publisher\_string(docmeta: Document) \-> str: '''Construct a publisher string to include in TEI header''' if docmeta.hostname and docmeta.sitename: publisher \= f'{docmeta.sitename.strip()} ({docmeta.hostname})' else: publisher \= docmeta.hostname or docmeta.sitename or 'N/A' if LOGGER.isEnabledFor(logging.WARNING) and publisher \== 'N/A': LOGGER.warning('no publisher for URL %s', docmeta.url) return publisher def write\_fullheader(teidoc: \_Element, docmeta: Document) \-> \_Element: '''Write TEI header based on gathered metadata''' \# todo: add language info header \= SubElement(teidoc, 'teiHeader') filedesc \= SubElement(header, 'fileDesc') bib\_titlestmt \= SubElement(filedesc, 'titleStmt') SubElement(bib\_titlestmt, 'title', type\='main').text \= docmeta.title if docmeta.author: SubElement(bib\_titlestmt, 'author').text \= docmeta.author publicationstmt\_a \= SubElement(filedesc, 'publicationStmt') publisher\_string \= \_define\_publisher\_string(docmeta) \# license, if applicable if docmeta.license: SubElement(publicationstmt\_a, 'publisher').text \= publisher\_string availability \= SubElement(publicationstmt\_a, 'availability') SubElement(availability, 'p').text \= docmeta.license \# insert an empty paragraph for conformity else: SubElement(publicationstmt\_a, 'p') notesstmt \= SubElement(filedesc, 'notesStmt') if docmeta.id: SubElement(notesstmt, 'note', type\='id').text \= docmeta.id SubElement(notesstmt, 'note', type\='fingerprint').text \= docmeta.fingerprint sourcedesc \= SubElement(filedesc, 'sourceDesc') source\_bibl \= SubElement(sourcedesc, 'bibl') sigle \= ', '.join(filter(None, \[docmeta.sitename, docmeta.date\])) if not sigle: LOGGER.warning('no sigle for URL %s', docmeta.url) source\_bibl.text \= ', '.join(filter(None, \[docmeta.title, sigle\])) SubElement(sourcedesc, 'bibl', type\='sigle').text \= sigle biblfull \= SubElement(sourcedesc, 'biblFull') bib\_titlestmt \= SubElement(biblfull, 'titleStmt') SubElement(bib\_titlestmt, 'title', type\='main').text \= docmeta.title if docmeta.author: SubElement(bib\_titlestmt, 'author').text \= docmeta.author publicationstmt \= SubElement(biblfull, 'publicationStmt') SubElement(publicationstmt, 'publisher').text \= publisher\_string if docmeta.url: SubElement(publicationstmt, 'ptr', type\='URL', target\=docmeta.url) SubElement(publicationstmt, 'date').text \= docmeta.date profiledesc \= SubElement(header, 'profileDesc') abstract \= SubElement(profiledesc, 'abstract') SubElement(abstract, 'p').text \= docmeta.description if docmeta.categories or docmeta.tags: textclass \= SubElement(profiledesc, 'textClass') keywords \= SubElement(textclass, 'keywords') if docmeta.categories: SubElement(keywords, 'term', type\='categories').text \= ','.join(docmeta.categories) if docmeta.tags: SubElement(keywords, 'term', type\='tags').text \= ','.join(docmeta.tags) creation \= SubElement(profiledesc, 'creation') SubElement(creation, 'date', type\="download").text \= docmeta.filedate encodingdesc \= SubElement(header, 'encodingDesc') appinfo \= SubElement(encodingdesc, 'appInfo') application \= SubElement(appinfo, 'application', version\=PKG\_VERSION, ident\='Trafilatura') SubElement(application, 'label').text \= 'Trafilatura' SubElement(application, 'ptr', target\='https://github.com/adbar/trafilatura') return header def \_handle\_text\_content\_of\_div\_nodes(element: \_Element) \-> None: "Wrap loose text in <div> within <p> elements for TEI conformity." if element.text and element.text.strip(): if len(element) \> 0 and element\[0\].tag \== "p": element\[0\].text \= f'{element.text} {element\[0\].text or ""}'.strip() else: new\_child \= Element("p") new\_child.text \= element.text element.insert(0, new\_child) element.text \= None if element.tail and element.tail.strip(): if len(element) \> 0 and element\[\-1\].tag \== "p": element\[\-1\].text \= f'{element\[\-1\].text or ""} {element.tail}'.strip() else: new\_child \= Element("p") new\_child.text \= element.tail element.append(new\_child) element.tail \= None def \_handle\_unwanted\_tails(element: \_Element) \-> None: "Handle tail on p and ab elements" element.tail \= element.tail.strip() if element.tail else None if not element.tail: return if element.tag \== "p": element.text \= " ".join(filter(None, \[element.text, element.tail\])) else: new\_sibling \= Element('p') new\_sibling.text \= element.tail parent \= element.getparent() if parent is not None: parent.insert(parent.index(element) + 1 , new\_sibling) element.tail \= None def \_tei\_handle\_complex\_head(element: \_Element) \-> \_Element: "Convert certain child elements to <ab> and <lb>." new\_element \= Element('ab', attrib\=element.attrib) new\_element.text \= element.text.strip() if element.text else None for child in element.iterchildren(): if child.tag \== 'p': if len(new\_element) \> 0 or new\_element.text: \# add <lb> if <ab> has no children or last tail contains text if len(new\_element) \== 0 or new\_element\[\-1\].tail: SubElement(new\_element, 'lb') new\_element\[\-1\].tail \= child.text else: new\_element.text \= child.text else: new\_element.append(child) tail \= element.tail.strip() if element.tail else None if tail: new\_element.tail \= tail return new\_element def \_wrap\_unwanted\_siblings\_of\_div(div\_element: \_Element) \-> None: "Wrap unwanted siblings of a div element in a new div element." new\_sibling \= Element("div") new\_sibling\_index \= None parent \= div\_element.getparent() if parent is None: return \# check siblings after target element for sibling in div\_element.itersiblings(): if sibling.tag \== "div": break if sibling.tag in TEI\_DIV\_SIBLINGS: new\_sibling\_index \= new\_sibling\_index or parent.index(sibling) new\_sibling.append(sibling) \# some elements (e.g. <lb/>) can appear next to div, but \# order of elements should be kept, thus add and reset new\_sibling else: if new\_sibling\_index and len(new\_sibling) \> 0: parent.insert(new\_sibling\_index, new\_sibling) new\_sibling \= Element("div") new\_sibling\_index \= None if new\_sibling\_index and len(new\_sibling) != 0: parent.insert(new\_sibling\_index, new\_sibling) def \_move\_element\_one\_level\_up(element: \_Element) \-> None: """ Fix TEI compatibility issues by moving certain p-elems up in the XML tree. There is always a n+2 nesting for p-elements with the minimal structure ./TEI/text/body/p """ parent \= element.getparent() grand\_parent \= parent.getparent() if parent is not None else None if parent is None or grand\_parent is None: return new\_elem \= Element("p") new\_elem.extend(list(element.itersiblings())) grand\_parent.insert(grand\_parent.index(parent) + 1, element) tail \= element.tail.strip() if element.tail else None if tail: new\_elem.text \= tail element.tail \= None tail \= parent.tail.strip() if parent.tail else None if tail: new\_elem.tail \= tail parent.tail \= None if len(new\_elem) \> 0 or new\_elem.text or new\_elem.tail: grand\_parent.insert(grand\_parent.index(element) + 1, new\_elem) if len(parent) \== 0 and not parent.text: grand\

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: \# type: ignore\[attr-defined\] result\_body \= recover\_wild\_text(backup\_tree, result\_body, options, potential\_tags) temp\_text \= ' '.join(result\_body.itertext()).strip() \# filter output strip\_elements(result\_body, 'done') strip\_tags(result\_body, 'div') \# return return result\_body, temp\_text, len(temp\_text) def process\_comments\_node(elem: \_Element, potential\_tags: Set\[str\], options: Extractor) \-> Optional\[\_Element\]: '''Process comment node and determine how to deal with its content''' if elem.tag in potential\_tags: \# print(elem.tag, elem.text\_content()) processed\_element \= handle\_textnode(elem, options, comments\_fix\=True) \# test length and remove if processed\_element is not None: \# and processed\_element.text not in COMMENTS\_BLACKLIST: processed\_element.attrib.clear() \# if textfilter(elem) is True: # ^Pingback \# return None return processed\_element return None [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.core.extract_comments) def extract\_comments(tree: HtmlElement, options: Extractor) \-> Tuple\[\_Element, str, int, HtmlElement\]: "Try to extract comments out of potential sections in the HTML." comments\_body \= Element("body") \# define iteration strategy potential\_tags \= set(TAG\_CATALOG) \# 'span' \# potential\_tags.add('div') trouble with

for expr in COMMENTS\_XPATH: \# 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 subtree \= prune\_unwanted\_nodes(subtree, COMMENTS\_DISCARD\_XPATH) \# todo: unified stripping function, taking include\_links into account strip\_tags(subtree, "a", "ref", "span") \# extract content \# for elem in subtree.xpath('.//\*'): \# processed\_elem = process\_comments\_node(elem, potential\_tags) \# if processed\_elem is not None: \# comments\_body.append(processed\_elem) \# processed\_elems = (process\_comments\_node(elem, potential\_tags, options) for elem in \# subtree.xpath('.//\*')) comments\_body.extend(filter(lambda x: x is not None, (process\_comments\_node(e, potential\_tags, options) for e in subtree.xpath(".//\*")))) \# type: ignore\[arg-type\] \# control if len(comments\_body) \> 0: \# if it has children LOGGER.debug(expr) \# remove corresponding subtree delete\_element(subtree, keep\_tail\=False) break \# lengths temp\_comments \= " ".join(comments\_body.itertext()).strip() return comments\_body, temp\_comments, len(temp\_comments), tree --- # trafilatura.sitemaps — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/sitemaps.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.sitemaps ==================================== """ Deriving link info from sitemaps. """ import logging import re from itertools import islice from time import sleep from typing import Callable, List, Set, Optional, Pattern from courlan import ( clean\_url, extract\_domain, filter\_urls, fix\_relative\_urls, get\_hostinfo, lang\_filter, ) from .deduplication import is\_similar\_domain from .downloads import fetch\_url, is\_live\_page from .settings import MAX\_LINKS, MAX\_SITEMAPS\_SEEN LOGGER \= logging.getLogger(\_\_name\_\_) LINK\_REGEX \= re.compile(r"(?:)?") XHTML\_REGEX \= re.compile(r"", re.DOTALL) HREFLANG\_REGEX \= re.compile(r'href=\["\\'\](.+?)\["\\'\]') WHITELISTED\_PLATFORMS \= re.compile( r"(?:blogger|blogpost|ghost|hubspot|livejournal|medium|typepad|squarespace|tumblr|weebly|wix|wordpress)\\." ) SITEMAP\_FORMAT \= re.compile(r"^.{0,5}<\\?xml| None: self.base\_url: str \= base\_url self.content: str \= "" self.domain: str \= domain self.external: bool \= external self.current\_url: str \= "" self.seen: Set\[str\] \= set() self.sitemap\_urls: List\[str\] \= sitemapsurls self.target\_lang: Optional\[str\] \= target\_lang self.urls: List\[str\] \= \[\] def fetch(self) \-> None: "Fetch a sitemap over the network." LOGGER.debug("fetching sitemap: %s", self.current\_url) self.content \= fetch\_url(self.current\_url) or "" self.seen.add(self.current\_url) def handle\_link(self, link: str) \-> None: """Examine a link and determine if it's valid and if it leads to a sitemap or a web page.""" if link \== self.current\_url: \# safety check return \# fix, check, clean and normalize link \= fix\_relative\_urls(self.base\_url, link) link \= clean\_url(link, self.target\_lang) or "" if not link or not lang\_filter(link, self.target\_lang): return newdomain \= extract\_domain(link, fast\=True) if newdomain is None: LOGGER.error("couldn't extract domain: %s", link) return \# don't take links from another domain and make an exception for main platforms \# also bypass: subdomains vs. domains if ( not self.external and not WHITELISTED\_PLATFORMS.search(newdomain) and not is\_similar\_domain(self.domain, newdomain) ): LOGGER.warning( "link discarded, diverging domain names: %s %s", self.domain, newdomain ) return if DETECT\_SITEMAP\_LINK.search(link): self.sitemap\_urls.append(link) else: self.urls.append(link) def extract\_links( self, regex: Pattern\[str\], index: int, handler: Callable\[\[str\], None\] ) \-> None: "Extract links from the content using pre-defined regex, index and handler." for match in ( m\[index\] for m in islice(regex.finditer(self.content), MAX\_LINKS) ): handler(match) LOGGER.debug( "%s sitemaps and %s links found for %s", len(self.sitemap\_urls), len(self.urls), self.current\_url, ) def extract\_sitemap\_langlinks(self) \-> None: "Extract links corresponding to a given target language." if "hreflang=" not in self.content: return lang\_regex \= re.compile( rf"hreflang=\[\\"'\]({self.target\_lang}.\*?|x-default)\[\\"'\]", re.DOTALL ) def handle\_lang\_link(attrs: str) \-> None: "Examine language code attributes." if lang\_regex.search(attrs): lang\_match \= HREFLANG\_REGEX.search(attrs) if lang\_match: self.handle\_link(lang\_match\[1\]) self.extract\_links(XHTML\_REGEX, 0, handle\_lang\_link) def extract\_sitemap\_links(self) \-> None: "Extract sitemap links and web page links from a sitemap file." self.extract\_links( LINK\_REGEX, 1, self.handle\_link ) \# process middle part of the match tuple def process(self) \-> None: "Download a sitemap and extract the links it contains." plausible \= is\_plausible\_sitemap(self.current\_url, self.content) \# safeguard if not plausible: return \# try to extract links from TXT file if not SITEMAP\_FORMAT.match(self.content): self.extract\_links(DETECT\_LINKS, 0, self.handle\_link) return \# process XML sitemap if self.target\_lang is not None: self.extract\_sitemap\_langlinks() if self.sitemap\_urls or self.urls: return self.extract\_sitemap\_links() [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.sitemaps.sitemap_search) def sitemap\_search( url: str, target\_lang: Optional\[str\] \= None, external: bool \= False, sleep\_time: float \= 2.0, max\_sitemaps: int \= MAX\_SITEMAPS\_SEEN, ) \-> List\[str\]: """Look for sitemaps for the given URL and gather links. Args: url: Webpage or sitemap URL as string. Triggers URL-based filter if the webpage isn't a homepage. target\_lang: Define a language to filter URLs based on heuristics (two-letter string, ISO 639-1 format). external: Similar hosts only or external URLs (boolean, defaults to False). sleep\_time: Wait between requests on the same website. max\_sitemaps: Maximum number of sitemaps to process. Returns: The extracted links as a list (sorted list of unique links). """ domainname, baseurl \= get\_hostinfo(url) if domainname is None: LOGGER.warning("invalid URL: %s", url) return \[\] if not is\_live\_page(baseurl): LOGGER.warning("base URL unreachable, dropping sitemap: %s", url) return \[\] urlfilter \= None if url.endswith((".gz", "sitemap", ".xml")): sitemapurls \= \[url\] else: sitemapurls \= \[\] \# set url filter to target subpages if len(url) \> len(baseurl) + 2: urlfilter \= url sitemap \= SitemapObject(baseurl, domainname, sitemapurls, target\_lang, external) \# try sitemaps in robots.txt file, additional URLs just in case if not sitemap.sitemap\_urls: sitemap.sitemap\_urls \= find\_robots\_sitemaps(baseurl) or \[\ f"{baseurl}/{g}" for g in GUESSES\ \] \# iterate through nested sitemaps and results while sitemap.sitemap\_urls and len(sitemap.seen) < max\_sitemaps: sitemap.current\_url \= sitemap.sitemap\_urls.pop() sitemap.fetch() sitemap.process() \# sanity check: keep track of visited sitemaps and exclude them sitemap.sitemap\_urls \= \[\ s for s in sitemap.sitemap\_urls if s not in sitemap.seen\ \] if len(sitemap.seen) < max\_sitemaps: sleep(sleep\_time) if urlfilter: sitemap.urls \= filter\_urls(sitemap.urls, urlfilter) LOGGER.debug("%s sitemap links found for %s", len(sitemap.urls), domainname) return sitemap.urls def is\_plausible\_sitemap(url: str, contents: Optional\[str\]) \-> bool: """Check if the sitemap corresponds to an expected format, i.e. TXT or XML.""" if contents is None: return False \# strip query and fragments url \= SCRUB\_REGEX.sub("", url) \# check content if ( POTENTIAL\_SITEMAP.search(url) and (not isinstance(contents, str) or not SITEMAP\_FORMAT.match(contents)) or " List\[str\]: """Guess the location of the robots.txt file and try to extract sitemap URLs from it""" robotstxt \= fetch\_url(baseurl + "/robots.txt") return extract\_robots\_sitemaps(robotstxt, baseurl) def extract\_robots\_sitemaps(robotstxt: Optional\[str\], baseurl: str) \-> List\[str\]: "Read a robots.txt file and find sitemap links." \# sanity check on length (cause: redirections) if robotstxt is None or len(robotstxt) \> 10000: return \[\] candidates \= \[\] \# source: https://github.com/python/cpython/blob/3.12/Lib/urllib/robotparser.py for line in robotstxt.splitlines(): \# remove optional comment and strip line i \= line.find("#") if i \>= 0: line \= line\[:i\] line \= line.strip() if not line: continue line\_parts \= line.split(":", 1) if len(line\_parts) \== 2: line\_parts\[0\] \= line\_parts\[0\].strip().lower() if line\_parts\[0\] \== "sitemap": \# urllib.parse.unquote(line\[1\].strip()) candidates.append(line\_parts\[1\].strip()) candidates \= list(dict.fromkeys(candidates)) sitemapurls \= \[fix\_relative\_urls(baseurl, u) for u in candidates if u\] LOGGER.debug("%s sitemaps found in robots.txt", len(sitemapurls)) return sitemapurls --- # trafilatura.metadata — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/metadata.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.metadata ==================================== """ Module bundling all functions needed to scrape metadata from webpages. """ import json import logging import re from copy import deepcopy from html import unescape from typing import Any, Dict, List, Optional, Set, Tuple, Union from courlan import ( extract\_domain, get\_base\_url, is\_valid\_url, normalize\_url, validate\_url, ) from htmldate import find\_date from lxml.etree import XPath from lxml.html import HtmlElement, tostring from .htmlprocessing import prune\_unwanted\_nodes from .json\_metadata import ( extract\_json, extract\_json\_parse\_error, normalize\_authors, normalize\_json, ) from .settings import Document, set\_date\_params from .utils import HTML\_STRIP\_TAGS, line\_processing, load\_html, trim from .xpaths import ( AUTHOR\_DISCARD\_XPATHS, AUTHOR\_XPATHS, CATEGORIES\_XPATHS, TAGS\_XPATHS, TITLE\_XPATHS, ) \_\_all\_\_ \= \["Document"\] LOGGER \= logging.getLogger(\_\_name\_\_) logging.getLogger("htmldate").setLevel(logging.WARNING) META\_URL \= re.compile(r"https?://(?:www\\.|w\[0-9\]+\\.)?(\[^/\]+)") JSON\_MINIFY \= re.compile(r'("(?:\\\\"|\[^"\])\*")|\\s') HTMLTITLE\_REGEX \= re.compile( r"^(.+)?\\s+\[–•·—|⁄\*⋆~‹«<›»>:-\]\\s+(.+)$" ) \# part without dots? CLEAN\_META\_TAGS \= re.compile(r'\["\\'\]') LICENSE\_REGEX \= re.compile( r"/(by-nc-nd|by-nc-sa|by-nc|by-nd|by-sa|by|zero)/(\[1-9\]\\.\[0-9\])" ) TEXT\_LICENSE\_REGEX \= re.compile( r"(cc|creative commons) (by-nc-nd|by-nc-sa|by-nc|by-nd|by-sa|by|zero) ?(\[1-9\]\\.\[0-9\])?", re.I, ) METANAME\_AUTHOR \= { "article:author", "atc-metaauthor", "author", "authors", "byl", "citation\_author", "creator", "dc.creator", "dc.creator.aut", "dc:creator", "dcterms.creator", "dcterms.creator.aut", "dcsext.author", "parsely-author", "rbauthors", "sailthru.author", "shareaholic:article\_author\_name", } \# questionable: twitter:creator METANAME\_DESCRIPTION \= { "dc.description", "dc:description", "dcterms.abstract", "dcterms.description", "description", "sailthru.description", "twitter:description", } METANAME\_PUBLISHER \= { "article:publisher", "citation\_journal\_title", "copyright", "dc.publisher", "dc:publisher", "dcterms.publisher", "publisher", "sailthru.publisher", "rbpubname", "twitter:site", } \# questionable: citation\_publisher METANAME\_TAG \= { "citation\_keywords", "dcterms.subject", "keywords", "parsely-tags", "shareaholic:keywords", "tags", } METANAME\_TITLE \= { "citation\_title", "dc.title", "dcterms.title", "fb\_title", "headline", "parsely-title", "sailthru.title", "shareaholic:title", "rbtitle", "title", "twitter:title", } METANAME\_URL \= {"rbmainurl", "twitter:url"} METANAME\_IMAGE \= { "image", "og:image", "og:image:url", "og:image:secure\_url", "twitter:image", "twitter:image:src", } PROPERTY\_AUTHOR \= {"author", "article:author"} TWITTER\_ATTRS \= {"twitter:site", "application-name"} \# also interesting: article:section EXTRA\_META \= {"charset", "http-equiv", "property"} OG\_PROPERTIES \= { "og:title": "title", "og:description": "description", "og:site\_name": "sitename", "og:image": "image", "og:image:url": "image", "og:image:secure\_url": "image", "og:type": "pagetype", } OG\_AUTHOR \= {"og:author", "og:article:author"} URL\_SELECTORS \= \[\ './/head//link\[@rel="canonical"\]',\ './/head//base',\ './/head//link\[@rel="alternate"\]\[@hreflang="x-default"\]'\ \] def normalize\_tags(tags: str) \-> str: """Remove special characters of tags""" trimmed \= trim(unescape(tags)) if not trimmed: return "" tags \= CLEAN\_META\_TAGS.sub(r"", trimmed) return ", ".join(filter(None, tags.split(", "))) def check\_authors(authors: str, author\_blacklist: Set\[str\]) \-> Optional\[str\]: "Check if the authors string correspond to expected values." author\_blacklist \= {a.lower() for a in author\_blacklist} new\_authors \= \[\ author.strip()\ for author in authors.split(";")\ if author.strip().lower() not in author\_blacklist\ \] if new\_authors: return "; ".join(new\_authors).strip("; ") return None def extract\_meta\_json(tree: HtmlElement, metadata: Document) \-> Document: """Parse and extract metadata from JSON-LD data""" for elem in tree.xpath( './/script\[@type="application/ld+json" or @type="application/settings+json"\]' ): if not elem.text: continue element\_text \= normalize\_json(JSON\_MINIFY.sub(r"\\1", elem.text)) try: schema \= json.loads(element\_text) metadata \= extract\_json(schema, metadata) except json.JSONDecodeError: metadata \= extract\_json\_parse\_error(element\_text, metadata) return metadata def extract\_opengraph(tree: HtmlElement) \-> Dict\[str, Optional\[str\]\]: """Search meta tags following the OpenGraph guidelines (https://ogp.me/)""" result \= dict.fromkeys( ("title", "author", "url", "description", "sitename", "image", "pagetype") ) \# detect OpenGraph schema for elem in tree.xpath('.//head/meta\[starts-with(@property, "og:")\]'): property\_name, content \= elem.get("property"), elem.get("content") \# safeguard if content and not content.isspace(): if property\_name in OG\_PROPERTIES: result\[OG\_PROPERTIES\[property\_name\]\] \= content elif property\_name \== "og:url" and is\_valid\_url(content): result\["url"\] \= content elif property\_name in OG\_AUTHOR: result\["author"\] \= normalize\_authors(None, content) \# og:locale \# elif elem.get('property') == 'og:locale': \# pagelocale = elem.get('content') return result def examine\_meta(tree: HtmlElement) \-> Document: """Search meta tags for relevant information""" \# bootstrap from potential OpenGraph tags metadata \= Document().from\_dict(extract\_opengraph(tree)) \# test if all values not assigned in the following have already been assigned if all( ( metadata.title, metadata.author, metadata.url, metadata.description, metadata.sitename, metadata.image, ) ): \# tags return metadata tags, backup\_sitename \= \[\], None \# iterate through meta tags for elem in tree.iterfind(".//head/meta\[@content\]"): \# content content\_attr \= HTML\_STRIP\_TAGS.sub("", elem.get("content", "")).strip() if not content\_attr: continue \# todo: image info \# ... \# property if "property" in elem.attrib: property\_attr \= elem.get("property", "").lower() \# no opengraph a second time if property\_attr.startswith("og:"): continue if property\_attr \== "article:tag": tags.append(normalize\_tags(content\_attr)) elif property\_attr in PROPERTY\_AUTHOR: metadata.author \= normalize\_authors(metadata.author, content\_attr) elif property\_attr \== "article:publisher": metadata.sitename \= metadata.sitename or content\_attr elif property\_attr in METANAME\_IMAGE: metadata.image \= metadata.image or content\_attr \# name attribute elif "name" in elem.attrib: name\_attr \= elem.get("name", "").lower() \# author if name\_attr in METANAME\_AUTHOR: metadata.author \= normalize\_authors(metadata.author, content\_attr) \# title elif name\_attr in METANAME\_TITLE: metadata.title \= metadata.title or content\_attr \# description elif name\_attr in METANAME\_DESCRIPTION: metadata.description \= metadata.description or content\_attr \# site name elif name\_attr in METANAME\_PUBLISHER: metadata.sitename \= metadata.sitename or content\_attr \# twitter elif name\_attr in TWITTER\_ATTRS or "twitter:app:name" in name\_attr: backup\_sitename \= content\_attr \# url elif ( name\_attr \== "twitter:url" and not metadata.url and is\_valid\_url(content\_attr) ): metadata.url \= content\_attr \# keywords elif name\_attr in METANAME\_TAG: \# 'page-topic' tags.append(normalize\_tags(content\_attr)) elif "itemprop" in elem.attrib: itemprop\_attr \= elem.get("itemprop", "").lower() if itemprop\_attr \== "author": metadata.author \= normalize\_authors(metadata.author, content\_attr) elif itemprop\_attr \== "description": metadata.description \= metadata.description or content\_attr elif itemprop\_attr \== "headline": metadata.title \= metadata.title or content\_attr \# to verify: \# elif itemprop\_attr == 'name': \# if title is None: \# title = elem.get('content') \# other types elif all(key not in elem.attrib for key in EXTRA\_META): LOGGER.debug( "unknown attribute: %s", tostring(elem, pretty\_print\=False, encoding\="unicode").strip(), ) \# backups metadata.sitename \= metadata.sitename or backup\_sitename \# copy metadata.tags \= tags \# metadata.set\_attributes(tags=tags) return metadata def extract\_metainfo( tree: HtmlElement, expressions: List\[XPath\], len\_limit: int \= 200 ) \-> Optional\[str\]: """Extract meta information""" \# try all XPath expressions for expression in expressions: \# examine all results results \= expression(tree) for elem in results: content \= trim(" ".join(elem.itertext())) if content and 2 < len(content) < len\_limit: return content if len(results) \> 1: LOGGER.debug( "more than one invalid result: %s %s", expression, len(results) ) return None def examine\_title\_element( tree: HtmlElement, ) \-> Tuple\[str, Optional\[str\], Optional\[str\]\]: """Extract text segments out of main element.""" title \= "" title\_element \= tree.find(".//head//title") if title\_element is not None: title \= trim(title\_element.text\_content()) if match := HTMLTITLE\_REGEX.match(title): return title, match\[1\], match\[2\] LOGGER.debug("no main title found") return title, None, None def extract\_title(tree: HtmlElement) \-> Optional\[str\]: """Extract the document title""" \# only one h1-element: take it h1\_results \= tree.findall(".//h1") if len(h1\_results) \== 1: title \= trim(h1\_results\[0\].text\_content()) if title: return title \# extract using x-paths title \= extract\_metainfo(tree, TITLE\_XPATHS) or "" if title: return title \# extract using title tag title, first, second \= examine\_title\_element(tree) for t in (first, second): if t and "." not in t: return t \# take first h1-title if h1\_results: return h1\_results\[0\].text\_content() \# take first h2-title try: title \= tree.xpath(".//h2")\[0\].text\_content() except IndexError: LOGGER.debug("no h2 title found") return title def extract\_author(tree: HtmlElement) \-> Optional\[str\]: """Extract the document author(s)""" subtree \= prune\_unwanted\_nodes(deepcopy(tree), AUTHOR\_DISCARD\_XPATHS) author \= extract\_metainfo(subtree, AUTHOR\_XPATHS, len\_limit\=120) if author: author \= normalize\_authors(None, author) \# copyright? return author def extract\_url(tree: HtmlElement, default\_url: Optional\[str\] \= None) \-> Optional\[str\]: """Extract the URL from the canonical link""" for selector in URL\_SELECTORS: element \= tree.find(selector) url \= element.attrib.get("href") if element is not None else None if url: break \# fix relative URLs if url and url.startswith("/"): for element in tree.iterfind(".//head//meta\[@content\]"): attrtype \= element.get("name") or element.get("property") or "" if attrtype.startswith("og:") or attrtype.startswith("twitter:"): base\_url \= get\_base\_url(element.attrib\["content"\]) if base\_url: \# prepend URL url \= base\_url + url break \# do not return invalid URLs if url: validation\_result, parsed\_url \= validate\_url(url) url \= normalize\_url(parsed\_url) if validation\_result else None return url or default\_url def extract\_sitename(tree: HtmlElement) \-> Optional\[str\]: """Extract the name of a site from the main title (if it exists)""" \_, \*parts \= examine\_title\_element(tree) return next((part for part in parts if part and "." in part), None) def extract\_catstags(metatype: str, tree: HtmlElement) \-> List\[str\]: """Find category and tag information""" results: List\[str\] \= \[\] regexpr \= "/" + metatype + "\[s|ies\]?/" xpath\_expression \= CATEGORIES\_XPATHS if metatype \== "category" else TAGS\_XPATHS \# search using custom expressions for catexpr in xpath\_expression: results.extend( elem.text\_content() for elem in catexpr(tree) if re.search(regexpr, elem.attrib\["href"\]) ) if results: break \# category fallback if metatype \== "category" and not results: for element in tree.xpath( './/head//meta\[@property="article:section" or contains(@name, "subject")\]\[@content\]' ): results.append(element.attrib\["content"\]) \# optional: search through links \# if not results: \# for elem in tree.xpath('.//a\[@href\]'): \# search for 'category' return \[r for r in dict.fromkeys(line\_processing(x) for x in results if x) if r\] def parse\_license\_element(element: HtmlElement, strict: bool \= False) \-> Optional\[str\]: """Probe a link for identifiable free license cues. Parse the href attribute first and then the link text.""" \# look for Creative Commons elements match \= LICENSE\_REGEX.search(element.get("href", "")) if match: return f"CC {match\[1\].upper()} {match\[2\]}" if element.text: \# check if it could be a CC license if strict: match \= TEXT\_LICENSE\_REGEX.search(element.text) return match\[0\] if match else None return trim(element.text) return None def extract\_license(tree: HtmlElement) \-> Optional\[str\]: """Search the HTML code for license information and parse it.""" \# look for links labeled as license for element in tree.findall('.//a\[@rel="license"\]\[@href\]'): result \= parse\_license\_element(element, strict\=False) if result is not None: return result \# probe footer elements for CC links for element in tree.xpath( './/footer//a\[@href\]|.//div\[contains(@class, "footer") or contains(@id, "footer")\]//a\[@href\]' ): result \= parse\_license\_element(element, strict\=True) if result is not None: return result return None [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.extract_metadata) def extract\_metadata( filecontent: Union\[HtmlElement, str\], default\_url: Optional\[str\] \= None, date\_config: Optional\[Any\] \= None, extensive: bool \= True, author\_blacklist: Optional\[Set\[str\]\] \= None, ) \-> Document: """Main process for metadata extraction. Args: filecontent: HTML code as string or parsed tree. default\_url: Previously known URL of the downloaded document. date\_config: Provide extraction parameters to htmldate as dict(). author\_blacklist: Provide a blacklist of Author Names as set() to filter out authors. Returns: A trafilatura.settings.Document containing the extracted metadata information or None. The Document class has .as\_dict() method that will return a copy as a dict. """ \# init author\_blacklist \= author\_blacklist or set() date\_config \= date\_config or set\_date\_params(extensive) \# load contents tree \= load\_html(filecontent) if tree is None: return Document() \# initialize dict and try to strip meta tags metadata \= examine\_meta(tree) \# to check: remove it and replace with author\_blacklist in test case if metadata.author and " " not in metadata.author: metadata.author \= None \# fix: try json-ld metadata and override try: metadata \= extract\_meta\_json(tree, metadata) except Exception as err: \# bugs in json\_metadata.py LOGGER.warning("error in JSON metadata extraction: %s", err) \# title if not metadata.title: metadata.title \= extract\_title(tree) \# check author in blacklist if metadata.author and author\_blacklist: metadata.author \= check\_authors(metadata.author, author\_blacklist) \# author if not metadata.author: metadata.author \= extract\_author(tree) \# recheck author in blacklist if metadata.author and author\_blacklist: metadata.author \= check\_authors(metadata.author, author\_blacklist) \# url if not metadata.url: metadata.url \= extract\_url(tree, default\_url) \# hostname if metadata.url: metadata.hostname \= extract\_domain(metadata.url, fast\=True) \# extract date with external module htmldate date\_config\["url"\] \= metadata.url metadata.date \= find\_date(tree, \*\*date\_config) \# sitename if not metadata.sitename: metadata.sitename \= extract\_sitename(tree) if metadata.sitename: \# fix: take 1st element (\['Westdeutscher Rundfunk'\]) if isinstance(metadata.sitename, list): metadata.sitename \= metadata.sitename\[0\] \# hotfix: probably an error coming from json\_metadata (#195) elif isinstance(metadata.sitename, dict): metadata.sitename \= str(metadata.sitename) \# scrap Twitter ID metadata.sitename \= metadata.sitename.lstrip("@") \# capitalize if ( metadata.sitename and "." not in metadata.sitename and not metadata.sitename\[0\].isupper() ): metadata.sitename \= metadata.sitename.title() \# use URL elif metadata.url: mymatch \= META\_URL.match(metadata.url) if mymatch: metadata.sitename \= mymatch\[1\] \# categories if not metadata.categories: metadata.categories \= extract\_catstags("category", tree) \# tags if not metadata.tags: metadata.tags \= extract\_catstags("tag", tree) \# license metadata.license \= extract\_license(tree) \# safety checks metadata.filedate \= date\_config\["max\_date"\] metadata.clean\_and\_trim() return metadata --- # trafilatura.spider — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/spider.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.spider ================================== \# pylint:disable-msg=E0611,E1101,I1101 """ Functions dedicated to website navigation and crawling/spidering. """ import logging from configparser import ConfigParser from time import sleep from typing import List, Optional, Tuple from urllib.robotparser import RobotFileParser from courlan import ( UrlStore, extract\_links, fix\_relative\_urls, get\_base\_url, is\_navigation\_page, is\_not\_crawlable, ) try: import py3langid \# type: ignore except ImportError: pass from lxml.etree import XPath, tostring from .core import baseline, prune\_unwanted\_nodes from .downloads import Response, fetch\_response, fetch\_url from .settings import DEFAULT\_CONFIG from .utils import LANGID\_FLAG, decode\_file, load\_html LOGGER \= logging.getLogger(\_\_name\_\_) URL\_STORE \= UrlStore(compressed\=False, strict\=False) ROBOTS\_TXT\_URL \= "/robots.txt" MAX\_SEEN\_URLS \= 10 MAX\_KNOWN\_URLS \= 100000 class CrawlParameters: "Store necessary information to manage a focused crawl." \_\_slots\_\_ \= \["start", "base", "lang", "rules", "ref", "i", "known\_num", "is\_on", "prune\_xpath"\] def \_\_init\_\_( self, start: str, lang: Optional\[str\] \= None, rules: Optional\[RobotFileParser\] \= None, prune\_xpath: Optional\[str\] \= None, ) \-> None: self.start: str \= start self.base: str \= self.\_get\_base\_url(start) self.ref: str \= self.\_get\_reference(start) self.lang: Optional\[str\] \= lang self.rules: Optional\[RobotFileParser\] \= rules or get\_rules(self.base) self.i: int \= 0 self.known\_num: int \= 0 self.is\_on: bool \= True self.prune\_xpath: Optional\[str\] \= prune\_xpath def \_get\_base\_url(self, start: str) \-> str: "Set reference domain for the crawl." base: str \= get\_base\_url(start) if not base: raise ValueError(f"cannot start crawl: {start}") return base def \_get\_reference(self, start: str) \-> str: "Determine the reference URL." return start.rsplit("/", 1)\[0\] if start.count("/") \>= 3 else start def update\_metadata(self, url\_store: UrlStore) \-> None: "Adjust crawl data based on URL store info." self.is\_on \= bool(url\_store.find\_unvisited\_urls(self.base)) self.known\_num \= len(url\_store.find\_known\_urls(self.base)) def filter\_list(self, todo: Optional\[List\[str\]\]) \-> List\[str\]: "Prepare the todo list, excluding invalid URLs." if not todo: return \[\] return \[u for u in todo if u != self.start and self.ref in u\] def is\_valid\_link(self, link: str) \-> bool: "Run checks: robots.txt rules, URL type and crawl breadth." return ( (not self.rules or self.rules.can\_fetch("\*", link)) and self.ref in link and not is\_not\_crawlable(link) ) def refresh\_detection( htmlstring: str, homepage: str ) \-> Tuple\[Optional\[str\], Optional\[str\]\]: "Check if there could be a redirection by meta-refresh tag." if '"refresh"' not in htmlstring and '"REFRESH"' not in htmlstring: return htmlstring, homepage html\_tree \= load\_html(htmlstring) if html\_tree is None: return htmlstring, homepage \# test meta-refresh redirection \# https://stackoverflow.com/questions/2318446/how-to-follow-meta-refreshes-in-python results \= html\_tree.xpath( './/meta\[@http-equiv="refresh" or @http-equiv="REFRESH"\]/@content' ) result \= results\[0\] if results else "" if not result or ";" not in result: logging.info("no redirect found: %s", homepage) return htmlstring, homepage url2 \= result.split(";")\[1\].strip().lower().replace("url=", "") if not url2.startswith("http"): \# Relative URL, adapt base\_url \= get\_base\_url(url2) url2 \= fix\_relative\_urls(base\_url, url2) \# second fetch newhtmlstring \= fetch\_url(url2) if newhtmlstring is None: logging.warning("failed redirect: %s", url2) return None, None \# else: logging.info("successful redirect: %s", url2) return newhtmlstring, url2 def probe\_alternative\_homepage( homepage: str, ) \-> Tuple\[Optional\[str\], Optional\[str\], Optional\[str\]\]: "Check if the homepage is redirected and return appropriate values." response \= fetch\_response(homepage, decode\=False) if not response or not response.data: return None, None, None \# get redirected URL here? if response.url not in (homepage, "/"): logging.info("followed homepage redirect: %s", response.url) homepage \= response.url \# decode response htmlstring \= decode\_file(response.data) \# is there a meta-refresh on the page? new\_htmlstring, new\_homepage \= refresh\_detection(htmlstring, homepage) if new\_homepage is None: \# malformed or malicious content return None, None, None logging.debug("fetching homepage OK: %s", new\_homepage) return new\_htmlstring, new\_homepage, get\_base\_url(new\_homepage) def parse\_robots(robots\_url: str, data: str) \-> Optional\[RobotFileParser\]: "Parse a robots.txt file with the standard library urllib.robotparser." \# https://github.com/python/cpython/blob/main/Lib/urllib/robotparser.py rules \= RobotFileParser() rules.set\_url(robots\_url) \# exceptions happening here try: rules.parse(data.splitlines()) except Exception as exc: LOGGER.error("cannot read robots.txt: %s", exc) return None return rules def get\_rules(base\_url: str) \-> Optional\[RobotFileParser\]: "Attempt to fetch and parse robots.txt file for a given website." robots\_url \= base\_url + ROBOTS\_TXT\_URL data \= fetch\_url(robots\_url) return parse\_robots(robots\_url, data) if data else None def is\_target\_language(htmlstring: str, language: Optional\[str\]) \-> bool: """Run a baseline extraction and use a language detector to check if the content matches the target language. Return True if language checks are bypassed.""" if htmlstring and language and LANGID\_FLAG: \_, text, \_ \= baseline(htmlstring) result, \_ \= py3langid.classify(text) return bool(result \== language) return True def is\_still\_navigation(todo: List\[str\]) \-> bool: """Probe if there are still navigation URLs in the queue.""" return any(is\_navigation\_page(url) for url in todo) def process\_links( htmlstring: str, params: CrawlParameters, url: Optional\[str\] \= "", ) \-> None: """Examine the HTML code and process the retrieved internal links. Extract and filter new internal links after an optional language check. Store the links in todo-list while prioritizing the navigation ones.""" if not is\_target\_language(htmlstring, params.lang): return if htmlstring and params.prune\_xpath is not None: if isinstance(params.prune\_xpath, str): params.prune\_xpath \= \[params.prune\_xpath\] \# type: ignore\[assignment\] tree \= load\_html(htmlstring) if tree is not None: tree \= prune\_unwanted\_nodes(tree, \[XPath(x) for x in params.prune\_xpath\]) htmlstring \= tostring(tree).decode() links, links\_priority \= \[\], \[\] for link in extract\_links( pagecontent\=htmlstring, url\=url or params.base, external\_bool\=False, language\=params.lang, with\_nav\=True, strict\=False, ): if not params.is\_valid\_link(link): continue if is\_navigation\_page(link): links\_priority.append(link) else: links.append(link) URL\_STORE.add\_urls(urls\=links, appendleft\=links\_priority) def process\_response( response: Optional\[Response\], params: CrawlParameters, ) \-> None: """Convert urllib3 response object and extract links.""" if response is None or not response.data: return \# add final document URL to known\_links URL\_STORE.add\_urls(\[response.url\], visited\=True) \# convert urllib3 response to string and proceed to link extraction process\_links(decode\_file(response.data), params, params.base) def init\_crawl( start: str, lang: Optional\[str\] \= None, rules: Optional\[RobotFileParser\] \= None, todo: Optional\[List\[str\]\] \= None, known: Optional\[List\[str\]\] \= None, prune\_xpath: Optional\[str\] \= None, ) \-> CrawlParameters: """Initialize crawl by setting variables, copying values to the URL store and retrieving the initial page if the crawl starts.""" params \= CrawlParameters(start, lang, rules, prune\_xpath) \# todo: just known or also visited? URL\_STORE.add\_urls(urls\=known or \[\], visited\=True) URL\_STORE.add\_urls(urls\=params.filter\_list(todo)) URL\_STORE.store\_rules(params.base, params.rules) \# visiting the start page if necessary if not todo: URL\_STORE.add\_urls(urls\=\[params.start\], visited\=False) params \= crawl\_page(params, initial\=True) else: params.update\_metadata(URL\_STORE) return params def crawl\_page( params: CrawlParameters, initial: bool \= False, ) \-> CrawlParameters: """Examine a webpage, extract navigation links and links.""" \# config=DEFAULT\_CONFIG url \= URL\_STORE.get\_url(params.base) if not url: params.is\_on \= False params.known\_num \= len(URL\_STORE.find\_known\_urls(params.base)) return params params.i += 1 if initial: \# probe and process homepage htmlstring, homepage, new\_base\_url \= probe\_alternative\_homepage(url) if htmlstring and homepage and new\_base\_url: \# register potentially new homepage URL\_STORE.add\_urls(\[homepage\]) \# extract links on homepage process\_links(htmlstring, params, url\=url) else: response \= fetch\_response(url, decode\=False) process\_response(response, params) \# optional backup of gathered pages without nav-pages ? ... params.update\_metadata(URL\_STORE) return params [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.spider.focused_crawler) def focused\_crawler( homepage: str, max\_seen\_urls: int \= MAX\_SEEN\_URLS, max\_known\_urls: int \= MAX\_KNOWN\_URLS, todo: Optional\[List\[str\]\] \= None, known\_links: Optional\[List\[str\]\] \= None, lang: Optional\[str\] \= None, config: ConfigParser \= DEFAULT\_CONFIG, rules: Optional\[RobotFileParser\] \= None, prune\_xpath: Optional\[str\] \= None, ) \-> Tuple\[List\[str\], List\[str\]\]: """Basic crawler targeting pages of interest within a website. Args: homepage: URL of the page to first page to fetch, preferably the homepage of a website. max\_seen\_urls: maximum number of pages to visit, stop iterations at this number or at the exhaustion of pages on the website, whichever comes first. max\_known\_urls: stop if the total number of pages "known" exceeds this number. todo: provide a previously generated list of pages to visit / crawl frontier. known\_links: provide a list of previously known pages. lang: try to target links according to language heuristics. config: use a different configuration (configparser format). rules: provide politeness rules (urllib.robotparser.RobotFileParser() format). prune\_xpath: remove unwanted elements from the HTML pages using XPath. Returns: List of pages to visit, deque format, possibly empty if there are no further pages to visit. Set of known links. """ params \= init\_crawl(homepage, lang, rules, todo, known\_links, prune\_xpath) sleep\_time \= URL\_STORE.get\_crawl\_delay( params.base, default\=config.getfloat("DEFAULT", "SLEEP\_TIME") ) \# visit pages until a limit is reached while ( params.is\_on and params.i < max\_seen\_urls and params.known\_num < max\_known\_urls ): params \= crawl\_page(params) sleep(sleep\_time) \# refocus todo-list on URLs without navigation? todo \= list(dict.fromkeys(URL\_STORE.find\_unvisited\_urls(params.base))) \# \[u for u in todo if not is\_navigation\_page(u)\] known\_links \= list(dict.fromkeys(URL\_STORE.find\_known\_urls(params.base))) return todo, known\_links --- # trafilatura.utils — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/utils.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.utils ================================= \# pylint:disable-msg=E0611,I1101 """ Module bundling functions related to HTML and text processing, content filtering and language detection. """ try: import gzip HAS\_GZIP \= True except ImportError: HAS\_GZIP \= False import logging import re try: import zlib HAS\_ZLIB \= True except ImportError: HAS\_ZLIB \= False from functools import lru\_cache from itertools import islice from typing import Any, List, Literal, Optional, Tuple, Union from unicodedata import normalize \# response compression try: import brotli \# type: ignore HAS\_BROTLI \= True except ImportError: HAS\_BROTLI \= False try: import zstandard HAS\_ZSTD \= True except ImportError: HAS\_ZSTD \= False \# language detection try: import py3langid \# type: ignore LANGID\_FLAG \= True except ImportError: LANGID\_FLAG \= False \# CChardet is faster and can be more accurate try: from cchardet import detect as cchardet\_detect \# type: ignore except ImportError: cchardet\_detect \= None from charset\_normalizer import from\_bytes from lxml.etree import \_Element from lxml.html import HtmlElement, HTMLParser, fromstring \# response types from urllib3.response import HTTPResponse LOGGER \= logging.getLogger(\_\_name\_\_) UNICODE\_ALIASES \= {'utf-8', 'utf\_8'} DOCTYPE\_TAG \= re.compile("^< ?! ?DOCTYPE.+?/ ?>", re.I) FAULTY\_HTML \= re.compile(r"(<html.\*?)\\s\*/>", re.I) HTML\_STRIP\_TAGS \= re.compile(r'(|<\[^>\]\*>)') \# note: htmldate could use HTML comments \# huge\_tree=True, remove\_blank\_text=True HTML\_PARSER \= HTMLParser(collect\_ids\=False, default\_doctype\=False, encoding\='utf-8', remove\_comments\=True, remove\_pis\=True) LINES\_TRIMMING \= re.compile(r'(?<!\[p{P}\>\])\\n', flags\=re.UNICODE|re.MULTILINE) URL\_BLACKLIST\_REGEX \= re.compile(r'^https?://|/+$') \# Regex to check image file extensions IMAGE\_EXTENSION \= re.compile(r'\[^\\s\]+\\.(avif|bmp|gif|hei\[cf\]|jpe?g|png|webp)(\\b|$)') FORMATTING\_PROTECTED \= {'cell', 'head', 'hi', 'item', 'p', 'quote', 'ref', 'td'} SPACING\_PROTECTED \= {'code', 'pre'} \# https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Language TARGET\_LANG\_ATTRS \= ('http-equiv="content-language"', 'property="og:locale"') RE\_HTML\_LANG \= re.compile(r'(\[a-z\]{2})') \# Mostly filters for social media RE\_FILTER \= re.compile(r'\\W\*(Drucken|E-?Mail|Facebook|Flipboard|Google|Instagram|' 'Linkedin|Mail|PDF|Pinterest|Pocket|Print|QQ|Reddit|Twitter|' 'WeChat|WeiBo|Whatsapp|Xing|Mehr zum Thema:?|More on this.{,8}$)$', flags\=re.IGNORECASE) \# COMMENTS\_BLACKLIST = ('( Abmelden / Ändern )') # Fill in your details below|Trage deine Daten unten|Kommentar verfassen|Bitte logge dich|Hinterlasse einen Kommentar| to %s| mit %s) def handle\_compressed\_file(filecontent: bytes) \-> bytes: """ Don't trust response headers and try to decompress a binary string with a cascade of installed packages. Use magic numbers when available. """ if not isinstance(filecontent, bytes): return filecontent \# source: https://stackoverflow.com/questions/3703276/how-to-tell-if-a-file-is-gzip-compressed if HAS\_GZIP and filecontent\[:3\] \== b"\\x1f\\x8b\\x08": try: return gzip.decompress(filecontent) except Exception: \# EOFError, OSError, gzip.BadGzipFile LOGGER.warning("invalid GZ file") \# try zstandard if HAS\_ZSTD and filecontent\[:4\] \== b"\\x28\\xb5\\x2f\\xfd": try: return zstandard.decompress(filecontent) \# max\_output\_size=??? except zstandard.ZstdError: LOGGER.warning("invalid ZSTD file") \# try brotli if HAS\_BROTLI: try: return brotli.decompress(filecontent) \# type: ignore\[no-any-return\] except brotli.error: pass \# logging.debug('invalid Brotli file') \# try zlib/deflate if HAS\_ZLIB: try: return zlib.decompress(filecontent) except zlib.error: pass \# return content unchanged if decompression failed return filecontent def isutf8(data: bytes) \-> bool: """Simple heuristic to determine if a bytestring uses standard unicode encoding""" try: data.decode('UTF-8') except UnicodeDecodeError: return False return True def detect\_encoding(bytesobject: bytes) \-> List\[str\]: """"Read all input or first chunk and return a list of encodings""" \# alternatives: https://github.com/scrapy/w3lib/blob/master/w3lib/encoding.py \# unicode-test if isutf8(bytesobject): return \['utf-8'\] guesses \= \[\] \# additional module if cchardet\_detect is not None: cchardet\_guess \= cchardet\_detect(bytesobject)\['encoding'\] if cchardet\_guess is not None: guesses.append(cchardet\_guess.lower()) \# try charset\_normalizer on first part, fallback on full document if len(bytesobject) < 10000: detection\_results \= from\_bytes(bytesobject) else: detection\_results \= from\_bytes(bytesobject\[:5000\] + bytesobject\[\-5000:\]) or \\ from\_bytes(bytesobject) \# return alternatives if len(detection\_results) \> 0: guesses.extend(\[r.encoding for r in detection\_results\]) \# it cannot be utf-8 (tested above) return \[g for g in guesses if g not in UNICODE\_ALIASES\] [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.decode_file) def decode\_file(filecontent: Union\[bytes, str\]) \-> str: """Check if the bytestring could be GZip and eventually decompress it, guess bytestring encoding and try to decode to Unicode string. Resort to destructive conversion otherwise.""" if isinstance(filecontent, str): return filecontent htmltext \= None \# GZip and Brotli test filecontent \= handle\_compressed\_file(filecontent) \# encoding for guessed\_encoding in detect\_encoding(filecontent): try: htmltext \= filecontent.decode(guessed\_encoding) except (LookupError, UnicodeDecodeError): \# VISCII: lookup LOGGER.warning('wrong encoding detected: %s', guessed\_encoding) htmltext \= None else: break \# return original content if nothing else succeeded return htmltext or str(filecontent, encoding\='utf-8', errors\='replace') def is\_dubious\_html(beginning: str) \-> bool: "Assess if the object is proper HTML (awith a corresponding tag or declaration)." return "html" not in beginning def repair\_faulty\_html(htmlstring: str, beginning: str) \-> str: "Repair faulty HTML strings to make then palatable for libxml2." \# libxml2/LXML issue: https://bugs.launchpad.net/lxml/+bug/1955915 if "doctype" in beginning: firstline, \_, rest \= htmlstring.partition("\\n") htmlstring \= DOCTYPE\_TAG.sub("", firstline, count\=1) + "\\n" + rest \# other issue with malformed documents: check first three lines for i, line in enumerate(iter(htmlstring.splitlines())): if "<html" in line and line.endswith("/>"): htmlstring \= FAULTY\_HTML.sub(r"\\1>", htmlstring, count\=1) break if i \> 2: break return htmlstring def fromstring\_bytes(htmlobject: str) \-> Optional\[HtmlElement\]: "Try to pass bytes to LXML parser." tree \= None try: tree \= fromstring(htmlobject.encode("utf8", "surrogatepass"), parser\=HTML\_PARSER) except Exception as err: LOGGER.error("lxml parser bytestring %s", err) return tree [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.load_html) def load\_html(htmlobject: Any) \-> Optional\[HtmlElement\]: """Load object given as input and validate its type (accepted: lxml.html tree, trafilatura/urllib3 response, bytestring and string) """ \# use tree directly if isinstance(htmlobject, HtmlElement): return htmlobject \# use trafilatura or urllib3 responses directly if isinstance(htmlobject, HTTPResponse) or hasattr(htmlobject, "data"): htmlobject \= htmlobject.data \# do not accept any other type after this point if not isinstance(htmlobject, (bytes, str)): raise TypeError("incompatible input type", type(htmlobject)) \# start processing tree \= None \# try to guess encoding and decode file: if None then keep original htmlobject \= decode\_file(htmlobject) \# sanity checks beginning \= htmlobject\[:50\].lower() check\_flag \= is\_dubious\_html(beginning) \# repair first htmlobject \= repair\_faulty\_html(htmlobject, beginning) \# first pass: use Unicode string fallback\_parse \= False try: tree \= fromstring(htmlobject, parser\=HTML\_PARSER) except ValueError: \# "Unicode strings with encoding declaration are not supported." tree \= fromstring\_bytes(htmlobject) fallback\_parse \= True except Exception as err: \# pragma: no cover LOGGER.error("lxml parsing failed: %s", err) \# second pass: try passing bytes to LXML if (tree is None or len(tree) < 1) and not fallback\_parse: tree \= fromstring\_bytes(htmlobject) \# rejection test: is it (well-formed) HTML at all? \# log parsing errors if tree is not None and check\_flag is True and len(tree) < 2: LOGGER.error( "parsed tree length: %s, wrong data type or not valid HTML", len(tree) ) tree \= None return tree @lru\_cache(maxsize\=2\*\*14) \# sys.maxunicode = 1114111 def return\_printables\_and\_spaces(char: str) \-> str: 'Return a character if it belongs to certain classes' return char if char.isprintable() or char.isspace() else '' def remove\_control\_characters(string: str) \-> str: '''Prevent non-printable and XML invalid character errors''' return ''.join(map(return\_printables\_and\_spaces, string)) def normalize\_unicode(string: str, unicodeform: Literal\['NFC', 'NFD', 'NFKC', 'NFKD'\] \= 'NFC') \-> str: 'Normalize the given string to the specified unicode format.' return normalize(unicodeform, string) @lru\_cache(maxsize\=1024) def line\_processing(line: str, preserve\_space: bool \= False, trailing\_space: bool \= False) \-> Optional\[str\]: '''Remove HTML space entities, then discard incompatible unicode and invalid XML characters on line level''' \# spacing HTML entities: https://www.w3.org/MarkUp/html-spec/html-spec\_13.html \# unique code spaces new\_line \= remove\_control\_characters(line.replace(' ', '\\r').replace(' ', '\\n').replace(' ', '\\u00A0')) if not preserve\_space: \# remove newlines that are not related to punctuation or markup \# remove non-printable chars and normalize space characters (including Unicode spaces) new\_line \= trim(LINES\_TRIMMING.sub(r" ", new\_line)) \# prune empty lines if all(map(str.isspace, new\_line)): new\_line \= None \# type: ignore\[assignment\] elif trailing\_space: space\_before \= " " if line\[0\].isspace() else "" space\_after \= " " if line\[\-1\].isspace() else "" new\_line \= "".join(\[space\_before, new\_line, space\_after\]) return new\_line [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.sanitize) def sanitize(text: str, preserve\_space: bool \= False, trailing\_space: bool \= False) \-> Optional\[str\]: '''Convert text and discard incompatible and invalid characters''' \# consider all text as a single line if trailing\_space: return line\_processing(text, preserve\_space, True) \# process line by line try: return '\\n'.join(filter(None, (line\_processing(l, preserve\_space) for l in text.splitlines()))).replace('\\u2424', '') except AttributeError: return None def sanitize\_tree(tree: \_Element) \-> \_Element: '''Trims spaces, removes control characters and normalizes unicode''' for elem in tree.iter(): parent \= elem.getparent() parent\_tag \= parent.tag if parent is not None else "" \# preserve space if the element or its parent is a specific tag, or if the element has text and children \# the last part is relevant for item elements with ref inside for example preserve\_space \= elem.tag in SPACING\_PROTECTED or parent\_tag in SPACING\_PROTECTED trailing\_space \= elem.tag in FORMATTING\_PROTECTED or parent\_tag in FORMATTING\_PROTECTED or preserve\_space \# remove invalid attributes for attribute in elem.attrib: if ':' in attribute: \# colon is reserved for namespaces in XML if not elem.attrib\[attribute\] or attribute.split(':', 1)\[0\] not in tree.nsmap: elem.attrib.pop(attribute) if elem.text: elem.text \= sanitize(elem.text, preserve\_space, trailing\_space) if elem.tail: elem.tail \= sanitize(elem.tail, preserve\_space, trailing\_space) return tree [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.utils.trim) @lru\_cache(maxsize\=1024) def trim(string: str) \-> str: "Remove unnecessary spaces within a text string." try: \# remove newlines that are not related to punctuation or markup + proper trimming return " ".join(string.split()).strip() except (AttributeError, TypeError): return "" def is\_image\_element(element: \_Element) \-> bool: '''Check if an element is a valid img element''' for attr in ("data-src", "src"): src \= element.get(attr, "") if is\_image\_file(src): return True else: \# take the first corresponding attribute for attr, value in element.attrib.items(): if attr.startswith("data-src") and is\_image\_file(value): return True return False def is\_image\_file(imagesrc: Optional\[str\]) \-> bool: '''Check if the observed string corresponds to a valid image extension. Use a length threshold and apply a regex on the content.''' if imagesrc is None or len(imagesrc) \> 8192: return False return bool(IMAGE\_EXTENSION.search(imagesrc)) def make\_chunks(iterable: Any, n: int) \-> Any: "Chunk data into smaller pieces." \# 3.12+: https://docs.python.org/3/library/itertools.html#itertools.batched iterator \= iter(iterable) while batch := tuple(islice(iterator, n)): yield batch def is\_acceptable\_length(my\_len: int, options: Any) \-> bool: "Check if the document length is within acceptable boundaries." if my\_len < options.min\_file\_size: LOGGER.error("too small/incorrect for URL %s", options.url) return False if my\_len \> options.max\_file\_size: LOGGER.error("too large: length %s for URL %s", my\_len, options.url) return False return True def check\_html\_lang(tree: HtmlElement, target\_language: str, strict: bool \= False) \-> bool: """Check HTML meta-elements for language information and split the result in case there are several languages.""" for attr in TARGET\_LANG\_ATTRS: elems \= tree.findall(f'.//meta\[@{attr}\]\[@content\]') if elems: if any(target\_language in RE\_HTML\_LANG.split(elem.get("content", "").lower()) for elem in elems): return True LOGGER.debug("%s lang attr failed", attr) return False \# HTML lang attribute: sometimes a wrong indication if strict: elems \= tree.xpath("//html\[@lang\]") if elems: if any(target\_language in RE\_HTML\_LANG.split(elem.get("lang", "").lower()) for elem in elems): return True LOGGER.debug("HTML lang failed") return False LOGGER.debug("No relevant lang elements found") return True def language\_classifier(temp\_text: str, temp\_comments: str) \-> Optional\[str\]: '''Run external component (if installed) for language identification''' if LANGID\_FLAG is True: result, \_ \= ( py3langid.classify(temp\_text) if len(temp\_text) \> len(temp\_comments) else py3langid.classify(temp\_comments) ) else: \# pragma: no cover LOGGER.warning('Language detector not installed, skipping detection') result \= None return result \# type: ignore\[no-any-return\] def language\_filter(temp\_text: str, temp\_comments: str, target\_language: str, docmeta: Any) \-> Tuple\[bool, Any\]: '''Filter text based on language detection and store relevant information''' \# todo: run and pass info along anyway? if target\_language is not None: \# more thorough: detection on actual text content docmeta.language \= language\_classifier(temp\_text, temp\_comments) \# HTML lang check? sometimes contradicted by detection above #if docmeta.language is None: \# if check\_html\_lang(tree, target\_language) is False: \# LOGGER.error('wrong HTML meta language for URL %s', url) \# raise ValueError if docmeta.language is not None and docmeta.language != target\_language: LOGGER.warning('wrong language: %s %s', docmeta.language, docmeta.url) return True, docmeta return False, docmeta def textfilter(element: \_Element) \-> bool: '''Filter out unwanted text''' testtext \= element.tail if element.text is None else element.text \# to check: line len → continue if len(line) <= 5 return not testtext or testtext.isspace() or any(map(RE\_FILTER.match, testtext.splitlines())) def text\_chars\_test(string: Optional\[str\]) \-> bool: '''Determine if a string is only composed of spaces and/or control characters''' \# or not re.search(r'\\w', string) \# return string is not None and len(string) != 0 and not string.isspace() return bool(string) and not string.isspace() \# type: ignore\[union-attr\] --- # trafilatura.xml — Trafilatura 2.0.0 documentation [Skip to main content](https://trafilatura.readthedocs.io/en/latest/_modules/trafilatura/xml.html#main-content) Back to top Ctrl+K * [GitHub](https://github.com/adbar/trafilatura "GitHub") Source code for trafilatura.xml =============================== \# pylint:disable-msg=E0611,I1101 """ All functions related to XML generation, processing and validation. """ import csv import logging from html import unescape from importlib.metadata import version from io import StringIO from json import dumps as json\_dumps from pathlib import Path from typing import List, Optional from lxml.etree import (\_Element, Element, SubElement, XMLParser, fromstring, tostring, DTD) from .settings import Document, Extractor from .utils import sanitize, sanitize\_tree, text\_chars\_test LOGGER \= logging.getLogger(\_\_name\_\_) PKG\_VERSION \= version("trafilatura") \# validation TEI\_SCHEMA \= str(Path(\_\_file\_\_).parent / "data" / "tei\_corpus.dtd") TEI\_VALID\_TAGS \= {'ab', 'body', 'cell', 'code', 'del', 'div', 'graphic', 'head', 'hi', \\ 'item', 'lb', 'list', 'p', 'quote', 'ref', 'row', 'table'} TEI\_VALID\_ATTRS \= {'rend', 'rendition', 'role', 'target', 'type'} TEI\_DTD \= None \# to be downloaded later if necessary TEI\_REMOVE\_TAIL \= {"ab", "p"} TEI\_DIV\_SIBLINGS \= {"p", "list", "table", "quote", "ab"} CONTROL\_PARSER \= XMLParser(remove\_blank\_text\=True) NEWLINE\_ELEMS \= {'code', 'graphic', 'head', 'lb', 'list', 'p', 'quote', 'row', 'table'} SPECIAL\_FORMATTING \= {'del', 'head', 'hi', 'ref'} WITH\_ATTRIBUTES \= {'cell', 'row', 'del', 'graphic', 'head', 'hi', 'item', 'list', 'ref'} NESTING\_WHITELIST \= {"cell", "figure", "item", "note", "quote"} META\_ATTRIBUTES \= \[\ 'sitename', 'title', 'author', 'date', 'url', 'hostname',\ 'description', 'categories', 'tags', 'license', 'id',\ 'fingerprint', 'language'\ \] HI\_FORMATTING \= {'#b': '\*\*', '#i': '\*', '#u': '\_\_', '#t': '\`'} MAX\_TABLE\_WIDTH \= 1000 \# https://github.com/lxml/lxml/blob/master/src/lxml/html/\_\_init\_\_.py def delete\_element(element: \_Element, keep\_tail: bool \= True) \-> None: """ Removes this element from the tree, including its children and text. The tail text is joined to the previous element or parent. """ parent \= element.getparent() if parent is None: return if keep\_tail and element.tail: previous \= element.getprevious() if previous is None: parent.text \= (parent.text or "") + element.tail else: previous.tail \= (previous.tail or "") + element.tail parent.remove(element) def merge\_with\_parent(element: \_Element, include\_formatting: bool \= False) \-> None: '''Merge element with its parent and convert formatting to markdown.''' parent \= element.getparent() if parent is None: return full\_text \= replace\_element\_text(element, include\_formatting) if element.tail is not None: full\_text += element.tail previous \= element.getprevious() if previous is not None: \# There is a previous node, append text to its tail previous.tail \= f'{previous.tail} {full\_text}' if previous.tail else full\_text elif parent.text is not None: parent.text \= f'{parent.text} {full\_text}' else: parent.text \= full\_text parent.remove(element) def remove\_empty\_elements(tree: \_Element) \-> \_Element: '''Remove text elements without text.''' for element in tree.iter('\*'): \# 'head', 'hi', 'item', 'p' if len(element) \== 0 and text\_chars\_test(element.text) is False and text\_chars\_test(element.tail) is False: parent \= element.getparent() \# not root element or element which is naturally empty \# do not remove elements inside <code> to preserve formatting if parent is not None and element.tag != "graphic" and parent.tag != 'code': parent.remove(element) return tree def strip\_double\_tags(tree: \_Element) \-> \_Element: "Prevent nested tags among a fixed list of tags." for elem in reversed(tree.xpath(".//head | .//code | .//p")): for subelem in elem.iterdescendants("code", "head", "p"): if subelem.tag \== elem.tag and subelem.getparent().tag not in NESTING\_WHITELIST: merge\_with\_parent(subelem) return tree def build\_json\_output(docmeta: Document, with\_metadata: bool \= True) \-> str: '''Build JSON output based on extracted information''' if with\_metadata: outputdict \= {slot: getattr(docmeta, slot, None) for slot in docmeta.\_\_slots\_\_} outputdict.update({ 'source': outputdict.pop('url'), 'source-hostname': outputdict.pop('sitename'), 'excerpt': outputdict.pop('description'), 'categories': ';'.join(outputdict.pop('categories') or \[\]), 'tags': ';'.join(outputdict.pop('tags') or \[\]), 'text': xmltotxt(outputdict.pop('body'), include\_formatting\=False), }) commentsbody \= outputdict.pop('commentsbody') else: outputdict \= {'text': xmltotxt(docmeta.body, include\_formatting\=False)} commentsbody \= docmeta.commentsbody outputdict\['comments'\] \= xmltotxt(commentsbody, include\_formatting\=False) return json\_dumps(outputdict, ensure\_ascii\=False) def clean\_attributes(tree: \_Element) \-> \_Element: '''Remove unnecessary attributes.''' for elem in tree.iter('\*'): if elem.tag not in WITH\_ATTRIBUTES: elem.attrib.clear() return tree def build\_xml\_output(docmeta: Document) \-> \_Element: '''Build XML output tree based on extracted information''' output \= Element('doc') add\_xml\_meta(output, docmeta) docmeta.body.tag \= 'main' \# clean XML tree output.append(clean\_attributes(docmeta.body)) docmeta.commentsbody.tag \= 'comments' output.append(clean\_attributes(docmeta.commentsbody)) return output def control\_xml\_output(document: Document, options: Extractor) \-> str: '''Make sure the XML output is conform and valid if required''' strip\_double\_tags(document.body) remove\_empty\_elements(document.body) func \= build\_xml\_output if options.format \== "xml" else build\_tei\_output output\_tree \= func(document) output\_tree \= sanitize\_tree(output\_tree) \# necessary for cleaning output\_tree \= fromstring(tostring(output\_tree, encoding\='unicode'), CONTROL\_PARSER) \# validate if options.format \== 'xmltei' and options.tei\_validation: LOGGER.debug('TEI validation result: %s %s', validate\_tei(output\_tree), options.source) return tostring(output\_tree, pretty\_print\=True, encoding\='unicode').strip() def add\_xml\_meta(output: \_Element, docmeta: Document) \-> None: '''Add extracted metadata to the XML output tree''' for attribute in META\_ATTRIBUTES: value \= getattr(docmeta, attribute, None) if value: output.set(attribute, value if isinstance(value, str) else ';'.join(value)) def build\_tei\_output(docmeta: Document) \-> \_Element: '''Build TEI-XML output tree based on extracted information''' \# build TEI tree output \= write\_teitree(docmeta) \# filter output (strip unwanted elements), just in case \# check and repair output \= check\_tei(output, docmeta.url) return output def check\_tei(xmldoc: \_Element, url: Optional\[str\]) \-> \_Element: '''Check if the resulting XML file is conform and scrub remaining tags''' \# convert head tags for elem in xmldoc.iter('head'): elem.tag \= 'ab' elem.set('type', 'header') parent \= elem.getparent() if parent is None: continue if len(elem) \> 0: new\_elem \= \_tei\_handle\_complex\_head(elem) parent.replace(elem, new\_elem) elem \= new\_elem if parent.tag \== "p": \_move\_element\_one\_level\_up(elem) \# convert <lb/> when child of <div> to <p> for elem in xmldoc.findall(".//text/body//div/lb"): if elem.tail and elem.tail.strip(): elem.tag, elem.text, elem.tail \= 'p', elem.tail, None \# look for elements that are not valid for elem in xmldoc.findall('.//text/body//\*'): \# check elements if elem.tag not in TEI\_VALID\_TAGS: \# disable warnings for chosen categories \# if element.tag not in ('div', 'span'): LOGGER.warning('not a TEI element, removing: %s %s', elem.tag, url) merge\_with\_parent(elem) continue if elem.tag in TEI\_REMOVE\_TAIL: \_handle\_unwanted\_tails(elem) elif elem.tag \== "div": \_handle\_text\_content\_of\_div\_nodes(elem) \_wrap\_unwanted\_siblings\_of\_div(elem) #if len(elem) == 0: \# elem.getparent().remove(elem) \# check attributes for attribute in \[a for a in elem.attrib if a not in TEI\_VALID\_ATTRS\]: LOGGER.warning('not a valid TEI attribute, removing: %s in %s %s', attribute, elem.tag, url) elem.attrib.pop(attribute) return xmldoc [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.xml.validate_tei) def validate\_tei(xmldoc: \_Element) \-> bool: '''Check if an XML document is conform to the guidelines of the Text Encoding Initiative''' global TEI\_DTD if TEI\_DTD is None: \# https://tei-c.org/release/xml/tei/custom/schema/dtd/tei\_corpus.dtd TEI\_DTD \= DTD(TEI\_SCHEMA) result \= TEI\_DTD.validate(xmldoc) if result is False: LOGGER.warning('not a valid TEI document: %s', TEI\_DTD.error\_log.last\_error) return result def replace\_element\_text(element: \_Element, include\_formatting: bool) \-> str: "Determine element text based on just the text of the element. One must deal with the tail separately." elem\_text \= element.text or "" \# handle formatting: convert to markdown if include\_formatting and element.text: if element.tag \== "head": try: number \= int(element.get("rend")\[1\]) \# type: ignore\[index\] except (TypeError, ValueError): number \= 2 elem\_text \= f'{"#" \* number} {elem\_text}' elif element.tag \== "del": elem\_text \= f"~~{elem\_text}~~" elif element.tag \== "hi": rend \= element.get("rend") if rend in HI\_FORMATTING: elem\_text \= f"{HI\_FORMATTING\[rend\]}{elem\_text}{HI\_FORMATTING\[rend\]}" elif element.tag \== "code": if "\\n" in element.text: elem\_text \= f"\`\`\`\\n{elem\_text}\\n\`\`\`" else: elem\_text \= f"\`{elem\_text}\`" \# handle links if element.tag \== "ref": if elem\_text: link\_text \= f"\[{elem\_text}\]" target \= element.get("target") if target: elem\_text \= f"{link\_text}({target})" else: LOGGER.warning("missing link attribute: %s %s'", elem\_text, element.attrib) elem\_text \= link\_text else: LOGGER.warning("empty link: %s %s", elem\_text, element.attrib) \# cells if element.tag \== "cell" and elem\_text and len(element) \> 0: if element\[0\].tag \== 'p': elem\_text \= f"{elem\_text} " if element.getprevious() is not None else f"| {elem\_text} " elif element.tag \== 'cell' and elem\_text: \# add | before first cell elem\_text \= f"{elem\_text}" if element.getprevious() is not None else f"| {elem\_text}" \# lists elif element.tag \== "item" and elem\_text: elem\_text \= f"- {elem\_text}\\n" return elem\_text def process\_element(element: \_Element, returnlist: List\[str\], include\_formatting: bool) \-> None: "Recursively convert a LXML element and its children to a flattened string representation." if element.text: \# this is the text that comes before the first child returnlist.append(replace\_element\_text(element, include\_formatting)) for child in element: process\_element(child, returnlist, include\_formatting) if not element.text and not element.tail: if element.tag \== "graphic": \# add source, default to '' text \= f'{element.get("title", "")} {element.get("alt", "")}' returnlist.append(f'!\[{text.strip()}\]({element.get("src", "")})') \# newlines for textless elements elif element.tag in NEWLINE\_ELEMS: \# add line after table head if element.tag \== "row": cell\_count \= len(element.xpath(".//cell")) \# restrict columns to a maximum of 1000 span\_info \= element.get("colspan") or element.get("span") if not span\_info or not span\_info.isdigit(): max\_span \= 1 else: max\_span \= min(int(span\_info), MAX\_TABLE\_WIDTH) \# row ended so draw extra empty cells to match max\_span if cell\_count < max\_span: returnlist.append(f'{"|" \* (max\_span \- cell\_count)}\\n') \# if this is a head row, draw the separator below if element.xpath("./cell\[@role='head'\]"): returnlist.append(f'\\n|{"---|" \* max\_span}\\n') else: returnlist.append("\\n") elif element.tag != "cell": \# cells still need to append vertical bars \# but nothing more to do with other textless elements return \# Process text \# Common elements (Now processes end-tag logic correctly) if element.tag in NEWLINE\_ELEMS and not element.xpath("ancestor::cell"): \# spacing hack returnlist.append("\\n\\u2424\\n" if include\_formatting and element.tag != 'row' else "\\n") elif element.tag \== "cell": returnlist.append(" | ") elif element.tag not in SPECIAL\_FORMATTING: returnlist.append(" ") \# this is text that comes after the closing tag, so it should be after any NEWLINE\_ELEMS if element.tail: returnlist.append(element.tail) [\[docs\]](https://trafilatura.readthedocs.io/en/latest/corefunctions.html#trafilatura.xml.xmltotxt) def xmltotxt(xmloutput: Optional\[\_Element\], include\_formatting: bool) \-> str: "Convert to plain text format and optionally preserve formatting as markdown." if xmloutput is None: return "" returnlist: List\[str\] \= \[\] process\_element(xmloutput, returnlist, include\_formatting) return unescape(sanitize("".join(returnlist)) or "") def xmltocsv(document: Document, include\_formatting: bool, \*, delim: str \= "\\t", null: str \= "null") \-> str: "Convert the internal XML document representation to a CSV string." \# preprocessing posttext \= xmltotxt(document.body, include\_formatting) or null commentstext \= xmltotxt(document.commentsbody, include\_formatting) or null \# output config output \= StringIO() outputwriter \= csv.writer(output, delimiter\=delim, quoting\=csv.QUOTE\_MINIMAL) \# organize fields outputwriter.writerow(\[d if d else null for d in (\ document.url,\ document.id,\ document.fingerprint,\ document.hostname,\ document.title,\ document.image,\ document.date,\ posttext,\ commentstext,\ document.license,\ document.pagetype\ )\]) return output.getvalue() def write\_teitree(docmeta: Document) \-> \_Element: '''Bundle the extracted post and comments into a TEI tree''' teidoc \= Element('TEI', xmlns\='http://www.tei-c.org/ns/1.0') write\_fullheader(teidoc, docmeta) textelem \= SubElement(teidoc, 'text') textbody \= SubElement(textelem, 'body') \# post postbody \= clean\_attributes(docmeta.body) postbody.tag \= 'div' postbody.set('type', 'entry') textbody.append(postbody) \# comments commentsbody \= clean\_attributes(docmeta.commentsbody) commentsbody.tag \= 'div' commentsbody.set('type', 'comments') textbody.append(commentsbody) return teidoc def \_define\_publisher\_string(docmeta: Document) \-> str: '''Construct a publisher string to include in TEI header''' if docmeta.hostname and docmeta.sitename: publisher \= f'{docmeta.sitename.strip()} ({docmeta.hostname})' else: publisher \= docmeta.hostname or docmeta.sitename or 'N/A' if LOGGER.isEnabledFor(logging.WARNING) and publisher \== 'N/A': LOGGER.warning('no publisher for URL %s', docmeta.url) return publisher def write\_fullheader(teidoc: \_Element, docmeta: Document) \-> \_Element: '''Write TEI header based on gathered metadata''' \# todo: add language info header \= SubElement(teidoc, 'teiHeader') filedesc \= SubElement(header, 'fileDesc') bib\_titlestmt \= SubElement(filedesc, 'titleStmt') SubElement(bib\_titlestmt, 'title', type\='main').text \= docmeta.title if docmeta.author: SubElement(bib\_titlestmt, 'author').text \= docmeta.author publicationstmt\_a \= SubElement(filedesc, 'publicationStmt') publisher\_string \= \_define\_publisher\_string(docmeta) \# license, if applicable if docmeta.license: SubElement(publicationstmt\_a, 'publisher').text \= publisher\_string availability \= SubElement(publicationstmt\_a, 'availability') SubElement(availability, 'p').text \= docmeta.license \# insert an empty paragraph for conformity else: SubElement(publicationstmt\_a, 'p') notesstmt \= SubElement(filedesc, 'notesStmt') if docmeta.id: SubElement(notesstmt, 'note', type\='id').text \= docmeta.id SubElement(notesstmt, 'note', type\='fingerprint').text \= docmeta.fingerprint sourcedesc \= SubElement(filedesc, 'sourceDesc') source\_bibl \= SubElement(sourcedesc, 'bibl') sigle \= ', '.join(filter(None, \[docmeta.sitename, docmeta.date\])) if not sigle: LOGGER.warning('no sigle for URL %s', docmeta.url) source\_bibl.text \= ', '.join(filter(None, \[docmeta.title, sigle\])) SubElement(sourcedesc, 'bibl', type\='sigle').text \= sigle biblfull \= SubElement(sourcedesc, 'biblFull') bib\_titlestmt \= SubElement(biblfull, 'titleStmt') SubElement(bib\_titlestmt, 'title', type\='main').text \= docmeta.title if docmeta.author: SubElement(bib\_titlestmt, 'author').text \= docmeta.author publicationstmt \= SubElement(biblfull, 'publicationStmt') SubElement(publicationstmt, 'publisher').text \= publisher\_string if docmeta.url: SubElement(publicationstmt, 'ptr', type\='URL', target\=docmeta.url) SubElement(publicationstmt, 'date').text \= docmeta.date profiledesc \= SubElement(header, 'profileDesc') abstract \= SubElement(profiledesc, 'abstract') SubElement(abstract, 'p').text \= docmeta.description if docmeta.categories or docmeta.tags: textclass \= SubElement(profiledesc, 'textClass') keywords \= SubElement(textclass, 'keywords') if docmeta.categories: SubElement(keywords, 'term', type\='categories').text \= ','.join(docmeta.categories) if docmeta.tags: SubElement(keywords, 'term', type\='tags').text \= ','.join(docmeta.tags) creation \= SubElement(profiledesc, 'creation') SubElement(creation, 'date', type\="download").text \= docmeta.filedate encodingdesc \= SubElement(header, 'encodingDesc') appinfo \= SubElement(encodingdesc, 'appInfo') application \= SubElement(appinfo, 'application', version\=PKG\_VERSION, ident\='Trafilatura') SubElement(application, 'label').text \= 'Trafilatura' SubElement(application, 'ptr', target\='https://github.com/adbar/trafilatura') return header def \_handle\_text\_content\_of\_div\_nodes(element: \_Element) \-> None: "Wrap loose text in <div> within <p> elements for TEI conformity." if element.text and element.text.strip(): if len(element) \> 0 and element\[0\].tag \== "p": element\[0\].text \= f'{element.text} {element\[0\].text or ""}'.strip() else: new\_child \= Element("p") new\_child.text \= element.text element.insert(0, new\_child) element.text \= None if element.tail and element.tail.strip(): if len(element) \> 0 and element\[\-1\].tag \== "p": element\[\-1\].text \= f'{element\[\-1\].text or ""} {element.tail}'.strip() else: new\_child \= Element("p") new\_child.text \= element.tail element.append(new\_child) element.tail \= None def \_handle\_unwanted\_tails(element: \_Element) \-> None: "Handle tail on p and ab elements" element.tail \= element.tail.strip() if element.tail else None if not element.tail: return if element.tag \== "p": element.text \= " ".join(filter(None, \[element.text, element.tail\])) else: new\_sibling \= Element('p') new\_sibling.text \= element.tail parent \= element.getparent() if parent is not None: parent.insert(parent.index(element) + 1 , new\_sibling) element.tail \= None def \_tei\_handle\_complex\_head(element: \_Element) \-> \_Element: "Convert certain child elements to <ab> and <lb>." new\_element \= Element('ab', attrib\=element.attrib) new\_element.text \= element.text.strip() if element.text else None for child in element.iterchildren(): if child.tag \== 'p': if len(new\_element) \> 0 or new\_element.text: \# add <lb> if <ab> has no children or last tail contains text if len(new\_element) \== 0 or new\_element\[\-1\].tail: SubElement(new\_element, 'lb') new\_element\[\-1\].tail \= child.text else: new\_element.text \= child.text else: new\_element.append(child) tail \= element.tail.strip() if element.tail else None if tail: new\_element.tail \= tail return new\_element def \_wrap\_unwanted\_siblings\_of\_div(div\_element: \_Element) \-> None: "Wrap unwanted siblings of a div element in a new div element." new\_sibling \= Element("div") new\_sibling\_index \= None parent \= div\_element.getparent() if parent is None: return \# check siblings after target element for sibling in div\_element.itersiblings(): if sibling.tag \== "div": break if sibling.tag in TEI\_DIV\_SIBLINGS: new\_sibling\_index \= new\_sibling\_index or parent.index(sibling) new\_sibling.append(sibling) \# some elements (e.g. <lb/>) can appear next to div, but \# order of elements should be kept, thus add and reset new\_sibling else: if new\_sibling\_index and len(new\_sibling) \> 0: parent.insert(new\_sibling\_index, new\_sibling) new\_sibling \= Element("div") new\_sibling\_index \= None if new\_sibling\_index and len(new\_sibling) != 0: parent.insert(new\_sibling\_index, new\_sibling) def \_move\_element\_one\_level\_up(element: \_Element) \-> None: """ Fix TEI compatibility issues by moving certain p-elems up in the XML tree. There is always a n+2 nesting for p-elements with the minimal structure ./TEI/text/body/p """ parent \= element.getparent() grand\_parent \= parent.getparent() if parent is not None else None if parent is None or grand\_parent is None: return new\_elem \= Element("p") new\_elem.extend(list(element.itersiblings())) grand\_parent.insert(grand\_parent.index(parent) + 1, element) tail \= element.tail.strip() if element.tail else None if tail: new\_elem.text \= tail element.tail \= None tail \= parent.tail.strip() if parent.tail else None if tail: new\_elem.tail \= tail parent.tail \= None if len(new\_elem) \> 0 or new\_elem.text or new\_elem.tail: grand\_parent.insert(grand\_parent.index(element) + 1, new\_elem) if len(parent) \== 0 and not parent.text: grand\_parent.remove(parent) ---