HelloWorld-FastAPI/utils.py at master · sam9111/HelloWorld-FastAPI · GitHub

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
importjson
importos
fromdatetimeimportdatetime

importpytz
fromdotenvimportload_dotenv
fromnewscatcherapiimportNewsCatcherApiClient

load_dotenv()

UTC=pytz.utc

# NewsCatcherAPI setup
newscatcherapi=NewsCatcherApiClient(x_api_key=os.getenv("API_KEY"))

# Azure Text Analytics Setup
key=os.getenv("AZURE_API_KEY")
endpoint=os.getenv("ENDPOINT")

fromazure.ai.textanalyticsimportTextAnalyticsClient
fromazure.core.credentialsimportAzureKeyCredential


defauthenticate_client():
ta_credential=AzureKeyCredential(key)
text_analytics_client=TextAnalyticsClient(
endpoint=endpoint, credential=ta_credential
 )
returntext_analytics_client


client=authenticate_client()


# Fetching and processing news from NewsCatcherApi


defmake_news(response):
articles=response["articles"]
news= {"countries": {}, "last_fetched": ""}
news_by_country=news["countries"]
forarticleinarticles:
ifarticle["country"] =="unknown":
continue
article_obj= {
"_id": article["_id"],
"title": article["title"],
"summary": article["summary"],
"topics": article["topic"],
"published_date": article["published_date"],
"link": article["link"],
"country": article["country"],
"media": article["media"],
 }

if (country:=article["country"]) notinnews_by_country:
country_obj= {"articles": []}
country_obj["articles"].append(article_obj)
news_by_country[country] =country_obj
else:
iflen(news_by_country[country]["articles"]) <10:
news_by_country[country]["articles"].append(article_obj)

news["last_fetched"] =datetime.now(UTC).strftime("%Y:%m:%d %H:%M:%S %Z %z")
returnnews


defupdate_news():
response=newscatcherapi.get_latest_headlines_all_pages(lang="en", when="24h")
withopen("./data/news.json", "w") asoutfile:
json.dump(make_news(response), outfile)


defget_news():
withopen("./data/news.json", "r") asinfile:
news_json=json.load(infile)
returnnews_json


# Process news to calculate average sentiments for articles of each country


defmake_data():
news=get_news()
news_by_country=news["countries"]
countries= {}

forcountryinnews_by_country:
documents= []
forarticleinnews_by_country[country]["articles"]:
documents.append(article["title"])
responses=client.analyze_sentiment(documents=documents)
positives= []
negatives= []
forresponseinresponses:
sentiment=response.sentiment
ifsentiment=="neutral":
continue
ifsentiment=="positive":
sentiment_score=response.confidence_scores.positive
positives.append(sentiment_score)
else:
sentiment_score=response.confidence_scores.negative
negatives.append(sentiment_score)

iflen(positives) ==0andlen(negatives) ==0:
continue
avg_positive_score=0
avg_negative_score=0
iflen(positives) >0:
avg_positive_score=sum(positives) /len(positives)
iflen(negatives) >0:
avg_negative_score=sum(negatives) /len(negatives)

ifavg_positive_score>avg_negative_score:
news_by_country[country]["sentiment"] ="positive"
news_by_country[country]["sentiment_score"] =avg_positive_score
else:
news_by_country[country]["sentiment"] ="negative"
news_by_country[country]["sentiment_score"] =avg_negative_score
countries[country] =news_by_country[country]

news["countries"] =countries
returnnews


defupdate_data():
withopen("./data/data.json", "w") asoutfile:
json.dump(make_data(), outfile)


defget_data():
withopen("./data/data.json", "r") asinfile:
data_json=json.load(infile)
returndata_json


# Process data to create marker points for the frontend


defmake_points():
data=get_data()

data_by_country=data["countries"]

withopen("./data/countries.json", "r") asinfile:
point_objs=json.load(infile)

forpoint_objinlist(point_objs):
country=point_obj["id"]
ifcountrynotindata_by_country:
point_objs.remove(point_obj)
continue

point_obj["color"] = (
"green"ifdata_by_country[country]["sentiment"] =="positive"else"red"
 )

point_obj["value"] =data_by_country[country]["sentiment_score"] *100

point_obj["sentiment"] =data_by_country[country]["sentiment"]

returnpoint_objs


defupdate_points():
withopen("./data/points.json", "w") asoutfile:
json.dump(make_points(), outfile)


defget_points():
withopen("./data/points.json", "r") asinfile:
points_json=json.load(infile)
returnpoints_json