[SK shieldus Rookies 29기] 9일차

• 오늘은 여태까지 배웠던 기술 스택을 활용하는 케이스 스터디를 진행하는 방식으로 수업을 진행하였다.
• 팀원들과 상의를 한 끝에 우리 조의 주제는 slack으로 4가지를 전송받기로 하였다. 여기서 4가지는 Google Calendar 일정, 날씨, 보안 뉴스, Github Push 알람 이벤트이다.
• 이 중에서 내가 맡은 파트는 날씨였다.
• 날씨에서 디벨롭된 방향은 기상청 API 기반 자동 날씨 수집·MongoDB캐싱·Slack 알림 시스템으로 결정하였다.

전체 흐름 구조

1. 기상청 API에서 데이터를 가져옴
2. MongoDB를 통해 캐시를 확인하거나 저장
3. Slack으로 알림을 전송

   weather_slack/
   ├── slack_weather.py          
   ├── models.py       
   ├── config.py       
   ├── requirements.txt 
   ├── .env  
   ├── .venv
   ├── __pycache__
   
   
   

먼저 코드를 제시하겠다.

slack_weather.py

import requests
from datetime import datetime, timedelta
import config  # .env 로드
import models  # MongoDB 캐싱

CITY_COORDS = {'seoul': {'nx': 60, 'ny': 127}}  # 가이드 2. 참고자료 - 단기예보 지점 좌표(X,Y) 참조: 서울 nx=60, ny=127 엑셀 값.

def get_latest_base_time():
    now = datetime.now()
    base_times = ['0200', '0500', '0800', '1100', '1400', '1700', '2000', '2300']  # 가이드 2. 참고자료 - 예보 발표시각 ❍ 단기예보 발표시각 참조: 1일 8회 시간대.
    current_hour = now.hour
    for bt in reversed(base_times):
        bt_hour = int(bt[:2])
        if current_hour >= bt_hour + (1 if int(bt[2:]) > 0 else 0):
            return bt
    return '2300'

def fetch_current_weather(city):
    coords = CITY_COORDS.get(city.lower())
    if not coords:
        return None
    base_date = datetime.now().strftime('%Y%m%d')  # 가이드 1.1 다. 상세기능내역 1) [초단기실황조회] b) 요청 메시지 명세 - base_date 필수 참조: YYYYMMDD 형식.
    base_time = (datetime.now() - timedelta(minutes=10)).strftime('%H00')  # 가이드 2. 참고자료 ❍ 초단기실황 발표시각 참조: 매시 10분 후 업데이트.
    url = 'https://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtNcst'  # 가이드 1.1 가. API 서비스 개요 - 서비스 URL 참조: VilageFcstInfoService_2.0.
    params = {
        'serviceKey': config.SERVICE_KEY,  # 가이드 1.1 다. 상세기능내역 1) b) 요청 메시지 명세 - serviceKey 필수 참조: URL Encode된 인증키.
        'numOfRows': '100',  # 가이드 1.1 다. 상세기능내역 1) b) numOfRows 참조: 한 페이지 결과 수 (Default:10).
        'dataType': 'JSON',  # 가이드 1.1 다. 상세기능내역 1) b) dataType 참조: JSON/XML (Default:XML).
        'base_date': base_date,
        'base_time': base_time,  # 가이드 1.1 다. 상세기능내역 1) b) base_time 참조: HH00 형식, 매시 10분 후 호출.
        'nx': coords['nx'],  # 가이드 1.1 다. 상세기능내역 1) b) nx 참조: 예보지점 X 좌표 (엑셀 참조).
        'ny': coords['ny']   # 가이드 1.1 다. 상세기능내역 1) b) ny 참조: 예보지점 Y 좌표.
    }
    response = requests.get(url, params=params)
    if response.status_code != 200:
        print('초단기실황 API 실패:', response.text)  # 가이드 2. 참고자료 ※ Open API 에러 코드 정리 참조: 에러 코드(00=정상, 30=키 오류 등) 확인.
        return None
    items = response.json()['response']['body']['items']['item']  # 가이드 1.1 다. 상세기능내역 1) d) 응답 메시지 예제 참조: items 배열.
    current = {'current_temp': None}
    for item in items:
        if item['category'] == 'T1H':  # 가이드 1.1 다. 상세기능내역 1) c) 응답 메시지 명세 - category 참조: T1H=기온 실황 값.
            current['current_temp'] = item['obsrValue']  # 가이드 1.1 다. 상세기능내역 1) c) obsrValue 참조: 실황 값 (실수/정수).
    return current

def fetch_weather(city):
    coords = CITY_COORDS.get(city.lower())
    if not coords:
        return None
    base_date = datetime.now().strftime('%Y%m%d')  # 가이드 1.1 다. 상세기능내역 3) [단기예보조회] b) 요청 메시지 명세 - base_date 필수 참조.
    base_time = get_latest_base_time()  # 가이드 1.1 다. 상세기능내역 3) b) base_time 참조: 0500 등 8회 시간대.
    url = 'https://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getVilageFcst'  # 가이드 1.1 다. 상세기능내역 3) a) Call Back URL 참조.
    params = {
        'serviceKey': config.SERVICE_KEY,
        'numOfRows': '1000',  # 가이드 1.1 다. 상세기능내역 3) b) numOfRows 참조: 충분히 가져옴 (Default:10).
        'dataType': 'JSON',
        'base_date': base_date,
        'base_time': base_time,
        'nx': coords['nx'],
        'ny': coords['ny']
    }
    response = requests.get(url, params=params)
    if response.status_code != 200:
        print('단기예보 API 실패:', response.text)  # 가이드 2. 참고자료 에러 코드 참조.
        return None
    items = response.json()['response']['body']['items']['item']
    weather_data = {'max_temp': None, 'min_temp': None, 'precip_prob': None, 'precip_amt': None}
    for item in items:
        if item['fcstDate'] == base_date:  # 가이드 1.1 다. 상세기능내역 3) c) fcstDate 참조: 오늘 예보 필터.
            if item['category'] == 'TMX':  # 가이드 1.1 다. 상세기능내역 3) c) category 참조: TMX=최고온도.
                weather_data['max_temp'] = item['fcstValue']  # 가이드 1.1 다. 상세기능내역 3) c) fcstValue 참조: 실수/정수 값.
            elif item['category'] == 'TMN':  # TMN=최저온도.
                weather_data['min_temp'] = item['fcstValue']
            elif item['category'] == 'POP':  # POP=강수확률.
                if weather_data['precip_prob'] is None:
                    weather_data['precip_prob'] = item['fcstValue']
            elif item['category'] == 'PCP':  # PCP=1시간 강수량, 가이드 2. 참고자료 - 강수량(PCP) 범주 참조: mm 단위 표시.
                if weather_data['precip_amt'] is None:
                    weather_data['precip_amt'] = item['fcstValue']
    return weather_data

def send_to_slack(city, data):
    webhook_url = config.SLACK_WEBHOOK_URL
    message = {
        'text': f"{city} 날씨: 현재 {data.get('current_temp', 'N/A')}도, 최고 {data.get('max_temp', 'N/A')}도, 최저 {data.get('min_temp', 'N/A')}도, 강수확률 {data.get('precip_prob', 'N/A')}%, 강수량 {data.get('precip_amt', 'N/A')}"
    }
    response = requests.post(webhook_url, json=message)
    if response.status_code == 200:
        print('Slack 메시지 전송 성공')
    else:
        print('Slack 실패:', response.text)

def main():
    city = 'seoul'
    cached = models.get_cached_data(city)
    if cached:
        send_to_slack(city, cached)
    else:
        current = fetch_current_weather(city) or {'current_temp': 'N/A'}
        weather = fetch_weather(city) or {'max_temp': 'N/A', 'min_temp': 'N/A', 'precip_prob': 'N/A', 'precip_amt': 'N/A'}
        data = {
            'city': city,
            'current_temp': current['current_temp'],
            'max_temp': weather['max_temp'],
            'min_temp': weather['min_temp'],
            'precip_prob': weather['precip_prob'],
            'precip_amt': weather['precip_amt']
        }
        models.save_data(city, data)
        send_to_slack(city, data)

if __name__ == '__main__':
    main()

models.py

from pymongo import MongoClient
from datetime import datetime, timedelta
import config  # config.py 임포트

client = MongoClient('mongodb://localhost:27017')
db = client['weather_db']
collection = db['weather_data']

#검증
try:
    client = MongoClient('mongodb://localhost:27017', serverSelectionTimeoutMS=3000)
    # 서버에 연결 가능한지 강제 확인
    client.server_info()
    db = client['weather_db']
    collection = db['weather_data']
    print("MongoDB: 연결 성공", 'mongodb://localhost:27017')
except Exception as e:
    print("MongoDB 연결 실패:", e)
    collection = None

def get_cached_data(city):
    # 1시간 내 캐싱된 데이터 찾기
    one_hour_ago = datetime.now() - timedelta(hours=1)
    data = collection.find_one({'city': city, 'timestamp': {'$gt': one_hour_ago}})
    return data

def save_data(city, data):
    # 데이터 저장 (timestamp 추가)
    data['city'] = city
    data['timestamp'] = datetime.now()
    collection.insert_one(data)

config.py

import os
from dotenv import load_dotenv

load_dotenv()

SERVICE_KEY = os.getenv('SERVICE_KEY')  # 가이드 1.1 가. API 서비스 개요 - serviceKey 필수 파라미터 참조: 공공데이터포털 발급 키 사용.
SLACK_WEBHOOK_URL = os.getenv('SLACK_WEBHOOK_URL')  # Slack Incoming Webhook URL

공공데이터 포털 사이트 참조

# Python3 샘플 코드 #


import requests

url = 'http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtNcst'
params ={'serviceKey' : '서비스키', 'pageNo' : '1', 'numOfRows' : '1000', 'dataType' : 'XML', 'base_date' : '20210628', 'base_time' : '0600', 'nx' : '55', 'ny' : '127' }

response = requests.get(url, params=params)
print(response.content)

.env example

SERVICE_KEY=your_service_key_here  # 실제 기상청 키 입력
SLACK_WEBHOOK_URL = os.getenv('SLACK_WEBHOOK_URL')

서울 격자 위경도(기상청-단기예보 조회서비스-오픈API활용가이드)

구분	행정구역코드	1단계	2단계	3단계	격자 X	격자 Y	경도(시)	경도(분)	경도(초)	위도(시)	위도(분)	위도(초)	경도(초/100)	위도(초/100)	위치업데이트	None
kor	1100000000	서울특별시			60	127	126	58	48.03	37	33	48.85	126.98	37.5636

기상청 단기 예보 가이드

문제점

seoul 날씨: 현재 5.1도, 최고 None도, 최저 None도, 강수확률 0%, 강수량 강수없음

• 왜 None값으로 반환이 되는가?
• 열심히 구글링을 돌린 결과 아래와 같은 결론이 나왔다.

  TMX는 오늘 날짜로 잘 찾았으나...

  TMN은 내일 fcstDate에 있어 필터링되어 버림.

  TMX (최고 기온): 보통 오늘의 예상 최고 기온으로, 이미 지나간 시간대의 예보를 기반으로 확정되는 경우가 많습니다.

  TMN (최저 기온): 보통 다음 날 새벽의 최저 기온을 의미하며, 이 데이터가 오늘 발표 예보에 포함될 때 fcstDate가 내일 날짜로 표기될 수 있습니다.

따라서
기상청 단기예보 특성상 오늘 최저기온은 대부분 “내일 새벽 기온”이라 fcstDate가 내일 날짜로 되어 있음 → 코드는 fcstDate == 오늘만 필터링하니까 TMN이 걸러져서 None 됨

이건 제 코드의 논리적 오류라기보다는 기상청 데이터 구조를 제가 아직 완벽히 반영하지 못한 부분입니다.

라고 결론을 냈습니다.

정리

① get_latest_base_time() 함수 로직 오류 - 현재 로직으로는 거의 항상 잘못된 base_time을 리턴함 - 예: 오전 9시 실행 → 0800을 써야 하는데 0200이나 2300을 리턴 - 결과적으로 오래된 예보를 가져오게 됨

② 초단기실황 base_time 계산 오류 - 지금 -10분만 빼면 최신 관측값을 못 가져올 때가 많음 - 실제로는 최소 40~50분 전 정시를 써야 안정적

③ 최고/최저 기온 필터링 문제 (★★★★ 가장 중요한 문제 ★★★★)

• 코드: if item['fcstDate'] == 오늘 날짜:
• TMX (오늘 최고기온): 대부분 오늘 fcstDate로 잘 나옴 → 정상 조회
  • TMN (오늘 최저기온): 기상청 특성상 다음 날 새벽(내일) 최저기온을 의미하는 경우가 많음 → TMN의 fcstDate가 내일 날짜로 되어 있음 → 오늘 날짜로만 필터링하면 TMN이 걸러져서 항상 None 또는 N/A 됨
• 결과: 최고기온은 나오는데 최저기온은 항상 안 나오는 현상 발생

④ 강수확률/강수량도 첫 번째 값만 저장 - 하루 중 여러 시간대 값이 나오는데, 첫 번째 값만 저장 → 의미 없는 값일 수 있음 - 보통은 하루 중 최고 강수확률(max POP)을 보여주는 게 좋음

⑤ 새벽 00:00~02:10 사이에는 전날 2300 예보를 써야 하는데 base_date 처리 안 됨 버그 발생 이유: 기상청 데이터 특성 때문에 호출을 하는 시점에는 N/A로 나오는 것이다다다!!!

---

고심 끝에 이렇게 결론을 내렸다.

디버깅 쉽지 않다....