import os import uuid from datetime import datetime, timedelta, timezone 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, TextMessage ) # ✅ LINE v3: use ImageMessage (not ImageSendMessage) from linebot.v3.messaging.models import ImageMessage from linebot.v3.webhooks import MessageEvent, TextMessageContent import requests import pandas as pd import plotly.express as px # --- Environment Variables --- CHANNEL_ACCESS_TOKEN = os.getenv("CHANNEL_ACCESS_TOKEN") CHANNEL_SECRET = os.getenv("CHANNEL_SECRET") HF_SPACE_URL = os.getenv("SPACEURL") # optional: https://.hf.space # --- Flask & LINE Bot Initialization --- app = Flask(__name__) os.makedirs("static", exist_ok=True) configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN) handler = WebhookHandler(CHANNEL_SECRET) # --- Welcome & Health --- @app.route("/", methods=["GET"]) def home(): return """ LINE Bot Server

✓ LINE Bot Server is Running

This is the backend service for the Earthquake Alert Bot.

The service is active and listening for webhook events from LINE.

""" @app.route("/healthz") def healthz(): return "ok" @app.route("/static/") def serve_static(filename): return send_from_directory("static", filename) # --- Earthquake Query Logic --- 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 = { "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 "✅ 過去 24 小時內，全球無規模 5.0 以上的顯著地震。" lines = ["🚨 近 24 小時全球顯著地震 (M≥5.0):", "-" * 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('%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=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"✅ 今年 ({now_utc.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}" # --- Offline-friendly Map (PNG) --- def create_and_save_map(df: pd.DataFrame) -> str: # scatter_geo → renders via kaleido without external tiles fig = px.scatter_geo( df, lat="latitude", lon="longitude", size="magnitude", color="magnitude", hover_name="place", hover_data={"magnitude": ":.1f", "time_utc": "|%Y-%m-%d %H:%M UTC"}, size_max=24, color_continuous_scale=px.colors.sequential.YlOrRd, projection="natural earth", ) fig.update_layout( title=f"今年 ({datetime.now(timezone.utc).year}) 台灣區域顯著地震 (M≥5.0)", margin=dict(r=0, t=40, l=0, b=0), ) fig.update_geos( lonaxis_range=[118.5, 123.5], lataxis_range=[20.5, 26.8], showcountries=True, showland=True, landcolor="#f8f8f8", countrycolor="#aaa", ) filename = f"map_{uuid.uuid4().hex}.png" filepath = os.path.join("static", filename) # Requires `kaleido` in requirements.txt fig.write_image(filepath, scale=2, width=900, height=600) return filename def _base_url_for_images() -> str: if HF_SPACE_URL: return HF_SPACE_URL.rstrip("/") return request.url_root.rstrip("/") # --- LINE Webhook --- @app.route("/callback", methods=["POST"]) def callback(): 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" # --- 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) # Taiwan map command if ("臺灣地震畫圖" in user_message) or ("台灣地震畫圖" in user_message): result = fetch_taiwan_df_this_year() if isinstance(result, pd.DataFrame): filename = create_and_save_map(result) image_url = f"{_base_url_for_images()}/static/{filename}" reply = ReplyMessageRequest( 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 # Help if user_message == "/help": text = ( "📖 地震預警 dayichen 指令說明\n\n" "➡️ /help\n 說明：顯示此幫助訊息。\n\n" "➡️ 地震\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 說明：顯示歡迎訊息。" ) line_bot_api.reply_message_with_http_info( ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=text)]) ) return # Taiwan list if ("臺灣地震" in user_message) or ("台灣地震" in user_message): result = fetch_taiwan_df_this_year() if isinstance(result, pd.DataFrame): count = len(result) lines = [f"🇹🇼 今年 ({datetime.now(timezone.utc).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 line_bot_api.reply_message_with_http_info( ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=reply_text)]) ) return # Global last 24h if ("地震" in user_message) or ("quake" in user_message): reply_text = fetch_global_last24h_text() line_bot_api.reply_message_with_http_info( ReplyMessageRequest(reply_token=event.reply_token, messages=[TextMessage(text=reply_text)]) ) return # Greeting 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 # No-op on unmatched text (keeps reply_token free for other handlers) return