ILRDF-Lowking's picture
Update app.py
c9c9424 verified
Raw
History Blame Contribute Delete
27.1 kB
# 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)