大里 即時空氣品質資訊 推播到 line Message API
大里 即時空氣品質資訊
https://airtw.moenv.gov.tw/AirQuality_APIs/WebWidget.aspx?site=7
需修改
LINE_CHANNEL_ACCESS_TOKEN =
LINE_TARGET_ID =
【台中市 大里】即時空品報告
細懸浮微粒 (PM2.5): 懸浮微粒 (PM10): 臭氧 (O3):
import tkinter as tk
from tkinter import ttk, messagebox
import requests
import threading
import json
import time
# ===============================================
# 全域設定區塊 (請修改此處)
# ===============================================
# --- 空氣品質測站資訊 ---
STATION_ID = "7" # 網站代碼。目前設定為 7 (新北市土城)。
STATION_NAME = "台中市 大里" # 這是 GUI 顯示名稱，若要修改測站請同時修改 STATION_ID
AJAX_URL = "https://airtw.moenv.gov.tw/AirQuality_APIs/ajax_Widget.aspx"
# --- LINE Messaging API 設定 (請務必修改此處) ---
# *** 1. 請在此處填入您的 LINE Channel Access Token ***
LINE_CHANNEL_ACCESS_TOKEN = "P7R4jd35usv1YhlJa1C9HGXwcqr6G0Yknp2vDxOb356AnOGHt5MPpzXmJrxj5L9OY5Z70h1DSdKRGr2/6Q8cN0bVoh6PcUMISbfncKvnMmv2HG5GCR+HMgpPj2LQYqOLDKgDqUGchzrkgkrGf1KhnhfnugdB04t89/1O/w1cDnyilFU="
# *** 2. 請在此處填入接收訊息的 User ID 或 Group ID ***
LINE_TARGET_ID = "U601f091afaaf1d41de21ace452105bfd3cf"
LINE_MESSAGE_API_URL = "https://api.line.me/v2/bot/message/push"
# 定義一個共用的 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:
print(f"獲取伺服器時間失敗: {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)
# 提取所有數據點
data = {
'Time': json_data[0].get('date', 'N/A'),
'AQI': json_data[4].get('AQI', 'N/A'),
'PM2.5_FIX': json_data[6].get('PM25_FIX', 'N/A'),
'AVPM25': json_data[5].get('AVPM25', 'N/A'),
'PM10_FIX': json_data[8].get('PM10_FIX', 'N/A'),
'AVPM10': json_data[7].get('AVPM10', 'N/A'),
'O3_FIX': json_data[10].get('O3_FIX', 'N/A'),
'AVO3': json_data[9].get('AVO3', 'N/A')
}
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 format_data_for_line(data):
"""將擷取到的數據格式化為 LINE 訊息所需的字串。"""
if not data or data.get('AQI') == 'N/A':
return f"\n【{STATION_NAME} 空氣品質】\n無法獲取最新的空氣品質數據。\n請檢查來源網站或網路連線。"
aqi_value = data.get('AQI')
description = get_aqi_description(aqi_value)
# 判斷空氣品質等級，用於訊息開頭
if aqi_value and str(aqi_value).isdigit() and int(aqi_value) >= 151:
header = f"🚨🚨🚨 警報：空氣品質對所有群族群不健康 🚨🚨🚨"
elif aqi_value and str(aqi_value).isdigit() and int(aqi_value) >= 101:
header = f"⚠️ 注意：空氣品質對敏感群族群不健康 ⚠️"
else:
header = f"✅ 空氣品質尚可"
message = (
f"\n{header}\n"
f"【{STATION_NAME} 即時空氣品質報告】\n"
f"============================\n"
f"📢 AQI 指數: {aqi_value}\n"
f"健康影響: {description}\n"
f"----------------------------\n"
f"💧 PM2.5 (小時濃度): {data.get('PM2.5_FIX')} μg/m³\n"
f"💧 PM2.5 (24H移動平均): {data.get('AVPM25')} μg/m³\n"
f"💨 PM10 (小時濃度): {data.get('PM10_FIX')} μg/m³\n"
f"💨 PM10 (小時移動平均): {data.get('AVPM10')} μg/m³\n"
f"🔆 O3 (小時濃度): {data.get('O3_FIX')} ppb\n"
f"🔆 O3 (8H移動平均): {data.get('AVO3')} ppb\n"
f"============================\n"
f"🕒 更新時間: {data.get('Time', 'N/A')}"
)
return message
def send_line_message_api(message):
"""
發送訊息到 LINE Messaging API (Push Message)。
使用 Channel Access Token 和 User/Group ID 發送 JSON 格式訊息。
"""
if LINE_CHANNEL_ACCESS_TOKEN == "YOUR_LINE_CHANNEL_ACCESS_TOKEN_HERE":
return False, "LINE Messaging API 錯誤：請先設定您的 Channel Access Token！"
if LINE_TARGET_ID == "YOUR_USER_OR_GROUP_ID_HERE":
return False, "LINE Messaging API 錯誤：請先設定接收訊息的 User ID 或 Group ID！"
headers = {
# Messaging API 使用 Bearer 授權
"Authorization": f"Bearer {LINE_CHANNEL_ACCESS_TOKEN}",
"Content-Type": "application/json"
}
# LINE Messaging API 採用 JSON Push 格式
payload = {
"to": LINE_TARGET_ID,
"messages": [
{
"type": "text",
"text": message
}
]
}
try:
response = requests.post(LINE_MESSAGE_API_URL, headers=headers, json=payload, timeout=10)
response.raise_for_status()
# Messaging API 成功時返回 status 200
if response.status_code == 200:
return True, "LINE 訊息發送成功！"
else:
# 嘗試解析錯誤訊息
response_json = response.json()
error_msg = response_json.get('message', '未知的 LINE API 錯誤')
return False, f"LINE API 錯誤 ({response.status_code}): {error_msg}"
except requests.exceptions.RequestException as e:
return False, f"LINE 訊息網路連線錯誤: {e}"
def start_line_notification():
"""
啟動 LINE 訊息發送流程 (必須先有數據)
"""
# 檢查是否有數據
if not hasattr(root, 'latest_data') or root.latest_data.get('AQI') == 'N/A':
messagebox.showwarning("無法發送", "請先點擊「獲取最新數據」成功抓取資料後再發送 LINE 訊息。")
return
# 設置按鈕狀態
line_button.config(state=tk.DISABLED)
status_var.set("狀態：正在準備並發送 LINE 訊息...")
def run_line_sender():
try:
message = format_data_for_line(root.latest_data)
# 使用新的 Messaging API 函式
success, result_msg = send_line_message_api(message)
# 更新狀態
root.after(0, lambda: status_var.set(f"狀態：{result_msg}"))
if not success:
root.after(0, lambda: messagebox.showerror("LINE 訊息失敗", result_msg))
finally:
# 恢復按鈕狀態
root.after(0, lambda: line_button.config(state=tk.NORMAL))
threading.Thread(target=run_line_sender).start()
def update_gui(data):
"""
在 Tkinter 視窗中更新數據顯示，並儲存最新數據以供 LINE 使用。
"""
# 儲存最新數據 (供 LINE 訊息使用)
root.latest_data = data
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("錯誤", "無法獲取空氣品質數據。請檢查網路連線或站點 ID。")
def start_scraping():
"""
啟動爬蟲並在一個獨立的線程中執行，避免 GUI 凍結。
"""
status_var.set("狀態：正在獲取伺服器時間，準備請求數據...")
# 禁用所有按鈕
fetch_button.config(state=tk.DISABLED)
update_button.config(state=tk.DISABLED)
line_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))
# 只有在成功取得資料後才啟用 LINE 訊息按鈕
if data and data.get('AQI') != 'N/A':
root.after(0, lambda: line_button.config(state=tk.NORMAL))
threading.Thread(target=run_scraper).start()
# ===============================================
# Tkinter GUI 設定區塊
# ===============================================
root = tk.Tk()
root.title(f"{STATION_NAME} 即時空氣品質資訊")
# 調整視窗高度以容納新增的 LINE 按鈕
root.geometry("550x750")
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="狀態：等待操作")
# 初始化一個儲存最新數據的屬性，用於 LINE 發送
root.latest_data = {'AQI': 'N/A'}
# 標題
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)
# --- 按鈕框架 1 (數據獲取) ---
fetch_button_frame = ttk.Frame(root)
fetch_button_frame.pack(fill='x', pady=10)
fetch_button = ttk.Button(fetch_button_frame, text="獲取最新數據", command=start_scraping)
fetch_button.pack(side=tk.LEFT, padx=30, expand=True)
update_button = ttk.Button(fetch_button_frame, text="更新數據", command=start_scraping)
update_button.pack(side=tk.RIGHT, padx=30, expand=True)
# --- 按鈕框架 2 (LINE 訊息) ---
line_button_frame = ttk.Frame(root)
line_button_frame.pack(fill='x', pady=10)
# 初始禁用 LINE 訊息按鈕，必須先抓到數據才能發送
line_button = ttk.Button(line_button_frame, text="發送 LINE 訊息 (Messaging API)", command=start_line_notification, state=tk.DISABLED)
line_button.pack(padx=30, fill='x')
# 狀態欄 (放在最後，讓它緊貼底部)
status_label = ttk.Label(root, textvariable=status_var, font=('Arial', 10), foreground='blue')
status_label.pack(fill='x', pady=(0, 10))
root.mainloop()
沒有留言:
張貼留言