2026年11月6日 星期五

結合桌面視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統

視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統


這份 gui_app_timer.py 程式是一個 結合桌面視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統。

整體架構採用 事件驅動 (Event-Driven) 與 多執行緒非同步設計 (Multithreading),解決了傳統 GUI 在進行網路連線時容易畫面卡死 (Freeze) 的問題。

核心功能模組

  1. 自動化 API 爬蟲與資料擷取

    • 使用 requests 連線中央氣象署 API(F-C0032-001),自動抓取全台 22 縣市的即時氣象資料(天氣現象、最低溫、最高溫、降雨機率)。

  2. Google Drive / Sheets 雲端資料庫同步

    • 透過 gspread 與 Service Account 金鑰,自動將資料寫入雲端試算表 Taiwan_Weather_Data。

    • 內建 智慧標頭校正 與 ID 動態累加機制,確保試算表第一列永遠是標準標頭,且流水號不重複、舊資料不覆蓋。

  3. 雙層多執行緒架構 (Thread Safety)

    • 背景執行緒 (Worker Thread):專門處理 API 請求、Google Sheet 寫入、Pandas 資料分析與 Telegram 圖片生成上傳。

    • 主執行緒 (Main GUI Thread):接收整理好的數據,安全地渲染 Matplotlib Canvas 圖表,確保畫面順暢不報錯。

  4. 10 分鐘自動計時與倒數顯示

    • 使用 Tkinter 原生 self.after(1000, ...) 實現每秒更新的 10 分鐘倒數計時器。當倒數歸零時,全自動發起「API 抓取 ➔ 雲端備份 ➔ Telegram 推播 ➔ GUI 刷屏」完整流程。

  5. 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()


Python Tkinter 的全自動全台氣候資料監控與視覺化系統

 Python Tkinter 的全自動全台氣候資料監控與視覺化系統






這份 gui_app_timer.py 程式是一個基於 Python Tkinter 的全自動全台氣候資料監控與視覺化系統。程式整合了 中央氣象署 API (CWA API)、Google Drive / Sheets 雲端資料庫、Matplotlib 圖表繪製 以及 獨立背景執行緒 (Threading),實現了資料定時自動抓取、雲端儲存與介面無卡頓刷屏。

以下為本程式的核心功能與完整動作流程說明:

核心功能說明

  1. 自動氣象 API 爬蟲

    • 自動連線中央氣象署 API,抓取全台 22 縣市的最新氣候資料(包含天氣狀況、最低溫、最高溫、降雨機率及發布時間)。

  2. Google 雲端試算表自動同步

    • 使用 gspread 與 Google 服務帳號(Service Account)金鑰,將抓取到的資料自動追加(Append)至雲端試算表 Taiwan_Weather_Data。

  3. 智慧標頭與流水號防錯

    • 自動檢測 Google Sheet 第一列是否存在標準欄位標頭,若遺失會自動補齊。

    • 自動搜尋現有資料的最大 ID 並順延編號,確保資料流水號不重複、不覆蓋舊紀錄。

  4. 10 分鐘自動定時器與倒數顯示

    • 內建每秒遞減的倒數計時器(顯示於右上角),每 10 分鐘(600 秒)自動發起全自動「抓取 API ➔ 寫入雲端 ➔ 刷屏繪圖」循環。

  5. 多執行緒(Threading)防卡頓介面

    • 將耗時的「網路 API 抓取」與「Google Drive 雲端讀寫」放進背景執行緒處理。

    • 將「Matplotlib 繪圖」與「GUI UI 更新」留在主執行緒 (Main Thread),完全避免跨執行緒崩潰與視窗無回應(卡死)問題。

  6. 動態選單與縣市切換

    • 提供熱門縣市快捷按鈕(臺中市、臺北市、高雄市、南投縣)及下拉式選單(ttk.Combobox),可自由切換欲檢視的縣市 24 小時氣溫變化趨勢圖。

程式動作流程(Execution Flow)

[程式啟動] 
    │
    ├── 1. 初始化 GUI 介面與控制元件 (預設選取「臺中市」)
    │
    ├── 2. 啟動倒數計時器迴圈 (_start_timer_loop) ➔ 每秒更新顯示
    │
    └── 3. 觸發第一次資料載入 (on_query_city: force_fetch_api=True)
            │
            ▼
┌─────────────────────────────────────────────────────────────┐
│ 步驟 A:背景執行緒 (_async_load_data)                        │
│ 1. 連線中央氣象署 API 取得最新天氣數據                         │
│ 2. 檢查 Google Sheet 標頭,計算下一個正值 ID                   │
│ 3. 將 22 筆縣市資料追加寫入 Google 雲端試算表                  │
│ 4. 讀取 Google Sheet 近 24 小時資料並過濾特定縣市 (Pandas)     │
└──────────────────────────────┬──────────────────────────────┘
                               │ (將整理好的 DataFrame 送回主執行緒)
                               ▼
┌─────────────────────────────────────────────────────────────┐
│ 步驟 B:主執行緒渲染 (_update_canvas_main_thread)            │
│ 1. 銷毀舊有 Canvas 與圖表物件 (釋放記憶體)                   │
│ 2. 繪製最高溫、最低溫、平均溫的 Matplotlib 折線圖            │
│ 3. 將圖表渲染至 Tkinter 視窗中央                              │
│ 4. 更新下拉選單之全台縣市清單,並顯示完成時間                │
└──────────────────────────────┬──────────────────────────────┘
                               │
                               ▼
[持續運作狀態]
    ├── 使用者點擊按鈕/選單 ➔ 重新篩選該縣市並繪圖 (可點擊「立即抓取」強制刷新)
    └── 倒數 10 分鐘歸零 ➔ 自動重新觸發【步驟 A ➔ 步驟 B】循環

模組架構總覽

  • AutoRefreshWeatherApp.__init__:系統初始化、介面排版、啟動首次 API 抓取與 Timer 迴圈。

  • _start_timer_loop:每秒遞減倒數,秒數歸零時觸發定時刷新。

  • _async_load_data:背景工作主體,安全隔離網路連線操作。

  • _fetch_cwa_api_and_write_sheets:解析氣象 JSON 並寫入 Google Sheets。

  • _prepare_data_from_sheets:使用 Pandas 進行資料清洗、時間排序與平均溫算術運算。

  • _update_canvas_main_thread:主執行緒繪圖渲染核心,產出嵌入式 Matplotlib Canvas。


要取得 Google 服務帳號(Service Account)憑證 JSON 檔,需要透過 Google Cloud Console 進行申請與下載。

服務帳號是專為 Python、Node.js 等自動化程式設計的獨立帳號,能讓您的程式在沒有使用者手動登入瀏覽器的情況下,自動存取 Google Drive 或 Google Sheets。

取得 Google 服務帳號憑證(JSON)四步驟

步驟 1:建立 Google Cloud 專案

  1. 開啟 Google Cloud Console 並登入您的 Google 帳號。

  2. 點擊頁面頂部的專案下拉選單,點擊 「新增專案 (New Project)」。

  3. 輸入專案名稱(例如:weather-monitor),然後點擊 「建立 (Create)」。

步驟 2:啟用必要的 API 服務

  1. 在左側選單點擊 「API 和服務 (APIs & Services)」 -> 「庫 (Library)」。

  2. 搜尋 Google Drive API,點進去後點擊 「啟用 (Enable)」。

  3. 再次返回 API 庫,搜尋 Google Sheets API(若需使用 Google 試算表),點擊 「啟用 (Enable)」。

步驟 3:建立服務帳號並下載 JSON 憑證

  1. 點擊左側選單 「IAM 和管理員 (IAM & Admin)」 -> 「服務帳號 (Service Accounts)」。

  2. 點擊頂部的 「+ 建立服務帳號 (+ CREATE SERVICE ACCOUNT)」。

  3. 輸入帳號名稱(例如 weather-bot),點擊 「建立並繼續 (Create and Continue)」。

  4. 角色(Role)可暫時選擇 「編輯者 (Editor)」 或直接跳過,點擊 「完成 (Done)」。

  5. 在服務帳號列表中找到剛剛建立的帳號,點擊右側的 「⋮(動作)」 -> 「管理金鑰 (Manage keys)」。

  6. 點擊 「新增金鑰 (Add Key)」 -> 「建立金鑰 (Create new key)」。

  7. 金鑰類型選擇 「JSON」,點擊 「建立 (Create)」。

  8. 瀏覽器會自動下載一個 .json 檔案(檔名類似 your-project-123456-abcdef.json)。

💡 提示:若要在 Python 中使用,您可以將下載的 JSON 內容複製貼入程式中的 SERVICE_ACCOUNT_INFO 字典,或是重新命名為 service_account.json 放置於專案資料夾下。

步驟 4:將 Google 試算表共用給服務帳號

  1. 打開剛下載的 JSON 檔案,找到裡面的 client_email 欄位(格式為 xxxx@xxxx.iam.gserviceaccount.com)。

  2. 前往您在 Google Drive 建立的 Taiwan_Weather_Data 試算表。

  3. 點擊右上角的 「共用」 按鈕。

  4. 將該 client_email 貼入,並將權限設為 「編輯者」,點擊 「傳送」。


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 = "CWA-40C25F4FF-12124-4250-B9D9-37353AAE17DBF"  # 請填入開頭為 CWA- 的有效金鑰

TELEGRAM_BOT_TOKEN = "88454708831:AAEuRreXFRyv1m--jGK7OC4n8MiEsW6XqpMT0"

TELEGRAM_CHAT_ID = "749652181469"

SPREADSHEET_NAME = "Taiwan_Weather_Data"


SERVICE_ACCOUNT_INFO = {

  "type": "service_account",

  "project_id": "weather-monitor-513909",

  "private_key_id": "bf332fe535a6025232b14ec734f109d0da9d4eddc",

  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvgIBADANBgkqhkiG9w0BAQEFAASCBKgwggSkAgEAAoIBAQDLS3P8oWzQlrAP\nP7wBMTspZK+0o8ISNOyyoJ9FWBeemWKMnUH+MZZrsNfGn0n/9ad3xhBfstysqnBC\nwAqKo8tLgcgjBBpN8Ff03xPwi02xOEB6+axnPmbblqzrRC9UHa08y9QHVoVMHZ1w\nbLykB1sBxtJOMWUatvxYL7vv6PGr2hJSXuBxPvaWotrrNW+E/F5ItUNeWyujMLJR\nVu94rEowIaeKbjwKPg3C16o6QdWquFHu5aeij7rxXckESWjKYW0T9YcUtf8mx/+O\n5buObPbUSUMDiKi2jxRFdETtSy3x+Wl6CzYGfBSS1mCQ+13GtpS285kVp2lXsa8A9\nNS+dYPQRAgMBAAECggEADksNCdlQpP6/PKc6wOu08OVZHC56TdzmOm3kHNRDft1i\naigBVI2oeYRau2aGJaagImHs00moTcSC9ZJ6e0JmX/XtPBcFAW9RQ9BuXrZwnTHr\nCKpmtnCS+0Pl/vmRiUy32iMrkmOB2Jjeb0uR2Q+MXgTG5JhG/fcgZSFr8z/KVEUO4\n9rLmU+2rf7QcPNlM3okfgAqKZUr4OFQccPA0vtgW9cmeg5+eV1fZHaGW91Q0mNk3\nM3EZhjtTVs763GmBbMO5ZNnvFYKWIRmXfRow8f24ykNRqHFpPPIvNCcwKo7kPkX3\nl4kz4Gs79+csPqLkRrY7g2gcUbXMiWgfefN6lpqZBQKBgQDr5lS8D20i4JmJ1dS+\nz01AlAkg+kp8R9W4FqJCY93I1aZDcvRTFrqQzXR4G+Yk7EwyRs1RayDdIC/y0nHG\nGG8OjMtxW9DrmKOiSEcNoCKfdgPX5mOfrOxrjagk/geYuqsGb7BB9DZSRd2A4qBK+\nmSb/AUyr9IvBZrNA2PB0ImiYNQKBgQDcnek44P5KsmCaPhYMki67adxLEapYaz2Z\ninAiZ4A4YlNCzHNs6zmR2Yb/dCUbtHlQs6hiIeodEQnkYYoTgTvX8UvLrNYt+jPoX\n31chl/iLHzTE5VUOPYR7gJeP/xhQVAqK16zjzRqo+crH+P0DVCJ/lzA4VoHOKrT2\nVqUuCZs/7QKBgQCQf3ZH++sOAiYKmEDI4JKhjAxLdwVJO3deW/7oS+P+sWA20KN4\ngeOl7zP4np5K37OM2HHL/aMT0DOOufmXLkBY2UWFGojOx6D0NvHPFtgedVvFVAqS\nNIpdF1h30GM22ptDt0YiU789m7IFAm748TdB3LnqWtGAljgW8MvZV3paQQKBgQDP\nKWkrA/tFYSLdM5e4zjwbuFgoFbqDtI/vcdPCOrZoO8w7WYi5qlgw4LanggSmc13H\n4UqQPZ5vNoxHnZtwljTeb3Wz8F/AwdjGOe6zmOTGntMKYN7xtEqHPkUT7lKCe6g6\noDO05ZA3Q/hqtx++1G5gqTOac4EIe4naF5qcFGE4MQKBgDvNtoReH7iLHW7/FecF\nMB2/8k3Q9sNStQdnOe5yHZkvcAiFzCwD+LaK+VXDi1o2KQloE3Ih+0TfvEz0QRiu\nNP1bH97yBV/hqYyTgfvc1tXsR9MhJQdioPCJbGNsHLOiiMNCM3t6Vw75yaipwJOU\nj8spBjjqlVjW3TzMpRh7qDki\n-----END PRIVATE KEY-----\n",

  "client_email": "weather-bot@weather-monitor-5131909.iam.gserviceaccount.com",

  "client_id": "10683561542120520182272",

  "auth_uri": "https://accounts.google.com/o/oauth2/auth",

  "token_uri": "https://oauth2.googleapis.com/token",

  "auth_provider_x5109_cert_url": "https://www.googleapis.com/oauth2/v1/certs",

  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x5109/weather-bot%40weather-monitor-5113909.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 主程式

# --------------------------------------------------

class AutoRefreshWeatherApp(tk.Tk):

    def __init__(self):

        super().__init__()


        self.title("台灣城市 24 小時氣溫監控 (自動抓取 API / 10分鐘定期刷屏)")

        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                      # 保持單一 Figure 物件參照


        self._create_widgets()


        # 啟動時:立即抓取 API,並預設載入「臺中市」

        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="🔄 立即抓取並刷新", 

            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 並刷新「{self.current_city}」...", 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. 非同步資料抓取 (純網路操作放 Thread,GUI繪圖移回 Main Thread)

    # --------------------------------------------------

    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 並讀取" 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 抓取與雲端試算表資料整理,不建立任何 GUI/Matplotlib 物件"""

        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)


            # 解析與準備 Pandas DataFrame

            filtered_df, all_cities = self._prepare_data_from_sheets(sheet, city_name)

            

            # 將整理好的 DataFrame 資料安全送回 Main Thread 進行繪圖

            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. API 抓取與寫入核心 (防止 ID 重複與補齊標頭)

    # --------------------------------------------------

    def _fetch_cwa_api_and_write_sheets(self, sheet):

        """從中央氣象署抓取資料並寫入 Google 試算表"""

        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 試算表。")


    # --------------------------------------------------

    # 6. 純資料解析 (非 GUI 操作)

    # --------------------------------------------------

    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


    # --------------------------------------------------

    # 7. 主執行緒安全的繪圖與 UI 渲染

    # --------------------------------------------------

    def _update_canvas_main_thread(self, filtered_df: pd.DataFrame, city_name: str, all_cities: list):

        """完全在 GUI 主執行緒 (Main Thread) 建立 Figure 並渲染至 Canvas"""

        # 銷毀舊 Canvas 視窗元件

        if self.canvas_widget:

            self.canvas_widget.destroy()

            self.canvas_widget = None


        # 顯式關閉舊的 Figure 釋放記憶體

        if self.fig:

            plt.close(self.fig)


        # 在 Main Thread 創建 Matplotlib Figure

        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()


        # 將 Figure 繪製至 Tkinter Canvas

        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}」圖表 (更新時間: {update_time_str})", 

            foreground="green"

        )


    def _handle_error(self, err_msg: str):

        self.lbl_status.config(text=f"❌ 刷屏失敗: {err_msg} (詳細請看 Terminal)", foreground="red")



# --------------------------------------------------

# 8. 程式啟動

# --------------------------------------------------

if __name__ == "__main__":

    app = AutoRefreshWeatherApp()

    app.mainloop()


結合桌面視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統

視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統 這份 gui_app_timer.py 程式是一個 結合桌面視窗介面 (GUI)、雲端資料庫與社群通訊推播的自動化氣象監控系統 。 整體架構採用 事件驅動 (Event-Driven) 與 多執行緒非同步設計...