-
Notifications
You must be signed in to change notification settings - Fork 8
/
Copy pathmain.py
169 lines (144 loc) · 6.84 KB
/
main.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
import eel
import datetime, os
from functools import cache
from email_validator import validate_email
from src.ai_models import ChatGPT
from src.google.__init__ import createService
from src.google.gmail_bot import GoogleGmailManager
from src.google.calendar_bot import GoogleCalendarManager
from src.google.contact_bot import GoogleContactManager
from src.google.task_bot import GoogleTaskManager
from langchain.schema import HumanMessage, AIMessage
eel.init('UI')
@eel.expose
def signUp():
createService('gmail', 'v1')
@eel.expose
def logIn():
if os.path.exists("src/configs/google_token.json"):
return True
else:
return False
@cache
def decodeEmail(text):
lines = text.split('\n')
subject_index = next((i for i, line in enumerate(lines) if "Subject:" in line), None)
body = '\n'.join(lines[subject_index+2:]) if subject_index is not None else text
subject = next((line.replace("Subject: ", "") for line in lines if "Subject" in line), None)
return body, subject
@cache
def get_ordinal_suffix(day): # This function helps determine the ordinal suffix for a day number.
if 4 <= day % 100 <= 20: suffix = "th"
else: suffix = ("st", "nd", "rd")[day % 10 if day % 10 < 4 else 0]
return suffix
@cache
def formatDatetime(timestamp: str):
timestamp= timestamp[:16]
if "T" not in timestamp:
parsed_date = datetime.datetime.strptime(timestamp, "%Y-%m-%d")
formatted_time = "whole day"
elif "T" in timestamp:
parsed_date = datetime.datetime.strptime(timestamp, "%Y-%m-%dT%H:%M")
formatted_time = parsed_date.strftime("%I:%M %p") # %I for 12-hour format, %p for AM/PM
timestamp = timestamp.split("T")[0]
parsed_date = datetime.datetime.strptime(timestamp, "%Y-%m-%d")
try:
# Get the day, month name, and day name
day = parsed_date.day
day_with_suffix = f"{day}{get_ordinal_suffix(day)}"
month_name = parsed_date.strftime("%b")
day_name = parsed_date.strftime("%A") # Full day name (Monday, Tuesday, etc.)
# Format the date with all desired components
formatted_date = f"{day_name}, {month_name} {day_with_suffix}, {parsed_date.year}"
except: formatted_date, formatted_time = "...", "..."
return formatted_date, formatted_time
@eel.expose
def validateEmail(emailId):
try:
validate_email(emailId)
return True
except ValueError as e:
print(f"Invalid email address: {e}")
return False
@eel.expose
def sendInstEmail(recipient_email, subject, body):
gmail = GoogleGmailManager()
gmail.sendEmail(recipient_email, subject, body)
@eel.expose
def createDraftEmail(subject, body, recipient_email=None):
print("called")
gmail = GoogleGmailManager()
if recipient_email != None:
gmail.createDraft(body=body, recipient_email=recipient_email, subject=subject)
else:
gmail.createDraft(body=body, subject=subject)
@eel.expose
def clearMemory():
chatHistory = []
chatHistory = []
@eel.expose
@cache
def giveResponseArray(query: str):
# try:
# print(chatHistory)
chatHistory.append(HumanMessage(content=query))
raw_response = ChatGPT.askAI(query, chatHistory)
if "order is composing email" in raw_response:
body, subject = decodeEmail(raw_response)
response = ["email", subject, body]
chatHistory.append(AIMessage(content=raw_response))
elif "order is to fetch upcoming events from Calendar" in raw_response:
calendar = GoogleCalendarManager()
amount = 10
chatHistory.append(AIMessage(content="order is to fetch upcoming events from Calendar"))
if "order is to fetch upcoming events from Calendar (amount: " in raw_response:
amount = int(raw_response.split("amount: ", 1)[1].split(")", 1)[0])
event_list = calendar.upcomingEvent(amount)
if event_list:
text = "Sure, here are your upcoming events: \n\n" + "\n".join(f"{formatDatetime(item[1])[0]}, {formatDatetime(item[1])[1]} : {item[2]}" for item in event_list)
else:
text = "Unfortunately, the event you are searching for does not appear to be exist"
response = ["calendar event", text]
chatHistory.append(AIMessage(content=text))
elif "order is to fetch today's events from Calendar" in raw_response:
calendar = GoogleCalendarManager()
event_list = calendar.todaysEvent()
chatHistory.append(AIMessage(content="order is to fetch today's events from Calendar"))
if event_list:
text = "Sure, here are your upcoming events for today: \n\n" + "\n".join(f"{formatDatetime(item[1])[0]}, {formatDatetime(item[1])[1]} - {item[2]}" for item in event_list)
else:
text = "I am unable to locate any event for today in Google Calendar"
response = ["calendar today's events", text]
chatHistory.append(AIMessage(content=text))
elif "order is fetching contact from Contact" in raw_response:
name = ChatGPT.findName(query)
contact = GoogleContactManager()
contact_info = contact.phoneNumber(name)
chatHistory.append(AIMessage(content="order is fetching contact from Contact"))
if contact_info:
text = "Sure, here is your contact: \n\n" + "\n".join(f"{item[0]} : {item[1]}" for item in contact_info)
else:
text = "I am unable to locate any contact number you are searching for in Google Contact"
response = ["contact", text]
chatHistory.append(AIMessage(content=text))
elif "order is to fetch task from Calendar" in raw_response:
task = GoogleTaskManager()
task_list = task.dueTask()
chatHistory.append(AIMessage(content="order is to fetch task from Calendar"))
if task_list:
text = "Sure, here are your due tasks: \n\n" + "\n".join(f"{formatDatetime(item[1])[0]}, {formatDatetime(item[1])[1]} - {item[0]}" for item in task_list)
else:
text = "Hooray! 🎉 you don't have any due tasks !"
response = ["calendar task", text]
chatHistory.append(AIMessage(content=text))
elif "order is fetching today's event and task from Calendar" in raw_response:
text = "Sorry, this feature is still not available, waiting for the next update"
response = ["calendar all", text]
chatHistory.append(AIMessage(content="order is fetching today's event and task from Calendar"))
else:
response = ["others", raw_response]
chatHistory.append(AIMessage(content=raw_response))
# except:
# response = "Sorry something went wrong, please try again"
return response
eel.start('index.html')