<jupyter_start><jupyter_text>ANALYSIS WITH FULL DATASET<jupyter_code>import pandas as pd
from pathlib import Path
import zipfile
import tarfile
basepath= '/content/drive/MyDrive/Colab_Notebooks/nlpAssignment'
folder= Path(basepath)
folder
#The spam csv has already been downloaded and saved in the data folder.
spam = folder / 'spam.csv'
!head {spam}
#Load Dataset
# convert to pandas dataframe
spam_file = pd.read_csv(spam,encoding='ISO-8859-1',
                        names= ['label','message','1','2','3'],
                        usecols=['label','message'], header=0)
spam_df = pd.DataFrame(spam_file)
spam_df.head()
!pip install -U spacy
import spacy
from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np
spacy_folder = Path('/content/drive/MyDrive/Colab_Notebooks/nlpAssignment/SPACY')
model = spacy_folder /'en_core_web_sm-3.1.0'/'en_core_web_sm'/'en_core_web_sm-3.1.0'
nlp = spacy.load(model, disable=['parser'])<jupyter_output><empty_output><jupyter_text>Custom Spacy Preprocessor<jupyter_code>class SpacyPreprocessor(BaseEstimator, TransformerMixin):
    np.random.seed(0)
    def __init__(self, lammetize=True, lower=True, remove_stop=True, 
                 remove_punct=True, remove_email=True, remove_url=True, 
                 remove_num=False, stemming = False,
                 add_user_mention_prefix=True, remove_hashtag_prefix=False):
        self.remove_stop = remove_stop
        self.remove_punct = remove_punct
        self.remove_num = remove_num
        self.remove_url = remove_url
        self.remove_email = remove_email
        self.lammetize = lammetize
        self.lower = lower
        self.stemming = stemming
        self.add_user_mention_prefix = add_user_mention_prefix
        self.remove_hashtag_prefix = remove_hashtag_prefix

 # helpfer functions for basic cleaning 

    def basic_clean(self, text):
        
        '''
        This fuction removes HTML tags from text
        '''
        if (bool(BeautifulSoup(text, "html.parser").find())==True):         
            soup = BeautifulSoup(text, "html.parser")
            text = soup.get_text()
        else:
            pass
        return re.sub(r'[\n\r]',' ', text) 

    # helper function for pre-processing with spacy and Porter Stemmer
    
    def spacy_preprocessor(self,texts):

        final_result = []
        nlp = spacy.load(model, disable=['parser','ner'])
        
        ## Add @ as a prefix so that we can separate the word from its token
        prefixes = list(nlp.Defaults.prefixes)

        if self.add_user_mention_prefix:
            prefixes += ['@']

        ## Remove # as a prefix so that we can keep hashtags and words together
        if self.remove_hashtag_prefix:
            prefixes.remove(r'#')

        prefix_regex = spacy.util.compile_prefix_regex(prefixes)
        nlp.tokenizer.prefix_search = prefix_regex.search

        matcher = Matcher(nlp.vocab)
        if self.remove_stop:
            matcher.add("stop_words", [[{"is_stop" : True}]])
        if self.remove_punct:
            matcher.add("punctuation",[ [{"is_punct": True}]])
        if self.remove_num:
            matcher.add("numbers", [[{"like_num": True}]])
        if self.remove_url:
            matcher.add("urls", [[{"like_url": True}]])
        if self.remove_email:
            matcher.add("emails", [[{"like_email": True}]])
            
        Token.set_extension('is_remove', default=False, force=True)

        cleaned_text = []
        for doc in nlp.pipe(texts,batch_size=64,disable=['parser','ner']):
            matches = matcher(doc)
            for _, start, end in matches:
                for token in doc[start:end]:
                    token._.is_remove =True
                    
            if self.lammetize:              
                text = ' '.join(token.lemma_ for token in doc 
                                if (token._.is_remove==False))
            elif self.stemming:
                text =' '.join(PorterStemmer().stem(token.text) for token in doc 
                                if (token._.is_remove==False))
            else:
                text = ' '.join(token.text for token in doc 
                                if (token._.is_remove==False))
                                   
            if self.lower:
                text=text.lower()
            cleaned_text.append(text)
        return cleaned_text

    def fit(self, X,y=None):
        return self

    def transform(self, X, y=None):
        try:
            if str(type(X)) not in ["<class 'list'>","<class 'numpy.ndarray'>"]:
                raise Exception('Expected list or numpy array got {}'.format(type(X)))
            x_clean = [self.basic_clean(text) for text in X]
            x_clean_final = self.spacy_preprocessor(x_clean)
            return x_clean_final
        except Exception as error:
            print('An exception occured: ' + repr(error))
!pip install pyspellchecker
from spellchecker import SpellChecker
import re
import re
from bs4 import BeautifulSoup

from scipy.sparse import hstack
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.feature_extraction.text import TfidfVectorizer, CountVectorizer, TfidfTransformer 
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.model_selection import GridSearchCV
from sklearn.metrics import classification_report

from xgboost import XGBClassifier
from spacy.matcher import Matcher
from spacy.tokens import Token
import nltk
from nltk.stem.porter import PorterStemmer<jupyter_output><empty_output><jupyter_text>Custom Feature Engineering Class<jupyter_code>class FeatureEngineering(BaseEstimator, TransformerMixin):
    np.random.seed(0)
    def __init__(self, word_count=False, char_count=False, 
                 char_count_wo_space=False, 
                 avg_word_length=True, digit_count=True,
                 punc_count= True, spell_check= True):
        self.word_count = word_count
        self.char_count = char_count
        self.char_count_wo_space = char_count_wo_space
        self.avg_word_length = avg_word_length
        self.digit_count = digit_count
        self.punc_count = punc_count
        self.spell_check = spell_check
        self.spell = SpellChecker()

    def fit(self, X,y=None):
        return self

    #Useful functions

    def wordCount(self,text):
        return len(text.split())

    def charCount(self,text):
        return len(text)

    def charCountWithoutSpace(self,text):
        count = 0
        for word in text.split():
            count += len(word)
        return count

    def avgWordLength(self,text):
        word_length = 0
        for token in text.split():
            word_length += len(token)
        word_count = len(text.split())
        if word_count == 0:
            return 0
        else:
            return word_length/word_count

    def digitCount(self,text):
        count = 0
        for i in text:
            if i.isdigit():
                count += 1
        return count
    
    def punctuationCount(self,text):
      line = re.findall(r'[!"$%&\'()*+,-./:;=#@?[\]^_`{|}~]*', text)     
      string="".join(line)     
      return len(list(string))
      # puncList = [token for token in self.nlpdocs if(token.is_punct)]
      # return len(puncList)

    def spellcheck(self, text,spell):
      misspelled= spell.unknown(text.split())
      return len(misspelled)

    def transform(self, X,y=None):
        try:
            if str(type(X)) not in ["<class 'list'>","<class 'numpy.ndarray'>"]:
                raise Exception('Expected list or numpy array got {}'.format(type(X)))
            final_result = []
            for index,item in enumerate(X):
                self.nlpdocs = None
                res = []
                if self.word_count:
                    res.append(self.wordCount(item))
                if self.char_count:
                    res.append(self.charCount(item))
                if self.char_count_wo_space:
                    res.append(self.charCountWithoutSpace(item))
                if self.avg_word_length:
                    res.append(self.avgWordLength(item))
                if self.digit_count:
                    res.append(self.digitCount(item))
                if self.punc_count:
                    res.append(self.punctuationCount(item))
                if self.spell_check:
                    res.append(self.spellcheck(item,self.spell))

                final_result.append(res)
            return np.array(final_result)
        except Exception as error:
            print('An exception occured: ' + repr(error))
from sklearn.model_selection import train_test_split
from sklearn.metrics import  f1_score<jupyter_output><empty_output><jupyter_text>Use Full Dataset to train and test the Pipeline From First Document<jupyter_code>X_full= spam_df['message'].values
Y_full= spam_df['label'].values
X_train_full, X_test_full, y_train_full, y_test_full = train_test_split(
    X_full, Y_full, random_state=0)
print(f'X_train: {X_train_full.shape} y_train: {y_train_full.shape}')
print(f'X_test: {X_test_full.shape} y_test: {y_test_full.shape}')<jupyter_output>X_train: (4179,) y_train: (4179,)
X_test: (1393,) y_test: (1393,)<jupyter_text>The Pipeline finalized from the mini dataset( ref: file_1) is now used to train the full dataset<jupyter_code>vectorization= Pipeline([('preprocessor', SpacyPreprocessor(lammetize= True)), 
                         ('vectorizer', TfidfVectorizer(analyzer='word', 
                                                        token_pattern=r"[\S]+", 
                                                        max_df= 0.8, 
                                                        max_features= 1000))])

feature_engineering = FeatureEngineering(digit_count= True, word_count= True)

combined_features = FeatureUnion([('vec', vectorization), 
                                  ('fe', feature_engineering)]) 

classifier_final = Pipeline([('cf', combined_features), 
                             ('classifier', XGBClassifier(scale_pos_weight=7))])

classifier_final.fit(X_train_full, y_train_full)
y_pred_train = classifier_final.predict(X_train_full)
trainScore_f1 = f1_score(y_train_full, y_pred_train, pos_label= 'spam')
y_pred_test = classifier_final.predict(X_test_full)
testScore_f1= f1_score(y_test_full, y_pred_test, pos_label= 'spam')
print('Train_score: {:.2f}'.format(trainScore_f1))
print('Test_score: {:.2f}'.format(testScore_f1))<jupyter_output>Train_score: 0.95
Test_score: 0.92<jupyter_text>This gives us a good Test f1 Score of **0.92** in classifying spam and ham. The train score is 0.95. Now, we will take a look at overall classification report to make sure both ham and spam are predicted correctly.1.   List item2.   List item<jupyter_code>from sklearn.metrics import  classification_report
print(classification_report(y_test_full,y_pred_test))<jupyter_output>precision    recall  f1-score   support

         ham       0.99      0.98      0.99      1196
        spam       0.91      0.93      0.92       197

    accuracy                           0.98      1393
   macro avg       0.95      0.96      0.95      1393
weighted avg       0.98      0.98      0.98      1393
{'fraction_non_alphanumeric': 0.100818651005517, 'fraction_numerical': 0.011923829862964941, 'mean_word_length': 2.3821847727956666, 'pattern_counts': {'":': 4, '<': 26, '<?xml version=': 0, '>': 26, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '6', 'dir': 'jupyter-structured-clean-dedup', 'id': '107105', 'license': 'Apache-2.0', 'n_tokens_mistral': 3553, 'n_tokens_neox': 3385, 'n_words': 748, 'path': 'Spam_detection/Classification_Pipelines_2.ipynb', 'repo_name': 'SumanLohit/NLP_Notebooks', 'size': '11238'}