app.py

app.py

@@ -1,321 +0,0 @@
-
-# app.py (mobile single-column, webhook/Sheets options, TW bbox + legend)
-import os
-import json
-import tempfile
-from datetime import datetime, timedelta
-from typing import List, Dict, Any, Tuple
-
-import gradio as gr
-import pandas as pd
-import requests
-import folium
-
-# Optional: Google Sheets (if available + secret provided)
-try:
-    import gspread
-    from google.oauth2.service_account import Credentials
-    HAS_SHEETS = True
-except Exception:
-    HAS_SHEETS = False
-
-BASE_URL = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001"
-
-INT_ORDER = {
-    "0": 0.0, "1": 1.0, "2": 2.0, "3": 3.0, "4": 4.0,
-    "5弱": 5.0, "5-": 5.0, "5+": 5.3, "5強": 5.3,
-    "6弱": 6.0, "6-": 6.0, "6+": 6.3, "6強": 6.3,
-    "7": 7.0
-}
-INT_FILTER_CHOICES = ["0","1","2","3","4","5弱","5強","6弱","6強","7"]
-
-def intensity_to_order(s: str) -> float:
-    if s is None:
-        return -1.0
-    s = str(s).strip().replace("級","").replace(" ", "")
-    return INT_ORDER.get(s, -1.0)
-
-def validate_iso(dt: str) -> str:
-    if not dt:
-        return ""
-    try:
-        datetime.strptime(dt, "%Y-%m-%dT%H:%M:%S")
-        return dt
-    except ValueError:
-        raise gr.Error("時間格式需為 yyyy-MM-ddThh:mm:ss")
-
-def build_params(limit: int|None, offset: int|None, fmt: str, sort: str|None, timeFrom: str|None, timeTo: str|None):
-    params = []
-    api_key = os.getenv("CWA_API_KEY")
-    if not api_key:
-        raise gr.Error("缺少授權碼：請到 Space 的 Settings → Repository secrets 新增 CWA_API_KEY。")
-    params.append(("Authorization", api_key))
-    if limit is not None: params.append(("limit", str(limit)))
-    if offset is not None: params.append(("offset", str(offset)))
-    if fmt: params.append(("format", fmt))
-    if sort: params.append(("sort", sort))
-    if timeFrom: params.append(("timeFrom", timeFrom))
-    if timeTo: params.append(("timeTo", timeTo))
-    return params
-
-def http_get(url: str, params: List[Tuple[str,str]]) -> Dict[str, Any]:
-    sess = requests.Session()
-    resp = sess.get(url, params=params, timeout=(5, 20))
-    resp.raise_for_status()
-    if "application/json" in resp.headers.get("Content-Type","").lower() or resp.text.strip().startswith("{"):
-        return resp.json()
-    else:
-        return {"raw": resp.text}
-
-def extract_records(payload: Dict[str, Any]):
-    recs = payload.get("records")
-    if isinstance(recs, dict):
-        for k, v in recs.items():
-            if isinstance(v, list):
-                return v
-    result = payload.get("result")
-    if isinstance(result, dict) and isinstance(result.get("records"), list):
-        return result["records"]
-    for key in ("Earthquake","earthquakes","data","items"):
-        v = payload.get(key)
-        if isinstance(v, list):
-            return v
-    return []
-
-def flatten_row(row: Dict[str, Any]) -> Dict[str, Any]:
-    out = {}
-    for key in ("EarthquakeNo","ReportImageURI","Web","ReportColor","ReportContent"):
-        if key in row:
-            out[key] = row.get(key)
-    eqi = row.get("EarthquakeInfo")
-    if isinstance(eqi, dict):
-        out["OriginTime"] = eqi.get("OriginTime")
-        out["Depth_km"] = eqi.get("FocalDepth")
-        mag = eqi.get("EarthquakeMagnitude") or {}
-        if isinstance(mag, dict):
-            out["Magnitude"] = mag.get("MagnitudeValue")
-            out["MagnitudeType"] = mag.get("MagnitudeType")
-        epic = eqi.get("Epicenter") or {}
-        if isinstance(epic, dict):
-            out["Epicenter"] = epic.get("Location")
-            out["EpicenterLon"] = epic.get("EpicenterLongitude")
-            out["EpicenterLat"] = epic.get("EpicenterLatitude")
-    for k in ("OriginTime","originTime","Time"):
-        if k in row and "OriginTime" not in out:
-            out["OriginTime"] = row.get(k)
-    for k in ("Depth","depth","FocalDepth"):
-        if k in row and "Depth_km" not in out:
-            out["Depth_km"] = row.get(k)
-    for k in ("Magnitude","mag"):
-        if k in row and "Magnitude" not in out:
-            out["Magnitude"] = row.get(k)
-    maxint = row.get("Intensity") or row.get("ShakingArea")
-    if isinstance(maxint, dict):
-        out["MaxIntensity"] = maxint.get("MaxIntensity")
-    return out
-
-def df_with_types(df: pd.DataFrame) -> pd.DataFrame:
-    if "OriginTime" in df.columns:
-        try:
-            df["OriginTime"] = pd.to_datetime(df["OriginTime"], format="%Y-%m-%dT%H:%M:%S", errors="coerce")
-        except Exception:
-            pass
-    if "Magnitude" in df.columns:
-        df["Magnitude"] = pd.to_numeric(df["Magnitude"], errors="coerce")
-    if "Depth_km" in df.columns:
-        df["Depth_km"] = pd.to_numeric(df["Depth_km"], errors="coerce")
-    if "MaxIntensity" in df.columns:
-        df["MaxIntensityOrder"] = df["MaxIntensity"].map(intensity_to_order)
-    return df
-
-def add_tw_bbox(m: folium.Map):
-    bounds = [(21.0, 119.0), (26.0, 123.0)]
-    folium.Rectangle(bounds=bounds, color="#444", fill=False, weight=2, dash_array="5").add_to(m)
-
-def add_legend(m: folium.Map):
-    html = """
-    <div style='position: fixed; bottom: 10px; right: 10px; z-index:9999; background: rgba(255,255,255,0.9); padding: 8px 10px; border:1px solid #999; border-radius:6px; font-size:12px;'>
-      <div style='font-weight:600; margin-bottom:4px;'>圖例</div>
-      <div><span style='display:inline-block;width:12px;height:12px;background:#abd9e9;margin-right:6px;border:1px solid #999;'></span> M4.0–4.9</div>
-      <div><span style='display:inline-block;width:12px;height:12px;background:#fdae61;margin-right:6px;border:1px solid #999;'></span> M5.0–5.9</div>
-      <div><span style='display:inline-block;width:12px;height:12px;background:#d7191c;margin-right:6px;border:1px solid #999;'></span> M≥6.0</div>
-      <div style='margin-top:4px;'>圓徑 ≈ 規模 × 2.5</div>
-    </div>
-    """
-    folium.Marker(location=[0,0], icon=folium.DivIcon(html=html)).add_to(m)
-
-def make_map(df: pd.DataFrame) -> str:
-    if not {"EpicenterLat","EpicenterLon"}.issubset(df.columns):
-        return "<p>沒有可用的經緯度資料。</p>"
-    valid = df.dropna(subset=["EpicenterLat","EpicenterLon"]).copy()
-    if valid.empty:
-        return "<p>沒有可用的經緯度資料。</p>"
-    try:
-        valid["EpicenterLat"] = pd.to_numeric(valid["EpicenterLat"], errors="coerce")
-        valid["EpicenterLon"] = pd.to_numeric(valid["EpicenterLon"], errors="coerce")
-    except Exception:
-        pass
-    valid = valid.dropna(subset=["EpicenterLat","EpicenterLon"])
-    if valid.empty:
-        return "<p>沒有可用的經緯度資料。</p>"
-    center_lat = valid["EpicenterLat"].mean()
-    center_lon = valid["EpicenterLon"].mean()
-    m = folium.Map(location=[center_lat, center_lon], zoom_start=6, tiles="OpenStreetMap")
-    add_tw_bbox(m)
-    for _, r in valid.iterrows():
-        lat = float(r["EpicenterLat"]); lon = float(r["EpicenterLon"])
-        mag = r.get("Magnitude", None); itx = r.get("MaxIntensity", "")
-        radius = 4.0
-        try:
-            if pd.notna(mag):
-                radius = max(4.0, min(20.0, float(mag) * 2.5))
-        except Exception:
-            pass
-        color = "#2c7bb6"
-        try:
-            if mag is not None and float(mag) >= 6.0:
-                color = "#d7191c"
-            elif mag is not None and float(mag) >= 5.0:
-                color = "#fdae61"
-            elif mag is not None and float(mag) >= 4.0:
-                color = "#abd9e9"
-        except Exception:
-            pass
-        popup = folium.Popup(html=f"<b>時間</b>: {r.get('OriginTime','')}<br>"
-                                  f"<b>震央</b>: {r.get('Epicenter','')}<br>"
-                                  f"<b>規模</b>: {mag}<br>"
-                                  f"<b>深度</b>: {r.get('Depth_km','')} km<br>"
-                                  f"<b>最大震度</b>: {itx}", max_width=320)
-        folium.CircleMarker(location=[lat, lon], radius=radius, color=color, fill=True, fill_opacity=0.7, popup=popup).add_to(m)
-    add_legend(m)
-    return m._repr_html_()
-
-def push_webhook(url: str, summary: str, df: pd.DataFrame) -> str:
-    if not url:
-        return "未設定 Webhook URL"
-    data = {
-        "summary": summary,
-        "count": int(len(df)),
-        "generated_at": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
-        "records": df.head(200).to_dict(orient="records")
-    }
-    try:
-        r = requests.post(url, json=data, timeout=10)
-        return f"Webhook 回應: {r.status_code}"
-    except Exception as e:
-        return f"Webhook 失敗: {e}"
-
-def push_sheets(spreadsheet_id: str, worksheet_name: str, df: pd.DataFrame) -> str:
-    if not HAS_SHEETS:
-        return "未安裝 Sheets 套件（gspread / google-auth）"
-    if not spreadsheet_id or not worksheet_name:
-        return "未設定 Spreadsheet ID 或工作表名稱"
-    sa_json = os.getenv("GDRIVE_SERVICE_ACCOUNT_JSON")
-    if not sa_json:
-        return "缺少 GDRIVE_SERVICE_ACCOUNT_JSON 憑證（請放在 Secrets）"
-    try:
-        info = json.loads(sa_json)
-        scopes = ["https://www.googleapis.com/auth/spreadsheets"]
-        creds = Credentials.from_service_account_info(info, scopes=scopes)
-        gc = gspread.authorize(creds)
-        sh = gc.open_by_key(spreadsheet_id)
-        try:
-            ws = sh.worksheet(worksheet_name)
-        except gspread.exceptions.WorksheetNotFound:
-            ws = sh.add_worksheet(title=worksheet_name, rows=1000, cols=40)
-        ws.clear()
-        if df.empty:
-            ws.update("A1", [["No data"]])
-            return "已寫入空資料到 Google Sheets"
-        values = [list(df.columns)] + df.astype(str).values.tolist()
-        ws.update("A1", values)
-        return "已寫入 Google Sheets"
-    except Exception as e:
-        return f"Sheets 失敗: {e}"
-
-def fetch(time_from, time_to, limit, offset, fmt, sort, min_intensity, webhook_url, enable_webhook, spreadsheet_id, worksheet_name, enable_sheets):
-    time_from = validate_iso(time_from) if time_from else None
-    time_to = validate_iso(time_to) if time_to else None
-    params = build_params(limit=limit, offset=offset, fmt=fmt, sort=sort, timeFrom=time_from, timeTo=time_to)
-    payload = http_get(BASE_URL, params)
-    records = extract_records(payload)
-    flat = [flatten_row(r) for r in records]
-    df = pd.DataFrame(flat)
-    df = df_with_types(df)
-    if min_intensity:
-        thr = intensity_to_order(min_intensity)
-        if "MaxIntensityOrder" in df.columns and thr >= 0:
-            df = df[df["MaxIntensityOrder"] >= thr]
-    if "OriginTime" in df.columns:
-        ascending = True if sort == "OriginTime" else False
-        df = df.sort_values("OriginTime", ascending=ascending)
-    tmpdir = tempfile.mkdtemp(prefix="cwa_")
-    csv_path = os.path.join(tmpdir, "cwa_quake.csv")
-    json_path = os.path.join(tmpdir, "raw.json")
-    df.to_csv(csv_path, index=False, encoding="utf-8")
-    with open(json_path, "w", encoding="utf-8") as f:
-        json.dump(payload, f, ensure_ascii=False, indent=2)
-    total = len(df)
-    earliest = latest = ""
-    if total and "OriginTime" in df.columns:
-        earliest_row = df.iloc[0]; latest_row = df.iloc[-1]
-        def fmt_row(r):
-            ot = r.get("OriginTime","")
-            if isinstance(ot, pd.Timestamp):
-                ot = ot.strftime("%Y-%m-%dT%H:%M:%S")
-            return f"{ot} | {r.get('Epicenter','')} | M{r.get('Magnitude','')} | 深{r.get('Depth_km','')}km | 震度{r.get('MaxIntensity','')}"
-        earliest = "最早: " + fmt_row(earliest_row)
-        latest = "最新: " + fmt_row(latest_row)
-    summary = f"取得筆數: {total}\n{earliest}\n{latest}"
-    html_map = make_map(df)
-    if "OriginTime" in df.columns:
-        df["OriginTime"] = df["OriginTime"].astype(str)
-    webhook_msg = ""
-    if enable_webhook and webhook_url:
-        webhook_msg = push_webhook(webhook_url, summary, df)
-    sheets_msg = ""
-    if enable_sheets:
-        sheets_msg = push_sheets(spreadsheet_id, worksheet_name, df)
-    notes = "\n".join([m for m in [webhook_msg, sheets_msg] if m])
-    return df, summary, csv_path, json_path, html_map, notes
-
-now = datetime.now()
-TF_DEFAULT = (now - timedelta(days=3)).strftime("%Y-%m-%dT%H:%M:%S")
-TT_DEFAULT = now.strftime("%Y-%m-%dT%H:%M:%S")
-
-with gr.Blocks(title="CWA 顯著有感地震報告 (E-A0015-001)") as demo:
-    gr.Markdown("# CWA 顯著有感地震報告 (E-A0015-001)")
-    gr.Markdown("**此 Space 只使用環境變數 `CWA_API_KEY` 作為授權。** 若要推送到 Google Sheets，請在 Secrets 提供 `GDRIVE_SERVICE_ACCOUNT_JSON`（Service Account JSON）。")
-    time_from = gr.Textbox(label="timeFrom yyyy-MM-ddThh:mm:ss", value=TF_DEFAULT)
-    time_to = gr.Textbox(label="timeTo yyyy-MM-ddThh:mm:ss", value=TT_DEFAULT)
-    sort = gr.Dropdown(choices=[None, "OriginTime"], value=None, label="sort（預設降冪；選 OriginTime 會升冪）")
-    min_intensity = gr.Dropdown(choices=INT_FILTER_CHOICES, value=None, label="最小最大震度（過濾）")
-    limit = gr.Number(label="limit（筆數上限）", precision=0, value=60)
-    offset = gr.Number(label="offset（起始偏移）", precision=0, value=0)
-    fmt = gr.Radio(choices=["JSON","XML"], value="JSON", label="回傳格式")
-    gr.Markdown("### 推送（選用）")
-    with gr.Row():
-        webhook_url = gr.Textbox(label="Webhook URL（選填）", placeholder="例如：你的 n8n / Dify / 自有 API")
-        enable_webhook = gr.Checkbox(label="查詢後自動 POST 到 Webhook", value=False)
-    with gr.Row():
-        spreadsheet_id = gr.Textbox(label="Google Sheets Spreadsheet ID（選填）", placeholder="例如：1AbC...")
-        worksheet_name = gr.Textbox(label="工作表名稱（例如：Sheet1）", value="Sheet1")
-    enable_sheets = gr.Checkbox(label="查詢後寫入 Google Sheets（需設定服務帳戶憑證）", value=False)
-    run_btn = gr.Button("查詢", variant="primary")
-    out_df = gr.Dataframe(label="查詢結果（扁平化）", interactive=False, wrap=True, datatype="str")
-    out_summary = gr.Textbox(label="摘要", interactive=False)
-    out_csv = gr.File(label="下載 CSV")
-    out_json = gr.File(label="下載原始 JSON")
-    out_map = gr.HTML(label="震央地圖")
-    out_notes = gr.Textbox(label="推送狀態", interactive=False)
-    def on_click(time_from, time_to, limit, offset, fmt, sort, min_intensity, webhook_url, enable_webhook, spreadsheet_id, worksheet_name, enable_sheets):
-        df, summary, csv_path, json_path, html_map, notes = fetch(
-            time_from, time_to, int(limit) if limit is not None else None, int(offset) if offset is not None else None,
-            fmt, sort, min_intensity, webhook_url, enable_webhook, spreadsheet_id, worksheet_name, enable_sheets
-        )
-        return df, summary, csv_path, json_path, html_map, notes
-    run_btn.click(on_click, inputs=[time_from, time_to, limit, offset, fmt, sort, min_intensity, webhook_url, enable_webhook, spreadsheet_id, worksheet_name, enable_sheets],
-                  outputs=[out_df, out_summary, out_csv, out_json, out_map, out_notes])
-
-if __name__ == "__main__":
-    demo.launch()