AWS Lambda, API Gateway로 대화형 질답봇 만들기 (Feat. 야구)

    목차
반응형

전편: 2024.03.07 - [AWS] - AWS Lambda와 selenium을 이용해서 날씨알림봇 만들기

 

0. 배경

이전에 AWS Lambda와 텔레그램을 이용하여 날씨알림봇을 만들었다. 또 만들만한 봇이 뭐가 있을까 고민하다 평소 관심이 많은 야구와 관련된 봇을 만들고 싶어졌다. 제일 먼저 떠오른 건 실시간 경기 알림봇이었다. 내가 응원하는 팀이 앞서가는 득점 혹은 역전을 했거나 이겼을 때 나에게 알림을 주는 걸 생각했다. 하지만 몇 가지 현실적인 문제에 부딪혔다. 첫째로, 실시간으로 알림을 받는 서비스를 구축하는데 있어 Lambda 같은 서버리스 서비스보다 EC2처럼 24시간 돌아가는 서비스가 오히려 더 유리하다는 것이었다. 둘째로, KBO, 네이버, 다음 등 문자 중계를 해주는 제공처 모두 API가 없어 스크래핑을 해야하는데 이것도 만만치 않았다. 그래서 야구 관련한 다른 봇을 생각해봤을 때 과거 경기 기록을 검색하면 좋겠다는 생각을 해서 대화형 질답봇을 만드는 걸 목표로 했다.

이런 느낌

 

1. 스크랩

네이버, 다음, KBO 모두 경기 결과를 제공하지만 반응형 웹이기 때문에 selenium이 필수적이 된다. 하지만 statiz는 정적 웹이기 때문에 beautifulsoup만으로도 충분히 스크래핑이 가능하여 statiz의 데이터를 긁어오도록 했다. (모든 데이터를 스크랩해서 DB에 저장해놓고 DB에서 검색하는 것도 생각해보았지만, 나만 쓸 거라 요청이 매우 적을 거라서 DB를 사용하는 게 더 비효율적이고 비용 측면에서도 스크랩하는 게 낫겠다고 생각했다.)

 

본격적으로 스크랩하기 전에 필요 라이브러리를 불러오자.

from datetime import datetime, timedelta

import pandas as pd
from bs4 import BeautifulSoup
import requests

 

1-1. 스크랩할 페이지

스탯티즈는 정말 고맙게도 날짜 정보만 있으면 해당 날짜에 있었던 모든 경기를 확인할 수 있다.

f'https://statiz.sporki.com/schedule/?m=daily&date={date}'

예를 들어 2023년 5월 6일인 경우 https://statiz.sporki.com/schedule/?m=daily&date=2023-05-06으로 이동하면 된다. 이 페이지에서 박스스코어와 승패세 결과도 바로 볼 수 있으며 우천취소 여부도 확인할 수 있다. 

 

1-2. 응원할 팀 찾기

위처럼 각 파란색으로 칠해진 부분이 하나의 박스이다. 박스 내에 내가 찾고자 하는 팀이 있으면 그 박스를 계속 사용하도록 할 것이며, 만약 우천취소라면 더 이상 진행하지 않아도 된다.

yyyymmdd = '2023-05-06'
team = '롯데'
url = f'https://statiz.sporki.com/schedule/?m=daily&date={yyyymmdd}'
r = requests.get(url).text
soup = BeautifulSoup(r, 'lxml')

boxes = soup.find_all("div", "box_cont")
i = 0
summaries = []
for i, box in enumerate(boxes):
    if team in box.text:
    	box_header = soup.find_all("div", "box_head")[i]
        if '경기취소' in box_header.text:
            summaries.append(['경기취소'])
            continue
            
        result_table = table(box)
        result_text = result_text(box, result_table)
        
        summaries.append((result_table, result_text))

 

 

1-3. 박스스코어 스크랩

내가 수집할 팀의 스코어를 수집해보자. 최대 문제점은 승리확률인지 뭔지 스코어와 함께 %가 있다는 점이다. 다행스럽게도 %는 `p`으로 감싸져 있어 `p` 태그를 제거한 후 텍스트를 추출하면 스코어만 정확하게 수집할 수 있었다.

def table(box):
    columns = [b.text for b in box.find_all('th')]

    team1 = []
    team2 = []

    for td in box.find_all('tr')[1].find_all('td'):
        for p_tag in td.find_all('p'):
            p_tag.decompose()  # <p> 태그 제거
        text = td.get_text(strip=True)  # 텍스트 부분 추출
        team1.append(text)


    for td in box.find_all('tr')[2].find_all('td'):
        for p_tag in td.find_all('p'):
            p_tag.decompose()  # <p> 태그 제거
        text = td.get_text(strip=True)  # 텍스트 부분 추출
        team2.append(text)

    result_table = pd.DataFrame([team1, team2], columns=columns).set_index('팀').T

    return result_table

참고로 수집을 완료하면 테이블이 스탯티즈에서 보이는대로 가로로 길게 나오는데, 텔레그램에서는 테이블이 가로로 길면 모양이 깨지기 때문에 일부러 세로로 모양을 바꿨다.

 

1-4. 경기 결과 수집

아까 수집했던 box와 테이블로부터 어느 팀이 이겼고 승리투수, 패전투수, 세이브투수를 수집하는 코드이다.

def result_text(soup, result_table):
    if result_table.iloc[-1, 0] == result_table.iloc[-1, 1]:
        result = "무승부"
    else:
        team1 = result_table.columns[0]
        team2 = result_table.columns[1]

        team1_score = result_table.loc['R', team1]
        team2_score = result_table.loc['R', team2]

        if team1_score > team2_score:
            result = f"(승) {team1} {team1_score} : {team2_score} {team2} (패)"
        else:
            result = f"(패) {team1} {team1_score} : {team2_score} {team2} (승)"

        try:
            winner = soup.find_all("ul", class_="win")[0].find_all(['span', 'a', 'em'])[:3]
            result += f'\n{winner[0].text}: {winner[1].text} {winner[2].text}'
                
            loser = soup.find_all("ul", class_="lose")[0].find_all(['span', 'a', 'em'])[:3]
            result += f'\n{loser[0].text}: {loser[1].text} {loser[2].text}'
                
            saver = soup.find_all("ul", class_="save")[0].find_all(['span', 'a', 'em'])[:3]
            result += f'\n{saver[0].text}: {saver[1].text} {saver[2].text}'
        except:
            0
    return result

- 테이블의 마지막 row(R)에 따라 승리팀을 확인한다.

- 승/패/세 투수 정보는 각각 `win`, `lose`, `save` 클래스로 확인할 수 있는데 best, worst도 같은 태그로 묶여있어 `[:3]`로 슬라이싱을 해주었다.

- 무승부 경기나 세이브가 없는 경우가 있기 때문에 `try`문으로 감싸주었다.

 

그리고 경기 요약 페이지로 가면 해당 경기의 best, worst, 득점장면을 확인할 수 있다.

해당 데이터들도 비슷한 방법으로 수집해주었다. 해당 페이지로의 이동은 아래와 같이 할 수 있다.

game_summary_button = box.find('button', text='경기요약')
if game_summary_button:
    link = game_summary_button.parent['href']
    link = f'http://statiz.sporki.com{link}'

 

 

1-5. 순위 데이터 수집

경기 뿐만 아니라 각 연도별 순위 데이터도 확인할 수 있으면 좋겠다는 생각이 들어서 추가 수집해보았다. 이 데이터는 스탯티즈가 아닌 KBO 홈페이지에서 가져왔다. (이상하게 이 페이지만큼은 정적 페이지로 작동하여 쉽게 수집이 가능하였다.)

url = f"https://www.koreabaseball.com/Record/History/Team/Record.aspx?startYear={y}&halfSc=T"

 

`startYear`는 1980, 1990, 2000, 2010, 2020이 될 수 있다. (전,후기 순위는 따로 수집하지 않았다.)

순위를 확인할 연도를 입력받으면 해당 연도가 들어있는 연대의 순위 페이지로 이동 후 올바른 순서의 데이터를 찾는다.

ex) 2015년 → 2010년대 순위 페이지로 이동 → 2015년이므로 index=5의 테이블 확인

ex) 1984년 → 1980년대 순위 페이지로 이동 → 원년이 1982년이므로 index=4-2=2의 테이블 확인

 

이후로는 일반적인 테이블 데이터를 수집하는 것과 같아 별도 설명은 생략한다.

 

 

1-6. 데이터 수집 전체 코드

▼ 경기 결과 수집 전체 코드

텔레그램 전송용으로 만들었기 때문에 수집한 데이터 모두 텍스트로 변경하도록 작성하였다.

더보기
class KBOResultBot:
    
    def __init__(self, date, team='롯데'):
        self.date = self.convert_to_yyyy_mm_dd(date)
        self.team = team
        
    def convert_to_yyyy_mm_dd(self, date_str):
        if date_str == '오늘':
            return datetime.now().strftime("%Y-%m-%d")
        
        formats = ["%Y년 %m월 %d일", "%Y/%m/%d", "%Y-%m-%d", "%Y %m %d", "%Y.%m.%d", "%Y.%m.%d",
                   "%y년 %m월 %d일", "%y/%m/%d", "%y-%m-%d", "%y %m %d", "%y.%m.%d", "%y.%m.%d", "%y%m%d",
                  ]

        for fmt in formats:
            try:
                date_obj = datetime.strptime(date_str, fmt)
                return date_obj.strftime("%Y-%m-%d")
            except ValueError:
                pass

        return "날짜 형식을 인식할 수 없습니다."
        
    def summary(self):
        if self.date == "날짜 형식을 인식할 수 없습니다.":
            return ["날짜 형식을 인식할 수 없습니다."]
        
        url = f'https://statiz.sporki.com/schedule/?m=daily&date={self.date}'
        r = requests.get(url).text
        soup = BeautifulSoup(r, 'lxml')

        boxes = soup.find_all("div", "box_cont")
        summaries = []
        
        for i, box in enumerate(boxes):
            if self.team in box.text:
                box_header = soup.find_all("div", "box_head")[i]
                if '경기취소' in box_header.text:
                    summaries.append(['경기취소'])
                    continue
                result_table = self.table(box)
                result_text = self.result_text(box, result_table)
                    
                summaries.append(f'<b>{self.date}</b>\n{result_text}\n<pre>{result_table.to_markdown(tablefmt="grid")}</pre>')
                    
        return summaries
                
    def table(self, box):
        columns = [b.text for b in box.find_all('th')]
        
        team1 = []
        team2 = []
        
        for td in box.find_all('tr')[1].find_all('td'):
            for p_tag in td.find_all('p'):
                p_tag.decompose()  # <p> 태그 제거
            text = td.get_text(strip=True)  # 텍스트 부분 추출
            team1.append(text)


        for td in box.find_all('tr')[2].find_all('td'):
            for p_tag in td.find_all('p'):
                p_tag.decompose()  # <p> 태그 제거
            text = td.get_text(strip=True)  # 텍스트 부분 추출
            team2.append(text)
        
        result_table = pd.DataFrame([team1, team2], columns=columns).set_index('팀').T
        
        return result_table
    
    def result_text(self, soup, result_table):
        if result_table.iloc[-1, 0] == result_table.iloc[-1, 1]:
            result = "무승부"
        else:
            team1 = result_table.columns[0]
            team2 = result_table.columns[1]
            
            team1_score = result_table.loc['R', team1]
            team2_score = result_table.loc['R', team2]
            
            if team1_score > team2_score:
                result = f"(승) {team1} {team1_score} : {team2_score} {team2} (패)"
            else:
                result = f"(패) {team1} {team1_score} : {team2_score} {team2} (승)"
            
            try:
                winner = soup.find_all("ul", class_="win")[0].find_all(['span', 'a', 'em'])[:3]
                result += f'\n{winner[0].text}: {winner[1].text} {winner[2].text}'
                
                loser = soup.find_all("ul", class_="lose")[0].find_all(['span', 'a', 'em'])[:3]
                result += f'\n{loser[0].text}: {loser[1].text} {loser[2].text}'
                
                saver = soup.find_all("ul", class_="save")[0].find_all(['span', 'a', 'em'])[:3]
                result += f'\n{saver[0].text}: {saver[1].text} {saver[2].text}'
                
            except:
                0
        return result
    
    
    def best_worst(self):
        if self.date == "날짜 형식을 인식할 수 없습니다.":
            return ["날짜 형식을 인식할 수 없습니다."]
        
        url = f'https://statiz.sporki.com/schedule/?m=daily&date={self.date}'
        r = requests.get(url).text
        soup = BeautifulSoup(r, 'lxml')

        boxes = soup.find_all("div", "box_cont")
        links = []
        
        for i, box in enumerate(boxes):
            if self.team in box.text:
                box_header = soup.find_all("div", "box_head")[i]
                if '경기취소' in box_header.text:
                    summaries.append(['경기취소'])
                    continue
                game_summary_button = box.find('a', text='경기요약')
                if game_summary_button:
                    link = game_summary_button['href']
                    links.append(f'https://statiz.sporki.com{link}')
        bests_worsts = []
        for link in links:
            best_worst = f'<b>{self.date}</b>'
            titles = ['경기 Best 5', '경기 Worst 5']
            if link != '경기취소':
                r = requests.get(link).text
                soup = BeautifulSoup(r, 'lxml')

                tables = soup.find_all("table")[1:3]
                
                for i, table in enumerate(tables):
                    columns = [col.text.strip() for col in table.find_all('th')]
                    
                    rows = table.find_all('tr')
                    data = []
                    for row in rows[1:]:
                        cols = row.find_all('td')
                        row_data = [col.text.strip() for col in cols]
                        data.append(row_data)
                    best_worst += f"\n{titles[i]}\n<pre>{pd.DataFrame(data, columns=columns)[['이름', '활약']].to_markdown(tablefmt='grid', index=None)}</pre>"
            
            else:
                best_worst += '경기취소'
                    
            bests_worsts.append(best_worst)
        
        return bests_worsts
        
    def player_record(self):
        if self.date == "날짜 형식을 인식할 수 없습니다.":
            return ["날짜 형식을 인식할 수 없습니다."]
        
        url = f'https://statiz.sporki.com/schedule/?m=daily&date={self.date}'
        r = requests.get(url).text
        soup = BeautifulSoup(r, 'lxml')

        boxes = soup.find_all("div", "box_cont")
        links = []
        
        for i, box in enumerate(boxes):
            if self.team in box.text:
                box_header = soup.find_all("div", "box_head")[i]
                if '경기취소' in box_header.text:
                    summaries.append(['경기취소'])
                    continue
                boxscore_button = box.find('a', text='박스스코어')
                if boxscore_button:
                    link = boxscore_button['href']
                    links.append(f'https://statiz.sporki.com{link}')

        records = []
        for link in links:
            if link != '경기취소':
                r = requests.get(link).text
                soup = BeautifulSoup(r, 'lxml')

                tables = soup.find_all("div", "sh_box")
                tables = [table for table in tables if (self.team in table.find("div", "box_head").text) and ('수비' not in table.find("div", "box_head").text)]
                for table in tables:
                    record = f'<b>{self.date}</b>'
                    columns = [col.text.strip() for col in table.find_all('th')]
    
                    rows = table.find_all('tr')
                    data = []
                    for row in rows[1:]:
                        cols = row.find_all('td')
                        row_data = [col.text.strip() for col in cols]
                        data.append(row_data)

                    if '타순' in columns:
                        record += f"\n<pre>{pd.DataFrame(data, columns=columns)[['타순', '이름', 'Pos.', 'PA', 'AB', 'H', ]].to_markdown(tablefmt='grid', index=None)}</pre>"
                        record += f"\n<pre>{pd.DataFrame(data, columns=columns)[['타순', '이름', 'R', 'HR', 'RBI', 'BB']].to_markdown(tablefmt='grid', index=None)}</pre>"
                        record += '\nPA=타석, AB=타수, R=득점, H=안타, HR=홈런, RBI=타점, BB=볼넷'
                    else:
                        record += f"\n<pre>{pd.DataFrame(data, columns=columns)[['이름', 'IP', 'TBF', 'H', 'R', 'ER']].to_markdown(tablefmt='grid', index=None)}</pre>"
                        record += f"\n<pre>{pd.DataFrame(data, columns=columns)[['이름', 'BB', 'HBP', 'K', 'HR']].to_markdown(tablefmt='grid', index=None)}</pre>"
                        record += '\nIP=이닝, TBF=상대 타자 수, H=피안타, R=실점, ER=자책점, BB=볼넷, HBP=사구, K=삼진, HR=피홈런'

                    logs = table.find_all('div', 'log_div')
                    log = '\n'.join([log.text for log in logs])

                    records.append(record)
                    records.append(log)

            else:
                record = f'<b>{self.date}</b>'
                record += '경기취소'    
                records.append(record)
        
        return records

 

▼ 순위 데이터 수집 전체 코드

더보기
class KBOYearRecordBot:
    
    def __init__(self, year):
        self.year = self.convert_to_yyyy(year)
        
        
    def convert_to_yyyy(self, year):
        try:
            year = int(year)
        except:
            if year == '오늘':
                year = int(datetime.today().year)
            else:
                return "날짜 형식을 인식할 수 없습니다."
        if year < 100:
            if year >= 82:
                year += 1900
            else:
                year += 2000
        
        elif year > int(datetime.today().year):
            year = "날짜 형식을 인식할 수 없습니다."
        
        return year
    
    
    def year_category(self, year):
        if year < 1990:
            return 1980
        elif year < 2000:
            return 1990
        elif year < 2010:
            return 2000
        elif year < 2020:
            return 2010
        else:
            return 2020
    
    def team_records(self):
        y = self.year_category(self.year)
        if y == "날짜 형식을 인식할 수 없습니다.":
            return [y]
        
        url = f"https://www.koreabaseball.com/Record/History/Team/Record.aspx?startYear={y}&halfSc=T"
        r = requests.get(url).text
        soup = BeautifulSoup(r, 'lxml')
        if self.year >= 1990:
            table_idx = self.year % 10
        else:
            table_idx = self.year % 10 - 2
        trs = soup.find_all('table', 'tData')[table_idx].find_all('tr')
        
        records = [[]]
        names = []
        column_name = [th.text for th in trs[0].find_all('th')]
        
        for row in trs[1:-1]:
            columns = row.find_all('td')
            if len(columns) > 0:
                team_record = [
                    row.find('th').text.strip(),
                    int(columns[0].text),
                    int(columns[1].text),
                    int(columns[2].text),
                    int(columns[3].text),
                    float(columns[4].text),
                    float(columns[5].text),
                    float(columns[6].text)
                ]

                records[-1].append(team_record)
            else:
                records.append([])
                names.append(row.find('th').text)

        if len(records) == 1:
            df = pd.DataFrame(records[0], columns=column_name, index=range(1, len(records[0])+1))
            df.index.name = '순위'
            return [f"<b>{self.year}년도 정규리그 순위</b>\n<pre>{df.reset_index().loc[:, :'무'].to_markdown(index=None, tablefmt='grid')}</pre>",
                    f"<pre>{df.reset_index().loc[:, [str(self.year), '타율', '평균자책점', '승률']].to_markdown(index=None, tablefmt='grid')}</pre>"]
        else:
            df1 = pd.DataFrame(records[1], columns=column_name, index=range(1, len(records[1])+1))
            df2 = pd.DataFrame(records[2], columns=column_name, index=range(1, len(records[2])+1))
            df1.index.name = names[0].replace('<', '').replace('>', '').strip()
            df2.index.name = names[1].replace('<', '').replace('>', '').strip()
        
            return [f"<b>{self.year}년도 정규리그 순위</b>\n<pre>{df1.loc[:, :'무'].to_markdown(tablefmt='grid')}</pre>", 
                    f"<pre>{df1.loc[:, [str(self.year), '타율', '평균자책점', '승률']].to_markdown(index=None, tablefmt='grid')}</pre>",
                    f"<pre>{df2.loc[:, :'무'].to_markdown(tablefmt='grid')}</pre>",
                   f"<pre>{df2.loc[:, [str(self.year), '타율', '평균자책점', '승률']].to_markdown(index=None, tablefmt='grid')}</pre>"]

 

 

2. 텔레그램 봇 생성 및 API Gateway 연결

API Gateway는 이름에서 알 수 있듯 API를 만들 수 있도록 돕는 서비스이다. 봇과 대화하는 건 API Gateway를 이용하여 쉽게 구현할 수 있었다. (텔레그램이 워낙 쉽게 잘 만들어놔서 나같은 초보자도 충분히 구현 가능했다.)

먼저 이전 글을 참고하여 텔레그램 봇과 Lambda 함수를 생성하자. 

 

2-1. API Gateway 생성 및 Lambda 연결

API Gateway에서 API 생성을 하자. 여러 유형의 API가 있지만 가장 간단한 HTTP API를 만들자.

 

HTTP API 구축을 누르고 통합 추가를 하여 앞서 생성했던 Lambda 함수를 연결하자.

 

다음으로 넘어가 경로 구성에는 `ANY`를 `POST`로 바꿔주자. 리소스 경로는 본인 필요에 따라 변경해도 된다. 나중에 여기서 지정한 리소스 경로를 포함한 API를 호출할 것이다.

 

스테이지는 default로 둬도 되고, 나는 `v1`으로 작성하였다.

그러면 이렇게 API Gateway와 Lambda를 연결이 완료되었다.

 

스테이지에 적혀있는 URL 맨 뒤에 리소스 경로를 붙인 `https://{apigateway_id}.execute-api.ap-northeast-2.amazonaws.com/v1/{리소스 경로}`가 API 주소가 된다.

 

Lambda 함수 - 구성 - 트리거에서도 해당 API 주소를 확인할 수 있다.

 

 

 

2-2. API Gateway와 텔레그램 봇 연결하기

텔레그램 봇이 있는 방과 API Gateway를 연결하는 건 한 번의 API 호출이면 끝이다.

포스트맨 같은 데서 POST로 `https://api.telegram.org/bot{tokenid}/setwebhook`에 파라미터 url={api_gateway_url}을 추가하여 호출하자.

return값으로 `Webhook was set`이 뜨면 웹훅 설정이 완료되어 텔레그램에서 메시지를 작성하면 API Gateway를 타고 Lambda를 거치게 된다.

 

 

3. Lambda 함수 작성하기

Lambda 함수에 layer 추가하는 건 이 글을 참고하자. 나는 아래와 같이 추가를 했다.

 

봇 클래스는 `bot.py`로 따로 빼놓은 뒤 아래와 같이 작성했다.

더보기
import json
import os
import requests
from bot import KBOResultBot, KBOYearRecordBot


TOKEN = os.environ['TOKEN']
TELEGRAM_URL = "https://api.telegram.org/bot{}/sendMessage".format(TOKEN)
API_GATEWAY_URL = os.environ['API_GATEWAY_URL']

def lambda_handler(event, context):
    
    headers = {'Content-Type':'application/json; charset=utf-8'}
    drop_pending_updates = requests.post(f'https://api.telegram.org/bot{TOKEN}/setwebhook?url={API_GATEWAY_URL}&drop_pending_updates=true', headers)
    
    msg = '' # bot의 답변 내용
    
    json_body = json.loads(event['body'])
    if 'message' in json_body.keys(): # 개인방
        json_body = json_body['message']
    else: # 단체방
        json_body = json_body['channel_post']
    CHAT_ID = json_body['chat']['id']
    text = json_body['text']
    
    def send_message(text, chat_id=CHAT_ID, url=TELEGRAM_URL):
        payload = {
                    "text": text,
                    "chat_id": chat_id,
                    "parse_mode":"HTML"
                }
        response = json.loads(requests.post(url, payload).text)
    
    params = text.split(' ')
    command = params[0]
    print(text)
    
    try:
        if not command.startswith('/'):
            msg += "알 수 없는 명령어입니다."
        elif command == '/t' or command == '/today':
            results = KBOResultBot('오늘').summary()
            if len(results) < 1:
                msg += "경기 정보가 없습니다."
                send_message(msg)
            else:
                for result in results:
                    send_message(result)
                    
        
        else:
            if len(params) < 2:
                date = '오늘'
            else:
                date = ' '.join(params[1:])
    
            if command == '/s' or command == '/summary':
                results = KBOResultBot(date).summary()
                if len(results) < 1:
                    msg += "경기 정보가 없습니다."
                    send_message(msg)
                
                else:
                    for result in results:
                        send_message(result)
            
            elif command == '/b' or command == '/best':
                results = KBOResultBot(date).best_worst()
                if len(results) < 1:
                    msg += "경기 정보가 없습니다."
                    send_message(msg)
                
                else:
                    for result in results:
                        send_message(result)
                            
            elif command == '/c' or command == '/scene':
                results = KBOResultBot(date).score_scene()
                if len(results) < 1:
                    msg += "경기 정보가 없습니다."
                    send_message(msg)
                
                else:
                    for result in results:
                        send_message(result)
            
            elif command == '/p' or command == '/player':
                results = KBOResultBot(date).player_record()
                if len(results) < 1:
                    msg += "경기 정보가 없습니다."
                    send_message(msg)
                
                else:
                    for result in results:
                        send_message(result)
                        
            elif command == '/y' or command == '/year':
                results = KBOYearRecordBot(date).team_records()
                for result in results:
                    send_message(result)
                
    
    except Exception as e:
        send_message(f"오류가 발생했습니다.\n{e}")
    
    # TODO implement
    return {
        'statusCode': 200,
        'body': json.dumps('Hello from Lambda!')
    }

- 환경 변수에 토큰과 API Gateway 주소를 저장해두고 불러와서 사용했다.

- `drop_pending_updates`를 하는 이유를 참조한 글에서 다음과 같이 설명하고 있었다.

메세지 외 다른 유형의 웹훅을 받을 때, 해당 업데이트 지점에서 다음 update_id로 못 넘어가는 이슈가 있어서 마지막에 전달받은 웹훅 시점으로 update_id 초기화 후, 웹훅 재설정

실제로 저 부분 없이 진행했더니 채팅을 날리면 여러 번의 답장을 하게 되는 오류가 있었다.

- 이후로는 커맨드에 따라 어떤 함수를 실행할 지 예외처리를 해준 것이다.

 

4. 완성

순위 데이터

 

 

경기 결과 데이터

 

전체 코드: github

 

 

5. 참조

- AWS Lambda Telegram Bot 구축(2/3) - 람다 셋팅편 (velog.io)

 

 

 

728x90
반응형