Update app.py

app.py

@@ -1,70 +1,51 @@
 import os
 import io
-import json
-from datetime import datetime, timedelta, timezone
-
+import tempfile
 import requests
 import pandas as pd
 import matplotlib.pyplot as plt
 import gradio as gr
-
+from datetime import datetime, timedelta, timezone
 
 # -----------------------------
-# 時區：台北 (UTC+8)
+# 台北時區 (UTC+8)
 # -----------------------------
 TAIPEI_TZ = timezone(timedelta(hours=8))
 
-def now_taipei():
-    return datetime.now(TAIPEI_TZ)
-
 def fmt_dt(dt: datetime) -> str:
     return dt.strftime("%Y-%m-%dT%H:%M:%S")
 
-
-# -----------------------------
-# 快速時間範圍
-# -----------------------------
-def set_time_range(hours: int | None = None, days: int | None = None):
-    """
-    依台北時間回傳 (timeFrom, timeTo) 字串（yyyy-MM-ddTHH:mm:ss）
-    """
-    now = now_taipei()
+def set_time_range(hours=None, days=None):
+    """依台北時間回傳 (timeFrom, timeTo)"""
+    now = datetime.now(TAIPEI_TZ)
     if hours is not None:
         t_from = now - timedelta(hours=hours)
     elif days is not None:
         t_from = now - timedelta(days=days)
     else:
-        t_from = now - timedelta(days=3)  # 預設最近 3 天
+        t_from = now - timedelta(days=3)
     return fmt_dt(t_from), fmt_dt(now)
 
-
 # -----------------------------
-# 呼叫 CWA E-A0015-001 API
+# 取 API 資料
 # -----------------------------
 API_URL = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001"
 
-def fetch_reports(time_from: str, time_to: str) -> dict:
-    """
-    以環境變數 CWA_API_KEY 做授權參數呼叫 API，回傳 JSON 物件（dict）
-    """
+def fetch_reports(time_from, time_to):
     api_key = os.getenv("CWA_API_KEY", "").strip()
     if not api_key:
-        raise RuntimeError("環境變數 CWA_API_KEY 未設定。請在 Space Secrets 設定。")
-
+        raise RuntimeError("請在環境變數設定 CWA_API_KEY")
     params = {
         "Authorization": api_key,
         "timeFrom": time_from,
-        "timeTo": time_to,
-        # 其餘參數保持預設；排序改由本地處理，以避免介面差異
+        "timeTo": time_to
     }
-    resp = requests.get(API_URL, params=params, timeout=30)
-    resp.raise_for_status()
-    return resp.json()
-
+    r = requests.get(API_URL, params=params, timeout=30)
+    r.raise_for_status()
+    return r.json()
 
 # -----------------------------
-# 解析 JSON → pandas.DataFrame
-# 只保留必要欄位（不含 area / station 等）
+# 解析 JSON → DataFrame
 # -----------------------------
 def safe_get(d, *keys, default=None):
     cur = d
@@ -75,49 +56,26 @@ def safe_get(d, *keys, default=None):
             return default
     return cur
 
-def parse_ea0015(json_obj: dict) -> pd.DataFrame:
-    """
-    嘗試容錯各種大小寫與路徑差異，輸出欄位：
-    OriginTime, Lat, Lon, Depth_km, Magnitude, Location, ReportURL
-    """
-    records = json_obj.get("records") or json_obj.get("Records") or {}
-    quakes = (
-        records.get("earthquake") or
-        records.get("Earthquake") or
-        records.get("data") or
-        []
-    )
+def _to_float(x):
+    try:
+        if x is None or x == "":
+            return None
+        return float(str(x).strip())
+    except Exception:
+        return None
 
+def parse_reports(data):
+    records = data.get("records") or {}
+    quakes = records.get("earthquake") or []
     rows = []
     for q in quakes:
-        origin = safe_get(q, "originTime") or safe_get(q, "OriginTime")
-        # 經緯度 & 深度
-        lat = (
-            safe_get(q, "epicenter", "epicenterLat") or
-            safe_get(q, "Epicenter", "EpicenterLat")
-        )
-        lon = (
-            safe_get(q, "epicenter", "epicenterLon") or
-            safe_get(q, "Epicenter", "EpicenterLon")
-        )
-        depth = (
-            safe_get(q, "depth") or
-            safe_get(q, "Depth")
-        )
-        # 規模
-        mag = (
-            safe_get(q, "magnitude", "magnitudeValue") or
-            safe_get(q, "Magnitude", "MagnitudeValue") or
-            safe_get(q, "magnitude", "magnitude") or
-            safe_get(q, "Magnitude", "Magnitude")
-        )
-        # 位置與連結
-        loc = (
-            safe_get(q, "epicenter", "location") or
-            safe_get(q, "Epicenter", "Location")
-        )
-        url = safe_get(q, "reportURL") or safe_get(q, "ReportURL")
-
+        origin = safe_get(q, "originTime")
+        lat = safe_get(q, "epicenter", "epicenterLat")
+        lon = safe_get(q, "epicenter", "epicenterLon")
+        depth = safe_get(q, "depth")
+        mag = safe_get(q, "magnitude", "magnitudeValue")
+        loc = safe_get(q, "epicenter", "location")
+        url = safe_get(q, "reportURL")
         rows.append({
             "OriginTime": origin,
             "Lat": _to_float(lat),
@@ -125,204 +83,119 @@ def parse_ea0015(json_obj: dict) -> pd.DataFrame:
             "Depth_km": _to_float(depth),
             "Magnitude": _to_float(mag),
             "Location": loc,
-            "ReportURL": url,
+            "ReportURL": url
         })
-
     df = pd.DataFrame(rows)
-
-    # 轉時間、排序（預設：OriginTime 由新到舊）
     if not df.empty:
         df["OriginTime"] = pd.to_datetime(df["OriginTime"], errors="coerce")
-        df = df.sort_values("OriginTime", ascending=False, na_position="last").reset_index(drop=True)
-
+        df = df.sort_values("OriginTime", ascending=False).reset_index(drop=True)
     return df
 
-def _to_float(x):
-    try:
-        if x is None or x == "":
-            return None
-        return float(str(x).strip())
-    except Exception:
-        return None
-
-
 # -----------------------------
-# 視覺化：趨勢圖、台灣地圖
+# 視覺化
 # -----------------------------
-TAIWAN_BBOX = (119, 123, 21, 26)  # lon_min, lon_max, lat_min, lat_max
-
-def plot_trend(df: pd.DataFrame) -> bytes | None:
-    """時間-規模散點圖，輸出 PNG bytes"""
+def plot_trend(df):
     if df.empty:
         return None
-    fig, ax = plt.subplots(figsize=(6, 3.8))
+    fig, ax = plt.subplots(figsize=(6, 4))
     ax.scatter(df["OriginTime"], df["Magnitude"])
     ax.set_xlabel("Origin Time (Taipei)")
     ax.set_ylabel("Magnitude")
     ax.grid(True, linestyle="--", alpha=0.4)
     fig.autofmt_xdate()
-
     buf = io.BytesIO()
-    fig.savefig(buf, format="png", dpi=160, bbox_inches="tight")
+    fig.savefig(buf, format="png", dpi=150, bbox_inches="tight")
     plt.close(fig)
     buf.seek(0)
     return buf.getvalue()
 
-def plot_taiwan_map(df: pd.DataFrame) -> bytes | None:
-    """
-    基礎台灣範圍框圖（非海岸線），用散點展示震央；
-    以規模對應 marker 大小，附簡易圖例。
-    """
+def plot_map(df):
     if df.empty:
         return None
-
-    lon_min, lon_max, lat_min, lat_max = TAIWAN_BBOX
+    lon_min, lon_max, lat_min, lat_max = 119, 123, 21, 26
     fig, ax = plt.subplots(figsize=(6, 6))
-
-    # 邊框
     ax.set_xlim(lon_min, lon_max)
     ax.set_ylim(lat_min, lat_max)
-    ax.set_xlabel("Longitude (°E)")
-    ax.set_ylabel("Latitude (°N)")
-    ax.set_title("Epicenters in Taiwan Region (119–123E, 21–26N)")
-
-    # 散點：大小反映規模
     mags = df["Magnitude"].fillna(0)
-    sizes = (mags.clip(lower=0) + 2.0) ** 3  # 調整一下讓差異更明顯
+    sizes = (mags + 2) ** 3
     ax.scatter(df["Lon"], df["Lat"], s=sizes, alpha=0.6, edgecolor="black")
-
-    # 簡單圖例：M3/4/5/6 對應大小
     for m in [3, 4, 5, 6]:
-        ax.scatter([], [], s=((m + 2.0) ** 3), alpha=0.6, edgecolor="black", label=f"M {m}")
-    ax.legend(title="Magnitude", loc="upper right", frameon=True)
-
+        ax.scatter([], [], s=((m + 2) ** 3), alpha=0.6, edgecolor="black", label=f"M {m}")
+    ax.legend(title="Magnitude")
     ax.grid(True, linestyle="--", alpha=0.3)
-
     buf = io.BytesIO()
-    fig.savefig(buf, format="png", dpi=160, bbox_inches="tight")
+    fig.savefig(buf, format="png", dpi=150, bbox_inches="tight")
     plt.close(fig)
     buf.seek(0)
     return buf.getvalue()
 
-
 # -----------------------------
-# 主流程：查詢 + 輸出
+# 主流程
 # -----------------------------
-def query_and_render(time_from: str, time_to: str, sort_order: str):
-    """
-    1) 取資料 → 2) 轉成 DataFrame → 3) 依排序輸出 → 4) 回傳表格、圖、CSV
-    Gradio 會接收回傳值更新元件。
-    """
-    try:
-        raw = fetch_reports(time_from, time_to)
-    except Exception as e:
-        return gr.update(value=f"查詢錯誤：{e}"), None, None, None
+def df_to_markdown(df, top_n=100):
+    if df.empty:
+        return "（查無資料）"
+    show_cols = ["OriginTime", "Magnitude", "Depth_km", "Lat", "Lon", "Location", "ReportURL"]
+    exist_cols = [c for c in show_cols if c in df.columns]
+    slim = df[exist_cols].head(top_n).copy()
+    if "OriginTime" in slim.columns:
+        slim["OriginTime"] = slim["OriginTime"].dt.tz_localize("UTC").dt.tz_convert(TAIPEI_TZ).dt.strftime("%Y-%m-%d %H:%M:%S %Z")
+    return slim.to_markdown(index=False)
 
+def query_and_render(time_from, time_to, sort_order):
     try:
-        df = parse_ea0015(raw)
+        raw = fetch_reports(time_from, time_to)
+        df = parse_reports(raw)
         if df.empty:
             return "（查無資料）", None, None, None
-
-        # 排序（本地）
         if sort_order == "OriginTime (舊→新)":
-            df = df.sort_values("OriginTime", ascending=True, na_position="last").reset_index(drop=True)
-        else:
-            df = df.sort_values("OriginTime", ascending=False, na_position="last").reset_index(drop=True)
-
-        # 表格（轉為 markdown 簡表，手機閱讀較清楚）
+            df = df.sort_values("OriginTime", ascending=True).reset_index(drop=True)
         md = df_to_markdown(df)
-
-        # 圖
         trend_png = plot_trend(df)
-        map_png = plot_taiwan_map(df)
-
-        # 下載 CSV（以 bytes 形式回傳給 DownloadButton）
+        map_png = plot_map(df)
         csv_bytes = df.to_csv(index=False).encode("utf-8-sig")
-
-        return md, trend_png, map_png, csv_bytes
+        # 建立���時檔讓 DownloadButton 可以下載
+        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".csv", prefix="CWA_E-A0015-001_")
+        tmp.write(csv_bytes)
+        tmp.flush()
+        tmp.close()
+        return md, trend_png, map_png, tmp.name
     except Exception as e:
-        return gr.update(value=f"解析錯誤：{e}"), None, None, None
-
-
-def df_to_markdown(df: pd.DataFrame, top_n: int = 100) -> str:
-    """
-    將前 top_n 筆轉為 markdown 表（欄位：OriginTime, Magnitude, Depth_km, Lat, Lon, Location, ReportURL）
-    """
-    if df.empty:
-        return "（查無資料）"
-
-    show_cols = ["OriginTime", "Magnitude", "Depth_km", "Lat", "Lon", "Location", "ReportURL"]
-    exist_cols = [c for c in show_cols if c in df.columns]
-    slim = df[exist_cols].head(top_n).copy()
-
-    # 時間顯示（台北時區）
-    if "OriginTime" in slim.columns:
-        slim["OriginTime"] = slim["OriginTime"].dt.tz_convert(TAIPEI_TZ).dt.strftime("%Y-%m-%d %H:%M:%S %Z")
-
-    return slim.to_markdown(index=False)
-
+        return f"錯誤：{e}", None, None, None
 
 # -----------------------------
-# Gradio 介面
+# 介面
 # -----------------------------
 default_from, default_to = set_time_range(days=3)
 
-with gr.Blocks(fill_height=True, analytics_enabled=False) as demo:
-    gr.Markdown(
-        """
-# CWA 顯著有感地震報告 (E-A0015-001)
-此 Space **只使用環境變數 `CWA_API_KEY` 作為授權**。  
-預設查詢 **最近 3 天（台北時間）**。手機版為單欄顯示。
-        """
-    )
+with gr.Blocks(fill_height=True) as demo:
+    gr.Markdown("## CWA 顯著有感地震報告 (E-A0015-001)\n預設查詢最近 3 天（台北時間）")
 
-    # 單欄（手機友好）
     with gr.Column():
-        with gr.Row():
-            time_from = gr.Textbox(label="timeFrom yyyy-MM-ddThh:mm:ss", value=default_from, scale=1)
-        with gr.Row():
-            time_to   = gr.Textbox(label="timeTo yyyy-MM-ddThh:mm:ss", value=default_to, scale=1)
+        time_from = gr.Textbox(label="timeFrom yyyy-MM-ddTHH:mm:ss", value=default_from)
+        time_to   = gr.Textbox(label="timeTo yyyy-MM-ddTHH:mm:ss", value=default_to)
 
-        # 快速時間範圍（已移除「最近6小時」）
         with gr.Row():
             btn_12h = gr.Button("最近 12 小時")
             btn_24h = gr.Button("最近 24 小時")
             btn_3d  = gr.Button("最近 3 天")
             btn_5d  = gr.Button("最近 5 天")
 
-        # 排序（在地端）
-        sort_dd = gr.Dropdown(
-            choices=["OriginTime (新→舊)", "OriginTime (舊→新)"],
-            value="OriginTime (新→舊)",
-            label="排序（本地）"
-        )
+        sort_dd = gr.Dropdown(choices=["OriginTime (新→舊)", "OriginTime (舊→新)"], value="OriginTime (新→舊)", label="排序")
 
         run_btn = gr.Button("查詢", variant="primary")
 
-        # 輸出
         table_out = gr.Markdown("（尚未查詢）")
-        trend_out = gr.Image(label="趨勢圖（時間-規模）", type="numpy")
-        map_out   = gr.Image(label="台灣範圍圖（119–123E, 21–26N）", type="numpy")
-        dl_btn    = gr.DownloadButton(label="下載 CSV", file_name="CWA_E-A0015-001.csv")
+        trend_out = gr.Image(label="趨勢圖", type="numpy")
+        map_out   = gr.Image(label="台灣範圍圖", type="numpy")
+        dl_btn    = gr.DownloadButton(label="下載 CSV")  # 不加 file_name
 
-    # 綁定：快速鍵 → 更新時間欄位
     btn_12h.click(lambda: set_time_range(hours=12), outputs=[time_from, time_to])
     btn_24h.click(lambda: set_time_range(hours=24), outputs=[time_from, time_to])
     btn_3d.click(lambda: set_time_range(days=3), outputs=[time_from, time_to])
     btn_5d.click(lambda: set_time_range(days=5), outputs=[time_from, time_to])
 
-    # 綁定：查詢 → 表格 / 圖 / CSV
-    def _on_query(tfrom, tto, sort_sel):
-        md, trend_png, map_png, csv_bytes = query_and_render(tfrom, tto, sort_sel)
-        # DownloadButton 需要返回 bytes-like 物件
-        return md, trend_png, map_png, csv_bytes
-
-    run_btn.click(
-        _on_query,
-        inputs=[time_from, time_to, sort_dd],
-        outputs=[table_out, trend_out, map_out, dl_btn]
-    )
+    run_btn.click(query_and_render, inputs=[time_from, time_to, sort_dd], outputs=[table_out, trend_out, map_out, dl_btn])
 
-# 注意：Hugging Face Spaces 會自動呼叫 demo.launch()
 if __name__ == "__main__":
     demo.launch()