app.py

app.py

@@ -1,172 +0,0 @@
-
-# app.py
-# Gradio app for Hugging Face Spaces: CWA 顯著有感地震報告 (E-A0015-001)
-import os
-import json
-import tempfile
-from datetime import datetime
-from typing import List, Dict, Any, Tuple
-
-import gradio as gr
-import pandas as pd
-import requests
-
-BASE_URL = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001"
-AREAS = ['宜蘭縣', '花蓮縣', '臺東縣', '澎湖縣', '金門縣', '連江縣', '臺北市', '新北市', '桃園市', '臺中市', '臺南市', '高雄市', '基隆市', '新竹縣', '新竹市', '苗栗縣', '彰化縣', '南投縣', '雲林縣', '嘉義縣', '嘉義市', '屏東縣']
-
-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(auth: str, limit: int|None, offset: int|None, fmt: str, areas: List[str], stations: List[str], sort: str|None, timeFrom: str|None, timeTo: str|None):
-    params = []
-    if auth:
-        params.append(("Authorization", auth))
-    elif os.getenv("CWA_API_KEY"):
-        params.append(("Authorization", os.getenv("CWA_API_KEY")))
-    else:
-        raise gr.Error("缺少授權碼：請在左側輸入 Authorization，或於 Spaces Secrets 設定 CWA_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))
-    for a in (areas or []):
-        params.append(("AreaName", a))
-    for s in (stations or []):
-        params.append(("StationName", s))
-    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 fetch(auth, time_from, time_to, limit, offset, fmt, sel_areas, stations, sort):
-    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(auth=auth, limit=limit, offset=offset, fmt=fmt, areas=sel_areas, stations=stations, 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]
-
-    # dataframe
-    df = pd.DataFrame(flat)
-    # Save files to temp and return handles
-    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)
-
-    # quick summary
-    total = len(df)
-    earliest = latest = ""
-    if total and "OriginTime" in df.columns:
-        try:
-            s = df.dropna(subset=["OriginTime"]).copy()
-            s["OriginTime_dt"] = pd.to_datetime(s["OriginTime"], format="%Y-%m-%dT%H:%M:%S", errors="coerce")
-            s = s.sort_values("OriginTime_dt")
-            if len(s):
-                earliest_row = s.iloc[0]
-                latest_row = s.iloc[-1]
-                earliest = f"最早: {earliest_row['OriginTime']} | {earliest_row.get('Epicenter','')} | M{earliest_row.get('Magnitude','')}"
-                latest = f"最新: {latest_row['OriginTime']} | {latest_row.get('Epicenter','')} | M{latest_row.get('Magnitude','')}"
-        except Exception:
-            pass
-
-    summary = f"取得筆數: {total}\n{earliest}\n{latest}"
-    return df, summary, csv_path, json_path
-
-with gr.Blocks(title="CWA 顯著有感地震報告 E-A0015-001") as demo:
-    gr.Markdown("# CWA 顯著有感地震報告 (E-A0015-001)")
-    gr.Markdown("在左側設定授權與查詢條件。Authorization 可留空並改用環境變數 **CWA_API_KEY**（建議在 Spaces Secrets 設定）。")
-    with gr.Row():
-        with gr.Column(scale=1):
-            auth = gr.Textbox(label="Authorization（可留空改用 CWA_API_KEY）", type="password", placeholder="留空則使用環境變數 CWA_API_KEY")
-            time_from = gr.Textbox(label="timeFrom yyyy-MM-ddThh:mm:ss", placeholder="例如 2025-08-01T00:00:00")
-            time_to = gr.Textbox(label="timeTo yyyy-MM-ddThh:mm:ss", placeholder="例如 2025-08-10T23:59:59")
-            areas = gr.CheckboxGroup(choices=AREAS, label="AreaName（可複選）")
-            stations = gr.Textbox(label="StationName（以逗號分隔，可留空）", placeholder="例如：台北、花蓮...")
-            sort = gr.Dropdown(choices=[None, "OriginTime"], value=None, label="sort（預設降冪；選 OriginTime 會升冪）")
-            with gr.Row():
-                limit = gr.Number(label="limit（筆數上限）", precision=0)
-                offset = gr.Number(label="offset（起始偏移）", precision=0, value=0)
-            fmt = gr.Radio(choices=["JSON","XML"], value="JSON", label="回傳格式")
-            run_btn = gr.Button("查詢", variant="primary")
-        with gr.Column(scale=2):
-            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")
-
-    def on_click(auth, time_from, time_to, limit, offset, fmt, areas_sel, stations_txt, sort):
-        stations_list = []
-        if stations_txt:
-            stations_list = [s.strip() for s in stations_txt.split(",") if s.strip()]
-        df, summary, csv_path, json_path = fetch(auth, time_from, time_to, int(limit) if limit is not None else None, int(offset) if offset is not None else None, fmt, areas_sel, stations_list, sort)
-        return df, summary, csv_path, json_path
-
-    run_btn.click(on_click, inputs=[auth, time_from, time_to, limit, offset, fmt, areas, stations, sort], outputs=[out_df, out_summary, out_csv, out_json])
-
-if __name__ == "__main__":
-    demo.launch()