LINE-ROBOT

Sleeping

App Files Files Community

cwadayi commited on Aug 16, 2025

Commit

2744551

verified ·

1 Parent(s): 3234236

Update app.py

Browse files

Files changed (1) hide show

app.py +43 -219

app.py CHANGED Viewed

@@ -1,5 +1,4 @@
 import os
-# ✅ Matplotlib config/cache 放入可寫路徑
 os.environ["MPLCONFIGDIR"] = "/tmp/matplotlib"
 import uuid
@@ -20,7 +19,7 @@ from linebot.v3.webhooks import MessageEvent, TextMessageContent
 import requests
 import pandas as pd
-# --- Matplotlib (headless) ---
 import matplotlib
 matplotlib.use("Agg")
 import matplotlib.pyplot as plt
@@ -28,7 +27,10 @@ from matplotlib.colors import Normalize
 import matplotlib.cm as cm
 from matplotlib import font_manager as fm
-# ✅ 嘗試載入支援中文的字型
 possible_fonts = [
     "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
     "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
@@ -39,82 +41,31 @@ for font_path in possible_fonts:
     if os.path.exists(font_path):
         font_prop = fm.FontProperties(fname=font_path)
         matplotlib.rcParams["font.family"] = font_prop.get_name()
-        print(f"✅ Using CJK font: {font_prop.get_name()}")
         break
-else:
-    print("⚠️ No CJK font found, Chinese glyphs may not display correctly.")
-# --- Environment Variables ---
 CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN")
 CHANNEL_SECRET = os.getenv("CHANNEL_SECRET")
-# ✅ HF_SPACE_URL 自動偵測（若未手動設定 SPACEURL）
 HF_SPACE_URL = os.getenv("SPACEURL")
 if not HF_SPACE_URL:
-    space_id = os.getenv("SPACE_ID")  # e.g., username/space-name
     if space_id and "/" in space_id:
-        author, name = space_id.split("/", 1)
-        HF_SPACE_URL = f"https://{author.replace('_','-')}-{name.replace('_','-')}.hf.space"
-        print(f"🔍 Auto-detected HF_SPACE_URL = {HF_SPACE_URL}")
     else:
-        HF_SPACE_URL = ""  # fallback 由 request.url_root 決定
-# --- 可寫入的靜態目錄 ---
 STATIC_DIR = os.getenv("STATIC_DIR", os.path.join(tempfile.gettempdir(), "static"))
 os.makedirs(STATIC_DIR, exist_ok=True)
-# --- Flask & LINE Bot ---
 app = Flask(__name__)
 configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
 handler = WebhookHandler(CHANNEL_SECRET)
-# -------------------- 共用小工具 --------------------
-def _iso(dt: datetime) -> str:
-    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
-def _base_url_for_images() -> str:
-    if HF_SPACE_URL:
-        return HF_SPACE_URL.rstrip("/")
-    return request.url_root.rstrip("/")
-def _fmt_dt_utc(ts_ms: int | float | None) -> str:
-    """將毫秒或秒級 timestamp 轉成 YYYY-MM-DD HH:MM (UTC)，未知回傳 '未知'"""
-    if ts_ms is None:
-        return "未知"
-    try:
-        # 多數 API 為毫秒；若是秒也能容錯
-        if ts_ms > 1e12:  # 毫秒
-            dt = datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc)
-        else:
-            dt = datetime.fromtimestamp(ts_ms, tz=timezone.utc)
-        return dt.strftime("%Y-%m-%d %H:%M")
-    except Exception:
-        return "未知"
-# -------------------- 路由 --------------------
 @app.route("/", methods=["GET"])
 def home():
-    return """
-    <html>
-      <head>
-        <title>LINE Bot Server</title>
-        <style>
-          body{font-family:Arial,sans-serif;display:flex;justify-content:center;align-items:center;height:100vh;background:#f0f2f5;margin:0;}
-          .container{max-width:720px;text-align:center;padding:40px;background:#fff;border-radius:12px;box-shadow:0 4px 16px rgba(0,0,0,.08)}
-          h1{color:#1dcd00;margin:0 0 8px}
-          p{color:#333;font-size:1.05rem;margin:.4rem 0}
-          .status{font-weight:700;color:#28a745}
-        </style>
-      </head>
-      <body>
-        <div class="container">
-          <h1>✓ LINE Bot Server is Running</h1>
-          <p>This is the backend service for the Earthquake Alert Bot.</p>
-          <p>The service is <span class="status">active</span> and listening for webhook events from LINE.</p>
-        </div>
-      </body>
-    </html>
-    """
 @app.route("/healthz")
 def healthz():
@@ -124,11 +75,13 @@ def healthz():
 def serve_static(filename):
     return send_from_directory(STATIC_DIR, filename)
-# -------------------- 資料存取：USGS --------------------
 USGS_API_BASE_URL = "https://earthquake.usgs.gov/fdsnws/event/1/query"
 def fetch_global_last24h_text(min_mag=5.0, limit=10) -> str:
-    """顯示近 24 小時全球 M≥min_mag 地震，含日期。"""
     now_utc = datetime.now(timezone.utc)
     since = now_utc - timedelta(hours=24)
     params = {
@@ -145,7 +98,6 @@ def fetch_global_last24h_text(min_mag=5.0, limit=10) -> str:
         features = r.json().get("features", [])
         if not features:
             return "✅ 過去 24 小時內，全球無規模 5.0 以上的顯著地震。"
         lines = ["🚨 近 24 小時全球顯著地震 (M≥5.0):", "-" * 20]
         for f in features:
             p = f["properties"]
@@ -158,20 +110,14 @@ def fetch_global_last24h_text(min_mag=5.0, limit=10) -> str:
         return f"❌ 查詢失敗: {e}"
 def fetch_taiwan_df_this_year(min_mag=5.0) -> pd.DataFrame | str:
-    """抓取今年台灣經緯度框內 M≥min_mag 地震的 DataFrame。"""
     now_utc = datetime.now(timezone.utc)
     start_of_year_utc = datetime(now_utc.year, 1, 1, tzinfo=timezone.utc)
     params = {
-        "format": "geojson",
-        "starttime": _iso(start_of_year_utc),
-        "endtime": _iso(now_utc),
         "minmagnitude": float(min_mag),
-        "minlatitude": 21,
-        "maxlatitude": 26,
-        "minlongitude": 119,
-        "maxlongitude": 123,
-        "limit": 250,
-        "orderby": "time",
     }
     try:
         r = requests.get(USGS_API_BASE_URL, params=params, timeout=20)
@@ -183,120 +129,19 @@ def fetch_taiwan_df_this_year(min_mag=5.0) -> pd.DataFrame | str:
         for f in features:
             p = f["properties"]
             lon, lat, *_ = f["geometry"]["coordinates"]
-            rows.append(
-                {
-                    "latitude": lat,
-                    "longitude": lon,
-                    "magnitude": p["mag"],
-                    "place": p.get("place", ""),
-                    "time_utc": datetime.fromtimestamp(p["time"] / 1000, tz=timezone.utc),
-                }
-            )
         return pd.DataFrame(rows)
     except Exception as e:
         return f"❌ 查詢失敗: {e}"
-# -------------------- 資料存取：CWA 地震預警 --------------------
-CWA_ALARM_API = "https://app-2.cwa.gov.tw/api/v1/earthquake/alarm/list"
-def fetch_cwa_alarm_list(limit: int = 5) -> str:
-    """
-    取得交通部中央氣象署「地震預警」清單並格式化為文字。
-    盡量容錯不同欄位名稱；最多輸出 limit 筆。
-    """
-    try:
-        r = requests.get(CWA_ALARM_API, timeout=10)
-        r.raise_for_status()
-        data = r.json()
-    except Exception as e:
-        return f"❌ 地震預警查詢失敗：{e}"
-    # 嘗試從常見包裝鍵取出陣列
-    items = None
-    for key in ("data", "records", "result", "list", "items"):
-        if isinstance(data, dict) and isinstance(data.get(key), list):
-            items = data.get(key)
-            break
-    if items is None:
-        # 也可能根本就是陣列
-        if isinstance(data, list):
-            items = data
-        else:
-            return "⚠️ 無法解析地震預警資料（未知格式）。"
-    if not items:
-        return "✅ 目前沒有地震預警。"
-    lines = ["🚨 地震預警（最新）:", "-" * 20]
-    count = 0
-    for it in items:
-        if count >= limit:
-            break
-        # 嘗試各種常見欄位名稱
-        event_id = it.get("eventId") or it.get("eventid") or it.get("id") or "—"
-        # 可能是 ISO 字串、毫秒或秒
-        origin = it.get("originTime") or it.get("time") or it.get("origin_time") or it.get("earthquake_time")
-        dt_str = "未知"
-        if isinstance(origin, (int, float)):
-            dt_str = _fmt_dt_utc(origin)
-        elif isinstance(origin, str):
-            # 直接顯示（若是 ISO 字串）
-            try:
-                # 嘗試 parse 並以 UTC 格式化
-                dt = datetime.fromisoformat(origin.replace("Z", "+00:00"))
-                dt_str = dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M")
-            except Exception:
-                dt_str = origin  # 無法解析就原樣輸出
-        mag = it.get("magnitude") or it.get("mag") or it.get("ml") or it.get("mw") or "—"
-        depth = it.get("depth") or "—"
-        lat = it.get("latitude") or it.get("lat") or "—"
-        lon = it.get("longitude") or it.get("lon") or "—"
-        place = it.get("location") or it.get("place") or it.get("region") or "—"
-        # 預警區域/縣市（可能是陣列或字串）
-        areas = it.get("areas") or it.get("alertAreas") or it.get("counties") or it.get("cities")
-        if isinstance(areas, list):
-            # 可能是字串陣列或物件陣列
-            try:
-                area_names = []
-                for a in areas:
-                    if isinstance(a, str):
-                        area_names.append(a)
-                    elif isinstance(a, dict):
-                        name = a.get("name") or a.get("area") or a.get("city") or a.get("county")
-                        if name:
-                            area_names.append(str(name))
-                areas_txt = "、".join(area_names) if area_names else "—"
-            except Exception:
-                areas_txt = "—"
-        elif isinstance(areas, str):
-            areas_txt = areas
-        else:
-            areas_txt = "—"
-        lines.append(
-            f"事件: {event_id}\n"
-            f"日期時間: {dt_str} (UTC)\n"
-            f"震級/深度: M{mag} / {depth} km\n"
-            f"位置: {place}（lat {lat}, lon {lon}）\n"
-            f"預警區域: {areas_txt}"
-        )
-        lines.append("")  # 空行
-        count += 1
-    if len(items) > limit:
-        lines.append(f"... 另有 {len(items) - limit} 筆。")
-    return "\n".join(lines).strip()
-# -------------------- 繪圖 --------------------
 def create_and_save_map(df: pd.DataFrame) -> str:
     fig, ax = plt.subplots(figsize=(9, 6), dpi=150)
-    lon_min, lon_max = 118.5, 123.5
-    lat_min, lat_max = 20.5, 26.8
-    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(f"今年 ({datetime.now(timezone.utc).year}) 台灣區域顯著地震 (M≥5.0) — UTC")
@@ -308,18 +153,10 @@ def create_and_save_map(df: pd.DataFrame) -> str:
     colors = cmap(norm(mags.values))
     sizes = 15 + (mags - mags.min()) * 25
-    ax.scatter(
-        df["longitude"].values,
-        df["latitude"].values,
-        s=sizes,
-        c=colors,
-        edgecolor="k",
-        linewidths=0.4,
-        alpha=0.9,
-    )
-    cbar = fig.colorbar(cm.ScalarMappable(norm=norm, cmap=cmap), ax=ax, pad=0.02)
-    cbar.set_label("Magnitude")
     filename = f"map_{uuid.uuid4().hex}.png"
     filepath = os.path.join(STATIC_DIR, filename)
@@ -328,7 +165,10 @@ def create_and_save_map(df: pd.DataFrame) -> str:
     plt.close(fig)
     return filename
-# -------------------- LINE Webhook --------------------
 @app.route("/callback", methods=["POST"])
 def callback():
     signature = request.headers.get("X-Line-Signature")
@@ -339,15 +179,13 @@ def callback():
         abort(400)
     return "OK"
-# -------------------- Message Handler --------------------
 @handler.add(MessageEvent, message=TextMessageContent)
 def handle_message(event):
     user_message = (event.message.text or "").strip().lower()
     with ApiClient(configuration) as api_client:
         line_bot_api = MessagingApi(api_client)
-        # ➊ 地震預警（新指令）
         if "地震預警" in user_message:
             reply_text = fetch_cwa_alarm_list(limit=5)
             line_bot_api.reply_message_with_http_info(
@@ -355,7 +193,6 @@ def handle_message(event):
             )
             return
-        # ➋ 臺灣地震畫圖
         if ("臺灣地震畫圖" in user_message) or ("台灣地震畫圖" in user_message):
             result = fetch_taiwan_df_this_year()
             if isinstance(result, pd.DataFrame):
@@ -365,37 +202,29 @@ def handle_message(event):
                     reply_token=event.reply_token,
                     messages=[
                         TextMessage(text="🗺️ 已為您繪製今年台灣區域 M≥5.0 地震分佈圖（UTC）。"),
-                        ImageMessage(
-                            original_content_url=image_url,
-                            preview_image_url=image_url,
-                        ),
                     ],
                 )
             else:
-                reply = ReplyMessageRequest(
-                    reply_token=event.reply_token,
-                    messages=[TextMessage(text=result)],
-                )
             line_bot_api.reply_message_with_http_info(reply)
             return
-        # ➌ 說明
         if user_message == "/help":
             text = (
                 "📖 地震預警 dayichen 指令說明\n\n"
-                "➡️ /help\n   說明：顯示此幫助訊息。\n\n"
-                "➡️ 地震 / quake\n   說明：查詢全球最近 24 小時內，M≥5.0 的顯著地震（含日期時間）。\n\n"
-                "➡️ 臺灣地震 / 台灣地震\n   說明：查詢今年以來台灣區�� (21–26°N, 119–123°E) M≥5.0 地震（含日期時間）。\n\n"
-                "➡️ 臺灣地震畫圖 / 台灣地震畫圖\n   說明：繪製今年台灣區域 M≥5.0 地震分佈圖並回傳圖片。\n\n"
-                "➡️ 地震預警\n   說明：取得中央氣象署地震預警清單（最新 5 筆）。\n\n"
-                "➡️ 你好\n   說明：顯示歡迎訊息。"
             )
             line_bot_api.reply_message_with_http_info(
                 ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=text)])
             )
             return
-        # ➍ 台灣清單
         if ("臺灣地震" in user_message) or ("台灣地震" in user_message):
             result = fetch_taiwan_df_this_year()
             if isinstance(result, pd.DataFrame):
@@ -414,7 +243,6 @@ def handle_message(event):
             )
             return
-        # ➎ 全球 24 小時
         if ("地震" in user_message) or ("quake" in user_message):
             reply_text = fetch_global_last24h_text()
             line_bot_api.reply_message_with_http_info(
@@ -422,12 +250,8 @@ def handle_message(event):
             )
             return
-        # ➏ 打招呼
         if ("你好" in user_message) or ("hi" in user_message):
             line_bot_api.reply_message_with_http_info(
-                ReplyMessageRequest(
-                    reply_token=event.reply_token,
-                    messages=[TextMessage(text="👋 你好！我是地震查詢機器人。\n\n輸入 /help 查看所有指令。")]
-                )
             )
             return

 import os
 os.environ["MPLCONFIGDIR"] = "/tmp/matplotlib"
 import uuid
 import requests
 import pandas as pd
+# Matplotlib headless + CJK
 import matplotlib
 matplotlib.use("Agg")
 import matplotlib.pyplot as plt
 import matplotlib.cm as cm
 from matplotlib import font_manager as fm
+# ✅ 匯入你剛拆出的模組
+from fetch_cwa_alarm_list import fetch_cwa_alarm_list
+# 設定 CJK 字型（如有）
 possible_fonts = [
     "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
     "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
     if os.path.exists(font_path):
         font_prop = fm.FontProperties(fname=font_path)
         matplotlib.rcParams["font.family"] = font_prop.get_name()
         break
 CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN")
 CHANNEL_SECRET = os.getenv("CHANNEL_SECRET")
+# 自動偵測 HF Space 公網位址
 HF_SPACE_URL = os.getenv("SPACEURL")
 if not HF_SPACE_URL:
+    space_id = os.getenv("SPACE_ID")  # username/space-name
     if space_id and "/" in space_id:
+        a, n = space_id.split("/", 1)
+        HF_SPACE_URL = f"https://{a.replace('_','-')}-{n.replace('_','-')}.hf.space"
     else:
+        HF_SPACE_URL = ""
 STATIC_DIR = os.getenv("STATIC_DIR", os.path.join(tempfile.gettempdir(), "static"))
 os.makedirs(STATIC_DIR, exist_ok=True)
 app = Flask(__name__)
 configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
 handler = WebhookHandler(CHANNEL_SECRET)
 @app.route("/", methods=["GET"])
 def home():
+    return "LINE Bot Server is Running"
 @app.route("/healthz")
 def healthz():
 def serve_static(filename):
     return send_from_directory(STATIC_DIR, filename)
+# ---------- USGS ----------
 USGS_API_BASE_URL = "https://earthquake.usgs.gov/fdsnws/event/1/query"
+def _iso(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
 def fetch_global_last24h_text(min_mag=5.0, limit=10) -> str:
     now_utc = datetime.now(timezone.utc)
     since = now_utc - timedelta(hours=24)
     params = {
         features = r.json().get("features", [])
         if not features:
             return "✅ 過去 24 小時內，全球無規模 5.0 以上的顯著地震。"
         lines = ["🚨 近 24 小時全球顯著地震 (M≥5.0):", "-" * 20]
         for f in features:
             p = f["properties"]
         return f"❌ 查詢失敗: {e}"
 def fetch_taiwan_df_this_year(min_mag=5.0) -> pd.DataFrame | str:
     now_utc = datetime.now(timezone.utc)
     start_of_year_utc = datetime(now_utc.year, 1, 1, tzinfo=timezone.utc)
     params = {
+        "format": "geojson", "starttime": _iso(start_of_year_utc), "endtime": _iso(now_utc),
         "minmagnitude": float(min_mag),
+        "minlatitude": 21, "maxlatitude": 26,
+        "minlongitude": 119, "maxlongitude": 123,
+        "limit": 250, "orderby": "time",
     }
     try:
         r = requests.get(USGS_API_BASE_URL, params=params, timeout=20)
         for f in features:
             p = f["properties"]
             lon, lat, *_ = f["geometry"]["coordinates"]
+            rows.append({
+                "latitude": lat, "longitude": lon, "magnitude": p["mag"],
+                "place": p.get("place", ""), "time_utc": datetime.fromtimestamp(p["time"]/1000, tz=timezone.utc)
+            })
         return pd.DataFrame(rows)
     except Exception as e:
         return f"❌ 查詢失敗: {e}"
+# ---------- 圖像 ----------
 def create_and_save_map(df: pd.DataFrame) -> str:
     fig, ax = plt.subplots(figsize=(9, 6), dpi=150)
+    ax.set_xlim(118.5, 123.5)
+    ax.set_ylim(20.5, 26.8)
     ax.set_xlabel("Longitude (°E)")
     ax.set_ylabel("Latitude (°N)")
     ax.set_title(f"今年 ({datetime.now(timezone.utc).year}) 台灣區域顯著地震 (M≥5.0) — UTC")
     colors = cmap(norm(mags.values))
     sizes = 15 + (mags - mags.min()) * 25
+    ax.scatter(df["longitude"].values, df["latitude"].values,
+               s=sizes, c=colors, edgecolor="k", linewidths=0.4, alpha=0.9)
+    fig.colorbar(cm.ScalarMappable(norm=norm, cmap=cmap), ax=ax, pad=0.02).set_label("Magnitude")
     filename = f"map_{uuid.uuid4().hex}.png"
     filepath = os.path.join(STATIC_DIR, filename)
     plt.close(fig)
     return filename
+def _base_url_for_images() -> str:
+    return HF_SPACE_URL.rstrip("/") if HF_SPACE_URL else request.url_root.rstrip("/")
+# ---------- LINE Webhook ----------
 @app.route("/callback", methods=["POST"])
 def callback():
     signature = request.headers.get("X-Line-Signature")
         abort(400)
     return "OK"
 @handler.add(MessageEvent, message=TextMessageContent)
 def handle_message(event):
     user_message = (event.message.text or "").strip().lower()
     with ApiClient(configuration) as api_client:
         line_bot_api = MessagingApi(api_client)
+        # 地震預警（使用外部模組）
         if "地震預警" in user_message:
             reply_text = fetch_cwa_alarm_list(limit=5)
             line_bot_api.reply_message_with_http_info(
             )
             return
         if ("臺灣地震畫圖" in user_message) or ("台灣地震畫圖" in user_message):
             result = fetch_taiwan_df_this_year()
             if isinstance(result, pd.DataFrame):
                     reply_token=event.reply_token,
                     messages=[
                         TextMessage(text="🗺️ 已為您繪製今年台灣區域 M≥5.0 地震分佈圖（UTC）。"),
+                        ImageMessage(original_content_url=image_url, preview_image_url=image_url),
                     ],
                 )
             else:
+                reply = ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=result)])
             line_bot_api.reply_message_with_http_info(reply)
             return
         if user_message == "/help":
             text = (
                 "📖 地震預警 dayichen 指令說明\n\n"
+                "➡️ /help\n"
+                "➡️ 地震 / quake\n"
+                "➡️ 臺灣地震 / 台灣地震\n"
+                "➡️ 臺灣地震畫圖 / 台灣地震畫圖\n"
+                "➡️ 地震預警（中央氣象署最新 5 筆）\n"
+                "➡️ 你好"
             )
             line_bot_api.reply_message_with_http_info(
                 ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=text)])
             )
             return
         if ("臺灣地震" in user_message) or ("台灣地震" in user_message):
             result = fetch_taiwan_df_this_year()
             if isinstance(result, pd.DataFrame):
             )
             return
         if ("地震" in user_message) or ("quake" in user_message):
             reply_text = fetch_global_last24h_text()
             line_bot_api.reply_message_with_http_info(
             )
             return
         if ("你好" in user_message) or ("hi" in user_message):
             line_bot_api.reply_message_with_http_info(
+                ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text="👋 你好！輸入 /help 查看指令。")])
             )
             return