import gradio as gr import requests import os from datetime import datetime, timedelta # --- 從 Hugging Face Secrets 讀取 API KEY --- CWA_API_KEY = os.getenv("CWA_API_KEY") # --- API 設定 --- CWA_API_BASE_URL = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/" # (DATASET_IDS 和 TAIWAN_LOCATIONS 的定義與前一版完全相同) DATASET_IDS = {"顯著有感地震": "E-A0015-001", "小區域有感地震": "E-A0016-001", "一般天氣預報": "F-C0032-001", "自動氣象站-氣象觀測": "O-A0001-001", "自動雨量站-雨量觀測": "O-A0002-001", "現在天氣觀測報告": "O-A0003-001"} TAIWAN_LOCATIONS = ["臺北市", "新北市", "基隆市", "桃園市", "新竹縣", "新竹市", "苗栗縣", "臺中市", "彰化縣", "南投縣", "雲林縣", "嘉義縣", "嘉義市", "臺南市", "高雄市", "屏東縣", "宜蘭縣", "花蓮縣", "臺東縣", "澎湖縣", "金門縣", "連江縣"] # ============================================================================== # API 請求與資料解析函式 (與前一版完全相同) # ============================================================================== def make_api_request(data_id, params=None): if not CWA_API_KEY: raise ValueError("錯誤：尚未在 Hugging Face Space 的 Secrets 中設定 CWA_API_KEY。") api_url = f"{CWA_API_BASE_URL}{data_id}" base_params = {"Authorization": CWA_API_KEY, "format": "JSON"} if params: base_params.update(params) response = requests.get(api_url, params=base_params) if response.status_code != 200: raise ConnectionError(f"API 請求失敗，狀態碼：{response.status_code}\n錯誤訊息：{response.text}") data = response.json() if not data.get('success'): raise ValueError("API 回應不成功或資料格式有誤。") return data.get('records', {}) def parse_earthquake_data(records, query_type): earthquakes = records.get('earthquake', []) if not earthquakes: return f"API 請求成功，在指定的時間範圍內沒有「{query_type}」的紀錄。" result = f"查詢到 {len(earthquakes)} 筆「{query_type}」紀錄 (最多顯示前5筆):\n" result += "========================================\n" for eq in earthquakes[:5]: info = eq['earthquakeInfo'] result += (f"報告內容：{eq.get('reportContent', 'N/A')}\n" f"發生時間：{info.get('originTime', 'N/A')}\n" f"震央位置：{info.get('epiCenter', {}).get('location', 'N/A')}\n" f"芮氏規模：{info.get('magnitude', {}).get('magnitudeValue', 'N/A')}\n" f"地震深度：{info.get('depth', {}).get('value', 'N/A')} 公里\n" "----------------------------------------\n") return result def parse_general_forecast_data(records, location): location_records = records.get('location', []) if not location_records: return f"找不到「{location}」的一般天氣預報資料。" weather_elements = location_records[0].get('weatherElement', []) forecasts = {} for element in weather_elements: element_name = element['elementName'] for time_period in element['time']: time_key = f"{time_period['startTime']} ~ {time_period['endTime']}" if time_key not in forecasts: forecasts[time_key] = {} value = time_period['parameter']['parameterName'] unit = time_period['parameter'].get('parameterUnit', '') forecasts[time_key][element_name] = f"{value} {unit}".strip() result = f"📍 **{location}** 一般天氣預報\n" + "---------------------------------\n" for time_period, values in forecasts.items(): result += (f"**預報時段：** {time_period}\n" f" - 天氣現象 (Wx): {values.get('Wx', 'N/A')}\n" f" - 最高溫度 (MaxT): {values.get('MaxT', 'N/A')}\n" f" - 最低溫度 (MinT): {values.get('MinT', 'N/A')}\n" f" - 舒適度 (CI): {values.get('CI', 'N/A')}\n" f" - 降雨機率 (PoP): {values.get('PoP', 'N/A')}\n\n") return result def parse_observation_data(records, query_type, location): location_records = records.get('location', []) if not location_records: return f"在「{location}」找不到任何「{query_type}」的觀測站資料。" result = f"📍 **{location}** - **{query_type}** (僅列出部分站點)\n" + "---------------------------------\n" if "站" in query_type: for station in location_records[:5]: result += f"**觀測站：{station['stationId']} {station.get('locationName', '')}**\n" elements = {elem['elementName']: elem['elementValue'] for elem in station['weatherElement']} if "氣象觀測" in query_type: result += (f" - 溫度: {elements.get('TEMP', 'N/A')} °C\n" f" - 濕度: {float(elements.get('HUMD', '-1'))*100:.1f} %\n" f" - 風速/風向: {elements.get('WDSD', 'N/A')} m/s / {elements.get('WDIR', 'N/A')}°\n") elif "雨量觀測" in query_type: result += (f" - 10分鐘雨量: {elements.get('RAIN', 'N/A')} mm\n" f" - 1小時雨量: {elements.get('HOUR_1', 'N/A')} mm\n" f" - 24小時雨量: {elements.get('HOUR_24', 'N/A')} mm\n") result += "\n" elif "現在天氣" in query_type: for loc in location_records: result += f"**區域：{loc.get('locationName', '')}**\n" elements = {elem['elementName']: elem['elementValue'] for elem in loc['weatherElement']} result += (f" - 天氣現象: {elements.get('Weather', 'N/A')}\n" f" - 溫度: {elements.get('Temperature', 'N/A')} °C\n" f" - 濕度: {float(elements.get('RelativeHumidity', '-1'))*100:.1f} %\n\n") return result # ============================================================================== # Gradio 觸發函式 # ============================================================================== # --- **這裡是修改的關鍵** --- def handle_earthquake_query(query_type, start_date_str, end_date_str): try: datetime.strptime(start_date_str, "%Y-%m-%d") datetime.strptime(end_date_str, "%Y-%m-%d") params = { "timeFrom": f"{start_date_str}T00:00:00", "timeTo": f"{end_date_str}T23:59:59", "limit": 10, "sort": "OriginTime" # 強制API依時間篩選與排序 } records = make_api_request(DATASET_IDS[query_type], params) return parse_earthquake_data(records, query_type) except ValueError: return "日期格式錯誤，請輸入 YYYY-MM-DD 格式。" except Exception as e: return str(e) def handle_general_forecast_query(location): try: records = make_api_request(DATASET_IDS["一般天氣預報"], {"locationName": location}) return parse_general_forecast_data(records, location) except Exception as e: return str(e) def handle_observation_query(query_type, location): try: records = make_api_request(DATASET_IDS[query_type], {"locationName": location}) return parse_observation_data(records, query_type, location) except Exception as e: return str(e) # ============================================================================== # Gradio UI 介面 (與前一版完全相同) # ============================================================================== with gr.Blocks(title="CWA 開放資料查詢平台") as demo: gr.Markdown("# 🇹🇼 CWA 中央氣象署開放資料查詢平台") with gr.Tabs(): with gr.TabItem("一般天氣預報(縣市)"): gr.Markdown("### 查詢各縣市天氣預報") fc_location_dropdown = gr.Dropdown(label="請選擇縣市", choices=TAIWAN_LOCATIONS, value="臺北市") fc_btn = gr.Button("查詢天氣") fc_output = gr.Textbox(lines=15, label="查詢結果") fc_btn.click(fn=handle_general_forecast_query, inputs=fc_location_dropdown, outputs=fc_output) with gr.TabItem("即時觀測"): gr.Markdown("### 查詢即時觀測資料") with gr.Row(): obs_type_dropdown = gr.Dropdown(label="觀測類型", choices=["自動氣象站-氣象觀測", "自動雨量站-雨量觀測", "現在天氣觀測報告"], value="自動氣象站-氣象觀測") obs_location_dropdown = gr.Dropdown(label="縣市", choices=TAIWAN_LOCATIONS, value="臺北市") obs_btn = gr.Button("查詢觀測資料") obs_output = gr.Textbox(lines=15, label="查詢結果") obs_btn.click(fn=handle_observation_query, inputs=[obs_type_dropdown, obs_location_dropdown], outputs=obs_output) with gr.TabItem("地震查詢"): gr.Markdown("### 查詢地震報告") eq_type_dropdown = gr.Dropdown(label="查詢類型", choices=["顯著有感地震", "小區域有感地震"], value="顯著有感地震") with gr.Row(): start_date_input = gr.Textbox(label="開始日期 (格式 YYYY-MM-DD)", value=(datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")) end_date_input = gr.Textbox(label="結束日期 (格式 YYYY-MM-DD)", value=datetime.now().strftime("%Y-%m-%d")) eq_btn = gr.Button("查詢地震") eq_output = gr.Textbox(lines=20, label="查詢結果") eq_btn.click(fn=handle_earthquake_query, inputs=[eq_type_dropdown, start_date_input, end_date_input], outputs=eq_output) gr.Markdown("
資料來源：[交通部中央氣象署](https://opendata.cwa.gov.tw/index)") demo.launch()