텔레그램 봇과 명령과 응답을 양방향으로 주고 받는 방법을 구현해보겠습니다. (Python 3 사용)
텔레그램 봇을 구현하는 방법에는 두가지가 있는데, 그 중 한가지는 웹훅을 사용하는 방식으로 고정된 주소를 가진 사용자 서버를 등록시켜놓고 이를 통해 서비스하는 방법인데 아무래도 개인 사용자가 하기에는 부담이 따르겠죠.
다른 한가지는 Long-Polling을 사용하는 방식인데, 서버를 열어놓지 않고 새로운 메시지가 있는지 여부를 약간 긴 주기로 계속적으로 확인하는 방법으로, 고정된 주소가 없는 개인 PC로도 서비스할 수 있는 장점이 있습니다. 여기서는 Long-Polling을 사용한 방식을 정리해보겠습니다.
다음은 Long-Polling에 의한 요청 처리가 구현되어, 기본적인 명령과 응답을 주고받는 기능이 동작하는 예제입니다.
입력된 요청 문자열을 텔레그램 봇 창에 메세지로 보내는 내용이 구현되어있습니다.
앱 상에서 요청 입력은 채널 포스팅에서 사용되었던 채널창이 아닌 텔레그램 봇 자체창에서 입력되어야 합니다
소스 중 <텔레그램 봇 토큰>은 아래 글을 참고하여 적당한 값으로 바꿔줍니다.
>> 텔레그램 봇 생성 및 봇 토큰 얻기
telegramagent.py :
# -*- coding: utf-8 -*-
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: # username 키 값 존재 체크하도록. 설정 안하면 없을 수 있음
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)
main.py :
# -*- coding: utf-8 -*-
from telegramagent import *
def myCallback(chatInfo):
print("chat id = %s" % chatInfo.chatId)
print("username = %s" % chatInfo.userName)
print("date = %s" % agent.convertDate(chatInfo.date, '%Y-%m-%d %H:%M:%S'))
print("text = %s" % chatInfo.text)
if chatInfo.text == "quit":
agent.setContinue(False)
else:
agent.sendMessage(chatInfo.chatId, "입력된 요청 : %s" % chatInfo.text)
agent = TelegramAgent()
agent.setDumpData(True)
agent.setToken("<텔레그램 봇 토큰>")
agent.setCallback(myCallback)
agent.loop()
telegramagent.py 주요 행 설명
7 : 요청을 보낸 사용자 정보가 담기는 클래스입니다.
77 : chatId가 채널 ID일 경우 따옴표로 감싸줍니다.
133 : Long-Polling 시, update id를 0으로 설정할 경우, 기존에 가지고 있는 상태 데이터를 모두 가져오고, 두번째 호출부터는 마지막 update id + 1로 설정하면 기존의 데이터는 무시하고 새로운 데이터만 가져오게 됩니다.
main.py 주요 행 설명
11 : quit 명령을 받으면 서버를 종료합니다.
13 : Callback 함수에서 입력된 요청 문자열을 텔레그램 봇 창에 메세지로 보내줍니다.
16 : 텔레그램 서버에서 받은 응답 정보를 출력합니다. 실사용시는 False로 합니다.
18 : 요청을 처리할 Callback 함수를 등록하여 사용합니다.
정찰 위성 이미지
참고 사이트
더 많은 텔레그램 봇 라이브러리들 : https://core.telegram.org/bots/samples
<테스트 환경>
- OS : Windows 10
- Python 버전 : 3.6
