大里 即時空氣品質資訊
https://airtw.moenv.gov.tw/AirQuality_APIs/WebWidget.aspx?site=7
import tkinter as tk
from tkinter import ttk, messagebox
import requests
import threading
import json
# 目標測站資訊
STATION_ID = "7" # site=7 對應 新北市土城測站
STATION_NAME = "台中市 大里"
AJAX_URL = "https://airtw.moenv.gov.tw/AirQuality_APIs/ajax_Widget.aspx"
# 定義一個共用的 headers 模板
COMMON_HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
'X-Requested-With': 'XMLHttpRequest',
'Referer': f'https://airtw.moenv.gov.tw/AirQuality_APIs/WebWidget.aspx?site={STATION_ID}'
}
# 定義 AQI 等級對應的中文描述
AQI_LEVELS = [
(0, 50, "良好 (一般人群可安心活動)"),
(51, 100, "普通 (對少數敏感族群有輕微影響)"),
(101, 150, "對敏感群族群不健康"),
(151, 200, "對所有群族群不健康"),
(201, 300, "非常不健康"),
(301, 500, "危害")
]
# ===============================================
# 函式定義區塊 (保持不變)
# ===============================================
def get_aqi_description(aqi_value):
"""根據 AQI 數值返回對應的中文健康影響描述。"""
try:
if not aqi_value or not str(aqi_value).isdigit():
return "有效數據不足"
aqi = int(aqi_value)
for lower, upper, description in AQI_LEVELS:
if lower <= aqi <= upper:
return description
if aqi > 500:
return "危害"
except Exception:
return "有效數據不足"
return "有效數據不足"
def get_server_post_time():
"""
第一步：發送請求獲取伺服器時間 (tw_post_time)。
"""
time_payload = {"Target": "system_time"}
try:
response = requests.post(AJAX_URL, data=time_payload, headers=COMMON_HEADERS, timeout=5)
response.raise_for_status()
raw_text = response.text.strip()
clean_text = raw_text.encode('utf-8').decode('utf-8-sig').strip()
if clean_text.startswith("error:"):
return None
time_data = json.loads(clean_text)
return time_data.get('tw_post_time')
except Exception as e:
return None
def fetch_aqi_data_json():
"""
第二步：使用獲取的伺服器時間，發送請求獲取空氣品質數據。
"""
post_time = get_server_post_time()
if not post_time:
return None
data_payload = {
"Target": "air_list",
"SiteID": STATION_ID,
"Datatime": post_time,
"Type": "home"
}
raw_text = ""
try:
response = requests.post(AJAX_URL, data=data_payload, headers=COMMON_HEADERS, timeout=10)
response.raise_for_status()
raw_text = response.text.strip()
clean_text = raw_text.encode('utf-8').decode('utf-8-sig').strip()
if clean_text.startswith("error:"):
print(f"後端返回數據錯誤: {clean_text}")
return None
json_data = json.loads(clean_text)
# 提取所有 6 個數據點
data = {
'Time': json_data[0].get('date', 'N/A'),
'AQI': json_data[4].get('AQI', 'N/A'),
# PM2.5
'PM2.5_FIX': json_data[6].get('PM25_FIX', 'N/A'), # 小時濃度
'AVPM25': json_data[5].get('AVPM25', 'N/A'), # 24小時移動平均 (索引 5)
# PM10
'PM10_FIX': json_data[8].get('PM10_FIX', 'N/A'), # 小時濃度
'AVPM10': json_data[7].get('AVPM10', 'N/A'), # 小時移動平均 (索引 7)
# O3
'O3_FIX': json_data[10].get('O3_FIX', 'N/A'), # 小時濃度
'AVO3': json_data[9].get('AVO3', 'N/A') # 8小時移動平均 (索引 9)
}
return data
except requests.exceptions.RequestException as e:
print(f"數據請求網路錯誤: {e}")
return None
except json.JSONDecodeError as e:
print(f"解析 JSON 響應失敗。錯誤詳情: {e}")
print(f"嘗試解析的原始文本: {raw_text[:200]}...")
return None
except IndexError:
print("JSON 數據結構不符合預期，無法提取 (索引錯誤)。")
return None
def update_gui(data):
"""
在 Tkinter 視窗中更新數據顯示，包含所有 6 個污染物數據。
"""
if data and data.get('AQI') != 'N/A':
aqi_value = data.get('AQI')
description = get_aqi_description(aqi_value)
# 更新 AQI
aqi_var.set(f"AQI 指數: {aqi_value} \n健康影響: {description}")
# 更新 PM2.5 數據
pm25_fix_var.set(f"PM2.5 (小時濃度): {data.get('PM2.5_FIX')} μg/m³")
avpm25_var.set(f"PM2.5 (24小時移動平均): {data.get('AVPM25')} μg/m³")
# 更新 PM10 數據
pm10_fix_var.set(f"PM10 (小時濃度): {data.get('PM10_FIX')} μg/m³")
avpm10_var.set(f"PM10 (小時移動平均): {data.get('AVPM10')} μg/m³")
# 更新 O3 數據
o3_fix_var.set(f"O3 (小時濃度): {data.get('O3_FIX')} ppb")
avo3_var.set(f"O3 (8小時移動平均): {data.get('AVO3')} ppb")
time_var.set(f"更新時間: {data.get('Time', 'N/A')}")
status_var.set("狀態：數據更新成功！")
else:
status_var.set("狀態：數據爬取失敗或無有效數據。")
# 清空數據顯示
aqi_var.set("AQI 指數: N/A \n健康影響: 有效數據不足")
pm25_fix_var.set("PM2.5 (小時濃度): N/A")
avpm25_var.set("PM2.5 (24小時移動平均): N/A")
pm10_fix_var.set("PM10 (小時濃度): N/A")
avpm10_var.set("PM10 (小時移動平均): N/A")
o3_fix_var.set("O3 (小時濃度): N/A")
avo3_var.set("O3 (8小時移動平均): N/A")
time_var.set("更新時間: N/A")
messagebox.showerror("錯誤", "無法獲取空氣品質數據。請檢查網路連線。")
def start_scraping():
"""
啟動爬蟲並在一個獨立的線程中執行，避免 GUI 凍結
"""
status_var.set("狀態：正在獲取伺服器時間，準備請求數據...")
# 禁用兩個按鈕
fetch_button.config(state=tk.DISABLED)
update_button.config(state=tk.DISABLED)
def run_scraper():
data = fetch_aqi_data_json()
root.after(0, lambda: update_gui(data))
# 爬取完成後重新啟用按鈕
root.after(0, lambda: fetch_button.config(state=tk.NORMAL))
root.after(0, lambda: update_button.config(state=tk.NORMAL))
threading.Thread(target=run_scraper).start()
# ===============================================
# Tkinter GUI 設定區塊
# ===============================================
root = tk.Tk()
root.title(f"{STATION_NAME} 即時空氣品質資訊")
# *** 調整視窗高度，給底部更多空間 ***
root.geometry("550x680") # 略微增加高度
root.resizable(False, False)
# 樣式設定
style = ttk.Style()
style.configure('TLabel', font=('Arial', 14))
style.configure('TButton', font=('Arial', 12))
# 數據變量
aqi_var = tk.StringVar(value="AQI 指數: N/A \n健康影響: 點擊獲取數據")
pm25_fix_var = tk.StringVar(value="PM2.5 (小時濃度): N/A")
avpm25_var = tk.StringVar(value="PM2.5 (24小時移動平均): N/A")
pm10_fix_var = tk.StringVar(value="PM10 (小時濃度): N/A")
avpm10_var = tk.StringVar(value="PM10 (小時移動平均): N/A")
o3_fix_var = tk.StringVar(value="O3 (小時濃度): N/A")
avo3_var = tk.StringVar(value="O3 (8小時移動平均): N/A")
time_var = tk.StringVar(value="更新時間: N/A")
status_var = tk.StringVar(value="狀態：等待操作")
# 標題
title_label = ttk.Label(root, text=f"{STATION_NAME} 即時空氣品質", font=('Arial', 18, 'bold'))
title_label.pack(pady=(20, 15))
# 數據顯示框架
data_frame = ttk.Frame(root, padding="20")
data_frame.pack(fill='x', padx=30)
# AQI
ttk.Label(data_frame, textvariable=aqi_var, justify=tk.LEFT).pack(anchor='w', pady=10)
# PM2.5
ttk.Separator(data_frame, orient='horizontal').pack(fill='x', pady=8)
ttk.Label(data_frame, text="--- PM2.5 ---", font=('Arial', 12, 'bold')).pack(anchor='w')
ttk.Label(data_frame, textvariable=pm25_fix_var).pack(anchor='w', pady=4)
ttk.Label(data_frame, textvariable=avpm25_var).pack(anchor='w', pady=4)
# PM10
ttk.Separator(data_frame, orient='horizontal').pack(fill='x', pady=8)
ttk.Label(data_frame, text="--- PM10 ---", font=('Arial', 12, 'bold')).pack(anchor='w')
ttk.Label(data_frame, textvariable=pm10_fix_var).pack(anchor='w', pady=4)
ttk.Label(data_frame, textvariable=avpm10_var).pack(anchor='w', pady=4)
# O3
ttk.Separator(data_frame, orient='horizontal').pack(fill='x', pady=8)
ttk.Label(data_frame, text="--- O3 ---", font=('Arial', 12, 'bold')).pack(anchor='w')
ttk.Label(data_frame, textvariable=o3_fix_var).pack(anchor='w', pady=4)
ttk.Label(data_frame, textvariable=avo3_var).pack(anchor='w', pady=4)
# 更新時間
ttk.Separator(data_frame, orient='horizontal').pack(fill='x', pady=8)
ttk.Label(data_frame, textvariable=time_var).pack(anchor='w', pady=15)
# *** 關鍵修正：調整按鈕框架和狀態欄的 pack 設置 ***
# 按鈕框架 (在所有數據之後 pack)
button_frame = ttk.Frame(root)
# 使用 fill='x' 讓 frame 佔滿寬度，並給予足夠的 pady
button_frame.pack(fill='x', pady=15)
fetch_button = ttk.Button(button_frame, text="獲取最新數據", command=start_scraping)
fetch_button.pack(side=tk.LEFT, padx=30, expand=True) # expand=True 讓按鈕在水平方向上有更多空間
update_button = ttk.Button(button_frame, text="更新數據", command=start_scraping)
update_button.pack(side=tk.RIGHT, padx=30, expand=True) # expand=True 讓按鈕在水平方向上有更多空間
# 狀態欄 (放在最後，讓它緊貼底部)
# 移除 side=tk.BOTTOM，讓 pack 按照順序堆疊到底部
status_label = ttk.Label(root, textvariable=status_var, font=('Arial', 10), foreground='blue')
status_label.pack(fill='x', pady=(0, 10)) # 僅在底部保留一些空間 (10)
# 啟動 Tkinter 主迴圈
root.mainloop()
沒有留言:
張貼留言