12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240224122422243224422452246224722482249225022512252225322542255225622572258225922602261226222632264226522662267226822692270227122722273227422752276227722782279228022812282228322842285228622872288228922902291229222932294229522962297229822992300230123022303230423052306230723082309231023112312231323142315231623172318231923202321232223232324232523262327232823292330233123322333 |
- """WARNING: this is only a legacy module.
- For newer versions use `bot.py`.
- This module relies on third party `telepot` library by Nick Lee (@Nickoala).
- The `telepot` repository was archived in may 2019 and will no longer be listed
- in requirements. For legacy code, install telepot manually.
- `pip install telepot`
- Subclass of third party telepot.aio.Bot, providing the following features.
- - It prevents hitting Telegram flood limits by waiting
- between text and photo messages.
- - It provides command, parser, button and other decorators to associate
- common Telegram actions with custom handlers.
- - It supports multiple bots running in the same script
- and allows communications between them
- as well as complete independency from each other.
- - Each bot is associated with a sqlite database
- using dataset, a third party library.
- Please note that you need Python3.5+ to run async code
- Check requirements.txt for third party dependencies.
- """
- # Standard library modules
- import asyncio
- import datetime
- import io
- import logging
- import os
- # Third party modules
- import dataset
- import telepot
- import telepot.aio
- # Project modules
- from davtelepot.utilities import (
- get_secure_key, Gettable, escape_html_chars, extract,
- line_drawing_unordered_list, make_lines_of_buttons, markdown_check, MyOD,
- pick_most_similar_from_list, remove_html_tags, sleep_until
- )
- def split_text_gracefully(text, limit, parse_mode):
- r"""Split text if it hits telegram limits for text messages.
- Split at `\n` if possible.
- Add a `[...]` at the end and beginning of split messages,
- with proper code markdown.
- """
- text = text.split("\n")[::-1]
- result = []
- while len(text) > 0:
- temp = []
- while len(text) > 0 and len("\n".join(temp + [text[-1]])) < limit:
- temp.append(text.pop())
- if len(temp) == 0:
- temp.append(text[-1][:limit])
- text[-1] = text[-1][limit:]
- result.append("\n".join(temp))
- if len(result) > 1:
- for i in range(1, len(result)):
- result[i] = "{tag[0]}[...]{tag[1]}\n{text}".format(
- tag=(
- ('`', '`') if parse_mode == 'Markdown'
- else ('<code>', '</code>') if parse_mode.lower() == 'html'
- else ('', '')
- ),
- text=result[i]
- )
- result[i-1] = "{text}\n{tag[0]}[...]{tag[1]}".format(
- tag=(
- ('`', '`') if parse_mode == 'Markdown'
- else ('<code>', '</code>') if parse_mode.lower() == 'html'
- else ('', '')
- ),
- text=result[i-1]
- )
- return result
- def make_inline_query_answer(answer):
- """Return an article-type answer to inline query.
- Takes either a string or a dictionary and returns a list.
- """
- if type(answer) is str:
- answer = dict(
- type='article',
- id=0,
- title=remove_html_tags(answer),
- input_message_content=dict(
- message_text=answer,
- parse_mode='HTML'
- )
- )
- if type(answer) is dict:
- answer = [answer]
- return answer
- class Bot(telepot.aio.Bot, Gettable):
- """telepot.aio.Bot (async Telegram bot framework) convenient subclass.
- === General functioning ===
- - While Bot.run() coroutine is executed, HTTP get requests are made
- to Telegram servers asking for new messages for each Bot instance.
- - Each message causes the proper Bot instance method coroutine
- to be awaited, according to its flavour (see routing_table)
- -- For example, chat messages cause `Bot().on_chat_message(message)`
- to be awaited.
- - This even-processing coroutine ensures the proper handling function
- a future and returns.
- -- That means that simpler tasks are completed before slower ones,
- since handling functions are not awaited but scheduled
- by `asyncio.ensure_future(handling_function(...))`
- -- For example, chat text messages are handled by
- `handle_text_message`, which looks for the proper function
- to elaborate the request (in bot's commands and parsers)
- - The handling function evaluates an answer, depending on the message
- content, and eventually provides a reply
- -- For example, `handle_text_message` sends its
- answer via `send_message`
- - All Bot.instances run simultaneously and faster requests
- are completed earlier.
- - All uncaught events are ignored.
- """
- instances = {}
- stop = False
- # Cooldown time between sent messages, to prevent hitting
- # Telegram flood limits
- # Current limits: 30 total messages sent per second,
- # 1 message per second per chat, 20 messages per minute per group
- COOLDOWN_TIME_ABSOLUTE = datetime.timedelta(seconds=1/30)
- COOLDOWN_TIME_PER_CHAT = datetime.timedelta(seconds=1)
- MAX_GROUP_MESSAGES_PER_MINUTE = 20
- # Max length of text field for a Telegram message (UTF-8 text)
- TELEGRAM_MESSAGES_MAX_LEN = 4096
- _path = '.'
- _unauthorized_message = None
- _unknown_command_message = None
- _maintenance_message = None
- _default_inline_query_answer = [
- dict(
- type='article',
- id=0,
- title="I cannot answer this query, sorry",
- input_message_content=dict(
- message_text="I'm sorry "
- "but I could not find an answer for your query."
- )
- )
- ]
- def __init__(self, token, db_name=None):
- """Instantiate Bot instance, given a token and a db name."""
- super().__init__(token)
- self.routing_table = {
- 'chat': self.on_chat_message,
- 'inline_query': self.on_inline_query,
- 'chosen_inline_result': self.on_chosen_inline_result,
- 'callback_query': self.on_callback_query
- }
- self.chat_message_handlers = {
- 'text': self.handle_text_message,
- 'pinned_message': self.handle_pinned_message,
- 'photo': self.handle_photo_message,
- 'location': self.handle_location
- }
- if db_name:
- self._db_url = 'sqlite:///{name}{ext}'.format(
- name=db_name,
- ext='.db' if not db_name.endswith('.db') else ''
- )
- self._database = dataset.connect(self.db_url)
- else:
- self._db_url = None
- self._database = None
- self._unauthorized_message = None
- self.authorization_function = lambda update, authorization_level: True
- self.get_chat_id = lambda update: (
- update['message']['chat']['id']
- if 'message' in update
- else update['chat']['id']
- )
- self.commands = dict()
- self.callback_handlers = dict()
- self.inline_query_handlers = MyOD()
- self._default_inline_query_answer = None
- self.chosen_inline_result_handlers = dict()
- self.aliases = MyOD()
- self.parsers = MyOD()
- self.custom_parsers = dict()
- self.custom_photo_parsers = dict()
- self.custom_location_parsers = dict()
- self.bot_name = None
- self.default_reply_keyboard_elements = []
- self._default_keyboard = dict()
- self.run_before_loop = []
- self.run_after_loop = []
- self.to_be_obscured = []
- self.to_be_destroyed = []
- self.last_sending_time = dict(
- absolute=(
- datetime.datetime.now()
- - self.__class__.COOLDOWN_TIME_ABSOLUTE
- )
- )
- self._maintenance = False
- self._maintenance_message = None
- self.chat_actions = dict(
- pinned=MyOD()
- )
- self.messages = dict()
- @property
- def name(self):
- """Bot name."""
- return self.bot_name
- @property
- def path(self):
- """custombot.py file path."""
- return self.__class__._path
- @property
- def db_url(self):
- """Return complete path to database."""
- return self._db_url
- @property
- def db(self):
- """Return the dataset.Database instance related to `self`.
- To start a transaction with bot's database, use a with statement:
- ```python3
- with bot.db as db:
- db['your_table'].insert(
- dict(
- name='John',
- age=45
- )
- )
- ```
- """
- return self._database
- @property
- def default_keyboard(self):
- """Get the default keyboard.
- It is sent when reply_markup is left blank and chat is private.
- """
- return self._default_keyboard
- @property
- def default_inline_query_answer(self):
- """Answer to be returned if inline query returned None."""
- if self._default_inline_query_answer:
- return self._default_inline_query_answer
- return self.__class__._default_inline_query_answer
- @property
- def unauthorized_message(self):
- """Return this if user is unauthorized to make a request.
- If instance message is not set, class message is returned.
- """
- if self._unauthorized_message:
- return self._unauthorized_message
- return self.__class__._unauthorized_message
- @property
- def unknown_command_message(self):
- """Message to be returned if user sends an unknown command.
- If instance message is not set, class message is returned.
- """
- if self._unknown_command_message:
- return self._unknown_command_message
- return self.__class__._unknown_command_message
- @property
- def maintenance(self):
- """Check whether bot is under maintenance.
- While under maintenance, bot will reply with
- `self.maintenance_message` to any request, with few exceptions.
- """
- return self._maintenance
- @property
- def maintenance_message(self):
- """Message to be returned if bot is under maintenance.
- If instance message is not set, class message is returned.
- """
- if self._maintenance_message:
- return self._maintenance_message
- if self.__class__.maintenance_message:
- return self.__class__._maintenance_message
- return "Bot is currently under maintenance! Retry later please."
- @classmethod
- def set_class_path(csl, path):
- """Set class path, where files will be looked for.
- For example, if send_photo receives `photo='mypic.png'`,
- it will parse it as `'{path}/mypic.png'.format(path=self.path)`
- """
- csl._path = path
- @classmethod
- def set_class_unauthorized_message(csl, unauthorized_message):
- """Set class unauthorized message.
- It will be returned if user is unauthorized to make a request.
- """
- csl._unauthorized_message = unauthorized_message
- @classmethod
- def set_class_unknown_command_message(cls, unknown_command_message):
- """Set class unknown command message.
- It will be returned if user sends an unknown command in private chat.
- """
- cls._unknown_command_message = unknown_command_message
- @classmethod
- def set_class_maintenance_message(cls, maintenance_message):
- """Set class maintenance message.
- It will be returned if bot is under maintenance.
- """
- cls._maintenance_message = maintenance_message
- @classmethod
- def set_class_default_inline_query_answer(cls,
- default_inline_query_answer):
- """Set class default inline query answer.
- It will be returned if an inline query returned no answer.
- """
- cls._default_inline_query_answer = default_inline_query_answer
- def set_unauthorized_message(self, unauthorized_message):
- """Set instance unauthorized message.
- If instance message is None, default class message is used.
- """
- self._unauthorized_message = unauthorized_message
- def set_unknown_command_message(self, unknown_command_message):
- """Set instance unknown command message.
- It will be returned if user sends an unknown command in private chat.
- If instance message is None, default class message is used.
- """
- self._unknown_command_message = unknown_command_message
- def set_maintenance_message(self, maintenance_message):
- """Set instance maintenance message.
- It will be returned if bot is under maintenance.
- If instance message is None, default class message is used.
- """
- self._maintenance_message = maintenance_message
- def set_default_inline_query_answer(self, default_inline_query_answer):
- """Set a custom default_inline_query_answer.
- It will be returned when no answer is found for an inline query.
- If instance answer is None, default class answer is used.
- """
- if type(default_inline_query_answer) in (str, dict):
- default_inline_query_answer = make_inline_query_answer(
- default_inline_query_answer
- )
- if type(default_inline_query_answer) is not list:
- return 1
- self._default_inline_query_answer = default_inline_query_answer
- return 0
- def set_maintenance(self, maintenance_message):
- """Put the bot under maintenance or ends it.
- While in maintenance, bot will reply to users with maintenance_message.
- Bot will accept /coma, /stop and /restart commands from admins.
- """
- self._maintenance = not self.maintenance
- if maintenance_message:
- self.set_maintenance_message(maintenance_message)
- if self.maintenance:
- return (
- "<i>Bot has just been put under maintenance!</i>\n\n"
- "Until further notice, it will reply to users "
- "with the following message:\n\n{}"
- ).format(
- self.maintenance_message
- )
- return "<i>Maintenance ended!</i>"
- def set_authorization_function(self, authorization_function):
- """Set a custom authorization_function.
- It should evaluate True if user is authorized to perform
- a specific action and False otherwise.
- It should take update and role and return a Boolean.
- Default authorization_function always evaluates True.
- """
- self.authorization_function = authorization_function
- def set_get_chat_id_function(self, get_chat_id_function):
- """Set a custom get_chat_id function.
- It should take and update and return the chat in which
- a reply should be sent.
- For instance, a bot could reply in private to group messages
- as a default behaviour.
- Default chat_id returned is current chat id.
- """
- self.get_chat_id = get_chat_id_function
- async def avoid_flooding(self, chat_id):
- """asyncio-sleep until COOLDOWN_TIME has passed.
- To prevent hitting Telegram flood limits, send_message and
- send_photo await this function.
- Consider cooldown time per chat and absolute.
- """
- if type(chat_id) is int and chat_id > 0:
- while (
- datetime.datetime.now() < (
- self.last_sending_time['absolute']
- + self.__class__.COOLDOWN_TIME_ABSOLUTE
- )
- ) or (
- chat_id in self.last_sending_time
- and (
- datetime.datetime.now() < (
- self.last_sending_time[chat_id]
- + self.__class__.COOLDOWN_TIME_PER_CHAT
- )
- )
- ):
- await asyncio.sleep(
- self.__class__.COOLDOWN_TIME_ABSOLUTE.seconds
- )
- self.last_sending_time[chat_id] = datetime.datetime.now()
- else:
- while (
- datetime.datetime.now() < (
- self.last_sending_time['absolute']
- + self.__class__.COOLDOWN_TIME_ABSOLUTE
- )
- ) or (
- chat_id in self.last_sending_time
- and len(
- [
- sending_datetime
- for sending_datetime in self.last_sending_time[chat_id]
- if sending_datetime >= (
- datetime.datetime.now()
- - datetime.timedelta(minutes=1)
- )
- ]
- ) >= self.__class__.MAX_GROUP_MESSAGES_PER_MINUTE
- ) or (
- chat_id in self.last_sending_time
- and len(self.last_sending_time[chat_id]) > 0
- and datetime.datetime.now() < (
- self.last_sending_time[chat_id][-1]
- + self.__class__.COOLDOWN_TIME_PER_CHAT
- )
- ):
- await asyncio.sleep(0.5)
- if chat_id not in self.last_sending_time:
- self.last_sending_time[chat_id] = []
- self.last_sending_time[chat_id].append(datetime.datetime.now())
- self.last_sending_time[chat_id] = [
- sending_datetime
- for sending_datetime in self.last_sending_time[chat_id]
- if sending_datetime >= (
- datetime.datetime.now()
- - datetime.timedelta(minutes=1)
- )
- ]
- self.last_sending_time['absolute'] = datetime.datetime.now()
- return
- def get_message(self, *fields, update=None, user_record=None,
- language=None, **format_kwargs):
- """Given a list of strings (`fields`), return proper message.
- If `language` is not passed, it is extracted from `update`.
- If `update` is not passed either, `language` is extracted from
- `user_record`.
- Fall back to English message if language is not available.
- Pass `format_kwargs` to format function.
- """
- if (
- language is None
- and isinstance(update, dict)
- and 'from' in update
- and 'language_code' in update['from']
- ):
- language = update['from']['language_code']
- if (
- language is None
- and isinstance(user_record, dict)
- and 'language_code' in user_record
- ):
- language = user_record['language_code']
- if language is None:
- language = 'en'
- result = self.messages
- for field in fields:
- if field not in result:
- logging.error(
- "Please define self.message{f}".format(
- f=''.join(
- '[\'{field}\']'.format(
- field=field
- )
- for field in fields
- )
- )
- )
- return "Invalid message!"
- result = result[field]
- if language not in result:
- language = extract(
- language,
- ender='-'
- )
- if language not in result:
- language = 'en'
- if language not in result:
- logging.error(
- "Please define self.message{f}['en']".format(
- f=''.join(
- '[\'{field}\']'.format(
- field=field
- )
- for field in fields
- )
- )
- )
- return "Invalid message!"
- return result[language].format(
- **format_kwargs
- )
- async def on_inline_query(self, update):
- """Schedule handling of received inline queries.
- Notice that handling is only scheduled, not awaited.
- This means that all Bot instances may now handle other requests
- before this one is completed.
- """
- asyncio.ensure_future(self.handle_inline_query(update))
- return
- async def on_chosen_inline_result(self, update):
- """Schedule handling of received chosen inline result events.
- Notice that handling is only scheduled, not awaited.
- This means that all Bot instances may now handle other requests
- before this one is completed.
- """
- asyncio.ensure_future(self.handle_chosen_inline_result(update))
- return
- async def on_callback_query(self, update):
- """Schedule handling of received callback queries.
- A callback query is sent when users press inline keyboard buttons.
- Bad clients may send malformed or deceiving callback queries:
- never use secret keys in buttons and always check request validity!
- Notice that handling is only scheduled, not awaited.
- This means that all Bot instances may now handle other requests
- before this one is completed.
- """
- # Reject malformed updates lacking of data field
- if 'data' not in update:
- return
- asyncio.ensure_future(self.handle_callback_query(update))
- return
- async def on_chat_message(self, update):
- """Schedule handling of received chat message.
- Notice that handling is only scheduled, not awaited.
- According to update type, the corresponding handler is
- scheduled (see self.chat_message_handlers).
- This means that all Bot instances may now handle other
- requests before this one is completed.
- """
- answer = None
- content_type, chat_type, chat_id = telepot.glance(
- update,
- flavor='chat',
- long=False
- )
- if content_type in self.chat_message_handlers:
- answer = asyncio.ensure_future(
- self.chat_message_handlers[content_type](update)
- )
- else:
- answer = None
- logging.debug("Unhandled message")
- return answer
- async def handle_inline_query(self, update):
- """Handle inline query and answer it with results, or log errors."""
- query = update['query']
- answer = None
- switch_pm_text, switch_pm_parameter = None, None
- if self.maintenance:
- answer = self.maintenance_message
- else:
- for condition, handler in self.inline_query_handlers.items():
- answerer = handler['function']
- if condition(query):
- if asyncio.iscoroutinefunction(answerer):
- answer = await answerer(update)
- else:
- answer = answerer(update)
- break
- if not answer:
- answer = self.default_inline_query_answer
- if type(answer) is dict:
- if 'switch_pm_text' in answer:
- switch_pm_text = answer['switch_pm_text']
- if 'switch_pm_parameter' in answer:
- switch_pm_parameter = answer['switch_pm_parameter']
- answer = answer['answer']
- if type(answer) is str:
- answer = make_inline_query_answer(answer)
- try:
- await self.answerInlineQuery(
- update['id'],
- answer,
- cache_time=10,
- is_personal=True,
- switch_pm_text=switch_pm_text,
- switch_pm_parameter=switch_pm_parameter
- )
- except Exception as e:
- logging.info("Error answering inline query\n{}".format(e))
- return
- async def handle_chosen_inline_result(self, update):
- """When an inline query result is chosen, perform an action.
- If chosen inline result id is in self.chosen_inline_result_handlers,
- call the related function passing the update as argument.
- """
- user_id = update['from']['id'] if 'from' in update else None
- if self.maintenance:
- return
- if user_id in self.chosen_inline_result_handlers:
- result_id = update['result_id']
- handlers = self.chosen_inline_result_handlers[user_id]
- if result_id in handlers:
- func = handlers[result_id]
- if asyncio.iscoroutinefunction(func):
- await func(update)
- else:
- func(update)
- return
- def set_inline_result_handler(self, user_id, result_id, func):
- """Associate a func to a result_id.
- When an inline result is chosen having that id, function will
- be passed the update as argument.
- """
- if type(user_id) is dict:
- user_id = user_id['from']['id']
- assert type(user_id) is int, "user_id must be int!"
- # Query result ids are parsed as str by telegram
- result_id = str(result_id)
- assert callable(func), "func must be a callable"
- if user_id not in self.chosen_inline_result_handlers:
- self.chosen_inline_result_handlers[user_id] = {}
- self.chosen_inline_result_handlers[user_id][result_id] = func
- return
- async def handle_callback_query(self, update):
- """Answer callback queries.
- Call the callback handler associated to the query prefix.
- The answer is used to edit the source message or send new ones
- if text is longer than single message limit.
- Anyway, the query is answered, otherwise the client would hang and
- the bot would look like idle.
- """
- answer = None
- if self.maintenance:
- answer = remove_html_tags(self.maintenance_message[:45])
- else:
- data = update['data']
- for start_text, handler in self.callback_handlers.items():
- answerer = handler['function']
- if data.startswith(start_text):
- if asyncio.iscoroutinefunction(answerer):
- answer = await answerer(update)
- else:
- answer = answerer(update)
- break
- if answer:
- if type(answer) is str:
- answer = {'text': answer}
- if type(answer) is not dict:
- return
- if 'edit' in answer:
- if 'message' in update:
- message_identifier = telepot.message_identifier(
- update['message']
- )
- else:
- message_identifier = telepot.message_identifier(update)
- edit = answer['edit']
- reply_markup = (
- edit['reply_markup']
- if 'reply_markup' in edit
- else None
- )
- text = (
- edit['text']
- if 'text' in edit
- else None
- )
- caption = (
- edit['caption']
- if 'caption' in edit
- else None
- )
- parse_mode = (
- edit['parse_mode']
- if 'parse_mode' in edit
- else None
- )
- disable_web_page_preview = (
- edit['disable_web_page_preview']
- if 'disable_web_page_preview' in edit
- else None
- )
- try:
- if 'text' in edit:
- if (
- len(text)
- > self.__class__.TELEGRAM_MESSAGES_MAX_LEN - 200
- ):
- if 'from' in update:
- await self.send_message(
- chat_id=update['from']['id'],
- text=text,
- reply_markup=reply_markup,
- parse_mode=parse_mode,
- disable_web_page_preview=(
- disable_web_page_preview
- )
- )
- else:
- await self.editMessageText(
- msg_identifier=message_identifier,
- text=text,
- parse_mode=parse_mode,
- disable_web_page_preview=(
- disable_web_page_preview
- ),
- reply_markup=reply_markup
- )
- elif 'caption' in edit:
- await self.editMessageCaption(
- msg_identifier=message_identifier,
- caption=caption,
- reply_markup=reply_markup
- )
- elif 'reply_markup' in edit:
- await self.editMessageReplyMarkup(
- msg_identifier=message_identifier,
- reply_markup=reply_markup
- )
- except Exception as e:
- logging.info("Message was not modified:\n{}".format(e))
- text = answer['text'][:180] if 'text' in answer else None
- show_alert = (
- answer['show_alert']
- if 'show_alert' in answer
- else None
- )
- cache_time = (
- answer['cache_time']
- if 'cache_time' in answer
- else None
- )
- try:
- await self.answerCallbackQuery(
- callback_query_id=update['id'],
- text=text,
- show_alert=show_alert,
- cache_time=cache_time
- )
- except telepot.exception.TelegramError as e:
- logging.error(e)
- else:
- try:
- await self.answerCallbackQuery(callback_query_id=update['id'])
- except telepot.exception.TelegramError as e:
- logging.error(e)
- return
- async def handle_text_message(self, update):
- """Answer to chat text messages.
- 1) Ignore bot name (case-insensitive) and search bot custom parsers,
- commands, aliases and parsers for an answerer.
- 2) Get an answer from answerer(update).
- 3) Send it to the user.
- """
- answerer, answer = None, None
- # Lower text and replace only bot's tag,
- # meaning that `/command@OtherBot` will be ignored.
- text = update['text'].lower().replace(
- '@{}'.format(
- self.name.lower()
- ),
- ''
- )
- user_id = update['from']['id'] if 'from' in update else None
- if self.maintenance and not any(
- text.startswith(x)
- for x in ('/coma', '/restart')
- ):
- if update['chat']['id'] > 0:
- answer = self.maintenance_message
- elif user_id in self.custom_parsers:
- answerer = self.custom_parsers[user_id]
- del self.custom_parsers[user_id]
- elif text.startswith('/'):
- command = text.split()[0].strip(' /@')
- if command in self.commands:
- answerer = self.commands[command]['function']
- elif update['chat']['id'] > 0:
- answer = self.unknown_command_message
- else:
- # If text starts with an alias
- # Aliases are case insensitive: text and alias are both .lower()
- for alias, parser in self.aliases.items():
- if text.startswith(alias.lower()):
- answerer = parser
- break
- # If update matches any parser
- for check_function, parser in self.parsers.items():
- if (
- parser['argument'] == 'text'
- and check_function(text)
- ) or (
- parser['argument'] == 'update'
- and check_function(update)
- ):
- answerer = parser['function']
- break
- if answerer:
- if asyncio.iscoroutinefunction(answerer):
- answer = await answerer(update)
- else:
- answer = answerer(update)
- if answer:
- try:
- return await self.send_message(answer=answer, chat_id=update)
- except Exception as e:
- logging.error(
- "Failed to process answer:\n{}".format(e),
- exc_info=True
- )
- async def handle_pinned_message(self, update):
- """Handle pinned message chat action."""
- if self.maintenance:
- return
- answerer = None
- for criteria, handler in self.chat_actions['pinned'].items():
- if criteria(update):
- answerer = handler['function']
- break
- if answerer is None:
- return
- elif asyncio.iscoroutinefunction(answerer):
- answer = await answerer(update)
- else:
- answer = answerer(update)
- if answer:
- try:
- return await self.send_message(
- answer=answer,
- chat_id=update['chat']['id']
- )
- except Exception as e:
- logging.error(
- "Failed to process answer:\n{}".format(
- e
- ),
- exc_info=True
- )
- return
- async def handle_photo_message(self, update):
- """Handle photo chat message."""
- user_id = update['from']['id'] if 'from' in update else None
- answerer, answer = None, None
- if self.maintenance:
- if update['chat']['id'] > 0:
- answer = self.maintenance_message
- elif user_id in self.custom_photo_parsers:
- answerer = self.custom_photo_parsers[user_id]
- del self.custom_photo_parsers[user_id]
- if answerer:
- if asyncio.iscoroutinefunction(answerer):
- answer = await answerer(update)
- else:
- answer = answerer(update)
- if answer:
- try:
- return await self.send_message(answer=answer, chat_id=update)
- except Exception as e:
- logging.error(
- "Failed to process answer:\n{}".format(
- e
- ),
- exc_info=True
- )
- return
- async def handle_location(self, update):
- """Handle location sent by user."""
- user_id = update['from']['id'] if 'from' in update else None
- answerer, answer = None, None
- if self.maintenance:
- if update['chat']['id'] > 0:
- answer = self.maintenance_message
- elif user_id in self.custom_location_parsers:
- answerer = self.custom_location_parsers[user_id]
- del self.custom_location_parsers[user_id]
- if answerer:
- if asyncio.iscoroutinefunction(answerer):
- answer = await answerer(update)
- else:
- answer = answerer(update)
- if answer:
- try:
- return await self.send_message(answer=answer, chat_id=update)
- except Exception as e:
- logging.error(
- "Failed to process answer:\n{}".format(
- e
- ),
- exc_info=True
- )
- return
- def set_custom_parser(self, parser, update=None, user=None):
- """Set a custom parser for the user.
- Any chat message update coming from the user will be handled by
- this custom parser instead of default parsers (commands, aliases
- and text parsers).
- Custom parsers last one single use, but their handler can call this
- function to provide multiple tries.
- """
- if user and type(user) is int:
- pass
- elif type(update) is int:
- user = update
- elif type(user) is dict:
- user = (
- user['from']['id']
- if 'from' in user
- and 'id' in user['from']
- else None
- )
- elif not user and type(update) is dict:
- user = (
- update['from']['id']
- if 'from' in update
- and 'id' in update['from']
- else None
- )
- else:
- raise TypeError(
- 'Invalid user.\nuser: {}\nupdate: {}'.format(
- user,
- update
- )
- )
- if not type(user) is int:
- raise TypeError(
- 'User {} is not an int id'.format(
- user
- )
- )
- if not callable(parser):
- raise TypeError(
- 'Parser {} is not a callable'.format(
- parser.__name__
- )
- )
- self.custom_parsers[user] = parser
- return
- def set_custom_photo_parser(self, parser, update=None, user=None):
- """Set a custom photo parser for the user.
- Any photo chat update coming from the user will be handled by
- this custom parser instead of default parsers.
- Custom photo parsers last one single use, but their handler can
- call this function to provide multiple tries.
- """
- if user and type(user) is int:
- pass
- elif type(update) is int:
- user = update
- elif type(user) is dict:
- user = (
- user['from']['id']
- if 'from' in user
- and 'id' in user['from']
- else None
- )
- elif not user and type(update) is dict:
- user = (
- update['from']['id']
- if 'from' in update
- and 'id' in update['from']
- else None
- )
- else:
- raise TypeError(
- 'Invalid user.\nuser: {}\nupdate: {}'.format(
- user,
- update
- )
- )
- if not type(user) is int:
- raise TypeError(
- 'User {} is not an int id'.format(
- user
- )
- )
- if not callable(parser):
- raise TypeError(
- 'Parser {} is not a callable'.format(
- parser.__name__
- )
- )
- self.custom_photo_parsers[user] = parser
- return
- def set_custom_location_parser(self, parser, update=None, user=None):
- """Set a custom location parser for the user.
- Any location chat update coming from the user will be handled by
- this custom parser instead of default parsers.
- Custom location parsers last one single use, but their handler can
- call this function to provide multiple tries.
- """
- if user and type(user) is int:
- pass
- elif type(update) is int:
- user = update
- elif type(user) is dict:
- user = (
- user['from']['id']
- if 'from' in user
- and 'id' in user['from']
- else None
- )
- elif not user and type(update) is dict:
- user = (
- update['from']['id']
- if 'from' in update
- and 'id' in update['from']
- else None
- )
- else:
- raise TypeError(
- 'Invalid user.\nuser: {}\nupdate: {}'.format(
- user,
- update
- )
- )
- if not type(user) is int:
- raise TypeError(
- 'User {} is not an int id'.format(
- user
- )
- )
- if not callable(parser):
- raise TypeError(
- 'Parser {} is not a callable'.format(
- parser.__name__
- )
- )
- self.custom_location_parsers[user] = parser
- return
- def command(self, command, aliases=None, show_in_keyboard=False,
- descr="", auth='admin'):
- """Define a bot command.
- Decorator: `@bot.command(*args)`
- When a message text starts with `/command[@bot_name]`, or with an
- alias, it gets passed to the decorated function.
- `command` is the command name (with or without /)
- `aliases` is a list of aliases
- `show_in_keyboard`, if True, makes first alias appear
- in default_keyboard
- `descr` is a description
- `auth` is the lowest authorization level needed to run the command
- """
- command = command.replace('/', '').lower()
- if not isinstance(command, str):
- raise TypeError('Command {} is not a string'.format(command))
- if aliases:
- if not isinstance(aliases, list):
- raise TypeError('Aliases is not a list: {}'.format(aliases))
- for alias in aliases:
- if not isinstance(alias, str):
- raise TypeError('Alias {} is not a string'.format(alias))
- def decorator(func):
- if asyncio.iscoroutinefunction(func):
- async def decorated(message):
- logging.info(
- "COMMAND({c}) @{n} FROM({f})".format(
- c=command,
- n=self.name,
- f=(
- message['from']
- if 'from' in message
- else message['chat']
- )
- )
- )
- if self.authorization_function(message, auth):
- return await func(message)
- return self.unauthorized_message
- else:
- def decorated(message):
- logging.info(
- "COMMAND({c}) @{n} FROM({f})".format(
- c=command,
- n=self.name,
- f=(
- message['from']
- if 'from' in message
- else message['chat']
- )
- )
- )
- if self.authorization_function(message, auth):
- return func(message)
- return self.unauthorized_message
- self.commands[command] = dict(
- function=decorated,
- descr=descr,
- auth=auth
- )
- if aliases:
- for alias in aliases:
- self.aliases[alias] = decorated
- if show_in_keyboard:
- self.default_reply_keyboard_elements.append(aliases[0])
- return decorator
- def parser(self, condition, descr='', auth='admin', argument='text'):
- """Define a message parser.
- Decorator: `@bot.parser(condition)`
- If condition evaluates True when run on a message text
- (not starting with '/'), such decorated function gets
- called on update.
- Conditions of parsers are evaluated in order; when one is True,
- others will be skipped.
- `descr` is a description
- `auth` is the lowest authorization level needed to run the command
- """
- if not callable(condition):
- raise TypeError(
- 'Condition {} is not a callable'.format(
- condition.__name__
- )
- )
- def decorator(func):
- if asyncio.iscoroutinefunction(func):
- async def decorated(message):
- logging.info(
- "TEXT MATCHING CONDITION({c}) @{n} FROM({f})".format(
- c=condition.__name__,
- n=self.name,
- f=(
- message['from']
- if 'from' in message
- else message['chat']
- )
- )
- )
- if self.authorization_function(message, auth):
- return await func(message)
- return self.unauthorized_message
- else:
- def decorated(message):
- logging.info(
- "TEXT MATCHING CONDITION({c}) @{n} FROM({f})".format(
- c=condition.__name__,
- n=self.name,
- f=(
- message['from']
- if 'from' in message
- else message['chat']
- )
- )
- )
- if self.authorization_function(message, auth):
- return func(message)
- return self.unauthorized_message
- self.parsers[condition] = dict(
- function=decorated,
- descr=descr,
- auth=auth,
- argument=argument
- )
- return decorator
- def pinned(self, condition, descr='', auth='admin'):
- """Handle pinned messages.
- Decorator: `@bot.pinned(condition)`
- If condition evaluates True when run on a pinned_message update,
- such decorated function gets called on update.
- Conditions are evaluated in order; when one is True,
- others will be skipped.
- `descr` is a description
- `auth` is the lowest authorization level needed to run the command
- """
- if not callable(condition):
- raise TypeError(
- 'Condition {c} is not a callable'.format(
- c=condition.__name__
- )
- )
- def decorator(func):
- if asyncio.iscoroutinefunction(func):
- async def decorated(message):
- logging.info(
- "PINNED MESSAGE MATCHING({c}) @{n} FROM({f})".format(
- c=condition.__name__,
- n=self.name,
- f=(
- message['from']
- if 'from' in message
- else message['chat']
- )
- )
- )
- if self.authorization_function(message, auth):
- return await func(message)
- return
- else:
- def decorated(message):
- logging.info(
- "PINNED MESSAGE MATCHING({c}) @{n} FROM({f})".format(
- c=condition.__name__,
- n=self.name,
- f=(
- message['from']
- if 'from' in message
- else message['chat']
- )
- )
- )
- if self.authorization_function(message, auth):
- return func(message)
- return
- self.chat_actions['pinned'][condition] = dict(
- function=decorated,
- descr=descr,
- auth=auth
- )
- return decorator
- def button(self, data, descr='', auth='admin'):
- """Define a bot button.
- Decorator: `@bot.button('example:///')`
- When a callback data text starts with <data>, it gets passed to the
- decorated function
- `descr` is a description
- `auth` is the lowest authorization level needed to run the command
- """
- if not isinstance(data, str):
- raise TypeError(
- 'Inline button callback_data {d} is not a string'.format(
- d=data
- )
- )
- def decorator(func):
- if asyncio.iscoroutinefunction(func):
- async def decorated(message):
- logging.info(
- "INLINE BUTTON({d}) @{n} FROM({f})".format(
- d=message['data'],
- n=self.name,
- f=(
- message['from']
- )
- )
- )
- if self.authorization_function(message, auth):
- return await func(message)
- return self.unauthorized_message
- else:
- def decorated(message):
- logging.info(
- "INLINE BUTTON({d}) @{n} FROM({f})".format(
- d=message['data'],
- n=self.name,
- f=(
- message['from']
- )
- )
- )
- if self.authorization_function(message, auth):
- return func(message)
- return self.unauthorized_message
- self.callback_handlers[data] = dict(
- function=decorated,
- descr=descr,
- auth=auth
- )
- return decorator
- def query(self, condition, descr='', auth='admin'):
- """Define an inline query.
- Decorator: `@bot.query(example)`
- When an inline query matches the `condition` function,
- decorated function is called and passed the query update object
- as argument.
- `descr` is a description
- `auth` is the lowest authorization level needed to run the command
- """
- if not callable(condition):
- raise TypeError(
- 'Condition {c} is not a callable'.format(
- c=condition.__name__
- )
- )
- def decorator(func):
- if asyncio.iscoroutinefunction(func):
- async def decorated(message):
- logging.info(
- "QUERY MATCHING CONDITION({c}) @{n} FROM({f})".format(
- c=condition.__name__,
- n=self.name,
- f=message['from']
- )
- )
- if self.authorization_function(message, auth):
- return await func(message)
- return self.unauthorized_message
- else:
- def decorated(message):
- logging.info(
- "QUERY MATCHING CONDITION({c}) @{n} FROM({f})".format(
- c=condition.__name__,
- n=self.name,
- f=message['from']
- )
- )
- if self.authorization_function(message, auth):
- return func(message)
- return self.unauthorized_message
- self.inline_query_handlers[condition] = dict(
- function=decorated,
- descr=descr,
- auth=auth
- )
- return decorator
- def additional_task(self, when='BEFORE'):
- """Add a task before or after message_loop.
- Decorator: such decorated async functions get awaited BEFORE or
- AFTER messageloop
- """
- when = when[0].lower()
- def decorator(func):
- if when == 'b':
- self.run_before_loop.append(func)
- elif when == 'a':
- self.run_after_loop.append(func)
- return decorator
- def set_default_keyboard(self, keyboard='set_default'):
- """Set a default keyboard for the bot.
- If a keyboard is not passed as argument, a default one is generated,
- based on aliases of commands.
- """
- if keyboard == 'set_default':
- btns = [
- dict(
- text=x
- )
- for x in self.default_reply_keyboard_elements
- ]
- row_len = 2 if len(btns) < 4 else 3
- self._default_keyboard = dict(
- keyboard=make_lines_of_buttons(
- btns,
- row_len
- ),
- resize_keyboard=True
- )
- else:
- self._default_keyboard = keyboard
- return
- async def edit_message(self, update, *args, **kwargs):
- """Edit given update with given *args and **kwargs.
- Please note, that it is currently only possible to edit messages
- without reply_markup or with inline keyboards.
- """
- try:
- return await self.editMessageText(
- telepot.message_identifier(update),
- *args,
- **kwargs
- )
- except Exception as e:
- logging.error("{}".format(e))
- async def delete_message(self, update, *args, **kwargs):
- """Delete given update with given *args and **kwargs.
- Please note, that a bot can delete only messages sent by itself
- or sent in a group which it is administrator of.
- """
- try:
- return await self.deleteMessage(
- telepot.message_identifier(update),
- *args,
- **kwargs
- )
- except Exception as e:
- logging.error("{}".format(e))
- async def send_message(self, answer=dict(), chat_id=None, text='',
- parse_mode="HTML", disable_web_page_preview=None,
- disable_notification=None, reply_to_message_id=None,
- reply_markup=None):
- """Send a message.
- Convenient method to call telepot.Bot(token).sendMessage
- All sendMessage **kwargs can be either **kwargs of send_message
- or key:val of answer argument.
- Messages longer than telegram limit will be split properly.
- Telegram flood limits won't be reached thanks to
- `await avoid_flooding(chat_id)`
- parse_mode will be checked and edited if necessary.
- Arguments will be checked and adapted.
- """
- if type(answer) is dict and 'chat_id' in answer:
- chat_id = answer['chat_id']
- # chat_id may simply be the update to which the bot should repy
- if type(chat_id) is dict:
- chat_id = self.get_chat_id(chat_id)
- if type(answer) is str:
- text = answer
- if (
- not reply_markup
- and chat_id > 0
- and text != self.unauthorized_message
- ):
- reply_markup = self.default_keyboard
- elif type(answer) is dict:
- if 'text' in answer:
- text = answer['text']
- if 'parse_mode' in answer:
- parse_mode = answer['parse_mode']
- if 'disable_web_page_preview' in answer:
- disable_web_page_preview = answer['disable_web_page_preview']
- if 'disable_notification' in answer:
- disable_notification = answer['disable_notification']
- if 'reply_to_message_id' in answer:
- reply_to_message_id = answer['reply_to_message_id']
- if 'reply_markup' in answer:
- reply_markup = answer['reply_markup']
- elif (
- not reply_markup
- and type(chat_id) is int
- and chat_id > 0
- and text != self.unauthorized_message
- ):
- reply_markup = self.default_keyboard
- assert type(text) is str, "Text is not a string!"
- assert (
- type(chat_id) is int
- or (type(chat_id) is str and chat_id.startswith('@'))
- ), "Invalid chat_id:\n\t\t{}".format(chat_id)
- if not text:
- return
- parse_mode = str(parse_mode)
- text_chunks = split_text_gracefully(
- text=text,
- limit=self.__class__.TELEGRAM_MESSAGES_MAX_LEN - 100,
- parse_mode=parse_mode
- )
- n = len(text_chunks)
- for text_chunk in text_chunks:
- n -= 1
- if parse_mode.lower() == "html":
- this_parse_mode = "HTML"
- # Check that all tags are well-formed
- if not markdown_check(
- text_chunk,
- [
- "<", ">",
- "code>", "bold>", "italic>",
- "b>", "i>", "a>", "pre>"
- ]
- ):
- this_parse_mode = "None"
- text_chunk = (
- "!!![invalid markdown syntax]!!!\n\n"
- + text_chunk
- )
- elif parse_mode != "None":
- this_parse_mode = "Markdown"
- # Check that all markdowns are well-formed
- if not markdown_check(
- text_chunk,
- [
- "*", "_", "`"
- ]
- ):
- this_parse_mode = "None"
- text_chunk = (
- "!!![invalid markdown syntax]!!!\n\n"
- + text_chunk
- )
- else:
- this_parse_mode = parse_mode
- this_reply_markup = reply_markup if n == 0 else None
- try:
- await self.avoid_flooding(chat_id)
- result = await self.sendMessage(
- chat_id=chat_id,
- text=text_chunk,
- parse_mode=this_parse_mode,
- disable_web_page_preview=disable_web_page_preview,
- disable_notification=disable_notification,
- reply_to_message_id=reply_to_message_id,
- reply_markup=this_reply_markup
- )
- except Exception as e:
- logging.debug(
- e,
- exc_info=False # Set exc_info=True for more information
- )
- result = e
- return result
- async def send_photo(self, chat_id=None, answer={},
- photo=None, caption='', parse_mode='HTML',
- disable_notification=None, reply_to_message_id=None,
- reply_markup=None, use_stored=True,
- second_chance=False):
- """Send a photo.
- Convenient method to call telepot.Bot(token).sendPhoto
- All sendPhoto **kwargs can be either **kwargs of send_message
- or key:val of answer argument.
- Captions longer than telegram limit will be shortened gently.
- Telegram flood limits won't be reached thanks to
- `await avoid_flooding(chat_id)`
- Most arguments will be checked and adapted.
- If use_stored is set to True, the bot will store sent photo
- telegram_id and use it for faster sending next times (unless
- future errors).
- Sending photos by their file_id already stored on telegram servers
- is way faster: that's why bot stores and uses this info,
- if required.
- A second_chance is given to send photo on error.
- """
- if 'chat_id' in answer:
- chat_id = answer['chat_id']
- # chat_id may simply be the update to which the bot should repy
- if type(chat_id) is dict:
- chat_id = self.get_chat_id(chat_id)
- assert (
- type(chat_id) is int
- or (type(chat_id) is str and chat_id.startswith('@'))
- ), "Invalid chat_id:\n\t\t{}".format(chat_id)
- if 'photo' in answer:
- photo = answer['photo']
- assert photo is not None, "Null photo!"
- if 'caption' in answer:
- caption = answer['caption']
- if 'parse_mode' in answer:
- parse_mode = answer['parse_mode']
- if 'disable_notification' in answer:
- disable_notification = answer['disable_notification']
- if 'reply_to_message_id' in answer:
- reply_to_message_id = answer['reply_to_message_id']
- if 'reply_markup' in answer:
- reply_markup = answer['reply_markup']
- already_sent = False
- if type(photo) is str:
- photo_url = photo
- with self.db as db:
- already_sent = db['sent_pictures'].find_one(
- url=photo_url,
- errors=False
- )
- if already_sent and use_stored:
- photo = already_sent['file_id']
- already_sent = True
- else:
- already_sent = False
- if not any(photo_url.startswith(x) for x in ['http', 'www']):
- with io.BytesIO() as buffered_picture:
- with open(
- "{}/{}".format(
- self.path,
- photo_url
- ),
- 'rb'
- ) as photo_file:
- buffered_picture.write(photo_file.read())
- photo = buffered_picture.getvalue()
- else:
- use_stored = False
- caption = escape_html_chars(caption)
- if len(caption) > 199:
- new_caption = ''
- tag = False
- tag_body = False
- count = 0
- temp = ''
- for char in caption:
- if tag and char == '>':
- tag = False
- elif char == '<':
- tag = True
- tag_body = not tag_body
- elif not tag:
- count += 1
- if count == 199:
- break
- temp += char
- if not tag_body:
- new_caption += temp
- temp = ''
- caption = new_caption
- sent = None
- try:
- await self.avoid_flooding(chat_id)
- sent = await self.sendPhoto(
- chat_id=chat_id,
- photo=photo,
- caption=caption,
- parse_mode=parse_mode,
- disable_notification=disable_notification,
- reply_to_message_id=reply_to_message_id,
- reply_markup=reply_markup
- )
- if isinstance(sent, Exception):
- raise Exception("SendingFailed")
- except Exception as e:
- logging.error(
- "Error sending photo\n{}".format(
- e
- ),
- exc_info=False # Set exc_info=True for more information
- )
- if already_sent:
- with self.db as db:
- db['sent_pictures'].update(
- dict(
- url=photo_url,
- errors=True
- ),
- ['url']
- )
- if not second_chance:
- logging.info("Trying again (only once)...")
- sent = await self.send_photo(
- chat_id=chat_id,
- answer=answer,
- photo=photo,
- caption=caption,
- parse_mode=parse_mode,
- disable_notification=disable_notification,
- reply_to_message_id=reply_to_message_id,
- reply_markup=reply_markup,
- second_chance=True
- )
- if (
- sent is not None
- and hasattr(sent, '__getitem__')
- and 'photo' in sent
- and len(sent['photo']) > 0
- and 'file_id' in sent['photo'][0]
- and (not already_sent)
- and use_stored
- ):
- with self.db as db:
- db['sent_pictures'].insert(
- dict(
- url=photo_url,
- file_id=sent['photo'][0]['file_id'],
- errors=False
- )
- )
- return sent
- async def forward_message(self, chat_id, update=None, from_chat_id=None,
- message_id=None, disable_notification=False):
- """Forward message from `from_chat_id` to `chat_id`.
- Set disable_notification to True to avoid disturbing recipient.
- """
- try:
- if from_chat_id is None or message_id is None:
- if (
- type(update) is not dict
- or 'chat' not in update
- or 'id' not in update['chat']
- or 'message_id' not in update
- ):
- raise Exception("Wrong parameters, cannot forward.")
- from_chat_id = update['chat']['id']
- message_id = update['message_id']
- await self.avoid_flooding(chat_id)
- sent = await self.forwardMessage(
- chat_id=chat_id,
- from_chat_id=from_chat_id,
- disable_notification=disable_notification,
- message_id=message_id,
- )
- if isinstance(sent, Exception):
- raise Exception("Forwarding failed.")
- except Exception as e:
- logging.error(
- "Error forwarding message:\n{}".format(
- e
- ),
- exc_info=False # Set exc_info=True for more information
- )
- async def send_and_destroy(self, chat_id, answer,
- timer=60, mode='text', **kwargs):
- """Send a message or photo and delete it after `timer` seconds."""
- if mode == 'text':
- sent_message = await self.send_message(
- chat_id=chat_id,
- answer=answer,
- **kwargs
- )
- elif mode == 'pic':
- sent_message = await self.send_photo(
- chat_id=chat_id,
- answer=answer,
- **kwargs
- )
- if sent_message is None:
- return
- self.to_be_destroyed.append(sent_message)
- await asyncio.sleep(timer)
- if await self.delete_message(sent_message):
- self.to_be_destroyed.remove(sent_message)
- return
- async def wait_and_obscure(self, update, when, inline_message_id):
- """Obscure messages which can't be deleted.
- Obscure an inline_message `timer` seconds after sending it,
- by editing its text or caption.
- At the moment Telegram won't let bots delete sent inline query results.
- """
- if type(when) is int:
- when = datetime.datetime.now() + datetime.timedelta(seconds=when)
- assert type(when) is datetime.datetime, (
- "when must be a datetime instance or a number of seconds (int) "
- "to be awaited"
- )
- if 'inline_message_id' not in update:
- logging.info(
- "This inline query result owns no inline_keyboard, so it "
- "can't be modified"
- )
- return
- inline_message_id = update['inline_message_id']
- self.to_be_obscured.append(inline_message_id)
- await sleep_until(when)
- try:
- await self.editMessageCaption(
- inline_message_id,
- text="Time over"
- )
- except Exception:
- try:
- await self.editMessageText(
- inline_message_id,
- text="Time over"
- )
- except Exception as e:
- logging.error(
- "Couldn't obscure message\n{}\n\n{}".format(
- inline_message_id,
- e
- )
- )
- self.to_be_obscured.remove(inline_message_id)
- return
- async def save_picture(self, update, file_name=None, path='img/',
- extension='jpg'):
- """Store `update` picture as `path`/`file_name`.`extension`."""
- if not path.endswith('/'):
- path = '{p}/'.format(
- p=path
- )
- if not os.path.isdir(path):
- path = '{path}/img/'.format(
- path=self.path
- )
- if file_name is None:
- file_name = get_secure_key(length=6)
- if file_name.endswith('.'):
- file_name = file_name[:-1]
- complete_file_name = '{path}{name}.{ext}'.format(
- path=self.path,
- name=file_name,
- ext=extension
- )
- while os.path.isfile(complete_file_name):
- file_name += get_secure_key(length=1)
- complete_file_name = '{path}{name}.{ext}'.format(
- path=self.path,
- name=file_name,
- ext=extension
- )
- try:
- await self.download_file(
- update['photo'][-1]['file_id'],
- complete_file_name
- )
- except Exception as e:
- return dict(
- result=1, # Error
- file_name=None,
- error=e
- )
- return dict(
- result=0, # Success
- file_name=complete_file_name,
- error=None
- )
- async def continue_running(self):
- """Get updates.
- If bot can be got, sets name and telegram_id,
- awaits preliminary tasks and starts getting updates from telegram.
- If bot can't be got, restarts all bots in 5 minutes.
- """
- await self.get_me()
- for task in self.run_before_loop:
- await task()
- self.set_default_keyboard()
- asyncio.ensure_future(
- self.message_loop(handler=self.routing_table)
- )
- return
- def stop_bots(self):
- """Exit script with code 0."""
- Bot.stop = True
- def restart_bots(self):
- """Restart the script exiting with code 65.
- Actually, you need to catch Bot.stop state when Bot.run() returns
- and handle the situation yourself.
- """
- Bot.stop = "Restart"
- @classmethod
- async def check_task(cls):
- """Await until cls.stop, then end session and return."""
- for bot in cls.instances.values():
- asyncio.ensure_future(bot.continue_running())
- while not cls.stop:
- await asyncio.sleep(10)
- return await cls.end_session()
- @classmethod
- async def end_session(cls):
- """Run after stop, before the script exits.
- Await final tasks, obscure and delete pending messages,
- log current operation (stop/restart).
- """
- for bot in cls.instances.values():
- for task in bot.run_after_loop:
- await task()
- for message in bot.to_be_destroyed:
- try:
- await bot.delete_message(message)
- except Exception as e:
- logging.error(
- "Couldn't delete message\n{}\n\n{}".format(
- message,
- e
- )
- )
- for inline_message_id in bot.to_be_obscured:
- try:
- await bot.editMessageCaption(
- inline_message_id,
- text="Time over"
- )
- except Exception:
- try:
- await bot.editMessageText(
- inline_message_id,
- text="Time over"
- )
- except Exception as e:
- logging.error(
- "Couldn't obscure message\n{}\n\n{}".format(
- inline_message_id,
- e
- )
- )
- if cls.stop == "Restart":
- logging.info("\n\t\t---Restart!---")
- elif cls.stop == "KeyboardInterrupt":
- logging.info("Stopped by KeyboardInterrupt.")
- else:
- logging.info("Stopped gracefully by user.")
- return
- @classmethod
- def run(cls, loop=None):
- """Call this method to run the async bots."""
- if loop is None:
- loop = asyncio.get_event_loop()
- logging.info(
- "{sep}{subjvb} STARTED{sep}".format(
- sep='-'*10,
- subjvb=('BOT HAS' if len(cls.instances) == 1 else 'BOTS HAVE')
- )
- )
- try:
- loop.run_until_complete(
- cls.check_task()
- )
- except KeyboardInterrupt:
- logging.info(
- (
- "\n\t\tYour script received a KeyboardInterrupt signal, "
- "your bot{} being stopped."
- ).format(
- 's are'
- if len(cls.instances) > 1
- else ' is'
- )
- )
- cls.stop = "KeyboardInterrupt"
- loop.run_until_complete(cls.end_session())
- except Exception as e:
- logging.error(
- '\nYour bot{vb} been stopped. with error \'{e}\''.format(
- e=e,
- vb='s have' if len(cls.instances) > 1 else ' has'
- ),
- exc_info=True
- )
- logging.info(
- "{sep}{subjvb} STOPPED{sep}".format(
- sep='-'*10,
- subjvb='BOT HAS' if len(cls.instances) == 1 else 'BOTS HAVE'
- )
- )
- return
- @classmethod
- async def _run_manual_mode(cls):
- available_bots = MyOD()
- for code, bot in enumerate(
- cls.instances.values()
- ):
- await bot.get_me()
- available_bots[code] = dict(
- bot=bot,
- code=code,
- name=bot.name
- )
- selected_bot = None
- while selected_bot is None:
- user_input = input(
- "\n=============================================\n"
- "Which bot would you like to control manually?\n"
- "Available bots:\n{}\n\n\t\t".format(
- line_drawing_unordered_list(
- list(
- "{b[code]:>3} - {b[bot].name}".format(
- b=bot,
- )
- for bot in available_bots.values()
- )
- )
- )
- )
- if (
- user_input.isnumeric()
- and int(user_input) in available_bots
- ):
- selected_bot = available_bots[int(user_input)]
- else:
- selected_bot = pick_most_similar_from_list(
- [
- bot['name']
- for bot in available_bots.values()
- ],
- user_input
- )
- selected_bot = available_bots.get_by_key_val(
- key='name',
- val=selected_bot,
- case_sensitive=False,
- return_value=True
- )
- if selected_bot is None:
- logging.error("Invalid selection.")
- continue
- logging.info(
- "Bot `{b[name]}` selected.".format(
- b=selected_bot
- )
- )
- exit_code = await selected_bot['bot']._run_manually()
- if exit_code == 0:
- break
- elif exit_code == 65:
- selected_bot = None
- return
- @classmethod
- def run_manual_mode(cls, loop=None):
- """Run in manual mode: send messages via bots."""
- if loop is None:
- loop = asyncio.get_event_loop()
- logging.info(
- "=== MANUAL MODE STARTED ==="
- )
- try:
- loop.run_until_complete(
- cls._run_manual_mode()
- )
- except KeyboardInterrupt:
- logging.info(
- (
- "\n\t\tYour script received a KeyboardInterrupt signal, "
- "your bot{} being stopped."
- ).format(
- 's are' if len(cls.instances) > 1 else ' is'
- )
- )
- except Exception as e:
- logging.error(
- '\nYour bot{vb} been stopped. with error \'{e}\''.format(
- e=e,
- vb='s have' if len(cls.instances) > 1 else ' has'
- ),
- exc_info=True
- )
- logging.info(
- "=== MANUAL MODE STOPPED ==="
- )
- async def _run_manually(self):
- user_input = ' choose_addressee'
- selected_user = None
- users = []
- while user_input:
- if user_input == ' choose_addressee':
- try:
- user_input = input(
- "Choose an addressee.\n"
- "Press enter to change bot.\n"
- "\n\t\t"
- )
- if len(user_input) == 0:
- return 65 # Let user select a different bot
- except KeyboardInterrupt:
- logging.error("Keyboard interrupt.")
- return 0 # Stop running
- if (
- selected_user is None
- and user_input.strip('-').isnumeric()
- ):
- user_input = int(user_input)
- users = list(
- filter(
- lambda user: user['telegram_id'] == user_input,
- users
- )
- )
- if len(users) == 0:
- users = [
- dict(
- telegram_id=user_input,
- name='Unknown user'
- )
- ]
- elif (
- selected_user is None
- and self.db_url is not None
- ):
- with self.db as db:
- if 'users' not in db.tables:
- db['users'].insert(
- dict(
- telegram_id=0,
- privileges=100,
- username='username',
- first_name='first_name',
- last_name='last_name'
- )
- )
- if 'contacts' not in db.tables:
- db['contacts'].insert(
- dict(
- telegram_id=0,
- name='ZZZXXXAAA',
- )
- )
- users = list(
- db.query(
- """SELECT telegram_id, MAX(name) name
- FROM (
- SELECT telegram_id,
- COALESCE(
- first_name || ' ' || last_name ||
- ' (' || username || ')',
- username,
- first_name || ' ' || last_name,
- last_name,
- first_name
- ) AS name
- FROM users
- WHERE COALESCE(
- first_name || last_name || username,
- first_name || username,
- last_name || username,
- first_name || last_name,
- username,
- last_name,
- first_name
- )
- LIKE '%{u}%'
- UNION ALL
- SELECT telegram_id, name
- FROM contacts
- WHERE name LIKE '%{u}%'
- )
- GROUP BY telegram_id
- """.format(
- u=user_input
- )
- )
- )
- if len(users) == 0:
- logging.info("Sorry, no user matches your query.")
- user_input = ' choose_addressee'
- elif len(users) > 1:
- user_input = input(
- "Please select one of the following users:\n"
- "\n"
- "{users}\n"
- "\n"
- "Paste their telegram_id\n"
- "\t\t".format(
- users=line_drawing_unordered_list(
- sorted(
- map(
- lambda user: (
- "{u[telegram_id]} - {u[name]}"
- ).format(
- u=user
- ),
- users
- )
- )
- )
- )
- )
- elif len(users) == 1:
- selected_user = users[0]
- while selected_user is not None:
- sent = None
- text = input(
- "What would you like to send "
- "{u[name]} ({u[telegram_id]})?\n"
- "Leave it blank if you want to select another "
- "addressee.\n"
- "\t\t\t".format(
- u=selected_user
- )
- )
- if len(text) == 0:
- selected_user = None
- user_input = ' choose_addressee'
- elif text.lower() == 'photo':
- caption = input(
- 'Write a caption (you can leave it blank)\n'
- '\t\t\t'
- )
- try:
- with io.BytesIO() as buffered_picture:
- with open(
- '{path}/sendme.png'.format(
- path=self.path
- ),
- 'rb' # Read bytes
- ) as photo_file:
- buffered_picture.write(
- photo_file.read()
- )
- photo = buffered_picture.getvalue()
- sent = await self.send_photo(
- chat_id=selected_user['telegram_id'],
- photo=photo,
- caption=caption,
- parse_mode='HTML',
- use_stored=False
- )
- except Exception as e:
- logging.error(e)
- else:
- try:
- sent = await self.send_message(
- chat_id=selected_user['telegram_id'],
- text=text,
- parse_mode='HTML'
- )
- except Exception as e:
- logging.error(e)
- if (
- sent is not None
- and not isinstance(sent, Exception)
- ):
- logging.info(
- '\n'
- 'Sent message:\n'
- '{s}\n'
- '\n'.format(
- s=sent
- )
- )
- while (
- self.db_url
- and selected_user['name'] == 'Unknown user'
- ):
- new_name = input(
- "Please enter a nickname for this user.\n"
- "Next time you may retrieve their telegram_id "
- "by passing this nickname (or a part of it).\n"
- "\t\t"
- )
- if len(new_name):
- selected_user['name'] = new_name
- with self.db as db:
- db['contacts'].upsert(
- selected_user,
- ['telegram_id'],
- ensure=True
- )
- else:
- logging.info("Invalid name, please try again.")
- return 65 # Keep running, making user select another bot
- def create_views(self, views, overwrite=False):
- """Take a list of `views` and add them to bot database.
- Each element of this list should have
- - a `name` field
- - a `query field`
- """
- with self.db as db:
- for view in views:
- try:
- if overwrite:
- db.query(
- f"""
- DROP VIEW IF EXISTS {view['name']}
- """
- )
- db.query(
- f"""
- CREATE VIEW IF NOT EXISTS {view['name']}
- AS {view['query']}
- """
- )
- except Exception as e:
- logging.error(f"{e}")
|