텔레그램 챗봇 사용하기 (봇 생성 과 통신) in python

예전에 라즈베리파이로 구현을 했었는데 일반 PC에서 사용이 필요하여 다시 정리해보았습니다. 소스도 PC에서 실행 가능하도록 변경하였습니다.

0. 사용 용도

휴대 기기와 bot을 대화형으로 bot과 연결된 장치의 상태를 전달 받을 수 있고 bot에게 명령을 내려 bot과 연결된 다른 장치를 조작할 수 있습니다.

대안으로는 카카오톡 OpenAPI나 slack을 사용해도 됩니다.

* 텔레그램 봇의 장점으로는 봇만드는것이 간단합니다.

* 메세지 보내는것에 제한이 없습니다. (카카오톡의 경우 현재 기준으로는 제한 있음)

* 가입이 간단합니다.

1. 일단 가입

모바일 장치에 어플을 설치하면 SMS 인증합니다. (SMS 되는 단말 필요)

2. 봇만들기

BotFather 챗봇과 대화형으로 이루어짐

검색에서 @BotFather 검색 후 선택 > 시작 버튼 누르기 ( BotFather 비슷한 이름이 많으니 잘 확인해서 선택해야합니다. )

BotFather에서 /newbot 입력해줍니다.

이름 입력 

봇이름 입력 ( 끝에 'bot' 이 들어가야함 )

HTTP API:로 시작하는 access token 값 저장해둠

3. python 라이브러리 설치

pip3 install telepot 

이제는 업데이트를 안해서 실행시키면 exception이 발생합니다.

fork된 프로젝트를 설치해야합니다.

window 경우 git 을 먼저 설치해줍니다. linux에서는 알아서 git 설치하면 됩니다.

https://git-scm.com/download/win

telepot을 업그레이드 합니다.

pip install git+https://github.com/MoumenKhadr/telepot

4. 챗봇 실행

챗봇을 대화방으로 초대하면 명령을 내리거나 챗봇이 특정 상황에 노티를 줍니다.

만들어진 소스 참고바랍니다.

4.1. 사용자 관리 (/start 들어올때마다 사용자 추가함, 사용자 목록을 파일로 저장합니다. data 폴더에 json 포맷으로 저장됩니다.)

4.3. wakup noti 기능 추가 (시스템 시작될때마다 모든 사용자에게 메세지 보냄)

전체 소스

#-*- coding: utf-8 -*-
# window git https://git-scm.com/download/win
# pip install git+https://github.com/MoumenKhadr/telepot
import time
import telepot
from telepot.loop import MessageLoop
import os
import sys
import psutil
import threading
import config_key
import traceback
import json

TELEGRAM_BOT_ACCESS_TOKEN = config_key.TELEGRAM_BOT_ACCESS_TOKEN #'12234567:tttttttttttttttttttttttttttttt'  # 텔레그램으로부터 받은 Bot 토큰

bot = None
all_users = []
base_folder = os.path.join(os.path.dirname(os.path.realpath(__file__)), '../data')
datafile = os.path.join(os.path.dirname(os.path.realpath(__file__)), '../data/users.dat')


# Bot init
def init():
	global bot
	global all_users
	bot = telepot.Bot(TELEGRAM_BOT_ACCESS_TOKEN)
	MessageLoop(bot, handler_func).run_as_thread()
	if base_folder!='' and not os.path.exists(base_folder):
		os.makedirs(base_folder)
	all_users = load_users(datafile)
	wakeup_noti()

# Wakeup notification
def wakeup_noti():
	noti_all_members('system restarted')

def noti_all_members(message):
	global all_users
	users=[]
	exception = False
	for user in all_users:
		try:
			bot.sendMessage(user, message)
			users.append(user)
		except telepot.exception.BotWasBlockedError:
			exception = True
		except:
			traceback.print_exc()
			time.sleep(2)
			pass
	if exception == True :
		all_users = list(set(users))
		save_users(datafile,all_users)

def handler_func(msg):
	try:
		content_type, chat_type, chat_id = telepot.glance(msg)
		#print(content_type, chat_type, chat_id)
		find = False
		if content_type == 'text':
			for command in COMMANDS_LIST:
				if msg['text'].lower().strip() == command[0]:
					ret, data = command[1](content_type, chat_type, chat_id)
					if ret == True :
						bot.sendMessage(chat_id, data)
					find = True
			if find == False:
				bot.sendMessage(chat_id, 'not supported command')
	except:
		pass

# bot command meminfo
def meminfo(content_type, chat_type, chat_id):
	if sys.platform == 'win32':
		return True, 'not supported command on windows'
	memory_usage = os.popen("cat /proc/meminfo").read()
	return True, memory_usage

# bot command diskusage
def diskusage(content_type, chat_type, chat_id):
	obj_Disk = psutil.disk_usage('/')
	str = "Disk Total %5fGB\nDisk used %5fGB\nDisk Free %5fGB\nusage %4f%%"%((obj_Disk.total / (1024.0 ** 3)) , (obj_Disk.used / (1024.0 ** 3)) , (obj_Disk.free / (1024.0 ** 3)) , (obj_Disk.percent))
	return True, str

# bot command cpuinfo
def cpuinfo(content_type, chat_type, chat_id):
	if sys.platform == 'win32':
		return True, 'not supported command on windows'
	text1 = psutil.sensors_temperatures()
	text2 = psutil.cpu_stats()
	text3 = os.popen("uptime").read()
	return True, str(text1)+"\n"+str(text2) + "\n"+str(text3)
	
# bot command start_chat
def start_chat(content_type, chat_type, chat_id):
	global all_users
	all_users.append(chat_id)
	all_users = list(set(all_users))
	save_users(datafile,all_users)
	return True, "system started"

COMMANDS_LIST=[('meminfo',meminfo),('/start',start_chat),('diskusage',diskusage),('cpuinfo',cpuinfo)]

def save_users(file,users):
	f = open(file,"w")
	json.dump(list(set(users)), f)
	f.close()
	os.chmod(file, 0o777)

def load_users(file):
	try:
		f = open(file,"r")
		users = json.load(f)
		f.close()
		return list(set(users))
	except:
		return []

def check_system_info(bot):
	obj_Disk = psutil.disk_usage('/')
	if obj_Disk.percent > 70 :
		# Warning
		for user in all_users:
			try:
				bot.sendMessage(user, 'Storage full:%f%%'%(obj_Disk.percent))
			except:
				pass

def thread(logger):
	init()
	# Main Loop
	print ('Listening ...')
	while(True):
		check_system_info(bot)
		time.sleep(5)

iThread = None

def Start(logger):
	global iThread
	iThread = threading.Thread(target=thread, args=[logger])
	iThread.daemon = True
	iThread.start()
	return iThread

if __name__ == "__main__":
	Start(None)
	while(True):
		time.sleep(60*60)

주기적으로 어떤 동작후에 메세지를 보내도록 구현하려면 아래 함수와 비슷하게 작업하면 됩니다.

check_system_info(bot)

위 소스는 config_key.TELEGRAM_BOT_ACCESS_TOKEN 부분이 빠져있습니다. 이것은 앞서 설명한 API token을 넣어주면 됩니다.

5. 디버깅 테스트 방법

챗봇과 대화하려면 그룹방은 안되고 1:1 대화를 하여야합니다.

BotFather대화 내용을 t.me/봇이름 으로된 링크가 보이는데 해당 링크를 누르면 챗봇과 대화가 가능합니다.

아니면 그룹을 만든뒤 그룹에서 봇을 추가해서 봇과 다시 1:1대화를 생성하면 됩니다.  

저장된 목록과 사용자가 mismatch하게되면 wakeup noti를 보내지 못합니다. 그래서 해당 동작의 sync가 필요하면 방을 삭제하고 다시 1:1대화를 새로 시작하던가, 봇에게 /start 명령을  수동으로 보내도록 합니다.

기본적으로 /start는 봇과 대화가 시작될때 자동으로 보내지게 되지만 디버그 용도로 수동으로 보내도 됩니다.