""" LINE Bot 相關工具函數 """ from linebot.models import ( TextSendMessage, QuickReply, QuickReplyButton, MessageAction, FlexSendMessage, BubbleContainer, BoxComponent, TextComponent, ButtonComponent, URIAction ) from typing import List, Dict, Any import logging logger = logging.getLogger(__name__) class LineMessageBuilder: """LINE 訊息建構器""" @staticmethod def create_quick_reply_message(text: str, quick_reply_items: List[Dict[str, str]]) -> TextSendMessage: """建立快速回覆訊息""" quick_reply_buttons = [] for item in quick_reply_items: button = QuickReplyButton( action=MessageAction( label=item["label"], text=item["text"] ) ) quick_reply_buttons.append(button) quick_reply = QuickReply(items=quick_reply_buttons) return TextSendMessage( text=text, quick_reply=quick_reply ) @staticmethod def create_product_flex_message(products: List[Dict[str, Any]]) -> FlexSendMessage: """建立商品展示的 Flex 訊息""" contents = [] for product in products[:10]: # 最多顯示10個商品 bubble = BubbleContainer( body=BoxComponent( layout="vertical", contents=[ TextComponent( text=product.get("name", "商品名稱"), weight="bold", size="lg" ), TextComponent( text=f"NT$ {product.get('price', 0):,}", size="md", color="#ff5551" ), TextComponent( text=f"庫存: {product.get('stock', 0)}", size="sm", color="#aaaaaa" ) ] ), footer=BoxComponent( layout="vertical", contents=[ ButtonComponent( action=MessageAction( label="查看詳情", text=f"查詢商品 {product.get('name', '')}" ), style="primary" ) ] ) ) contents.append(bubble) return FlexSendMessage( alt_text="商品列表", contents={ "type": "carousel", "contents": [bubble.as_json_dict() for bubble in contents] } ) @staticmethod def create_help_message() -> TextSendMessage: """建立說明訊息""" help_text = """ 🤖 歡迎使用智能查詢機器人! 📝 支援的查詢類型: • 用戶查詢:「查詢用戶 張三」 • 訂單查詢:「查詢訂單 ORD001」 • 商品查詢:「查詢商品 iPhone」 • 價格範圍:「價格 1000 到 5000 的商品」 • 統計分析:「統計用戶數量」 💡 使用範例: • 「找用戶名叫王小明的資料」 • 「我的訂單狀態如何」 • 「有什麼手機商品」 • 「總共有多少筆訂單」 ❓ 如需協助,請輸入「幫助」或「說明」 """ quick_reply_items = [ {"label": "查詢用戶", "text": "查詢用戶"}, {"label": "查詢訂單", "text": "查詢訂單"}, {"label": "查詢商品", "text": "查詢商品"}, {"label": "統計資料", "text": "統計"} ] return LineMessageBuilder.create_quick_reply_message(help_text, quick_reply_items) @staticmethod def create_error_message(error_msg: str = None) -> TextSendMessage: """建立錯誤訊息""" default_msg = "抱歉,我無法理解您的訊息。請輸入「幫助」查看使用說明。" text = error_msg if error_msg else default_msg quick_reply_items = [ {"label": "幫助", "text": "幫助"}, {"label": "重新開始", "text": "開始"} ] return LineMessageBuilder.create_quick_reply_message(text, quick_reply_items) def get_user_display_name(line_bot_api, user_id: str) -> str: """取得用戶顯示名稱""" try: profile = line_bot_api.get_profile(user_id) return profile.display_name except Exception as e: logger.error(f"取得用戶資料失敗: {str(e)}") return "用戶" def validate_line_signature(body: bytes, signature: str, channel_secret: str) -> bool: """驗證 LINE Webhook 簽名""" import hashlib import hmac import base64 hash_value = hmac.new( channel_secret.encode('utf-8'), body, hashlib.sha256 ).digest() expected_signature = base64.b64encode(hash_value).decode('utf-8') return hmac.compare_digest(signature, expected_signature)