tokenize.py

# -*- coding: utf-8 -*-
#
# Copyright (c) nexB Inc. and others. All rights reserved.
# ScanCode is a trademark of nexB Inc.
# SPDX-License-Identifier: Apache-2.0
# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
# See https://github.com/nexB/scancode-toolkit for support or download.
# See https://aboutcode.org for more information about nexB OSS projects.
#

import re

from binascii import crc32
from collections import defaultdict
from itertools import islice

from licensedcode.spans import Span
from licensedcode.stopwords import STOPWORDS
from textcode.analysis import numbered_text_lines

"""
Utilities to break texts in lines and tokens (aka. words),
and handle required phrases in texts through these tokens,
with specialized version for queries and rules texts.
"""


def query_lines(
    location=None,
    query_string=None,
    strip=True,
    start_line=1,
    plain_text=False,
):
    """
    Return an iterable of tuples (line number, text line) given a file at
    `location` or a `query string`. Include empty lines.
    Line numbers start at ``start_line`` which is 1-based by default.

    If `plain_text` is True treat the file as a plain text file and do not
    attempt to detect its type and extract its content with special procedures.
    This is used mostly when loading license texts and rules.
    """
    # TODO: OPTIMIZE: tokenizing line by line may be rather slow
    # we could instead get lines and tokens at once in a batch?
    numbered_lines = []
    if location:
        numbered_lines = numbered_text_lines(
            location,
            demarkup=False,
            start_line=start_line,
            plain_text=plain_text,
        )

    elif query_string:
        if strip:
            keepends = False
        else:
            keepends = True

        numbered_lines = enumerate(
            query_string.splitlines(keepends),
            start_line,
        )

    for line_number, line in numbered_lines:
        if strip:
            yield line_number, line.strip()
        else:
            yield line_number, line.rstrip('\n') + '\n'

# Split on whitespace and punctuations: keep only characters and numbers and +
# when in the middle or end of a word. Keeping the trailing + is important for
# licenses name such as GPL2+. The use a double negation "not non word" meaning
# "words" to define the character ranges


query_pattern = '[^_\\W]+\\+?[^_\\W]*'
word_splitter = re.compile(query_pattern, re.UNICODE).findall

required_phrase_pattern = '(?:' + query_pattern + '|\\{\\{|\\}\\})'
required_phrase_splitter = re.compile(required_phrase_pattern, re.UNICODE).findall


extra_phrase_pattern = '(?:' + query_pattern + r'|\[\[|\]\])'
extra_phrase_splitter = re.compile(extra_phrase_pattern, re.UNICODE).findall


# pattern to match and remove extra phrases like [[1]], [[4]]..etc from the text
extra_phrase_removal_pattern = re.compile(r'\[\[\d+\]\]')

REQUIRED_PHRASE_OPEN = '{{'
REQUIRED_PHRASE_CLOSE = '}}'

EXTRA_PHRASE_OPEN ='[['
EXTRA_PHRASE_CLOSE =']]'

# FIXME: this should be folded in a single pass tokenization with the index_tokenizer


def extra_phrase_tokenizer(text, stopwords=STOPWORDS, preserve_case=False):
    """
    Yield tokens from a rule ``text`` including extra phrases [[n]] markers.
    This n denotes maximum number of extra-words i.e valide at that position.
    This is same as ``required_phrase_tokenizer``.
    """
    if not text:
        return
    if not preserve_case:
        text = text.lower()

    for token in extra_phrase_splitter(text):
        if token and token not in stopwords:
            yield token


def get_extra_phrase_spans(text):
    """
    Return a list of tuples `(Span, int)`, one for each [[n]] extra phrase found in ``text``.
    Here, `n` should always be a digit token inside the extra phrase brackets.

    Example:
    >>> text = 'Neither the name [[3]] of nor the names of its'
    >>> #          0    1    2     3   4  5    6   7    8   9
    >>> x = get_extra_phrase_spans(text)
    >>> assert x == [(Span([3]), 3)], x
    """
    ipos = 0
    in_extra_phrase = False
    current_phrase_value = []
    extra_phrase_spans = []

    for token in extra_phrase_tokenizer(text):
        if token == EXTRA_PHRASE_OPEN:
            in_extra_phrase = True
            current_phrase_value = []
            continue

        elif token == EXTRA_PHRASE_CLOSE:
            if in_extra_phrase:
                # token must be digit and token must be present in double square bracket ``[[token]]``
                # and between extra phrases there must only one token exist
                if len(current_phrase_value) == 1 and current_phrase_value[0].isdigit():
                    extra_phrase_spans.append((Span([ipos - 1]), int(current_phrase_value[0])))

            in_extra_phrase = False
            current_phrase_value = []
            continue

        if in_extra_phrase:
            # consider one token after double open square bracket ``[[``
            if len(current_phrase_value) == 0:
                current_phrase_value.append(token)

        ipos += 1

    return extra_phrase_spans   


def required_phrase_tokenizer(text, stopwords=STOPWORDS, preserve_case=False):
    """
    Yield tokens from a rule ``text`` including required phrases {{brace}} markers.
    This tokenizer behaves the same as as the ``index_tokenizer`` returning also
    REQUIRED_PHRASE_OPEN and REQUIRED_PHRASE_CLOSE as separate tokens so that they can be
    used to parse required phrases.

    >>> x = list(required_phrase_splitter('{{AGPL-3.0  GNU Affero License v3.0}}'))
    >>> assert x == ['{{', 'AGPL', '3', '0', 'GNU', 'Affero', 'License', 'v3', '0', '}}'], x

    >>> x = list(required_phrase_splitter('{{{AGPL{{{{Affero }}License}}0}}'))
    >>> assert x == ['{{', 'AGPL', '{{', '{{', 'Affero', '}}', 'License', '}}', '0', '}}'], x

    >>> list(index_tokenizer('')) == []
    True

    >>> x = list(index_tokenizer('{{AGPL-3.0  GNU Affero License v3.0}}'))
    >>> assert x == ['agpl', '3', '0', 'gnu', 'affero', 'license', 'v3', '0']

    >>> x = list(required_phrase_tokenizer('{{AGPL-3.0  GNU Affero License v3.0}}'))
    >>> assert x == ['{{', 'agpl', '3', '0', 'gnu', 'affero', 'license', 'v3', '0', '}}']
    """
    if not text:
        return
    if not preserve_case:
        text = text.lower()

    for token in required_phrase_splitter(text):
        if token and token not in stopwords:
            yield token


def get_existing_required_phrase_spans(text):
    """
    Return a list of token position Spans, one for each {{tagged}} required phrase found in  ``text``.

    For example:

    >>> text = 'This is enclosed in {{double curly braces}}'
    >>> #       0    1  2        3    4      5     6
    >>> x = get_existing_required_phrase_spans(text)
    >>> assert x == [Span(4, 6)], x

    >>> text = 'This is {{enclosed}} a  {{double curly braces}} or not'
    >>> #       0    1    2          SW   3      4     5        6  7
    >>> x = get_existing_required_phrase_spans(text)
    >>> assert x == [Span(2), Span(3, 5)], x

    >>> text = 'This {{is}} enclosed a  {{double curly braces}} or not'
    >>> #       0    1      2        SW   3      4     5        6  7
    >>> x = get_existing_required_phrase_spans(text)
    >>> assert x == [Span([1]), Span([3, 4, 5])], x

    >>> text = '{{AGPL-3.0  GNU Affero General Public License v3.0}}'
    >>> #         0    1 2  3   4      5       6      7       8  9
    >>> x = get_existing_required_phrase_spans(text)
    >>> assert x == [Span(0, 9)], x

    >>> assert get_existing_required_phrase_spans('{This}') == []

    >>> def check_exception(text):
    ...     try:
    ...         return get_existing_required_phrase_spans(text)
    ...     except InvalidRuleRequiredPhrase:
    ...         pass

    >>> check_exception('This {{is')
    >>> check_exception('This }}is')
    >>> check_exception('{{This }}is{{')
    >>> check_exception('This }}is{{')
    >>> check_exception('{{}}')
    >>> check_exception('{{This is')
    >>> check_exception('{{This is{{')
    >>> check_exception('{{This is{{ }}')
    >>> check_exception('{{{{This}}}}')
    >>> check_exception('}}This {{is}}')
    >>> check_exception('This }} {{is}}')
    >>> check_exception('{{This}}')
    [Span(0)]
    >>> check_exception('{This}')
    []
    >>> check_exception('{{{This}}}')
    [Span(0)]
    """
    return list(get_phrase_spans(text))


class InvalidRuleRequiredPhrase(Exception):
    pass



def get_phrase_spans(text):
    """
    Yield position Spans for each tagged required phrase found in ``text``.
    """
    ipos = 0
    in_required_phrase = False
    current_phrase_positions = []
    for token in required_phrase_tokenizer(text):
        if token == REQUIRED_PHRASE_OPEN:
            if in_required_phrase:
                raise InvalidRuleRequiredPhrase('Invalid rule with nested required phrase {{ {{ braces', text)
            in_required_phrase = True

        elif token == REQUIRED_PHRASE_CLOSE:
            if in_required_phrase:
                if current_phrase_positions:
                    yield Span(current_phrase_positions)
                    current_phrase_positions = []
                else:
                    raise InvalidRuleRequiredPhrase('Invalid rule with empty required phrase {{}} braces', text)
                in_required_phrase = False
            else:
                raise InvalidRuleRequiredPhrase(f'Invalid rule with dangling required phrase missing closing braces', text)
            continue
        else:
            if in_required_phrase:
                current_phrase_positions.append(ipos)
            ipos += 1

    if current_phrase_positions or in_required_phrase:
        raise InvalidRuleRequiredPhrase(f'Invalid rule with dangling required phrase missing final closing braces', text)




def index_tokenizer(text, stopwords=STOPWORDS, preserve_case=False):
    """
    Return an iterable of tokens from a rule or query ``text`` using index
    tokenizing rules. Ignore words that exist as lowercase in the ``stopwords``
    set.

    For example::
    >>> list(index_tokenizer(''))
    []
    >>> x = list(index_tokenizer('some Text with   spAces! + _ -'))
    >>> assert x == ['some', 'text', 'with', 'spaces']

    >>> x = list(index_tokenizer('{{}some }}Text with   spAces! + _ -'))
    >>> assert x == ['some', 'text', 'with', 'spaces']

    >>> x = list(index_tokenizer('{{Hi}}some {{}}Text with{{noth+-_!@ing}}   {{junk}}spAces! + _ -{{}}'))
    >>> assert x == ['hi', 'some', 'text', 'with', 'noth+', 'ing', 'junk', 'spaces']

    >>> stops = set(['quot', 'lt', 'gt'])
    >>> x = list(index_tokenizer('some &quot&lt markup &gt&quot', stopwords=stops))
    >>> assert x == ['some', 'markup']
    """
    if not text:
        return []
    if not preserve_case:
        text = text.lower()
    words = word_splitter(text)
    return (token for token in words if token and token not in stopwords)


def index_tokenizer_with_stopwords(text, stopwords=STOPWORDS):
    """
    Return a tuple of (tokens, stopwords_by_pos) for a rule
    ``text`` using index tokenizing rules where tokens is a list of tokens and
    stopwords_by_pos is a mapping of {pos: stops count} where "pos" is a token
    position and "stops count" is the number of stopword tokens after this
    position if any. For stopwords at the start, the position is using the magic
    -1 key. Use the lowercase ``stopwords`` set.

    For example::

    >>> toks, stops = index_tokenizer_with_stopwords('')
    >>> assert toks == [], (toks, stops)
    >>> assert stops == {}

    >>> toks, stops = index_tokenizer_with_stopwords('some Text with   spAces! + _ -')
    >>> assert toks == ['some', 'text', 'with', 'spaces'], (toks, stops)
    >>> assert stops == {}

    >>> toks, stops = index_tokenizer_with_stopwords('{{}some }}Text with   spAces! + _ -')
    >>> assert toks == ['some', 'text', 'with', 'spaces'], (toks, stops)
    >>> assert stops == {}

    >>> toks, stops = index_tokenizer_with_stopwords('{{Hi}}some {{}}Text with{{noth+-_!@ing}}   {{junk}}spAces! + _ -{{}}')
    >>> assert toks == ['hi', 'some', 'text', 'with', 'noth+', 'ing', 'junk', 'spaces'], (toks, stops)
    >>> assert stops == {}

    >>> stops = set(['quot', 'lt', 'gt'])
    >>> toks, stops = index_tokenizer_with_stopwords('some &quot&lt markup &gt&quot', stopwords=stops)
    >>> assert toks == ['some', 'markup'], (toks, stops)
    >>> assert stops == {0: 2, 1: 2}

    >>> toks, stops = index_tokenizer_with_stopwords('{{g', stopwords=stops)
    >>> assert toks == ['g'], (toks, stops)
    >>> assert stops == {}
    """
    if not text:
        return [], {}
    
    text = extra_phrase_removal_pattern.sub('', text)

    tokens = []
    tokens_append = tokens.append

    # we use a defaultdict as a convenience at construction time
    # TODO: use the actual words and not just a count
    stopwords_by_pos = defaultdict(int)

    pos = -1

    for token in word_splitter(text.lower()):
        if token:
            if token in stopwords:
                # If we have not yet started, then all tokens seen so far
                # are stopwords and we keep a count of them in the magic
                # "-1" position.
                stopwords_by_pos[pos] += 1
            else:
                pos += 1
                tokens_append(token)

    return tokens, dict(stopwords_by_pos)


def query_tokenizer(text):
    """
    Return an iterable of tokens from a unicode query text. Do not ignore stop
    words. They are handled at a later stage in a query.

    For example::
    >>> list(query_tokenizer(''))
    []
    >>> x = list(query_tokenizer('some Text with   spAces! + _ -'))
    >>> assert x == ['some', 'text', 'with', 'spaces']

    >>> x = list(query_tokenizer('{{}some }}Text with   spAces! + _ -'))
    >>> assert x == ['some', 'text', 'with', 'spaces']

    >>> x = list(query_tokenizer('{{Hi}}some {{}}Text with{{noth+-_!@ing}}   {{junk}}spAces! + _ -{{}}'))
    >>> assert x == ['hi', 'some', 'text', 'with', 'noth+', 'ing', 'junk', 'spaces']
    """
    if not text:
        return []
    words = word_splitter(text.lower())
    return (token for token in words if token)


# Alternate pattern which is the opposite of query_pattern used for
# matched text collection
not_query_pattern = '[_\\W\\s\\+]+[_\\W\\s]?'

# collect tokens and non-token texts in two different groups
_text_capture_pattern = (
    '(?P<token>' +
        query_pattern +
    ')' +
    '|' +
    '(?P<punct>' +
        not_query_pattern +
    ')'
)
tokens_and_non_tokens = re.compile(_text_capture_pattern, re.UNICODE).finditer


def matched_query_text_tokenizer(text):
    """
    Return an iterable of tokens and non-tokens punctuation from a unicode query
    text keeping everything (including punctuations, line endings, etc.)
    The returned iterable contains 2-tuples of:
    - True if the string is a text token or False if this is not
      (such as punctuation, spaces, etc).
    - the corresponding string.
    This is used to reconstruct the matched query text for reporting.
    """
    if not text:
        return
    for match in tokens_and_non_tokens(text):
        if match:
            mgd = match.groupdict()
            token = mgd.get('token')
            punct = mgd.get('punct')
            if token:
                yield True, token
            elif punct:
                yield False, punct
            else:
                # this should never happen
                raise Exception('Internal error in matched_query_text_tokenizer')


def ngrams(iterable, ngram_length):
    """
    Return an iterable of ngrams of length `ngram_length` given an `iterable`.
    Each ngram is a tuple of `ngram_length` items.

    The returned iterable is empty if the input iterable contains less than
    `ngram_length` items.

    Note: this is a fairly arcane but optimized way to compute ngrams.

    For example:
    >>> list(ngrams([1,2,3,4,5], 2))
    [(1, 2), (2, 3), (3, 4), (4, 5)]

    >>> list(ngrams([1,2,3,4,5], 4))
    [(1, 2, 3, 4), (2, 3, 4, 5)]

    >>> list(ngrams([1,2,3,4], 2))
    [(1, 2), (2, 3), (3, 4)]

    >>> list(ngrams([1,2,3], 2))
    [(1, 2), (2, 3)]

    >>> list(ngrams([1,2], 2))
    [(1, 2)]

    >>> list(ngrams([1], 2))
    []

    This also works with arrays or tuples:

    >>> from array import array
    >>> list(ngrams(array('h', [1,2,3,4,5]), 2))
    [(1, 2), (2, 3), (3, 4), (4, 5)]

    >>> list(ngrams(tuple([1,2,3,4,5]), 2))
    [(1, 2), (2, 3), (3, 4), (4, 5)]
    """
    return zip(*(islice(iterable, i, None) for i in range(ngram_length)))


def select_ngrams(ngrams, with_pos=False):
    """
    Return an iterable as a subset of a sequence of ngrams using the hailstorm
    algorithm. If `with_pos` is True also include the starting position for the
    ngram in the original sequence.

    Definition from the paper: http://www2009.eprints.org/7/1/p61.pdf

      The algorithm first fingerprints every token and then selects a shingle s
      if the minimum fingerprint value of all k tokens in s occurs at the first
      or the last position of s (and potentially also in between). Due to the
      probabilistic properties of Rabin fingerprints the probability that a
      shingle is chosen is 2/k if all tokens in the shingle are different.

    For example:
    >>> list(select_ngrams([(2, 1, 3), (1, 1, 3), (5, 1, 3), (2, 6, 1), (7, 3, 4)]))
    [(2, 1, 3), (1, 1, 3), (5, 1, 3), (2, 6, 1), (7, 3, 4)]

    Positions can also be included. In this case, tuple of (pos, ngram) are returned:
    >>> list(select_ngrams([(2, 1, 3), (1, 1, 3), (5, 1, 3), (2, 6, 1), (7, 3, 4)], with_pos=True))
    [(0, (2, 1, 3)), (1, (1, 1, 3)), (2, (5, 1, 3)), (3, (2, 6, 1)), (4, (7, 3, 4))]

    This works also from a generator:
    >>> list(select_ngrams(x for x in [(2, 1, 3), (1, 1, 3), (5, 1, 3), (2, 6, 1), (7, 3, 4)]))
    [(2, 1, 3), (1, 1, 3), (5, 1, 3), (2, 6, 1), (7, 3, 4)]
    """
    ngram = None
    last = None
    for pos, ngram in enumerate(ngrams):
        # FIXME: use a proper hash
        nghs = []
        for ng in ngram:
            if isinstance(ng, str):
                ng = bytearray(ng, encoding='utf-8')
            else:
                ng = bytearray(str(ng).encode('utf-8'))
            nghs.append(crc32(ng) & 0xffffffff)
        min_hash = min(nghs)
        if with_pos:
            ngram = (pos, ngram,)
        if min_hash in (nghs[0], nghs[-1]):
            yield ngram
            last = ngram
        else:
            # always yield the first or last ngram too.
            if pos == 0:
                yield ngram
                last = ngram
    if last != ngram:
        yield ngram