File size: 6,462 Bytes
c576592 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 |
from fastapi import FastAPI, HTTPException
from utils import bs4_extractor, SentimentAnalyzer, SemanticGrouping, save_audio
from llm_utils import *
import openai
import asyncio
app = FastAPI()
# Initialize sentiment analyzer and semantic grouping
sentiment_analyzer = SentimentAnalyzer()
semantic_grouping = SemanticGrouping()
class CompareNewsRequest(BaseModel):
api_key: str
model_name: str = "gpt-4o-mini"
company_name: str
def check_api_key(api_key: str):
if api_key == None:
return False
elif api_key != None:
client = openai.OpenAI(api_key=api_key)
try:
client.models.list()
except openai.AuthenticationError:
return False
else:
return True
def check_model_name(model_name: str, api_key: str):
openai.api_key = api_key
model_list = [model.id for model in openai.models.list()]
return True if model_name in model_list else False
# Helper function to get articles and article sentiments
def get_articles(company_name: str):
if not company_name:
raise HTTPException(status_code=500, detail="The company name is required.")
news_articles = bs4_extractor(company_name)
if not news_articles:
raise HTTPException(status_code=500, detail="No news found")
articles_data = [
{"title": article["title"], "summary": article["summary"]}
for article in news_articles
]
analyzed_articles = sentiment_analyzer.classify_sentiments(articles_data)
return news_articles, analyzed_articles
def get_formatted_output(
company_name,
analyzed_articles,
topic_extraction_results,
topic_overlap_results,
comparative_analysis_results,
final_analysis,
):
articles = analyzed_articles
sentiment_distribution = {"positive": 0, "negative": 0, "neutral": 0}
for i in range(len(articles)):
articles[i]["topics"] = topic_extraction_results[i]
sentiment = articles[i]["sentiment"]
sentiment_distribution[sentiment] += 1
comparative_sentiment_score = {
"Sentiment Distribution": sentiment_distribution,
"Coverage Differences": comparative_analysis_results,
"Topic Overlap": topic_overlap_results,
}
final_output = {
"Company": company_name,
"Articles": articles,
"Comparative Sentiment Score": comparative_sentiment_score,
"Final Sentiment Analysis": final_analysis,
}
return final_output
@app.get("/news/{company_name}")
def get_news(company_name: str):
"""
API endpoint to get news for a company.
Fetches news articles from NYTimes and BBC.
"""
try:
news_articles = bs4_extractor(company_name)
app.state.company_name = company_name
if not news_articles:
raise HTTPException(status_code=404, detail="No news found")
app.state.news_articles = news_articles
return {"company": company_name, "articles": news_articles}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/analyze-news")
def analyze_news():
"""
API endpoint to analyze news articles.
Performs sentiment analysis on a list of articles.
"""
if not app.state.news_articles:
HTTPException(
status_code=500, detail="Type in the name before the news analysis."
)
try:
articles_data = [
{"title": article["title"], "summary": article["summary"]}
for article in app.state.news_articles
]
analyzed_articles = sentiment_analyzer.classify_sentiments(articles_data)
app.state.articles_with_sentiments = analyzed_articles
return {"analyzed_articles": analyzed_articles}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/compare-news")
async def compare_news(request_info: CompareNewsRequest):
"""
API endpoint to perform comparative analysis.
Uses semantic similarity to find the most related articles.
"""
api_key = request_info.api_key
company_name = request_info.company_name
model_name = request_info.model_name
if not check_api_key(api_key):
HTTPException(
status_code=500,
detail="The entered API key does not seem to be right. Please enter a valid API key",
)
if not check_model_name(model_name, api_key):
HTTPException(
status_code=500,
detail="The model you specified does not exist.",
)
news_articles, analyzed_articles = get_articles(company_name)
try:
articles_text = [
f"{article['title']}. {article['summary']}" for article in news_articles
]
if len(articles_text) < 2:
raise HTTPException(
status_code=400, detail="At least two articles required for comparison."
)
top_similar_articles = semantic_grouping.find_top_k_similar_articles(
articles_text, k=5
)
llm_chatbot = ChatBot(
api_key,
model_name,
analyzed_articles,
company_name,
)
llm_result = await llm_chatbot.main(top_similar_articles)
(
topic_extraction_results,
topic_overlap_results,
comparative_analysis_results,
) = llm_result.values()
final_analysis_eng, final_analysis_hi = llm_chatbot.final_analysis(
comparative_analysis_results
)
final_output = get_formatted_output(
company_name,
analyzed_articles,
topic_extraction_results,
topic_overlap_results,
comparative_analysis_results,
final_analysis_eng,
)
app.state.hindi_summary = final_analysis_hi
return final_output
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/hindi-summary")
def get_hindi_summary():
if not app.state.hindi_summary:
raise HTTPException(
status_code=500, detail="Generate the Comparative Analysis first."
)
save_audio(app.state.hindi_summary)
return {"hindi_summary": app.state.hindi_summary}
|