File size: 5,246 Bytes
cd9bca9 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 |
"""
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) |