기상청 초단기 API 활용 텔레그램 알림 봇

import requests
from datetime import datetime, timedelta
import json
import telegram
import asyncio

# 텔레그램 봇 정보
TOKEN = ""
CHAT_ID = ""

# 날씨 API 정보
API_KEY = '='
API_URL = 'http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst'
LOCATION_X = ''
LOCATION_Y = ''

# 날씨 코드
RAIN_CODE = {0 : '강수 없음', 1 : '☔', 2 : '☔/❄️', 3 : '❄️', 5 : '💦', 6 : '❄️', 7 : '💨❄️'}
#RAIN_CODE = {0 : '강수 없음', 1 : '비', 2 : '비/눈', 3 : '눈', 5 : '빗방울', 6 : '진눈깨비', 7 : '눈날림'}

SKY_CODE = {1 : '☀️', 3 : '⛅', 4 : '☁️'}
#sky_code = {1 : '맑음', 3 : '구름많음', 4 : '흐림'}
# 요일 정보
WEEKDAYS = ['월', '화', '수', '목', '금', '토', '일']

async def send_notice(forecast_text):
    """텔레그램으로 예보 전송"""
    chat = telegram.Bot(TOKEN)
    await chat.send_message(chat_id=CHAT_ID, text=forecast_text)

def get_weather_data():
    """날씨 데이터 가져오기"""
    current_time = datetime.now()
    three_hours_ago = current_time - timedelta(hours=1)

    # API 요청정보 설정
    params = {
        'serviceKey': API_KEY,
        'pageNo': '1',
        'numOfRows': '1000',
        'dataType': 'json',
        'base_date': current_time.strftime('%Y%m%d'),
        'base_time': three_hours_ago.strftime('%H%M'),
        'nx': LOCATION_X,
        'ny': LOCATION_Y,
    }

    # API 요청 및 응답
    response = requests.get(API_URL, params=params)
    weather_data = json.loads(response.text)

    return weather_data, params['base_date']

def parse_weather_data(weather_data):
    """날씨 데이터 파싱"""
    forecast_dict = {}

    # API 응답에서 날씨 예보 정보 추출
    for item in weather_data['response']['body']['items']['item']:
        forecast_time = item['fcstTime']
        weather_category = item['category']
        forecast_value = item['fcstValue']

        # 예보 정보 초기화
        if forecast_time not in forecast_dict:
            forecast_dict[forecast_time] = {}

        # 날씨 예보 정보 저장
        forecast_dict[forecast_time][weather_category] = forecast_value

    return forecast_dict

def create_forecast_text(forecast_dict, base_date):
    """날씨 예보 정보 문자열 변환"""
    result_str = []

    for time, forecast in forecast_dict.items():
        forecast_date = datetime.strptime(base_date, '%Y%m%d')
        day = forecast_date.day
        month = forecast_date.month
        weekday = WEEKDAYS[forecast_date.weekday()]

        forecast_text = f"[{month}/{day}({weekday}) {time[:2]}시] "

        # 하늘 상태
        if 'SKY' in forecast:
            forecast_text += SKY_CODE[int(forecast['SKY'])]

        # 기온
        if 'T1H' in forecast:
            forecast_text += f"{int(forecast['T1H']):3d}℃"
        # 강수 형태
        if 'PTY' in forecast:

            # 강수량(1시간당)
            if 'RN1' in forecast and forecast['RN1'] != '강수없음':
                forecast_text += f"\n"
                forecast_text += RAIN_CODE[int(forecast['PTY'])]
                forecast_text += f"  🌧️시간당 강수량 {forecast['RN1']}mm"


        result_str.append(forecast_text)

    return "\n".join(result_str)

def main():
    """메인 함수"""
    weather_data, base_date = get_weather_data()
    forecast_dict = parse_weather_data(weather_data)
    result_text = create_forecast_text(forecast_dict, base_date)

    print(result_text)
    asyncio.run(send_notice(result_text))

if __name__ == "__main__":
    main()