AWS Lambda만을 이용해서 날씨알림봇 만들기

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

AWS Lambda와 selenium을 이용해서 날씨알림봇 만들기

 

 

이전에 Lambda와 Selenium을 이용하여 네이버 날씨에서 여러 제공사의 날씨를 수집하는 봇을 만들었다. 근데 언제 생겼는지, 네이버에서 예보 비교가 가능했다. 생각보다 꽤 된 것 같은데 모르고 있었다.

예보 비교 기능 : 날씨 고객센터 (naver.com)

 

그래서 해당 페이지를 활용하여 selenium 없이 날씨 데이터를 수집해보기로 했다.

 

 

1. 데이터 수집

import requests
from bs4 import BeautifulSoup

url = f'https://weather.naver.com/compare/{지역코드}'
r = requests.get(url).text
soup = BeautifulSoup(r, 'lxml')

url의 지역코드는 네이버 날씨에 직접 들어가서, 내 위치 정보를 통해 확인하도록 하자.

 

`soup`을 출력해보면, 테이블 태그가 없어 날씨 비교 페이지도 반응형인 것을 알 수 있다. 그러면 결국 이전과 똑같이 [selenium을 써야 하느냐?] 할 수 있지만 답은 [그렇지 않다]이다. 출력 결과를 아래로 쭉 내려보면 js 스크립트 부분에 json 형태로 데이터가 박혀있다.

 

이런 느낌으로...

<script charset="utf-8" type="text/javascript">
    async function DecompressBlob(blob) {
        const ds = new DecompressionStream("gzip");
        const decompressedStream = blob.stream().pipeThrough(ds);
        return await new Response(decompressedStream).blob();
    }
    var blockApiResult = {"success":true,"message":"OK","resultCode":200,"version":"v1","results":{"1":"xxx", "2":"yyy", "3":{"4":"zzz"}}}
</script>

 

우린 `var blockApiResult =` 이후 `{~~}`를 추출해야한다.

 

import re
import json

s = [s.string for s in soup.find_all('script')][-1]
pattern = r'var blockApiResult = ({.*})'
match = re.search(pattern, s)
if match:
    json_str = match.group(1)
    data = json.loads(json_str)['results']['choiceResult']['compareHourlyFcast~~1']['domesticHourlyListMap']

스크립트가 여럿 있는데 그 중 가장 마지막 스크립트에 날씨 데이터가 박혀있고, 거기서도 여러 단계를 타고 타고 가면 데이터를 찾을 수 있게 된다. 이 데이터는 각 제공사의 단축명이 키로 되어있는 딕셔너리 구조이다.

 

 

(참고) 주간 날씨는 아래처럼 불러올 수 있다.

data = json.loads(json_str)['results']['choiceResult']['compareWeeklyFcast~~1']['domesticWeeklyListMap']

 

 

 

이제 제공사별로 필요한 데이터를 추출하면 된다.

import pandas as pd

df = pd.DataFrame()
forecast_kor = {'KMA':'기상청', 'ACCUWEATHER':'아큐웨더', 'TWC':'웨더채널', 'WEATHERNEWS':'웨더뉴스'}
for key in forecast_kor.keys():
    df_one = pd.DataFrame(data[key])[[
             'aplYmd', 'aplTm', 'tmpr', 'rainProb', 'rainAmt'
         ]]
    df_one.columns=['날짜', '시간', '기온', '강수확률', '강수량']
    df_one = df_one[(df_one['날짜']==today) & (df_one['시간'].astype(int) > hour)].drop(columns='날짜')
    df_one = df_one.set_index('시간').T
    df_one.index = [[forecast_kor[key] for _ in range(3)], ['기온', '강수확률', '강수량']]
    
    df = pd.concat([df, df_one], axis=0)

내가 필요로 한 정보는 기온, 강수확률, 강수량이기 때문에 그에 해당하는 컬럼만 따로 추출해서 사용했다.

 

 

2. 데이터 정제

이렇게 데이터를 뽑아내면 소수점 때문에 지저분해보인다.

그래서 기온과 강수확률은 int형으로, 강수량은 0이 아닐 때만 소수점 1자리까지만 출력하도록 수정해주었다.

 

def to_int(x):
    try:
        return int(round(float(x), 0))
    except:
        return x

def to_float(x):
    try:
        x = round(float(x), 1)
        if x == 0:
            return 0
        else:
            return x
    except:
        return x
        
        
for col in df.columns:
    df.loc[df.index.get_level_values(1)!='강수량', col] = df.loc[df.index.get_level_values(1)!='강수량', col].apply(to_int)
    df.loc[df.index.get_level_values(1)=='강수량', col] = df.loc[df.index.get_level_values(1)=='강수량', col].apply(to_float)

 

 

이외에도 평균, 최소, 최대값을 추가해주고 자잘한 설정을 바꿔주면...

minimum = df.replace('~1', '0.5').replace('-', None).astype(float).min(axis=1)
maximum = df.replace('~1', '0.5').replace('-', None).astype(float).max(axis=1)
average = df.replace('~1', '0.5').replace('-', None).astype(float).mean(axis=1).round(1)

df['최저'] = minimum
df['최고'] = maximum
df['평균'] = average

df.index.names = [today[:4] + '년', str(int(today[4:6])) + '월 ' + today[6:8] + '일']

print(df.reset_index().to_markdown(index=None, tablefmt="grid"))

이런 표를 완성할 수 있게 된다.

 

 

이후에는 이전글을 참고하여 텔레그램 메시지를 보내면 끝이다.

 

 

3. 전체 코드 (lambda)

import json
import re
import os
import requests
import pandas as pd
from bs4 import BeautifulSoup
from datetime import datetime, timedelta

today = datetime.now() + timedelta(hours=9)
hour = today.hour
today = today.strftime('%Y%m%d')


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


def to_int(x):
    try:
        return int(round(float(x), 0))
    except:
        return x

def to_float(x):
    try:
        x = round(float(x), 1)
        if x == 0:
            return 0
        else:
            return x
    except:
        return x


def lambda_handler(event, context):
    
    url = f'https://weather.naver.com/compare/{code}'
    r = requests.get(url).text
    soup = BeautifulSoup(r, 'lxml')
    s = soup.find_all('script')[-1].string
    
    pattern = r'var blockApiResult = ({.*})'
    match = re.search(pattern, s)
    if match:
        json_str = match.group(1)
        data = json.loads(json_str)['results']['choiceResult']['compareHourlyFcast~~1']['domesticHourlyListMap']
        
        df = pd.DataFrame()
        forecast_kor = {'KMA':'기상청', 'ACCUWEATHER':'아큐웨더', 'TWC':'웨더채널', 'WEATHERNEWS':'웨더뉴스'}
        for key in forecast_kor.keys():
            df_one = pd.DataFrame(data[key])[[
                     'aplYmd', 'aplTm', 'tmpr', 'rainProb', 'rainAmt'
                 ]]
            df_one.columns=['날짜', '시간', '기온', '강수확률', '강수량']
            df_one = df_one[(df_one['날짜']==today) & (df_one['시간'].astype(int) > hour)].drop(columns='날짜')
            df_one = df_one.set_index('시간').T
            df_one.index = [[forecast_kor[key] for _ in range(3)], ['기온', '강수확률', '강수량']]
            
            df = pd.concat([df, df_one], axis=0)
            
        df = df.dropna(axis=1)
        minimum = df.replace('~1', '0.5').replace('-', None).astype(float).min(axis=1)
        maximum = df.replace('~1', '0.5').replace('-', None).astype(float).max(axis=1)
        average = df.replace('~1', '0.5').replace('-', None).astype(float).mean(axis=1).round(1)
        df['최저'] = minimum
        df['최고'] = maximum
        df['평균'] = average
        
        df.index.names = [today[:4] + '년', str(int(today[4:6])) + '월 ' + today[6:8] + '일']
        for col in df.columns:
            df.loc[df.index.get_level_values(1)!='강수량', col] = df.loc[df.index.get_level_values(1)!='강수량', col].apply(to_int)
            df.loc[df.index.get_level_values(1)=='강수량', col] = df.loc[df.index.get_level_values(1)=='강수량', col].apply(to_float)
        
        for i in range(0, len(df.columns), 3):
            tmp = df.iloc[:, i:i+3].reset_index()
            
            if i == 0:
                msg = '<b>날씨 알림</b>'
            else:
                msg = ''
            msg += f'<pre>{tmp.to_markdown(index=None, tablefmt="grid")}</pre>'
        
            try:
                payload = {
                    "text": msg,
                    "chat_id": CHAT_ID,
                    "parse_mode":"HTML"
                }
                response = json.loads(requests.post(TELEGRAM_URL, payload).text)
                if not response['ok']:
                    raise Exception(response['description'])
            except Exception as e:
                raise e
            
        return {"status":"OK"}
    
    else:
        raise Exception('No matched data')