Jean-Baptiste/email_parser
20
1import logging2import os3import regex4from transformers import AutoModelForTokenClassification, AutoTokenizer, pipeline5import pandas as pd6import numpy as np7 8from . import utils, _models_signatures9from .utils import timing10from langid.langid import LanguageIdentifier11from langid.langid import model as model_langid12 13# Creating language_identifier object for usage in function f_detect_language14language_identifier = LanguageIdentifier.from_modelstring(model_langid, norm_probs=True)15language_identifier.set_languages(['en', 'fr'])16 17 18logging.info(f"Reading config file from folder:{os.path.join(os.path.dirname(__file__))}")19 20config = utils.f_read_config(os.path.join(os.path.dirname(__file__), 'config.ini'))21 22device = int(config["DEFAULT"]["device"])23default_lang = config["DEFAULT"]["default_lang"]24 25tokenizer_dict = {}26models_dict = {}27nlp_dict = {}28 29 30dict_regex_pattern = dict(EMAIL=r'[\p{L}\p{M}\-\d._]{1,}@[\p{L}\p{M}\d\-_]{1,}(\.[\p{L}\p{M}]{1,}){1,}',31 TEL=r'(?<!\d)(\+?\d{1,2}[ -]?)?\(?\d{3}\)?[ .-]?\d{3}[ .-]?\d{4}(?!\d|\p{P}\d)',32 POST=r'\b([A-z][0-9][A-z][ -]?[0-9][A-z][0-9]|[A-z][0-9][A-z])\b',33 PRICE=r"(([\s:,]|^){1}\$*(CA|CAD|USD|EUR|GBP|\$|\€|\£|\¢){1}\$*[\d., ]*[\d]{1,}\b)" +34 "|([\d]{1,}[\d., ]*(CA|CAD|USD|EUR|GBP|\$|\€|\£|k|m|\¢){1,}\$*(?=\s|\p{P}|$))",35 WEB=r"((www(\.[\p{L}\p{M}\-0-9]]{1,}){2,})" +36 "|(https?:[^ ]*)"+37 # r"|(([\p{L}\p{M}\.]{3,}){2,})|"38 r"|((?<=[\s:]|^)([\p{L}\p{M}\-0-9]{1,}\.){1,}(com|ca|org|fr){1,}\b))")39 # WEB=r"(http(s)?:\/\/)?[a-z0-9]{1}[a-z0-9-._~]+[.]{1}(com|ca)(?![\p{L}\p{M}])")40 41def f_load_tokenizer_and_model_for_nlp(model_name, pipeline_type='ner'):42 """43 Loading model and tokenizer takes a long time.44 We do it once and store the model and tokenizer in global dict for next usage45 Args:46 name: Name of the model that should be loaded and stored47 pipeline_type: type of pipeline that should be initialized48 49 Returns: tokenizer, model50 51 """52 global tokenizer_dict, models_dict, nlp_dict53 auto_model = None54 if pipeline_type == "ner":55 auto_model = AutoModelForTokenClassification56 57 if model_name not in tokenizer_dict.keys() or model_name not in models_dict.keys() or model_name not in nlp_dict.keys():58 logging.info(59 f"Loading tokenizer and model: {model_name}")60 try:61 tokenizer_dict[model_name] = AutoTokenizer.from_pretrained(model_name)62 models_dict[model_name] = auto_model.from_pretrained(model_name)63 except OSError as exc:64 raise OSError(65 f"Failed to load Hugging Face model '{model_name}'. "66 "Check outbound network access and make sure 'sentencepiece' is installed "67 "for CamemBERT-based tokenizers."68 ) from exc69 if pipeline_type == 'ner':70 nlp_dict[model_name] = pipeline(pipeline_type, model=models_dict[model_name], tokenizer=tokenizer_dict[model_name],71 aggregation_strategy="simple", device=device)72 73 74def f_ner(text, lang=default_lang):75 df_result = f_ner_regex(text)76 df_result = f_ner_model(text, lang=lang, df_result=df_result)77 return df_result78 79 80@timing81def f_ner_model(text, lang=default_lang, df_result=pd.DataFrame()):82 list_result = []83 # We split the text by sentence and run model on each one84 sentence_tokenizer = f_split_text_by_lines(text)85 for start, end, value in sentence_tokenizer:86 if value != "":87 results = f_ner_model_by_sentence(value, lang=lang, pos_offset=start)88 if len(results) != 0:89 list_result += results90 return f_concat_results(df_result, list_result)91 92 93@timing94def f_ner_model_by_sentence(sentence, lang=default_lang, df_result=pd.DataFrame(), pos_offset=0):95 """ Run ner algorithm96 97 Args:98 sentence : sentence on which to run model99 lang : lang to determine which model to use100 df_result : If results of f_ner should be combined with previous value101 (in this case we will keep the previous values if tags overlapsed)102 103 Returns:104 Dataframe with identified entities105 106 """107 108 if not config.has_option('DEFAULT', 'ner_model_' + lang):109 raise ValueError(f"No model was defined for ner in {lang}")110 111 model_name = config['DEFAULT']['ner_model_' + lang]112 f_load_tokenizer_and_model_for_nlp(model_name)113 logging.debug(f"starting {model_name} on sentence:'{sentence}'")114 115 results = nlp_dict[model_name](sentence)116 list_result = []117 for result in results:118 if result["word"] != "" and result['entity_group'] in ["PER", "LOC", "ORG", "DATE"]:119 120 # Required because sometimes spaces are included in result["word"] value, but not in start/end position121 value = sentence[result["start"]:result["end"]]122 123 # We remove any special character at the beginning124 pattern = r"[^.,'’` \":()\n].*"125 result_regex = regex.search(pattern, value, flags=regex.IGNORECASE)126 127 if result_regex is not None:128 word_raw = result_regex.group()129 word = word_raw130 real_word_start = result["start"] + result_regex.start()131 real_word_end = result["start"] + result_regex.start() + len(word_raw)132 # We check if entity might be inside a longer word, if this is the case we ignore133 letter_before = sentence[max(0, real_word_start - 1): real_word_start]134 letter_after = sentence[real_word_end: min(len(sentence), real_word_end + 1)]135 if regex.match(r"[A-z]", letter_before) or regex.match(r"[A-z]", letter_after):136 logging.debug(f"Ignoring entity {value} because letter before is"137 f" '{letter_before}' or letter after is '{letter_after}'")138 continue139 140 list_result.append(141 [result["entity_group"],142 word,143 real_word_start + pos_offset,144 real_word_end + pos_offset,145 result["score"]])146 147 return list_result148 149 150@timing151def f_concat_results(df_result, list_result_new):152 """ Merge results between existing dataframe and a list of new values153 154 Args:155 df_result: dataframe of entities156 list_result_new: list of new entities to be added in df_result157 158 Returns:159 Dataframe with all entities. Entities in list_result_new that were overlapping position of another entity in160 df_result are ignored.161 162 """163 # If df_result and list_result_new are both empty, we return an empty dataframe164 list_columns_names = ["entity", "value", "start", "end", "score"]165 if (df_result is None or len(df_result) == 0) and (list_result_new is None or len(list_result_new) == 0):166 return pd.DataFrame()167 elif len(list_result_new) > 0:168 if df_result is None or len(df_result) == 0:169 return pd.DataFrame(list_result_new,170 columns=list_columns_names)171 list_row = []172 for row in list_result_new:173 df_intersect = df_result.query("({1}>=start and {0}<=end)".format(row[2], row[3]))174 if len(df_intersect) == 0:175 list_row.append(row)176 df_final = pd.concat([df_result,177 pd.DataFrame(list_row,178 columns=list_columns_names)],179 ignore_index=True) \180 .sort_values(by="start")181 return df_final182 else:183 # If list_result_new was empty we just return df_result184 return df_result185 186 187@timing188def f_detect_language(text, default=default_lang):189 """ Detect language190 191 Args:192 text: text on which language should be detected193 default: default value if there is an error or score of predicted value is to low (default nlp.default_lang)194 195 Returns:196 "fr" or "en"197 198 """199 lang = default200 try:201 if text.strip() != "":202 lang, score = language_identifier.classify(text.strip().replace("\n"," ").lower())203 # If scroe is not high enough we will take default value instead204 if score < 0.8:205 lang = default_lang206 except Exception as e:207 logging.error("following error occurs when trying to detect language: {}".format(e))208 finally:209 return lang210 211@timing212def f_find_regex_pattern(text, type_, pattern):213 """ Find all occurences of a pattern in a text and return a list of results214 Args:215 text: the text to be analyzed216 type_: the entity type (value is added in result)217 pattern: regex pattern to be found218 219 Returns:220 A list containing type, matched value, position start and end of each result221 222 """223 list_result = []224 results = regex.finditer(pattern, text, flags=regex.IGNORECASE)225 for match in results:226 value = match.string[match.start(): match.end()].replace("\n", " ").strip()227 list_result.append([type_,228 value,229 match.start(),230 match.end(),231 1])232 return list_result233 234 235@timing236def f_ner_regex(text, dict_pattern=dict_regex_pattern,237 df_result=pd.DataFrame()):238 """Run a series of regex expression to detect email, tel and postal codes in a full text.239 240 Args:241 text: the text to be analyzed242 dict_pattern: dictionary of regex expression to be ran successively (default nlp.dict_regex_pattern)243 df_result: results of this function will be merged with values provided here.244 If value is already found at an overlapping position in df_results, the existing value will be kept245 246 Returns:247 Dataframe containing results merged with provided argument df_result (if any)248 """249 logging.debug("Starting regex")250 list_result = []251 252 # we run f_find_regex_pattern for each pattern in dict_regex253 for type_, pattern in dict_pattern.items():254 result = f_find_regex_pattern(text, type_, pattern)255 if len(result) != 0:256 list_result += result257 258 df_result = f_concat_results(df_result, list_result)259 return df_result260 261@timing262def f_split_text_by_lines(text, position_offset=0):263 """264 :param text: text that should be split265 :return: list containing for each line: [position start, position end, sentence]266 """267 results = []268 # iter_lines = regex.finditer(".*(?=\n|$)", text)269 iter_lines = regex.finditer("[^>\n]((.*?([!?.>] ){1,})|.*(?=\n|$))", text)270 for line_match in iter_lines:271 start_line = line_match.start()272 end_line = line_match.end()273 line = line_match.group()274 if len(line.strip()) > 1:275 results.append([start_line + position_offset, end_line + position_offset, line])276 return results277 278 279def f_detect_email_signature(text, df_ner=None, cut_off_score=0.6, lang=default_lang):280 # with tf.device("/cpu:0"):281 if text.strip() == "":282 return None283 if df_ner is None:284 df_ner = f_ner(text, lang=lang)285 286 try:287 df_features = _models_signatures.f_create_email_lines_features(text, df_ner=df_ner)288 289 if len(df_features)==0:290 return None291 292 # We add a dummy value for signature in order to use the same function as training.293 df_features["is_signature"] = -2294 295 x, y_out, y_mask, _, _ = _models_signatures.generate_x_y(296 df_features,297 _models_signatures.minmax_scaler,298 _models_signatures.standard_scaler,299 )300 301 y_predict = _models_signatures.predict_signature_scores(x)302 y_predict_value = (y_predict[y_mask != -1] > cut_off_score).reshape([-1])303 y_predict_value = np.pad(y_predict_value, (len(df_features) - len(y_predict_value), 0), constant_values=0)[304 -len(df_features):]305 y_predict_score = y_predict[y_mask != -1].reshape([-1])306 y_predict_score = np.pad(y_predict_score, (len(df_features) - len(y_predict_score), 0), constant_values=1)[307 -len(df_features):]308 except Exception as exc:309 logging.warning("Email signature detection is unavailable with the current runtime: %s", exc)310 return None311 312 # return(y_predict, y_mask)313 df_features["prediction"] = y_predict_value314 df_features["score"] = y_predict_score315 # return df_features316 series_position_body = df_features.query(f"""prediction==0""")['end']317 if len(series_position_body) > 0:318 body_end_pos = max(series_position_body)319 else:320 # In this case everything was detected as a signature321 body_end_pos = 0322 score = df_features.query(f"""prediction==1""")["score"].mean()323 signature_text = text[body_end_pos:].strip().replace("\n", " ")324 if signature_text != "":325 list_result = [326 # ["body", text[:body_end_pos], 0 + pos_start_email, body_end_pos + pos_start_email, 1, ""],327 ["SIGNATURE", signature_text, body_end_pos, len(text), score]]328 329 df_result = f_concat_results(pd.DataFrame(), list_result)330 else:331 df_result = None332 333 return df_result334 335 336