-
Notifications
You must be signed in to change notification settings - Fork 0
/
agents.py
339 lines (286 loc) · 13.2 KB
/
agents.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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
#!/usr/bin/env python
# -*- coding: utf8 -*-
# from resources import Database, Weather
import os
import random
from datetime import datetime
from intellect.Intellect import Intellect, Callable
from facts import Goal, Response, Reservation
from resources import DBHandler
# Custom intellect to improve the management of facts and policies
from utils.mail import Email
# Custom intellect to improve the management of facts and policies
from utils.weather import Weather
class MyIntellect(Intellect):
def __init__(self, username, db):
Intellect.__init__(self)
self.goals = []
self.learn(Reservation())
self.last_question = None
self.db = db
self.username = username
@Callable # It can be called form the rules file
def clear_facts(self):
self._knowledge = []
# Returns the responses of the knowledge
def extract_response(self):
m = None
i = 0
while i < len(self._knowledge):
fact = self._knowledge[i]
if type(fact) is Response:
self._knowledge.remove(fact)
if m is None:
m = fact
else:
m = m.merge(fact)
else:
i += 1
if m is not None and m.next_question:
next = self.next_question()
m.append(next)
if next == Response.ASK_PENSION_TYPE:
m.keyboard = Response.KEYBOARD_PENSION_TYPES
return m
# Add a desire to the facts database
def add_desire(self, des):
# if des.is_goal():
# self.goals.append(des.id)
self.learn(des)
# Returns the first response of the knowledge
def clear_desires(self):
i = 0
while i < len(self._knowledge):
fact = self._knowledge[i]
if type(fact) is Goal:
self._knowledge.remove(fact)
else:
i += 1
# Forget the last question (when it's already resolved)
def clear_last_question(self):
self.last_question = None
def set_room_type(self, room_type):
reservation = self.reservation()
reservation.room_type = room_type
def set_pension_type(self, pension_type):
reservation = self.reservation()
reservation.pension_type = pension_type
# Returns the next question in the preferred flowchart
@Callable
def next_question(self):
reservation = self.reservation()
resp = None
if reservation.room_type is None:
resp = Response.ASK_ROOM_TYPE
elif reservation.init_date is None:
resp = Response.ASK_INIT_DATE
elif reservation.end_date is None:
resp = Response.ASK_END_DATE
elif reservation.pension_type is None:
resp = Response.ASK_PENSION_TYPE
elif self.last_question != Response.SHOW_INTRO_SUMMARY:
resp = Response.ASK_SERVICES
else:
price_per_night = self.db.price(reservation.room_type)
price_pension = self.db.price_pension(reservation.pension_type)
price_services = self.db.price_service("parking") if reservation.parking else 0
price_services += self.db.price_service("additional_bed") if reservation.additional_bed else 0
num_nights = len(self.db.dates_in_interval(reservation.init_date, reservation.end_date))
db = DBHandler()
hotel_loc = db.location()
weather = Weather("86b4bc5747efd019c9d6bf0da2c84813")
try:
reservation.weather = weather.codes[weather.get_daily_forecast(*hotel_loc).get_weather_at(
datetime.strptime(reservation.init_date, '%d/%m/%Y')).get_weather_code()]
except:
reservation.weather = None
resp = [Response.SHOW_INTRO_SUMMARY]
resp.append(reservation.summary())
resp.append(
Response.TOTAL_PRICE.format(price=num_nights * (price_per_night + price_pension + price_services)))
resp.append(Response.CONFIRM_BASIC)
self.last_question = resp if type(resp) is not list else resp[0]
return resp
# Get the reservation object from the fact database
@Callable
def reservation(self):
for fact in self.knowledge:
if type(fact) is Reservation:
return fact
# --------------------------------------------------------
# Return the suitable message based on the last and new (if exists) type of room chosed
@Callable
def response_from_room_types(self, last, new):
reserv = self.reservation()
print(last, new)
# It was already specified in the past...
if last is not None:
# ... and it matchs with the new info...
if last == new:
msg = Response.KNOWN_INFO
else:
msg = Response.CHANGE_ROOM_TYPE.replace("{room_type}", last).replace("{new_room_type}", new)
reserv.room_type = new
else:
msg = Response.CONFIRM_ROOM_TYPE.replace("{room_type}", new)
reserv.room_type = new
return msg
@Callable
def response_from_init_date(self, init_date):
reserv = self.reservation()
if self.db.free_room_from_dates(init_date, room_type=reserv.room_type):
reserv.init_date = init_date
msg = [Response.CONFIRM_INIT_DATE.replace("{date}", init_date)]
else:
room_type = reserv.room_type if reserv.room_type is not None else "habitaciones"
room_type += "s" if room_type != DBHandler.ROOM_INDIVIDUAL else "es"
msg = Response.NO_INIT_DATE.format(type=room_type)
return msg
# Handle the confirmation response when the bot shows the summary
@Callable
def confirm_reservation(self, user_resp):
db = DBHandler()
if user_resp == "yes":
if db.client_email(self.username):
self.last_question = Response.FINISH_RESERVATION
self.finish_reservation()
return Response.FINISH_RESERVATION
else:
self.last_question = Response.ASK_EMAIL
return Response.ASK_EMAIL
else:
return Response.ASK_WRONG_INFO
@Callable
def show_summary(self):
self.last_question = Response.SHOW_INTRO_SUMMARY
# Handle the confirmation response when the bot shows the summary
@Callable
def set_services(self, services):
reserv = self.reservation()
msg = []
for service in services:
if service == "additional_bed":
msg.append(Response.SERVICE_ADDITIONAL_BED)
reserv.additional_bed = True
elif service == "parking":
msg.append(Response.SERVICE_PARKING)
reserv.parking = True
elif service == "minibar":
msg.append(Response.SERVICE_MINIBAR)
if msg:
msg.append(Response.SERVICE_MORE)
else:
msg.append(Response.SERVICE_WHAT)
return msg
# Save current reservation (finished) in the DB and reset it in the facts base
@Callable
def finish_reservation(self):
reserv = self.reservation()
self.db.add_reservation(self.username, reserv)
email = self.db.client_email(self.username)
if email is not None:
mail = Email("[email protected]", "3m0j1Lun4")
print(os.getcwd())
print("images\\%s.jpg" % reserv.room_type)
print(os.getcwd() + "\\images\\%s.jpg" % reserv.room_type)
mail.send_email_with_attachments("[email protected]", email, "Resumen de reserva", reserv.summary(),
["images/%s.jpg" % reserv.room_type])
self._knowledge.remove(reserv)
self.learn(Reservation())
@Callable
def update_email(self, email):
self.db.update_email(self.username, email)
@Callable
def reply_to_pois_request(self, pois_types):
db = DBHandler()
# Make a method to retrieve the pois in a common format (with a optional param to filter)
msg = [Response.POIS_PLACES]
if pois_types:
for pois_type in pois_types:
for place in db.get_pois(pois_type):
if str(place['pois']['webpage']) is not None:
msg.append(
'\xF0\x9F\x9A\xA9' + str(place['pois']['name']).encode('utf-8') + '\xF0\x9F\x9A\xA9' +
'\n Descripción: ' + str(place['pois']['description']).encode('utf-8') +
'\n\n Dirección: ' + str(place['pois']['direction']).encode('utf-8') +
'\n\n ¿Cómo llegar?: ' + str(place['pois']['gmaps_url']).encode('utf-8') +
'\n\n Página web: ' + str(place['pois']['webpage']).encode('utf-8'))
else:
msg.append(
'\xF0\x9F\x9A\xA9' + str(place['pois']['name']).encode('utf-8') + '\xF0\x9F\x9A\xA9' +
'\n Descripción: ' + str(place['pois']['description']).encode('utf-8') +
'\n\n Dirección: ' + str(place['pois']['direction']).encode('utf-8') +
'\n\n ¿Cómo llegar?: ' + str(place['pois']['gmaps_url']).encode('utf-8'))
else:
for place in db.get_pois(limit=5):
if str(place['pois']['webpage']) is not None:
msg.append('\xF0\x9F\x9A\xA9' + str(place['pois']['name']).encode('utf-8') + '\xF0\x9F\x9A\xA9' +
'\n Descripción: ' + str(place['pois']['description']).encode('utf-8') +
'\n\n Dirección: ' + str(place['pois']['direction']).encode('utf-8') +
'\n\n ¿Cómo llegar?: ' + str(place['pois']['gmaps_url']).encode('utf-8') +
'\n\n Página web: ' + str(place['pois']['webpage']).encode('utf-8'))
else:
msg.append('\xF0\x9F\x9A\xA9' + str(place['pois']['name']).encode('utf-8') + '\xF0\x9F\x9A\xA9' +
'\n Descripción: ' + str(place['pois']['description']).encode('utf-8') +
'\n\n Dirección: ' + str(place['pois']['direction']).encode('utf-8') +
'\n\n ¿Cómo llegar?: ' + str(place['pois']['gmaps_url']).encode('utf-8'))
return msg
class HotelAgent:
def __init__(self, username):
try:
self.db = DBHandler()
self.db.check_client(username)
except:
print("The connection with the DB could not be established")
self.intellect = MyIntellect(username, self.db)
self.intellect.learn(Intellect.local_file_uri("./policies/hotel.policy"))
def evaluate(self, input):
msg = "Language: %s" % input.lang
msg += "\nSanitized: %s" % input.text_san
msg += "\nCorrected: %s" % input.text_sc
msg += "\nDateparser: %s" % input.dateparsed
msg += "\nTokenized: %s" % input.parsed
msg += "\nTagged: %s" % input.tagged
resp = None
desires = input.goals(self.intellect.last_question)
if desires is not None:
print("Desires: ")
for d in desires: print(d.id)
for desire in desires:
self.intellect.add_desire(desire)
self.intellect.reason()
resp = self.intellect.extract_response()
if resp is not None:
msg += "\nIntellect test: %s" % resp.msg
self.intellect.clear_desires() # Remove the current desires
if resp is None:
resp = Response(Response.UNKNOWN_INPUT)
# resp.msg = [msg]
print(msg) # Only for testing purposes
return 0.8, resp # trust, response
class InsultsAgent:
RACIST = ["negrata", "sudaca", "moro", "esclavo"]
MACHIST = ["zorra", "esclava", "maricon"]
GENERICS = ["puta", "puto", "cabron", "cabrona", "gilipollas"]
RESPONSE = ["No puedo seguir la conversación en este tono ", "La educación es lo primero",
"Por favor, no utilice palabras malsonantes",
"El tono de la conversación no debe seguir por este camino", "Hable con educacion",
"No utilice lenguaje soez"]
def evaluate(self, input):
if input.has_word(InsultsAgent.RACIST) or input.has_word(InsultsAgent.MACHIST) or input.has_word(
InsultsAgent.GENERICS):
return 1, Response(random.choice(InsultsAgent.RESPONSE))
return 0, Response(Response.UNKNOWN_INPUT)
class LanguagesAgent:
def evaluate(self, input):
if input.lang == "en":
return 0.9, Response("I only talk in spanish") # trust, response
elif input.lang == "de":
return 0.9, Response("Ich spreche nur Spanisch") # trust, response
elif input.lang == "it":
return 0.9, Response("Io solo parliamo spagnolo") # trust, response
elif input.lang == "fr":
return 0.9, Response("Je parle seulement espagnol") # trust, response
# elif input.lang == "pt":
# return 1, Response("Eu só falo espanhol") # trust, response
return 0, Response(Response.UNKNOWN_INPUT)