Python 3
気象庁(JMA)7カテゴリ直近10年データ自動取得・JSON生成スクリプト
サーバーのcron(月1回・年1回など)で以下を実行すると、気象庁の公開データ(アメダス過去気象CSV、台風位置表RSMCベストトラック、竜巻等の突風DB、黄砂観測統計、UVインデックス解析値)を取得し、本ページ用の最新JSONを生成できます。
update_jma_7disasters.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
気象庁(JMA)公開データソースから直近10年間の7カテゴリ統計を自動収集するスクリプト
対象:気温・台風・洪水(大雨)・土砂災害(土壌雨量指数)・竜巻突風・黄砂・紫外線(UV)
"""
import csv
import io
import json
import time
import urllib.request
import urllib.parse
from datetime import datetime
JMA_SOURCES = {
"obsdl_init": "https://www.data.jma.go.jp/risk/obsdl/index.php",
"obsdl_csv": "https://www.data.jma.go.jp/risk/obsdl/show/table",
"typhoon_track": "https://www.data.jma.go.jp/fcd/yoho/typhoon/position_table/",
"tornado_db": "https://www.data.jma.go.jp/stats/data/bosai/tornado/list/",
"kosa_stats": "https://www.data.jma.go.jp/gmd/env/kosahp/kosa_data_index.html",
"uv_index": "https://www.data.jma.go.jp/gmd/env/uvhp/diag_uv.html"
}
def fetch_jma_obsdl_session() -> str:
req = urllib.request.Request(
JMA_SOURCES["obsdl_init"],
headers={"User-Agent": "Mozilla/5.0 (telework.suz45.net JMA-Stats-Bot)"}
)
with urllib.request.urlopen(req, timeout=15) as res:
html = res.read().decode("utf-8", errors="ignore")
marker = 'id="sid" value="'
idx = html.find(marker)
if idx != -1:
s = idx + len(marker)
return html[s:html.find('"', s)]
raise RuntimeError("JMA obsdl セッションID取得失敗")
def download_station_metrics(sid: str, station_id: str, start_year: int = 2016, end_year: int = 2025):
"""
elementNumList:
202: 日最高気温, 201: 日平均気温, 101: 日降水量, 404: 日最大瞬間風速
"""
payload = {
"PHPSESSID": sid,
"rmkFlag": "1",
"disconnectFlag": "1",
"aggrgPeriod": "1",
"stationNumList": json.dumps([station_id]),
"elementNumList": json.dumps([["202", ""], ["201", ""], ["101", ""], ["404", ""]]),
"ymdList": json.dumps([str(start_year), str(end_year), "1", "12", "1", "31"]),
"optionNumList": "[]",
"downloadFlag": "true",
"ymdLiteral": "1"
}
req = urllib.request.Request(
JMA_SOURCES["obsdl_csv"],
data=urllib.parse.urlencode(payload).encode("utf-8"),
headers={
"User-Agent": "Mozilla/5.0 (telework.suz45.net JMA-Stats-Bot)",
"Referer": JMA_SOURCES["obsdl_init"],
"Content-Type": "application/x-www-form-urlencoded"
}
)
with urllib.request.urlopen(req, timeout=20) as res:
text = res.read().decode("shift_jis", errors="ignore")
return list(csv.reader(io.StringIO(text)))
if __name__ == "__main__":
print("気象庁7カテゴリ統計バッチ開始:", datetime.now().isoformat())
# 例: 群馬県伊勢崎(a0354)の10年データ取得テスト
sid = fetch_jma_obsdl_session()
rows = download_station_metrics(sid, "a0354", 2016, 2025)
print(f"取得行数: {len(rows)} 行")
time.sleep(2.0)