Spaces:
Runtime error
Runtime error
| import os | |
| import time | |
| import warnings | |
| import openai | |
| import pandas as pd | |
| from dotenv import find_dotenv, load_dotenv | |
| from pandas.core.common import SettingWithCopyWarning | |
| from twitterscraper import TwitterScraper | |
| from sentence_transformers import SentenceTransformer | |
| from scipy import spatial | |
| from datetime import date, timedelta | |
| warnings.simplefilter(action="ignore", category=SettingWithCopyWarning) | |
| # Set one directory up into ROOT_PATH | |
| ROOT_PATH = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) | |
| dotenv_path = find_dotenv() | |
| load_dotenv(dotenv_path) | |
| OPENAI_API_KEY = os.environ.get("OPENAI_API_KEY") | |
| class TextClassifier: | |
| def __init__(self, model_name="text-davinci-002", from_date='2022-01-01', to_date=str(date.today()), | |
| user_list=['jimmieakesson'], | |
| num_tweets=20): | |
| """ | |
| Initializes the TextClassifier. | |
| :param model_name: name of the model from openai. | |
| :param from_date: string of the format 'YYYY-MM-DD'. | |
| :param to_date: string of the format 'YYYY-MM-DD'. | |
| :param num_tweets: integer value of the maximum number of tweets to be scraped. | |
| """ | |
| # Make sure user_name is not empty | |
| assert user_list is not None, "user_name cannot be empty" | |
| self.ts = TwitterScraper.TwitterScraper(from_date, to_date, num_tweets) | |
| self.model_name = model_name | |
| self.from_date = from_date | |
| self.to_date = to_date | |
| self.num_tweets = num_tweets | |
| self.user_name = user_list | |
| # Assure that scrape_by_user actually gets num_tweets | |
| # add timer in time-loop and stop after 10 seconds | |
| # self.df = self.ts.scrape_by_user(user_name) | |
| self.df = self.ts.scrape_by_several_users(user_list) | |
| # Check if 'id' is in self.df | |
| if 'id' in self.df.columns: | |
| # Make id as type int64 | |
| self.df.loc[:, 'id'] = self.df.id.copy().apply(lambda x: int(x)) | |
| else: | |
| # If not do nothing | |
| pass | |
| openai.api_key = OPENAI_API_KEY | |
| def classify_all(self, tweet: str): | |
| """ | |
| Classifies the topic, subtopic, sentiment and target of a user's tweets. | |
| """ | |
| import os | |
| import openai | |
| valid_tweet = len(tweet.split()) > 4 | |
| if valid_tweet: | |
| openai.api_key = os.getenv("OPENAI_API_KEY") | |
| promptstring = "Decide a Tweet's political TOPIC and SUBTOPIC, without classifying it as 'politics'. Also " \ | |
| "decide whether a political Tweet's " \ | |
| "SENTIMENT is " \ | |
| "positive, " \ | |
| "negative or neutral. Also give the TARGET of the sentiment. \nGive the answer in the form ' (" \ | |
| "TOPIC, SUBTOPIC, SENTIMENT, TARGET)'\n\nTweet: {} \nAnswer: ".format(tweet) | |
| response = openai.Completion.create( | |
| model="text-davinci-002", | |
| prompt=promptstring, | |
| temperature=0, | |
| max_tokens=30, | |
| top_p=1, | |
| frequency_penalty=0.5, | |
| presence_penalty=0 | |
| ) | |
| classification_unclean = response.choices[0]['text'] | |
| classification_clean = self.cleanup_topic_results(classification_unclean) | |
| if classification_clean.lower() == "(topic, subtopic, sentiment, target)": | |
| classification_clean = "(none, none, none, none)" | |
| else: | |
| classification_clean = "(none, none, none, none)" | |
| return classification_clean.lower() | |
| def classify_all_list(self): | |
| """ | |
| Classifies the topics of a user's tweets. | |
| """ | |
| df_topic = self.df.copy() | |
| df_topic['class_tuple'] = df_topic['tweet'].apply(self.classify_all) | |
| self.df = df_topic | |
| self.split_tuple_into_columns() | |
| return self.df | |
| def cleanup_topic_results(text): | |
| """ | |
| Cleanup response from GPT-3 to a string matching the format: "(main_topic, sub_topic, sentiment, target)" | |
| :param text: GPT-3 response | |
| :return: A string on the format: "(main_topic, sub_topic, sentiment, target)" | |
| """ | |
| new_item = text.strip() | |
| new_item = new_item.replace("\n", "") | |
| new_item = new_item.replace(" ", "") | |
| item_control = new_item.replace("(", "") | |
| item_control = item_control.replace(")", "") | |
| item_control = item_control.split(",") | |
| if ' ' or '' in item_control: | |
| item_control = [s.strip() if not (s == ' ' or s == '') else 'none' for s in | |
| item_control] # Replace empty classifications with 'none' | |
| diff = 4 - len(item_control) | |
| if diff < 0: # If response gave more than four predictions | |
| cutout = item_control[diff - 1:] # Cut out the superflous predictions | |
| item_control = item_control[:diff - 1] # Save the rest | |
| new_s = "" | |
| for i in range(len(cutout)): | |
| new_s += cutout[i] | |
| if i < -diff: | |
| new_s += " and " # Merge superflous predictions. E.g. target = 's', 'mp', 'v' -> target = 's and mp and v' | |
| item_control.append(new_s) | |
| elif diff > 0: # If response gave less than four predictions | |
| for i in range(diff): | |
| item_control.append("none") # Fill out tuple with nones | |
| new_item = str(tuple(item_control)) | |
| new_item = new_item.replace("'", "") | |
| return new_item | |
| def df_to_csv(self, filename="{}/data/twitterdata.csv".format(ROOT_PATH)): | |
| """ | |
| Writes pandas df to csv file. If it already exists, it appends. If not, it creates. It also removes duplicates. | |
| :param filename: | |
| :return: | |
| """ | |
| if not os.path.exists(filename): | |
| self.df.to_csv(filename, index=False) | |
| else: | |
| self.df.to_csv(filename, mode='a', header=False, index=False) | |
| self.remove_duplicates_from_csv(filename) | |
| def remove_duplicates_from_csv(filename="{}/data/twitterdata.csv".format(ROOT_PATH)): | |
| """ | |
| Removes duplicates from csv file. | |
| :param filename: filename of csv file | |
| :return: None | |
| """ | |
| with open(filename, 'r', encoding="utf8") as f: | |
| lines = f.readlines() | |
| with open(filename, 'w', encoding="utf8") as f: | |
| for line in lines: | |
| if line not in lines[lines.index(line) + 1:]: | |
| f.write(line) | |
| def remove_already_classified_tweets(self, filename="{}/data/twitterdata.csv".format(ROOT_PATH)): | |
| """ | |
| Removes tweets that have already been classified. | |
| :param filename: filename of csv file | |
| :return: None | |
| """ | |
| df = self.df | |
| df = df[df['sentiment'].isnull()] | |
| self.df = df | |
| self.df_to_csv(filename) | |
| def split_tuple_into_columns(self): | |
| """ | |
| Splits the topics (topic, subtopic, sentiment, target) into columns. | |
| :return: None | |
| """ | |
| df_topic = self.df.copy() | |
| df_topic['topics_temp'] = df_topic['class_tuple'].apply(lambda x: tuple(x[1:-1].split(","))) | |
| df_topic_split = pd.DataFrame(df_topic['topics_temp'].tolist(), | |
| columns=['main_topic', 'sub_topic', 'sentiment', 'target']) | |
| # Manually add columns to self.df | |
| self.df['main_topic'] = df_topic_split['main_topic'].tolist() | |
| self.df['main_topic'] = self.df['main_topic'].replace(["n/a", "", " "], "none", regex=True) | |
| self.df['main_topic'] = self.df['main_topic'].apply( | |
| lambda x: x.strip() if not (len(x) == 1 and x == "-") else "none") | |
| self.df['sub_topic'] = df_topic_split['sub_topic'].tolist() | |
| # In a few of the outputs from GPT-3 the sub_topic = "sentiment" | |
| self.df['sub_topic'] = self.df['sub_topic'].replace(["n/a", "sentiment", "", " "], "none", regex=True) | |
| self.df['sub_topic'] = self.df['sub_topic'].apply( | |
| lambda x: x.strip() if not (len(x) == 1 and x == "-") else "none") | |
| self.df['sentiment'] = df_topic_split['sentiment'].tolist() | |
| self.df['sentiment'] = self.df['sentiment'].replace(["n/a", "sentiment", "", " "], "none", regex=True) | |
| self.df['sentiment'] = self.df['sentiment'].apply( | |
| lambda x: x.strip() if not (len(x) == 1 and x == "-") else "none") | |
| self.df['target'] = df_topic_split['target'].tolist() | |
| self.df['target'] = self.df['target'].replace(["n/a", "", " "], "none", regex=True) | |
| self.df['target'] = self.df['target'].apply(lambda x: x.strip() if not (len(x) == 1 and x == "-") else "none") | |
| self.df.fillna('none', inplace=True) | |
| def get_dataframe(self): | |
| """ | |
| Returns the dataframe. | |
| :return: dataframe | |
| """ | |
| return self.df | |
| def __repr__(self): | |
| """ | |
| Gives a string that describes which user is classified | |
| :return: | |
| """ | |
| return "Classifier for user: " + self.user_name + " with model: " + self.model_name + "." | |
| def get_database(self, filename="{}/data/twitterdata.csv".format(ROOT_PATH)): | |
| """ | |
| Returns the database containing all dataframes. | |
| :param filename: filename of csv file | |
| :return: | |
| """ | |
| db = pd.read_csv(filename) | |
| return db | |
| def cleanup_list(self, uncleaned_list): | |
| """ | |
| Cleans up faulty predictions. | |
| :param uncleaned_list: the list to be cleaned | |
| :return: cleaned list | |
| """ | |
| uncleaned_list = [s if not isinstance(s, float) else "none" for s in uncleaned_list] | |
| uncleaned_list = [s if not len(s.split()) > 5 else "none" for s in uncleaned_list] | |
| uncleaned_list = [s if not "swedish" in s else s.replace("swedish", " ") for s in uncleaned_list] | |
| uncleaned_list = [s if not "politics" in s else s.replace("politics", "none") for s in uncleaned_list] | |
| uncleaned_list = [s.replace(" ", " ") for s in uncleaned_list] | |
| cleaned_list = [s.strip() for s in uncleaned_list] | |
| return cleaned_list | |
| def merge_lists(self, main_topic_list, sub_topic_list): | |
| """ | |
| Merges the topic lists. If either topic is a faulty classification, only the non-faulty topic wil be used. | |
| If both are faulty, the merged topic will be labeled as faulty (ERROR_496). | |
| :param main_topic_list: A list containing main topics | |
| :param sub_topic_list: A list containing sub topics | |
| :return: A list containing string items on the form "main_topic and sub_topic" | |
| """ | |
| new_list = [] | |
| main_topic_list = self.clean_party_names(main_topic_list) | |
| sub_topic_list = self.clean_party_names(sub_topic_list) | |
| for i in range(len(main_topic_list)): | |
| if main_topic_list[i].lower() == "none" and sub_topic_list[ | |
| i].lower() == "none": # If the predictions are faulty | |
| new_list.append("ERROR_496") # Label as ERROR_496 (faulty prediction) | |
| elif main_topic_list[i].lower() == "none": | |
| new_list.append(sub_topic_list[i]) | |
| elif sub_topic_list[i].lower() == "none": | |
| new_list.append(main_topic_list[i]) | |
| else: | |
| new_list.append(main_topic_list[i] + " and " + sub_topic_list[i]) | |
| return new_list | |
| def file_to_mat(self, classification_type): | |
| """ | |
| Converts a synonym textfile to a matrix in which the rows contain a general topic/target and its related words. | |
| :param classification_type: The type of classification: topic or target | |
| :return: a matrix in which the first element of each row is a general topic/target, and the rest are words related to | |
| the topic | |
| """ | |
| filename = "{}/data/".format(ROOT_PATH) | |
| filename += classification_type + "_synonyms.txt" | |
| with open(filename, encoding='utf-8') as f: | |
| lines = f.read() | |
| lines = lines.split("\n") | |
| topic_list = [] | |
| temp_list = [] | |
| for topic in lines: | |
| if not topic.endswith("####"): | |
| temp_list.append(topic) | |
| else: | |
| temp_list.append(topic[:-4]) # Remove the marker (####) | |
| topic_list.append(temp_list) | |
| temp_list = [] | |
| return topic_list | |
| def mat_to_list(self, mat): | |
| """ | |
| Converts a matrix from file_to_mat() into one list containing all topics and synonyms, and one list with | |
| mappings for the synonyms. | |
| :param mat: a matrix from file_to_mat() | |
| :return: | |
| """ | |
| full_list = [] | |
| mapped_synonyms = [] | |
| for syns in mat: | |
| for topic in syns: | |
| full_list.append(topic) | |
| mapped_synonyms.append(syns[0]) | |
| return full_list, mapped_synonyms | |
| def clean_party_names(self, old_topic_list): | |
| """ | |
| Encodes all party names to sentences that will yield a high cosine similarity value when merged with another | |
| topic, without taking the actual party name into account. These sentences have deliberately been composed such | |
| that they pose a low risk of being close (in the sentence embedding-space) to any possible merged topic or | |
| target that may be encountered. | |
| :param old_topic_list: list of topics | |
| :return: list of encoded topics | |
| """ | |
| # Problem 1: When a party name is encountered, we want to bias the merging towards that party since the | |
| # occurrence of a very general main topic (as in the example below) plus a party name as subtopic is frequent. | |
| # Example: main_topic = "politics", sub_topic = "sweden democrats" -> | |
| # combined_topics = "politics and sweden democrats" | |
| # Problem 2: The party names themselves are biased towards certain topics/targets and lead to faulty merges. | |
| # Example: Variations of words such as "Sweden" as the target/topic will be biased towards getting merged with | |
| # "Sweden Democrats". | |
| # Solution: Encode party names with sentences that are HIGHLY unlikely to be close to anything in the embedding | |
| # space and thus enforcing a strong bias in the cosine similarity computation towards the party if encountered. | |
| party_names = {} | |
| party_names["m"] = "parrot computer is swimming as screen time" | |
| party_names["moderaterna"] = "parrot computer is swimming as screen time" | |
| party_names["moderates"] = "parrot computer is swimming as screen time" | |
| party_names["the moderates"] = "parrot computer is swimming as screen time" | |
| party_names["moderate party"] = "parrot computer is swimming as screen time" | |
| party_names["the moderate party"] = "parrot computer is swimming as screen time" | |
| party_names["the moderaterna party"] = "parrot computer is swimming as screen time" | |
| party_names["sd"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["sverigedemokraterna"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["sweden democrats"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["the sweden democrats"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["the swedish democrats"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["swedish democrats"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["@jimmieakesson"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["l"] = "red weather jokes with music and the mathematician" | |
| party_names["liberalerna"] = "red weather jokes with music and the mathematician" | |
| party_names["liberals"] = "red weather jokes with music and the mathematician" | |
| party_names["the liberals"] = "red weather jokes with music and the mathematician" | |
| party_names["the liberal party"] = "red weather jokes with music and the mathematician" | |
| party_names["liberal people's party"] = "red weather jokes with music and the mathematician" | |
| party_names["@johanpehrson"] = "red weather jokes with music and the mathematician" | |
| party_names["mp"] = "ice piano flies with pencil as direction" | |
| party_names["miljöpartiet"] = "ice piano flies with pencil as direction" | |
| party_names["de gröna"] = "ice piano flies with pencil as direction" | |
| party_names["green party"] = "ice piano flies with pencil as direction" | |
| party_names["the green party"] = "ice piano flies with pencil as direction" | |
| party_names["miljopartiet"] = "ice piano flies with pencil as direction" | |
| party_names["@bolund"] = "ice piano flies with pencil as direction" | |
| party_names["@martastenevi"] = "ice piano flies with pencil as direction" | |
| party_names["s"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["socialdemokraterna"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["social democratic party"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["the social democratic party"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["social democrats"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["the social democrats"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["sosse"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["sossen"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["sossar"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["sossarna"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["sossarnas"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["swedish social democrats"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["@swedishpm"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["v"] = "rooftop cats play physics with cardboard fire" | |
| party_names["vänsterpartiet"] = "rooftop cats play physics with cardboard fire" | |
| party_names["left party"] = "rooftop cats play physics with cardboard fire" | |
| party_names["the left party"] = "rooftop cats play physics with cardboard fire" | |
| party_names["@dadgostarnooshi"] = "rooftop cats play physics with cardboard fire" | |
| party_names["c"] = "differential donuts program sunny waters" | |
| party_names["centerpartiet"] = "differential donuts program sunny waters" | |
| party_names["center party"] = "differential donuts program sunny waters" | |
| party_names["centre party"] = "differential donuts program sunny waters" | |
| party_names["the center party"] = "differential donuts program sunny waters" | |
| party_names["@annieloof"] = "differential donuts program sunny waters" | |
| party_names["kd"] = "cauchy-riemann met sunglasses after rolling yellow" | |
| party_names["kristdemokraterna"] = "cauchy-riemann met sunglasses after rolling yellow" | |
| party_names["christian democrats"] = "cauchy-riemann met sunglasses after rolling yellow" | |
| party_names["the christian democrats"] = "cauchy-riemann met sunglasses after rolling yellow" | |
| party_names["@buschebba"] = "cauchy-riemann met sunglasses after rolling yellow" | |
| for i, topic in enumerate(old_topic_list): | |
| topic = topic.lower() | |
| topic = topic.replace(" ", " ") | |
| topic = topic.strip() | |
| if topic in party_names: | |
| old_topic_list[i] = party_names.get(topic) | |
| return old_topic_list | |
| def reset_party_names(self, old_topic_list): | |
| """ | |
| Decodes the encoded party names. | |
| :param old_topic_list: list of topics | |
| :return: list of encoded topics | |
| """ | |
| party_names = {} | |
| party_names["m"] = "parrot computer is swimming as screen time" | |
| party_names["sd"] = "keyboard can hire the yellow elephant in cosmos" | |
| party_names["l"] = "red weather jokes with music and the mathematician" | |
| party_names["mp"] = "ice piano flies with pencil as direction" | |
| party_names["s"] = "lamp of fire walks bird gladly tomorrow" | |
| party_names["v"] = "rooftop cats play physics with cardboard fire" | |
| party_names["c"] = "differential donuts program sunny waters" | |
| party_names["kd"] = "cauchy-riemann met sunglasses after rolling yellow" | |
| inverted_dict = {} | |
| # Invert dictionary | |
| for k, v in party_names.items(): | |
| if v not in inverted_dict: | |
| inverted_dict[v] = k | |
| # Update values in old_topic_list | |
| for i, topic in enumerate(old_topic_list): | |
| if topic in inverted_dict.keys(): | |
| old_topic_list[i] = inverted_dict.get(topic) | |
| return old_topic_list | |
| def merge_classifications(self, old_list, classification_type): | |
| """ | |
| Merges topics/targets from GPT-3 according to a list of predefined topics/targets. | |
| :param old_list: list of the topics/targets to be merged | |
| :param classification_type: type of classifications: topic or target | |
| :return: list of new topics/targets | |
| """ | |
| # Get the tuple of lists containing all synonyms and general topics/targets | |
| tup_list = self.mat_to_list(self.file_to_mat(classification_type)) | |
| # Save list of synonyms | |
| synonym_list = tup_list[0] | |
| # Save list of mappings between synonym and general topic/target | |
| synonym_mappings = tup_list[1] | |
| # Load embedding model-names | |
| model_list = ['sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2', 'all-MiniLM-L6-v2'] | |
| result_dict = {} | |
| # Encode party names | |
| old_list = self.clean_party_names(old_list) | |
| for model_name in model_list: | |
| model = SentenceTransformer(model_name) | |
| # Encode the topics/targets with the sentence transformer model | |
| old_list_embeddings = model.encode(old_list, batch_size=64, show_progress_bar=True, | |
| convert_to_tensor=True) | |
| # Encode the synonyms with the sentence transformer model | |
| synonym_list_embeddings = model.encode(synonym_list, batch_size=64, show_progress_bar=True, | |
| convert_to_tensor=True) | |
| for i, embedded_classification in enumerate(old_list_embeddings): | |
| result_list = [] | |
| for embedded_synonyms in synonym_list_embeddings: | |
| # Compute the cosine similarity between every classification and synonym | |
| result = 1 - spatial.distance.cosine(embedded_classification, embedded_synonyms) | |
| result_list.append(result) | |
| max_value = max(result_list) | |
| max_index = result_list.index(max_value) | |
| old_classification = old_list[i] | |
| # Extract the general topic/target | |
| new_classification = synonym_mappings[max_index] | |
| # Save the topic/target that yielded the highest cosine similarity value | |
| if old_classification not in result_dict: | |
| result_dict[old_classification] = [(new_classification, max_value, synonym_list[max_index])] | |
| # When we have found the best topics/targets after using the first transformer model | |
| else: | |
| # Append the results from the next model | |
| result_dict[old_classification].append((new_classification, max_value, synonym_list[max_index])) | |
| new_dict = {} | |
| # Time to replace the old values with the new ones | |
| for old_values in result_dict: | |
| tup_list = result_dict[old_values] | |
| max_tup = max(tup_list, key=lambda item: item[1]) | |
| if classification_type == "topic": | |
| limit = 0.4 | |
| else: | |
| limit = 0.75 | |
| # Discard classification if the old topic/target is not similar to anything in our synonym lists | |
| if max_tup[1] < limit: | |
| max_tup = ("ERROR_9000", "{:.2f}".format(round(max_tup[1], 2)), "none") | |
| else: | |
| max_tup = (max_tup[0], "{:.2f}".format(round(max_tup[1], 2)), max_tup[2]) | |
| new_classification = max_tup | |
| if old_values not in new_dict: | |
| new_dict[old_values] = new_classification | |
| new_list = [] | |
| for old_value in old_list: | |
| new_list.append(new_dict[old_value]) | |
| return new_list | |
| def merge_all(self): | |
| """ | |
| Merges main+subtopics, targets, and updates the dataframe. | |
| :param df: | |
| :return: | |
| """ | |
| df_topics = self.df.copy() | |
| sub_topics = df_topics['sub_topic'] | |
| sub_topics = sub_topics.tolist() | |
| sub_topics = self.cleanup_list(sub_topics) | |
| main_topics = df_topics['main_topic'] | |
| main_topics = main_topics.tolist() | |
| main_topics = self.cleanup_list(main_topics) | |
| merged_topic_list = self.merge_lists(main_topics, sub_topics) | |
| targets = df_topics['target'] | |
| targets = targets.tolist() | |
| targets = self.cleanup_list(targets) | |
| merged_topics = self.merge_classifications(merged_topic_list, "topic") | |
| merged_targets = self.merge_classifications(targets, "target") | |
| print("The following merges were made: ") | |
| for i, top in enumerate(merged_topic_list): | |
| print("TOPICS: ", top, " -> ", merged_topics[i]) | |
| t_list = [] | |
| for i in range(len(merged_topics)): | |
| t_list.append(tuple(merged_topics[i]) + tuple(merged_targets[i])) | |
| merged_tuples = t_list | |
| df_topics['merged_tuple'] = merged_tuples | |
| df = self.split_merged_tuple_into_columns(df_topics) | |
| print("Merging finished...") | |
| self.df = df | |
| def split_merged_tuple_into_columns(self, df): | |
| """ | |
| Splits the merged tuple (merged topic, merged target) into columns. | |
| :return: None | |
| """ | |
| df_topic = df.copy() | |
| df_topic_split = pd.DataFrame(df_topic['merged_tuple'].tolist(), | |
| columns=['merged_topic', 'cos_sim_topic', 'synonym_topic', 'merged_target', | |
| 'cos_sim_target', 'synonym_target']) | |
| self.df['merged_tuple'] = df_topic['merged_tuple'].tolist() | |
| # Manually add columns to self.df | |
| self.df['merged_topic'] = df_topic_split['merged_topic'].tolist() | |
| self.df['cos_sim_topic'] = df_topic_split['cos_sim_topic'].tolist() | |
| self.df['synonym_topic'] = self.reset_party_names(df_topic_split['synonym_topic'].tolist()) | |
| self.df['merged_target'] = df_topic_split['merged_target'].tolist() | |
| self.df['cos_sim_target'] = df_topic_split['cos_sim_target'].tolist() | |
| self.df['synonym_target'] = self.reset_party_names(df_topic_split['synonym_target'].tolist()) | |
| return self.df | |
| def run_main_pipeline(self, filename="{}/data/twitterdata.csv".format(ROOT_PATH)): | |
| """ | |
| Classifies the topics/sentiments of a user's tweets. | |
| #We presume that all tweets inside the twitterdata.csv file are already classified. | |
| :return: None | |
| """ | |
| # Check if file exists, if not, create it | |
| if os.path.exists(filename): | |
| # Fetch tweets from csv file | |
| already_classified_df = pd.read_csv(filename, on_bad_lines='skip') | |
| print("Already classified tweets: {}".format(already_classified_df.shape[0])) | |
| # Create a temporary df where values from already_classified_df that are not it self.df are stored | |
| temp_df = already_classified_df[already_classified_df['id'].isin(self.df['id'])] | |
| # Remove rows from self.df that are not in already_classified_df | |
| self.df = self.df[~self.df['id'].isin(already_classified_df['id'])] | |
| # Only classify non-empty rows | |
| if self.df.shape[0] > 0: | |
| time.sleep(10) | |
| print("Classifying topic, subtopic, sentiment and target of {} tweets...".format(self.df.shape[0])) | |
| self.df = self.classify_all_list() | |
| self.df = self.df.replace({'': 'none'}, regex=True) | |
| self.df = self.df.replace({' ': 'none'}, regex=True) | |
| print("Merging topics...") | |
| self.merge_all() | |
| print("Writing to csv...") | |
| self.df_to_csv(filename) | |
| # Concatenate temp_df and self.df | |
| self.df = pd.concat([temp_df, self.df], ignore_index=True) | |
| print("Appended {}.".format(filename)) | |
| return None | |
| else: | |
| self.df = pd.concat([temp_df, self.df], ignore_index=True) | |
| print("No new tweets to classify.") | |
| return None | |
| else: | |
| print("No csv file found. Continuing without removing already classified tweets.") | |
| print("Classifying topic, subtopic, sentiment and target of {} tweets...".format(self.df.shape[0])) | |
| self.df = self.classify_all_list() | |
| self.df = self.df.replace({'': 'none'}, regex=True) | |
| self.df = self.df.replace({' ': 'none'}, regex=True) | |
| print("Merging topics...") | |
| self.merge_all() | |
| print("Writing to csv file...") | |
| self.df_to_csv(filename) | |
| print("Created {}.".format(filename)) | |
| return None | |
| if __name__ == "__main__": | |
| # $6.39 @ 3431 tweets | |
| # $18.00 @ 4608 tweets | |
| # $11.61 to classify 1177 tweets ~ $0.01 / tweet | |
| # This code snippet allows for scraping and classifying by simply specifying a start and end date. | |
| USER_LIST = ['jimmieakesson', 'BuschEbba', 'annieloof', 'JohanPehrson', 'bolund', 'martastenevi', 'SwedishPM', | |
| 'dadgostarnooshi'] | |
| start_date = date(2022, 8, 4) | |
| end_date = date(2022, 8, 4) | |
| delta = timedelta(days=1) | |
| while start_date <= end_date: | |
| from_date = start_date.strftime("%Y-%m-%d") | |
| start_date += delta | |
| to_date = start_date.strftime("%Y-%m-%d") | |
| print("curr_date: ", from_date) | |
| tc = TextClassifier(from_date=from_date, to_date=to_date, user_list=USER_LIST, num_tweets=6000) | |
| tc.run_main_pipeline() | |