|
|
""" |
|
|
Source: DPR Implementation from Facebook Research |
|
|
https://github.com/facebookresearch/DPR/tree/master/dpr |
|
|
Original license: https://github.com/facebookresearch/DPR/blob/main/LICENSE |
|
|
""" |
|
|
|
|
|
import regex |
|
|
import unicodedata |
|
|
|
|
|
|
|
|
class Tokens(object): |
|
|
"""A class to represent a list of tokenized text.""" |
|
|
TEXT = 0 |
|
|
TEXT_WS = 1 |
|
|
SPAN = 2 |
|
|
POS = 3 |
|
|
LEMMA = 4 |
|
|
NER = 5 |
|
|
|
|
|
def __init__(self, data, annotators, opts=None): |
|
|
self.data = data |
|
|
self.annotators = annotators |
|
|
self.opts = opts or {} |
|
|
|
|
|
def __len__(self): |
|
|
"""The number of tokens.""" |
|
|
return len(self.data) |
|
|
|
|
|
def slice(self, i=None, j=None): |
|
|
"""Return a view of the list of tokens from [i, j).""" |
|
|
new_tokens = copy.copy(self) |
|
|
new_tokens.data = self.data[i: j] |
|
|
return new_tokens |
|
|
|
|
|
def untokenize(self): |
|
|
"""Returns the original text (with whitespace reinserted).""" |
|
|
return ''.join([t[self.TEXT_WS] for t in self.data]).strip() |
|
|
|
|
|
def words(self, uncased=False): |
|
|
"""Returns a list of the text of each token |
|
|
|
|
|
Args: |
|
|
uncased: lower cases text |
|
|
""" |
|
|
if uncased: |
|
|
return [t[self.TEXT].lower() for t in self.data] |
|
|
else: |
|
|
return [t[self.TEXT] for t in self.data] |
|
|
|
|
|
def offsets(self): |
|
|
"""Returns a list of [start, end) character offsets of each token.""" |
|
|
return [t[self.SPAN] for t in self.data] |
|
|
|
|
|
def pos(self): |
|
|
"""Returns a list of part-of-speech tags of each token. |
|
|
Returns None if this annotation was not included. |
|
|
""" |
|
|
if 'pos' not in self.annotators: |
|
|
return None |
|
|
return [t[self.POS] for t in self.data] |
|
|
|
|
|
def lemmas(self): |
|
|
"""Returns a list of the lemmatized text of each token. |
|
|
Returns None if this annotation was not included. |
|
|
""" |
|
|
if 'lemma' not in self.annotators: |
|
|
return None |
|
|
return [t[self.LEMMA] for t in self.data] |
|
|
|
|
|
def entities(self): |
|
|
"""Returns a list of named-entity-recognition tags of each token. |
|
|
Returns None if this annotation was not included. |
|
|
""" |
|
|
if 'ner' not in self.annotators: |
|
|
return None |
|
|
return [t[self.NER] for t in self.data] |
|
|
|
|
|
def ngrams(self, n=1, uncased=False, filter_fn=None, as_strings=True): |
|
|
"""Returns a list of all ngrams from length 1 to n. |
|
|
|
|
|
Args: |
|
|
n: upper limit of ngram length |
|
|
uncased: lower cases text |
|
|
filter_fn: user function that takes in an ngram list and returns |
|
|
True or False to keep or not keep the ngram |
|
|
as_string: return the ngram as a string vs list |
|
|
""" |
|
|
|
|
|
def _skip(gram): |
|
|
if not filter_fn: |
|
|
return False |
|
|
return filter_fn(gram) |
|
|
|
|
|
words = self.words(uncased) |
|
|
ngrams = [(s, e + 1) |
|
|
for s in range(len(words)) |
|
|
for e in range(s, min(s + n, len(words))) |
|
|
if not _skip(words[s:e + 1])] |
|
|
|
|
|
|
|
|
if as_strings: |
|
|
ngrams = ['{}'.format(' '.join(words[s:e])) for (s, e) in ngrams] |
|
|
|
|
|
return ngrams |
|
|
|
|
|
def entity_groups(self): |
|
|
"""Group consecutive entity tokens with the same NER tag.""" |
|
|
entities = self.entities() |
|
|
if not entities: |
|
|
return None |
|
|
non_ent = self.opts.get('non_ent', 'O') |
|
|
groups = [] |
|
|
idx = 0 |
|
|
while idx < len(entities): |
|
|
ner_tag = entities[idx] |
|
|
|
|
|
if ner_tag != non_ent: |
|
|
|
|
|
start = idx |
|
|
while (idx < len(entities) and entities[idx] == ner_tag): |
|
|
idx += 1 |
|
|
groups.append((self.slice(start, idx).untokenize(), ner_tag)) |
|
|
else: |
|
|
idx += 1 |
|
|
return groups |
|
|
|
|
|
|
|
|
class Tokenizer(object): |
|
|
"""Base tokenizer class. |
|
|
Tokenizers implement tokenize, which should return a Tokens class. |
|
|
""" |
|
|
|
|
|
def tokenize(self, text): |
|
|
raise NotImplementedError |
|
|
|
|
|
def shutdown(self): |
|
|
pass |
|
|
|
|
|
def __del__(self): |
|
|
self.shutdown() |
|
|
|
|
|
|
|
|
class SimpleTokenizer(Tokenizer): |
|
|
ALPHA_NUM = r'[\p{L}\p{N}\p{M}]+' |
|
|
NON_WS = r'[^\p{Z}\p{C}]' |
|
|
|
|
|
def __init__(self, **kwargs): |
|
|
""" |
|
|
Args: |
|
|
annotators: None or empty set (only tokenizes). |
|
|
""" |
|
|
self._regexp = regex.compile( |
|
|
'(%s)|(%s)' % (self.ALPHA_NUM, self.NON_WS), |
|
|
flags=regex.IGNORECASE + regex.UNICODE + regex.MULTILINE |
|
|
) |
|
|
if len(kwargs.get('annotators', {})) > 0: |
|
|
logger.warning('%s only tokenizes! Skipping annotators: %s' % |
|
|
(type(self).__name__, kwargs.get('annotators'))) |
|
|
self.annotators = set() |
|
|
|
|
|
def tokenize(self, text): |
|
|
data = [] |
|
|
matches = [m for m in self._regexp.finditer(text)] |
|
|
for i in range(len(matches)): |
|
|
|
|
|
token = matches[i].group() |
|
|
|
|
|
|
|
|
span = matches[i].span() |
|
|
start_ws = span[0] |
|
|
if i + 1 < len(matches): |
|
|
end_ws = matches[i + 1].span()[0] |
|
|
else: |
|
|
end_ws = span[1] |
|
|
|
|
|
|
|
|
data.append(( |
|
|
token, |
|
|
text[start_ws: end_ws], |
|
|
span, |
|
|
)) |
|
|
return Tokens(data, self.annotators) |
|
|
|
|
|
|
|
|
def has_answer(tokenized_answers, text): |
|
|
text = DPR_normalize(text) |
|
|
|
|
|
for single_answer in tokenized_answers: |
|
|
for i in range(0, len(text) - len(single_answer) + 1): |
|
|
if single_answer == text[i: i + len(single_answer)]: |
|
|
return True |
|
|
|
|
|
return False |
|
|
|
|
|
|
|
|
def locate_answers(tokenized_answers, text): |
|
|
""" |
|
|
Returns each occurrence of an answer as (offset, endpos) in terms of *characters*. |
|
|
""" |
|
|
tokenized_text = DPR_tokenize(text) |
|
|
occurrences = [] |
|
|
|
|
|
text_words, text_word_positions = tokenized_text.words(uncased=True), tokenized_text.offsets() |
|
|
answers_words = [ans.words(uncased=True) for ans in tokenized_answers] |
|
|
|
|
|
for single_answer in answers_words: |
|
|
for i in range(0, len(text_words) - len(single_answer) + 1): |
|
|
if single_answer == text_words[i: i + len(single_answer)]: |
|
|
(offset, _), (_, endpos) = text_word_positions[i], text_word_positions[i+len(single_answer)-1] |
|
|
occurrences.append((offset, endpos)) |
|
|
|
|
|
return occurrences |
|
|
|
|
|
|
|
|
STokenizer = SimpleTokenizer() |
|
|
|
|
|
|
|
|
def DPR_tokenize(text): |
|
|
return STokenizer.tokenize(unicodedata.normalize('NFD', text)) |
|
|
|
|
|
|
|
|
def DPR_normalize(text): |
|
|
return DPR_tokenize(text).words(uncased=True) |
|
|
|
|
|
|
|
|
|
|
|
def strip_accents(text): |
|
|
"""Strips accents from a piece of text.""" |
|
|
text = unicodedata.normalize("NFD", text) |
|
|
output = [] |
|
|
for char in text: |
|
|
cat = unicodedata.category(char) |
|
|
if cat == "Mn": |
|
|
continue |
|
|
output.append(char) |
|
|
return "".join(output) |
|
|
|
