app.py

import gradio as gr
import feedparser
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(dummy_input=None) -> str:
    """
    Fetches, filters, and formats public disaster warnings from the CWA PWS feed.

    This function retrieves the latest disaster warnings from the official feed at
    https://cbs.tw/files/rssatomfeed.xml, filters them to show only alerts
    issued by the Central Weather Administration (CWA), and formats the output
    in Markdown. For earthquake alerts, it performs additional parsing to
    extract geocodes and translate them into human-readable county names.
    """
    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
            
            if any(keyword in summary_text for keyword in keywords):
                messages_found += 1
                
                title_tag = entry.find('title')
                title = title_tag.text if title_tag is not None else "無標題"

                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"

                if "地震" in title:
                    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}"

# ===============================================================
# 功能二：Feed 結構偵測器
# ===============================================================
def inspector(dummy_input=None):
    """
    Investigates the CWA PWS feed and reports its raw data structure for debugging.

    This function connects to the official feed at https://cbs.tw/files/rssatomfeed.xml,
    parses it, and generates a detailed report on the first two entries. The report
    includes all available data fields (keys), their data types, and a value preview.
    Its purpose is to help developers understand the feed's structure for accurate parsing.
    """
    feed_url = "https://cbs.tw/files/rssatomfeed.xml"
    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=15)
        response.raise_for_status()
        feed = feedparser.parse(response.content)
        if not feed.entries:
            return "Error: Successfully downloaded, but the feed contains no entries."
        entries_to_inspect = feed.entries[:2]
        report = f"## Debug Report: Feed Structure (Total: {len(feed.entries)}, Showing first {len(entries_to_inspect)})\n\n"
        for i, entry in enumerate(entries_to_inspect):
            report += f"---\n\n### Entry #{i + 1}\n\n"
            keys = list(entry.keys())
            report += "This entry contains the following **Keys** and their **Data Types**:\n\n"
            for key in sorted(keys):
                value = entry[key]
                value_type = type(value).__name__
                report += f"* **Key:** `{key}`\n"
                report += f"  * **Type:** `{value_type}`\n"
                try:
                    display_value = str(value)[:200]
                    report += f"  * **Value Preview:** `{html.escape(display_value)}`\n\n"
                except:
                    report += f"  * **Value Preview:** `Could not be displayed`\n\n"
        return report
    except Exception as e:
        return f"An error occurred during the inspection: {e}"

# ===============================================================
# 建立 Gradio 介面
# ===============================================================
warnings_app = gr.Interface(
    fn=get_disaster_warnings,
    inputs=gr.Button("Click to Update Latest Warnings"),
    outputs=gr.Markdown(),
    title="Disaster Warnings",
    description="Fetches the latest disaster warnings and filters for content published by the 'Central Weather Administration' (CWA).",
    api_name="get_disaster_warnings"
)

inspector_app = gr.Interface(
    fn=inspector,
    inputs=gr.Button("Start Inspection"),
    outputs=gr.Markdown(),
    title="Feed Structure Inspector",
    description="This tool fetches the first two entries from the specified feed and reports their internal structure to identify the correct field names for development.",
    api_name="inspector"
)

app = gr.TabbedInterface([warnings_app, inspector_app], ["Disaster Warnings", "Feed Inspector"])

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