File size: 4,837 Bytes
f5dd236 220d7b5 2199ce9 f5dd236 94c21a4 f5dd236 94c21a4 0021684 94c21a4 2199ce9 f5dd236 485e12a 94c21a4 f5dd236 2199ce9 f5dd236 94c21a4 f5dd236 2199ce9 f5dd236 0021684 f5dd236 2199ce9 f5dd236 2199ce9 f5dd236 0021684 2199ce9 f5dd236 5352272 f5dd236 2199ce9 f5dd236 fd6a623 f5dd236 627e458 f5dd236 1b9ba83 |
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 |
from fastapi import FastAPI, Query,HTTPException
from fastapi.responses import JSONResponse, FileResponse, StreamingResponse
from google.cloud import texttospeech
from google.oauth2.service_account import Credentials
from langchain.schema import HumanMessage
from langchain_groq import ChatGroq
import json
from dotenv import load_dotenv
import os
from utils import (
extract_titles_and_summaries,
perform_sentiment_analysis,
extract_topics_with_hf,
compare_articles
)
load_dotenv()
GROQ_API_KEY = os.getenv('GROQ_API_KEY')
PRIVATE_KEY = os.getenv('PRIVATE_KEY').replace("\\n", "\n")
CLIENT_EMAIL = os.getenv('CLIENT_EMAIL')
app = FastAPI(title="Company Sentiment API", description="Get company news summaries with sentiment analysis")
llm=ChatGroq(api_key=GROQ_API_KEY, model="llama-3.1-8b-instant")
JSON_FILE_PATH = "final_summary.json"
AUDIO_FILE_PATH = "hindi_summary.mp3"
def get_tts_client():
credentials = Credentials.from_service_account_info({
"type": "service_account",
"private_key": PRIVATE_KEY,
"client_email": CLIENT_EMAIL,
"token_uri": "https://oauth2.googleapis.com/token"
})
return texttospeech.TextToSpeechClient(credentials=credentials)
def generate_summary(company_name):
news_articles = extract_titles_and_summaries(company_name)
news_articles, sentiment_counts = perform_sentiment_analysis(news_articles)
news_articles = extract_topics_with_hf(news_articles)
final_summary = compare_articles(news_articles, sentiment_counts)
hindi_text = ""
if PRIVATE_KEY and CLIENT_EMAIL:
hindi_prompt = f"Just Translate this text into Hindi: {final_summary['Final Sentiment Analysis']}"
hindi_response = llm.invoke([HumanMessage(content=hindi_prompt)]).content
hindi_text = hindi_response.strip() if hindi_response else "Translation not available."
if hindi_text:
print(f"Generated Hindi Text: {hindi_text}")
else:
print("Hindi Text not generated")
try:
client = get_tts_client()
input_text = texttospeech.SynthesisInput(text=hindi_text)
voice = texttospeech.VoiceSelectionParams(
language_code="hi-IN",
name="hi-IN-Chirp3-HD-Kore"
)
audio_config = texttospeech.AudioConfig(audio_encoding=texttospeech.AudioEncoding.MP3)
response = client.synthesize_speech(input=input_text, voice=voice, audio_config=audio_config)
with open(AUDIO_FILE_PATH, "wb") as out:
out.write(response.audio_content)
print(f"Audio content written to file: {AUDIO_FILE_PATH}")
except Exception as e:
print(f"Error generating audio: {e}")
if not os.path.exists(AUDIO_FILE_PATH):
print(f"Audio file could not be found at {AUDIO_FILE_PATH}.")
final_summary["Audio"] = AUDIO_FILE_PATH
with open(JSON_FILE_PATH,"w",encoding="utf-8") as f:
json.dump(final_summary,f,ensure_ascii=False, indent=4)
return {
'Company': final_summary["Company"],
'Articles': [
{
'Title': article.get('Title', 'No Title'),
'Summary': article.get('Summary', 'No Summary'),
'Sentiment': article.get('Sentiment', 'Unknown'),
'Score': article.get('Score', 0.0),
'Topics': article.get('Topics', [])
}
for article in final_summary["Articles"]
],
'Comparative Sentiment Score': {
'Sentiment Distribution': sentiment_counts,
'Coverage Differences': final_summary["Comparative Sentiment Score"].get("Coverage Differences", []),
'Topic Overlap': {
'Common Topics': final_summary["Comparative Sentiment Score"].get("Topic Overlap", {}).get("Common Topics", []),
'Unique Topics': final_summary["Comparative Sentiment Score"].get("Topic Overlap", {}).get("Unique Topics", {})
}
},
'Final Sentiment Analysis': final_summary["Final Sentiment Analysis"],
'Audio': AUDIO_FILE_PATH
}
@app.get("/")
def home():
return {"message": "Welcome to the Company Sentiment API"}
@app.post("/generateSummary")
def get_summary(company_name: str = Query(..., description="Enter company name")):
structured_summary = generate_summary(company_name)
return structured_summary
@app.get("/downloadJson")
def download_json():
return FileResponse(JSON_FILE_PATH, media_type="application/json", filename="final_summary.json")
@app.get("/downloadHindiAudio")
def download_audio():
return FileResponse(AUDIO_FILE_PATH, media_type="audio/mp3", filename="hindi_summary.mp3")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
|