昨天我們把 HTTP 的基礎和一個帶重試的請求函式準備好了。今天要拿它去做真的事:查台灣的天氣。
為什麼選天氣當第一個 API?
到 中央氣象署開放資料平臺:https://opendata.cwa.gov.tw
CWA-XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX 格式的授權碼拿到之後,照 Day 12 的做法放進 .env:
CWA_API_KEY=CWA-XXXXXXXX-XXXX-XXXX-XXXX-XXXXXXXXXXXX
不要寫進程式碼裡。
昨天說過,寫程式之前先看清楚資料結構。我們要用的是 F-C0032-001:一般天氣預報-今明 36 小時天氣預報。
在瀏覽器打開(把 YOUR_KEY 換成你的授權碼):
https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-C0032-001?Authorization=YOUR_KEY&locationName=臺北市
回傳的 JSON 結構大概長這樣(我把不重要的部分省略了):
{
"success": "true",
"records": {
"location": [
{
"locationName": "臺北市",
"weatherElement": [
{
"elementName": "Wx",
"time": [
{
"startTime": "2026-09-28 12:00:00",
"endTime": "2026-09-28 18:00:00",
"parameter": {"parameterName": "多雲", "parameterValue": "4"}
}
]
},
{"elementName": "PoP", "time": [...]},
{"elementName": "MinT", "time": [...]},
{"elementName": "MaxT", "time": [...]},
{"elementName": "CI", "time": [...]}
]
}
]
}
}
五個天氣要素的意思:
| elementName | 中文 | 單位 |
|---|---|---|
Wx |
天氣現象 | 文字(晴、多雲、陰時多雲短暫陣雨…) |
PoP |
降雨機率 | % |
MinT |
最低溫度 | °C |
MaxT |
最高溫度 | °C |
CI |
舒適度 | 文字(舒適、悶熱…) |
而 time 陣列有三個時段,各 12 小時,涵蓋未來 36 小時。
這個結構有點難用——資料是「以要素分組」,但我們想要的是「以時段分組」。這就是今天要處理的主要工作。
⚠️ 一個很容易卡住的地方:城市名稱必須用正體的「臺」,不是「台」。
臺北市可以,台北市會查不到。而且要包含「市」或「縣」。
建立 tools/weather.py:
"""tools/weather.py — 中央氣象署天氣查詢。"""
import requests
CWA_BASE = "https://opendata.cwa.gov.tw/api/v1/rest/datastore"
FORECAST_36H = "F-C0032-001"
def fetch_raw_forecast(city, api_key):
"""向氣象署取得原始的 36 小時預報資料。"""
response = requests.get(
f"{CWA_BASE}/{FORECAST_36H}",
params={
"Authorization": api_key,
"locationName": city,
"format": "JSON",
},
timeout=10,
)
response.raise_for_status()
return response.json()
# 試跑
import config
data = fetch_raw_forecast("臺北市", config.CWA_API_KEY)
print(data["records"]["location"][0]["locationName"])
能跑就代表串通了。接下來是把資料整理成好用的形狀。
我們想要的結果是這樣:
{
"city": "臺北市",
"periods": [
{
"start": "2026-09-28 12:00:00",
"end": "2026-09-28 18:00:00",
"weather": "多雲",
"rain_probability": 20,
"min_temp": 26,
"max_temp": 31,
"comfort": "悶熱",
},
...
],
}
轉換的程式碼:
# 氣象署的欄位名 → 我們自己的欄位名
ELEMENT_MAP = {
"Wx": "weather",
"PoP": "rain_probability",
"MinT": "min_temp",
"MaxT": "max_temp",
"CI": "comfort",
}
NUMERIC_FIELDS = {"rain_probability", "min_temp", "max_temp"}
def parse_forecast(raw):
"""把氣象署的巢狀 JSON 攤平成好用的結構。"""
locations = raw.get("records", {}).get("location", [])
if not locations:
raise ValueError("氣象署沒有回傳任何地點資料,請確認城市名稱(要用「臺」)")
location = locations[0]
# 用時段的 startTime 當 key,把各要素的值收集起來
periods = {}
for element in location.get("weatherElement", []):
field = ELEMENT_MAP.get(element["elementName"])
if field is None:
continue # 不認識的要素就跳過
for slot in element.get("time", []):
key = slot["startTime"]
period = periods.setdefault(key, {
"start": slot["startTime"],
"end": slot["endTime"],
})
value = slot["parameter"]["parameterName"]
if field in NUMERIC_FIELDS:
try:
value = int(value)
except ValueError:
value = None # 氣象署偶爾會給空字串
period[field] = value
return {
"city": location["locationName"],
# 依開始時間排序(ISO 風格的字串排序剛好等於時間排序)
"periods": [periods[k] for k in sorted(periods)],
}
這段程式碼把 Day 5、6、8、10 學的東西全用上了:
.get() 與 .setdefault():安全地取值、不存在就建立int() 轉換失敗時不要讓整件事崩掉sorted():把 dict 的 key 排序setdefault() 這個方法很好用:「如果這個 key 存在就拿出來,不存在就用預設值建立再拿出來」。一行取代了三行 if-else。
天氣預報不會每秒變。如果使用者在十分鐘內問三次天氣,我們打三次 API 是浪費——浪費時間(每次要等 1~2 秒)、也浪費額度。
用 Day 9、10 學的檔案與 JSON 來做快取:
import json
import time
from pathlib import Path
CACHE_DIR = Path("data/cache")
CACHE_TTL = 30 * 60 # 30 分鐘
def _cache_path(city):
return CACHE_DIR / f"weather_{city}.json"
def read_cache(city):
"""讀取快取,過期或不存在回傳 None。"""
path = _cache_path(city)
if not path.exists():
return None
try:
with open(path, "r", encoding="utf-8") as f:
cached = json.load(f)
except json.JSONDecodeError:
return None
if time.time() - cached.get("cached_at", 0) > CACHE_TTL:
return None # 過期了
return cached["data"]
def write_cache(city, data):
CACHE_DIR.mkdir(parents=True, exist_ok=True)
with open(_cache_path(city), "w", encoding="utf-8") as f:
json.dump(
{"cached_at": time.time(), "data": data},
f, ensure_ascii=False, indent=2,
)
快取的核心概念就三件事:
cached_at)把三個部分組合起來,就是可以直接用的工具:
"""tools/weather.py — 中央氣象署天氣查詢。"""
import json
import time
from pathlib import Path
import requests
CWA_BASE = "https://opendata.cwa.gov.tw/api/v1/rest/datastore"
FORECAST_36H = "F-C0032-001"
CACHE_DIR = Path("data/cache")
CACHE_TTL = 30 * 60
ELEMENT_MAP = {
"Wx": "weather",
"PoP": "rain_probability",
"MinT": "min_temp",
"MaxT": "max_temp",
"CI": "comfort",
}
NUMERIC_FIELDS = {"rain_probability", "min_temp", "max_temp"}
class WeatherError(Exception):
"""天氣查詢失敗。"""
# --- 快取 ---(略,同上)
# --- 主要入口 ---
def get_weather(city, api_key, use_cache=True):
"""查詢城市的 36 小時天氣預報。
Args:
city: 城市名稱,例如「臺北市」(注意是「臺」不是「台」)
api_key: 中央氣象署授權碼
use_cache: 是否使用快取
Returns:
dict: {"city": ..., "periods": [...]}
Raises:
WeatherError: 查詢失敗
"""
city = normalize_city(city)
if use_cache:
cached = read_cache(city)
if cached is not None:
return cached
try:
response = requests.get(
f"{CWA_BASE}/{FORECAST_36H}",
params={"Authorization": api_key, "locationName": city, "format": "JSON"},
timeout=10,
)
response.raise_for_status()
raw = response.json()
except requests.Timeout as e:
raise WeatherError("查詢天氣逾時,請稍後再試") from e
except requests.ConnectionError as e:
raise WeatherError("無法連線到氣象署,請檢查網路") from e
except requests.HTTPError as e:
code = e.response.status_code
if code == 401:
raise WeatherError("氣象署授權碼無效,請檢查 CWA_API_KEY") from e
raise WeatherError(f"氣象署回應錯誤(HTTP {code})") from e
except ValueError as e:
raise WeatherError("氣象署回應格式異常") from e
if raw.get("success") not in ("true", True):
raise WeatherError("氣象署回報查詢失敗")
try:
data = parse_forecast(raw)
except (KeyError, ValueError) as e:
raise WeatherError(f"無法解析天氣資料:{e}") from e
write_cache(city, data)
return data
def normalize_city(city):
"""把常見的城市寫法修正成氣象署認得的格式。"""
city = city.strip()
city = city.replace("台", "臺")
if not city.endswith(("市", "縣")):
# 簡單的補字規則
counties = {"新竹", "嘉義", "苗栗", "彰化", "南投", "雲林",
"屏東", "宜蘭", "花蓮", "臺東", "澎湖", "金門", "連江"}
city += "縣" if city in counties else "市"
return city
def summarize(data):
"""把天氣資料轉成一段給人(或給 AI)讀的文字。"""
lines = [f"【{data['city']}】未來 36 小時天氣"]
for p in data["periods"]:
start = p["start"][5:16] # 只保留「09-28 12:00」
end = p["end"][11:16]
lines.append(
f"{start}~{end} {p['weather']},"
f"{p['min_temp']}–{p['max_temp']}°C,"
f"降雨機率 {p['rain_probability']}%,{p['comfort']}"
)
return "\n".join(lines)
試跑:
import config
from tools.weather import get_weather, summarize, WeatherError
try:
data = get_weather("台北", config.CWA_API_KEY) # 故意用「台」
print(summarize(data))
except WeatherError as e:
print(f"查詢失敗:{e}")
輸出:
【臺北市】未來 36 小時天氣
09-28 12:00~18:00 多雲,26–31°C,降雨機率 20%,悶熱
09-28 18:00~06:00 多雲,25–28°C,降雨機率 10%,舒適
09-29 06:00~18:00 短暫陣雨,25–30°C,降雨機率 60%,悶熱
今天這支程式看起來只是「查天氣」,但它的結構是刻意設計的,值得回頭看一下:
WeatherErrorclass WeatherError(Exception):
"""天氣查詢失敗。"""
我們把所有可能的失敗(逾時、斷線、授權錯誤、格式異常)統一包裝成一種例外,而且訊息是人看得懂的中文。
這樣呼叫端只要寫:
except WeatherError as e:
print(f"查詢失敗:{e}")
就能處理所有情況,不需要知道底層用了 requests、也不需要知道氣象署的 API 長什麼樣。這就是 Day 7 講的「封裝」。
而且這些錯誤訊息,等到 Day 21 之後可以直接丟回給 AI:「工具執行失敗,原因是『氣象署授權碼無效』」——AI 就知道不該再重試這個工具。
normalize_city() 幫使用者擦屁股使用者不會知道要打「臺」不是「台」。AI 也常常會給出「台北」。與其讓查詢失敗,不如在工具裡面幫忙修正。
好的工具應該對輸入寬容、對輸出嚴格。
summarize() 分離資料與呈現get_weather() 回傳結構化資料(給程式用),summarize() 把它轉成文字(給人或 AI 用)。兩件事分開,以後要改顯示格式不用動到查詢邏輯。
這在 Agent 裡特別重要——同一份資料,可能要:
三種格式,一份資料。
我們的生活助理,現在有了第一個真實的資訊來源。
回想一下 Day 1 的情境:
使用者:「幫我安排明天的行程」
AI 取得天氣、行程等資訊 → 分析 → 給出建議
今天完成的是第一個箭頭的一部分。接下來:
注意最後這一點——現在是我們決定什麼時候呼叫 get_weather()。到了 Day 21,會變成AI 自己決定。那就是「Workflow」跟「Agent」的差別。
WeatherError)把底層細節包起來,錯誤訊息要人看得懂get_weather)與呈現(summarize)分開明天我們要把今天寫的東西,正式升格成 Agent 可以使用的工具(Tool)。