src/preprocessing.py

import re
import string
from collections import OrderedDict
from typing import Callable, List, Optional, Tuple

import numpy as np
import pandas as pd
import spacy
import streamlit as st
from pandas.core.series import Series
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.preprocessing import LabelEncoder
from stqdm import stqdm
from textacy.preprocessing import make_pipeline, normalize, remove, replace

from .configs import Languages

stqdm.pandas()


def encode(text: pd.Series, labels: pd.Series):
    """
    Encodes text in mathematical object ameanable to training algorithm
    """
    tfidf_vectorizer = TfidfVectorizer(
        input="content",  # default: file already in memory
        encoding="utf-8",  # default
        decode_error="strict",  # default
        strip_accents=None,  # do nothing
        lowercase=False,  # do nothing
        preprocessor=None,  # do nothing - default
        tokenizer=None,  # default
        stop_words=None,  # do nothing
        analyzer="word",
        ngram_range=(1, 3),  # maximum 3-ngrams
        min_df=0.001,
        max_df=0.75,
        sublinear_tf=True,
    )
    label_encoder = LabelEncoder()

    with st.spinner("Encoding text using TF-IDF and Encoding labels"):
        X = tfidf_vectorizer.fit_transform(text.values)
        y = label_encoder.fit_transform(labels.values)

    return {
        "X": X,
        "y": y,
        "X_names": np.array(tfidf_vectorizer.get_feature_names()),
        "y_names": label_encoder.classes_,
    }


# 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):
    return _re_normalize_acronyms.sub(t.translate(str.maketrans("", "", string.punctuation)).upper(), t)


_re_non_word = re.compile(r"\W")
def remove_non_word(t):
    return _re_non_word.sub(" ", t)


_re_space = re.compile(r" {2,}")
def normalize_useless_spaces(t):
    return _re_space.sub(" ", t)


_re_rep = re.compile(r"(\S)(\1{2,})")
def normalize_repeating_chars(t):
    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):
    def _replace_wrep(m):
        c, cc, e = m.groups()
        return c

    return _re_wrep.sub(_replace_wrep, t)


# fmt: on
class Lemmatizer:
    """Creates lemmatizer based on spacy"""

    def __init__(
        self, language: str, remove_stop: bool = True, lemmatization: bool = True
    ) -> None:
        self.language = language
        self.nlp = spacy.load(
            Languages[language].value, exclude=["parser", "ner", "pos", "tok2vec"]
        )
        self._lemmatizer_fn = self._get_lemmatization_fn(remove_stop, lemmatization)
        self.lemmatization = lemmatization

    def _get_lemmatization_fn(
        self, remove_stop: bool, lemmatization: bool
    ) -> Optional[Callable]:
        """Return the correct spacy Doc-level lemmatizer"""
        if remove_stop and lemmatization:

            def lemmatizer_fn(doc: spacy.tokens.doc.Doc) -> str:
                return " ".join(
                    [t.lemma_ for t in doc if t.lemma_ != "-PRON-" and not t.is_stop]
                )

        elif remove_stop and not lemmatization:

            def lemmatizer_fn(doc: spacy.tokens.doc.Doc) -> str:
                return " ".join([t for t in doc if not t.is_stop])

        elif lemmatization and not remove_stop:

            def lemmatizer_fn(doc: spacy.tokens.doc.Doc) -> str:
                return " ".join([t.lemma_ for t in doc if t.lemma_ != "-PRON-"])

        else:
            self.status = False
            return

        return lemmatizer_fn

    def __call__(self, series: Series) -> Series:
        """
        Apply spacy pipeline to transform string to spacy Doc and applies lemmatization
        """
        res = []
        pbar = stqdm(total=len(series), desc="Lemmatizing")
        for doc in self.nlp.pipe(series, batch_size=500):
            res.append(self._lemmatizer_fn(doc))
            pbar.update(1)
        pbar.close()
        return pd.Series(res)


class PreprocessingPipeline:
    def __init__(
        self, pre_steps: List[str], lemmatizer: Lemmatizer, post_steps: List[str]
    ):

        # build pipeline
        self.pre_pipeline, self.lemmatizer, self.post_pipeline = self.make_pipeline(
            pre_steps, lemmatizer, post_steps
        )

    def __call__(self, series: Series) -> Series:
        with st.spinner("Pre-lemmatization cleaning"):
            res = series.progress_map(self.pre_pipeline)

        with st.spinner("Lemmatizing"):
            res = self.lemmatizer(series)

        with st.spinner("Post-lemmatization cleaning"):
            res = series.progress_map(self.post_pipeline)

        return res

    def make_pipeline(
        self, pre_steps: List[str], lemmatizer: Lemmatizer, post_steps: List[str]
    ) -> Tuple[Callable]:

        # pre-lemmatization steps
        pre_steps = [
            self.pipeline_components()[step]
            for step in pre_steps
            if step in self.pipeline_components()
        ]
        pre_steps = make_pipeline(*pre_steps) if pre_steps else lambda x: x

        # lemmatization
        lemmatizer = lemmatizer if lemmatizer.lemmatization else lambda x: x

        # post lemmatization steps
        post_steps = [
            self.pipeline_components()[step]
            for step in post_steps
            if step in self.pipeline_components()
        ]
        post_steps = make_pipeline(*post_steps) if post_steps else lambda x: x

        return pre_steps, lemmatizer, post_steps

    @staticmethod
    def pipeline_components() -> "OrderedDict[str, Callable]":
        """Returns available cleaning steps in order"""
        return OrderedDict(
            [
                ("lower", lambda x: x.lower()),
                ("normalize_unicode", normalize.unicode),
                ("normalize_bullet_points", normalize.bullet_points),
                ("normalize_hyphenated_words", normalize.hyphenated_words),
                ("normalize_quotation_marks", normalize.quotation_marks),
                ("normalize_whitespace", 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),
                ("normalize_useless_spaces", normalize_useless_spaces),
                ("normalize_repeating_chars", normalize_repeating_chars),
                ("normalize_repeating_words", normalize_repeating_words),
                ("strip", lambda x: x.strip()),
            ]
        )