Commit 7497da0b authored by Malik Koné's avatar Malik Koné

meilleure gestion du text wrangling

parent 0f6bc3ca
Pipeline #615 failed with stages
#-*-coding:utf-8-*-
import pandas as pd
import numpy as np
import re
from bs4 import BeautifulSoup as BS
from langdetect import detect
from langdetect.lang_detect_exception import LangDetectException
from nltk.tokenize import PunktSentenceTokenizer
import csv
import langdetect as lgd
import logging
from bs4 import BeautifulSoup as BS
import numpy as np
import os
import pandas as pd
import re
# #Rangement des fonctions à faire
logging.getLogger()
path = os.path.abspath('./')
......@@ -133,14 +139,15 @@ class UnigeData():
return self
def prepare_text(self):
"""Attention change la df"""
"""Attention change la df, créer plusieurs colones avec des (features du text) 'lang, tokens, nb token long bigest token, bigest token text, img, code, contact, links"""
if len(self.df) == 0:
self.df = self.load_df(filterCols=False)
logging.debug(f'>>>>self.df.columns={self.df.columns}<<<<')
mask = self.df.discussion_answer_content.isna() == True
dqid = S.newNames['ufm']["discussion_question_id"]
# dqid = S.newNames['ufm']["discussion_question_id"]
dqid = "discussion_question_id"
for col in ['discussion_answer_parent_discussion_answer_id', 'discussion_answer_id']:
self.df.loc[mask, col] = self.df.loc[mask, dqid]
pass
......@@ -149,24 +156,143 @@ class UnigeData():
self.df.index = pd.Index(range(len(self.df)))
# commonTags = {'em', 'a', 'u', 'strong', 'li', 'list', 'code', 'sup', 'text', 'img', 'co-content'}
for cmd in ['text', 'link', 'img', 'code']:
for cmd in ['link', 'img', 'code', "contact", "text"]: # text en dernier car on supprime aussi à ce moment là.
self.df.loc[:, cmd] = self.df.discussion_answer_content.apply(lambda d: wrangle_text(d, cmd=cmd))
self.df.loc[:, "lang"] = self.df.text.apply(detect_language)
self.df.loc[:, 'tokens'] = self.df.text.apply(UNIGEtokenize)
self.df.loc[:, 'lent'] = self.df.tokens.apply(len)
self.df = self.df.loc[self.df.lent != 0] # phrase sans token
def big_token(tokens):
return sorted([(t, len(t)) for t in tokens], key=lambda x: x[1])[-1]
# big tokens
self.df.loc[:, 'bigtok'] = self.df.tokens.apply(lambda toks: big_token(toks)[0])
self.df.loc[:, 'lenbigt'] = self.df.tokens.apply(lambda toks: big_token(toks)[1])
# on filtre
# les message dont le token est plus grand que Honorificabilitudinitatibus (shakespear)
# les message avec au moins un token # à voir
# les message non anglais
mask = (self.df.lenbigt < 27) & (self.df.lent > 0) & (self.df.lang.str.contains('en').values)
self.df = self.df.loc[mask]
return self.df
def wrangle_text(message, cmd='text'):
"""Enlève les tags, en récupère quelques uns"""
bs = BS(message, features="xml")
if cmd == 'text':
return re.sub(r'([^ ])([A-Z])', r'\1 \2', bs.text)
"""Enlève les tags, en récupère quelques uns
récupère des bouts importants du text"""
# try to get urls ands emails
speA = "[]!*'();:&=+$,/?#[A-Za-z0-9_.~%-]"
speR = "[A-Za-z0-9]"
urlpat = f"(?:https?|file)://(?:{speA}|@)+"
mailpat = f'(?:(?:{speR}+{speA}*|\s)@{speA}*{speR}+)'
if cmd == 'text':
# on supprime les liens et référence à quelqu'uns des messages
message, n = re.subn(urlpat, '', message)
message, n = re.subn(mailpat, '', message)
message = message.replace('><', '> <') # pour eviter que les mots se colle (pas de point ou \n car span)
text = BS(message, features="xml").get_text()
# Faits des réponse quantitatives d'une unique expression numérique, une phrase.
qtypat = '((?:\d|<)(?:\d|\s|-|=|%|/|<)*)'
if re.fullmatch(qtypat, text):
text = re.sub(qtypat, lambda m: f"My answer is {m.group(1)}.", text)
return cap_sent(text, own=True) # pour aider detection de la langue
elif cmd == 'link':
return list({link.get('href') for link in bs.find_all('a')})
return re.findall(urlpat, message)
elif cmd == "img":
return bs.find_all('img')
return BS(message, features="xml").find_all('img')
elif cmd == "code":
return bs.find_all('code')
return BS(message, features="xml").find_all('code')
elif cmd == "contact":
return re.findall(mailpat, message)
else:
raise(Exception("command not known"))
pass
def langexception(f):
"""wrapper pour la langue detect"""
def f_with_exception(args):
try:
return f(args)
except LangDetectException as lde:
# gestion des messages avec pas assez de mots pour détecter la langue
if "No features in text" in lde.__repr__():
return None
else:
raise(lde)
pass
pass
return f_with_exception
@langexception
def detect_language(msg):
"""# ### Détecter la langue"""
# à paralléliser
langue = detect(msg)
if langue != 'en':
langue = filtre_lang(msg, nb_estim=10, seuil=.95)
return langue
def filtre_lang(msg, nb_estim=10, seuil=.95, deflang=['en', 'af'], possible_langs=['pt', 'es', 'fr', 'ca', 'zh-cn']):
"""effectuer plusieurs estimation de language pour un message. La langue par défaut sera deflang[0]"""
altl = {}
# l'idée c'est de voir la moyenne sur un nombre k d'estimation. Si > seuil alors lang sinon, doute. en ?
# si au cours des estimation il y a l'anglais alors on le prend
for i in range(nb_estim):
estim = lgd.detect_langs(msg)
langs = {l.lang: l.prob for l in estim}
if set(deflang) & set(langs.keys()):
return deflang[0]
# on fait un dict avec les langues trouvées et leur proba cumulées
for lg, proba in langs.items():
altl[lg] = altl.get(lg, 0) + proba
pass
pass
# si la prova de max_lang est forte et que ce n'est pas une langue authorisée renvois None
max_lang, max_prob = sorted(list(altl.items()), key=lambda x: x[1])[-1]
if max_prob/nb_estim < seuil:
return deflang[0]
elif max_lang in possible_langs:
return max_lang
else:
return None
def UNIGEtokenize(msg):
"""tokenizer pour le corpus d'UNIGE. Les tokenizer publiques sont souvent pleins d'assomptions qu'il faut maîtriser voir # see http://aclweb.org/anthology/W18-2502 pour avoir de bon résultats avec la veditisation (lemùing) qui suit logiquement
- msg : est une chaine de charactère représentant un message dans un forum du MOOC"""
pat = r'\b\w\w+\b' # mot de 2 char ou plus
tokens = re.findall(pat, msg.lower())
return tokens
def cap_sent(text, own=True):
"""capitalize sentence using my own sent tokenizer or that of nltk"""
if own:
# ~10 µs mais ne gère pas peut-être pas bien les abréviations eg Mlle."
endsent = r'(?:(?<!\b\w)([.?!]|\A)+)' # exclus les phrase d'une lettre
sentpat = endsent + r'\s*(?<=\b)(\w)'
wtext = re.subn(sentpat, lambda m: m.group(1) + '\n' + m.group(2).upper(), text)[0].strip()
else:
# ~310 µs
sent_tokenizer = PunktSentenceTokenizer(text)
sents = [x.capitalize() for x in sent_tokenizer.tokenize(text)]
wtext = "\n".join(sents)
return wtext
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment