"""
- TelegramAPIError
- ValidationError
- Throttled
- BadRequest
- MessageError
- MessageNotModified
- MessageToForwardNotFound
- MessageIdInvalid
- MessageToDeleteNotFound
- MessageToPinNotFound
- MessageIdentifierNotSpecified
- MessageTextIsEmpty
- MessageCantBeEdited
- MessageCantBeDeleted
- MessageCantBeForwarded
- MessageToEditNotFound
- MessageToReplyNotFound
- ToMuchMessages
- PollError
- PollCantBeStopped
- PollHasAlreadyClosed
- PollsCantBeSentToPrivateChats
- PollSizeError
- PollMustHaveMoreOptions
- PollCantHaveMoreOptions
- PollsOptionsLengthTooLong
- PollOptionsMustBeNonEmpty
- PollQuestionMustBeNonEmpty
- MessageWithPollNotFound (with MessageError)
- MessageIsNotAPoll (with MessageError)
- ObjectExpectedAsReplyMarkup
- InlineKeyboardExpected
- ChatNotFound
- ChatDescriptionIsNotModified
- InvalidQueryID
- InvalidPeerID
- InvalidHTTPUrlContent
- ButtonURLInvalid
- URLHostIsEmpty
- StartParamInvalid
- ButtonDataInvalid
- FileIsTooBig
- WrongFileIdentifier
- GroupDeactivated
- BadWebhook
- WebhookRequireHTTPS
- BadWebhookPort
- BadWebhookAddrInfo
- BadWebhookNoAddressAssociatedWithHostname
- NotFound
- MethodNotKnown
- PhotoAsInputFileRequired
- InvalidStickersSet
- NoStickerInRequest
- ChatAdminRequired
- NeedAdministratorRightsInTheChannel
- MethodNotAvailableInPrivateChats
- CantDemoteChatCreator
- CantRestrictSelf
- NotEnoughRightsToRestrict
- PhotoDimensions
- UnavailableMembers
- TypeOfFileMismatch
- WrongRemoteFileIdSpecified
- PaymentProviderInvalid
- CurrencyTotalAmountInvalid
- CantParseUrl
- UnsupportedUrlProtocol
- CantParseEntities
- ResultIdDuplicate
- MethodIsNotAvailable
- ConflictError
- TerminatedByOtherGetUpdates
- CantGetUpdates
- Unauthorized
- BotKicked
- BotBlocked
- UserDeactivated
- CantInitiateConversation
- CantTalkWithBots
- NetworkError
- RetryAfter
- MigrateToChat
- RestartingTelegram
- AIOGramWarning
- TimeoutWarning
"""
import time
# TODO: Use exceptions detector from `aiograph`.
# TODO: aiogram.utils.exceptions.BadRequest: Bad request: can't parse entities: unsupported start tag "function" at byte offset 0
# TODO: aiogram.utils.exceptions.TelegramAPIError: Gateway Timeout
_PREFIXES = ['error: ', '[error]: ', 'bad request: ', 'conflict: ', 'not found: ']
def _clean_message(text):
for prefix in _PREFIXES:
if text.startswith(prefix):
text = text[len(prefix):]
return (text[0].upper() + text[1:]).strip()
[docs]class TelegramAPIError(Exception):
def __init__(self, message=None):
super(TelegramAPIError, self).__init__(_clean_message(message))
class _MatchErrorMixin:
match = ''
text = None
__subclasses = []
def __init_subclass__(cls, **kwargs):
super(_MatchErrorMixin, cls).__init_subclass__(**kwargs)
# cls.match = cls.match.lower() if cls.match else ''
if not hasattr(cls, f"_{cls.__name__}__group"):
cls.__subclasses.append(cls)
@classmethod
def check(cls, message) -> bool:
"""
Compare pattern with message
:param message: always must be in lowercase
:return: bool
"""
return cls.match.lower() in message
@classmethod
def detect(cls, description):
description = description.lower()
for err in cls.__subclasses:
if err is cls:
continue
if err.check(description):
raise err(cls.text or description)
raise cls(description)
[docs]class AIOGramWarning(Warning):
pass
[docs]class TimeoutWarning(AIOGramWarning):
pass
[docs]class FSMStorageWarning(AIOGramWarning):
pass
[docs]class ValidationError(TelegramAPIError):
pass
[docs]class BadRequest(TelegramAPIError, _MatchErrorMixin):
__group = True
[docs]class MessageError(BadRequest):
__group = True
[docs]class MessageNotModified(MessageError):
"""
Will be raised when you try to set new text is equals to current text.
"""
match = 'message is not modified'
[docs]class MessageToForwardNotFound(MessageError):
"""
Will be raised when you try to forward very old or deleted or unknown message.
"""
match = 'message to forward not found'
[docs]class MessageIdInvalid(MessageError):
text = 'Invalid message id'
match = 'message_id_invalid'
[docs]class MessageToDeleteNotFound(MessageError):
"""
Will be raised when you try to delete very old or deleted or unknown message.
"""
match = 'message to delete not found'
[docs]class MessageToPinNotFound(MessageError):
"""
Will be raised when you try to pin deleted or unknown message.
"""
match = 'message to pin not found'
[docs]class MessageToReplyNotFound(MessageError):
"""
Will be raised when you try to reply to very old or deleted or unknown message.
"""
match = 'Reply message not found'
[docs]class MessageIdentifierNotSpecified(MessageError):
match = 'message identifier is not specified'
[docs]class MessageTextIsEmpty(MessageError):
match = 'Message text is empty'
[docs]class MessageCantBeEdited(MessageError):
match = 'message can\'t be edited'
[docs]class MessageCantBeDeleted(MessageError):
match = 'message can\'t be deleted'
[docs]class MessageCantBeForwarded(MessageError):
match = 'message can\'t be forwarded'
[docs]class MessageToEditNotFound(MessageError):
match = 'message to edit not found'
[docs]class ToMuchMessages(MessageError):
"""
Will be raised when you try to send media group with more than 10 items.
"""
match = 'Too much messages to send as an album'
[docs]class ObjectExpectedAsReplyMarkup(BadRequest):
match = 'object expected as reply markup'
[docs]class InlineKeyboardExpected(BadRequest):
match = 'inline keyboard expected'
[docs]class PollError(BadRequest):
__group = True
[docs]class PollCantBeStopped(PollError):
match = "poll can't be stopped"
[docs]class PollHasAlreadyBeenClosed(PollError):
match = 'poll has already been closed'
[docs]class PollsCantBeSentToPrivateChats(PollError):
match = "polls can't be sent to private chats"
[docs]class PollSizeError(PollError):
__group = True
[docs]class PollMustHaveMoreOptions(PollSizeError):
match = "poll must have at least 2 option"
[docs]class PollCantHaveMoreOptions(PollSizeError):
match = "poll can't have more than 10 options"
[docs]class PollOptionsMustBeNonEmpty(PollSizeError):
match = "poll options must be non-empty"
[docs]class PollQuestionMustBeNonEmpty(PollSizeError):
match = "poll question must be non-empty"
[docs]class PollCanBeRequestedInPrivateChatsOnly(PollError):
match = "Poll can be requested in private chats only"
[docs]class MessageWithPollNotFound(PollError, MessageError):
"""
Will be raised when you try to stop poll with message without poll
"""
match = 'message with poll to stop not found'
[docs]class MessageIsNotAPoll(PollError, MessageError):
"""
Will be raised when you try to stop poll with message without poll
"""
match = 'message is not a poll'
[docs]class ChatNotFound(BadRequest):
match = 'chat not found'
[docs]class ChatIdIsEmpty(BadRequest):
match = 'chat_id is empty'
[docs]class InvalidUserId(BadRequest):
match = 'user_id_invalid'
text = 'Invalid user id'
[docs]class ChatDescriptionIsNotModified(BadRequest):
match = 'chat description is not modified'
[docs]class InvalidQueryID(BadRequest):
match = 'query is too old and response timeout expired or query id is invalid'
[docs]class InvalidPeerID(BadRequest):
match = 'PEER_ID_INVALID'
text = 'Invalid peer ID'
[docs]class InvalidHTTPUrlContent(BadRequest):
match = 'Failed to get HTTP URL content'
[docs]class URLHostIsEmpty(BadRequest):
match = 'URL host is empty'
[docs]class StartParamInvalid(BadRequest):
match = 'START_PARAM_INVALID'
text = 'Start param invalid'
[docs]class FileIsTooBig(BadRequest):
match = 'File is too big'
[docs]class WrongFileIdentifier(BadRequest):
match = 'wrong file identifier/HTTP URL specified'
[docs]class GroupDeactivated(BadRequest):
match = 'Group chat was deactivated'
[docs]class NoStickerInRequest(BadRequest):
match = 'there is no sticker in the request'
[docs]class ChatAdminRequired(BadRequest):
match = 'CHAT_ADMIN_REQUIRED'
text = 'Admin permissions is required!'
[docs]class NeedAdministratorRightsInTheChannel(BadRequest):
match = 'need administrator rights in the channel chat'
text = 'Admin permissions is required!'
[docs]class NotEnoughRightsToPinMessage(BadRequest):
match = 'not enough rights to pin a message'
[docs]class MethodNotAvailableInPrivateChats(BadRequest):
match = 'method is available only for supergroups and channel'
[docs]class CantDemoteChatCreator(BadRequest):
match = 'can\'t demote chat creator'
[docs]class CantRestrictSelf(BadRequest):
match = "can't restrict self"
text = "Admin can't restrict self."
[docs]class NotEnoughRightsToRestrict(BadRequest):
match = 'not enough rights to restrict/unrestrict chat member'
[docs]class PhotoDimensions(BadRequest):
match = 'PHOTO_INVALID_DIMENSIONS'
text = 'Invalid photo dimensions'
[docs]class UnavailableMembers(BadRequest):
match = 'supergroup members are unavailable'
[docs]class TypeOfFileMismatch(BadRequest):
match = 'type of file mismatch'
[docs]class WrongRemoteFileIdSpecified(BadRequest):
match = 'wrong remote file id specified'
[docs]class PaymentProviderInvalid(BadRequest):
match = 'PAYMENT_PROVIDER_INVALID'
text = 'payment provider invalid'
[docs]class CurrencyTotalAmountInvalid(BadRequest):
match = 'currency_total_amount_invalid'
text = 'currency total amount invalid'
[docs]class BadWebhook(BadRequest):
__group = True
[docs]class WebhookRequireHTTPS(BadWebhook):
match = 'HTTPS url must be provided for webhook'
text = 'bad webhook: ' + match
[docs]class BadWebhookPort(BadWebhook):
match = 'Webhook can be set up only on ports 80, 88, 443 or 8443'
text = 'bad webhook: ' + match
[docs]class BadWebhookAddrInfo(BadWebhook):
match = 'getaddrinfo: Temporary failure in name resolution'
text = 'bad webhook: ' + match
[docs]class BadWebhookNoAddressAssociatedWithHostname(BadWebhook):
match = 'failed to resolve host: no address associated with hostname'
[docs]class CantParseUrl(BadRequest):
match = 'can\'t parse URL'
[docs]class UnsupportedUrlProtocol(BadRequest):
match = 'unsupported URL protocol'
[docs]class CantParseEntities(BadRequest):
match = 'can\'t parse entities'
[docs]class ResultIdDuplicate(BadRequest):
match = 'result_id_duplicate'
text = 'Result ID duplicate'
[docs]class BotDomainInvalid(BadRequest):
match = 'bot_domain_invalid'
text = 'Invalid bot domain'
[docs]class MethodIsNotAvailable(BadRequest):
match = "Method is available only for supergroups"
[docs]class CantRestrictChatOwner(BadRequest):
"""
Raises when bot restricts the chat owner
"""
match = 'Can\'t remove chat owner'
[docs]class UserIsAnAdministratorOfTheChat(BadRequest):
"""
Raises when bot restricts the chat admin
"""
match = 'User is an administrator of the chat'
[docs]class NotFound(TelegramAPIError, _MatchErrorMixin):
__group = True
[docs]class MethodNotKnown(NotFound):
match = 'method not found'
[docs]class ConflictError(TelegramAPIError, _MatchErrorMixin):
__group = True
[docs]class TerminatedByOtherGetUpdates(ConflictError):
match = 'terminated by other getUpdates request'
text = 'Terminated by other getUpdates request; ' \
'Make sure that only one bot instance is running'
[docs]class CantGetUpdates(ConflictError):
match = 'can\'t use getUpdates method while webhook is active'
[docs]class Unauthorized(TelegramAPIError, _MatchErrorMixin):
__group = True
[docs]class BotKicked(Unauthorized):
match = 'bot was kicked from'
[docs]class BotBlocked(Unauthorized):
match = 'bot was blocked by the user'
[docs]class UserDeactivated(Unauthorized):
match = 'user is deactivated'
[docs]class CantInitiateConversation(Unauthorized):
match = 'bot can\'t initiate conversation with a user'
[docs]class CantTalkWithBots(Unauthorized):
match = 'bot can\'t send messages to bots'
[docs]class NetworkError(TelegramAPIError):
pass
[docs]class RestartingTelegram(TelegramAPIError):
def __init__(self):
super(RestartingTelegram, self).__init__('The Telegram Bot API service is restarting. Wait few second.')
[docs]class RetryAfter(TelegramAPIError):
def __init__(self, retry_after):
super(RetryAfter, self).__init__(f"Flood control exceeded. Retry in {retry_after} seconds.")
self.timeout = retry_after
[docs]class MigrateToChat(TelegramAPIError):
def __init__(self, chat_id):
super(MigrateToChat, self).__init__(f"The group has been migrated to a supergroup. New id: {chat_id}.")
self.migrate_to_chat_id = chat_id
[docs]class Throttled(TelegramAPIError):
def __init__(self, **kwargs):
from ..dispatcher.storage import DELTA, EXCEEDED_COUNT, KEY, LAST_CALL, RATE_LIMIT, RESULT
self.key = kwargs.pop(KEY, '<None>')
self.called_at = kwargs.pop(LAST_CALL, time.time())
self.rate = kwargs.pop(RATE_LIMIT, None)
self.result = kwargs.pop(RESULT, False)
self.exceeded_count = kwargs.pop(EXCEEDED_COUNT, 0)
self.delta = kwargs.pop(DELTA, 0)
self.user = kwargs.pop('user', None)
self.chat = kwargs.pop('chat', None)
def __str__(self):
return f"Rate limit exceeded! (Limit: {self.rate} s, " \
f"exceeded: {self.exceeded_count}, " \
f"time delta: {round(self.delta, 3)} s)"