Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
UPDATES_OFFSET_KEY = '_telegram_updates_offset'
try:
import telegram
except ImportError:
log.exception("Could not start the Telegram back-end")
log.fatal(
"You need to install the telegram support in order "
"to use the Telegram backend.\n"
"You should be able to install this package using:\n"
"pip install errbot[telegram]"
)
sys.exit(1)
class RoomsNotSupportedError(RoomError):
def __init__(self, message=None):
if message is None:
message = (
"Room operations are not supported on Telegram. "
"While Telegram itself has groupchat functionality, it does not "
"expose any APIs to bots to get group membership or otherwise "
"interact with groupchats."
)
super().__init__(message)
class TelegramBotFilter(object):
"""
This is a filter for the logging library that filters the
"No new updates found." log message generated by telegram.bot.
def destroy(self):
try:
if self.id.startswith('C'):
log.info("Archiving channel %s (%s)" % (str(self), self.id))
self._bot.api_call('channels.archive', data={'channel': self.id})
else:
log.info("Archiving group %s (%s)" % (str(self), self.id))
self._bot.api_call('groups.archive', data={'channel': self.id})
except SlackAPIResponseError as e:
if e.error == "user_is_bot":
raise RoomError("Unable to archive channel. " + USER_IS_BOT_HELPTEXT)
else:
raise RoomError(e)
self._id = None
def invite(self, *args):
users = {user['name']: user['id'] for user in self._bot.api_call('users.list')['members']}
for user in args:
if user not in users:
raise UserDoesNotExistError("User '%s' not found" % user)
log.info("Inviting %s into %s (%s)" % (user, str(self), self.id))
method = 'groups.invite' if self.private else 'channels.invite'
response = self._bot.api_call(
method,
data={'channel': self.id, 'user': users[user]},
raise_errors=False
)
if not response['ok']:
if response['error'] == "user_is_bot":
raise RoomError("Unable to invite people. " + USER_IS_BOT_HELPTEXT)
elif response['error'] != "already_in_channel":
raise SlackAPIResponseError(error="Slack API call to %s failed: %s" % (method, response['error']))
def create(self, private=False):
try:
if private:
log.info("Creating group %s" % str(self))
self._bot.api_call('groups.create', data={'name': self.name})
else:
log.info("Creating channel %s" % str(self))
self._bot.api_call('channels.create', data={'name': self.name})
except SlackAPIResponseError as e:
if e.error == "user_is_bot":
raise RoomError("Unable to create channel. " + USER_IS_BOT_HELPTEXT)
else:
raise RoomError(e)
def create(self, private=False):
try:
if private:
log.info("Creating group %s" % str(self))
self._bot.api_call('groups.create', data={'name': self.name})
else:
log.info("Creating channel %s" % str(self))
self._bot.api_call('channels.create', data={'name': self.name})
except SlackAPIResponseError as e:
if e.error == "user_is_bot":
raise RoomError("Unable to create channel. " + USER_IS_BOT_HELPTEXT)
else:
raise RoomError(e)
from typing import Iterable, Optional
from errbot.backends.base import Message
from errbot.backends.base import Person
from errbot.backends.base import Room, RoomError
from errbot.errBot import ErrBot
from google.cloud import pubsub
from oauth2client.service_account import ServiceAccountCredentials
from cachetools import LRUCache
from markdownconverter import hangoutschat_markdown_converter
log = logging.getLogger('errbot.backends.hangoutschat')
class RoomsNotSupportedError(RoomError):
def __init__(self, message=None):
if message is None:
message = (
"Most Room operations are not supported in Google Hangouts Chat."
"While Rooms are a _concept_, the API is minimal and does not "
"expose this functionality to bots"
)
super().__init__(message)
class GoogleHangoutsChatAPI:
"""
Represents the Google Hangouts REST API
See: https://developers.google.com/hangouts/chat/reference/rest/
"""
base_url = 'https://chat.googleapis.com/v1'
def join(self, username=None, password=None):
log.info("Joining channel %s" % str(self))
try:
self._bot.api_call('channels.join', data={'name': self.name})
except SlackAPIResponseError as e:
if e.error == "user_is_bot":
raise RoomError("Unable to join channel. " + USER_IS_BOT_HELPTEXT)
else:
raise RoomError(e)
def destroy(self):
try:
if self.id.startswith('C'):
log.info("Archiving channel %s (%s)" % (str(self), self.id))
self._bot.api_call('channels.archive', data={'channel': self.id})
else:
log.info("Archiving group %s (%s)" % (str(self), self.id))
self._bot.api_call('groups.archive', data={'channel': self.id})
except SlackAPIResponseError as e:
if e.error == "user_is_bot":
raise RoomError("Unable to archive channel. " + USER_IS_BOT_HELPTEXT)
else:
raise RoomError(e)
self._id = None
def invite(self, *args) -> None:
"""
Invite one or more people into the room.
:*args:
One or more identifiers to invite into the room.
"""
raise NotImplementedError("It should be implemented specifically for your backend")
class RoomError(Exception):
"""General exception class for MUC-related errors"""
class RoomNotJoinedError(RoomError):
"""Exception raised when performing MUC operations
that require the bot to have joined the room"""
class RoomDoesNotExistError(RoomError):
"""Exception that is raised when performing an operation
on a room that doesn't exist"""
class UserDoesNotExistError(Exception):
"""Exception that is raised when performing an operation
on a user that doesn't exist"""
class Message(object):
"""
def leave(self, reason=None):
try:
if self.id.startswith('C'):
log.info("Leaving channel %s (%s)" % (str(self), self.id))
self._bot.api_call('channels.leave', data={'channel': self.id})
else:
log.info("Leaving group %s (%s)" % (str(self), self.id))
self._bot.api_call('groups.leave', data={'channel': self.id})
except SlackAPIResponseError as e:
if e.error == "user_is_bot":
raise RoomError("Unable to leave channel. " + USER_IS_BOT_HELPTEXT)
else:
raise RoomError(e)
self._id = None