# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Implementing Feature Union
# Using the given custom transformer, `StartingVerbExtractor`, add a feature union to your pipeline to incorporate a feature that indicates with a boolean value whether the starting token of a post is identified as a verb.

import nltk
nltk.download(['punkt', 'wordnet', 'averaged_perceptron_tagger'])

# +
import re
import numpy as np
import pandas as pd
from nltk.tokenize import word_tokenize
from nltk.stem import WordNetLemmatizer

from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.metrics import confusion_matrix
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.feature_extraction.text import CountVectorizer, TfidfTransformer
# -

from custom_transformer import StartingVerbExtractor

url_regex = 'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\(\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+'


# ### Build your pipeline to have this structure:
# - Pipeline
#     - feature union
#         - text pipeline
#             - count vectorizer
#             - TFIDF transformer
#         - starting verb extractor
#     - classifier

def model_pipeline():
    pipeline = Pipeline(
        [
            ('union', FeatureUnion(
                [
                    (
                        'text', Pipeline(
                            [
                                ('vect', CountVectorizer(tokenizer=tokenize)),
                                ('tfidf', TfidfTransformer())
                            ]
                        )
                    ),
                    (
                        'starting_verb', 
                        StartingVerbExtractor()
                    ),
                ]
            )),
            ('clf', RandomForestClassifier()),
        ]
    )
    return pipeline


# ### Run program to test

# +
def load_data():
    df = pd.read_csv('corporate_messaging.csv', encoding='latin-1')
    df = df[(df["category:confidence"] == 1) & (df['category'] != 'Exclude')]
    X = df.text.values
    y = df.category.values
    return X, y


def tokenize(text):
    detected_urls = re.findall(url_regex, text)
    for url in detected_urls:
        text = text.replace(url, "urlplaceholder")

    tokens = word_tokenize(text)
    lemmatizer = WordNetLemmatizer()

    clean_tokens = []
    for tok in tokens:
        clean_tok = lemmatizer.lemmatize(tok).lower().strip()
        clean_tokens.append(clean_tok)

    return clean_tokens


def display_results(y_test, y_pred):
    labels = np.unique(y_pred)
    confusion_mat = confusion_matrix(y_test, y_pred, labels=labels)
    accuracy = (y_pred == y_test).mean()

    print("Labels:", labels)
    print("Confusion Matrix:\n", confusion_mat)
    print("Accuracy:", accuracy)


def main():
    X, y = load_data()
    X_train, X_test, y_train, y_test = train_test_split(X, y)

    model = model_pipeline()
    model.fit(X_train, y_train)
    y_pred = model.predict(X_test)

    display_results(y_test, y_pred)

main()
# -



StarcoderdataJupyterScriptsDedupFiltered