import spacy import string from HTMLParser import HTMLParser from re import compile as _Re from CommonServerPython import * import sys reload(sys) sys.setdefaultencoding('utf-8') # pylint: disable=no-member MAX_TEXT_LENGTH = 10 ** 5 NUMBER_PATTERN = "NUMBER_PATTERN" URL_PATTERN = "URL_PATTERN" EMAIL_PATTERN = "EMAIL_PATTERN" reserved_tokens = set([NUMBER_PATTERN, URL_PATTERN, EMAIL_PATTERN]) CLEAN_HTML = (demisto.args()['cleanHtml'] == 'yes') REMOVE_LINE_BREAKS = (demisto.args()['removeLineBreaks'] == 'yes') TEXT_ENCODE = demisto.args()['encoding'] HASH_SEED = demisto.args().get('hashWordWithSeed') FILTER_ENGLISH_WORDS = demisto.args()['removeNonEnglishWords'] == 'yes' REMOVE_STOP_WORDS = demisto.args()['removeStopWords'] == 'yes' REMOVE_PUNCT = demisto.args()['removePunctuation'] == 'yes' REMOVE_NON_ALPHA = demisto.args()['removeNonAlphaWords'] == 'yes' REPLACE_EMAIL = demisto.args()['replaceEmails'] == 'yes' REPLACE_URLS = demisto.args()['replaceUrls'] == 'yes' REPLACE_NUMBERS = demisto.args()['replaceNumbers'] == 'yes' LEMMATIZER = demisto.args()['useLemmatization'] == 'yes' VALUE_IS_JSON = demisto.args()['isValueJson'] == 'yes' HTML_PATTERNS = [ re.compile(r"(?is)<(script|style).*?>.*?()"), re.compile(r"(?s)[\n]?"), re.compile(r"(?s)<.*?>"), re.compile(r" "), re.compile(r" +") ] # define global parsers html_parser = HTMLParser() LANGUAGES_TO_MODEL_NAMES = {'English': 'en_core_web_sm', 'German': 'de_core_news_sm', 'French': 'fr_core_news_sm', 'Spanish': 'es_core_news_sm', 'Portuguese': 'pt_core_news_sm', 'Italian': 'it_core_news_sm', 'Dutch': 'nl_core_news_sm' } _unicode_chr_splitter = _Re('(?s)((?:[\ud800-\udbff][\udc00-\udfff])|.)').split nlp = None def clean_html(text): if not CLEAN_HTML: return text cleaned = text for pattern in HTML_PATTERNS: cleaned = pattern.sub(" ", cleaned) return html_parser.unescape(cleaned).strip() def remove_line_breaks(text): if not REMOVE_LINE_BREAKS: return text return text.replace("\r", " ").replace("\n", " ") def remove_multiple_whitespaces(text): return re.sub(r"\s+", " ", text).strip() def hash_word(word): return str(hash_djb2(word, int(HASH_SEED))) def tokenize_text(text): try: unicode_text = unicode(text) except Exception: unicode_text = text language = demisto.args()['language'] if language in LANGUAGES_TO_MODEL_NAMES: original_words_to_tokens, tokens_list = tokenize_text_spacy(unicode_text, language) else: original_words_to_tokens, tokens_list = tokenize_text_other(unicode_text) hashed_tokens_list = [] if HASH_SEED: for word in tokens_list: word_hashed = hash_word(word) hashed_tokens_list.append(word_hashed) hashed_words_to_tokens = {word: [hash_word(t) for t in tokens_list] for word, tokens_list in original_words_to_tokens.items()} else: hashed_words_to_tokens = {} return ' '.join(tokens_list).strip(), ' '.join(hashed_tokens_list) if len( hashed_tokens_list) > 0 else None, original_words_to_tokens, hashed_words_to_tokens def tokenize_text_other(unicode_text): tokens_list = [] tokenization_method = demisto.args()['tokenizationMethod'] if tokenization_method == 'byWords': original_words_to_tokens = {} for t in unicode_text.split(): token_without_punct = ''.join([c for c in t if c not in string.punctuation]) if len(token_without_punct) > 0: tokens_list.append(token_without_punct) original_words_to_tokens[token_without_punct] = t elif tokenization_method == 'byLetters': for t in unicode_text: tokens_list += [chr for chr in _unicode_chr_splitter(t) if chr] original_words_to_tokens = {c: t for c in tokens_list} else: return_error('Unsupported tokenization method: when language is "Other" ({})'.format(tokenization_method)) return original_words_to_tokens, tokens_list def tokenize_text_spacy(unicode_text, language): global nlp if nlp is None: nlp = spacy.load(LANGUAGES_TO_MODEL_NAMES[language], disable=['tagger', 'parser', 'ner', 'textcat']) doc = nlp(unicode(unicode_text)) original_text_indices_to_words = map_indices_to_words(unicode_text) tokens_list = [] original_words_to_tokens = {} # type: ignore for word in doc: if word.is_space: continue elif REMOVE_STOP_WORDS and word.is_stop: continue elif REMOVE_PUNCT and word.is_punct: continue elif REPLACE_EMAIL and '@' in word.text: tokens_list.append(EMAIL_PATTERN) elif REPLACE_URLS and word.like_url: tokens_list.append(URL_PATTERN) elif REPLACE_NUMBERS and (word.like_num or word.pos_ == 'NUM'): tokens_list.append(NUMBER_PATTERN) elif REMOVE_NON_ALPHA and not word.is_alpha: continue elif FILTER_ENGLISH_WORDS and word.text not in nlp.vocab: continue else: if LEMMATIZER and word.lemma_ != '-PRON-': token_to_add = word.lemma_ else: token_to_add = word.lower_ tokens_list.append(token_to_add) original_word = original_text_indices_to_words[word.idx] if original_word not in original_words_to_tokens: original_words_to_tokens[original_word] = [] original_words_to_tokens[original_word].append(token_to_add) return original_words_to_tokens, tokens_list def map_indices_to_words(unicode_text): original_text_indices_to_words = {} word_start = 0 while word_start < len(unicode_text) and unicode_text[word_start].isspace(): word_start += 1 for word in unicode_text.split(): for char_idx, char in enumerate(word): original_text_indices_to_words[word_start + char_idx] = word # find beginning of next word word_start += len(word) while word_start < len(unicode_text) and unicode_text[word_start].isspace(): word_start += 1 return original_text_indices_to_words def handle_long_text(t, input_length): if input_length == 1: return_error("Input text length ({}) exceeds the legal maximum length for preprocessing".format(len(t))) else: return '', '', {}, {} def word_tokenize(text): if VALUE_IS_JSON: try: text = json.loads(text) except Exception: pass if not isinstance(text, list): text = [text] result = [] for t in text: original_text = t t = remove_line_breaks(t) t = clean_html(t) t = remove_multiple_whitespaces(t) if len(t) < MAX_TEXT_LENGTH: tokenized_text, hash_tokenized_text, original_words_to_tokens, words_to_hashed_tokens = tokenize_text(t) else: tokenized_text, hash_tokenized_text, original_words_to_tokens, words_to_hashed_tokens =\ handle_long_text(t, input_length=len(text)) text_result = { 'originalText': original_text, 'tokenizedText': tokenized_text, 'originalWordsToTokens': original_words_to_tokens, } if hash_tokenized_text: text_result['hashedTokenizedText'] = hash_tokenized_text text_result['wordsToHashedTokens'] = words_to_hashed_tokens result.append(text_result) if len(result) == 1: result = result[0] # type: ignore return { 'Type': entryTypes['note'], 'Contents': result, 'ContentsFormat': formats['json'], 'HumanReadable': tableToMarkdown('Tokenized Text', result, headers=['tokenizedText']), 'HumanReadableFormat': formats['markdown'], 'EntryContext': { 'WordTokenizeNLPOutput': result } } if __name__ in ['__main__', '__builtin__', 'builtins']: demisto.results(word_tokenize(demisto.args()['value']))