import http.client, urllib
import json
import datetime
class ChatInfo:
def __init__(self):
self.chatId = None
self.firstName = None
self.lastName= None
self.userName = None
self.text = None
self.date = None
class TelegramAgent:
API_URL = "api.telegram.org"
CONTENT_TYPE = "Content-type"
APPLICATION_JSON = "application/json"
BOT_PATH = "/bot%s"
SEND_MESSAGE_PATH = "/sendMessage"
SEND_MESSAGE_DATA = '{"chat_id": %s, "text": "%s"}'
GET_ME_PATH = "/getMe"
GET_UPDATES_PATH = "/getUpdates"
GET_UPDATES_DATA = '{"offset": "%d"}'
SEND_PHOTO_PATH = "/sendPhoto"
SEND_PHOTO_DATA = '{"chat_id": %s, "photo": "%s"}'
def __init__(self):
self.bContinue = True
self.response = None
self.callback = None
self.bDumpData = False
def setToken(self, token):
self.token = token
def setCallback(self, callback):
self.callback = callback
def setContinue(self, bContinue):
self.bContinue = bContinue
def setDumpData(self, bDumpData):
self.bDumpData = bDumpData
def convertDate(self, date, format):
return datetime.datetime.fromtimestamp(date).strftime(format)
def packChatId(self, chatId):
if len(chatId) > 0 and chatId[:1] == "@":
return "\"%s\"" % chatId
return chatId
def postRequest(self, url, path, headers, paramsRaw):
try:
params = paramsRaw
conn = http.client.HTTPSConnection(url)
conn.request("POST", path, params.encode("UTF-8"), headers)
self.response = conn.getresponse()
self.data = self.response.read().decode("UTF-8")
if self.bDumpData:
print(self.data)
conn.close()
except (ConnectionError, TimeoutError) as e:
print(f'Catched : {e}')
def sendCommon(self, actionPath, data):
path = self.BOT_PATH % self.token + actionPath
self.postRequest(self.API_URL, path,
{self.CONTENT_TYPE : self.APPLICATION_JSON}, data);
def sendNoData(self, actionPath):
self.sendCommon(actionPath, "")
def sendMessage(self, chatId, message):
chatId = self.packChatId(chatId)
data = self.SEND_MESSAGE_DATA % (chatId, message)
self.sendCommon(self.SEND_MESSAGE_PATH, data)
def sendPhoto(self, chatId, photoUrl):
data = self.SEND_PHOTO_DATA % (chatId, photoUrl)
self.sendCommon(self.SEND_PHOTO_PATH, data)
def getMe(self):
self.sendNoData(self.GET_ME_PATH)
def getUpdates(self, offset):
data = self.GET_UPDATES_DATA % offset
self.sendCommon(self.GET_UPDATES_PATH, data)
def processResult(self, result):
chatInfo = ChatInfo()
try:
message = result['message']
if message == None:
print("message is null")
else:
chat = message['chat']
if chat == None:
print("chat is null")
else:
chatInfo.chatId = "%d" % chat['id']
chatInfo.firstName = chat['first_name']
chatInfo.lastName = chat['last_name']
chatInfo.userName = None
if 'username' in chat:
chatInfo.userName = chat['username']
chatInfo.text = message['text']
chatInfo.date = message['date']
if self.callback != None:
self.callback(chatInfo)
except KeyError as e:
return
def loop(self):
self.bContinue = True
updateId = 0
while (self.bContinue):
self.getUpdates(updateId)
if self.response.status == 200:
respDict = json.loads(self.data)
results = respDict['result']
resultsNum = len(results)
if resultsNum > 0:
lastResult = results[resultsNum - 1]
updateId = lastResult['update_id'] + 1
self.processResult(lastResult)
