import pymongo import os import pandas as pd import numpy as np import cloudpickle import logging from datetime import datetime, timedelta import re from pymongo import MongoClient from flask import Flask, Response from flask_cors import cross_origin, CORS os.environ["TOKENIZERS_PARALLELISM"] = "false" DB_URL = os.getenv('DB_URL') DB_NAME = "indian_news_predictions_db" COLL_NAME = "indian_news_predictions_coll" AUTO_VALIDATE_PROB_THRESHOLD = 0.7 # DB_NAME = "TEST_indian_news_predictions_db" # COLL_NAME = "TEST_indian_news_predictions_coll" keywords = {'BUSINESS': ['markets', 'stocks', 'startup', 'wealth', 'corporate', 'personal-finance', 'small-biz', 'nri/invest', '/funding', 'market-news'], 'ASTROLOGY': ['astrology', 'zodiac', 'horoscope'], 'EDUCATION': ['education', 'jobs/exams-results'], 'ENTERTAINMENT': ['entertainment/', 'web-series', 'tv/', 'box-office', 'bollywood', 'hollywood', '/movies', '/music', 'celeb-style', '/wwe', 'life-style/fashion'], 'HEALTH': ['health-fitness', 'health'], 'SPORTS': ['sports/', 'sport/', '/sports', '/sport', 'cricket', 'tennis', 'football'], 'SCIENCE': ['science', 'home/science'], 'NATION': ['city/', 'india-news', 'india', 'news/national', 'lok-sabha', 'politics-and-nation', '/cities/', 'assembly-elections', 'delhi-news', 'karnataka-today'], 'TECHNOLOGY': ['technology/', '\technology', '/electronics', 'tech-tips', 'gadgets-news', 'laptops', 'mobiles'], 'WORLD': ['world-news', '/world', 'world/', 'us-canada-news', 'nri/other-news', 'newsletter-international', 'nri/migrate', 'news/international']} def load_model(): logging.warning('Entering load transformer') with open("models/label_encoder.bin", "rb") as model_file_obj: label_encoder = cloudpickle.load(model_file_obj) with open("models/calibrated_model.bin", "rb") as model_file_obj: calibrated_model = cloudpickle.load(model_file_obj) tflite_model_path = os.path.join("models", "model.tflite") calibrated_model.estimator.tflite_model_path = tflite_model_path logging.warning('Exiting load transformer') return calibrated_model, label_encoder calibrated_model, label_encoder = load_model() app = Flask(__name__) CORS(app) class DBOperations: def __init__(self, n_days_trailing: int=10): self.url = DB_URL self.database = DB_NAME self.collection = COLL_NAME self.__client = None self.n_days_trailing = n_days_trailing self.__error = 0 def __connect(self): try: self.__client = pymongo.MongoClient(self.url) _ = self.__client.list_database_names() except Exception as conn_exception: self.__error = 1 self.__client = None raise def __read(self): try: db = self.__client[self.database] coll = db[self.collection] results = coll.find() docs = [] for doc in results: docs.append(doc) rss_df = pd.DataFrame(docs) except Exception as insert_err: self.__error = 1 rss_df = None raise return rss_df def __close_connection(self): if self.__client is not None: self.__client.close() self.__client = None def read_from_db(self): rss_df = None if self.url is not None: if self.__error == 0: self.__connect() if self.__error == 0: rss_df = self.__read() if self.__error == 0: logging.warning("Read Successful") if self.__client is not None: self.__close_connection() return rss_df def find_path(url): if url == '': return '' url = url.replace("-/-", "-") url_split = url.replace("https://", "") url_split = url_split.replace("www.", "") url_split = url_split.strip() url = url.replace("//", "/") url = url.replace("https/timesofindia-indiatimes-com", "") url_split = url_split.split("/") url_split = [u for u in url_split if (u != "") and (u != "articleshow") and (u.find(".cms")==-1) and (u.find(".ece")==-1) and (u.find(".htm")==-1) and (len(u.split('-')) <= 5) and (u.find(" ") == -1) ] if len(url_split) > 2: url_split = "/".join(url_split[1:]) else: if len(url_split) > 0: url_split = url_split[-1] else: url_split = '-' return url_split def find_valid_paths_of_class(df, class_, keywords): paths_for_particular_class = [*df.loc[df['y_true']==class_, 'path'].unique()] valid_paths_for_particular_class = [i for i in paths_for_particular_class if len(re.findall(rf'{"|".join(keywords[class_])}', i))>0] return valid_paths_for_particular_class def find_valid_examples(df, class_, keywords): df = df.copy() valid_keywords = find_valid_paths_of_class(df, class_, keywords) return df.loc[(df['path'].isin(valid_keywords)) & (df['y_true']==class_) & (df['y_true_proba']!=1) & (df['y_true_proba']>AUTO_VALIDATE_PROB_THRESHOLD), ['url', 'text', 'path', 'y_true', 'y_true_proba']] def connect_db(): try: client = MongoClient(DB_URL) db = client[DB_NAME] collection = db[COLL_NAME] logging.warning('Successfully connected to DB') except Exception as e: client = None db = None collection = None logging.warning(f"Couldn't connect to DB: {e}") raise return client, db, collection def close_db_connection(client): if client is not None: client.close() client = None logging.warning('Successfully closed DB connection') def update_valid_examples_in_db(df, class_, keywords): logging.warning('Entering update_valid_examples_in_db()') client = None collection = None try: new_data = {'y_true_proba': 1, 'allocated': 1} valid_df = find_valid_examples(df, class_, keywords) valid_urls = [*valid_df['url']] client, db, collection = connect_db() if collection is not None: for valid_url in valid_urls: collection.update_one({'url': valid_url}, {'$set': new_data}) logging.warning('Successfully updated in DB') close_db_connection(client) logging.warning('Exiting update_valid_examples_in_db()') except Exception as e: close_db_connection(client) logging.warning(f'Exiting update_valid_examples_in_db(): {e}') raise def parse_prediction(tflite_pred, label_encoder): tflite_pred_argmax = np.argmax(tflite_pred, axis=1) tflite_pred_label = label_encoder.inverse_transform(tflite_pred_argmax) tflite_pred_prob = np.max(tflite_pred, axis=1) return tflite_pred_label, tflite_pred_prob def inference(text, calibrated_model, label_encoder): logging.warning('Entering inference()') logging.warning(f'Samples to predict: {len(text)}') if text != "": tflite_pred = calibrated_model.predict_proba(text) tflite_pred = parse_prediction(tflite_pred, label_encoder) logging.warning('Exiting inference()') return tflite_pred def get_predictions(df, calibrated_model, label_encoder): df = df.copy() urls = df['url'].to_list() texts = df['text'].to_list() paths = df['path'].to_list() headlines_desc = [f"{p}: {t}" for p, t in zip(paths, texts)] label, prob = inference(headlines_desc, calibrated_model, label_encoder) preds = [[u, l, p] for u, l, p in zip(urls, label, prob)] return preds def update_predictions_in_db(preds): logging.warning('Entering update_predictions_in_db()') client = None collection = None try: client, db, collection = connect_db() if collection is not None: for url, label, prob in preds: new_data = {'y_true_proba': prob, 'y_true': label, 'updated_prediction': 1} collection.update_many({'url': url}, {'$set': new_data}, upsert=False) logging.warning('Successfully updated predictions in DB') close_db_connection(client) logging.warning('Exiting update_predictions_in_db()') except Exception as e: close_db_connection(client) logging.warning(f'Exiting update_predictions_in_db(): {e}') raise def auto_validate(calibrated_model, label_encoder): classes = ['ASTROLOGY', 'BUSINESS', 'EDUCATION', 'ENTERTAINMENT', 'HEALTH', 'SPORTS', 'SCIENCE', 'NATION', 'TECHNOLOGY', 'WORLD'] db = DBOperations() df=db.read_from_db() df.dropna(subset='url', inplace=True) df['path'] = df['url'].map(find_path) if 'updated_prediction' in [*df.columns]: df['updated_prediction'] = df['updated_prediction'].fillna(0) else: df['updated_prediction'] = 0 to_update_pred_df = df.loc[((df['y_true_proba'] != 1) & (df['updated_prediction'] == 0)), ['url', 'text', 'path',]].copy() if len(to_update_pred_df) > 0: preds = get_predictions(to_update_pred_df, calibrated_model, label_encoder) update_predictions_in_db(preds) else: logging.warning("All the predictions are updated. Nothing to update") for i in range(len(classes)): CLASS_TO_UPDATE = classes[i] logging.warning(f"Updating: {CLASS_TO_UPDATE}") update_valid_examples_in_db(df, CLASS_TO_UPDATE, keywords) logging.warning(f"Updated: {CLASS_TO_UPDATE}") logging.warning('Auto-validation Successfully Completed') @app.route("/") @cross_origin() def main(): logging.warning('Entering Application') status_json = "{'status':'success'}" status_code = 200 try: auto_validate(calibrated_model, label_encoder) logging.warning('Exiting Application') except Exception as e: logging.warning(f'Error in Application: {e}') status_json = "{'status':'failure'}" status_code = 500 return Response(status_json, status=status_code, mimetype='application/json') if __name__ == "__main__": app.run(host="0.0.0.0", port=7860, timeout=12000, workers=1, threads=1)