[한라대학교 공지 알림 봇] 코드분석 (telegram bot)

Intro

이번 챕터에서는 telegram bot을 활용해서 공지가 생기면 알람이 오도록 구현해보겠다.

지금까지 만든 코드를 실행하면 위 같은 결과가 나온다.
공지를 제외한 1page에 있는 게시글을 전부 크롤링한 모습이다.

telegram bot

텔레그램 봇 정리본
위 정리본에 링크 되어 있는 문서를 참고하자.

봇 만들기

BotFather

텔레그램 봇은 BotFather을 통해 만들 수 있다.

만드는 방법은 스크린샷을 보고 따라하면 된다.
/start 메세지를 보내고
/newbot 메세지를 보내고
봇 이름을 적어주고
봇 사용자 이름을 적어주면 된다.
그러면 봇 주소와 토큰이 발행된다.
봇 주소로 들어가면 방금 만든 봇이 나타난다.

채널 만들기

봇을 이용하지 않고 채널을 통해 만든 이유는 봇은 사용자가 메세지를 보내면 응답을 해주는 방식이다.
하지만 채널은 새로운 공지가 있을경우 봇이 채널에 메세지를 보내는 방식이다.
사용자가 메세지를 보내야 한다는 조건이 붙지 않아서 채널을 이용했다.
그리고 채널은 관리자를 제외한 일반 유저들은 채팅을 치지 못한다.
그래서 공지 알림만 오게 하는 용도로 적합하다고 생각했다.

채널 개설 방법

스크린샷을 보고 잘 따라오면 채널을 쉽게 만들 수 있다.
중간에 채널 주소는 꼭 기억해야한다.

테스트 채널 만들기

실 서비스 채널이 있으면, 테스트 채널도 있어야 한다.

테스트 채널 개설 방법

테스트 채널은 위 채널 만드는 방식이랑 똑같이 만든 다음에

https://api.telegram.org/bot1081363002:AAGpV0UOaOLu7nbY-M4sTHCav6YIDoorHKQ/sendMessage?chat_id=@hallatest&text=h

https://api.telegram.org/bot(봇 토큰)/sendMessage?chat_id=@(채널 링크)&text=hi

위 주소로 들어가서 chat id를 확인한다.
그리고 설정에 들어가서 채널을 공개 채널에서 비공개 채널로 바꿔주면 된다.

코딩

import requests
from bs4 import BeautifulSoup
import sys
import telegram



my_token = '봇 토큰'
my_chat_id = '채널 주소 / 서비스 채널은 @*** / 테스트 체널은 -***'
req = requests.get('http://www.halla.ac.kr/mbs/kr/jsp/board/list.jsp?boardId=23401&mcategoryId=&id=kr_060101000000')

client_errors = [400, 401, 403, 404, 408]
server_errors = [500,502, 503, 504]

if req.status_code in client_errors:
    print(req.status_code + ": 클라이언트 에러")
    sys.exit(1)
elif req.status_code in server_errors:
    print(req.status_code + ": 서버 에러")
    sys.exit(1)

bot = telegram.Bot(token=my_token)
html = req.text
soup = BeautifulSoup(html, "html.parser")
posts = soup.select("td > a")
num = soup.find_all(title='공지')

for i in range(len(num)):           # 공지로 위로 올라간 게시글 제외한 최신 게시글 분류
    del posts[0]

bot.sendMessage(chat_id=my_chat_id, text='일반공지 : ' + posts[0].text)

코드를 실행시키면 텔레그램 채널로 봇이 메세지를 보낸다.
그러면 봇 구현은 완성이다.
남은 건 최신 게시글인지 확인 후 최신 게시글이면 봇을 통해 메세지를 보내서 알리고, 아니면 메세지를 보내지 말아야한다.

open

os 정리본

이제 우린 어떤 글이 최신 글인지 확인하는 방법을 생각해야한다.

1. 공지 알림 봇은 기본적으로 텔레그램 채널에서 실행된다.
2. 코드는 서버에서 1시간마다 실행된다.

그러면 우리가 말하는 최신 글은, 글 옆에 주황색 N 박스가 아닌
서버에서 코드를 실행하는 주기(1시간) 안에 게시되는 모든 글이다.
다르게 말하자면, 텔레그램 채널에 봇이 알리지 못한 게시글을 뜻한다.

import requests
from bs4 import BeautifulSoup
import sys
import telegram
import os

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

my_token = '봇 토큰'
my_chat_id = '채널 주소 / 서비스 채널은 @*** / 테스트 체널은 -***'
req = requests.get('http://www.halla.ac.kr/mbs/kr/jsp/board/list.jsp?boardId=23401&mcategoryId=&id=kr_060101000000')

client_errors = [400, 401, 403, 404, 408]
server_errors = [500,502, 503, 504]

if req.status_code in client_errors:
    print(req.status_code + ": 클라이언트 에러")
    sys.exit(1)
elif req.status_code in server_errors:
    print(req.status_code + ": 서버 에러")
    sys.exit(1)

bot = telegram.Bot(token=my_token)
html = req.text
soup = BeautifulSoup(html, "html.parser")
posts = soup.select("td > a")
num = soup.find_all(title='공지')

for i in range(len(num)):           # 공지로 위로 올라간 게시글 제외한 최신 게시글 분류
    del posts[0]

if not(os.path.isfile(os.path.join(BASE_DIR, 'latest.txt'))):
    new_file = open('latest.txt', 'w+', encoding='utf-8')
    new_file.write(posts[0].text)
    new_file.close()

with open(os.path.join(BASE_DIR, 'latest.txt'), 'r+', encoding='utf-8') as f_read:
    before = f_read.readline()

    for post in posts:
        print('post = ' + post.text)
        print('before = ' + before)
        if before == post.text:
            print('최신글입니다.')
            break
        elif before != post.text:
            url = "http://www.halla.ac.kr/mbs/kr/jsp/board/" + post.get('href')
            print(url)
            bot.sendMessage(chat_id=my_chat_id, text="일반공지 : " + post.text)
            bot.sendMessage(chat_id=my_chat_id, text=url)

with open(os.path.join(BASE_DIR, 'latest.txt'), 'w+', encoding='utf--8') as f_write:
    f_write.write(posts[0].text)

가장 최근에 크롤링한 게시글.text를 txt파일에 저장하기로 결정했다.

1. latest.txt 파일이 있는 경우
2. latest.txt 파일이 없는 경우

두 경우가 나온다.
파일이 없을 경우 새로 만들어 줘야 한다.

if not(os.path.isfile(os.path.join(BASE_DIR, 'latest.txt'))):
    new_file = open('latest.txt', 'w+', encoding='utf-8')
    new_file.write(posts[0].text)
    new_file.close()

위 코드 처럼 파일을 새로 만든다음에 닫아주자

파일이 있는 경우는 파일을 열어서 저장된 내용을 변수에 저장한다.
파일의 내용은 게시글.text이기 때문에 크롤링한 게시글과 쉽게 비교할 수 있다.
크롤링한 데이터와 파일에 저장된 데이터를 비교하다가 게시글이 일치하면 더 비교할 필요 없이 종료하면 된다.
데이터가 일치하지 않을 경우는 bot을 통해 크롤링한 테이더를 메세지로 보내면 된다.

    for post in posts:
        print('post = ' + post.text)
        print('before = ' + before)
        if before == post.text:
            print('최신글입니다.')
            break
        elif before != post.text:
            url = "http://www.halla.ac.kr/mbs/kr/jsp/board/" + post.get('href')
            print(url)
            bot.sendMessage(chat_id=my_chat_id, text="일반공지 : " + post.text)
            bot.sendMessage(chat_id=my_chat_id, text=url)

마지막으로 코드를 한번 실행 했으므로 latest.txt 파일에는 최신 게시글(크롤링한 게시글 중 최신)이 들어가야 한다.
고로 마지막에 따로 저장해주자.

with open(os.path.join(BASE_DIR, 'latest.txt'), 'w+', encoding='utf--8') as f_write:
    f_write.write(posts[0].text)

에러처리

테스트를 여러번 해본 결과 텔레그램 봇은 url을 메세지로 보낼때 한번에 최대로 보낼 수 있는 양이 정해져 있는 듯하다.
한번에 최대 10개까지만 보낼 수 있다.
그 이상 보내면 코드의 속도도 느려질 뿐 아니라 timeout에러 메세지를 나타내면서 코드가 강제로 꺼진다.

            try:
                if post != posts[10]:
                    bot.sendMessage(chat_id=my_chat_id, text="일반공지 : " + post.text)
                    bot.sendMessage(chat_id=my_chat_id, text=url)
                else:
                    break
            except Exception as ex:
                print("timeout")
                break

따라서 위 예외 처리를 bot.sendMessage 부분에 넣어주면 된다.
어차피 1시간 간격으로 코드가 실행되기 때문에 최신 게시글이 10개 이상 나올 확률은 적지만, 가능성이 적다고해서 발생 가능한 에러를 무시하고 그냥 지나치는건.. 좀 곤란하다.

최종 코드

import requests
from bs4 import BeautifulSoup
import sys
import telegram
import os

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

my_token = '봇 토큰'
my_chat_id = '채널 주소 / 서비스 채널은 @*** / 테스트 체널은 -***'
req = requests.get('http://www.halla.ac.kr/mbs/kr/jsp/board/list.jsp?boardId=23401&mcategoryId=&id=kr_060101000000')

client_errors = [400, 401, 403, 404, 408]
server_errors = [500,502, 503, 504]

if req.status_code in client_errors:
    print(req.status_code + ": 클라이언트 에러")
    sys.exit(1)
elif req.status_code in server_errors:
    print(req.status_code + ": 서버 에러")
    sys.exit(1)

bot = telegram.Bot(token=my_token)
html = req.text
soup = BeautifulSoup(html, "html.parser")
posts = soup.select("td > a")
num = soup.find_all(title='공지')

for i in range(len(num)):           # 공지로 위로 올라간 게시글 제외한 최신 게시글 분류
    del posts[0]

if not(os.path.isfile(os.path.join(BASE_DIR, 'latest.txt'))):
    new_file = open('latest.txt', 'w+', encoding='utf-8')
    new_file.write(posts[0].text)
    new_file.close()

with open(os.path.join(BASE_DIR, 'latest.txt'), 'r+', encoding='utf-8') as f_read:
    before = f_read.readline()

    for post in posts:
        print('post = ' + post.text)
        print('before = ' + before)
        if before == post.text:
            print('최신글입니다.')
            break
        elif before != post.text:
            url = "http://www.halla.ac.kr/mbs/kr/jsp/board/" + post.get('href')
            print(url)
            try:
                if post != posts[10]:
                    bot.sendMessage(chat_id=my_chat_id, text="일반공지 : " + post.text)
                    bot.sendMessage(chat_id=my_chat_id, text=url)
                else:
                    break
            except Exception as ex:
                print("timeout")
                break

with open(os.path.join(BASE_DIR, 'latest.txt'), 'w+', encoding='utf--8') as f_write:
    f_write.write(posts[0].text)

다음시간에

다음 챕터에서는 서버에 코드를 올려서 자동으로 실행시키는 스케쥴링에 관해 다룰것이다.

참고문헌

Python Telegram Bot document, "https://blex.me/@mildsalmon/telegram", [mildsalmon]

OS document, "https://blex.me/@mildsalmon/os", [mildsalmon]

notice_alarm, "https://github.com/mildsalmon/notice_alarm", [mildsalmon]