forked from home-assistant/core
/
hangouts_bot.py
346 lines (297 loc) · 12.5 KB
/
hangouts_bot.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
340
341
342
343
344
345
346
"""The Hangouts Bot."""
import asyncio
import io
import logging
import aiohttp
import hangups
from hangups import ChatMessageEvent, ChatMessageSegment, Client, get_auth, hangouts_pb2
from homeassistant.helpers import dispatcher, intent
from homeassistant.helpers.aiohttp_client import async_get_clientsession
from .const import (
ATTR_DATA,
ATTR_MESSAGE,
ATTR_TARGET,
CONF_CONVERSATION_ID,
CONF_CONVERSATION_NAME,
CONF_CONVERSATIONS,
CONF_MATCHERS,
DOMAIN,
EVENT_HANGOUTS_CONNECTED,
EVENT_HANGOUTS_CONVERSATIONS_CHANGED,
EVENT_HANGOUTS_CONVERSATIONS_RESOLVED,
EVENT_HANGOUTS_DISCONNECTED,
EVENT_HANGOUTS_MESSAGE_RECEIVED,
INTENT_HELP,
)
from .hangups_utils import HangoutsCredentials, HangoutsRefreshToken
_LOGGER = logging.getLogger(__name__)
class HangoutsBot:
"""The Hangouts Bot."""
def __init__(
self, hass, refresh_token, intents, default_convs, error_suppressed_convs
):
"""Set up the client."""
self.hass = hass
self._connected = False
self._refresh_token = refresh_token
self._intents = intents
self._conversation_intents = None
self._client = None
self._user_list = None
self._conversation_list = None
self._default_convs = default_convs
self._default_conv_ids = None
self._error_suppressed_convs = error_suppressed_convs
self._error_suppressed_conv_ids = None
dispatcher.async_dispatcher_connect(
self.hass,
EVENT_HANGOUTS_MESSAGE_RECEIVED,
self._async_handle_conversation_message,
)
def _resolve_conversation_id(self, obj):
if CONF_CONVERSATION_ID in obj:
return obj[CONF_CONVERSATION_ID]
if CONF_CONVERSATION_NAME in obj:
conv = self._resolve_conversation_name(obj[CONF_CONVERSATION_NAME])
if conv is not None:
return conv.id_
return None
def _resolve_conversation_name(self, name):
for conv in self._conversation_list.get_all():
if conv.name == name:
return conv
return None
def async_update_conversation_commands(self):
"""Refresh the commands for every conversation."""
self._conversation_intents = {}
for intent_type, data in self._intents.items():
if data.get(CONF_CONVERSATIONS):
conversations = []
for conversation in data.get(CONF_CONVERSATIONS):
conv_id = self._resolve_conversation_id(conversation)
if conv_id is not None:
conversations.append(conv_id)
data["_" + CONF_CONVERSATIONS] = conversations
elif self._default_conv_ids:
data["_" + CONF_CONVERSATIONS] = self._default_conv_ids
else:
data["_" + CONF_CONVERSATIONS] = [
conv.id_ for conv in self._conversation_list.get_all()
]
for conv_id in data["_" + CONF_CONVERSATIONS]:
if conv_id not in self._conversation_intents:
self._conversation_intents[conv_id] = {}
self._conversation_intents[conv_id][intent_type] = data
try:
self._conversation_list.on_event.remove_observer(
self._async_handle_conversation_event
)
except ValueError:
pass
self._conversation_list.on_event.add_observer(
self._async_handle_conversation_event
)
def async_resolve_conversations(self, _):
"""Resolve the list of default and error suppressed conversations."""
self._default_conv_ids = []
self._error_suppressed_conv_ids = []
for conversation in self._default_convs:
conv_id = self._resolve_conversation_id(conversation)
if conv_id is not None:
self._default_conv_ids.append(conv_id)
for conversation in self._error_suppressed_convs:
conv_id = self._resolve_conversation_id(conversation)
if conv_id is not None:
self._error_suppressed_conv_ids.append(conv_id)
dispatcher.async_dispatcher_send(
self.hass, EVENT_HANGOUTS_CONVERSATIONS_RESOLVED
)
async def _async_handle_conversation_event(self, event):
if isinstance(event, ChatMessageEvent):
dispatcher.async_dispatcher_send(
self.hass,
EVENT_HANGOUTS_MESSAGE_RECEIVED,
event.conversation_id,
event.user_id,
event,
)
async def _async_handle_conversation_message(self, conv_id, user_id, event):
"""Handle a message sent to a conversation."""
user = self._user_list.get_user(user_id)
if user.is_self:
return
message = event.text
_LOGGER.debug("Handling message '%s' from %s", message, user.full_name)
intents = self._conversation_intents.get(conv_id)
if intents is not None:
is_error = False
try:
intent_result = await self._async_process(intents, message, conv_id)
except (intent.UnknownIntent, intent.IntentHandleError) as err:
is_error = True
intent_result = intent.IntentResponse()
intent_result.async_set_speech(str(err))
if intent_result is None:
is_error = True
intent_result = intent.IntentResponse()
intent_result.async_set_speech("Sorry, I didn't understand that")
message = (
intent_result.as_dict().get("speech", {}).get("plain", {}).get("speech")
)
if (message is not None) and not (
is_error and conv_id in self._error_suppressed_conv_ids
):
await self._async_send_message(
[{"text": message, "parse_str": True}],
[{CONF_CONVERSATION_ID: conv_id}],
None,
)
async def _async_process(self, intents, text, conv_id):
"""Detect a matching intent."""
for intent_type, data in intents.items():
for matcher in data.get(CONF_MATCHERS, []):
match = matcher.match(text)
if not match:
continue
if intent_type == INTENT_HELP:
return await self.hass.helpers.intent.async_handle(
DOMAIN, intent_type, {"conv_id": {"value": conv_id}}, text
)
return await self.hass.helpers.intent.async_handle(
DOMAIN,
intent_type,
{key: {"value": value} for key, value in match.groupdict().items()},
text,
)
async def async_connect(self):
"""Login to the Google Hangouts."""
session = await self.hass.async_add_executor_job(
get_auth,
HangoutsCredentials(None, None, None),
HangoutsRefreshToken(self._refresh_token),
)
self._client = Client(session)
self._client.on_connect.add_observer(self._on_connect)
self._client.on_disconnect.add_observer(self._on_disconnect)
self.hass.loop.create_task(self._client.connect())
def _on_connect(self):
_LOGGER.debug("Connected!")
self._connected = True
dispatcher.async_dispatcher_send(self.hass, EVENT_HANGOUTS_CONNECTED)
async def _on_disconnect(self):
"""Handle disconnecting."""
if self._connected:
_LOGGER.debug("Connection lost! Reconnect...")
await self.async_connect()
else:
dispatcher.async_dispatcher_send(self.hass, EVENT_HANGOUTS_DISCONNECTED)
async def async_disconnect(self):
"""Disconnect the client if it is connected."""
if self._connected:
self._connected = False
await self._client.disconnect()
async def async_handle_hass_stop(self, _):
"""Run once when Home Assistant stops."""
await self.async_disconnect()
async def _async_send_message(self, message, targets, data):
conversations = []
for target in targets:
conversation = None
if CONF_CONVERSATION_ID in target:
conversation = self._conversation_list.get(target[CONF_CONVERSATION_ID])
elif CONF_CONVERSATION_NAME in target:
conversation = self._resolve_conversation_name(
target[CONF_CONVERSATION_NAME]
)
if conversation is not None:
conversations.append(conversation)
if not conversations:
return False
messages = []
for segment in message:
if messages:
messages.append(
ChatMessageSegment(
"", segment_type=hangouts_pb2.SEGMENT_TYPE_LINE_BREAK
)
)
if "parse_str" in segment and segment["parse_str"]:
messages.extend(ChatMessageSegment.from_str(segment["text"]))
else:
if "parse_str" in segment:
del segment["parse_str"]
messages.append(ChatMessageSegment(**segment))
image_file = None
if data:
if data.get("image_url"):
uri = data.get("image_url")
try:
websession = async_get_clientsession(self.hass)
async with websession.get(uri, timeout=5) as response:
if response.status != 200:
_LOGGER.error(
"Fetch image failed, %s, %s", response.status, response
)
image_file = None
else:
image_data = await response.read()
image_file = io.BytesIO(image_data)
image_file.name = "image.png"
except (asyncio.TimeoutError, aiohttp.ClientError) as error:
_LOGGER.error("Failed to fetch image, %s", type(error))
image_file = None
elif data.get("image_file"):
uri = data.get("image_file")
if self.hass.config.is_allowed_path(uri):
try:
image_file = open(uri, "rb")
except OSError as error:
_LOGGER.error(
"Image file I/O error(%s): %s", error.errno, error.strerror
)
else:
_LOGGER.error('Path "%s" not allowed', uri)
if not messages:
return False
for conv in conversations:
await conv.send_message(messages, image_file)
async def _async_list_conversations(self):
(
self._user_list,
self._conversation_list,
) = await hangups.build_user_conversation_list(self._client)
conversations = {}
for i, conv in enumerate(self._conversation_list.get_all()):
users_in_conversation = []
for user in conv.users:
users_in_conversation.append(user.full_name)
conversations[str(i)] = {
CONF_CONVERSATION_ID: str(conv.id_),
CONF_CONVERSATION_NAME: conv.name,
"users": users_in_conversation,
}
self.hass.states.async_set(
f"{DOMAIN}.conversations",
len(self._conversation_list.get_all()),
attributes=conversations,
)
dispatcher.async_dispatcher_send(
self.hass, EVENT_HANGOUTS_CONVERSATIONS_CHANGED, conversations
)
async def async_handle_send_message(self, service):
"""Handle the send_message service."""
await self._async_send_message(
service.data[ATTR_MESSAGE],
service.data[ATTR_TARGET],
service.data.get(ATTR_DATA, {}),
)
async def async_handle_update_users_and_conversations(self, _=None):
"""Handle the update_users_and_conversations service."""
await self._async_list_conversations()
async def async_handle_reconnect(self, _=None):
"""Handle the reconnect service."""
await self.async_disconnect()
await self.async_connect()
def get_intents(self, conv_id):
"""Return the intents for a specific conversation."""
return self._conversation_intents.get(conv_id)