app.py

"""
app.py
------
LINE Messaging API webhook（Flask）。
  • 收文字 → 回股價 Flex 圖卡 / 指令回覆
  • 收 postback → 處理「加入自選 / 到價提醒」按鈕
  • /tasks/check-alerts → 供排程呼叫，觸發到價提醒推播
"""

import os
import logging

from flask import Flask, request, abort, jsonify

from linebot.v3 import WebhookHandler
from linebot.v3.exceptions import InvalidSignatureError
from linebot.v3.messaging import (
    Configuration,
    ApiClient,
    MessagingApi,
    ReplyMessageRequest,
    PushMessageRequest,
    TextMessage,
    FlexMessage,
    FlexContainer,
    ShowLoadingAnimationRequest,
)
from linebot.v3.webhooks import MessageEvent, TextMessageContent, PostbackEvent

import services

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

CHANNEL_SECRET = os.environ.get("LINE_CHANNEL_SECRET")
CHANNEL_ACCESS_TOKEN = os.environ.get("LINE_CHANNEL_ACCESS_TOKEN")
CRON_SECRET = os.environ.get("CRON_SECRET")  # 保護 /tasks/check-alerts

if not CHANNEL_SECRET or not CHANNEL_ACCESS_TOKEN:
    logger.warning("尚未設定 LINE_CHANNEL_SECRET / LINE_CHANNEL_ACCESS_TOKEN。")
if not os.environ.get("GEMINI_API_KEY"):
    logger.warning("尚未設定 GEMINI_API_KEY，新聞/展望與中文股名解析將無法使用。")

app = Flask(__name__)
configuration = Configuration(access_token=CHANNEL_ACCESS_TOKEN)
handler = WebhookHandler(CHANNEL_SECRET or "dummy")


# ------------------------------------------------------------------ #
# 健康檢查（順便顯示環境變數是否設好，方便除錯）
# ------------------------------------------------------------------ #
@app.route("/", methods=["GET"])
def health():
    return jsonify({
        "ok": True,
        "line_secret": bool(CHANNEL_SECRET),
        "line_token": bool(CHANNEL_ACCESS_TOKEN),
        "gemini": bool(os.environ.get("GEMINI_API_KEY")),
        "model": services.GEMINI_MODEL,
    })


@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:
        logger.warning("簽章驗證失敗，請確認 Channel secret。")
        abort(400)
    return "OK"


# ------------------------------------------------------------------ #
# 訊息與按鈕
# ------------------------------------------------------------------ #
def _to_messages(result):
    if result.get("kind") == "text":
        return [TextMessage(text=result["text"])]
    display_name, quote, body = result["display_name"], result["quote"], result["body"]
    try:
        bubble = services.build_stock_bubble(display_name, quote, body)
        return [FlexMessage(alt_text=services.stock_alt_text(display_name, quote),
                            contents=FlexContainer.from_dict(bubble))]
    except Exception:  # noqa: BLE001
        logger.exception("建立 Flex 訊息失敗，改用純文字")
        return [TextMessage(text=services.stock_summary_text(display_name, quote, body))]


def _reply(line_bot_api, event, messages):
    line_bot_api.reply_message(
        ReplyMessageRequest(reply_token=event.reply_token, messages=messages))


@handler.add(MessageEvent, message=TextMessageContent)
def handle_message(event):
    user_id = getattr(event.source, "user_id", None) or "anon"
    with ApiClient(configuration) as api_client:
        line_bot_api = MessagingApi(api_client)
        try:
            if user_id != "anon":
                line_bot_api.show_loading_animation(
                    ShowLoadingAnimationRequest(chat_id=user_id, loading_seconds=30))
        except Exception as e:  # noqa: BLE001
            logger.info("loading 動畫略過: %s", e)

        try:
            messages = _to_messages(services.build_reply(event.message.text, user_id))
        except Exception:  # noqa: BLE001
            logger.exception("build_reply 發生錯誤")
            messages = [TextMessage(text="查詢時發生錯誤，請稍後再試 🙏")]
        _reply(line_bot_api, event, messages)


@handler.add(PostbackEvent)
def handle_postback(event):
    user_id = getattr(event.source, "user_id", None) or "anon"
    with ApiClient(configuration) as api_client:
        line_bot_api = MessagingApi(api_client)
        try:
            text = services.handle_postback(event.postback.data, user_id)
        except Exception:  # noqa: BLE001
            logger.exception("handle_postback 發生錯誤")
            text = "操作失敗，請稍後再試 🙏"
        _reply(line_bot_api, event, [TextMessage(text=text)])


# ------------------------------------------------------------------ #
# 到價提醒檢查（供 cron / 排程呼叫）
#   例：每 5 分鐘打一次
#   curl -X POST https://你的網址/tasks/check-alerts -H "X-Cron-Secret: <CRON_SECRET>"
# ------------------------------------------------------------------ #
@app.route("/tasks/check-alerts", methods=["POST", "GET"])
def check_alerts():
    if CRON_SECRET and request.headers.get("X-Cron-Secret") != CRON_SECRET:
        abort(403)

    def push_fn(uid, text):
        with ApiClient(configuration) as api_client:
            MessagingApi(api_client).push_message(
                PushMessageRequest(to=uid, messages=[TextMessage(text=text)]))

    try:
        n = services.check_alerts(push_fn)
    except Exception:  # noqa: BLE001
        logger.exception("check_alerts 失敗")
        return jsonify({"ok": False}), 500
    return jsonify({"ok": True, "triggered": n})


if __name__ == "__main__":
    port = int(os.environ.get("PORT", 5000))
    app.run(host="0.0.0.0", port=port)