我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用telebot.TeleBot()。
def listener(messages): """ When new messages arrive TeleBot will call this function. """ for m in messages: if m.content_type == 'text': # print the sent message to the console print (str(m.chat.first_name) + " [" + str(m.chat.id) + "]: " + m.text)
def test_send_venue_dis_noti(self): tb = telebot.TeleBot(TOKEN) lat = 26.3875591 lon = -161.2901042 ret_msg = tb.send_venue(CHAT_ID, lat, lon, "Test Venue", "1123 Test Venue address", disable_notification=True) assert ret_msg.venue.title == "Test Venue"
def __init__(self, config): if config["token"] == "": logging.error("No telegram token specified, exiting") exit(1) self.token = config["token"] self.bot = telebot.TeleBot(self.token, threaded=False) self.event_handlers = {} self.own_user = None self._telebot_event_handler(self.cmd_start, commands=["start"]) self._telebot_event_handler(self.cmd_help, commands=["help"]) # FIXME: not a portable way of registering commands self._telebot_event_handler(self.cmd_me, commands=["me"]) for k, v in mapped_content_type.items(): if v == "": continue self._telebot_event_handler_passthrough(v, content_types=[k]) for k in content_types_media: self._telebot_event_handler(self.on_media, content_types=[k]) self._telebot_event_handler(self.on_content_type_none, content_types=[None])
def sethook(req): bot = telebot.TeleBot(config.token) bot.remove_webhook() # ?????? ?????? ?????? bot.set_webhook(url=config.WEBHOOK_URL) return HttpResponse('OK')
def listener(messages): """ When new messages arrive TeleBot will call this function. """ for m in messages: if m.content_type == 'text': print (str(m.chat.first_name) + " [" + str(m.chat.id) + "]: " + m.text)
def test_message_listener(self): msg_list = [] for x in range(100): msg_list.append(self.create_text_message('Message ' + str(x))) def listener(messages): assert len(messages) == 100 tb = telebot.TeleBot('') tb.set_update_listener(listener)
def test_message_handler(self): tb = telebot.TeleBot('') msg = self.create_text_message('/help') @tb.message_handler(commands=['help', 'start']) def command_handler(message): message.text = 'got' tb.process_new_messages([msg]) time.sleep(1) assert msg.text == 'got'
def test_message_handler_reg(self): bot = telebot.TeleBot('') msg = self.create_text_message(r'https://web.telegram.org/') @bot.message_handler(regexp='((https?):((//)|(\\\\))+([\w\d:#@%/;$()~_?\+-=\\\.&](#!)?)*)') def command_url(message): msg.text = 'got' bot.process_new_messages([msg]) time.sleep(1) assert msg.text == 'got'
def test_message_handler_lambda(self): bot = telebot.TeleBot('') msg = self.create_text_message(r'lambda_text') @bot.message_handler(func=lambda message: r'lambda' in message.text) def command_url(message): msg.text = 'got' bot.process_new_messages([msg]) time.sleep(1) assert msg.text == 'got'
def test_message_handler_reg_fail(self): bot = telebot.TeleBot('') msg = self.create_text_message(r'web.telegram.org/') @bot.message_handler(regexp='((https?):((//)|(\\\\))+([\w\d:#@%/;$()~_?\+-=\\\.&](#!)?)*)') def command_url(message): msg.text = 'got' bot.process_new_messages([msg]) time.sleep(1) assert not msg.text == 'got'
def test_send_message_with_markdown(self): tb = telebot.TeleBot(TOKEN) markdown = """ *bold text* _italic text_ [text](URL) """ ret_msg = tb.send_message(CHAT_ID, markdown, parse_mode="Markdown") assert ret_msg.message_id
def test_send_message_with_disable_notification(self): tb = telebot.TeleBot(TOKEN) markdown = """ *bold text* _italic text_ [text](URL) """ ret_msg = tb.send_message(CHAT_ID, markdown, parse_mode="Markdown", disable_notification=True) assert ret_msg.message_id
def test_send_file(self): file_data = open('../examples/detailed_example/kitten.jpg', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_document(CHAT_ID, file_data) assert ret_msg.message_id ret_msg = tb.send_document(CHAT_ID, ret_msg.document.file_id) assert ret_msg.message_id
def test_send_file_dis_noti(self): file_data = open('../examples/detailed_example/kitten.jpg', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_document(CHAT_ID, file_data, disable_notification=True) assert ret_msg.message_id ret_msg = tb.send_document(CHAT_ID, ret_msg.document.file_id) assert ret_msg.message_id
def test_send_video(self): file_data = open('./test_data/test_video.mp4', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_video(CHAT_ID, file_data) assert ret_msg.message_id
def test_send_video_dis_noti(self): file_data = open('./test_data/test_video.mp4', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_video(CHAT_ID, file_data, disable_notification=True) assert ret_msg.message_id
def test_send_video_more_params(self): file_data = open('./test_data/test_video.mp4', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_video(CHAT_ID, file_data, 1) assert ret_msg.message_id
def test_send_video_more_params_dis_noti(self): file_data = open('./test_data/test_video.mp4', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_video(CHAT_ID, file_data, 1, disable_notification=True) assert ret_msg.message_id
def test_send_file_exception(self): tb = telebot.TeleBot(TOKEN) try: tb.send_document(CHAT_ID, None) assert False except Exception as e: print(e) assert True
def test_send_photo_dis_noti(self): file_data = open('../examples/detailed_example/kitten.jpg', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_photo(CHAT_ID, file_data) assert ret_msg.message_id ret_msg = tb.send_photo(CHAT_ID, ret_msg.photo[0].file_id, disable_notification=True) assert ret_msg.message_id
def test_send_audio(self): file_data = open('./test_data/record.mp3', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_audio(CHAT_ID, file_data, 1, 'eternnoir', 'pyTelegram') assert ret_msg.content_type == 'audio' assert ret_msg.audio.performer == 'eternnoir' assert ret_msg.audio.title == 'pyTelegram'
def test_send_audio_dis_noti(self): file_data = open('./test_data/record.mp3', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_audio(CHAT_ID, file_data, 1, 'eternnoir', 'pyTelegram', disable_notification=True) assert ret_msg.content_type == 'audio' assert ret_msg.audio.performer == 'eternnoir' assert ret_msg.audio.title == 'pyTelegram'
def test_send_voice(self): file_data = open('./test_data/record.ogg', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_voice(CHAT_ID, file_data) assert ret_msg.voice.mime_type == 'audio/ogg'
def test_send_voice_dis_noti(self): file_data = open('./test_data/record.ogg', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_voice(CHAT_ID, file_data, disable_notification=True) assert ret_msg.voice.mime_type == 'audio/ogg'
def test_get_file_dis_noti(self): file_data = open('./test_data/record.ogg', 'rb') tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_voice(CHAT_ID, file_data, disable_notification=True) file_id = ret_msg.voice.file_id file_info = tb.get_file(file_id) assert file_info.file_id == file_id
def test_send_message(self): text = 'CI Test Message' tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_message(CHAT_ID, text) assert ret_msg.message_id
def test_send_message_dis_noti(self): text = 'CI Test Message' tb = telebot.TeleBot(TOKEN) ret_msg = tb.send_message(CHAT_ID, text, disable_notification=True) assert ret_msg.message_id
def test_send_message_with_markup(self): text = 'CI Test Message' tb = telebot.TeleBot(TOKEN) markup = types.ReplyKeyboardMarkup() markup.add(types.KeyboardButton("1")) markup.add(types.KeyboardButton("2")) ret_msg = tb.send_message(CHAT_ID, text, disable_notification=True, reply_markup=markup) assert ret_msg.message_id
def test_send_message_with_markup_use_string(self): text = 'CI Test Message' tb = telebot.TeleBot(TOKEN) markup = types.ReplyKeyboardMarkup() markup.add("1") markup.add("2") markup.add("3") markup.add("4") ret_msg = tb.send_message(CHAT_ID, text, disable_notification=True, reply_markup=markup) assert ret_msg.message_id
def test_forward_message(self): text = 'CI forward_message Test Message' tb = telebot.TeleBot(TOKEN) msg = tb.send_message(CHAT_ID, text) ret_msg = tb.forward_message(CHAT_ID, CHAT_ID, msg.message_id) assert ret_msg.forward_from
def test_forward_message_dis_noti(self): text = 'CI forward_message Test Message' tb = telebot.TeleBot(TOKEN) msg = tb.send_message(CHAT_ID, text) ret_msg = tb.forward_message(CHAT_ID, CHAT_ID, msg.message_id, disable_notification=True) assert ret_msg.forward_from
def test_reply_to(self): text = 'CI reply_to Test Message' tb = telebot.TeleBot(TOKEN) msg = tb.send_message(CHAT_ID, text) ret_msg = tb.reply_to(msg, text + ' REPLY') assert ret_msg.reply_to_message.message_id == msg.message_id
def test_register_for_reply(self): text = 'CI reply_to Test Message' tb = telebot.TeleBot(TOKEN) msg = tb.send_message(CHAT_ID, text, reply_markup=types.ForceReply()) reply_msg = tb.reply_to(msg, text + ' REPLY') def process_reply(message): assert msg.message_id == message.reply_to_message.message_id tb.register_for_reply(msg, process_reply) tb.process_new_messages([reply_msg])
def test_send_location(self): tb = telebot.TeleBot(TOKEN) lat = 26.3875591 lon = -161.2901042 ret_msg = tb.send_location(CHAT_ID, lat, lon) assert int(ret_msg.location.longitude) == int(lon) assert int(ret_msg.location.latitude) == int(lat)
def test_send_venue(self): tb = telebot.TeleBot(TOKEN) lat = 26.3875591 lon = -161.2901042 ret_msg = tb.send_venue(CHAT_ID, lat, lon, "Test Venue", "1123 Test Venue address") assert ret_msg.venue.title == "Test Venue"
def test_Chat(self): tb = telebot.TeleBot(TOKEN) me = tb.get_me() msg = tb.send_message(CHAT_ID, 'Test') assert me.id == msg.from_user.id assert msg.chat.id == int(CHAT_ID)
def test_edit_message_text(self): tb = telebot.TeleBot(TOKEN) msg = tb.send_message(CHAT_ID, 'Test') new_msg = tb.edit_message_text('Edit test', chat_id=CHAT_ID, message_id=msg.message_id) assert new_msg.text == 'Edit test'
def test_edit_message_caption(self): file_data = open('../examples/detailed_example/kitten.jpg', 'rb') tb = telebot.TeleBot(TOKEN) msg = tb.send_document(CHAT_ID, file_data, caption="Test") new_msg = tb.edit_message_caption(caption='Edit test', chat_id=CHAT_ID, message_id=msg.message_id) assert new_msg.caption == 'Edit test'
def test_get_chat_administrators(self): tb = telebot.TeleBot(TOKEN) cas = tb.get_chat_administrators(GROUP_ID) assert len(cas) > 0
def test_get_chat_members_count(self): tb = telebot.TeleBot(TOKEN) cn = tb.get_chat_members_count(GROUP_ID) assert cn > 1
def test_edit_markup(self): text = 'CI Test Message' tb = telebot.TeleBot(TOKEN) markup = types.InlineKeyboardMarkup() markup.add(types.InlineKeyboardButton("Google", url="http://www.google.com")) markup.add(types.InlineKeyboardButton("Yahoo", url="http://www.yahoo.com")) ret_msg = tb.send_message(CHAT_ID, text, disable_notification=True, reply_markup=markup) markup.add(types.InlineKeyboardButton("Google2", url="http://www.google.com")) markup.add(types.InlineKeyboardButton("Yahoo2", url="http://www.yahoo.com")) new_msg = tb.edit_message_reply_markup(chat_id=CHAT_ID, message_id=ret_msg.message_id, reply_markup=markup) assert new_msg.message_id
def listener(messages): """ When new messages arrive TeleBot will call this function. """ for m in messages: if m.content_type == 'text': # print the sent message to the console print str(m.chat.first_name) + " [" + str(m.chat.id) + "]: " + m.text
def upload_list(audio_list): bot = telebot.TeleBot(config.bot_token) filename_mp3 = 'temp.mp3' filename_ogg = 'temp.ogg' counter = 0 start = time.time() for audio in list(audio_list): try: api_query = 'http://api.?.ws/api.php?method=get.audio&ids=' + str(audio[1]) + '_' + str(audio[0]) + '&key=' + config.audio_api_key api_callback = requests.get(api_query) if api_callback.text == 'wrong ids or Limit exceeded(10)' or api_callback.status_code != 200: audio_list.remove(audio) continue audio_url = (json.loads(api_callback.text))[0][2] if get_local_ogg(audio_url, filename_mp3, filename_ogg): audio_list.remove(audio) continue stop = time.time() if stop - start < 1: #avoiding of exeption: 'Too many requests' time.sleep(1) start = stop audio_file_id = upload_ogg(bot, filename_ogg) audio[11] = audio_file_id os.remove(filename_ogg) except: audio_list.remove(audio) if os.path.exists(filename_ogg): os.remove(filename_ogg) continue with open('uploaded_list.txt', 'w') as outfile: json.dump(audio_list, outfile) return audio_list
def init(self): token = self.config.get('token') self.bot = telebot.TeleBot(token) self.telegram_cache = JSONFile(create_file(os.path.join(get_var_directory(), 'telegram-cache.json'), { 'chat_ids': {}, 'version': __version__, })) # print([vars(u.message.chat) for u in updates])
def get_bot(): '''Return the Telegram bot with token is drawn from config''' return telebot.TeleBot(botcfg['TELEGRAM_TOKEN'])
def check_args(): print("checking args") parser = argparse.ArgumentParser() parser.add_argument('-t', '--token', help="Token received from botfather.") parser.add_argument('-o', '--owner', help="Telegram chat ID to send status messages.", type=int) parser.add_argument('-d', '--dict', help="Keep database as a dict object stored in ram", default=False) args = parser.parse_args() if(args.dict): global triggers_dict triggers_dict = True if (args.token): if (args.owner): try: dummy_bot = telebot.TeleBot(args.token) bot_info = dummy_bot.get_me() bot_user, created = TGUserModel.get_or_create( chat_id=bot_info.id, first_name=bot_info.first_name, username=bot_info.username ) bot_user.save() try: dummy_bot.send_message(args.owner, "This bot is ready!") bot_cfg = ConfigModel.create(bot_user=bot_user, token=args.token, owner=args.owner) bot_cfg.save() except ApiException as ae: print('''Make sure you have started your bot https://telegram.me/{}. And configured the owner variable. ApiException: {}'''.format(bot_info.username, ae)) exit(1) except ApiException as ApiError: print("Invalid token[{}]: {}".format(args.token, ApiError)) exit(1) except Exception as e: print(e) exit(1) else: print("Owner ID not supplied") exit(1)
