텔레그램봇을 활용한 유저 채팅 데이터 수집 및 활용(feat. telepot, telegram)

텔레그램봇을 활용한 유저데이터 수집 및 활용

메시지 서비스인 텔레그램을 이용해서 다양한 유저 데이터 수집과 유저 반응에 따른 자동 서비스를 구현할 수 있다. [목차] 1. 텔레그램 봇 만들기 2. 텔레그램 봇 설정   - 그룹 메시지 보기(disabled 설정)   - 관리자 권한 획득 3. 그룹챗에서 getupdates를 이용하여 데이터 수집 4. 그룹챗에 메시지 보내기 5. 유저반응에 따라 대응하는 자동 응답형 봇 만들기

 

1. 텔레그램 봇 만들기

- BotFather를 이용한 텔레그램 봇 만들기

• 텔레그램 방에 BotFather를 검색하여 대화 시작한다.
• ‘/명령어’ 를 이용해 다양한 봇 생성 및 세팅을 할수 있다.

/newbot - create a new bot /mybots - edit your bots [beta] Edit Bots /setname - change a bot's name /setdescription - change bot description /setabouttext - change bot about info /setuserpic - change bot profile photo /setcommands - change the list of commands /deletebot - delete a bot Bot Settings /token - generate authorization token /revoke - revoke bot access token /setinline - toggle inline mode /setinlinegeo - toggle inline location requests /setinlinefeedback - change inline feedback settings /setjoingroups - can your bot be added to groups? /setprivacy - toggle privacy mode in groups Games /mygames - edit your games [beta] /newgame - create a new game /listgames - get a list of your games /editgame - edit a game /deletegame - delete an existing game

- 생성 순서는 간단하다 :  /newbot → ‘봇이름 입력 ~Bot or _bot’ → 봇이름 입력 → api key를 준다.

2. 텔레그램 봇 설정

- 수집하고자 하는 챗방이 있으면 봇은 해당 챗방에 꼭 들어가야함! --> 옵저버 역할이라고 생각하면 된다.

- 텔레그램 봇을 이용하여 채팅을 수집하고 보내려면 간단한 설정이 필요하다.

- 그룹 채팅방에서 관리자외 다른 유저의 메시지 까지 확인하려면 아래 두가지 설정을 해주어야 한다. 

1. 채팅방 내 봇 관리자 계정

2. /setprivacy 를 disable로 변경

 

3. 그룹챗에서 getupdates를 이용하여 데이터 수집

- 여기서는 telepot이라는 라이브러리를 활용하여 데이터를 수집한다.

• getUpdates 는 봇이 있는 챗방에서 업데이트된 모든 형태를 가지고 온다.
• 이미 가져온 업데이트 내역은 다시 조회할 필요가 없으므로, 가장 최근의 update_id를 이용하여 offset을 입력해줘, 새로운 update내역만 받아올수 있도록 한다.
• getUpdates(offset=(update_id + 1))
• json형태의 update목록이 나오므로, 수집하고자하는 데이터에 맞게 parse해서 구성하면 된다.

[코드]

import telepot
from pprint import pprint
import time
from telepot.loop import MessageLoop
import pandas as pd



def get_bot(api_key):
   bot = telepot.Bot(api_key)
   info = bot.getMe()
   pprint(info)

   return bot


def get_update(bot, offset=None):
   '''
   offset : last update_id +1 -> 최근 받아온 업데이트 아이디 이후 것으로 설정
   '''
   resp = bot.getUpdates(offset)

   return resp


def get_offset(updates):
   last_update_id = updates[-1]['update_id']
   offset = last_update_id + 1

   return offset


def get_datetime(timestamp):
   from datetime import datetime
   from pytz import timezone

   KST = timezone('Asia/Seoul')
   date = datetime.fromtimestamp(timestamp).astimezone(KST).strftime('%Y-%m-%d')
   date_time = datetime.fromtimestamp(timestamp).astimezone(KST).strftime('%Y-%m-%d %H:%M:%S')

   return date, date_time


def get_username(upt):
    temp = ''
    lst = ['username','first_name','last_name']
    name_lst = [i for i in lst if i in upt['message']['from']]
    if name_lst[0] == lst[0]:
        return upt['message']['from'][name_lst[0]]
    else:
        for i in name_lst:
            temp += upt['message']['from'][i]
        return temp


def get_df(updates):
    title, author, author_name, datetime, message, basis_dt = [], [], [], [], [], []
    for upt in updates:
        if 'text' in upt['message']:
            date, date_time = get_datetime(upt['message']['date'])
            user_name = get_username(upt)
            title.append(upt['message']['chat']['title'])
            author.append(upt['message']['chat']['id'])
            author_name.append(user_name)
            datetime.append(date_time)
            message.append(upt['message']['text'])
            basis_dt.append(date)

    return pd.DataFrame(
        {'title': title, 'author': author, 'author_name': author_name, 'datetime': datetime, 'message': message,
         'basis_dt': basis_dt})


def main(api_key, offset, destination_table, project_id, cycle_sec, limit_cnt):
    bot = get_bot(api_key)
    cnt = 0
    while True:
        updates = get_update(bot, offset)
        if len(updates) > 0:
            offset = get_offset(updates)
            df = get_df(updates)
            table_schema = [{'name': 'title', 'type': 'STRING'},
                            {'name': 'author', 'type': 'STRING'},
                            {'name': 'author_name', 'type': 'STRING'},
                            {'name': 'datetime', 'type': 'DATETIME'},
                            {'name': 'message', 'type': 'STRING'},
                            {'name': 'basis_dt', 'type': 'DATE'},
                            ]
            if len(df) > 0:
                df.to_gbq(destination_table=destination_table, project_id=project_id, if_exists='append',
                          table_schema=table_schema)
                print(f'SUCCESS : telegram data load, rows:{len(df)}')

        cnt += 1
        if cnt > limit_cnt:
            break

        time.sleep(cycle_sec)

    return 'SUCCESS'


if __name__ == '__main__':
    API_KEY = <telegrambot API_KEY xxxxx:yyyyy>
    offset = None
    destination_table = <data 적재 테이블>
    project_id= <data 적재 프로젝트>
    sleep_sec = <적재 주기>
    limit_cnt = <제한 적재 횟수>
    main(API_KEY, offset, destination_table, project_id, cycle_sec, limit_cnt)

[수집데이터 결과]

 

4. 그룹챗에 메시지 보내기

- 메시지를 보낼 chat_id를 이용해 간단하게 전송 가능

bot = telepot.Bot(API_KEY)
bot.sendMessage(chat_id,'hello~')

 

5. 유저반응에 따라 대응하는 자동 응답형 봇 만들기

- 이부분은 https://jow1025.tistory.com/317 로부터 가져옴

- 간단히 핸들러를 만들어서 예상되는 text에 따라 자동 응답 기능을 만들 수 있다.

import telegram
from telegram.ext import Updater
from telegram.ext import MessageHandler, Filters
 
token = "토큰 값"
id = "id 값"
 
bot = telegram.Bot(token)
 
 
# updater
updater = Updater(token=token, use_context=True)
dispatcher = updater.dispatcher
#봇 
updater.start_polling()
 
def handler(update, context):
    user_text = update.message.text 
    if user_text == "ㅋㅋ": 
        bot.send_message(chat_id=id, text="안녕?") # 답장 보내기
    elif user_text == "넌 누구": 
        bot.send_message(chat_id=id, text="난 봇이야") # 답장 보내기
 
echo_handler = MessageHandler(Filters.text, handler)
dispatcher.add_handler(echo_handler)

 

https://core.telegram.org/bots/api#getchatmember

Telegram Bot API