Spaces:
Sleeping
Sleeping
| 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://<space-name>.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 --- | |
| 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> | |
| """ | |
| def healthz(): | |
| return "ok" | |
| 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"<b>今年 ({datetime.now(timezone.utc).year}) 台灣區域顯著地震 (M≥5.0)</b>", | |
| 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 --- | |
| 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 --- | |
| 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 |