| import pandas as pd | |
| from gensim import corpora | |
| from gensim import similarities | |
| from gensim.models import TfidfModel | |
| from gensim.parsing import strip_tags, strip_numeric, \ | |
| strip_multiple_whitespaces, stem_text, strip_punctuation, \ | |
| remove_stopwords, preprocess_string | |
| import re | |
| from typing import List | |
| from utils.constants import TEST_INPUTS | |
| import argparse | |
| from random import choice | |
| transform_to_lower = lambda s: s.lower() | |
| remove_single_char = lambda s: re.sub(r'\s+\w{1}\s+', '', s) | |
| class PaperRecommender: | |
| def __init__(self, | |
| num_samples=3000, | |
| corpus_dictionary_path="30Ktokens", | |
| arxiv_dataset_path="/Users/luis.morales/Desktop/arxiv-paper-recommender/data/processed/reduced_arxiv_papers.parquet.gzip", | |
| save_dict=False, | |
| query=""): | |
| self.num_samples = num_samples | |
| self.corpus_dictionary_path = corpus_dictionary_path | |
| self.arxiv_dataset_path = arxiv_dataset_path | |
| self.save_dict = save_dict | |
| self.query = query | |
| self.cleaning_filters = [ | |
| strip_tags, | |
| strip_numeric, | |
| strip_punctuation, | |
| strip_multiple_whitespaces, | |
| transform_to_lower, | |
| remove_stopwords, | |
| remove_single_char | |
| ] | |
| self.dictionary = None | |
| self.index = None | |
| self.tfidf_model = None | |
| self.df = None | |
| def gensim_tokenizer(self, docs: List[str]): | |
| tokenized_docs = list() | |
| for doc in docs: | |
| processed_words = preprocess_string(doc, self.cleaning_filters) | |
| tokenized_docs.append(processed_words) | |
| return tokenized_docs | |
| def cleaning_pipe(self, document): | |
| processed_words = preprocess_string(document, self.cleaning_filters) | |
| return processed_words | |
| def get_gensim_dictionary(self, tokenized_docs: List[str], dict_name: str = "corpus"): | |
| dictionary = corpora.Dictionary(tokenized_docs) | |
| if self.save_dict: | |
| parent_folder = "/Users/luis.morales/Desktop/arxiv-paper-recommender/models/nlp_dictionaries" | |
| dictionary.save(f'{parent_folder}/{dict_name}.dict') | |
| return dictionary | |
| def get_closest_n(self, query: str, n: int): | |
| query_document = self.cleaning_pipe(query) | |
| query_bow = self.dictionary.doc2bow(query_document) | |
| sims = self.index[self.tfidf_model[query_bow]] | |
| top_idx = sims.argsort()[-1 * n:][::-1] | |
| return top_idx | |
| def get_recommendations_metadata(self, query: str, n: int): | |
| recommendations_idxs = self.get_closest_n(query, n) | |
| recommendations_metadata = self.df.iloc[recommendations_idxs] | |
| recommendations_metadata = recommendations_metadata.reset_index(drop=True) | |
| return recommendations_metadata | |
| def run_recommender(self): | |
| if self.num_samples is None: | |
| self.df = pd.read_parquet(self.arxiv_dataset_path) | |
| self.df = pd.read_parquet(self.arxiv_dataset_path).sample(self.num_samples).reset_index(drop=True) | |
| corpus = self.df['cleaned_abstracts'].to_list() | |
| tokenized_corpus = self.gensim_tokenizer(corpus) | |
| self.dictionary = self.get_gensim_dictionary(tokenized_docs=tokenized_corpus, dict_name=self.corpus_dictionary_path) | |
| BoW_corpus = [self.dictionary.doc2bow(doc, allow_update=True) for doc in tokenized_corpus] | |
| self.tfidf_model = TfidfModel(BoW_corpus) | |
| self.index = similarities.SparseMatrixSimilarity(self.tfidf_model[BoW_corpus], num_features=len(self.dictionary)) | |
| if self.query is None: | |
| self.query = choice(TEST_INPUTS) | |
| return self.results | |