app.py

import os
import re
import contractions
import unicodedata
import spacy
import keras
import requests
import shutil
import json
import gradio as gr
import pandas as pd
import numpy as np
from PIL import Image
from keras import backend as K
from keras.utils.data_utils import pad_sequences
from gensim.models import Word2Vec
from gensim.models.callbacks import CallbackAny2Vec

import nltk
nltk.download('punkt')
nltk.download('stopwords')

os.system('python -m spacy download en_core_web_sm')

import en_core_web_sm
nlp = en_core_web_sm.load()


def recall_m(y_true, y_pred):
  true_positives = K.sum(K.round(K.clip(y_true * y_pred, 0, 1)))
  possible_positives = K.sum(K.round(K.clip(y_true, 0, 1)))
  recall = true_positives / (possible_positives + K.epsilon())
  return recall

def precision_m(y_true, y_pred):
  true_positives = K.sum(K.round(K.clip(y_true * y_pred, 0, 1)))
  predicted_positives = K.sum(K.round(K.clip(y_pred, 0, 1)))
  precision = true_positives / (predicted_positives + K.epsilon())
  return precision

def f1_m(y_true, y_pred):
  precision = precision_m(y_true, y_pred)
  recall = recall_m(y_true, y_pred)
  return 2*((precision*recall)/(precision+recall+K.epsilon()))


#initialise callback class
class callback(CallbackAny2Vec):
  """
  Print the loss value after each epoch
  """
  def __init__(self):
    self.epoch = 0
    #gensim loss is cumulative, so we record previous values to print
    self.loss_previous_step = 0 

  def on_epoch_end(self, model):
    loss = model.get_latest_training_loss()
    if self.epoch % 100 == 0:
      print('Loss after epoch {}: {}'.format(self.epoch, loss-self.loss_previous_step))

    self.epoch+= 1
    self.loss_previous_step = loss



def spacy_lemmatize_text(text):
  text = nlp(text)
  text = ' '.join([word.lemma_ if word.lemma_ != '-PRON-' else word.text for word in text])
  return text

def remove_accented_chars(text):
  text = unicodedata.normalize('NFC', text).encode('ascii', 'ignore').decode('utf-8', 'ignore')
  return text
 
def remove_special_characters(text, remove_digits=False):
  pattern = r'[^a-zA-Z0-9\s]' if not remove_digits else r'[^a-zA-Z\s]'
  text = re.sub(pattern, '', text)
  return text
  
def remove_stopwords(text, is_lower_case=False, stopwords=None):
  if not stopwords:
      stopwords = nltk.corpus.stopwords.words('english')
  tokens = nltk.word_tokenize(text)
  tokens = [token.strip() for token in tokens]
  
  if is_lower_case:
      filtered_tokens = [token for token in tokens if token not in stopwords]
  else:
      filtered_tokens = [token for token in tokens if token.lower() not in stopwords]
  
  filtered_text = ' '.join(filtered_tokens)    
  return filtered_text


def pre_process(df):
  opo_texto_data = df['opo_texto']
  opo_texto_ele_data = df['opo_texto_ele']
  opo_texto_final = []

  for i in range(len(opo_texto_data)):
    if opo_texto_data[i] == opo_texto_ele_data[i]:
      opo_texto_final.append(opo_texto_data[i])
    elif pd.isna(opo_texto_ele_data[i]):
      opo_texto_final.append(opo_texto_data[i])
    elif len(nltk.word_tokenize(opo_texto_data[i])) < 4000:
      opo_texto_final.append(opo_texto_data[i]+". "+opo_texto_ele_data[i])
    else:
      opo_texto_final.append(opo_texto_data[i])
        
  pre_processed_data = []
  for opo in opo_texto_final:
    opo_texto_sem_caracteres_especiais = (remove_accented_chars(opo))
    sentenceExpanded = contractions.fix(opo_texto_sem_caracteres_especiais)
    sentenceWithoutPunctuation = remove_special_characters(sentenceExpanded , remove_digits=True)
    sentenceLowered = sentenceWithoutPunctuation.lower()
    sentenceLemmatized = spacy_lemmatize_text(sentenceLowered)
    sentenceLemStopped = remove_stopwords(sentenceLemmatized, is_lower_case=False)
    sentenceTokenized = nltk.word_tokenize(sentenceLemStopped)
    pre_processed_data.append(sentenceTokenized)

  df['opo_pre_tkn'] = pre_processed_data
  return df

def classify(df, new_column = True, pre_processed = False):
  sentencesMCTIList_xp8 = df['opo_pre_tkn']
    
  formatted_sentences = []
  if not pre_processed:
    for sentence in sentencesMCTIList_xp8:
      formatted_sentences.append(json.loads(sentence.replace("'",'"')))
    del sentencesMCTIList_xp8
  else:
    formatted_sentences = sentencesMCTIList_xp8

  words = list(reloaded_w2v_model.wv.vocab) 
  item_shape = np.shape(reloaded_w2v_model.wv[words[0]])

  MCTIinput_vector = []
  for sentence in formatted_sentences:
    aux_vector = []
    for word in sentence:
      if word in reloaded_w2v_model.wv.vocab:
        aux_vector.append(reloaded_w2v_model.wv[word])
      else:
        aux_vector.append(np.zeros(item_shape))
    MCTIinput_vector.append(aux_vector)
  del formatted_sentences

  MCTIinput_padded = pad_sequences(MCTIinput_vector, maxlen=2726, padding='pre')
  del MCTIinput_vector

  predictions = reconstructed_model_CNN.predict(MCTIinput_padded)
  del MCTIinput_padded
  print(predictions)

  cleaned_up_predictions = []
  for prediction in predictions:
    cleaned_up_predictions.append(1 if prediction >= 0.5 else 0);
  del predictions

  df['classification'] = cleaned_up_predictions
  if not new_column:
    df = df.loc[df['classification'] == 1]

  return df

def gen_output(data):
  data.to_excel("output.xlsx", index=False)
  return "output.xlsx"
    

reloaded_w2v_model = Word2Vec.load('word2vec_xp8.model')

reconstructed_model_CNN = keras.models.load_model("best weights CNN.h5", 
                                                   custom_objects={'f1_m':f1_m, 
                                                                   "precision_m":precision_m, 
                                                                   "recall_m":recall_m})

def app(operacao, resultado, dados):
  data = pd.read_excel(dados)
  print("Dados Carregados!")

  if operacao == "Pré-processamento + Classificação" :
    preprocessed = pre_process(data)
    df = classify(preprocessed, resultado == "Nova Coluna", pre_processed = True)
    output = gen_output(df)
    
    return output
  elif operacao == "Apenas Pré-processamento" :
    df = pre_process(data)
    output = gen_output(df)

    return output
  elif operacao == "Apenas Classificação" :
    df = classify(data, resultado == "Nova Coluna")
    output = gen_output(df)

    return output

iface = gr.Interface(
    fn=app, 
    inputs=[
            gr.Radio(["Pré-processamento + Classificação", "Apenas Pré-processamento", "Apenas Classificação"]),
            gr.Radio(["Nova Coluna", "Filtrar planilha"]),
            "file"
           ], 
    outputs="file"
)
iface.launch()