라즈베리파이 텔레그램 챗봇 사용하기 (봇 생성 과 통신) Raspberry Pi creating telegram bot

Telegram Desktop
다운로드
https://desktop.telegram.org/

1. 일단 가입

가입을 위해서는 SMS 되는 단말 필요 SMS인증함
이미 가입 하여 사용중이라면 추가 해당 내용 필요 없습니다. 다음 내용은 모바일로 진행해도 무방합니다.

2. 검색 목록에서 @botfather 입력

3. /start -> /newbot -> name생성 -> username생성(중복이 안되어야함)

4. access token 정보는 저장해야 두어야 하며, 다른 사람과 공유하지 않도록 해야합니다.

5. 라즈베리파이에서 설치

sudo pip3 install telepot
아래와 같은 에러 발생시 재시도
TypeError: unsupported operand type(s) for -=: 'Retry' and 'int'

6. get id 봇 검색 (내 계정으로 메시지를 보낼때 사용)

start 하면 id 알려줌
Your Chat ID = XXXXXXXXXX
범용적인 챗봇에서는 사용하지 않고, 특정 사용자에게 보낼때만 필요합니다.

7. 특정 사용자에게 메신저 보내기

test.py

import telepot
 
my_token = '123456789:........................' # bot 의 token 입력
bot = telepot.Bot(my_token) #봇을 생성해줍니다.

msg = 'message test'
telegram_id = '123412345' # 수신처 아이디 입력

bot.sendMessage(chat_id = telegram_id, text = msg) #메시지를 보냅니다.

다음과 같은 오류가 날때
raise exception.TelegramError(description, error_code, data)
telepot.exception.TelegramError: ('Bad Request: chat not found', 400, {'ok': False, 'error_code': 400, 'description': 'Bad Request: chat not found'})

위에서 만든 chat bot과 대화가 start가 되어있는지 확인 필요, 없다면 대화 시작된 상태에서만 메신저가 갑니다.

8. 응답 챗봇

이번에는 좀 더 복잡한 예제입니다. 메신저가 들어온 사용자에게 답변을 하는 예제입니다.

#-*- coding: utf-8 -*-

import time
import telepot
from telepot.loop import MessageLoop
 
def handle(msg):
    content_type, chat_type, chat_id = telepot.glance(msg)
 
    if content_type == 'text':
        if msg['text'].upper() == 'hello':
            bot.sendMessage(chat_id, 'good morning')
        elif msg['text'] == '/start':
            pass
        else:
            bot.sendMessage(chat_id, '지원하지 않는 기능입니다')
 
 
TOKEN = '1234567:................_....'  # 텔레그램으로부터 받은 Bot 토큰

bot = telepot.Bot(TOKEN)
MessageLoop(bot, handle).run_as_thread()

print ('Listening ...')
while True:
    time.sleep(1000)

실행 예제

9. 몇가지 기능 추가

9.1. 사용자 관리 (/start 들어올때마다 saving.dat 파일에 사용자 추가함)
9.2. 더 많은 명령어 (cpu 정보와, disk정보보기 명령 추가)
9.3. wakup noti 기능 추가 (시스템 시작될때마다 메세지 보냄)
9.4. DiskFull 일때 알림 기능

#-*- coding: utf-8 -*-

import time
import telepot
from telepot.loop import MessageLoop
import os
import sys
import psutil
datafile = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'saving.dat')
all_users = []


def load_users(file):
 try:
  f = open(file,"r")
  datas = f.readlines()
  f.close()
  users = []
  for data in datas :
   if len(data.strip())>0 :
    users.append(data.strip())
  return list(set(users))
 except:
  return []

def save_users(file,users):
 f = open(file,"w")
 for user in users :
  f.write(str(user).strip()+"\n")
 f.close()
 os.chmod(file, 0o777)

def handle(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

def meminfo(content_type, chat_type, chat_id):
 memory_usage = os.popen("cat /proc/meminfo").read()
 return True, memory_usage

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

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"

def cpuinfo(content_type, chat_type, chat_id):
 text1 = psutil.sensors_temperatures()
 text2 = psutil.cpu_stats()
 text3 = os.popen("uptime").read()
 return True, str(text1)+"\n"+str(text2) + "\n"+str(text3)

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

# Wakeup notification
def wakeup_noti(bot):
 global all_users
 users=[]
 exception = False
 for user in all_users:
  try:
   bot.sendMessage(user, 'system restarted')
   users.append(user)
   
  except telepot.exception.BotWasBlockedError:
   exception = True
  except:
   time.sleep(5)
   pass
 if exception == True :
  all_users = list(set(users))
  save_users(datafile,all_users)

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

# 시스템 network 될때까지 시간이 걸립니다.
time.sleep(20)

TOKEN = '12234567:tttttttttttttttttttttttttttttt'  # 텔레그램으로부터 받은 Bot 토큰
all_users = load_users(datafile)

# Bot Setup
bot = telepot.Bot(TOKEN)

wakeup_noti(bot)
MessageLoop(bot, handle).run_as_thread()

# Main Loop
print ('Listening ...')
while True:
 check_system_info(bot)
 time.sleep(60*60) # 1 hours

10. 켜질때마다 챗봇 서버 동작하기

https://swlock.blogspot.com/2019/03/python-auto-start-python-script-at.html