import gradio as gr import torch from transformers import AutoTokenizer, AutoModelForSequenceClassification,AutoModel import re from textblob import TextBlob from nltk import pos_tag, word_tokenize from nltk.corpus import stopwords import emoji import string import nltk from nltk import pos_tag from nltk.tokenize import word_tokenize from nltk.corpus import stopwords import textstat import pandas as pd from transformers import pipeline from torch.utils.data import Dataset, DataLoader import torch.nn as nn #Loading author details def average_word_length(tweet): words = tweet.split() return sum(len(word) for word in words) / len(words) def lexical_diversity(tweet): words = tweet.split() unique_words = set(words) return len(unique_words) / len(words) def count_capital_letters(tweet): return sum(1 for char in tweet if char.isupper()) def count_words_surrounded_by_colons(tweet): # Define a regular expression pattern to match words surrounded by ':' pattern = r':(\w+):' # Use re.findall to find all matches in the tweet matches = re.findall(pattern, tweet) # Return the count of matched words return len(matches) def count_emojis(tweet): # Convert emoji symbols to their corresponding names tweet_with_names = emoji.demojize(tweet) return count_words_surrounded_by_colons(tweet_with_names) def hashtag_frequency(tweet): hashtags = re.findall(r'#\w+', tweet) return len(hashtags) def mention_frequency(tweet): mentions = re.findall(r'@\w+', tweet) return len(mentions) def count_special_characters(tweet): special_characters = [char for char in tweet if char in string.punctuation] return len(special_characters) def stop_word_frequency(tweet): stop_words = set(stopwords.words('english')) words = [word for word in tweet.split() if word.lower() in stop_words] return len(words) nltk.download('punkt') nltk.download('averaged_perceptron_tagger') nltk.download('stopwords') def get_linguistic_features(tweet): # Tokenize the tweet words = word_tokenize(tweet) # Remove stopwords stop_words = set(stopwords.words('english')) filtered_words = [word.lower() for word in words if word.isalnum() and word.lower() not in stop_words] # Get parts of speech tags pos_tags = pos_tag(filtered_words) # Count various linguistic features noun_count = sum(1 for word, pos in pos_tags if pos.startswith('N')) verb_count = sum(1 for word, pos in pos_tags if pos.startswith('V')) participle_count = sum(1 for word, pos in pos_tags if pos.startswith('V') and ('ing' in word or 'ed' in word)) interjection_count = sum(1 for word, pos in pos_tags if pos == 'UH') pronoun_count = sum(1 for word, pos in pos_tags if pos.startswith('PRP')) preposition_count = sum(1 for word, pos in pos_tags if pos.startswith('IN')) adverb_count = sum(1 for word, pos in pos_tags if pos.startswith('RB')) conjunction_count = sum(1 for word, pos in pos_tags if pos.startswith('CC')) return { 'Noun_Count': noun_count, 'Verb_Count': verb_count, 'Participle_Count': participle_count, 'Interjection_Count': interjection_count, 'Pronoun_Count': pronoun_count, 'Preposition_Count': preposition_count, 'Adverb_Count': adverb_count, 'Conjunction_Count': conjunction_count } def readability_score(tweet): return textstat.flesch_reading_ease(tweet) def get_url_frequency(tweet): urls = re.findall(r'http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+', tweet) return len(urls) # Define a function to extract features from a single tweet def extract_features(tweet): features = { 'Average_Word_Length': average_word_length(tweet), # 'Average_Sentence_Length': average_sentence_length(tweet), 'Lexical_Diversity': lexical_diversity(tweet), 'Capital_Letters_Count': count_capital_letters(tweet), # Uncomment if you want to include this feature 'Hashtag_Frequency': hashtag_frequency(tweet), 'Mention_Frequency': mention_frequency(tweet), 'count_emojis': count_emojis(tweet), 'special_chars_count': count_special_characters(tweet), 'Stop_Word_Frequency': stop_word_frequency(tweet), **get_linguistic_features(tweet), # Include linguistic features 'Readability_Score': readability_score(tweet), 'URL_Frequency': get_url_frequency(tweet) # Assuming you have the correct function for this } return features # # Extract features for all tweets # features_list = [extract_features(tweet) for tweet in X['text']] # # Create a Pandas DataFrame # X_new = pd.DataFrame(features_list) # Loading personality model def personality_detection(text, threshold=0.05, endpoint= 1.0): tokenizer = AutoTokenizer.from_pretrained ("Nasserelsaman/microsoft-finetuned-personality",token=PERSONALITY_TOKEN) model = AutoModelForSequenceClassification.from_pretrained ("Nasserelsaman/microsoft-finetuned-personality",token=PERSONALITY_TOKEN) inputs = tokenizer(text, truncation=True, padding=True, return_tensors="pt") outputs = model(**inputs) predictions = outputs.logits.squeeze().detach().numpy() # Get raw logits logits = model(**inputs).logits # Apply sigmoid to squash between 0 and 1 probabilities = torch.sigmoid(logits) # # Set values less than the threshold to 0.05 # predictions[predictions < threshold] = 0.05 # predictions[predictions > endpoint] = 1.0 # print("per",probabilities[0][0].detach().numpy()) # print("per",probabilities[0][1].detach().numpy()) # print("per",probabilities[0][2].detach().numpy()) # print("per",probabilities[0][3].detach().numpy()) # print("per",probabilities[0][4].detach().numpy()) # label_names = ['Agreeableness', 'Conscientiousness', 'Extraversion', 'Neuroticism', 'Openness'] # # result = {label_names[i]: f"{predictions[i]*100:.0f}%" for i in range(len(label_names))} # result = {label_names[i]: f"{probabilities}%" for i in range(len(label_names))} # probabilities return [probabilities[0][0].detach().numpy() ,probabilities[0][1].detach().numpy() ,probabilities[0][2].detach().numpy() ,probabilities[0][3].detach().numpy() ,probabilities[0][4].detach().numpy()] # tokenizer = AutoTokenizer.from_pretrained("Nasserelsaman/microsoft-finetuned-personality") # model = AutoModelForSequenceClassification.from_pretrained("Nasserelsaman/microsoft-finetuned-personality") #Loading emotion model # tokenizer = AutoTokenizer.from_pretrained("cardiffnlp/twitter-roberta-base-emotion-multilabel-latest") # model = AutoModelForSequenceClassification.from_pretrained("cardiffnlp/twitter-roberta-base-emotion-multilabel-latest") ##use this for gpu # pipe = pipeline("text-classification", model="cardiffnlp/twitter-roberta-base-emotion-multilabel-latest", return_all_scores=True,device=device ) ##use this for cpu def calc_emotion_score(tweet): pipe = pipeline("text-classification", model="cardiffnlp/twitter-roberta-base-emotion-multilabel-latest", return_all_scores=True ) emotions = pipe(tweet)[0] for i in emotions: print(i) return [emotions[0]['score'],emotions[1]['score'],emotions[2]['score'],emotions[3]['score'],emotions[4]['score'],emotions[5]['score'],emotions[6]['score'],emotions[7]['score'],emotions[8]['score'],emotions[9]['score'],emotions[10]['score']] #DCL model launching def load_model(tweet): # model = torch.load("./authormodel.pt",map_location ='cpu') # print(model) model_name = "vinai/bertweet-base" PADDING_MAX_LENGTH = 45 tokenizer = AutoTokenizer.from_pretrained(model_name) inputs = tokenizer(tweet, truncation=True, padding='max_length',max_length=PADDING_MAX_LENGTH,add_special_tokens=True, return_tensors="pt") print(inputs) emotion_list = calc_emotion_score(tweet) print(emotion_list) features_list = extract_features(tweet) for i in features_list.values(): emotion_list.append(i) print("emotion + author",emotion_list) # print() # print(features_list) personality_list = personality_detection(tweet) print("personality",personality_list) # person_list = [personality_list["Extraversion"],personality_list['Neuroticism'],personality_list['Agreeableness'],personality_list['Conscientiousness'],personality_list['Openness']] emotion_list.extend(personality_list) print("final list",emotion_list) # print(str(features_list["Average_Word_Length"])) inputs['emotion_author_vector'] = torch.tensor([emotion_list]) print("final inputs ",inputs) # [] # inputs["emotion_author_vector"] = # train_dataloader=DataLoader(inputs, batch_size=1 , shuffle=False) # print(train_dataloader) device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") # def tokenize_function(examples): # return tokenizer.batch_encode_plus(examples["text"], padding='max_length',max_length=PADDING_MAX_LENGTH,add_special_tokens=True,truncation=True) class EmotionAuthorGuidedDCLModel(nn.Module): def __init__(self,dcl_model:nn.Module,dropout:float=0.5): super(EmotionAuthorGuidedDCLModel, self).__init__() self.dcl_model = dcl_model self.dim = 802 self.dropout = nn.Dropout(dropout) self.linear = nn.Linear(self.dim, 1) # Freeze all layers for param in self.dcl_model.parameters(): param.requires_grad = False def forward(self,batch_tokenized): input_ids = batch_tokenized['input_ids'] attention_mask = batch_tokenized['attention_mask'] emotion_vector = batch_tokenized['emotion_author_vector'] bert_output = self.dcl_model(input_ids, attention_mask=attention_mask, output_hidden_states=True) bert_cls_hidden_state = bert_output[1] combined_vector =torch.cat((bert_cls_hidden_state,emotion_vector), 1) d_combined_vector=self.dropout(combined_vector) linear_output = self.linear(d_combined_vector) pred_linear = linear_output.squeeze(1) return pred_linear # twee checkpoint = { "model_state_dict":torch.load("./model.pt",map_location ='cpu') , } # checkpoint=load_checkpoint(run=run_dcl_study,check_point_name="model_checkpoints/") class DCLArchitecture(nn.Module): def __init__(self,dropout:float,bert_model_name:str='vinai/bertweet-base'): super(DCLArchitecture, self).__init__() self.bert = AutoModel.from_pretrained(bert_model_name) self.dim = 768 self.dense = nn.Linear(self.dim, 1) self.dropout = nn.Dropout(dropout) def forward(self,batch_tokenized, if_train=False): input_ids = batch_tokenized['input_ids'] attention_mask = batch_tokenized['attention_mask'] bert_output = self.bert(input_ids, attention_mask=attention_mask, output_hidden_states=True) bert_cls_hidden_state = bert_output[1] torch.cuda.empty_cache() if if_train: bert_cls_hidden_state_aug = self.dropout(bert_cls_hidden_state) bert_cls_hidden_state = torch.cat((bert_cls_hidden_state, bert_cls_hidden_state_aug), dim=1).reshape(-1, self.dim) else: bert_cls_hidden_state = self.dropout(bert_cls_hidden_state) linear_output = self.dense(bert_cls_hidden_state) linear_output = linear_output.squeeze(1) return bert_cls_hidden_state, linear_output # dcl_model = DCLArchitecture(bert_model_name=model_name,dropout=best_prams["DROPOUT"]) dcl_model = DCLArchitecture(bert_model_name=model_name,dropout=0.5) dcl_model.to(device) DROPOUT = 0.5 fined_tuned_bert_model=dcl_model.bert model = EmotionAuthorGuidedDCLModel(dcl_model=fined_tuned_bert_model,dropout=DROPOUT) model.to(device) model.load_state_dict(checkpoint["model_state_dict"]) # def test_loop(model, test_dataloader, device): # # collection_metric = MetricCollection( # # BinaryAccuracy(), # # MulticlassPrecision(num_classes=2,average=average), # # MulticlassRecall(num_classes=2,average=average), # # MulticlassF1Score(num_classes=2,average=average), # # BinaryConfusionMatrix() # # ) # # collection_metric.to(device) # model.eval() # print(test_dataloader) # # total_test_loss = 0.0 # for batch in test_dataloader: # print(batch) # batch = {k: v.to(device) for k, v in batch.items()} # # labels = batch["labels"] # with torch.no_grad(): # pred = model(batch) # # loss = criteon(pred, labels.float()) # pred = torch.round(torch.sigmoid(pred)) # return pred # result_metrics=test_loop(model=model, test_dataloader=train_dataloader,device=device) # print("Hate speech result",result_metrics) def predict_single_text(model, inputs,device): # Preprocess the text # inputs = tokenizer(text, return_tensors="pt", padding=True, truncation=True) inputs = {k: v.to(device) for k, v in inputs.items()} # Pass the preprocessed text through the model with torch.no_grad(): model.eval() pred = model(inputs) # Assuming your model returns a single value for prediction pred = torch.round(torch.sigmoid(pred)).item() return pred predicted_class = predict_single_text(model, inputs, device) return predicted_class # print("Hate speech result",predicted_class) #Gradio interface def greet(tweet): print("start") predicted_class = load_model(tweet) # features_list = extract_features(tweet) # print(personality_detection(tweet)) # print(str(features_list["Average_Word_Length"])) # print(calc_emotion_score(tweet)) print("end") return str(predicted_class) demo = gr.Interface(fn=greet, inputs="text", outputs="text") demo.launch()