視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統
這份 gui_app_timer.py 程式是一個 結合桌面視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統。
整體架構採用 事件驅動 (Event-Driven) 與 多執行緒非同步設計 (Multithreading),解決了傳統 GUI 在進行網路連線時容易畫面卡死 (Freeze) 的問題。
核心功能模組
自動化 API 爬蟲與資料擷取
使用
requests連線中央氣象署 API(F-C0032-001),自動抓取全台 22 縣市的即時氣象資料(天氣現象、最低溫、最高溫、降雨機率)。
Google Drive / Sheets 雲端資料庫同步
透過
gspread與 Service Account 金鑰,自動將資料寫入雲端試算表Taiwan_Weather_Data。內建 智慧標頭校正 與 ID 動態累加機制,確保試算表第一列永遠是標準標頭,且流水號不重複、舊資料不覆蓋。
雙層多執行緒架構 (Thread Safety)
背景執行緒 (Worker Thread):專門處理 API 請求、Google Sheet 寫入、Pandas 資料分析與 Telegram 圖片生成上傳。
主執行緒 (Main GUI Thread):接收整理好的數據,安全地渲染 Matplotlib Canvas 圖表,確保畫面順暢不報錯。
10 分鐘自動計時與倒數顯示
使用 Tkinter 原生
self.after(1000, ...)實現每秒更新的 10 分鐘倒數計時器。當倒數歸零時,全自動發起「API 抓取 ➔ 雲端備份 ➔ Telegram 推播 ➔ GUI 刷屏」完整流程。
Telegram Bot 主動圖片推播
當自動定時觸發或手動點擊「立即抓取並 Push」時,背景會將 Matplotlib 趨勢圖繪製至記憶體緩衝區(
io.BytesIO),連同文字摘要發送至指定 Telegram Chat ID。
程式詳細執行流程圖
┌───────────────────────────────────────────────────────────┐
│ 1. 系統啟動與初始化 │
│ - 載入 Service Account 與 API 金鑰 │
│ - 建立 Tkinter 視窗、控制按鈕與下拉選單 (預設「臺中市」) │
│ - 啟動 10 分鐘倒數計時器迴圈 │
└─────────────────────────────┬─────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────┐
│ 2. 啟動背景執行緒 (_async_load_data) │
│ (僅在系統啟動、10分鐘到期或點擊「立即抓取」時執行 API 抓取) │
└─────────────────────────────┬─────────────────────────────┘
│
┌───────────────┴───────────────┐
▼ ▼
┌───────────────────────────┐ ┌───────────────────────────┐
│ 2-A. API 與雲端備份 │ │ 2-B. Telegram 圖片推播 │
│ - 呼叫 CWA 氣象 API │ │ - 背景繪製 PNG 圖表 Stream │
│ - 檢查與修復 Sheet 標頭 │ │ - 組合最新天氣數據文字摘要 │
│ - 計算流水號 ID 並寫入 │ │ - 呼叫 API sendPhoto 推播 │
└─────────────┬─────────────┘ └─────────────┬─────────────┘
│ │
└───────────────┬───────────────┘
│
▼
┌───────────────────────────────────────────────────────────┐
│ 3. Pandas 資料整理 (_prepare_data) │
│ - 讀取 Google Sheet 數據 │
│ - 篩選指定縣市近 24 小時紀錄 │
│ - 計算平均溫度 ((最高+最低)/2) │
└─────────────────────────────┬─────────────────────────────┘
│ (將整理好的 DataFrame 送回主執行緒)
▼
┌───────────────────────────────────────────────────────────┐
│ 4. 主執行緒 GUI 圖表渲染 (Main Thread) │
│ - 銷毀舊 Canvas 與 Figure (防止記憶體洩漏) │
│ - 繪製最高溫/最低溫/平均溫折線圖 │
│ - 渲染至 Tkinter 視窗中央 │
│ - 更新下拉選單之全台縣市選單與成功狀態 │
└───────────────────────────────────────────────────────────┘
主要控制介面與觸發邏輯
「📊 快捷城市按鈕」/「選單查詢」:
動作:僅針對雲端試算表已有數據進行重新篩選與繪圖,不會重新呼叫 CWA API 或推送 Telegram,避免洗版。
「🔄 立即抓取並 Push」按鈕:
動作:強制立即觸發 API 抓取、寫入 Google 雲端、推送 Telegram 並重新繪圖,同時重置 10 分鐘倒數計時。
「倒數計時歸零 (00:00)」:
動作:自動執行完整的 API 抓取、雲端同步、Telegram 推播與介面刷屏。
import io
import traceback
import threading
import requests
import tkinter as tk
from tkinter import ttk, messagebox
import pandas as pd
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg
import gspread
from google.oauth2.service_account import Credentials
# --------------------------------------------------
# 1. 內嵌金鑰與憑證設定
# --------------------------------------------------
CWA_API_KEY = "CWB-402C25FFF-12224-4250-B9D9-3735AAE17DBF" # 請確保填入開頭為 CWA- 的有效金鑰
TELEGRAM_BOT_TOKEN = "884527208831:AAEuRreXFRyv1m--jGK7OCn8MiEsWXqpMT0"
TELEGRAM_CHAT_ID = "796522218469"
SPREADSHEET_NAME = "Taiwan_Weather_Data"
SERVICE_ACCOUNT_INFO = {
"type": "service_account",
"project_id": "weather-monitor-513909",
"private_key_id": "bf332fe535a602532b14ec734f109d0da9d4eddc",
"private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDLS3P8oWzQlrAP\nP7wBMTspZK+0o8ISNOyyoJ9FWBeemWKMnUH+MZZrsNfGn0n/9ad3xhBfstysqnBC\nwAqKo8tLgcgjBBpN8Ff03xPwi02xOEB6+axnPmbblqzrRC9UHa08y9QHVoVMHZ1w\nbLykB1sBxtJOMWUatvxYL7vv6PGr2hJSXuBxPvaWotrrNW+E/F5ItUNeWyujMLJR\nVu94rEowIaeKbjwKPg3C16o6QdWquFHu5aeij7rxXckESWjKYW0T9YcUtf8mx/+O\n5buObPbUSUMDiKi2jxRFdETtSy3x+Wl6CzYGfBSSmCQ+13GtpS285kVp2lXsa8A9\nNS+dYPQRAgMBAAECggEADksNCdlQpP6/PKc6wOu08OVZHC56TdzmOm3kHNRDft1i\naigBVI2oeYRau2aGJaagImHs00moTcSC9ZJ6e0JmX/XtPBcFAW9RQ9BuXrZwnTHr\nCKpmtnCS+0Pl/vmRiUy32iMrkmOBJjeb0uR2Q+MXgTG5JhG/fcgZSFr8z/KVEUO4\n9rLmU+2rf7QcPNlM3okfgAqKZUr4OFQccPA0vtgW9cmeg5+eV1fZHaGW91Q0mNk3\nM3EZhjtTVs7163GmBbMO5ZNnvFYKWIRmXfRow8f24ykNRqHFpPPIvNCcwKo7kPkX3\nl4kz4Gs79+csPqLkRrY7g2gcUbXMiWgfefN6lpqZBQKBgQDr5lS8D20i4JmJ1dS+\nz01AlAkg+kp8R9W4FqJCY93I1aZDcvRTFrqQzXR4G+Yk7EwyRs1RayDdIC/y0nHG\nGG8OjMtxW9DrmKOiSEcNoCKfdgPX5mOfrOxrjagk/geYuqGb7BB9DZSRd2A4qBK+\nmSb/AUyr9IvBZrNA2PB0ImiYNQKBgQDcnek44P5KsmCaPhYMki67adxLEapYaz2Z\ninAiZ4A4YlNCzHNs6zmR2Yb/dCUbtHlQ6hiIeodEQnkYYoTgTvX8UvLrNYt+jPoX\n31chl/iLHzTE5VUOPYR7gJeP/xhQVAqK16zjzRqo+crH+P0DVCJ/lzA4VoHOKrT2\nVqUuCZs/7QKBgQCQf3ZH++sOAiYKmEDI4JKhjAxLdw1VJO3deW/7oS+P+sWA20KN4\ngeOl7zP4np5K37OM2HHL/aMT0DOOufmXLkB1Y2UWFGojOx6D0NvHPFtgedVvFVAqS\nNIpdF1h30GM22ptDt0YiU789m7IFAm748TdB3LnqWtGAljgW8MvZV3paQQKBgQDP\nKWkrA/tFYSLdM5e4zjwbuFgoFbqDtI/vcdPCOrZoO8w7WYi5qlgw4LanggSmc13H\n4UqQPZ5vNoxHnZtwljTeb3Wz8F/AwdjGOe6zmOTGntMKYN7xtEqHPkUT7lKCe6g6\noDO05ZA3Q/hqtx++1G5gqTOac4EIe4n1aF5qcFGE4MQKBgDvNtoReH7iLHW7/FecF\nMB2/8k3Q9sNStQdnOe5yHZkvcAiFzCwD+LaK+VXDi1oBKQloE3Ih+0TfvEz0QRiu\nNP1bH97yBV/hqYyTgfvc1tXsR9MhJQdioPCJbGNsHLOiiMNCM3t6Vw75yaipwJOU\nj8spBjjqlVjW3TzMpRh7qDki\n-----END PRIVATE KEY-----\n",
"client_email": "weather-bot@weather-monitor-513909.iam.gserviceaccount.com",
"client_id": "1068356542212025201182272",
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"token_uri": "https://oauth2.googleapis.com/token",
"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/weather-bot%40weather-monitor-513909.iam.gserviceaccount.com",
"universe_domain": "googleapis.com"
}
# 設定 Matplotlib 中文字型
plt.rcParams['font.sans-serif'] = ['Microsoft JhengHei', 'Arial Unicode MS', 'sans-serif']
plt.rcParams['axes.unicode_minus'] = False
# --------------------------------------------------
# 2. Tkinter 主程式 (含 Telegram 自動推送功能)
# --------------------------------------------------
class AutoRefreshWeatherApp(tk.Tk):
def __init__(self):
super().__init__()
self.title("台灣城市 24 小時氣溫監控 (自動 API / 10分鐘刷屏 / Push Telegram)")
self.geometry("980x680")
self.resizable(True, True)
self.current_city = "臺中市"
self.refresh_interval = 600 # 10 分鐘 (600 秒)
self.countdown_seconds = self.refresh_interval
self.timer_job = None
self.canvas_widget = None
self.fig = None
self._create_widgets()
# 啟動時:立即抓取 API、寫入雲端並 Push 至 Telegram
self.on_query_city(self.current_city, force_fetch_api=True)
self._start_timer_loop()
def _create_widgets(self):
"""建立 UI 控制介面"""
control_frame = ttk.LabelFrame(self, text=" 城市選擇與定時設定 ", padding=10)
control_frame.pack(fill=tk.X, padx=15, pady=10)
# 快捷城市按鈕
cities = ["臺中市", "臺北市", "高雄市", "南投縣"]
for city in cities:
btn = ttk.Button(
control_frame,
text=f"📊 {city}",
command=lambda c=city: self.on_query_city(c, force_fetch_api=False)
)
btn.pack(side=tk.LEFT, padx=4)
# 下拉選單 (ttk.Combobox)
ttk.Label(control_frame, text="選擇縣市:").pack(side=tk.LEFT, padx=(15, 2))
self.cbo_city = ttk.Combobox(control_frame, width=10, state="readonly")
self.cbo_city['values'] = ("臺中市", "臺北市", "高雄市", "南投縣", "彰化縣", "雲林縣", "嘉義市", "新北市", "桃園市", "臺南市", "花蓮縣", "宜蘭縣", "臺東縣")
self.cbo_city.set("臺中市")
self.cbo_city.pack(side=tk.LEFT, padx=4)
btn_custom = ttk.Button(
control_frame,
text="查詢",
command=lambda: self.on_query_city(self.cbo_city.get().strip(), force_fetch_api=False)
)
btn_custom.pack(side=tk.LEFT, padx=4)
# 手動刷新按鈕
btn_refresh = ttk.Button(
control_frame,
text="🔄 立即抓取並 Push",
command=lambda: self.on_query_city(self.current_city, force_fetch_api=True)
)
btn_refresh.pack(side=tk.RIGHT, padx=5)
# 狀態列與倒數標籤
status_frame = ttk.Frame(self)
status_frame.pack(fill=tk.X, padx=15, pady=(0, 5))
self.lbl_status = ttk.Label(
status_frame,
text="系統初始化中...",
font=("Microsoft JhengHei", 10),
foreground="blue"
)
self.lbl_status.pack(side=tk.LEFT)
self.lbl_timer = ttk.Label(
status_frame,
text="⏳ 下次自動更新倒數: 10:00",
font=("Microsoft JhengHei", 10, "bold"),
foreground="#555555"
)
self.lbl_timer.pack(side=tk.RIGHT)
# 圖表顯示區域
self.chart_frame = ttk.Frame(self, relief=tk.SUNKEN, borderwidth=1)
self.chart_frame.pack(fill=tk.BOTH, expand=True, padx=15, pady=(0, 15))
# --------------------------------------------------
# 3. 定時器 (Timer) 邏輯
# --------------------------------------------------
def _start_timer_loop(self):
"""10 分鐘倒數計時器"""
mins, secs = divmod(self.countdown_seconds, 60)
self.lbl_timer.config(text=f"⏳ 下次自動更新倒數: {mins:02d}:{secs:02d}")
if self.countdown_seconds <= 0:
self.lbl_status.config(text=f"⏰ [10分鐘定時觸發] 正在抓取 API 並 Push「{self.current_city}」至 Telegram...", foreground="purple")
self._fetch_in_background(self.current_city, fetch_api=True)
self.countdown_seconds = self.refresh_interval
else:
self.countdown_seconds -= 1
self.timer_job = self.after(1000, self._start_timer_loop)
def _reset_timer(self):
"""重置倒數計時器"""
self.countdown_seconds = self.refresh_interval
mins, secs = divmod(self.countdown_seconds, 60)
self.lbl_timer.config(text=f"⏳ 下次自動更新倒數: {mins:02d}:{secs:02d}")
# --------------------------------------------------
# 4. 非同步資料抓取與推送
# --------------------------------------------------
def on_query_city(self, city_name: str, force_fetch_api: bool = False):
if not city_name:
messagebox.showwarning("警告", "請選擇有效的縣市名稱!")
return
self.current_city = city_name
self.cbo_city.set(city_name)
self._reset_timer()
action_msg = "正在抓取 API 並 Push Telegram" if force_fetch_api else "正在讀取"
self.lbl_status.config(text=f"⏳ {action_msg}「{city_name}」數據...", foreground="orange")
self._fetch_in_background(city_name, fetch_api=force_fetch_api)
def _fetch_in_background(self, city_name: str, fetch_api: bool = False):
threading.Thread(target=self._async_load_data, args=(city_name, fetch_api), daemon=True).start()
def _async_load_data(self, city_name: str, fetch_api: bool):
"""背景處理 API 抓取、寫入試算表,並推送圖表至 Telegram"""
try:
scopes = [
"https://www.googleapis.com/auth/spreadsheets",
"https://www.googleapis.com/auth/drive"
]
creds = Credentials.from_service_account_info(SERVICE_ACCOUNT_INFO, scopes=scopes)
client = gspread.authorize(creds)
sheet = client.open(SPREADSHEET_NAME).sheet1
if fetch_api:
self._fetch_cwa_api_and_write_sheets(sheet)
# 解析數據
filtered_df, all_cities = self._prepare_data_from_sheets(sheet, city_name)
# 產生圖表圖片並推送到 Telegram (只在抓取 API 或定時更新時推送)
if fetch_api:
self._push_chart_to_telegram(filtered_df, city_name)
# 安全送回 Main Thread 進行 Tkinter 介面繪圖
self.after(0, self._update_canvas_main_thread, filtered_df, city_name, all_cities)
except Exception as e:
print("\n" + "="*60)
print(f"❌ [執行失敗詳細原因] 時間: {pd.Timestamp.now()}")
print("="*60)
traceback.print_exc()
print("="*60 + "\n")
self.after(0, self._handle_error, str(e))
# --------------------------------------------------
# 5. Telegram 圖表 Push 核心函數 (含詳細發送 Log)
# --------------------------------------------------
def _push_chart_to_telegram(self, filtered_df: pd.DataFrame, city_name: str):
"""在背景產生無 GUI 的 Matplotlib 圖片 Stream,並透過 Telegram Bot API 發送照片"""
try:
print(f"[{pd.Timestamp.now().strftime('%H:%M:%S')}] 📡 開始準備發送 Telegram 推播...")
fig, ax = plt.subplots(figsize=(8, 4.5), dpi=100)
ax.plot(filtered_df['更新時間'], filtered_df['最高溫(°C)'], marker='o', color='#FF5722', linestyle='--', label='最高溫 (°C)')
ax.plot(filtered_df['更新時間'], filtered_df['最低溫(°C)'], marker='o', color='#03A9F4', linestyle='--', label='最低溫 (°C)')
ax.plot(filtered_df['更新時間'], filtered_df['平均溫(°C)'], marker='s', color='#4CAF50', linewidth=2.5, label='平均溫 (°C)')
ax.set_title(f"{city_name} 近 24 小時氣溫變化趨勢圖", fontsize=14, fontweight='bold', pad=12)
ax.set_xlabel("時間 (MM/DD HH:MM)", fontsize=10)
ax.set_ylabel("氣溫 (°C)", fontsize=10)
ax.xaxis.set_major_formatter(mdates.DateFormatter('%m/%d %H:%M'))
ax.xaxis.set_major_locator(mdates.HourLocator(interval=2))
fig.autofmt_xdate(rotation=30)
for _, row in filtered_df.iterrows():
ax.text(row['更新時間'], row['平均溫(°C)'] + 0.3, f"{row['平均溫(°C)']:.1f}°", ha='center', fontsize=8)
ax.grid(True, linestyle=':', alpha=0.6)
ax.legend(loc='best')
fig.tight_layout()
# 輸出成 PNG BytesIO 緩衝 Stream
buf = io.BytesIO()
plt.savefig(buf, format='png')
buf.seek(0)
plt.close(fig)
# 組合最新數據摘要
latest_row = filtered_df.iloc[-1]
caption = (
f"📊 <b>{city_name} 氣候自動播報</b>\n"
f"─────────────────\n"
f"🌤 天氣狀況:{latest_row['天氣狀況']}\n"
f"🌡 最低 / 最高溫:{latest_row['最低溫(°C)']}°C / {latest_row['最高溫(°C)']}°C\n"
f"💧 降雨機率:{latest_row['降雨機率']}\n"
f"🕒 更新時間:{pd.to_datetime(latest_row['更新時間']).strftime('%Y-%m-%d %H:%M:%S')}"
)
# 呼叫 Telegram sendPhoto API
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendPhoto"
files = {'photo': ('weather_chart.png', buf, 'image/png')}
data = {'chat_id': TELEGRAM_CHAT_ID, 'caption': caption, 'parse_mode': 'HTML'}
resp = requests.post(url, files=files, data=data, timeout=15)
if resp.status_code == 200:
print(f"[{pd.Timestamp.now().strftime('%H:%M:%S')}] 🚀 成功將「{city_name}」氣象圖表 Push 至 Telegram!")
else:
print("\n" + "❌"*20)
print(f"⚠️ Telegram Push 發送失敗!HTTP 狀態碼: {resp.status_code}")
print(f"⚠️ API 回應內容: {resp.text}")
print("❌"*20 + "\n")
except Exception as err:
print("\n" + "❌"*20)
print(f"⚠️ Telegram Push 發生例外異常: {err}")
traceback.print_exc()
print("❌"*20 + "\n")
# --------------------------------------------------
# 6. API 抓取與寫入雲端核心
# --------------------------------------------------
def _fetch_cwa_api_and_write_sheets(self, sheet):
url = f"https://opendata.cwa.gov.tw/api/v1/rest/datastore/F-C0032-001?Authorization={CWA_API_KEY}"
res = requests.get(url, timeout=10)
res.raise_for_status()
data = res.json()
locations = data['records']['location']
standard_headers = ["ID", "城市名稱", "天氣狀況", "最低溫(°C)", "最高溫(°C)", "降雨機率", "更新時間"]
all_values = sheet.get_all_values()
if not all_values:
sheet.append_row(standard_headers)
all_values = sheet.get_all_values()
max_id = 0
elif "城市名稱" not in all_values[0]:
sheet.insert_row(standard_headers, 1)
all_values = sheet.get_all_values()
max_id = 0
else:
id_list = []
for row in all_values[1:]:
if row and row[0].isdigit():
id_list.append(int(row[0]))
max_id = max(id_list) if id_list else 0
current_id = max_id + 1
fetch_time_str = pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
new_rows = []
for loc in locations:
city_name = str(loc['locationName'])
weather_elements = loc['weatherElement']
weather_state = str(weather_elements[0]['time'][0]['parameter']['parameterName'])
pop = str(weather_elements[1]['time'][0]['parameter']['parameterName'])
min_temp = float(weather_elements[2]['time'][0]['parameter']['parameterName'])
max_temp = float(weather_elements[4]['time'][0]['parameter']['parameterName'])
new_rows.append([current_id, city_name, weather_state, min_temp, max_temp, f"{pop}%", fetch_time_str])
current_id += 1
sheet.append_rows(new_rows)
print(f"[{fetch_time_str}] 成功抓取氣象 API 並追加 {len(new_rows)} 筆資料至 Google 試算表。")
# --------------------------------------------------
# 7. 純資料解析
# --------------------------------------------------
def _prepare_data_from_sheets(self, sheet, target_city: str, hours: int = 24):
all_rows = sheet.get_all_values()
if not all_rows:
raise ValueError("雲端試算表中目前沒有任何資料。")
standard_headers = ["ID", "城市名稱", "天氣狀況", "最低溫(°C)", "最高溫(°C)", "降雨機率", "更新時間"]
if "城市名稱" not in all_rows[0]:
df = pd.DataFrame(all_rows, columns=standard_headers)
else:
headers = [str(h).strip() for h in all_rows[0]]
df = pd.DataFrame(all_rows[1:], columns=headers)
col_map = {}
for c in df.columns:
if '城市' in c or 'city' in c.lower():
col_map[c] = '城市名稱'
elif '最低' in c or 'min' in c.lower():
col_map[c] = '最低溫(°C)'
elif '最高' in c or 'max' in c.lower():
col_map[c] = '最高溫(°C)'
elif '時間' in c or 'time' in c.lower():
col_map[c] = '更新時間'
df = df.rename(columns=col_map)
all_cities = sorted(list(set(df['城市名稱'].astype(str))))
df['更新時間'] = pd.to_datetime(df['更新時間'], errors='coerce')
df = df.dropna(subset=['更新時間']).sort_values(by='更新時間')
df['最低溫(°C)'] = pd.to_numeric(df['最低溫(°C)'], errors='coerce')
df['最高溫(°C)'] = pd.to_numeric(df['最高溫(°C)'], errors='coerce')
df = df.dropna(subset=['最低溫(°C)', '最高溫(°C)'])
df['平均溫(°C)'] = (df['最低溫(°C)'] + df['最高溫(°C)']) / 2
city_df = df[df['城市名稱'] == target_city].copy()
if city_df.empty:
raise ValueError(f"找不到縣市「{target_city}」的歷史紀錄。目前已有縣市:{all_cities}")
latest_time = city_df['更新時間'].max()
start_time = latest_time - pd.Timedelta(hours=hours)
filtered_df = city_df[city_df['更新時間'] >= start_time]
if filtered_df.empty:
raise ValueError(f"「{target_city}」近 {hours} 小時內無數據。")
return filtered_df, all_cities
# --------------------------------------------------
# 8. 主執行緒安全的繪圖與 UI 渲染
# --------------------------------------------------
def _update_canvas_main_thread(self, filtered_df: pd.DataFrame, city_name: str, all_cities: list):
if self.canvas_widget:
self.canvas_widget.destroy()
self.canvas_widget = None
if self.fig:
plt.close(self.fig)
self.fig, ax = plt.subplots(figsize=(8, 4.5), dpi=100)
ax.plot(filtered_df['更新時間'], filtered_df['最高溫(°C)'], marker='o', color='#FF5722', linestyle='--', label='最高溫 (°C)')
ax.plot(filtered_df['更新時間'], filtered_df['最低溫(°C)'], marker='o', color='#03A9F4', linestyle='--', label='最低溫 (°C)')
ax.plot(filtered_df['更新時間'], filtered_df['平均溫(°C)'], marker='s', color='#4CAF50', linewidth=2.5, label='平均溫 (°C)')
ax.set_title(f"{city_name} 近 24 小時氣溫變化趨勢圖", fontsize=14, fontweight='bold', pad=12)
ax.set_xlabel("時間 (MM/DD HH:MM)", fontsize=10)
ax.set_ylabel("氣溫 (°C)", fontsize=10)
ax.xaxis.set_major_formatter(mdates.DateFormatter('%m/%d %H:%M'))
ax.xaxis.set_major_locator(mdates.HourLocator(interval=2))
self.fig.autofmt_xdate(rotation=30)
for _, row in filtered_df.iterrows():
ax.text(row['更新時間'], row['平均溫(°C)'] + 0.3, f"{row['平均溫(°C)']:.1f}°", ha='center', fontsize=8)
ax.grid(True, linestyle=':', alpha=0.6)
ax.legend(loc='best')
self.fig.tight_layout()
canvas = FigureCanvasTkAgg(self.fig, master=self.chart_frame)
self.canvas_widget = canvas.get_tk_widget()
self.canvas_widget.pack(fill=tk.BOTH, expand=True)
canvas.draw()
if all_cities:
self.cbo_city['values'] = all_cities
update_time_str = pd.Timestamp.now().strftime("%H:%M:%S")
self.lbl_status.config(
text=f"✅ 已成功更新「{city_name}」圖表並 Push 至 Telegram (更新時間: {update_time_str})",
foreground="green"
)
def _handle_error(self, err_msg: str):
self.lbl_status.config(text=f"❌ 刷屏失敗: {err_msg} (詳細請看 Terminal)", foreground="red")
# --------------------------------------------------
# 9. 程式啟動
# --------------------------------------------------
if __name__ == "__main__":
app = AutoRefreshWeatherApp()
app.mainloop()




