1
0
mirror of https://github.com/eternnoir/pyTelegramBotAPI.git synced 2023-08-10 21:12:57 +03:00
pyTelegramBotAPI/telebot/__init__.py
CongNT 24491f467f make telebot compatible with python3
also two small PEP8 fixes
2015-07-01 09:49:41 +07:00

202 lines
6.6 KiB
Python

# -*- coding: utf-8 -*-
from __future__ import print_function
import json
import time
import threading
from telebot import apihelper, types
"""
Module : telebot
"""
API_URL = r"https://api.telegram.org/"
class TeleBot:
""" This is TeleBot Class
Methods:
getMe
sendMessage
forwardMessage
sendPhoto
sendAudio
sendDocument
sendSticker
sendVideo
sendLocation
sendChatAction
getUserProfilePhotos
getUpdates
setWebhook
"""
def __init__(self, token):
self.token = token
self.update_listener = []
self.polling_thread = None
self.__stop_polling = False
self.interval = 3
self.last_update_id = 0
def get_update(self):
result = apihelper.get_updates(self.token, offset=(self.last_update_id + 1))
updates = result['result']
notify_updates = []
for update in updates:
if update['update_id'] > self.last_update_id:
self.last_update_id = update['update_id']
msg = types.Message.de_json(json.dumps(update['message']))
notify_updates.append(msg)
self.__notify_update(notify_updates)
def __notify_update(self, new_messages):
for listener in self.update_listener:
t = threading.Thread(target=listener, args=new_messages)
t.start()
def polling(self, interval=3):
"""
Always get updates.
:param interval: iterval secs.
:return:
"""
self.interval = interval
# clear thread.
self.__stop_polling = True
time.sleep(interval + 1)
self.__stop_polling = False
self.polling_thread = threading.Thread(target=self.__polling, args=())
self.polling_thread.daemon = True
self.polling_thread.start()
def __polling(self):
print('telegram bot start polling')
while not self.__stop_polling:
try:
self.get_update()
except Exception as e:
print(e)
time.sleep(self.interval)
print('telegram bot stop polling')
def stop_polling(self):
self.__stop_polling = True
def set_update_listener(self, listener):
self.update_listener.append(listener)
def get_me(self):
result = apihelper.get_me(self.token)
if result['ok'] is not True:
raise Exception('getMe Error.' + json.dumps(result))
u = types.User.de_json(json.dumps(result['result']))
return u
def send_message(self, chat_id, text, disable_web_page_preview=None, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send text messages.
:param chat_id:
:param text:
:param disable_web_page_preview:
:param reply_to_message_id:
:param reply_markup:
:return:
"""
return apihelper.send_message(self.token, chat_id, text, disable_web_page_preview, reply_to_message_id,
reply_markup)
def forward_message(self, chat_id, from_chat_id, message_id):
"""
Use this method to forward messages of any kind.
:param chat_id: which chat to forward
:param from_chat_id: which chat message from
:param message_id: message id
:return:
"""
return apihelper.forward_message(self.token, chat_id, from_chat_id, message_id)
def send_photo(self, chat_id, photo, caption=None, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send photos.
:param chat_id:
:param photo:
:param caption:
:param reply_to_message_id:
:param reply_markup:
:return:
"""
return apihelper.send_photo(self.token, chat_id, photo, caption, reply_to_message_id, reply_markup)
def send_audio(self, chat_id, data, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send audio files, if you want Telegram clients to display the file as a playable
voice message. For this to work, your audio must be in an .ogg file encoded with OPUS
:param chat_id:
:param data:
:param reply_to_message_id:
:param reply_markup:
:return:
"""
return apihelper.send_data(self.token, chat_id, data, 'audio', reply_to_message_id, reply_markup)
def send_document(self, chat_id, data, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send general files.
:param chat_id:
:param data:
:param reply_to_message_id:
:param reply_markup:
:return:
"""
return apihelper.send_data(self.token, chat_id, data, 'document', reply_to_message_id, reply_markup)
def send_sticker(self, chat_id, data, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send .webp stickers.
:param chat_id:
:param data:
:param reply_to_message_id:
:param reply_markup:
:return:
"""
return apihelper.send_data(self.token, chat_id, data, 'sticker', reply_to_message_id, reply_markup)
def send_video(self, chat_id, data, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send video files, Telegram clients support mp4 videos.
:param chat_id:
:param data:
:param reply_to_message_id:
:param reply_markup:
:return:
"""
return apihelper.send_data(self.token, chat_id, data, 'video', reply_to_message_id, reply_markup)
def send_location(self, chat_id, latitude, longitude, reply_to_message_id=None, reply_markup=None):
"""
Use this method to send point on the map.
:param chat_id:
:param latitude:
:param longitude:
:param reply_to_message_id:
:param reply_markup:
:return:
"""
return apihelper.send_location(self.token, chat_id, latitude, longitude, reply_to_message_id, reply_markup)
def send_chat_action(self, chat_id, action):
"""
Use this method when you need to tell the user that something is happening on the bot's side.
The status is set for 5 seconds or less (when a message arrives from your bot, Telegram clients clear
its typing status).
:param chat_id:
:param action: string . typing,upload_photo,record_video,upload_video,record_audio,upload_audio,upload_document,
find_location.
:return:
"""
return apihelper.send_chat_action(self.token, chat_id, action)