[Python] 서버 없이 텔레그램 봇 만들기 예제

텔레그램 봇과 명령과 응답을 양방향으로 주고 받는 방법을 구현해보겠습니다. (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