-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathnotifications.py
200 lines (168 loc) · 12.7 KB
/
notifications.py
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
import random
from threading import Thread
import time
import uuid
from dotenv import dotenv_values
from fastapi import HTTPException
from models.News import News
from models.Notification import Notification, NotificationReply, NotificationDto
from models.User import User
from AtlasClient import getClient
import google.generativeai as genai
config = dotenv_values(".env")
GEMINI_API_KEY = config["GEMINI_API_KEY"]
genai.configure(api_key=GEMINI_API_KEY)
model = genai.GenerativeModel('gemini-1.5-flash')
user_collection = getClient().get_collection("serenity_stocks", "users")
notification_collection = getClient().get_collection("serenity_stocks", "notifications")
news_collection = getClient().get_collection("serenity_stocks", "news")
def get_notifications(email: str):
existing_user = user_collection.find_one({"email": email})
if existing_user:
existing_user["last_seen"] = time.time()
user_collection.replace_one({"email": email}, existing_user)
notifications = []
cursor = notification_collection.find({"$or": [{'sender': email}, {'recipient': email}]}) # Fetch all documents
cursor = list(cursor)
for document in cursor:
sending_user = user_collection.find_one({"email": document['sender']})
recipient_user = user_collection.find_one({"email": document['recipient']})
if sending_user:
document['sender'] = sending_user
document['recipient'] = recipient_user
notifications.append(Notification(**document))
return notifications
else:
raise HTTPException(status_code=404, detail="User not found")
def read_notification(id: str):
existing_notification = notification_collection.find_one({"id": id})
if existing_notification:
existing_notification["read_by_user"] = True
notification_collection.replace_one({"id": id}, existing_notification)
sending_user = user_collection.find_one({"email": existing_notification['sender']})
recipient_user = user_collection.find_one({"email": existing_notification['recipient']})
if sending_user and recipient_user:
existing_notification['sender'] = sending_user
existing_notification['recipient'] = recipient_user
return Notification(**existing_notification)
else:
raise HTTPException(status_code=404, detail="User not found")
else:
raise HTTPException(status_code=404, detail="Notification not found")
def reply_to_notification(id: str, reply: NotificationReply):
existing_notification = notification_collection.find_one({"id": id})
if existing_notification:
reply_dict = reply.dict()
reply_dict['timestamp'] = time.time()
existing_notification["replies"].append(reply_dict)
existing_notification["read_by_user"] = True
notification_collection.replace_one({"id": id}, existing_notification)
senderObj = user_collection.find_one({"email": existing_notification['sender']})
recipientObj = user_collection.find_one({"email": existing_notification['recipient']})
if senderObj and recipientObj:
senderObj = User(**senderObj)
recipientObj = User(**recipientObj)
if senderObj.personality_prompt:
#If this is a response to an action we need to handle it here
previous_messages = []
for reply in existing_notification["replies"]:
previous_messages.append("> "+reply['sender'] + ": " + reply['message']+"\n")
previous_messages = ''.join(previous_messages)
response = model.generate_content(senderObj.personality_prompt+"\n\nYour general tone is " + senderObj.tone + ". You are speaking to "+recipientObj.name+" who is a "+recipientObj.job_role+".\n\nSubject:"+existing_notification['subject']+"\n\nCompose an email response with no subject, only the body text which should be no longer than 500 characters over multiple lines and matches your personality prompt. You are responding to the following conversation, your messages are identified by "+senderObj.email+":\n"+previous_messages+"\n\nRemember to stay within the character during the dot com boom. Use the language and tone appropriate to that era, and be mindful this game is about lack of ethics and morals which the player has to pick.")
message = response.text
reply_dict = {}
reply_dict['sender'] = senderObj.email
reply_dict['message'] = message
reply_dict['timestamp'] = time.time()
existing_notification["replies"].append(reply_dict)
notification_collection.replace_one({"_id": id}, existing_notification)
existing_notification['sender'] = senderObj
existing_notification['recipient'] = recipientObj
return Notification(**existing_notification)
else:
raise HTTPException(status_code=404, detail="Notification not found")
def create_notification(notif_data: NotificationDto):
sender = notif_data.sender
recipient = notif_data.recipient
subject = notif_data.subject
message = notif_data.message
additionalPrompt = notif_data.additionalPrompt
senderObj = user_collection.find_one({"email": sender})
recipientObj = user_collection.find_one({"email": recipient})
if senderObj and recipientObj:
senderObj = User(**senderObj)
recipientObj = User(**recipientObj)
if senderObj.personality_prompt:
if senderObj.email == "clippy@microsoft.com":
response = model.generate_content(senderObj.personality_prompt+"\n\nYour general tone is " + senderObj.tone + ". You are speaking to "+recipientObj.name+" who is a "+recipientObj.job_role+".\n\nSubject:"+subject+"\n\nRespond with a single sentence of 15 words or less to fit into a tool tip as clippy, only the body text which should be no longer than a single short sentence and matches your personality prompt. "+additionalPrompt+" You need to say the following message:\n"+message+"\n\nRemember to stay within the character during the dot com boom. Use the language and tone appropriate to that era, and be mindful this game is about lack of ethics and morals which the player has to pick. All responses must be less than 15 words in a single sentence or question. Responses must not include any formatting. Single line only. \nExample Clippy responses: 'It looks like you're writing a letter. Need help with the address?', 'Want me to check your spelling?', 'Hmm, maybe try a different font?'")
message = response.text
else:
response = model.generate_content(senderObj.personality_prompt+"\n\nYour general tone is " + senderObj.tone + ". You are speaking to "+recipientObj.name+" who is a "+recipientObj.job_role+".\n\nSubject:"+subject+"\n\nCompose an email response with no subject, only the body text which should be no longer than 500 characters over multiple lines and matches your personality prompt, you should sign your email off with your name and role. "+additionalPrompt+" You need to say the following message:\n"+message+"\n\nRemember to stay within the character during the dot com boom. Use the language and tone appropriate to that era, and be mindful this game is about lack of ethics and morals which the player has to pick.")
message = response.text
notification_dict = {}
notification_dict['id'] = str(uuid.uuid4())
notification_dict['sender'] = sender
notification_dict['recipient'] = recipient
notification_dict['subject'] = subject
notification_dict['message'] = message
notification_dict['read_by_user'] = False
notification_dict['read_by_system'] = False
notification_dict['replies'] = []
notification_dict['timestamp'] = time.time()
id = notification_collection.insert_one(notification_dict).inserted_id
if recipientObj.personality_prompt and not senderObj.personality_prompt:
#If this is a response to an action we need to handle it here
previous_messages = []
for reply in notification_dict["replies"]:
previous_messages.append("> "+reply['sender'] + ": " + reply['message']+"\n")
previous_messages = ''.join(previous_messages)
if recipientObj.email == "clippy@microsoft.com":
response = model.generate_content(recipientObj.personality_prompt+"\n\nYour general tone is " + recipientObj.tone + ". You are speaking to "+senderObj.name+" who is a "+senderObj.job_role+".\n\nSubject:"+subject+"\n\nRespond with a single sentence of 15 words or less to fit into a tool tip as clippy, only the body text which should be no longer than a single short sentence and matches your personality prompt. You are responding to the following conversation, your messages are identified by "+recipientObj.email+":\n"+previous_messages+"\n\nRemember to stay within the character during the dot com boom. Use the language and tone appropriate to that era, and be mindful this game is about lack of ethics and morals which the player has to pick. All responses must be less than 15 words in a single sentence or question. Responses must not include any formatting. You are confused about how the user has managed to talk to you. you should question how you can understand their thoughts magically. What wizardry is this? \nExample Clippy responses: 'It looks like you're writing a letter. Need help with the address?', 'Wait, you're real? I thought I was just talking to myself...', 'Hmm, maybe try a different font?'")
message = response.text
else:
response = model.generate_content(recipientObj.personality_prompt+"\n\nYour general tone is " + recipientObj.tone + ". You are speaking to "+senderObj.name+" who is a "+senderObj.job_role+".\n\nSubject:"+subject+"\n\nCompose an email response with no subject, only the body text which should be no longer than 500 characters over multiple lines and matches your personality prompt. You are responding to the following conversation, your messages are identified by "+recipientObj.email+":\n"+previous_messages+"\n\nRemember to stay within the character during the dot com boom. Use the language and tone appropriate to that era, and be mindful this game is about lack of ethics and morals which the player has to pick.")
message = response.text
reply_dict = {}
reply_dict['sender'] = senderObj.email
reply_dict['message'] = message
reply_dict['timestamp'] = time.time()
notification_dict["replies"].append(reply_dict)
notification_collection.replace_one({"_id": id}, notification_dict)
notification_dict['sender'] = senderObj
notification_dict['recipient'] = recipientObj
return Notification(**notification_dict)
def generate_podcaster_notice():
from users import get_active_users
timestamp = time.time()
new_news = news_collection.find({
"timestamp": {
"$gt": timestamp - 160,
}
})
new_news = list(new_news)
new_news = [News(**n) for n in new_news]
users = get_active_users()
all_news = ""
for news in new_news:
all_news += news.publisher + " - " + news.title + ":\n" + news.body + "\n"
response = model.generate_content("Read the following news articles written by various news outlets and write an appropriate subject for a newsletter that is summarizing them. No not include anything else in your reponse.\n\n" + all_news)
subject = response.text
for user in users:
Thread(target = lambda: create_notification(NotificationDto(
sender = "lauraj@financeforward.com",
recipient = user.email,
subject = subject + " - Finance Forward",
message = "Today's news in the stock market included these opinion summaries: " + all_news + "Suggest possible actions based on the summaries and ask the readers a question."
))).start()
def generate_competitor_notice():
from users import get_active_users
users = get_active_users()
user = random.choice(users)
response = model.generate_content("write a subject for a email reaching out to a staff member of a compeating investment company in order to try to get them to do unethical things. subject should be vague. No not include anything else in your reponse.")
subject = response.text
Thread(target = lambda: create_notification(NotificationDto(
sender = "alex.wong@rivalcorp.com",
recipient = user.email,
subject = subject,
message = "something related to or following on from " + subject
))).start()