Day 27 — 生活資訊面板：整合天氣、AQI、新聞 - iT 邦幫忙::一起幫忙解決難題，拯救 IT 人的一天

2025 iThome 鐵人賽

DAY 27

Modern Web

每天一點 API：打造我的生活小工具系列第 27 篇

Day 27 — 生活資訊面板：整合天氣、AQI、新聞

17th鐵人賽

alice__

2025-10-09 23:06:42

19 瀏覽

分享至

我已經學會如何使用單一 API，比如查詢天氣、取得最新新聞或貓咪圖片。

今天我想挑戰多個 API 的串接，也就是同時整合多個資料來源。

我希望能一次取得天氣、空氣品質指數（AQI）和最新新聞，然後把這些資料整理成一份自動更新的報表。

這樣的練習能幫助我了解如何將不同來源的 JSON 資料整合在一起，並學會讓這些資料能協同運作。這對我理解資料整合流程非常有幫助，也能讓我更熟悉多 API 同時使用的技巧。

今天的小專案目標

今天的目標是用 Python 程式自動連接三種不同的 API，學習如何整合多個資料來源。

首先，我會使用 OpenWeatherMap 的天氣 API，取得指定城市的即時天氣資訊，接著利用 OpenWeatherMap 的空氣污染 API，得到該城市的空氣品質指標，包括 AQI、PM2.5 與 PM10 數值。最後，我會從 Spaceflight News API 抓取最新的五則新聞，這樣可以練習整合文字資料。

程式執行時會同時呼叫這三個 API，並將取得的資料整合成一份完整的 JSON 和 CSV 檔案。

輸出內容會包含：

城市名稱、溫度、濕度和天氣描述
空氣品質指標（AQI）及 PM2.5、PM10 的數值
最新新聞的標題、來源和連結

透過這樣的練習，可以幫助我們理解如何把不同來源的 JSON 資料整合起來，學習如何讓多個 API 資料相互配合，製作出一份自動更新且易於閱讀的資訊報表。

實作流程

準備環境

建立虛擬環境並安裝套件

python -m venv .venv
.\.venv\Scripts\Activate.ps1 
pip install requests python-dotenv

準備 .env

OWM_API_KEY=我們的OpenWeatherMap金鑰
CITY=Taipei
NEWS_LIMIT=5
UNITS=![https://ithelp.ithome.com.tw/upload/images/20251009/20178708tShAq8Vbei.png](https://ithelp.ithome.com.tw/upload/images/20251009/20178708tShAq8Vbei.png)
LANG=zh_tw

.gitignore 記得忽略 .env

.env

這一步是為了避免金鑰外洩。

匯入、常數與路徑

import os, csv, json, argparse
from pathlib import Path
from datetime import datetime
import requests
from dotenv import load_dotenv

API_WEATHER = "https://api.openweathermap.org/data/2.5/weather"
API_AIR   = "https://api.openweathermap.org/data/2.5/air_pollution"  # AQI
API_NEWS  = "https://api.spaceflightnewsapi.net/v4/articles/"         # 免金鑰

ROOT = Path(__file__).resolve().parent
OUT_DIR = ROOT / "output"
RAW_DIR = OUT_DIR / "raw"

共用小工具

def ts():
    return datetime.now().strftime("%Y%m%d_%H%M%S")

def ensure_dir(p: Path):
    p.mkdir(parents=True, exist_ok=True)

def save_json(data, path: Path):
    ensure_dir(path.parent)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def save_csv_row(path: Path, fieldnames, row: dict):
    ensure_dir(path.parent)
    newfile = not path.exists()
    with open(path, "a", newline="", encoding="utf-8-sig") as f:
        w = csv.DictWriter(f, fieldnames=fieldnames)
        if newfile:
            w.writeheader()
        w.writerow(row)

def fetch_json(url, *, params=None, timeout=15):
    r = requests.get(url, params=params, timeout=timeout)
    r.raise_for_status()
    ct = r.headers.get("Content-Type","")
    # httpbin 之類有時候會回 text/html，這裡保守處理
    return r.json()

三個資料源

整合天氣 API

def get_weather(city: str, *, units="metric", lang="zh_tw"):
    # OpenWeatherMap 目前天氣（需金鑰）→ 回傳精簡資料與原始 JSON
    load_dotenv(ROOT / ".env")
    key = os.getenv("OWM_API_KEY")
    if not key:
        raise RuntimeError("找不到 OWM_API_KEY（請在 .env 或環境變數設定）")

    params = {"q": city, "appid": key, "units": units, "lang": lang}
    r = requests.get(API_WEATHER, params=params, timeout=15)
    if r.status_code == 401:
        raise RuntimeError("401 未授權：API Key 無效或未啟用")
    if r.status_code == 404:
        raise RuntimeError(f"找不到城市：{city}")
    r.raise_for_status()
    data = r.json()

    # 精簡欄位
    name = data.get("name") or city
    main = data.get("main") or {}
    weather0 = (data.get("weather") or [{}])[0]
    sys = data.get("sys") or {}
    country_code = sys.get("country")  # 例如 "TW"

    slim = {
        "city": name,
        "country_code": country_code,
        "temp_c": main.get("temp"),            # units=metric 下是 °C
        "feels_like_c": main.get("feels_like"),
        "humidity": main.get("humidity"),
        "weather": weather0.get("description"),
        "raw": data
    }
    return slim

整合空氣品質 API

def get_aqi_by_coord(lat: float, lon: float, *, key: str, timeout=15):
    """OpenWeatherMap Air Pollution：以經緯度取得 AQI 與污染物濃度。"""
    params = {"lat": lat, "lon": lon, "appid": key}
    r = requests.get(API_AIR, params=params, timeout=timeout)
    r.raise_for_status()
    data = r.json()
    # 典型結構：{"list":[{"main":{"aqi":1..5},"components":{"pm2_5":...,"pm10":...}}]}
    first = (data.get("list") or [{}])[0]
    main = first.get("main") or {}
    comps = first.get("components") or {}
    aqi = main.get("aqi")  # 1~5（1=Good, 5=Very Poor）

    return {
        "aqi": aqi,
        "pm2_5": comps.get("pm2_5"),
        "pm10": comps.get("pm10"),
        "no2": comps.get("no2"),
        "so2": comps.get("so2"),
        "o3": comps.get("o3"),
        "co": comps.get("co"),
        "raw": data
    }

整合新聞 API

def get_news(limit=5, timeout=15):
    # Spaceflight News 最新文章清單（免金鑰）。
    r = requests.get(API_NEWS, params={"limit": limit}, timeout=timeout)
    r.raise_for_status()
    data = r.json()
    items = data.get("results") or data.get("articles") or []
    # 只取精簡欄位
    slim = []
    for x in items:
        slim.append({
            "title": x.get("title"),
            "source": x.get("news_site") or (x.get("source",{}) or {}).get("name"),
            "published": x.get("published_at") or x.get("publishedAt"),
            "url": x.get("url"),
        })
    return {"count": len(slim), "items": slim, "raw": data}

主流程

def main():
    load_dotenv(ROOT / ".env")

    # 參數
    ap = argparse.ArgumentParser(description="Day27 多 API 串接：天氣 + 空氣品質 + 新聞")
    ap.add_argument("-c", "--city", default=os.getenv("CITY", "Taipei"), help="城市（預設 .env 的 CITY）")
    ap.add_argument("--units", choices=["metric","imperial"], default="metric", help="metric=攝氏, imperial=華氏")
    ap.add_argument("--lang", default="zh_tw", help="天氣回應語系（預設 zh_tw）")
    ap.add_argument("--aqi", action="store_true", help="同時查詢空氣品質（使用天氣回傳的經緯度）")
    ap.add_argument("--news-limit", type=int, default=int(os.getenv("NEWS_LIMIT", 5)),
                help="抓取新聞數量（預設 .env 的 NEWS_LIMIT，預設 5）")
    args = ap.parse_args()

讀 .env，用 argparse 取得城市、單位、語系，以及空氣品質、新聞數量等。

    ensure_dir(OUT_DIR); ensure_dir(RAW_DIR)

    weather = get_weather(args.city, units=args.units, lang=args.lang)
    save_json(weather["raw"], RAW_DIR / f"weather_{ts()}.json")

    # 從天氣回應拿經緯度
    coord = (weather["raw"].get("coord") or {})
    lat, lon = coord.get("lat"), coord.get("lon")

    # 讀 OWM key（已在 get_weather 裡 load 過 .env，但這邊再拿一次）
    key = os.getenv("OWM_API_KEY")


    aqi = get_aqi_by_coord(lat, lon, key=key)
    save_json(aqi["raw"], RAW_DIR / f"aqi_{ts()}.json")


    news_limit = int(os.getenv("NEWS_LIMIT", "5"))
    news = get_news(limit=news_limit)
    save_json(news["raw"], RAW_DIR / f"news_{ts()}.json")

程式會先抓天氣並把原始回應存到 raw/。

然後從天氣回應取出 lat/lon 再查 AQI，也把原始回應存檔。

讀 NEWS_LIMIT 抓新聞並存原始回應。

    now_iso = datetime.now().isoformat(timespec="seconds")
    unit_symbol = "°C" if args.units == "metric" else "°F"

    merged = {
    "timestamp": now_iso,
    "city": weather["city"],
    "country_code": weather.get("country_code"),
    "temp": weather["temp_c"],
    "feels_like": weather["feels_like_c"],
    "humidity": weather["humidity"],
    "weather_desc": weather["weather"],

    # AQI
    "aqi": aqi["aqi"],           # 1=Good, 5=Very Poor
    "pm2_5": aqi["pm2_5"],
    "pm10": aqi["pm10"],

    # News（摘要）
    "news_count": news["count"],
    "news_top_title": (news["items"][0]["title"] if news["count"] else None),
    "news_top_source": (news["items"][0]["source"] if news["count"] else None),
    "news_top_url": (news["items"][0]["url"] if news["count"] else None),

    "sources": "OpenWeatherMap Weather/Air, SpaceflightNews"
    }

彙整三方資料成一個字典，便於後續列印與寫檔。

    # 印出漂亮摘要
    print(f"時間：{merged['timestamp']}")
    print(f"地點：{merged['city']}（{merged.get('country_code') or '-'}）")
    print(f"天氣：{merged['weather_desc']}，{merged['temp']}{unit_symbol}（體感 {merged['feels_like']}{unit_symbol}），濕度 {merged['humidity']}%")
    print(f"AQI ：{merged['aqi']}（1=好、5=很差），PM2.5={merged['pm2_5']} μg/m³，PM10={merged['pm10']} μg/m³")
    if merged['news_count']:
        print(f"新聞：共 {merged['news_count']} 則｜Top1：{merged['news_top_title']}｜{merged['news_top_source']}")
        print(f"連結：{merged['news_top_url']}")
    else:
        print("新聞：無")

依單位顯示 °C 或 °F，並把 AQI、PM2.5/PM10 與新聞 Top1 印在終端機。

    # 存檔
    json_path = OUT_DIR / f"day27_summary_{ts()}.json"
    csv_path  = OUT_DIR / "day27_summary.csv"

    save_json(merged, json_path)

    fields = [
    "timestamp","city","country_code",
    "temp","feels_like","humidity","weather_desc",
    "aqi","pm2_5","pm10",
    "news_count","news_top_title","news_top_source","news_top_url",
    "sources"
]

    save_csv_row(csv_path, fields, merged)

    print(f"\n 已輸出：\n- {json_path}\n- {csv_path}\n- 原始回應在 {RAW_DIR}")

把彙整結果同時輸出成一份帶時間戳的 JSON 與累積型的 CSV，最後在印出檔案路徑與 raw/ 位置。

程式入口與網路錯誤處理

if __name__ == "__main__":
    try:
        main()
    except requests.exceptions.Timeout:
        print("連線逾時，請稍後再試。")
    except requests.exceptions.ConnectionError:
        print("連線錯誤，請檢查網路。")
    except requests.exceptions.HTTPError as e:
        print("HTTP 錯誤：", e)
    except RuntimeError as e:
        print("設定錯誤：", e)

執行結果：

時間：2025-10-09T12:34:54
地點：Taipei（TW）
天氣：多雲，31.96°C（體感 34.53°C），濕度 51%
AQI ：2（1=好、5=很差），PM2.5=23.39 μg/m³，PM10=30.75 μg/m³
新聞：共 5 則｜Top1：Stoke Space raises $510 million｜SpaceNews
連結：https://spacenews.com/stoke-space-raises-510-million/

 已輸出：
- C:\Users\Ariel\day27-1\output\day27_summary_20251009_123454.json
- C:\Users\Ariel\day27-1\output\day27_summary.csv
- 原始回應在 C:\Users\Ariel\day27-1\output\raw

.json 檔

指定城市：Tokyo

時間：2025-10-09T12:36:37
地點：Tokyo（JP）
天氣：陰，多雲，20.79°C（體感 20.31°C），濕度 53%
AQI ：2（1=好、5=很差），PM2.5=5.83 μg/m³，PM10=14.24 μg/m³
新聞：共 5 則｜Top1：Stoke Space raises $510 million｜SpaceNews
連結：https://spacenews.com/stoke-space-raises-510-million/

 已輸出：
- C:\Users\Ariel\day27-1\output\day27_summary_20251009_123637.json
- C:\Users\Ariel\day27-1\output\day27_summary.csv
- 原始回應在 C:\Users\Ariel\day27-1\output\raw

.json 檔

常見問題與解決方法

為什麼 AQI（空氣品質指數）顯示 None？
因為有些地區暫時沒有空氣品質資料，這時候可以換其他城市試試看。

為什麼會出現 HTTP 401 Unauthorized 錯誤？
這表示我們的 API 金鑰輸入錯誤或還沒有啟用，我們需要重新檢查 .env 檔案裡的 OWM_API_KEY 是否正確。金鑰一定要完全對應且有效，才能成功使用 API。

為什麼新聞 API 回傳 0 筆資料？
Spaceflight News API 有時資料更新有延遲，我們可以過幾分鐘再試，或是將抓取數量（NEWS_LIMIT）調大一點，通常就能抓到資料了。

今日總結

今天的練習讓我學會了如何整合多個 API，並處理不同類型的資料格式，例如氣象的數值資料和文字新聞資訊，然後把這些結果整合成一份報表。我發現只要設計好資料的結構，後續就能輕鬆加入更多 API，擴充功能。這次做出來的成果就像是一個生活資訊面板，裡面同時可以看到天氣狀況、空氣品質和新聞摘要。這個過程讓我更清楚了解資料整合在真實生活應用中的重要性和價值。