Spaces:
Sleeping
Sleeping
| from flask import Flask, request, jsonify | |
| from flask_cors import CORS | |
| import os | |
| from dotenv import load_dotenv | |
| from transformers import pipeline | |
| import feedparser | |
| import json | |
| from dateutil import parser | |
| import re | |
| load_dotenv() | |
| # Load Setiment Classifier | |
| sentiment_analysis = pipeline( | |
| "sentiment-analysis", model="siebert/sentiment-roberta-large-english" | |
| ) | |
| app = Flask(__name__, static_url_path="", static_folder="dist") | |
| CORS(app) | |
| def index(): | |
| return app.send_static_file("index.html") | |
| def get_news(): | |
| feed_url = request.args.get("feed_url") | |
| # check if string is a valid | |
| # file name for cache | |
| file_name = "".join(re.split(r"https://|\.|/", feed_url)) | |
| feed_entries = get_feed(feed_url) | |
| # filter only titles for sentiment analysis | |
| try: | |
| with open(f"{file_name}_cache.json") as file: | |
| cache = json.load(file) | |
| except: | |
| cache = {} | |
| # if new homepage is newer than cache, update cache and return | |
| print("new date", feed_entries["last_update"]) | |
| print("old date", cache["last_update"] if "last_update" in cache else "None") | |
| if not cache or parser.parse(feed_entries["last_update"]) > parser.parse( | |
| cache["last_update"] | |
| ): | |
| print("Updating cache with new preditions") | |
| titles = [entry["title"] for entry in feed_entries["entries"]] | |
| # run sentiment analysis on titles | |
| predictions = [sentiment_analysis(sentence) for sentence in titles] | |
| # parse Negative and Positive, normalize to -1 to 1 | |
| predictions = [ | |
| -prediction[0]["score"] | |
| if prediction[0]["label"] == "NEGATIVE" | |
| else prediction[0]["score"] | |
| for prediction in predictions | |
| ] | |
| # merge rss data with predictions | |
| entries_predicitons = [ | |
| {**entry, "sentiment": prediction} | |
| for entry, prediction in zip(feed_entries["entries"], predictions) | |
| ] | |
| output = { | |
| "entries": entries_predicitons, | |
| "last_update": feed_entries["last_update"], | |
| } | |
| # update last precitions cache | |
| with open(f"{file_name}_cache.json", "w") as file: | |
| json.dump(output, file) | |
| # send back json | |
| return jsonify(output) | |
| else: | |
| print("Returning cached predictions") | |
| return jsonify(cache) | |
| def predict(): | |
| # get data from POST | |
| if request.method == "POST": | |
| # get current news | |
| # get post body data | |
| data = request.get_json() | |
| if data.get("sentences") is None: | |
| return jsonify({"error": "No text provided"}) | |
| # get post expeceted to be under {'sentences': ['text': '...']} | |
| sentences = data.get("sentences") | |
| # prencit sentiments | |
| predictions = [sentiment_analysis(sentence) for sentence in sentences] | |
| # parse Negative and Positive, normalize to -1 to 1 | |
| predictions = [ | |
| -prediction[0]["score"] | |
| if prediction[0]["label"] == "NEGATIVE" | |
| else prediction[0]["score"] | |
| for prediction in predictions | |
| ] | |
| output = [ | |
| dict(sentence=sentence, sentiment=prediction) | |
| for sentence, prediction in zip(sentences, predictions) | |
| ] | |
| # send back json | |
| return jsonify(output) | |
| def get_feed(feed_url): | |
| feed = feedparser.parse(feed_url) | |
| return {"entries": feed["entries"], "last_update": feed["feed"]["updated"]} | |
| if __name__ == "__main__": | |
| app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860))) |