Source code for linkify_it.main

import copy
import re
import types

from .ucre import build_re

# py>=37: re.Pattern, else: _sre.SRE_Pattern
RE_TYPE = type(re.compile(r""))


def _escape_re(string):
    return re.sub(r"([.?*+^$[\]\\(){}|-])", r"\\\1", string)


def _index_of(text, search_value):
    try:
        result = text.index(search_value)
    except ValueError:
        result = -1

    return result


[docs] class SchemaError(Exception): """Linkify schema error""" def __init__(self, name, val): message = "(LinkifyIt) Invalid schema '{}': '{}'".format(name, val) super().__init__(message)
[docs] class Match: """Match result. Attributes: schema (str): Prefix (protocol) for matched string. index (int): First position of matched string. last_index (int): Next position after matched string. raw (str): Matched string. text (str): Notmalized text of matched string. url (str): Normalized url of matched string. Args: linkifyit (:class:`linkify_it.main.LinkifyIt`) LinkifyIt object shift (int): text searh position """ def __repr__(self): return "{}.{}({!r})".format( self.__class__.__module__, self.__class__.__name__, self.__dict__ ) def __init__(self, linkifyit, shift): start = linkifyit._index end = linkifyit._last_index text = linkifyit._text_cache[start:end] self.schema = linkifyit._schema.lower() self.index = start + shift self.last_index = end + shift self.raw = text self.text = text self.url = text
[docs] class LinkifyIt: """Creates new linkifier instance with optional additional schemas. By default understands: - ``http(s)://...`` , ``ftp://...``, ``mailto:...`` & ``//...`` links - "fuzzy" links and emails (example.com, foo@bar.com). ``schemas`` is an dict where each key/value describes protocol/rule: - **key** - link prefix (usually, protocol name with ``:`` at the end, ``skype:`` for example). `linkify-it` makes shure that prefix is not preceeded with alphanumeric char. Only whitespaces and punctuation allowed. - **value** - rule to check tail after link prefix - *str* - just alias to existing rule - *dict* - *validate* - either a ``re.Pattern``, ``re str`` (start with ``^``, and don't include the link prefix itself), or a validator ``function`` which, given arguments *self*, *text* and *pos* returns the length of a match in *text* starting at index *pos*. *pos* is the index right after the link prefix. - *normalize* - optional function to normalize text & url of matched result (for example, for @twitter mentions). ``options`` is an dict: - **fuzzyLink** - recognige URL-s without ``http(s):`` prefix. Default ``True``. - **fuzzyIP** - allow IPs in fuzzy links above. Can conflict with some texts like version numbers. Default ``False``. - **fuzzyEmail** - recognize emails without ``mailto:`` prefix. - **---** - set `True` to terminate link with `---` (if it's considered as long dash). Args: schemas (dict): Optional. Additional schemas to validate (prefix/validator) options (dict): { fuzzy_link | fuzzy_email | fuzzy_ip: True | False }. Default: {"fuzzy_link": True, "fuzzy_email": True, "fuzzy_ip": False}. """ def _validate_http(self, text, pos): tail = text[pos:] if not self.re.get("http"): # compile lazily, because "host"-containing variables can change on # tlds update. self.re["http"] = ( "^\\/\\/" + self.re["src_auth"] + self.re["src_host_port_strict"] + self.re["src_path"] ) founds = re.search(self.re["http"], tail, flags=re.IGNORECASE) if founds: return len(founds.group()) return 0 def _validate_double_slash(self, text, pos): tail = text[pos:] if not self.re.get("not_http"): # compile lazily, because "host"-containing variables can change on # tlds update. self.re["not_http"] = ( "^" + self.re["src_auth"] + "(?:localhost|(?:(?:" + self.re["src_domain"] + ")\\.)+" + self.re["src_domain_root"] + ")" + self.re["src_port"] + self.re["src_host_terminator"] + self.re["src_path"] ) founds = re.search(self.re["not_http"], tail, flags=re.IGNORECASE) if founds: if pos >= 3 and text[pos - 3] == ":": return 0 if pos >= 3 and text[pos - 3] == "/": return 0 return len(founds.group(0)) return 0 def _validate_mailto(self, text, pos): tail = text[pos:] if not self.re.get("mailto"): self.re["mailto"] = ( "^" + self.re["src_email_name"] + "@" + self.re["src_host_strict"] ) founds = re.search(self.re["mailto"], tail, flags=re.IGNORECASE) if founds: return len(founds.group(0)) return 0 def _reset_scan_cache(self): self._index = -1 self._text_cache = "" def _create_validator(self, regex): def func(text, pos): tail = text[pos:] if isinstance(regex, str): founds = re.search(regex, tail, flags=re.IGNORECASE) else: # re.Pattern founds = re.search(regex, tail) if founds: return len(founds.group(0)) return 0 return func def _create_normalizer(self): def func(match): self.normalize(match) return func def _create_match(self, shift): match = Match(self, shift) self._compiled[match.schema]["normalize"](match) return match def __init__(self, schemas=None, options=None): self.default_options = { "fuzzy_link": True, "fuzzy_email": True, "fuzzy_ip": False, } self.default_schemas = { "http:": {"validate": self._validate_http}, "https:": "http:", "ftp:": "http:", "//": {"validate": self._validate_double_slash}, "mailto:": {"validate": self._validate_mailto}, } # RE pattern for 2-character tlds (autogenerated by ./support/tlds_2char_gen.js) self.tlds_2ch_src_re = "a[cdefgilmnoqrstuwxz]|b[abdefghijmnorstvwyz]|c[acdfghiklmnoruvwxyz]|d[ejkmoz]|e[cegrstu]|f[ijkmor]|g[abdefghilmnpqrstuwy]|h[kmnrtu]|i[delmnoqrst]|j[emop]|k[eghimnprwyz]|l[abcikrstuvy]|m[acdeghklmnopqrstuvwxyz]|n[acefgilopruz]|om|p[aefghklmnrstwy]|qa|r[eosuw]|s[abcdeghijklmnortuvxyz]|t[cdfghjklmnortvwz]|u[agksyz]|v[aceginu]|w[fs]|y[et]|z[amw]" # noqa: E501 # DON'T try to make PRs with changes. Extend TLDs with LinkifyIt.tlds() instead self.tlds_default = "biz|com|edu|gov|net|org|pro|web|xxx|aero|asia|coop|info|museum|name|shop|рф".split( # noqa: E501 "|" ) if options: self.default_options.update(options) self._opts = self.default_options else: self._opts = self.default_options # Cache last tested result. Used to skip repeating steps on next `match` call. self._index = -1 self._last_index = -1 # Next scan position self._schema = "" self._text_cache = "" if schemas: self.default_schemas.update(schemas) self._schemas = self.default_schemas else: self._schemas = self.default_schemas self._compiled = {} self._tlds = self.tlds_default self._tlds_replaced = False self.re = {} self._compile() def _compile(self): """Schemas compiler. Build regexps.""" # Load & clone RE patterns. self.re = build_re(self._opts) # Define dynamic patterns tlds = copy.deepcopy(self._tlds) self._on_compile() if not self._tlds_replaced: tlds.append(self.tlds_2ch_src_re) tlds.append(self.re["src_xn"]) self.re["src_tlds"] = "|".join(tlds) def untpl(tpl): return tpl.replace("%TLDS%", self.re["src_tlds"]) self.re["email_fuzzy"] = untpl(self.re["tpl_email_fuzzy"]) self.re["link_fuzzy"] = untpl(self.re["tpl_link_fuzzy"]) self.re["link_no_ip_fuzzy"] = untpl(self.re["tpl_link_no_ip_fuzzy"]) self.re["host_fuzzy_test"] = untpl(self.re["tpl_host_fuzzy_test"]) # # Compile each schema # aliases = [] self._compiled = {} for name, val in self._schemas.items(): # skip disabled methods if val is None: continue compiled = {"validate": None, "link": None} self._compiled[name] = compiled if isinstance(val, dict): if isinstance(val.get("validate"), RE_TYPE): compiled["validate"] = self._create_validator(val.get("validate")) elif isinstance(val.get("validate"), str): compiled["validate"] = self._create_validator(val.get("validate")) elif isinstance(val.get("validate"), types.MethodType): compiled["validate"] = val.get("validate") # Add custom handler elif isinstance(val.get("validate"), types.FunctionType): setattr(LinkifyIt, "func", val.get("validate")) compiled["validate"] = self.func else: raise SchemaError(name, val) if isinstance(val.get("normalize"), types.MethodType): compiled["normalize"] = val.get("normalize") # Add custom handler elif isinstance(val.get("normalize"), types.FunctionType): setattr(LinkifyIt, "func", val.get("normalize")) compiled["normalize"] = self.func elif not val.get("normalize"): compiled["normalize"] = self._create_normalizer() else: raise SchemaError(name, val) continue if isinstance(val, str): aliases.append(name) continue raise SchemaError(name, val) # # Compile postponed aliases # for alias in aliases: if not self._compiled.get(self._schemas.get(alias)): continue self._compiled[alias]["validate"] = self._compiled[self._schemas[alias]][ "validate" ] self._compiled[alias]["normalize"] = self._compiled[self._schemas[alias]][ "normalize" ] # Fake record for guessed links self._compiled[""] = {"validate": None, "normalize": self._create_normalizer()} # # Build schema condition # slist = "|".join( [ _escape_re(name) for name, val in self._compiled.items() if len(name) > 0 and val ] ) re_schema_test = ( "(^|(?!_)(?:[><\uff5c]|" + self.re["src_ZPCc"] + "))(" + slist + ")" ) # (?!_) cause 1.5x slowdown self.re["schema_test"] = re_schema_test self.re["schema_search"] = re_schema_test self.re["schema_at_start"] = "^" + self.re["schema_search"] self.re["pretest"] = ( "(" + re_schema_test + ")|(" + self.re["host_fuzzy_test"] + ")|@" ) # Cleanup self._reset_scan_cache()
[docs] def add(self, schema, definition): """Add new rule definition. (chainable) See :class:`linkify_it.main.LinkifyIt` init description for details. ``schema`` is a link prefix (``skype:``, for example), and ``definition`` is a ``str`` to alias to another schema, or an ``dict`` with ``validate`` and optionally `normalize` definitions. To disable an existing rule, use ``.add(<schema>, None)``. Args: schema (str): rule name (fixed pattern prefix) definition (`str` or `re.Pattern`): schema definition Return: :class:`linkify_it.main.LinkifyIt` """ self._schemas[schema] = definition self._compile() return self
[docs] def set(self, options): """Override default options. (chainable) Missed properties will not be changed. Args: options (dict): ``keys``: [``fuzzy_link`` | ``fuzzy_email`` | ``fuzzy_ip``]. ``values``: [``True`` | ``False``] Return: :class:`linkify_it.main.LinkifyIt` """ self._opts.update(options) return self
[docs] def test(self, text): """Searches linkifiable pattern and returns ``True`` on success or ``False`` on fail. Args: text (str): text to search Returns: bool: ``True`` if a linkable pattern was found, otherwise it is ``False``. """ self._text_cache = text self._index = -1 if not len(text): return False if re.search(self.re["schema_test"], text, flags=re.IGNORECASE): regex = self.re["schema_search"] last_index = 0 matched_iter = re.finditer(regex, text[last_index:], flags=re.IGNORECASE) for matched in matched_iter: last_index = matched.end(0) m = (matched.group(), matched.groups()[0], matched.groups()[1]) length = self.test_schema_at(text, m[2], last_index) if length: self._schema = m[2] self._index = matched.start(0) + len(m[1]) self._last_index = matched.start(0) + len(m[0]) + length break if self._opts.get("fuzzy_link") and self._compiled.get("http:"): # guess schemaless links matched_tld = re.search( self.re["host_fuzzy_test"], text, flags=re.IGNORECASE ) if matched_tld: tld_pos = matched_tld.start(0) else: tld_pos = -1 if tld_pos >= 0: # if tld is located after found link - no need to check fuzzy pattern if self._index < 0 or tld_pos < self._index: if self._opts.get("fuzzy_ip"): pattern = self.re["link_fuzzy"] else: pattern = self.re["link_no_ip_fuzzy"] ml = re.search(pattern, text, flags=re.IGNORECASE) if ml: shift = ml.start(0) + len(ml.groups()[0]) if self._index < 0 or shift < self._index: self._schema = "" self._index = shift self._last_index = ml.start(0) + len(ml.group()) if self._opts.get("fuzzy_email") and self._compiled.get("mailto:"): # guess schemaless emails at_pos = _index_of(text, "@") if at_pos >= 0: # We can't skip this check, because this cases are possible: # 192.168.1.1@gmail.com, my.in@example.com me = re.search(self.re["email_fuzzy"], text, flags=re.IGNORECASE) if me: shift = me.start(0) + len(me.groups()[0]) next_shift = me.start(0) + len(me.group()) if ( self._index < 0 or shift < self._index or (shift == self._index and next_shift > self._last_index) ): self._schema = "mailto:" self._index = shift self._last_index = next_shift return self._index >= 0
[docs] def pretest(self, text): """Very quick check, that can give false positives. Returns true if link MAY BE can exists. Can be used for speed optimization, when you need to check that link NOT exists. Args: text (str): text to search Returns: bool: ``True`` if a linkable pattern was found, otherwise it is ``False``. """ if re.search(self.re["pretest"], text, flags=re.IGNORECASE): return True return False
[docs] def test_schema_at(self, text, name, position): """Similar to :meth:`linkify_it.main.LinkifyIt.test` but checks only specific protocol tail exactly at given position. Args: text (str): text to scan name (str): rule (schema) name position (int): length of found pattern (0 on fail). Returns: int: text (str): text to search """ # If not supported schema check requested - terminate if not self._compiled.get(name.lower()): return 0 return self._compiled.get(name.lower()).get("validate")(text, position)
[docs] def match(self, text): """Returns ``list`` of found link descriptions or ``None`` on fail. We strongly recommend to use :meth:`linkify_it.main.LinkifyIt.test` first, for best speed. Args: text (str): text to search Returns: ``list`` or ``None``: Result match description: * **schema** - link schema, can be empty for fuzzy links, or ``//`` for protocol-neutral links. * **index** - offset of matched text * **last_index** - offset of matched text * **raw** - offset of matched text * **text** - normalized text * **url** - link, generated from matched text """ shift = 0 result = [] # try to take previous element from cache, if .test() called before if self._index >= 0 and self._text_cache == text: result.append(self._create_match(shift)) shift = self._last_index # Cut head if cache was used tail = text[shift:] if shift else text # Scan string until end reached while self.test(tail): result.append(self._create_match(shift)) tail = tail[self._last_index :] shift += self._last_index if len(result): return result return None
[docs] def match_at_start(self, text): """Returns fully-formed (not fuzzy) link if it starts at the beginning of the string, and null otherwise. Args: text (str): text to search Retuns: ``Match`` or ``None`` """ # Reset scan cache self._text_cache = text self._index = -1 if not len(text): return None founds = re.search(self.re["schema_at_start"], text, flags=re.IGNORECASE) if not founds: return None m = (founds.group(), founds.groups()[0], founds.groups()[1]) length = self.test_schema_at(text, m[2], len(m[0])) if not length: return None self._schema = m[2] self._index = founds.start(0) + len(m[1]) self._last_index = founds.start(0) + len(m[0]) + length return self._create_match(0)
[docs] def tlds(self, list_tlds, keep_old=False): """Load (or merge) new tlds list. (chainable) Those are user for fuzzy links (without prefix) to avoid false positives. By default this algorythm used: * hostname with any 2-letter root zones are ok. * biz|com|edu|gov|net|org|pro|web|xxx|aero|asia|coop|info|museum|name|shop|рф are ok. * encoded (`xn--...`) root zones are ok. If list is replaced, then exact match for 2-chars root zones will be checked. Args: list_tlds (list or str): ``list of tlds`` or ``tlds string`` keep_old (bool): merge with current list if q`True`q (q`Falseq` by default) """ _list = list_tlds if isinstance(list_tlds, list) else [list_tlds] if not keep_old: self._tlds = _list self._tlds_replaced = True self._compile() return self self._tlds.extend(_list) self._tlds = sorted(list(set(self._tlds)), reverse=True) self._compile() return self
[docs] def normalize(self, match): """Default normalizer (if schema does not define it's own). Args: match (:class:`linkify_it.main.Match`): Match result """ if not match.schema: match.url = "http://" + match.url if match.schema == "mailto:" and not re.search( "^mailto:", match.url, flags=re.IGNORECASE ): match.url = "mailto:" + match.url
def _on_compile(self): """Override to modify basic RegExp-s.""" pass