from flask import Blueprint, request, abort, jsonify, current_app from linebot import LineBotApi, WebhookHandler from linebot.exceptions import InvalidSignatureError from linebot.models import MessageEvent, TextMessage, TextSendMessage, ImageSendMessage import database from routes.main import main_bp import os import math import time # 嘗試載入 Pillow 影像處理套件 try: from PIL import Image, ImageDraw except ImportError: Image = None ImageDraw = None line_bp = Blueprint('line', __name__) # @850urstz line_bot_api = LineBotApi('BcFuVwO4mi8R5ORj0OQDDUbaRcDevCE62LMH/yXuRjMmHDIbzye8awOHW9XaQFXxrNJpvad52R015Smi32bu86OXZB/LZHOqcqSfyHF127yd8Zc/6QhceDTDc3WdZb07y0JKgfySeaU5v48veHBLZwdB04t89/1O/w1cDnyilFU=') handler = WebhookHandler('d629e2c90aaee9c49a71343488e7d4db') # # @586xqzkx # line_bot_api = LineBotApi('jRl2RVie79yr1f3ysUHZXBZFFmtYpA0MYNWm5Dbfxi9seIVTIVMIMgS5jPcVvLue5gyJ4OIjxLdOJCQMBckn29n/m0WIXs5WTdKRX5B1ad3iJJ5WasGdlAX6HwMhNn7HltvONuzp5osVopFL3d2jrQdB04t89/1O/w1cDnyilFU=') # handler = WebhookHandler('2cf53ea67c0687197ab3900578ef60e5') # 圖片合成 def generate_location_image(uid, x_str, y_str): """根據 x, y 座標生成地圖圖片，並回傳 Line 的 ImageSendMessage 物件""" if not Image or not ImageDraw: print("❌ LINE 圖片生成失敗：伺服器未安裝 Pillow 套件") return None try: x_val = float(x_str) if x_str not in [None, '---', ''] else float('nan') y_val = float(y_str) if y_str not in [None, '---', ''] else float('nan') current_dir = os.path.dirname(os.path.abspath(__file__)) # 取得 line.py 的目錄 (routes) current_dir = os.path.dirname(current_dir) # 退回 mmwave 主目錄 # bg_path = os.path.join(current_dir, 'static', 'images', 'map_304.png') bg_path = os.path.join(current_dir, 'static', 'images', 'map_lab.png') if not os.path.exists(bg_path): print(f"❌ LINE 圖片生成失敗：找不到地圖底圖，路徑錯誤 ({bg_path})") return None out_dir = os.path.dirname(bg_path) out_filename = f'loc_{uid}.png' out_path = os.path.join(out_dir, out_filename) raw_img = Image.open(bg_path).convert("RGBA") raw_width, raw_height = raw_img.size padding = 30 new_width = raw_width + padding * 2 new_height = raw_height + padding * 2 white_bg = Image.new("RGBA", (new_width, new_height), (255, 255, 255, 255)) white_bg.paste(raw_img, (padding, padding), raw_img) img = white_bg.convert("RGB") if not math.isnan(x_val) and not math.isnan(y_val): draw = ImageDraw.Draw(img) scale = raw_width / 3.62 radar_x = raw_width + padding radar_y = raw_height + padding r = math.sqrt(x_val**2 + y_val**2) theta = math.atan2(x_val, y_val) screen_angle = 1.25 * math.pi + theta px = radar_x + r * scale * math.cos(screen_angle) py = radar_y + r * scale * math.sin(screen_angle) dot_r = 10 draw.ellipse( [(px - dot_r, py - dot_r), (px + dot_r, py + dot_r)], fill="#0d6efd", outline="white", width=2 ) img.save(out_path) print(f"✅ 成功生成地圖圖片：{out_path}") ts = int(time.time()) img_url = f"https://mmwave.duckdns.org/mmwave/static/images/{out_filename}?t={ts}" return ImageSendMessage(original_content_url=img_url, preview_image_url=img_url) except Exception as e: print(f"❌ LINE 地圖生成發生未知錯誤: {e}") return None def build_event_message(event_type, owner_name, device_name, receiver_name, receiver_role, event_id, zone="無人"): """組合通知訊息內容 (提示使用圖文選單)""" title_map = { 'fall': '🚨 跌倒警報', 'no_move': '⚠️ 異常靜止', 'no_home': 'ℹ️ 人已外出', 'fall_unresponsive': '🆘 跌倒未處理緊急通報' } action_map = { 'fall': '跌倒', 'no_move': '長時間不動', 'no_home': '不在家 / 外出', 'fall_unresponsive': '跌倒後倒地不起 (具失去意識風險)' } role_map = {'admin': '管理者', 'family': '家屬', 'caregiver': '照護人員'} title = title_map.get(event_type, '⚠️ 異常事件') action_name = action_map.get(event_type, event_type) now_str = database.datetime.now(database.pytz.timezone('Asia/Taipei')).strftime('%Y-%m-%d %H:%M:%S') role_name = role_map.get(receiver_role, receiver_role) return (f"{title}\n" f"接收帳號：{role_name}（{receiver_name}）\n" f"受照護者：{owner_name}\n" f"發生位置：{zone}\n" f"狀態：{action_name}\n" f"裝置：{device_name}\n" f"時間：{now_str}\n" "──────────────\n" "請確認安全後點擊下方選單的「解除警報」") # ========================================================= # 主動發送警報的共用函式 (供 database.py 呼叫) def broadcast_new_alert(event_type, owner_name, device_name, event_id, x_str, y_str): """當系統偵測到新警報時，呼叫此函式來發送包含圖片的 LINE 推播給所有家屬""" print(f"準備發送 {event_type} 警報推播...") # 1. 產生當下位置的圖片 img_msg = generate_location_image(f"alert_{event_id}", x_str, y_str) try: db = database.get_db_connection() with db.cursor() as cursor: # 撈出所有有綁定且開啟通知的用戶 cursor.execute("SELECT user_id, name, role, line_user_id FROM users WHERE line_user_id IS NOT NULL AND notify_enabled = 1 AND is_active = 1") recipients = cursor.fetchall() if not recipients: print("⚠️ 找不到符合條件的收件人 (請檢查是否綁定且開啟通知)") return for r in recipients: uid = r['line_user_id'] # 組合文字訊息 text_str = build_event_message( event_type, owner_name, device_name, r['name'], r['role'], event_id ) msgs = [TextSendMessage(text=text_str)] if img_msg: msgs.append(img_msg) # 把圖片加進推播清單 # 執行發送推播 line_bot_api.push_message(uid, msgs) print(f"✅ 成功發送警報給 {r['name']}") # 自動寫入推播紀錄 notifications cursor.execute("INSERT INTO notifications (event_id, user_id, sent_time) VALUES (%s, %s, NOW())", (event_id, r['user_id'])) db.commit() except Exception as e: print(f"❌ 主動發送警報推播發生錯誤: {e}") finally: if 'db' in locals() and db: db.close() # ========================================================= @line_bp.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' @handler.add(MessageEvent, message=TextMessage) def handle_message(event): msg = event.message.text.strip().lower() uid = event.source.user_id if msg.startswith('bind '): _, code = msg.split(maxsplit=1) _, reply = database.bind_line_user(code.upper(), uid) line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply)) elif msg.startswith('unbind '): _, code = msg.split(maxsplit=1) _, reply = database.unbind_line_user(uid, code.upper()) line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply)) elif msg in ['status', '目前狀態']: state = database.get_current_ui_state(line_id=uid) user = database.get_user_by_line_id(uid) role = user['role'] if user else 'guest' # 針對 Caregiver 的極簡回覆 if role == 'caregiver': is_home = state.get('raw_activity', 'null') != 'null' status_text = "在家" if is_home else "不在家 / 外出" text_reply = ( f"【目前狀態】\n" f"狀態：{status_text}\n" f"更新時間：{state.get('last_activity_time', '---')}" ) line_bot_api.reply_message(event.reply_token, [TextSendMessage(text=text_reply)]) else: text_reply = ( f"【目前狀態】\n" f"狀態：{state.get('title', '未知')}\n" f"位置：{state.get('zone', '無人')}\n" f"更新時間：{state.get('last_activity_time', '---')}\n" f"說明：{state.get('desc', '無')}" ) reply_msgs = [TextSendMessage(text=text_reply)] img_msg = generate_location_image(uid, state.get('x'), state.get('y')) if img_msg: reply_msgs.append(img_msg) line_bot_api.reply_message(event.reply_token, reply_msgs) elif msg == '解除警報': resolver_name = database.resolve_all_alerts_by_line(uid) if resolver_name == "NO_ALERTS": line_bot_api.reply_message(event.reply_token, TextSendMessage(text="✅ 目前沒有需要解除的警報喔！")) elif resolver_name: state = database.get_current_ui_state() current_app.socketio.emit('ui_update', state, namespace='/') current_app.socketio.emit('force_reload', namespace='/') user = database.get_user_by_line_id(uid) role = user['role'] if user else 'guest' # 本人的回覆 if role == 'caregiver': is_home = state.get('raw_activity', 'null') != 'null' status_text = "在家" if is_home else "不在家 / 外出" reply_text = ( f"🛡️ 警報已由 {resolver_name} 解除！\n" "──────────────\n" f"【目前狀態】\n" f"狀態：{status_text}\n" f"更新時間：{state.get('last_activity_time', '---')}" ) line_bot_api.reply_message(event.reply_token, [TextSendMessage(text=reply_text)]) else: reply_text = ( f"🛡️ 警報已由 {resolver_name} 解除！\n" "──────────────\n" f"【目前狀態】\n" f"狀態：{state.get('title', '未知')}\n" f"位置：{state.get('zone', '無人')}\n" f"更新時間：{state.get('last_activity_time', '---')}\n" f"說明：{state.get('desc', '無')}" ) reply_msgs = [TextSendMessage(text=reply_text)] img_msg = generate_location_image('broadcast', state.get('x'), state.get('y')) if img_msg: reply_msgs.append(img_msg) line_bot_api.reply_message(event.reply_token, reply_msgs) # 推播給其他人 try: db = database.get_db_connection() with db.cursor() as cursor: cursor.execute("SELECT line_user_id, role FROM users WHERE line_user_id IS NOT NULL AND notify_enabled = 1 AND is_active = 1") recipients = cursor.fetchall() db.close() if recipients: for r in recipients: target_line_id = r['line_user_id'] target_role = r['role'] if target_line_id != uid: # 針對照護者的精簡推播 if target_role == 'caregiver': is_home = state.get('raw_activity', 'null') != 'null' status_text = "在家" if is_home else "不在家 / 外出" cg_text = ( f"🛡️ 警報已由 {resolver_name} 解除！\n" "──────────────\n" f"【目前狀態】\n" f"狀態：{status_text}\n" f"更新時間：{state.get('last_activity_time', '---')}" ) line_bot_api.push_message(target_line_id, TextSendMessage(text=cg_text)) else: # 給一般家屬的推播 (含詳細文字與圖片) line_bot_api.push_message(target_line_id, reply_msgs) except Exception as e: print(f"LINE解除警報後，同步推播給其他人失敗: {e}") else: line_bot_api.reply_message(event.reply_token, TextSendMessage(text="⚠️ 找不到您的帳號，請確認是否已綁定系統。")) else: line_bot_api.reply_message(event.reply_token, TextSendMessage(text="🤖 系統指令：\n1. 目前狀態\n2. 解除警報\n3. 綁定Line：bind [代碼]"))