텔레그램 봇과 MySQL 연동하기

Preview

새로운 게시글 알림 봇, Open API를 이용한 업데이트 등 다양한 기능을 할 수 있는 텔레그램 봇을 여러 개 운영하고 있다. 봇을 더 사용자 친화적으로 사용할 수 있는 방법이 없을까 고민하던 중 메시지마다 경험치를 부여하여 채팅을 할수록 레벨업을 하는 기능을 구현해 보았다. 단체 채팅방에서 사용하면 재미있는 기능이며, 텔레그램 메시지 핸들러 사용을 위한 python과 서버 구축을 위한 MySQL을 통해 구현하였다.

 

[동작환경]
Raspberry pi 3+
python 3.8

pip install

일반 알림만을 제공하는 Cron데몬을 사용하는 텔레그램 봇과는 다르게 사용자와 상호 작용을 해야 하기 때문에 코드가 서버 상에서 24시간 동작해야 한다. 메시지 갱신여부, 종류 판단(텍스트인지, 사진인지, 영상인지), 대화처리 핸들러가 포함되어 있는 telegram.ext 모듈을 import 하여 사용한다.

공식 문서는 다음과 같다.

https://python-telegram-bot.readthedocs.io/en/stable/telegram.ext.html

telegram.ext package — Python Telegram Bot 13.0 documentation

pip install python-telegram-bot --upgrade

그리고 사용을 위해 import 해준다.

from telegram.ext import Updater, MessageHandler, Filters, CommandHandler, ConversationHandler

 

토큰 발급

기존 발급 게시글이 있어 참고하여 발급받는다.

https://gomguk.tistory.com/7

[BeautifulSoup4] Python 게시판 크롤러 텔레그램 봇 제작기(2)

def main

메시지 처리 핸들러에서 메시지가 업데이트 되면 그 정보를 전달받아야 한다. 텔레그램 메시지에는 우리가 대화창에서 볼 수 있는 정보를 포함하여 채팅방 id, 보낸 사람의 id 등 여러가지 정보를 json으로 담고 있다.

 

class Telebot:
    def __init__(self):
        self.my_token = my_token
        self.fixdict['galgori'] = 100
        
    def get_message(self, update, context) :
        # context.bot.send_chat_action(chat_id=update.effective_message.chat_id, action=context.bot.ChatAction.TYPING)
        logging.info(f"message >> {update.message.from_user['first_name']} {update.message.text}")
    
        if update.message.text == '?':
            output = f"{update.message.from_user['first_name']}님, 다음 패치까지 갈고리 {self.fixdict['galgori']}개 남음! v｡◕‿◕｡v"
            self.fixdict['galgori'] -= 1
            update.message.reply_text(output)
            
    def main(self):
    	updater = Updater(self.my_token, use_context=True)
    	dp = updater.dispatcher
        dp.add_handler(MessageHandler(Filters.text, self.get_message))
    
    	updater.start_polling(timeout=3, clean=True)
    	updater.idle()

main 함수를 보면 dp 를 통해 디스패처를 선언하고 add_handler를 통해 핸들러를 추가한다. Filters를 통해 그 메시지가 텍스트인지, 영상인지 종류를 판단한다. 여기에서는 메시지 핸들러가 text 메시지를 감지했을 때 두 번째 인자인 함수를 호출하는 것이다. 미리 정의해둔 get_message에서 메시지에 대한 정보를 처리하고 메시지를 보내든, 서버에 값을 저장하든 할 수 있는 것이다.

또한 메시지를 계속해서 polling 방식으로 수신하면서 update는 idle 상태로 대기하도록 한다. python 코드가 계속 실행되면서 메시지를 가져오는 것이다.

예제에서는 사용자가 '?' 메시지를 보냈을 때 output에 담긴 문자열을 사용자의 메시지에 '답장(reply)'하는 형태로 전송하도록 하였다.

메시지를 보내면, 봇이 답장을 합니다

코드에서도 알 수 있듯 update.message.from_user['first_name']에는 메시지를 보낸 사람의 이름, update.message.text는 메시지의 문자열이 전달 되는 것을 알 수 있다. 이는 디스패처에서 update, context를 통해 전달된 정보이며 다른 많은 정보도 포함하고 있으니 공식 문서를 참고하면 된다.

https://python-telegram-bot.readthedocs.io/en/stable/telegram.message.html

telegram.Message — Python Telegram Bot 13.0 documentation

MySQL 과 연동하기

텔레그램 메시지와 파이썬과의 연결을 마쳤으니, 파이썬과 MySQL과 연동해주면 된다. 

MySQL에서 디비와 테이블을 생성한다. 과정은 따로 설명하지 않으며 사용한 테이블의 스키마는 사진과 같다.

 

id, username, level, chat_id, exp

import pymysql
class Telebot:
    def __init__(self):
        self.leveldic = {}
        self.fixdict = {}
        self.burningflag = 1 # 경험치 배수 이벤트용 (기본 : 1배)
        self.exp_table = [math.floor(pow(math.ceil(((lv - 1) * 50 / 49)), 2.5) * 6) for lv in range(1, 70)]
        
    def putlevel(self, update, context, user, score=2):
        try:
            expdb = pymysql.connect(host='input.server.host.here', user='input.userid', password='input.password', database='input.db.info')
            expcur = expdb.cursor()
            expcur.execute(f"SELECT level, exp FROM accounts WHERE chat_id = '{user}'")
            level, exp = expcur.fetchone()

            if self.burningflag != 1:
                if score > 0:
                    score = score * self.burningflag
            if user in self.leveldic.keys():
                if exp + score > 0:
                    expcur.execute(f"UPDATE `accounts` SET EXP=EXP+{score} WHERE `chat_id` = '{user}'")
            else:
                expcur.execute(f"UPDATE `accounts` SET EXP=EXP+{score} WHERE `chat_id` = '{user}'")
            expdb.commit()
            expcur.execute(f"SELECT level, exp FROM accounts WHERE chat_id = '{user}'")
            level, exp = expcur.fetchone()
            print(f"현재 레벨 {level}, 현재경험치 {exp}, 테이블 경험치{self.exp_table[level]}")
            if exp > self.exp_table[level]:
                print(f"현재경험치 {exp}, 테이블 경험치{self.exp_table[level]}")
                print(level)
                level += 1
                expcur.execute(f"UPDATE `accounts` SET level=level+1 WHERE `chat_id` = '{user}'")
                expdb.commit()
                context.bot.send_message(chat_id=update.effective_message.chat_id,
                                 text=f"####레벨####\n{str(user)}님 레벨업! <<LEVEL {level}>>")
            if exp < self.exp_table[level-1] and exp >= 0:
                print(f"현재경험치 {exp}, 테이블 경험치{self.exp_table[level]}")
                level -= 1
                expcur.execute(f"UPDATE `accounts` SET level=level-1 WHERE `chat_id` = '{user}'")
                expdb.commit()
                context.bot.send_message(chat_id=update.effective_message.chat_id,
                                 text=f"####레벨####\n{str(user)}님 레벨 강등! <<LEVEL {level}>>\n한끗에 5억을 태울건가요?")
            elif exp < 0:
                context.bot.send_message(chat_id=update.effective_message.chat_id,
                                 text=f"####레벨####\n{str(user)}님 레벨 위험! <<LEVEL {level}>>\n레벨이 너무 낮아요! 점수는 유지되었어요!")
        finally:
            expdb.commit()
            expdb.close()

메시지 하나 당 2 경험치씩 쌓이도록 했으며, 경험치 배수 이벤트 플래그(burningflag)도 설정하여 메시지당 점수도 다르게 할 수 있도록 하였다. 경험치테이블(exp_table)은 실제 다른 게임에서 사용하고 있는 것을 참고하였으며 변수를 조절하여 그 간격을 조정할 수 있다. 레벨 단위별 기준 경험치를 준비하여 경험치가 기준 레벨보다 높으면 레벨 업을 한다. 그 반대의 경우는 강등도 구현하였고, 강등 구현 시 0점 이하로 점수가 떨어지지 않도록 예외처리하였다.

 

기능 추가하기

        elif '잘자요' in update.message.text:
            if random.random() >= 0.8:
                update.message.reply_text("슬기도 잘자요![+1000]")
                self.putlevel(update, context, update.message.from_user['first_name'], 1000)
            else:
                update.message.reply_text("자요?? [-100]")
                self.putlevel(update, context, update.message.from_user['first_name'], -100)

메시지와 다양한 기능을 결합하여 재미를 추가할 수 있다. 위의 코드는 "잘자요" 메시지를 받고 20% 확률로 잘자요와 함께 보너스 점수를 답하고, 80%의 확률로 마이너스를 부여하도록 했다. 

실패하면 강등뿐

테이블 값 확인하기

일정 조건에 의해서만 값을 확인할 수 있다면 랭킹확인 및 본인 경험치 확인에 어려움이 있다. 이를 위한 기능을 추가한다.

    def printProgressBar(self, iteration, total, prefix='->', suffix='', decimals=1, length=15, fill='■'):
        percent = ("{0:." + str(decimals) + "f}").format(100*(iteration/float(total)))
        filledLength = int(length * iteration // total)
        bar = fill * filledLength + '□' *(length - filledLength)
        return '\r%s |%s| %s%% %s' % (prefix, bar, percent, suffix)
        
    def getlevel(self, update, context):
        logging.info('>>> GET LEVEL')
        if context.args[0] is None:
            user = str(update.message.from_user["first_name"])
        else:
            user = context.args[0]

        try:
            expdb = pymysql.connect(host='', user='', password='', database='')
            expcur = expdb.cursor()
            expcur.execute(f"SELECT level, exp FROM accounts WHERE chat_id = '{user}'")
            level, exp = expcur.fetchone()

            pre_exp = self.exp_table[level-1]
            next_exp = self.exp_table[level+1]

            p_total = next_exp - pre_exp
            p_now = exp - pre_exp
            p_remain = p_total - p_now

            bar = self.printProgressBar(p_now, p_total, prefix=f"lv. {level}")
            context.bot.send_message(chat_id=update.effective_message.chat_id,
                             text=f"####레벨####\n{user}님 점수\n{bar}\n다음 레벨까지 {p_remain}점 남았어요!")

        finally:
            expdb.commit()
            expdb.close()

명령어를 통한 경험치 확인

ProgressBar함수를 통해서 텍스트로 그래프를 만들고 점수, 현재 레벨, 퍼센트를 표시하도록 하였다. 사용하기 위해서 핸들러를 추가 등록하여 봇 명령어를 사용하였다. 사용 방법은 다음 포스트에서 소개한다.

 

Review

사이드프로젝트에 텔레그램 봇만큼 손쉽게 할 수 있는 게 없는 것 같다. 같은 기능을 웹이나 앱으로 구현하고자 했다면 프론트엔드 구현에 더 많은 시간이 필요했을 것이다. 현재는 소개한 기능 외에도 더 많은 기능을 추가하여 단체 채팅방에서 운영중이다. 라즈베리파이 서버에서 구동중이지만 큰 문제없이 잘 동작하고 있다. 다음 포스트에서는 명령어 핸들러 추가를 통한 동작, 사진 데이터 처리에 대해 알아보려 한다.