Upload 8 files

ai_service.py

@@ -0,0 +1,82 @@
+# ai_service.py
+import os
+from config import (
+    LLM_DEVICE, LLM_THREADS, LLM_MODEL, TRANSFORMERS_CACHE,
+    LLM_MAX_NEW_TOKENS, LLM_TOP_K, LLM_TEMPERATURE
+)
+
+# 用於延遲載入語言模型的字典
+_LLM = {"loaded": False, "ok": False, "err": None, "model": None, "tokenizer": None, "device": "cpu"}
+
+def _ensure_llm():
+    """在首次使用時載入 AI 模型與分詞器。"""
+    if _LLM["loaded"]:
+        return _LLM["ok"], _LLM["err"]
+
+    _LLM["loaded"] = True
+    try:
+        import torch
+        from transformers import AutoTokenizer, AutoModelForCausalLM
+
+        device = LLM_DEVICE
+        if device not in ("cuda", "cpu"):
+            device = "cuda" if torch.cuda.is_available() else "cpu"
+            torch.set_num_threads(max(1, int(LLM_THREADS)))
+
+        tok = AutoTokenizer.from_pretrained(LLM_MODEL, cache_dir=TRANSFORMERS_CACHE)
+        mdl = AutoModelForCausalLM.from_pretrained(LLM_MODEL, cache_dir=TRANSFORMERS_CACHE)
+        
+        try:
+            mdl = mdl.to(device)
+        except Exception:
+            device = "cpu"
+            mdl = mdl.to(device)
+
+        _LLM.update({"ok": True, "model": mdl, "tokenizer": tok, "device": device})
+        return True, None
+    except Exception as e:
+        _LLM["err"] = f"{e}"
+        _LLM["ok"] = False
+        return False, _LLM["err"]
+
+def generate_ai_text(user_prompt: str) -> str:
+    """使用已載入的 AI 模型生成文字回應。"""
+    ok, err = _ensure_llm()
+    if not ok:
+        return (
+            "🤖 AI 尚未啟用：缺少依賴或模型未下載。\n"
+            "請在 requirements.txt 加入 transformers、torch、accelerate、safetensors 等。\n"
+            f"詳細錯誤：{err}"
+        )
+
+    import torch
+    tok = _LLM["tokenizer"]
+    mdl = _LLM["model"]
+    device = _LLM["device"]
+
+    sys_prefix = (
+        "你是一個地震資訊與一般問答的 LINE 助理。回答要精簡、清楚；"
+        "若與地震相關可加入注意事項；若無關則一般回覆。\n\n使用者："
+    )
+    prompt = sys_prefix + user_prompt
+
+    try:
+        inputs = tok(prompt, return_tensors="pt").to(device)
+        with torch.no_grad():
+            output = mdl.generate(
+                input_ids=inputs["input_ids"],
+                attention_mask=inputs.get("attention_mask"),
+                max_new_tokens=LLM_MAX_NEW_TOKENS,
+                do_sample=True,
+                top_k=LLM_TOP_K,
+                temperature=LLM_TEMPERATURE,
+                pad_token_id=tok.eos_token_id,
+            )
+        text = tok.decode(output[0], skip_special_tokens=True)
+        if sys_prefix in text:
+            text = text.split(sys_prefix, 1)[-1]
+        if user_prompt in text:
+            text = text.split(user_prompt, 1)[-1].strip()
+        return (text or "（沒有產生內容）")[:1200]
+    except Exception as e:
+        return f"AI 產生發生錯誤：{e}"

app.py

@@ -0,0 +1,127 @@
+# app.py
+# 首先匯入 config，以設定環境變數
+import config
+
+from flask import Flask, request, abort, send_from_directory
+from linebot.v3 import WebhookHandler
+from linebot.v3.exceptions import InvalidSignatureError
+from linebot.v3.messaging import (
+    Configuration, ApiClient, MessagingApi,
+    ReplyMessageRequest
+)
+from linebot.v3.webhooks import MessageEvent, TextMessageContent
+
+# 匯入指令處理器
+from command_handler import process_message
+
+# ------------------------------------------------------------------------------
+# Flask & LINE Bot 設定
+# ------------------------------------------------------------------------------
+app = Flask(__name__)
+line_config = Configuration(access_token=config.CHANNEL_ACCESS_TOKEN)
+handler = WebhookHandler(config.CHANNEL_SECRET)
+
+# ------------------------------------------------------------------------------
+# Web 伺服器路由
+# ------------------------------------------------------------------------------
+@app.route("/", methods=["GET"])
+def home():
+    """渲染首頁，包含說明與狀態。"""
+    base = (config.HF_SPACE_URL or request.url_root).rstrip("/")
+    webhook_url = f"{base}/callback"
+    static_hint = f"{base}/static/<filename>"
+    channel_ok = "✅" if config.CHANNEL_ACCESS_TOKEN and config.CHANNEL_SECRET else "⚠️"
+    space_ok = "✅" if config.HF_SPACE_URL else "ℹ️"
+
+    return f"""
+<!doctype html>
+<html lang="zh-Hant"><head>
+<meta charset="utf-8"/><meta name="viewport" content="width=device-width,initial-scale=1"/>
+<title>地震預警 dayichen – LINE Bot Server</title>
+<style>
+:root{{--bg:#0f1115;--card:#151821;--text:#e6e8ef;--muted:#9aa4b2;--border:rgba(255,255,255,.08)}}
+*{{box-sizing:border-box}} body{{margin:0;background:#0f1115;color:#e6e8ef;
+font:16px/1.6 -apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,"Noto Sans TC","PingFang TC",sans-serif;
+padding:32px 16px;display:flex;justify-content:center}}
+.wrap{{width:100%;max-width:980px}} .hero{{background:linear-gradient(135deg,#1f2937,#0f172a);
+border:1px solid var(--border);border-radius:16px;padding:28px;margin-bottom:20px;box-shadow:0 8px 30px rgba(0,0,0,.25)}}
+.title{{margin:0 0 6px;font-size:28px;font-weight:800}} .subtitle{{margin:0;color:#9aa4b2}}
+.grid{{display:grid;gap:16px;grid-template-columns:repeat(auto-fit,minmax(260px,1fr));margin-top:18px}}
+.card{{background:#151821;border:1px solid var(--border);border-radius:14px;padding:16px 18px}}
+h3{{margin:0 0 8px;font-size:18px}} .kbd{{padding:2px 6px;border:1px solid var(--border);border-radius:6px;background:#0b0e14}}
+.mono,code{{font-family:ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;color:#e6e8ef}}
+a{{color:#60a5fa;text-decoration:none}} a:hover{{text-decoration:underline}}
+.badge{{display:inline-block;padding:2px 8px;border-radius:999px;background:#1f2937;border:1px solid var(--border);font-size:12px;color:#9aa4b2}}
+.foot{{color:#9aa4b2;font-size:13px;margin-top:18px;text-align:center}}
+</style></head>
+<body><div class="wrap"><section class="hero">
+<div class="badge">狀態：<span style="color:{'#86efac' if channel_ok=='✅' else '#fbbf24'}">{channel_ok}</span> LINE 金鑰　·　HF Space：{space_ok}</div>
+<h1 class="title">地震預警 dayichen – LINE Bot</h1>
+<p class="subtitle">指令：/help、地震/quake、臺灣地震/台灣地震、臺灣地震畫圖/台灣地震畫圖、地震預警、AI（ai + 問題）。</p>
+<div class="grid">
+<div class="card"><h3>🚀 快速開始</h3><ul>
+<li><span class="kbd">/help</span>：顯示所有指令</li>
+<li><span class="kbd">地震</span>/<span class="kbd">quake</span>：全球近 24 小時 M≥5.0</li>
+<li><span class="kbd">臺灣地震</span>/<span class="kbd">台灣地震</span>：今年台灣區域清單（含日期時間）</li>
+<li><span class="kbd">臺灣地震畫圖</span>/<span class="kbd">台灣地震畫圖</span>：回傳地圖圖片</li>
+<li><span class="kbd">地震預警</span>：CWA 地震預警（最新 5 筆）</li>
+<li><span class="kbd">ai 你的問題</span>：AI 對話（模型：<span class="mono">{config.LLM_MODEL}</span>）</li>
+</ul></div>
+<div class="card"><h3>🛠️ Webhook / 靜態檔</h3><ul>
+<li>Webhook：<span class="mono"><a href="{webhook_url}">{webhook_url}</a></span></li>
+<li>靜態圖片：<span class="mono">{static_hint}</span></li>
+<li>健康檢查：<span class="mono"><a href="{base}/healthz">{base}/healthz}</a></span></li>
+</ul></div>
+<div class="card"><h3>ℹ️ 備註</h3><ul>
+<li>AI 快取位置：<span class="mono">{config.TRANSFORMERS_CACHE}</span></li>
+<li>若 AI 未安裝依賴，機器人會提示安裝，不會影響其他功能。</li>
+</ul></div>
+</div>
+<p class="foot">© {config.CURRENT_YEAR} dayichen · server: {base}</p>
+</section></div></body></html>"""
+
+@app.route("/healthz")
+def healthz():
+    """健康檢查端點。"""
+    return "ok"
+
+@app.route("/static/<path:filename>")
+def serve_static(filename):
+    """提供���態檔案（例如，生成的地圖）。"""
+    return send_from_directory(config.STATIC_DIR, filename)
+
+# ------------------------------------------------------------------------------
+# LINE Webhook 處理器
+# ------------------------------------------------------------------------------
+@app.route("/callback", methods=["POST"])
+def callback():
+    """處理來自 LINE 平台的傳入 Webhooks。"""
+    signature = request.headers.get("X-Line-Signature")
+    body = request.get_data(as_text=True)
+    try:
+        handler.handle(body, signature)
+    except InvalidSignatureError:
+        abort(400)
+    return "OK"
+
+@handler.add(MessageEvent, message=TextMessageContent)
+def handle_message(event):
+    """
+    處理來自使用者的文字訊息並回覆。
+    所有邏輯都委派給 command_handler。
+    """
+    # 決定用於生成圖片連結的基礎 URL
+    base_url = request.url_root.rstrip("/")
+    
+    # 從處理器獲取回覆訊息
+    reply_messages = process_message(event.message.text, base_url)
+    
+    # 發送回覆
+    with ApiClient(line_config) as api_client:
+        line_bot_api = MessagingApi(api_client)
+        line_bot_api.reply_message_with_http_info(
+            ReplyMessageRequest(
+                reply_token=event.reply_token,
+                messages=reply_messages
+            )
+        )

command_handler.py

@@ -0,0 +1,87 @@
+# command_handler.py
+import pandas as pd
+from linebot.v3.messaging import TextMessage, ImageMessage
+
+# 匯入服務函式
+from cwa_service import fetch_cwa_alarm_list
+from usgs_service import fetch_global_last24h_text, fetch_taiwan_df_this_year
+from plotting_service import create_and_save_map
+from ai_service import generate_ai_text
+from config import CURRENT_YEAR, HF_SPACE_URL
+
+def get_help_message() -> TextMessage:
+    """回傳包含所有可用指令的說明訊息。"""
+    text = (
+        "📖 指令\n\n"
+        "• /help\n"
+        "• 地震 / quake（全球近24小時，含日期時間）\n"
+        "• 臺灣地震 / 台灣地震（今年台灣區域清單）\n"
+        "• 臺灣地震畫圖 / 台灣地震畫圖（今年台灣區域分佈圖）\n"
+        "• 地震預警（CWA 最新 5 筆）\n"
+        "• ai 你的問題（AI 對話）\n"
+        "• 你好"
+    )
+    return TextMessage(text=text)
+
+def get_taiwan_earthquake_list() -> TextMessage:
+    """回傳近期的台灣地震文字列表。"""
+    result = fetch_taiwan_df_this_year()
+    if isinstance(result, pd.DataFrame):
+        count = len(result)
+        lines = [f"🇹🇼 今年 ({CURRENT_YEAR} 年) 台灣區域顯著地震 (M≥5.0)，共 {count} 筆:", "-" * 20]
+        for _, row in result.head(15).iterrows():
+            t = row["time_utc"].strftime("%Y-%m-%d %H:%M")
+            lines.append(f"震級: {row['magnitude']:.1f} | 日期時間: {t} (UTC)\n地點: {row['place']}")
+        if count > 15:
+            lines.append(f"... (還有 {count - 15} 筆，可用「臺灣地震畫圖」查看全部)")
+        reply_text = "\n\n".join(lines)
+    else:
+        reply_text = result
+    return TextMessage(text=reply_text)
+
+def get_taiwan_earthquake_map(base_url: str) -> list:
+    """產生並回傳台灣地震地圖的訊息。"""
+    result = fetch_taiwan_df_this_year()
+    if isinstance(result, pd.DataFrame):
+        filename = create_and_save_map(result)
+        # 如果 HF_SPACE_URL 存在就使用它，否則使用請求的 base_url
+        image_url = f"{(HF_SPACE_URL or base_url)}/static/{filename}"
+        return [
+            TextMessage(text="🗺️ 已為您繪製今年台灣區域 M≥5.0 地震分佈圖（UTC）。"),
+            ImageMessage(original_content_url=image_url, preview_image_url=image_url),
+        ]
+    else:
+        return [TextMessage(text=result)]
+
+def process_message(user_message_raw: str, request_base_url: str) -> list:
+    """處理使用者的文字訊息並回傳一個包含回覆訊息的列表。"""
+    user_message = (user_message_raw or "").strip().lower()
+
+    if "地震預警" in user_message:
+        reply_text = fetch_cwa_alarm_list(limit=5)
+        return [TextMessage(text=reply_text)]
+
+    if "臺灣地震畫圖" in user_message or "台灣地震畫圖" in user_message:
+        return get_taiwan_earthquake_map(request_base_url)
+    
+    if "臺灣地震" in user_message or "台灣地震" in user_message:
+        return [get_taiwan_earthquake_list()]
+
+    if user_message == "/help":
+        return [get_help_message()]
+
+    if "地震" in user_message or "quake" in user_message:
+        reply_text = fetch_global_last24h_text()
+        return [TextMessage(text=reply_text)]
+
+    if user_message.startswith("ai ") or user_message.startswith("ai:") or user_message.startswith("ai："):
+        prompt = user_message_raw[2:].lstrip(" ：:").strip() or "請簡要介紹你的功能。"
+        ai_text = generate_ai_text(prompt)
+        return [TextMessage(text=ai_text)]
+
+    if "你好" in user_message or "hi" in user_message:
+        return [TextMessage(text="👋 你好！輸入 /help 查看指令。")]
+    
+    # 對於所有其他訊息，使用 AI 作為備援回覆
+    fallback_text = generate_ai_text(user_message_raw)
+    return [TextMessage(text=fallback_text)]

config.py

@@ -0,0 +1,43 @@
+# config.py
+import os
+import tempfile
+from datetime import datetime
+
+# --- 環境設定 ---
+# 設定 Matplotlib 與 Hugging Face 模型的快取目錄
+os.environ.setdefault("MPLCONFIGDIR", "/tmp/matplotlib")
+os.environ.setdefault("TRANSFORMERS_CACHE", "/tmp/huggingface")
+os.makedirs(os.environ["TRANSFORMERS_CACHE"], exist_ok=True)
+
+# --- LINE Bot 憑證 ---
+CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN")
+CHANNEL_SECRET = os.getenv("CHANNEL_SECRET")
+
+# --- Hugging Face Space URL ---
+HF_SPACE_URL = os.getenv("SPACEURL")
+if not HF_SPACE_URL:
+    sid = os.getenv("SPACE_ID")
+    if sid and "/" in sid:
+        author, name = sid.split("/", 1)
+        HF_SPACE_URL = f"https://{author.replace('_', '-')}-{name.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)
+
+# --- API 端點 ---
+CWA_ALARM_API = "https://app-2.cwa.gov.tw/api/v1/earthquake/alarm/list"
+USGS_API_BASE_URL = "https://earthquake.usgs.gov/fdsnws/event/1/query"
+
+# --- AI 模型設定 ---
+LLM_DEVICE = os.getenv("LLM_DEVICE")
+LLM_THREADS = os.getenv("LLM_THREADS", "1")
+LLM_MODEL = os.getenv("LLM_MODEL", "ckiplab/gpt2-base-chinese")
+LLM_MAX_NEW_TOKENS = int(os.getenv("LLM_MAX_NEW_TOKENS", "120"))
+LLM_TOP_K = int(os.getenv("LLM_TOP_K", "50"))
+LLM_TEMPERATURE = float(os.getenv("LLM_TEMPERATURE", "0.7"))
+
+# --- 顯示用當年年份 ---
+CURRENT_YEAR = datetime.now().year

cwa_service.py

@@ -0,0 +1,97 @@
+# cwa_service.py
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+import requests
+from datetime import datetime, timedelta, timezone
+from config import CWA_ALARM_API
+
+def _parse_cwa_time(s: str) -> tuple[str, str]:
+    """回傳 (台灣時間, UTC)；若字串無時區，預設視為台灣時間。"""
+    if not s:
+        return ("未知", "未知")
+    try:
+        if "T" in s or s.endswith("Z") or "+" in s:
+            dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
+        else:
+            dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S")
+            dt = dt.replace(tzinfo=timezone(timedelta(hours=8)))
+        tw = dt.astimezone(timezone(timedelta(hours=8))).strftime("%Y-%m-%d %H:%M")
+        utc = dt.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M")
+        return (tw, utc)
+    except Exception:
+        return (s, "未知")
+
+def fetch_cwa_alarm_list(limit: int = 5) -> str:
+    """抓 CWA 地震預警並格式化輸出。"""
+    try:
+        r = requests.get(CWA_ALARM_API, timeout=10)
+        r.raise_for_status()
+        payload = r.json()
+    except Exception as e:
+        return f"❌ 地震預警查詢失敗：{e}"
+
+    items = None
+    if isinstance(payload, dict):
+        items = payload.get("data") or payload.get("records") or payload.get("list") or payload.get("items")
+    if items is None and isinstance(payload, list):
+        items = payload
+    if not items:
+        return "✅ 目前沒有地震預警。"
+
+    def _key(it):
+        s = it.get("originTime") or ""
+        try:
+            if "T" in s or s.endswith("Z") or "+" in s:
+                dt = datetime.fromisoformat(s.replace("Z", "+00:00"))
+            else:
+                dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S").replace(tzinfo=timezone(timedelta(hours=8)))
+            return dt.astimezone(timezone.utc)
+        except Exception:
+            return datetime.min.replace(tzinfo=timezone.utc)
+
+    try:
+        items = sorted(items, key=_key, reverse=True)
+    except Exception:
+        pass
+
+    def _num(x):
+        xs = str(x)
+        ok = xs.replace(".", "", 1).replace("-", "", 1).isdigit()
+        return float(xs) if ok else None
+
+    lines = ["🚨 地震預警（最新）:", "-" * 20]
+    for idx, it in enumerate(items[:limit]):
+        identifier = it.get("identifier") or it.get("eventId") or it.get("id") or "—"
+        status = it.get("status") or "—"
+        msg_type = it.get("msgType") or "—"
+        msg_no = it.get("msgNo") or it.get("msgSeq") or "—"
+        mag = _num(it.get("magnitudeValue") or it.get("magnitude") or it.get("ml") or it.get("mw"))
+        mag_str = f"{mag:.1f}" if mag is not None else "—"
+        depth = _num(it.get("depth"))
+        depth_str = f"{depth:.0f}" if depth is not None else "—"
+        lat = _num(it.get("epicenterLat") or it.get("latitude") or it.get("lat"))
+        lon = _num(it.get("epicenterLon") or it.get("longitude") or it.get("lon"))
+        lat_str = f"{lat:.2f}" if lat is not None else "—"
+        lon_str = f"{lon:.2f}" if lon is not None else "—"
+        origin = it.get("originTime") or ""
+        tw_str, utc_str = _parse_cwa_time(origin)
+        areas = it.get("locationDesc") or it.get("areas") or it.get("alertAreas")
+        if isinstance(areas, list):
+            areas_txt = "、".join(str(a) for a in areas if a)
+        elif isinstance(areas, str):
+            areas_txt = areas
+        else:
+            areas_txt = "—"
+        lines.append(
+            f"事件: {identifier} | 狀態: {status} | 類型: {msg_type}#{msg_no}\n"
+            f"震級/深度: M{mag_str} / {depth_str} km\n"
+            f"震中: lat {lat_str}, lon {lon_str}\n"
+            f"時間: {tw_str}（台灣） / {utc_str}（UTC）\n"
+            f"預警地區: {areas_txt}"
+        )
+        lines.append("")
+
+    if len(items) > limit:
+        lines.append(f"... 另有 {len(items) - limit} 筆。")
+
+    return "\n".join(lines).strip()

plotting_service.py

@@ -0,0 +1,53 @@
+# plotting_service.py
+import os
+import uuid
+import pandas as pd
+import matplotlib
+matplotlib.use("Agg")  # 使用非互動式後端
+import matplotlib.pyplot as plt
+from matplotlib.colors import Normalize
+import matplotlib.cm as cm
+from matplotlib import font_manager as fm
+from config import STATIC_DIR, CURRENT_YEAR
+
+def setup_chinese_font():
+    """如果系統中存在，則為 Matplotlib 設定中文字體。"""
+    font_paths = [
+        "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc",
+        "/usr/share/fonts/truetype/noto/NotoSansCJK-Regular.ttc",
+        "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc",
+        "/usr/share/fonts/truetype/arphic/ukai.ttc",
+    ]
+    for fp in font_paths:
+        if os.path.exists(fp):
+            matplotlib.rcParams["font.family"] = fm.FontProperties(fname=fp).get_name()
+            break
+
+def create_and_save_map(df: pd.DataFrame) -> str:
+    """建立地震地圖，儲存圖片並回傳檔案名稱。"""
+    setup_chinese_font()
+    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"今年 ({CURRENT_YEAR}) 台灣區域顯著地震 (M≥5.0) — UTC")
+    ax.grid(True, linestyle="--", linewidth=0.5, alpha=0.4)
+
+    mags = df["magnitude"].astype(float).clip(lower=0)
+    norm = Normalize(vmin=max(4.5, mags.min()), vmax=max(6.5, mags.max()))
+    cmap = cm.get_cmap("YlOrRd")
+    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)
+    fig.tight_layout()
+    fig.savefig(filepath)
+    plt.close(fig)
+    return filename

requirements.txt

@@ -0,0 +1,10 @@
+flask
+gunicorn
+line-bot-sdk>=3.0.0,<4
+requests
+pandas
+matplotlib
+transformers
+torch
+accelerate
+safetensors

usgs_service.py

@@ -0,0 +1,65 @@
+# usgs_service.py
+import requests
+import pandas as pd
+from datetime import datetime, timedelta, timezone
+from config import USGS_API_BASE_URL, CURRENT_YEAR
+
+def _iso(dt: datetime) -> str:
+    """將 datetime 物件格式化為 USGS API 需要的 ISO 8601 字串。"""
+    return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")
+
+def fetch_global_last24h_text(min_mag: float = 5.0, limit: int = 10) -> str:
+    """從 USGS 擷取過去 24 小時的全球顯著地震。"""
+    now_utc = datetime.now(timezone.utc)
+    since = now_utc - timedelta(hours=24)
+    params = {
+        "format": "geojson",
+        "starttime": _iso(since),
+        "endtime": _iso(now_utc),
+        "minmagnitude": float(min_mag),
+        "limit": int(limit),
+        "orderby": "time",
+    }
+    try:
+        r = requests.get(USGS_API_BASE_URL, params=params, timeout=15)
+        r.raise_for_status()
+        features = r.json().get("features", [])
+        if not features:
+            return f"✅ 過去 24 小時內，全球無規模 {min_mag} 以上的顯著地震。"
+        lines = [f"🚨 近 24 小時全球顯著地震 (M≥{min_mag}):", "-" * 20]
+        for f in features:
+            p = f["properties"]
+            t_utc = datetime.fromtimestamp(p["time"] / 1000, tz=timezone.utc)
+            lines.append(f"震級: {p['mag']:.1f} | 日期時間: {t_utc.strftime('%Y-%m-%d %H:%M')} (UTC)\n地點: {p.get('place','')}")
+        return "\n\n".join(lines)
+    except Exception as e:
+        return f"❌ 查詢失敗: {e}"
+
+def fetch_taiwan_df_this_year(min_mag: float = 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)
+        r.raise_for_status()
+        features = r.json().get("features", [])
+        if not features:
+            return f"✅ 今年 ({CURRENT_YEAR} 年) 以來，台灣區域無 M≥{min_mag:.1f} 的顯著地震。"
+        rows = []
+        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}"