app.py

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)