Update app.py

app.py

@@ -1,96 +1,130 @@
-from fastapi import FastAPI, HTTPException, Query
-from starlette.responses import JSONResponse
-from obspy.clients.fdsn import Client
-from obspy.core.utcdatetime import UTCDateTime
-import logging
-from datetime import date
+# app.py
 
-# Configure logging
-logging.basicConfig(level=logging.INFO)
+import gradio as gr
+import html
+import requests
+import re
+import xml.etree.ElementTree as ET
+from datetime import datetime, timedelta, timezone
 
-# Initialize the FastAPI app
-app = FastAPI(
-    title="Dynamic Earthquake Catalog API",
-    description="A robust API to fetch earthquake data from the IRIS FDSN web service using URL parameters.",
-    version="2.1.0",
-)
+# ===============================================================
+# 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": "金門縣"
+}
 
-# Initialize the IRIS FDSN client
-try:
-    client = Client("IRIS")
-    logging.info("Successfully initialized IRIS FDSN client.")
-except Exception as e:
-    client = None
-    logging.error(f"Could not initialize IRIS FDSN client on startup: {e}")
-
-@app.get("/")
-def greet_json():
-    """
-    A simple 'Hello World' endpoint.
+# ===============================================================
+# 主要功能：讀取並過濾地震速報
+# ===============================================================
+def get_disaster_warnings() -> str:
     """
-    return {"Hello": "World!", "message": "Go to /docs to try the dynamic earthquake endpoint.  cwadayi \n https://cwadayi-python-app.hf.space/earthquakes?start_date=2024-07-01&end_date=2024-07-07&min_magnitude=5.0"}
+    Fetches and formats Earthquake Early Warnings from the CWA PWS feed.
 
-@app.get("/earthquakes")
-async def get_earthquake_catalog(
-    start_date: date = Query(default="2024-01-01", description="Start date in YYYY-MM-DD format"),
-    end_date: date = Query(default="2024-01-07", description="End date in YYYY-MM-DD format"),
-    min_magnitude: float = Query(default=4.5, description="Minimum earthquake magnitude (e.g., 5.5)", gt=0, le=10)
-):
-    """
-    Fetches earthquake data based on date and magnitude parameters provided in the URL.
+    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.
     """
-    if not client:
-        raise HTTPException(status_code=503, detail="The IRIS FDSN client is not available.")
+    # 使用時間戳避免快取
+    timestamp = int(datetime.now().timestamp())
+    feed_url = f"https://cbs.tw/files/rssatomfeed.xml?_={timestamp}"
+    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'
+    }
     
-    if start_date > end_date:
-        raise HTTPException(status_code=400, detail="The start_date cannot be after the end_date.")
-
     try:
-        starttime = UTCDateTime(start_date)
-        endtime = UTCDateTime(end_date)
+        response = requests.get(feed_url, headers=headers, timeout=15)
+        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)
 
-        logging.info(f"Fetching events from {starttime} to {endtime} with min magnitude {min_magnitude}.")
-        catalog = client.get_events(
-            starttime=starttime,
-            endtime=endtime,
-            minmagnitude=min_magnitude,
-        )
-        logging.info(f"Found {len(catalog)} events. Now processing.")
+        # 取得台灣時區
+        tw_tz = timezone(timedelta(hours=8))
+        current_time_str = datetime.now(tw_tz).strftime('%Y-%m-%d %H:%M:%S')
+        
+        markdown_output = f"## 最新地震速報 (僅顯示來自：中央氣象署)\n---\n*最後更新時間：{current_time_str} (GMT+8)*\n\n"
+        messages_found = 0
 
-        earthquake_list = []
-        for event in catalog:
-            try:
-                origin = event.preferred_origin()
-                magnitude = event.preferred_magnitude()
+        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 not origin or not magnitude:
-                    continue
+            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 "無發布時間"
 
-                # --- Key Change Start ---
-                # Safely get the description using getattr. This avoids errors if the 'descriptions'
-                # attribute doesn't exist, by providing a default empty list [].
-                place = "N/A"
-                descriptions = getattr(event, 'descriptions', [])
-                if descriptions:
-                    place = descriptions[0].text
-                # --- Key Change End ---
+                markdown_output += f"### {html.escape(title)}\n"
+                markdown_output += f"**發布時間:** {html.escape(published_time)}\n\n"
 
-                earthquake_list.append({
-                    "time": origin.time.isoformat(),
-                    "latitude": origin.latitude,
-                    "longitude": origin.longitude,
-                    "depth_km": origin.depth / 1000.0 if origin.depth is not None else None,
-                    "magnitude": magnitude.mag,
-                    "magnitude_type": magnitude.magnitude_type,
-                    "place": place,
-                })
-            except Exception as e:
-                logging.warning(f"Skipping event due to unexpected processing error: {e}")
-                continue
-        
-        logging.info(f"Successfully processed {len(earthquake_list)} events.")
-        return JSONResponse(content={"earthquakes": earthquake_list})
+                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 markdown_output + "目前沒有最新的地震速報。"
+            
+        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:
-        logging.error(f"A major error occurred during the request: {e}")
-        raise HTTPException(status_code=500, detail=f"An internal error occurred: {e}")
+        return f"讀取訊息時發生未預期的錯誤: {e}"
+
+# ===============================================================
+# 建立 Gradio 介面
+# ===============================================================
+with gr.Blocks(title="台灣地震速報") as app:
+    gr.Markdown("## 台灣地震速報")
+    gr.Markdown("自動讀取最新的地震速報，並過濾只顯示由「中央氣象署」發布的內容。")
+    output_markdown = gr.Markdown()
+    app.load(get_disaster_warnings, outputs=output_markdown, every=60) # 每60秒自動更新一次
+
+
+# 啟動 Gradio 網頁介面
+if __name__ == "__main__":
+    app.launch()