Upload app.py

app.py

@@ -0,0 +1,349 @@
+
+# app.py (adds hourly auto-refresh + GeoJSON/KML downloads)
+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
+
+BASE_URL = "https://opendata.cwa.gov.tw/api/v1/rest/datastore/E-A0015-001"
+
+TILE_CHOICES = {
+    "OpenStreetMap": "OpenStreetMap",
+    "CartoDB Positron": "CartoDB positron",
+    "Stamen Terrain": "Stamen Terrain",
+    "Esri World Imagery (衛星)": "Esri.WorldImagery"
+}
+
+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")
+    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, tile_choice: str) -> 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>"
+
+    tiles = TILE_CHOICES.get(tile_choice, "OpenStreetMap")
+    if tiles == "Esri.WorldImagery":
+        m = folium.Map(location=[valid["EpicenterLat"].mean(), valid["EpicenterLon"].mean()], zoom_start=6, tiles=None)
+        folium.TileLayer(
+            tiles="https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{z}/{y}/{x}",
+            attr="Tiles © Esri — Source: Esri, i-cubed, USDA, USGS, AEX, GeoEye, Getmapping, Aerogrid, IGN, IGP, UPR-EGP, and the GIS User Community",
+            name="Esri World Imagery"
+        ).add_to(m)
+    else:
+        m = folium.Map(location=[valid["EpicenterLat"].mean(), valid["EpicenterLon"].mean()], zoom_start=6, tiles=tiles)
+
+    add_tw_bbox(m)
+
+    for _, r in valid.iterrows():
+        lat = float(r["EpicenterLat"]); lon = float(r["EpicenterLon"])
+        mag = r.get("Magnitude", None)
+        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", 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 to_geojson(df: pd.DataFrame, path: str) -> str:
+    if not {"EpicenterLat","EpicenterLon"}.issubset(df.columns):
+        open(path, "w", encoding="utf-8").write(json.dumps({"type":"FeatureCollection","features":[]}))
+        return path
+    features = []
+    for _, r in df.iterrows():
+        try:
+            lat = float(r.get("EpicenterLat"))
+            lon = float(r.get("EpicenterLon"))
+        except (TypeError, ValueError):
+            continue
+        props = {
+            "OriginTime": str(r.get("OriginTime","")),
+            "Epicenter": r.get("Epicenter",""),
+            "Magnitude": r.get("Magnitude",""),
+            "Depth_km": r.get("Depth_km","")
+        }
+        features.append({
+            "type": "Feature",
+            "geometry": {"type": "Point", "coordinates": [lon, lat]},
+            "properties": props
+        })
+    fc = {"type":"FeatureCollection", "features": features}
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(fc, f, ensure_ascii=False, indent=2)
+    return path
+
+def to_kml(df: pd.DataFrame, path: str) -> str:
+    # Simple KML without external libs
+    def esc(s):
+        return str(s).replace("&","&amp;").replace("<","&lt;").replace(">","&gt;")
+    kml = [
+        '<?xml version="1.0" encoding="UTF-8"?>',
+        '<kml xmlns="http://www.opengis.net/kml/2.2">',
+        "<Document>"
+    ]
+    if {"EpicenterLat","EpicenterLon"}.issubset(df.columns):
+        for _, r in df.iterrows():
+            try:
+                lat = float(r.get("EpicenterLat"))
+                lon = float(r.get("EpicenterLon"))
+            except (TypeError, ValueError):
+                continue
+            name = f"M{r.get('Magnitude','')} {r.get('Epicenter','')}"
+            desc = f"時間: {r.get('OriginTime','')}\n深度: {r.get('Depth_km','')} km"
+            kml.extend([
+                "<Placemark>",
+                f"<name>{esc(name)}</name>",
+                f"<description>{esc(desc)}</description>",
+                "<Point>",
+                f"<coordinates>{lon},{lat},0</coordinates>",
+                "</Point>",
+                "</Placemark>"
+            ])
+    kml.append("</Document></kml>")
+    with open(path, "w", encoding="utf-8") as f:
+        f.write("\n".join(kml))
+    return path
+
+def fetch(time_from, time_to, limit, offset, fmt, sort, tile_choice):
+    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 "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")
+    geojson_path = os.path.join(tmpdir, "cwa_quake.geojson")
+    kml_path = os.path.join(tmpdir, "cwa_quake.kml")
+    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)
+    to_geojson(df, geojson_path)
+    to_kml(df, kml_path)
+    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"
+        earliest = "最早: " + fmt_row(earliest_row)
+        latest = "最新: " + fmt_row(latest_row)
+    summary = f"取得筆數: {total}\n{earliest}\n{latest}"
+    html_map = make_map(df, tile_choice)
+    if "OriginTime" in df.columns:
+        df["OriginTime"] = df["OriginTime"].astype(str)
+    return df, summary, csv_path, json_path, html_map, geojson_path, kml_path
+
+def quick_range(hours: int):
+    now = datetime.now()
+    tf = (now - timedelta(hours=hours)).strftime("%Y-%m-%dT%H:%M:%S")
+    tt = now.strftime("%Y-%m-%dT%H:%M:%S")
+    return tf, tt
+
+with gr.Blocks(title="CWA 顯著有感地震報告 (E-A0015-001)") as demo:
+    gr.Markdown("# CWA 顯著有感地震報告 (E-A0015-001)")
+    gr.Markdown("**此 Space 只使用環境變數 `CWA_API_KEY` 作為授權。** 預設查詢最近 3 天。")
+
+    # defaults computed at runtime
+    tf_default = (datetime.now() - timedelta(days=3)).strftime("%Y-%m-%dT%H:%M:%S")
+    tt_default = datetime.now().strftime("%Y-%m-%dT%H:%M:%S")
+
+    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)
+    with gr.Row():
+        btn6 = gr.Button("最近 6 小時")
+        btn12 = gr.Button("最近 12 小時")
+        btn24 = gr.Button("最近 24 小時")
+        btn3d = gr.Button("最近 3 天")
+    sort = gr.Dropdown(choices=[None, "OriginTime"], value=None, label="sort（預設降冪；選 OriginTime 會升冪）")
+    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="回傳格式")
+    tile_choice = gr.Dropdown(choices=list(TILE_CHOICES.keys()), value="OpenStreetMap", label="地圖底圖")
+
+    # Auto refresh (hourly)
+    auto_on = gr.Checkbox(label="每小時自動刷新（固定使用目前 timeFrom/timeTo）", value=False)
+    timer = gr.Timer(3600.0)
+
+    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_geojson = gr.File(label="下載 GeoJSON")
+    out_kml = gr.File(label="下載 KML")
+
+    def on_click(time_from, time_to, limit, offset, fmt, sort, tile_choice):
+        df, summary, csv_path, json_path, html_map, geojson_path, kml_path = 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, tile_choice
+        )
+        return df, summary, csv_path, json_path, html_map, geojson_path, kml_path
+
+    def on_tick(auto_on, time_from, time_to, limit, offset, fmt, sort, tile_choice):
+        if not auto_on:
+            # return no updates
+            return gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip()
+        return on_click(time_from, time_to, limit, offset, fmt, sort, tile_choice)
+
+    run_btn.click(on_click, inputs=[time_from, time_to, limit, offset, fmt, sort, tile_choice],
+                  outputs=[out_df, out_summary, out_csv, out_json, out_map, out_geojson, out_kml])
+
+    btn6.click(lambda: quick_range(6), inputs=[], outputs=[time_from, time_to])
+    btn12.click(lambda: quick_range(12), inputs=[], outputs=[time_from, time_to])
+    btn24.click(lambda: quick_range(24), inputs=[], outputs=[time_from, time_to])
+    btn3d.click(lambda: ((datetime.now() - timedelta(days=3)).strftime("%Y-%m-%dT%H:%M:%S"), datetime.now().strftime("%Y-%m-%dT%H:%M:%S")), inputs=[], outputs=[time_from, time_to])
+
+    timer.tick(on_tick, inputs=[auto_on, time_from, time_to, limit, offset, fmt, sort, tile_choice],
+               outputs=[out_df, out_summary, out_csv, out_json, out_map, out_geojson, out_kml])
+
+if __name__ == "__main__":
+    demo.launch()