Upload 13 files

README.md

@@ -1,6 +1,6 @@
 ---
 python_version: '3.11.5'
-title: newsdb
+title: latest_news_backend_with_cat_pred
 emoji: 🔥
 colorFrom: red
 colorTo: red

app.py

@@ -1,25 +1,46 @@
 from news_extractor import get_news
-from db_operations import DBOperations
+from db_operations.db_write import DBWrite
+from db_operations.db_read import DBRead
+from news_category_prediction import predict_news_category
 import json
 from flask import Flask, Response
 from flask_cors import cross_origin, CORS
 import logging
+import tensorflow as tf 
+import cloudpickle
+from transformers import DistilBertTokenizerFast
+import os
 
 app = Flask(__name__)
 CORS(app)
 logging.warning('Initiated')
 
 
+def load_model():
+    interpreter = tf.lite.Interpreter(model_path=os.path.join("models/news_classification_hf_distilbert.tflite"))
+    with open("models/news_classification_labelencoder.bin", "rb") as model_file_obj:
+        label_encoder = cloudpickle.load(model_file_obj)
+        
+    model_checkpoint = "distilbert-base-uncased"
+    tokenizer = DistilBertTokenizerFast.from_pretrained(model_checkpoint)
+    return interpreter, label_encoder, tokenizer
+
+interpreter, label_encoder, tokenizer = load_model()
+
+
 @app.route("/")
 @cross_origin()
 def update_news():
     status_json = "{'status':'success'}"
     status_code = 200
     try:
-        db = DBOperations()
-        news_df = get_news()
+        db_read = DBRead()
+        db_write = DBWrite()
+        old_news = db_read.read_news_from_db()
+        new_news = get_news()
+        news_df = predict_news_category(old_news, new_news, interpreter, label_encoder, tokenizer)
         news_json = [*json.loads(news_df.reset_index(drop=True).to_json(orient="index")).values()]
-        db.insert_news_into_db(news_json)
+        db_write.insert_news_into_db(news_json)
     except:
         status_json = "{'status':'failure'}"
         status_code = 500

db_operations/db_read.py

@@ -0,0 +1,59 @@
+import pymongo
+import os
+import pandas as pd
+
+
+class DBRead:
+    """
+    Reads news from MongoDB
+    """
+    def __init__(self):
+        self.url = os.getenv('DB_URL')
+        self.database = "rss_news_db"
+        self.collection = "rss_news"
+        self.__client = None
+        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]
+            docs = []
+            for doc in coll.find():
+                docs.append(doc)
+            rss_df = pd.DataFrame(docs)
+        except Exception as insert_err:
+            self.__error = 1
+            rss_df = pd.DataFrame({'_id': '', 'title': '', 'url': '',
+                                   'description': '', 'parsed_date': '',
+                                   'src': ''}, index=[0])
+        return rss_df
+
+    def __close_connection(self):
+        if self.__client is not None:
+            self.__client.close()
+            self.__client = None
+
+    def read_news_from_db(self):
+        rss_df = pd.DataFrame({'_id': '', 'title': '', 'url': '',
+                               'description': '', 'parsed_date': '',
+                               'src': ''}, index=[0])
+        if self.url is not None:
+            if self.__error == 0:
+                self.__connect()
+            if self.__error == 0:
+                rss_df = self.__read()
+            if self.__error == 0:
+                print("Read Successful")
+            if self.__client is not None:
+                self.__close_connection()
+        return rss_df

db_operations/db_write.py

@@ -0,0 +1,52 @@
+import pymongo
+import os
+
+
+class DBWrite:
+    """
+    Inserts processed news into MongoDB
+    """
+    def __init__(self):
+        self.url = os.getenv('DB_URL')
+        self.database = "rss_news_db"
+        self.collection = "rss_news"
+        self.__client = None
+        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.__close_connection()
+            self.__client = None
+            raise
+
+    def __insert(self, documents):
+        try:
+
+            db = self.__client[self.database]
+            coll = db[self.collection]
+            coll.drop()
+            coll.insert_many(documents=documents)
+        except Exception as insert_err:
+            self.__error = 1
+            self.__close_connection()
+            raise
+
+    def __close_connection(self):
+        if self.__client is not None:
+            self.__client.close()
+            self.__client = None
+
+    def insert_news_into_db(self, documents: list):
+        if self.url is not None:
+            if self.__error == 0:
+                self.__connect()
+            if self.__error == 0:
+                self.__insert(documents=documents)
+            if self.__error == 0:
+                print("Insertion Successful")
+            if self.__client is not None:
+                self.__close_connection()

models/news_classification_hf_distilbert.tflite

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:283b191892f95889a79e18f4362f207617e56b5c9f93160b61be7db1c480938e
+size 66788520

models/news_classification_labelencoder.bin

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:65ddceef60d9f1dc95d70a1940c5b382bb58d47ebb7145bf32e887f62e054535
+size 327

news_category_prediction.py

@@ -0,0 +1,71 @@
+import pandas as pd
+import numpy as np
+import tensorflow as tf 
+
+
+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, interpreter, label_encoder, tokenizer):
+    batch_size = len(text)
+    MAX_LEN = 80
+    N_CLASSES = 8
+    if text != "":
+        tokens = tokenizer(text, max_length=MAX_LEN, padding="max_length", truncation=True, return_tensors="tf")
+        # tflite model inference  
+        interpreter.allocate_tensors()
+        input_details = interpreter.get_input_details()
+        output_details = interpreter.get_output_details()[0]
+        attention_mask, input_ids = tokens['attention_mask'], tokens['input_ids']
+        interpreter.resize_tensor_input(input_details[0]['index'],[batch_size, MAX_LEN])
+        interpreter.resize_tensor_input(input_details[1]['index'],[batch_size, MAX_LEN])
+        interpreter.resize_tensor_input(output_details['index'],[batch_size, N_CLASSES])
+        interpreter.allocate_tensors()
+        interpreter.set_tensor(input_details[0]["index"], attention_mask)
+        interpreter.set_tensor(input_details[1]["index"], input_ids)
+        interpreter.invoke()
+        tflite_pred = interpreter.get_tensor(output_details["index"])
+        tflite_pred = parse_prediction(tflite_pred)
+    return tflite_pred
+
+def cols_check(new_cols, old_cols):
+    return all([new_col==old_col for new_col, old_col in zip(new_cols, old_cols)])
+
+    
+def predict_news_category(old_news: pd.DataFrame, new_news: pd.DataFrame, interpreter, label_encoder, tokenizer):
+    old_news = old_news.copy()
+    new_news = new_news.copy()
+    # dbops = DBOperations()
+    # old_news = dbops.read_news_from_db()
+    old_news.drop(columns='_id', inplace=True)
+    # new_news = get_news()  
+    if 'category' not in [*old_news.columns]:
+        print('no prior predictions found')
+        if not cols_check([*new_news.columns], [*old_news.columns]):
+            raise Exeption("New and old cols don't match")
+        final_df = pd.concat([old_news, new_news], axis=0, ignore_index=True)
+        final_df.drop_duplicates(subset='url', keep='first', inplace=True)
+        headlines = [*final_df['title']].copy()
+        label, prob = inference(headlines, interpreter, label_encoder, tokenizer)
+        final_df['category'] = label
+        final_df['pred_proba'] = prob
+    else:
+        print('prior predictions found')
+        if not cols_check([*new_news.columns], [*old_news.columns][:-2]):
+            raise Exeption("New and old cols don't match")
+        old_urls = [*old_news['url']]
+        new_news = new_news.loc[new_news['url'].isin(old_urls) == False, :]
+        headlines = [*new_news['title']].copy()
+        label, prob = inference(headlines, interpreter, label_encoder, tokenizer)
+        new_news['category'] = label
+        new_news['pred_proba'] = prob
+        final_df = pd.concat([old_news, new_news], axis=0, ignore_index=True)
+        final_df.drop_duplicates(subset='url', keep='first', inplace=True)
+    final_df.reset_index(drop=True, inplace=True)
+    final_df.loc[final_df['pred_proba']<0.65, 'category'] = 'OTHERS'
+    return final_df
+    

news_extractor/__init__.py

@@ -0,0 +1 @@
+from news_extractor.news_extractor import *

news_extractor/news_extractor.py

@@ -0,0 +1,192 @@
+import pandas as pd
+import numpy as np
+from bs4 import BeautifulSoup
+import requests as r
+import regex as re
+from dateutil import parser
+import logging
+import multiprocessing
+
+
+def date_time_parser(dt):
+    """
+    Computes the minutes elapsed since published time.
+    :param dt: date
+    :return: int, minutes elapsed.
+    """
+    return int(np.round((dt.now(dt.tz) - dt).total_seconds() / 60, 0))
+
+def text_clean(desc):
+    """
+    Cleans the text by removing special chars.
+    :param desc: string containing description
+    :return: str, cleaned description.
+    """
+    desc = desc.replace("&lt;", "<")
+    desc = desc.replace("&gt;", ">")
+    desc = re.sub("<.*?>", "", desc)
+    desc = desc.replace("#39;", "'")
+    desc = desc.replace('&quot;', '"')
+    desc = desc.replace('&nbsp;', ' ')
+    desc = desc.replace('#32;', ' ')
+    return desc
+
+
+def rss_parser(i):
+    """
+    Returns a data frame of parsed news item.
+    :param i: single news item in RSS feed.
+    :return: Data frame of parsed news item.
+    """
+    b1 = BeautifulSoup(str(i), "xml")
+    title = "" if b1.find("title") is None else b1.find("title").get_text()
+    title = text_clean(title)
+    url = "" if b1.find("link") is None else b1.find("link").get_text()
+    desc = "" if b1.find("description") is None else b1.find("description").get_text()
+    desc = text_clean(desc)
+    desc = f'{desc[:300]}...' if len(desc) >= 300 else desc
+    date = "Sat, 12 Aug 2000 13:39:15 +05:30" if ((b1.find("pubDate") == "") or (b1.find("pubDate") is None)) else b1.find("pubDate").get_text()
+    if url.find("businesstoday.in") >= 0:
+        date = date.replace("GMT", "+0530")
+    
+    date1 = parser.parse(date)
+    return pd.DataFrame({"title": title,
+                         "url": url,
+                         "description": desc,
+                         "parsed_date": date1}, index=[0])
+
+
+def src_parse(rss):
+    """
+    Returns the root domain name (eg. livemint.com is extracted from www.livemint.com
+    :param rss: RSS URL
+    :return: str, string containing the source name
+    """
+    if rss.find('ndtvprofit') >= 0:
+        rss = 'ndtv profit'
+    if rss.find('ndtv') >= 0:
+        rss = 'ndtv.com'
+    if rss.find('telanganatoday') >= 0:
+        rss = 'telanganatoday.com'
+    
+    rss = rss.replace("https://www.", "")
+    rss = rss.split("/")
+    return rss[0]
+
+
+def news_agg(rss):
+    """
+    Returns feeds from each 'rss' URL.
+    :param rss: RSS URL.
+    :return: Data frame of processed articles.
+    """
+    try:
+        rss_df = pd.DataFrame()
+        # user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/109.0.0.0 Safari/537.36"
+        headers = {
+                    'authority': 'www.google.com',
+                    'accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
+                    'accept-language': 'en-US,en;q=0.9',
+                    'cache-control': 'max-age=0',
+                    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36'
+        }
+
+        timeout = 5
+        
+        resp = r.get(rss, timeout=timeout, headers=headers)
+        logging.warning(f'{rss}: {resp.status_code}')
+        b = BeautifulSoup(resp.content, "xml")
+        items = b.find_all("item")
+        for i in items:
+            # rss_df = rss_df.append(rss_parser(i)).copy()
+            rss_df = pd.concat([rss_df, rss_parser(i)], axis=0)
+        rss_df.reset_index(drop=True, inplace=True)
+        rss_df["description"] = rss_df["description"].replace([" NULL", ''], np.nan)
+
+
+        #### UNCOMMENT IN CASE OF OOM ERROR IN RENDER
+        # rss_df.dropna(inplace=True)
+
+        ####
+
+        
+        rss_df["src"] = src_parse(rss)
+        rss_df["elapsed_time"] = rss_df["parsed_date"].apply(date_time_parser)
+        rss_df["parsed_date"] = rss_df["parsed_date"].astype("str")
+        # rss_df["elapsed_time_str"] = rss_df["elapsed_time"].apply(elapsed_time_str)
+    except Exception as e:
+        print(e)
+        pass
+    return rss_df
+
+
+# List of RSS feeds
+rss = ['https://www.economictimes.indiatimes.com/rssfeedstopstories.cms',
+       'https://www.thehindu.com/news/feeder/default.rss',
+       # 'https://telanganatoday.com/feed',
+       'https://www.businesstoday.in/rssfeeds/?id=225346',
+       'https://feeds.feedburner.com/ndtvnews-latest',
+       'https://www.hindustantimes.com/feeds/rss/world-news/rssfeed.xml',
+       'https://www.indiatoday.in/rss/1206578',
+       
+       'https://www.moneycontrol.com/rss/latestnews.xml',
+       'https://www.livemint.com/rss/news',
+       
+       'https://www.zeebiz.com/latest.xml/feed',
+       'https://www.timesofindia.indiatimes.com/rssfeedmostrecent.cms']
+
+
+def get_news_rss(url):
+    # final_df = pd.DataFrame()
+    # for i in rss:
+    #     # final_df = final_df.append(news_agg(i))
+    #     final_df = pd.concat([final_df, news_agg(i)], axis=0)
+    final_df = news_agg(url)
+    final_df.reset_index(drop=True, inplace=True)
+
+    
+
+    final_df.sort_values(by="elapsed_time", inplace=True)
+    # final_df['src_time'] = final_df['src'] + ("&nbsp;" * 5) + final_df["elapsed_time_str"]
+    # final_df.drop(columns=['date', 'parsed_date', 'src', 'elapsed_time', 'elapsed_time_str'], inplace=True)
+    final_df.drop(columns=['elapsed_time'], inplace=True)
+
+
+    #### UNCOMMENT 1ST STATEMENT AND REMOVE 2ND STATEMENT IN CASE OF OOM ERROR IN RENDER
+    # final_df.drop_duplicates(subset='description', inplace=True)
+    final_df.drop_duplicates(subset='url', inplace=True)
+    
+    ####
+    
+    final_df = final_df.loc[(final_df["title"] != ""), :].copy()
+
+    final_df.loc[(final_df['description'].isna()) | (final_df['description']=='')| (final_df['description']==' '), 'description'] = final_df.loc[(final_df['description'].isna()) | (final_df['description']=='')| (final_df['description']==' '), 'title']
+    
+    return final_df
+
+def get_news_multi_process(urls):
+    '''
+    Get the data shape by parallely calculating lenght of each chunk and 
+    aggregating them to get lenght of complete training dataset
+    '''
+    pool = multiprocessing.Pool(processes=multiprocessing.cpu_count())
+    
+    results = []
+    for url in urls:
+        f = pool.apply_async(get_news_rss, [url]) # asynchronously applying function to chunk. Each worker parallely begins to work on the job
+        results.append(f) # appending result to results
+        
+    final_df = pd.DataFrame()
+    for f in results:
+#         print(f.get())
+        final_df = pd.concat([final_df, f.get(timeout=120)], axis=0) # getting output of each parallel job
+    
+    final_df.reset_index(drop=True, inplace=True)
+    logging.warning(final_df['src'].unique())
+    pool.close()
+    pool.join()
+    return final_df
+
+
+def get_news():
+   return get_news_multi_process(rss)

requirements.txt

@@ -10,3 +10,7 @@ flask_cors==3.0.10
 gunicorn==20.1.0
 pymongo==4.3.3
 Werkzeug==2.2.2
+tensorflow
+scikit-learn==1.2.2
+cloudpickle
+transformers