app.py

import gradio as gr
import html
import requests
import re
import xml.etree.ElementTree as ET

# ===============================================================
# Geocode 與縣市名稱的對照字典
# ===============================================================
GEOCODE_MAP = {
    "10002": "宜蘭縣", "10004": "新竹縣", "10005": "苗栗縣",
    "10007": "彰化縣", "10008": "南投縣", "10009": "雲林縣",
    "10010": "嘉義縣", "10013": "屏東縣", "10014": "臺東縣",
    "10015": "花蓮縣", "10016": "澎湖縣", "10017": "基隆市",
    "10018": "新竹市", "10020": "嘉義市", "63": "臺北市",
    "64": "高雄市", "65": "新北市", "66": "臺中市",
    "67": "臺南市", "10003": "桃園市", "09007": "連江縣",
    "09020": "金門縣"
}

# ===============================================================
# 主要功能：讀取並過濾地震速報
# ===============================================================
def get_disaster_warnings() -> str:
    """
    Fetches and formats Earthquake Early Warnings from the CWA PWS feed.

    This function retrieves the latest disaster warnings, filters them to show
    only 'Earthquake Early Warning' (地震速報) alerts issued by the CWA, and
    formats the output in Markdown, including affected counties.
    """
    feed_url = "https://cbs.tw/files/rssatomfeed.xml"
    keywords = ["中央氣象署", "氣象署"]
    
    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',
        'Cache-Control': 'no-cache',
        'Pragma': 'no-cache'
    }
    
    try:
        response = requests.get(feed_url, headers=headers, timeout=10)
        response.raise_for_status()
        
        xml_content = response.content.decode('utf-8')
        xml_content = re.sub(' xmlns="[^"]+"', '', xml_content, count=1)
        root = ET.fromstring(xml_content)
        
        markdown_output = f"## 最新地震速報 (僅顯示來自：中央氣象署)\n---\n"
        messages_found = 0

        for entry in root.findall('entry'):
            summary_text = ''
            for tag_name in ['summary', 'description']:
                content_tag = entry.find(tag_name)
                if content_tag is not None and content_tag.text:
                    summary_text = content_tag.text
                    break
            
            title_tag = entry.find('title')
            title = title_tag.text if title_tag is not None else ""
            
            # --- (修改處) ---
            # 新增篩選條件：標題必須包含 "地震速報"
            is_earthquake_alert = "地震速報" in title
            is_from_cwa = any(keyword in summary_text for keyword in keywords)

            if is_earthquake_alert and is_from_cwa:
                messages_found += 1
                
                sent_tag = entry.find('sent')
                published_time = sent_tag.text if sent_tag is not None else "無發布時間"

                markdown_output += f"### {html.escape(title)}\n"
                markdown_output += f"**發布時間:** {html.escape(published_time)}\n\n"

                # 因為已確認是地震速報，直接處理影響範圍
                area_tag = entry.find('area')
                if area_tag is not None:
                    area_desc_tag = area_tag.find('areadesc')
                    area_desc = area_desc_tag.text if area_desc_tag is not None else ''

                    affected_locations = []
                    for geocode_tag in area_tag.findall('geocode'):
                        value_tag = geocode_tag.find('value')
                        if value_tag is not None:
                            code = value_tag.text
                            location_name = GEOCODE_MAP.get(code, f"未知代碼({code})")
                            affected_locations.append(location_name)
                    
                    area_details = ""
                    if affected_locations:
                        area_details = f" ({'、'.join(affected_locations)})"
                    
                    full_area_info = f"{html.escape(area_desc)}{area_details}"
                    markdown_output += f"**影響範圍:** {full_area_info}\n\n"
                
                markdown_output += f"**內容:**\n{html.escape(summary_text)}\n\n"
                markdown_output += "---\n"

        if messages_found == 0:
            return "目前沒有最新的地震速報。"
            
        return markdown_output
    except requests.exceptions.RequestException as e:
        return f"網路請求失敗，無法取得資料: {e}"
    except ET.ParseError as e:
        return f"XML 資料解析失敗，請稍後再試: {e}"
    except Exception as e:
        return f"讀取訊息時發生未預期的錯誤: {e}"

# ===============================================================
# 建立 Gradio 介面
# ===============================================================
with gr.Blocks() as app:
    gr.Markdown("## 地震速報")
    gr.Markdown("自動讀取最新的地震速報，並過濾只顯示由「中央氣象署」發布的內容。")
    output_markdown = gr.Markdown()
    app.load(get_disaster_warnings, outputs=output_markdown, api_name="get_disaster_warnings")


# 啟動 Gradio 網頁介面
if __name__ == "__main__":
    app.launch(mcp_server=True, show_error=True)