97 lines
3.2 KiB
Python
97 lines
3.2 KiB
Python
import os
|
|
import requests
|
|
import csv
|
|
from datetime import datetime, timedelta
|
|
|
|
# 1. API 인증키 설정
|
|
def get_env_variable(var_name):
|
|
if os.path.exists(".env"):
|
|
with open(".env", "r") as f:
|
|
for line in f:
|
|
if "=" in line:
|
|
k, v = line.split("=", 1)
|
|
if k.strip() == var_name:
|
|
return v.strip()
|
|
return os.getenv(var_name)
|
|
|
|
API_KEY = get_env_variable("WEATHER_KEY") or get_env_variable("KMA_API_KEY")
|
|
|
|
# 2. 좌표 설정
|
|
TARGET_LAT = 34.92470
|
|
TARGET_LON = 127.71207
|
|
|
|
# 3. 시점 로드
|
|
def load_target_times(file_path):
|
|
times = []
|
|
if not os.path.exists(file_path): return []
|
|
with open(file_path, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line: continue
|
|
try:
|
|
dt = datetime.strptime(line, "%Y-%m-%d: %H:%M")
|
|
times.append(dt)
|
|
except ValueError: continue
|
|
return times
|
|
|
|
target_times = load_target_times("visible-times.txt")
|
|
|
|
# 4. API 설정
|
|
BASE_URL = "https://apihub.kma.go.kr/api/typ01/url/sfc_nc_var.php"
|
|
OBS_ELEMENTS = "ta,hm,td,wd_10m,ws_10m,uu,vv,rn_ox,vs"
|
|
INTERVAL = 5
|
|
|
|
def fetch_data(t_time):
|
|
print(f"--- [{t_time.strftime('%Y-%m-%d %H:%M')}] 추출 중 ---")
|
|
start_time = t_time - timedelta(hours=1)
|
|
end_time = t_time + timedelta(hours=1)
|
|
|
|
queries = [(start_time, t_time), (t_time + timedelta(minutes=INTERVAL), end_time)]
|
|
extracted_data = []
|
|
|
|
for tm1, tm2 in queries:
|
|
params = {
|
|
"tm1": tm1.strftime("%Y%m%d%H%M"),
|
|
"tm2": tm2.strftime("%Y%m%d%H%M"),
|
|
"obs": OBS_ELEMENTS,
|
|
"itv": INTERVAL,
|
|
"lon": TARGET_LON,
|
|
"lat": TARGET_LAT,
|
|
"help": 0,
|
|
"authKey": API_KEY
|
|
}
|
|
try:
|
|
response = requests.get(BASE_URL, params=params, timeout=30)
|
|
response.raise_for_status()
|
|
|
|
for line in response.text.strip().split('\n'):
|
|
if line.startswith('#') or not line.strip(): continue
|
|
parts = [p.strip() for p in line.replace(',', ' ').split()]
|
|
if len(parts) >= 10:
|
|
extracted_data.append({
|
|
"tm": parts[0],
|
|
"ta": parts[1],
|
|
"hm": parts[2],
|
|
"td": parts[3],
|
|
"wd_10m": parts[4],
|
|
"ws_10m": parts[5],
|
|
"uu": parts[6],
|
|
"vv": parts[7],
|
|
"rn_ox": parts[8],
|
|
"vs": parts[9]
|
|
})
|
|
except Exception as e: print(f" Error: {e}")
|
|
|
|
if extracted_data:
|
|
extracted_data.sort(key=lambda x: x['tm'])
|
|
filename = f"weather_data_{t_time.strftime('%Y%m%d_%H%M')}.csv"
|
|
with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
|
|
writer = csv.DictWriter(f, fieldnames=extracted_data[0].keys())
|
|
writer.writeheader()
|
|
writer.writerows(extracted_data)
|
|
print(f" >> 저장됨: {filename} ({len(extracted_data)}개)")
|
|
else:
|
|
print(f" >> 데이터 없음")
|
|
|
|
if __name__ == "__main__":
|
|
for t in target_times: fetch_data(t)
|