# app.py import os import uuid import requests import logging import time # 💡 導入時間模組進行效能測試 from fastapi import FastAPI, Request, HTTPException from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse from linebot import LineBotApi, WebhookHandler from linebot.exceptions import InvalidSignatureError from linebot.models import ( MessageEvent, TextMessage, TextSendMessage, AudioSendMessage, FlexSendMessage, VideoMessage, ImageMessage, AudioMessage, ImageSendMessage, PostbackEvent, FollowEvent ) # 💡 導入專案模組 import image_processor import config import translator import audio_processor import video_processor import ui_templates import db_manager import deep_analyzer import visual_renderer # 設定日誌 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI() # 確保靜態資料夾存在 if not os.path.exists("static"): os.makedirs("static") app.mount("/static", StaticFiles(directory="static"), name="static") line_bot_api = LineBotApi(config.LINE_TOKEN) handler = WebhookHandler(config.LINE_SECRET) # ---------------------------------------------------------------- # 🔮 智慧翻譯快取攔截器 (Priority 2: LRU Cache) # ---------------------------------------------------------------- from functools import lru_cache @lru_cache(maxsize=1024) def cached_translate(text: str, tribe: str): """記憶體快取：如果相同的句子與族語查過，直接秒回，不再去遠端 API 排隊 5 秒""" logger.info(f"🔮 [Cache Miss] 正在向遠端 API 請求全新翻譯: {text[:10]}...") return translator.translate(text, tribe) # ---------------------------------------------------------------- # 輔助功能 # ---------------------------------------------------------------- @app.get("/download_file/{filename}") async def download_file(filename: str): """供使用者下載生成的音檔或檔案""" file_path = os.path.join("static", filename) if os.path.exists(file_path): return FileResponse(path=file_path, filename=f"TRK_AI_{filename}", media_type='application/octet-stream') return {"error": "檔案不存在"} def show_loading_animation(user_id, loading_seconds=20): """發送 LINE 官方的讀取中動畫""" url = "https://api.line.me/v2/bot/chat/loading/start" headers = {"Content-Type": "application/json", "Authorization": f"Bearer {config.LINE_TOKEN}"} data = {"chatId": user_id, "loadingSeconds": loading_seconds} try: requests.post(url, headers=headers, json=data, timeout=5) except Exception as e: logger.error(f"Loading Animation Error: {e}") def get_line_nickname(user_id): """小工具：向 LINE 伺服器請求使用者暱稱""" try: profile = line_bot_api.get_profile(user_id) return profile.display_name except: return "新朋友" # ---------------------------------------------------------------- # 路由設定 # ---------------------------------------------------------------- @app.get("/") async def root(): return {"message": "16族智慧翻譯小幫手運行中", "base_url": config.BASE_URL} @app.post("/callback") async def callback(request: Request): signature = request.headers.get("X-Line-Signature") body = await request.body() try: handler.handle(body.decode("utf-8"), signature) except InvalidSignatureError: raise HTTPException(status_code=400) return "OK" # ---------------------------------------------------------------- # 1. 處理初次加入好友 (迎賓與暱稱紀錄) # ---------------------------------------------------------------- @handler.add(FollowEvent) def handle_follow(event): user_id = event.source.user_id # 抓取暱稱並存入資料庫 user_nickname = get_line_nickname(user_id) db_manager.set_user_tribe(user_id, "太魯閣", display_name=user_nickname) # 合成 static 資料夾內的圖片網址 (對齊最新視覺設計) image_filename = "welcome_hero.png" full_image_url = f"{config.BASE_URL}/static/{image_filename}" # 創意總監潤色版歡迎文案 welcome_text = ( f"Embiyax su hug！{user_nickname} 您好 ❤️\n\n" "我是您的專屬「16族智慧翻譯小幫手」。\n" "不僅能翻譯文字，還能聽懂語音、看懂圖片！\n\n" "👇 請點擊下方「快速選單」，開啟您的南島語言探索之旅。" ) carousel_card = ui_templates.create_welcome_carousel(hero_image_url=full_image_url) line_bot_api.reply_message(event.reply_token, [ TextSendMessage(text=welcome_text), FlexSendMessage(alt_text="歡迎使用16族智慧翻譯小幫手", contents=carousel_card) ]) # ---------------------------------------------------------------- # 2. 處理互動按鈕與回饋 # ---------------------------------------------------------------- @handler.add(PostbackEvent) def handle_postback(event): user_id = event.source.user_id current_tribe = db_manager.get_user_tribe(user_id) data = event.postback.data if data == "feedback_good": db_manager.log_feedback(user_id, current_tribe, "good") line_bot_api.reply_message(event.reply_token, TextSendMessage(text="✅ 感謝您的肯定！這將幫助我們將模型訓練得更好。")) elif data == "feedback_bad": db_manager.log_feedback(user_id, current_tribe, "bad") line_bot_api.reply_message(event.reply_token, TextSendMessage(text="📝 感謝您的回饋！我們已將此翻譯標記，將持續優化品質。")) # ---------------------------------------------------------------- # 3. 處理多媒體 (影片/圖片/語音) # ---------------------------------------------------------------- @handler.add(MessageEvent, message=VideoMessage) def handle_video(event): user_id = event.source.user_id msg_id = event.message.id current_tribe = db_manager.get_user_tribe(user_id) show_loading_animation(user_id, 30) video_path = f"static/{msg_id}.mp4" audio_path = f"static/{msg_id}_extracted.wav" try: message_content = line_bot_api.get_message_content(msg_id) with open(video_path, 'wb') as fd: for chunk in message_content.iter_content(): fd.write(chunk) if video_processor.extract_audio(video_path, audio_path): recognized_text = audio_processor.speech_to_text(audio_path, current_tribe) if recognized_text: res = cached_translate(recognized_text, current_tribe) reply_text = f"🎬 【{current_tribe}語影片辨識】\n━━━━━━━━━━━━━━\n原文：\n{recognized_text}\n\n翻譯結果：\n{res.get('target', '翻譯失敗')}" db_manager.log_interaction(user_id, current_tribe, "video", recognized_text) else: reply_text = f"⚠️ 抱歉，影片中的{current_tribe}語音量太小或不夠清晰。" else: reply_text = "❌ 影片音軌提取失敗，請檢查檔案格式。" line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply_text)) except Exception as e: line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"影片處理錯誤: {e}")) finally: for p in [video_path, audio_path]: if os.path.exists(p): os.remove(p) @handler.add(MessageEvent, message=ImageMessage) def handle_image(event): user_id = event.source.user_id msg_id = event.message.id current_tribe = db_manager.get_user_tribe(user_id) show_loading_animation(user_id, 10) image_path = f"static/{msg_id}.jpg" try: message_content = line_bot_api.get_message_content(msg_id) with open(image_path, 'wb') as fd: for chunk in message_content.iter_content(): fd.write(chunk) # ⏱️ 啟動唯一的一次計時 start_time = time.time() # 💡 呼叫一體化模組 result = image_processor.extract_and_translate(image_path, current_tribe) duration = time.time() - start_time if result and "items" in result: originals = [i['original'] for i in result['items']] translates = [i['translated'] for i in result['items']] reply_text = ( f"👁️ 【視覺辨識結果】總耗時 {duration:.2f}s\n" f"━━━━━━━━━━━━━━\n" f"📝 擷取原文：\n" + "\n".join(originals) + "\n\n" f"🏹 【{current_tribe}語翻譯】：\n" + "\n".join(translates) ) else: reply_text = f"⚠️ 辨識結束 (耗時 {duration:.2f}s)，未能偵測文字。" line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply_text)) except Exception as e: line_bot_api.reply_message(event.reply_token, TextSendMessage(text="❌ 處理失敗")) finally: if os.path.exists(image_path): os.remove(image_path) @handler.add(MessageEvent, message=AudioMessage) def handle_audio(event): user_id = event.source.user_id msg_id = event.message.id current_tribe = db_manager.get_user_tribe(user_id) show_loading_animation(user_id, 15) audio_path = f"static/{msg_id}.m4a" try: message_content = line_bot_api.get_message_content(msg_id) with open(audio_path, 'wb') as fd: for chunk in message_content.iter_content(): fd.write(chunk) recognized_text = audio_processor.speech_to_text(audio_path, current_tribe) if recognized_text: res = cached_translate(recognized_text, current_tribe) reply_text = ( f"🎤 【{current_tribe}語音辨識】\n━━━━━━━━━━━━━━\n" f"聽寫原文：\n{recognized_text}\n\n" f"中文翻譯：\n{res.get('target', '翻譯失敗')}" ) db_manager.log_interaction(user_id, current_tribe, "audio", recognized_text) else: reply_text = f"⚠️ 無法聽辨{current_tribe}語。" line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply_text)) except Exception as e: line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"語音處理失敗: {e}")) finally: if os.path.exists(audio_path): os.remove(audio_path) # ---------------------------------------------------------------- # 4. 文字訊息處理 (核心邏輯) # ---------------------------------------------------------------- @handler.add(MessageEvent, message=TextMessage) def handle_message(event): user_text = event.message.text.strip() user_id = event.source.user_id current_tribe = db_manager.get_user_tribe(user_id) # --------------------------------------------------------- # 💡 意見回饋處理邏輯 # --------------------------------------------------------- if user_text.startswith("建議："): feedback_content = user_text.replace("建議：", "").strip() db_manager.log_interaction(user_id, current_tribe, "text_feedback", feedback_content) reply_msg = "💖 感謝您的寶貴建議！\n開發團隊已經收到您的留言，我們會持續讓翻譯官變得更好！" line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply_msg)) return if user_text == "我要留言": reply_msg = "💡 請直接在對話框輸入您的建議！\n\n(請記得以「建議：」開頭，例如：建議：字體可以大一點嗎？)" line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply_msg)) return # --------------------------------------------------------- # 💡 執行長最新版 Rich Menu (2x3 六項配置對接) # --------------------------------------------------------- # 1. 切換族語 (上排左) if user_text in ["切換族語", "選單", "切換專屬族語"]: show_loading_animation(user_id, 3) menu = ui_templates.create_tribe_menu() line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="請選擇族語模式", contents=menu)) return # 2. 意見回饋 (上排中) if user_text in ["意見回饋", "填寫意見回饋"]: feedback_ui = ui_templates.create_feedback_card() line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="意見回饋", contents=feedback_ui)) return # 4. 構詞語法分析 (下排left) if user_text in ["構詞語法分析", "語法分析", "什麼是構詞語法分析", "什麼是構詞語法分析？"]: # 🎯 解決 UnboundLocalError：直接從已經通車的 deep_analyzer 借調 supabase 客戶端 from deep_analyzer import supabase current_tribe = "太魯閣" # 安全防禦預設值 try: pref_res = supabase.table("user_preferences").select("tribe").eq("user_id", user_id).execute() if pref_res.data: current_tribe = pref_res.data[0]["tribe"] except Exception as e: logger.error(f"選單介紹圖卡即時撈取族語偏好失敗: {e}") card = ui_templates.create_glossing_intro_card(current_tribe) line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="構詞語法分析介紹", contents=card)) return # 5. 多媒體翻譯 (下排中) if user_text in ["多媒體翻譯", "拍照翻譯", "什麼是多媒體翻譯", "什麼是多媒體翻譯？"]: # 🎯 解決 UnboundLocalError from deep_analyzer import supabase current_tribe = "太魯閣" try: pref_res = supabase.table("user_preferences").select("tribe").eq("user_id", user_id).execute() if pref_res.data: current_tribe = pref_res.data[0]["tribe"] except Exception as e: logger.error(f"選單介紹圖卡即時撈取族語偏好失敗: {e}") card = ui_templates.create_multimedia_intro_card(current_tribe) line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="多媒體翻譯教學", contents=card)) return # 6. 深度文化翻譯 (下排right) if user_text in ["深度文化翻譯", "深度翻譯", "什麼是深度文化翻譯", "什麼是深度文化翻譯？"]: # 🎯 解決 UnboundLocalError from deep_analyzer import supabase current_tribe = "太魯閣" try: pref_res = supabase.table("user_preferences").select("tribe").eq("user_id", user_id).execute() if pref_res.data: current_tribe = pref_res.data[0]["tribe"] except Exception as e: logger.error(f"選單介紹圖卡即時撈取族語偏好失敗: {e}") card = ui_templates.create_deep_translation_intro_card(current_tribe) line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="深度文化翻譯介紹", contents=card)) return # --------------------------------------------------------- # 💡 處理多媒體測試引導 (亮點功能導覽) # --------------------------------------------------------- if user_text == "🎙️ 我要測試語音/影片解析": reply_msg = ( "👇 請直接使用 LINE 的內建功能傳送：\n\n" "1️⃣ 【錄製語音】：按住右下角的「🎤 麥克風」，說一段太魯閣語（例如：Embiyax su hug?）。\n\n" "2️⃣ 【上傳影片】：點擊左下角「➕」>「相簿」，選取一段有族語對話的短影片。\n\n" "傳送後，我會立刻幫您「聽寫原文」並「翻譯內容」喔！🚀" ) line_bot_api.reply_message(event.reply_token, [ TextSendMessage(text=reply_msg), TextSendMessage(text="💡 小提示：錄音時記得大聲一點，AI 老師才聽得清楚喔！") ]) return # --------------------------------------------------------- # 💡 處理指令式功能 # --------------------------------------------------------- if user_text.startswith("📍 切換族語："): tribe = user_text.replace("📍 切換族語：", "").strip() nickname = get_line_nickname(user_id) db_manager.set_user_tribe(user_id, tribe, display_name=nickname) line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"✅ 已成功切換至【{tribe}語】模式！")) return if user_text.startswith("🔊 聽發音："): target_native = user_text.replace("🔊 聽發音：", "").strip() show_loading_animation(user_id, 5) db_manager.log_interaction(user_id, current_tribe, "tts", target_native) try: u_id = str(uuid.uuid4()) rel_path = audio_processor.text_to_speech(target_native, current_tribe, f"tts_{u_id}") if config.BASE_URL and rel_path: audio_url = f"{config.BASE_URL}/{rel_path}" download_url = f"{config.BASE_URL}/download_file/tts_{u_id}.wav" line_bot_api.reply_message(event.reply_token, [ AudioSendMessage(original_content_url=audio_url, duration=2000), FlexSendMessage(alt_text="儲存音檔", contents=ui_templates.create_download_button(download_url)) ]) except Exception as e: logger.error(f"TTS Error: {e}") line_bot_api.reply_message(event.reply_token, TextSendMessage(text="⚠️ 語音生成失敗。")) return # ================================================================= # 💡 語法分析： 階段一 (純文字快速版 + 觸發繪圖按鈕) # ================================================================= if user_text.startswith("💡 語法分析："): target_sentence = user_text.replace("💡 語法分析：", "").strip() show_loading_animation(user_id, 10) try: start_time = time.time() analysis_data = deep_analyzer.get_structured_analysis(target_sentence, current_tribe) duration = time.time() - start_time logger.info(f"⏱️ [語法分析文字版] 耗時: {duration:.2f}s") # 1. 建立純文字回覆內容 reply_text = f"📖 【{current_tribe}語 語法深度解析】\n━━━━━━━━━━━━━━\n" if not analysis_data.get("is_long"): glossing = analysis_data.get("glossing", {}) l1 = glossing.get("line1", []) l2 = glossing.get("line2", []) l3 = glossing.get("line3", []) comp_text = "🎯 【逐詞分析】\n" for i in range(len(l1)): if l1[i] and l1[i].strip(): # 🎴 這裡精準替換成「構詞語法(族語)」與「構詞語法(華語)」 comp_text += f"▪️ {l1[i]}\n ├ 族語: {l2[i] if i < len(l2) else ''}\n └ 華語: {l3[i] if i < len(l3) else ''}\n" reply_text += comp_text + "\n" reply_text += f"💡 【深入解說】\n{analysis_data.get('explanation', '無詳細說明')}\n\n" reply_text += f"🏹 【翻譯結果】\n{analysis_data.get('translation', target_sentence)}" # 2. 建立「需要四行分析圖」的按鈕卡片 (Flex Message) if not analysis_data.get("is_long"): button_card = { "type": "bubble", "size": "kilo", "body": { "type": "box", "layout": "vertical", "contents": [ {"type": "text", "text": "需要將以上分析轉為精美圖卡嗎？", "wrap": True, "size": "sm", "color": "#666666"} ] }, "footer": { "type": "box", "layout": "vertical", "contents": [ { "type": "button", "style": "primary", "color": "#2B579A", "action": { "type": "message", "label": "🖼️ 生成四行分析圖卡", "text": f"🎨 繪製圖卡：{target_sentence}" } } ] } } # 同時發送「文字」與「按鈕」 line_bot_api.reply_message(event.reply_token, [ TextSendMessage(text=reply_text), FlexSendMessage(alt_text="生成圖卡選項", contents=button_card) ]) else: # 若為長句，就不顯示繪圖按鈕（避免畫出來擠在一起） line_bot_api.reply_message(event.reply_token, TextSendMessage(text=reply_text)) db_manager.log_interaction(user_id, current_tribe, "analysis_text", target_sentence) except Exception as e: logger.error(f"語法分析邏輯錯誤: {e}") line_bot_api.reply_message(event.reply_token, TextSendMessage(text="⚠️ 系統解析中，請稍後再試。")) return # ================================================================= # 🎨 繪製圖卡： 階段二 (接收繪圖指令並產出圖片) # ================================================================= if user_text.startswith("🎨 繪製圖卡："): target_sentence = user_text.replace("🎨 繪製圖卡：", "").strip() show_loading_animation(user_id, 15) # 畫圖需要比較久 try: # 重新獲取一次分析資料 (為了畫圖) analysis_data = deep_analyzer.get_structured_analysis(target_sentence, current_tribe) # 準備圖片路徑 img_filename = f"analysis_{uuid.uuid4().hex[:8]}.png" img_path = os.path.join("static", img_filename) # 呼叫美術部門畫圖 visual_renderer.draw_glossing_card(analysis_data, img_path) image_url = f"{config.BASE_URL}/{img_path}" # 發送圖片 line_bot_api.reply_message( event.reply_token, ImageSendMessage(original_content_url=image_url, preview_image_url=image_url) ) except Exception as e: logger.error(f"圖卡繪製錯誤: {e}") line_bot_api.reply_message(event.reply_token, TextSendMessage(text="⚠️ 抱歉，圖卡生成失敗，請確認您的 BASE_URL 網址設定是否正確。")) return # ================================================================= # 🧠 【新增關鍵路由】核心處理：執行「深度文化翻譯潤飾」 # ================================================================= if "深度翻譯" in user_text and "|" in user_text: # 1. 移除非必要的 Emoji 標題前綴 clean_text = user_text.replace("🧠 深度翻譯：", "").replace("深度翻譯：", "").strip() # 2. 切開中文字串與一般翻譯初稿 parts = clean_text.split("|") if len(parts) == 2: source_text = parts[0].strip() # 華語原文 target_text = parts[1].strip() # 族語初稿 show_loading_animation(user_id, loading_seconds=15) try: # 3. 呼叫大腦進行潤飾 refined_result = deep_analyzer.deep_translation(source_text, target_text, current_tribe) db_manager.log_interaction(user_id, current_tribe, "deep_translation", user_text) # 4. 直接回覆經過深度潤飾與說明的結果 line_bot_api.reply_message(event.reply_token, TextSendMessage(text=refined_result)) return except Exception as e: logger.error(f"深度翻譯處理失敗: {e}") line_bot_api.reply_message(event.reply_token, TextSendMessage(text="⚠️ 深度翻譯功能暫時無回應，請稍後再試。")) return # --------------------------------------------------------- # 💡 保險限制：攔截一般翻譯的過長文本 (只對一般翻譯生效) # --------------------------------------------------------- if len(user_text) > 180: limit_msg = ( "⚠️ 您要翻譯的句子太長了（超過 180 字），\n" "AI 會沒辦法快速思考，\n" "麻煩您將句子調整在 180 個字以內，感謝您。❤️" ) line_bot_api.reply_message(event.reply_token, TextSendMessage(text=limit_msg)) return # --------------------------------------------------------- # 一般文字翻譯 (NLLB 模型 - 極速版) # --------------------------------------------------------- # 在開始運算前「立刻」告訴 LINE 要顯示動畫 show_loading_animation(user_id, loading_seconds=15) start_time = time.time() res = cached_translate(user_text, current_tribe) # 🎯 這裡已經被精準修正為 user_text 了！ duration = time.time() - start_time logger.info(f"⏱️ [NLLB 一般翻譯效能] 輸入字數: {len(user_text)} | 耗時: {duration:.2f}s") db_manager.log_interaction(user_id, current_tribe, "text", user_text) target_output = res.get('target', '翻譯失敗') # 雖已有 180 字輸入限制，此處仍保留 200 字輸出保險，避免極端狀況燒毀 Token if len(target_output) > 200: line_bot_api.reply_message( event.reply_token, TextSendMessage(text=f"🏹 【{current_tribe}語結果】\n\n{target_output}") ) return try: card = ui_templates.create_translation_card( tribe=current_tribe, source_text=user_text, target_text=target_output, confidence_level=res.get("confidence_level", "high"), confidence_desc=f"耗時 {duration:.1f}s" # 💡 讓使用者也看到我們的極速成果 ) line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="翻譯結果", contents=card)) except Exception as e: logger.error(f"Flex Message 發送失敗: {e}") line_bot_api.push_message(user_id, TextSendMessage(text=f"🏹 【{current_tribe}語】\n{target_output}")) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=7860)