LINE-ROBOT

Sleeping

File size: 10,552 Bytes

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 ---
@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():
    return "ok"

@app.route("/static/<path:filename>")
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 ---
@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