wordify / src /preprocessing.py
Pietro Lesci
formatting
dc4ad9e
import re
import string
from collections import OrderedDict
from typing import Callable, List, Optional, Union
import spacy
import vaex
from pandas.core.frame import DataFrame
from textacy.preprocessing import make_pipeline, normalize, remove, replace
from .configs import Languages
# more [here](https://github.com/fastai/fastai/blob/master/fastai/text/core.py#L42)
# and [here](https://textacy.readthedocs.io/en/latest/api_reference/preprocessing.html)
# fmt: off
_re_normalize_acronyms = re.compile(r"(?:[a-zA-Z]\.){2,}")
def normalize_acronyms(t: str) -> str:
return _re_normalize_acronyms.sub(t.translate(str.maketrans("", "", string.punctuation)).upper(), t)
_re_non_word = re.compile(r"[^A-Za-z]+")
def remove_non_word(t: str) -> str:
"Removes non-words characters and digits from the text using the regex `[^A-Za-z]+`"
return _re_non_word.sub(" ", t)
_re_space = re.compile(r" {2,}")
def normalize_useless_spaces(t: str) -> str:
return _re_space.sub(" ", t)
_re_rep = re.compile(r"(\S)(\1{2,})")
def normalize_repeating_chars(t: str) -> str:
def _replace_rep(m):
c, cc = m.groups()
return c
return _re_rep.sub(_replace_rep, t)
_re_wrep = re.compile(r"(?:\s|^)(\w+)\s+((?:\1\s+)+)\1(\s|\W|$)")
def normalize_repeating_words(t: str) -> str:
def _replace_wrep(m):
c, cc, e = m.groups()
return c
return _re_wrep.sub(_replace_wrep, t)
_re_remove_numbers = re.compile(r"\d+")
def remove_numbers(t: str) -> str:
return _re_remove_numbers.sub(" ", t)
def lowercase(t: str) -> str:
"Lowercases the text"
return t.lower()
def strip(t: str) -> str:
return t.strip()
def lemmatize_remove_stopwords(doc: spacy.tokens.doc.Doc) -> str:
return " ".join(
[t.lemma_ for t in doc if t.lemma_ != "-PRON-" and not t.is_stop]
)
def remove_stopwords(doc: spacy.tokens.doc.Doc) -> str:
return " ".join([t.text for t in doc if not t.is_stop])
def lemmatize_keep_stopwords(doc: spacy.tokens.doc.Doc) -> str:
return " ".join([t.lemma_ for t in doc if t.lemma_ != "-PRON-"])
def identity(t):
return t
# fmt: on
class PreprocessingPipeline:
def __init__(
self,
language: str,
pre_steps: Optional[List[str]],
lemmatization_step: Optional[str],
post_steps: Optional[List[str]],
):
self.language = language
self.pre_steps = pre_steps
self.lemmatization_step = lemmatization_step
self.post_steps = post_steps
self.pre = self.make_pipe_component(self.pre_steps, self.language)
self.post = self.make_pipe_component(self.post_steps, self.language)
self.nlp = self.make_nlp(self.lemmatization_step, self.language)
self.lemma = self.make_lemma(self.lemmatization_step, self.language)
# def apply_multiproc(fn, series):
# with mp.Pool(mp.cpu_count()) as pool:
# new_series = pool.map(fn, series)
# return new_series
def vaex_process(self, df: DataFrame, text_column: str) -> DataFrame:
def fn(t):
return self.post(self.lemma(self.nlp(self.pre(t))))
vdf = vaex.from_pandas(df)
vdf["processed_text"] = vdf.apply(
fn, arguments=[vdf[text_column]], vectorize=False
)
df = vdf.to_pandas_df()
return df
@classmethod
def make_pipe_component(cls, steps: Optional[List[str]], language: str) -> Callable:
if not steps:
return identity
elif language in ("MultiLanguage", "Chinese") and "remove_non_words" in steps:
idx = steps.index("remove_non_words")
steps = (
steps[:idx]
+ ["remove_numbers", "remove_punctuation"]
+ steps[idx + 1 :]
)
components = [cls.pipeline_components()[step] for step in steps]
return make_pipeline(*components)
@staticmethod
def make_nlp(
lemmatization_step: Optional[str], language: str
) -> Union[spacy.language.Language, Callable]:
if (
lemmatization_step is None
or lemmatization_step == "Disable lemmatizer"
or (
lemmatization_step == "Spacy lemmatizer (keep stopwords)"
and language in ("MultiLanguage", "Chinese")
)
):
return identity
return spacy.load(Languages[language].value, disable=["parser", "ner"])
@classmethod
def make_lemma(cls, lemmatization_step: Optional[str], language: str) -> Callable:
if (
lemmatization_step is None
or lemmatization_step == "Disable lemmatizer"
or (
lemmatization_step == "Spacy lemmatizer (keep stopwords)"
and language in ("MultiLanguage", "Chinese")
)
):
return identity
elif (
lemmatization_step == "Spacy lemmatizer (remove stopwords)"
and language in ("MultiLanguage", "Chinese")
):
return cls.lemmatization_component().get("Remove stopwords")
return cls.lemmatization_component().get(lemmatization_step)
@staticmethod
def pipeline_components() -> "OrderedDict[str, Callable]":
"""Returns available cleaning steps in order"""
return OrderedDict(
[
("lowercase", lowercase),
("normalize_unicode", normalize.unicode),
("normalize_bullet_points", normalize.bullet_points),
("normalize_hyphenated_words", normalize.hyphenated_words),
("normalize_quotation_marks", normalize.quotation_marks),
("normalize_whitespaces", normalize.whitespace),
("replace_urls", replace.urls),
("replace_currency_symbols", replace.currency_symbols),
("replace_emails", replace.emails),
("replace_emojis", replace.emojis),
("replace_hashtags", replace.hashtags),
("replace_numbers", replace.numbers),
("replace_phone_numbers", replace.phone_numbers),
("replace_user_handles", replace.user_handles),
("normalize_acronyms", normalize_acronyms),
("remove_accents", remove.accents),
("remove_brackets", remove.brackets),
("remove_html_tags", remove.html_tags),
("remove_punctuation", remove.punctuation),
("remove_non_words", remove_non_word),
("remove_numbers", remove_numbers),
("normalize_useless_spaces", normalize_useless_spaces),
("normalize_repeating_chars", normalize_repeating_chars),
("normalize_repeating_words", normalize_repeating_words),
("strip", strip),
]
)
@staticmethod
def lemmatization_component() -> "OrderedDict[str, Optional[Callable]]":
return OrderedDict(
[
("Spacy lemmatizer (keep stopwords)", lemmatize_keep_stopwords),
("Spacy lemmatizer (remove stopwords)", lemmatize_remove_stopwords),
("Disable lemmatizer", identity),
("Remove stopwords", remove_stopwords),
]
)