| """ |
| Source: DPR Implementation from Facebook Research |
| https://github.com/facebookresearch/DPR/tree/master/dpr |
| """ |
|
|
| import string |
| import spacy |
| 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) |
|
|
