Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def on_chat_message(event):
if event.user.is_self:
# don't send my own messages
return
event_timestamp = event.timestamp
conversation_id = event.conv_id
conversation_name = get_conv_name(event.conv)
conversation_text = event.text
user_full_name = event.user.full_name
user_id = event.user_id
url = sender._config["HUBOT_URL"] + conversation_id
payload = {"from" : str(user_id.chat_id), "message" : conversation_text}
headers = {'content-type': 'application/json'}
r = requests.post(url, data = json.dumps(payload), headers = headers, verify=False)
bot.send_message(event.conv, 'Vote "{}" cancelled.'.format(UtilBot.get_vote_subject(event.conv_id)))
UtilBot.end_vote(event.conv_id)
else:
bot.send_message(event.conv, 'No vote currently started.')
return
# Starts a new vote
if not UtilBot.is_vote_started(event.conv_id) and set_vote == "start":
vote_subject = ' '.join(args)
vote_callback = None
# TODO Refactor this into a more easily extensible system.
if vote_subject.lower().strip() == "admin": # For the special Conversation Admin case.
vote_subject = '{} for Conversation Admin for chat {}'.format(event.user.full_name,
get_conv_name(event.conv))
def set_conv_admin(won):
if won:
try:
bot.config["conversations"][event.conv_id]["conversation_admin"] = event.user.id_[0]
except (KeyError, TypeError):
bot.config["conversations"][event.conv_id] = {}
bot.config["conversations"][event.conv_id]["admin"] = event.user.id_[0]
bot.config.save()
vote_callback = set_conv_admin
UtilBot.set_vote_subject(event.conv_id, vote_subject)
UtilBot.init_new_vote(event.conv_id, event.conv.users)
if vote_callback is not None:
UtilBot.set_vote_callback(event.conv_id, vote_callback)
))
if show_notification:
# We have to escape angle brackets because freedesktop.org
# notifications support markup.
if self._discreet_notification:
user = NOTIFY_ESCAPER("hangups")
message = NOTIFY_ESCAPER("New message")
else:
user = NOTIFY_ESCAPER(user.full_name)
message = NOTIFY_ESCAPER(conv_event.text)
cmd = [arg.format(
sender_name=user,
msg_text=message,
replaces_id=self._replaces_id,
convo_name=NOTIFY_ESCAPER(get_conv_name(conv)),
) for arg in NOTIFY_CMD]
# Run the notification and parse out the replaces_id. Since the
# command is a list of arguments, and we're not using a shell, this
# should be safe.
logger.info('Creating notification with command: {}'.format(cmd))
try:
output = subprocess.check_output(
cmd, stderr=subprocess.STDOUT
).decode()
except (subprocess.CalledProcessError, FileNotFoundError) as e:
# Only log this at INFO level to prevent log spam when gdbus
# isn't available.
logger.info('Notification command failed: {}'.format(e))
return
try:
def hangouts(bot, event, *args):
"""
**Hangouts:**
Usage: /hangouts
Purpose: Lists all Hangouts this Bot is currently in.
"""
segments = [hangups.ChatMessageSegment('Currently In These Hangouts:', is_bold=True),
hangups.ChatMessageSegment('\n', hangups.SegmentType.LINE_BREAK)]
for c in bot.list_conversations():
s = '{} [commands: {:d}, forwarding: {:d}, autoreplies: {:d}]'.format(get_conv_name(c, truncate=True),
bot.get_config_suboption(c.id_,
'commands_enabled'),
bot.get_config_suboption(c.id_,
'forwarding_enabled'),
bot.get_config_suboption(c.id_,
'autoreplies_enabled'))
segments.append(hangups.ChatMessageSegment(s))
segments.append(hangups.ChatMessageSegment('\n', hangups.SegmentType.LINE_BREAK))
bot.send_message_segments(event.conv, segments)
def print_conversations(self):
print('Conversations:')
for c in self.list_conversations():
print(' {} ({}) u:{}'.format(get_conv_name(c, truncate=True), c.id_, len(c.users)))
for u in c.users:
print(' {} ({}) {}'.format(u.first_name, u.full_name, u.id_.chat_id))
print()
def _on_connect(self, initial_data):
"""Handle connecting for the first time"""
print(_('Connected!'))
self._user_list = hangups.UserList(self._client,
initial_data.self_entity,
initial_data.entities,
initial_data.conversation_participants)
self._conv_list = hangups.ConversationList(self._client,
initial_data.conversation_states,
self._user_list,
initial_data.sync_timestamp)
self._conv_list.on_event.add_observer(self._on_event)
print(_('Conversations:'))
for c in self.list_conversations():
print(' {} ({})'.format(get_conv_name(c, truncate=True), c.id_))
print()
def conv_list(bot, event, conv_name='', *args):
"""List all conversations where bot is wreaking havoc
Usage: /bot conv_list [conversation_name]
Legend: c ... commands, f ... forwarding, a ... autoreplies"""
conv_name = strip_quotes(conv_name)
convs = bot.list_conversations() if not conv_name else bot.find_conversations(conv_name)
convs_text = []
for c in convs:
s = '{} [c: {:d}, f: {:d}, a: {:d}]'.format(
get_conv_name(c, truncate=True),
bot.get_config_suboption(c.id_, 'commands_enabled'),
bot.get_config_suboption(c.id_, 'forwarding_enabled'),
bot.get_config_suboption(c.id_, 'autoreplies_enabled')
)
convs_text.append(s)
text = _('**Active conversations:**\n'
'{}').format('\n'.join(convs_text))
yield from event.conv.send_message(text_to_segments(text))
def set_title(self):
"""Update this conversation's tab title."""
title = get_conv_name(self.conv, truncate=True)
conv_widget_id = self.tab_parent.conversationsTabWidget.indexOf(self)
num_unread = self.get_num_unread()
if num_unread > 0:
title += ' ({})'.format(num_unread)
self.tab_parent.conversationsTabWidget.tabBar().setTabTextColor(conv_widget_id, QtCore.Qt.darkBlue)
else:
self.tab_parent.conversationsTabWidget.tabBar().setTabTextColor(conv_widget_id, QtGui.QColor())
self.tab_parent.conversationsTabWidget.setTabText(conv_widget_id, title)
self.tab_parent.conversationsTabWidget.setTabToolTip(conv_widget_id, title)
def external_send_message_parsed(self, conversation_id, html):
conversation = self._conv_list.get(conversation_id)
print('sending parsed message, conversation name:', get_conv_name(conversation))
self.send_message_parsed(conversation, html)