📋 ILRDF AI 戰情室大廳 (🔒 已連線至雲端金庫)
🗣️ 累計對話次數
{total_chats}
句練習成果
✍️ 累計修正筆數
{total_corrections}
黃金數據
🔥 最活躍族語
{top_lang_text}
歡迎度排行
📢 族人心聲與滿意度留言板
來自族人的最真實回饋與建議。📥 下載 Excel 報表
📋 完整修正回饋紀錄 (即時連線)
這些是族人針對 AI 翻譯提出的正確版本，包含完整的對話上下文。
| 時間 | 族別 | 用戶輸入情境 | AI 錯誤版本 | 族人修正建議 | 使用者暱稱 |
|---|
{table_rows}
"""
return HTMLResponse(content=content)
# 📥 修正回饋 Excel 下載 (Supabase 雲端版)
@app.get("/export_feedback")
async def export_feedback(username: str = Depends(get_current_username)):
res = supabase.table("feedback_v2").select("*").order("timestamp", desc=True).execute()
csv_filename = "feedback_export.csv"
csv_path = os.path.join("static", csv_filename)
with open(csv_path, 'w', newline='', encoding='utf-8-sig') as f:
writer = csv.writer(f)
writer.writerow(['序號', '回報時間', '使用者暱稱', '使用者ID', '族別', '用戶輸入情境', 'AI原本輸出', '用戶修正建議'])
for row in res.data:
writer.writerow([
row.get('id'), row.get('timestamp'), row.get('user_name') or '未知用戶',
row.get('user_id'), row.get('tribe'), row.get('user_input'),
row.get('ai_output'), row.get('correction')
])
today_str = datetime.now(tw_tz).strftime('%Y%m%d_%H%M')
return FileResponse(
path=csv_path,
filename=f"ILRDF_回饋意見總表_{today_str}.csv",
media_type='text/csv'
)
# 🤖 4. LINE Callback (雲端連線版)
@app.post("/callback")
async def callback(request: Request):
signature = request.headers.get('X-Line-Signature', '')
body = await request.body()
try:
handler.handle(body.decode(), signature)
except InvalidSignatureError:
raise HTTPException(status_code=400)
return 'OK'
@handler.add(MessageEvent, message=TextMessage)
def handle_text(event):
text = event.message.text.strip()
user_id = event.source.user_id
# 💡 魔法啟動：讓使用者看到輸入中的點點點
show_loading_animation(user_id)
# ==========================================
# 👑 執行長專屬：成就測試外掛 (Supabase 雲端版)
# ==========================================
if text.startswith("#外掛"):
try:
parts = text.split(" ")
cheat_type = parts[1]
target_num = int(parts[2]) - 1
# 先確保您的帳號有在資料庫裡 (用 upsert 寫入或更新)
supabase.table("users").upsert({"user_id": user_id}).execute()
if cheat_type == "互動":
supabase.table("users").update({"count": target_num}).eq("user_id", user_id).execute()
msg = f"✅ 【外掛成功】已將您的互動次數設定為 {target_num}！\n👉 請隨便打一句中文或族語，即可解鎖第 {target_num+1} 次的成就圖卡！"
elif cheat_type == "建議":
# 刪除舊的假建議並重新塞入
supabase.table("user_surveys").delete().eq("user_id", user_id).not_is("suggestion", "null").execute()
fake_surveys = [{"timestamp": "2026-01-01", "user_id": user_id, "suggestion": "假建議"} for _ in range(target_num)]
if fake_surveys:
supabase.table("user_surveys").insert(fake_surveys).execute()
msg = f"✅ 【外掛成功】已為您塞入 {target_num} 筆歷史建議！\n👉 請輸入「#建議 測試」，即可解鎖第 {target_num+1} 次的成就圖卡！"
elif cheat_type == "天數":
yesterday_str = (datetime.now(tw_tz) - timedelta(days=1)).strftime('%Y-%m-%d')
supabase.table("users").update({"streak": target_num, "last_active_date": yesterday_str}).eq("user_id", user_id).execute()
msg = f"✅ 【外掛成功】已將您的連續天數設定為 {target_num} 天，並把最後上線日改為昨天！\n👉 請隨便打一句話，即可解鎖連續 {target_num+1} 天的成就圖卡！"
else:
msg = "❌ 外掛類型錯誤，請輸入：互動、建議、天數"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=msg))
except Exception as e:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"外掛指令錯誤: {e}"))
return
# ==========================================
# ==========================================
# 💡 背景自動同步使用者暱稱 (Supabase 雲端版)
# ==========================================
user_name = "未知名稱"
try:
profile = line_bot_api.get_profile(user_id)
user_name = profile.display_name
today_str = datetime.now(tw_tz).strftime('%Y-%m-%d')
# 檢查使用者是否存在，不存在就建立，存在就更新暱稱
res = supabase.table("users").select("user_id").eq("user_id", user_id).execute()
if not res.data:
supabase.table("users").insert({"user_id": user_id, "join_date": today_str, "user_name": user_name}).execute()
else:
supabase.table("users").update({"user_name": user_name}).eq("user_id", user_id).execute()
except Exception as e:
print(f"背景同步暱稱失敗: {e}")
# ==========================================
# 💡 3.0 營運核心：捕獲滿意度與具體建議 (Supabase 雲端版)
# ==========================================
if text.startswith("回饋："):
score = text.replace("回饋：", "").strip()
timestamp = datetime.now(tw_tz).strftime('%Y-%m-%d %H:%M:%S')
supabase.table("user_surveys").insert({
"timestamp": timestamp, "user_id": user_id, "user_name": user_name, "score": score
}).execute()
if score == "我有具體建議":
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="太棒了！我們非常需要您的聲音👂\n請在下方直接輸入「#建議：您的想法」，我們一定會納入後續改版參考！"))
else:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"收到！感謝您的「{score}」評價，我們會持續為語言傳承努力！💪"))
return
if text.startswith("#建議"):
suggestion = text.replace("#建議", "").strip()
timestamp = datetime.now(tw_tz).strftime('%Y-%m-%d %H:%M:%S')
supabase.table("user_surveys").insert({
"timestamp": timestamp, "user_id": user_id, "user_name": user_name, "suggestion": suggestion
}).execute()
try:
awards = update_user_stats_and_get_awards(user_id, DB_FILE, action="feedback")
except:
awards = []
reply_msgs = [TextSendMessage(text="您的寶貴建議已經安全送達【原語會信箱】！📨\n感謝您為族語的貢獻。")]
for award in awards:
reply_msgs.append(FlexSendMessage(alt_text="🎊 解鎖新成就！", contents=create_achievement_flex(award, user_name)))
line_bot_api.reply_message(event.reply_token, reply_msgs)
return
# ==========================================
mode = user_modes.get(user_id, "chat")
# === 切換族語模式 ===
if text in TRIBE_CONFIG:
today_str = datetime.now(tw_tz).strftime('%Y-%m-%d')
res = supabase.table("users").select("user_id").eq("user_id", user_id).execute()
if not res.data:
supabase.table("users").insert({"user_id": user_id, "join_date": today_str, "tribe": text, "user_name": user_name}).execute()
else:
supabase.table("users").update({"tribe": text}).eq("user_id", user_id).execute()
chat_sessions.pop(user_id, None)
user_modes[user_id] = "chat"
welcome_msg = (
f"✅ 已切換至【{text}語】模式！\n\n"
f"💡 小提醒：除了用語音聊天或打字聊天，您現在也可以「傳送照片」給我，我會告訴您照片裡的東西怎麼說喔！\n\n"
f"現在，請按住麥克風跟我說說話或打出族語，或是傳張照片來吧！\n\n"
f"💡 溫馨提醒：等待AI回覆時間約15-25秒！請耐心等待💡 "
)
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=welcome_msg))
return
# 💡 進入修正模式
elif text == "💡 提出修正":
if user_id not in last_bot_reply:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="目前沒有可以修正的句子喔，請先跟我聊聊天或傳照片！"))
else:
user_modes[user_id] = "correcting"
ai_native = last_bot_reply[user_id]['native']
ai_zh = last_bot_reply[user_id]['zh']
msg1 = f"感謝您的協助！請複製下方的族語拼音，修改成正確的版本後傳送給我：\n\n(原本的中文：{ai_zh})\n(若想取消請輸入「選單」)"
msg2 = ai_native
line_bot_api.reply_message(event.reply_token, [TextSendMessage(text=msg1), TextSendMessage(text=msg2)])
# 💡 接收修正內容
elif user_modes.get(user_id) == "correcting":
if text in ["選單", "取消"]:
user_modes[user_id] = "chat"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="已取消修正，返回聊天模式！"))
return
last_msg = user_last_msg.get(user_id, {"input": "無紀錄", "output": "無紀錄"})
corrected_native = text
timestamp = datetime.now(tw_tz).strftime('%Y-%m-%d %H:%M:%S')
res = supabase.table("users").select("tribe").eq("user_id", user_id).execute()
tribe = res.data[0]['tribe'] if res.data else "未知"
try:
# 寫入全新 feedback_v2 雲端資料庫
supabase.table("feedback_v2").insert({
"timestamp": timestamp, "user_id": user_id, "tribe": tribe,
"user_input": last_msg["input"], "ai_output": last_msg["output"],
"correction": corrected_native, "user_name": user_name
}).execute()
fb_res = supabase.table("feedback_v2").select("id", count="exact").eq("user_id", user_id).execute()
feedback_count = fb_res.count or 0
thank_you_msg = (
f"✅ 感謝您的指正！這句話已寫入 ILRDF AI語言庫。\n\n"
f"✅ 累積至今，您已提供 {feedback_count} 筆建議回饋，再次感謝您，我們會持續讓族語AI更聰明、更精確。"
)
reply_msgs = [TextSendMessage(text=thank_you_msg)]
try:
target_native = corrected_native
config = TRIBE_CONFIG.get(tribe)
if config:
unique_str = f"{tribe}_{target_native}"
safe_name = hashlib.md5(unique_str.encode('utf-8')).hexdigest()
filepath = f"static/tts_{safe_name}.wav"
if not os.path.exists(filepath):
speaker = get_clean_value(tts_client.predict(ethnicity=config["mt"], api_name="/lambda"))
if tribe == "阿美": speaker = "阿美_秀姑巒_女聲1"
temp_tts = tts_client.predict(ref=speaker, gen_text_input=target_native, api_name="/default_speaker_tts")
shutil.move(temp_tts, filepath)
audio_dur = AudioSegment.from_file(filepath)
duration_ms = len(audio_dur)
base_url = "https://lowking-ilrdf-ai-line-bot.hf.space"
play_url = f"{base_url}/{filepath}"
just_filename = filepath.replace("static/", "")
download_url = f"{base_url}/download_file/{just_filename}?openExternalBrowser=1"
download_button = FlexSendMessage(
alt_text="下載語音",
contents={
"type": "bubble", "size": "micro",
"body": {
"type": "box", "layout": "vertical", "paddingAll": "10px",
"contents": [{"type": "button", "action": {"type": "uri", "label": "💾 儲存音檔", "uri": download_url}, "style": "secondary", "height": "sm"}]
}
}
)
reply_msgs.append(AudioSendMessage(original_content_url=play_url, duration=duration_ms))
reply_msgs.append(download_button)
except Exception as tts_e:
print(f"修正後的語音生成失敗: {tts_e}")
line_bot_api.reply_message(event.reply_token, reply_msgs)
except Exception as e:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"系統儲存回饋時遇到狀況: {e}"))
finally:
user_modes[user_id] = "chat"
return
# 💡 語音隨選 (附加強制下載功能)
elif text.startswith("🔊 聽發音："):
target_native = text.replace("🔊 聽發音：", "").strip()
res = supabase.table("users").select("tribe").eq("user_id", user_id).execute()
if not res.data or not res.data[0].get('tribe'):
return
tribe = res.data[0]['tribe']
config = TRIBE_CONFIG[tribe]
unique_str = f"{tribe}_{target_native}"
safe_name = hashlib.md5(unique_str.encode('utf-8')).hexdigest()
filepath = f"static/tts_{safe_name}.wav"
if not os.path.exists(filepath):
try:
speaker = get_clean_value(tts_client.predict(ethnicity=config["mt"], api_name="/lambda"))
if tribe == "阿美": speaker = "阿美_秀姑巒_女聲1"
temp_tts = tts_client.predict(ref=speaker, gen_text_input=target_native, api_name="/default_speaker_tts")
shutil.move(temp_tts, filepath)
except Exception as e:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"語音生成失敗，請稍後再試：{e}"))
return
audio_dur = AudioSegment.from_file(filepath)
duration_ms = len(audio_dur)
base_url = "https://lowking-ilrdf-ai-line-bot.hf.space"
play_url = f"{base_url}/{filepath}"
just_filename = filepath.replace("static/", "")
download_url = f"{base_url}/download_file/{just_filename}?openExternalBrowser=1"
download_button = FlexSendMessage(
alt_text="下載語音",
contents={
"type": "bubble", "size": "micro",
"body": {
"type": "box", "layout": "vertical", "paddingAll": "10px",
"contents": [{"type": "button", "action": {"type": "uri", "label": "💾 儲存音檔", "uri": download_url}, "style": "secondary", "height": "sm"}]
}
}
)
line_bot_api.reply_message(event.reply_token, [
AudioSendMessage(original_content_url=play_url, duration=duration_ms),
download_button
])
return
# 🚀 v2.0 正式全體發布指令
elif text == "執行長正式發布2.0":
announcement = (
"✨ 【ILRDF AI 族語學習夥伴 v2.0 重磅升級】 ✨\n━━━━━━━━━━━━━━━━\n親愛的族語學習夥伴，我們進化了！🚀\n\n"
"1. ⚡ 雙模對話：可採文字和語音輸入，族語AI與你雙模式對話！\n2. 📸 看圖學語：傳照片給我，透過AI視力教你族語！\n"
"3. 📊 視覺護照：新增學習進度條，掌握個人多語學習狀況及分佈！\n4. 🎯 發音挑戰：從對話出題，提供AI語音精準評分建議！\n"
"5. 🔊 語音隨選：點擊綠色按鈕，聽取AI擬真族語發音。\n\n6. 💡 回饋建議：點選提出修正，回饋內容將作為訓練AI的重要養分。\n\n"
"━━━━━━━━━━━━━━━━\n🌅 讓我們一起，透過族語 AI 輔助學習自己的語言。\n請輸入「選單」體驗最新功能！"
)
res = supabase.table("users").select("user_id").execute()
success_count, fail_count = 0, 0
for user in res.data:
uid = user['user_id']
try:
line_bot_api.push_message(uid, TextSendMessage(text=announcement))
success_count += 1
except Exception as e:
fail_count += 1
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"📢 v2.0 公告發布完畢！\n✅ 成功：{success_count} 位\n❌ 失敗：{fail_count} 位"))
return
# 🚀 執行長一鍵更新舊用戶暱稱
elif text == "執行長更新名單":
res = supabase.table("users").select("user_id").is_("user_name", "null").execute()
success = 0
for r in res.data:
uid = r['user_id']
try:
profile = line_bot_api.get_profile(uid)
supabase.table("users").update({"user_name": profile.display_name}).eq("user_id", uid).execute()
success += 1
except Exception as e:
pass
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"✅ 報告執行長！成功抓回 {success} 人的真實暱稱！請至戰情室重新整理。"))
return
# 🎯 發音挑戰區塊
elif "發音挑戰" in text or "我的金句" in text or "趣味發音" in text:
res = supabase.table("users").select("tribe").eq("user_id", user_id).execute()
tribe = res.data[0]['tribe'] if res.data else None
if not tribe:
show_tribe_menu(event, 0)
return
phrase_res = supabase.table("phrases").select("native, zh").eq("user_id", user_id).execute()
if phrase_res.data:
chosen = random.choice(phrase_res.data)
target_native, target_zh = chosen['native'], chosen['zh']
else:
target_native, target_zh = ("Nga'ay ho", "你好嗎") if tribe=="阿美" else ("Embiyax su hug", "你好嗎")
user_modes[user_id] = "challenge"
challenge_target[user_id] = {"native": target_native, "zh": target_zh}
config = TRIBE_CONFIG[tribe]
try:
speaker = get_clean_value(tts_client.predict(ethnicity=config["mt"], api_name="/lambda"))
if tribe == "阿美": speaker = "阿美_秀姑巒_女聲1"
temp_tts = tts_client.predict(ref=speaker, gen_text_input=target_native, api_name="/default_speaker_tts")
unique_str = f"chal_{tribe}_{target_native}"
safe_name = hashlib.md5(unique_str.encode('utf-8')).hexdigest()
filepath = f"static/chal_{safe_name}.wav"
shutil.move(temp_tts, filepath)
audio_dur = AudioSegment.from_file(filepath)
duration_ms = len(audio_dur)
just_filename = filepath.replace("static/", "")
audio_url = f"https://lowking-ilrdf-ai-line-bot.hf.space/download_file/{just_filename}"
target_native = format_native_text(target_native, tribe)
line_bot_api.reply_message(event.reply_token, [
TextSendMessage(text=f"🎯 【發音挑戰開始】\n\n請按住麥克風跟著唸：\n「{target_native}」\n({target_zh})"),
AudioSendMessage(original_content_url=audio_url, duration=duration_ms)
])
except Exception as e:
user_modes[user_id] = "chat"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"產生考題失敗，請稍後再試：{e}"))
return
elif text in ["選單", "🔄 切換語別", "切換語別", "切換族語", "換族語"]:
user_modes[user_id] = "chat"
show_tribe_menu(event, 0)
return
elif text == "更多族別":
show_tribe_menu(event, 10)
return
elif text in ["功能", "介紹", "說明", "怎麼用", "教學", "help", "功能介紹與特色"]:
messages_to_send = [FlexSendMessage(alt_text="功能介紹與特點", contents=create_intro_card())]
sop_text = (
"🚀【ILRDF 族語AI學習夥伴 (Beta測試版)】新手上路 SOP 🚀\n\n"
"1️⃣ **設定語別**：點擊下方「🔄 切換族語」選擇族語。\n\n"
"2️⃣ **開始聊天**：傳送「你好」或語音訊息。 族語AI會經翻譯後並播放擬真人發音🔊。\n\n"
"3️⃣ **進階探索**：\n📸 拍張照：傳送照片，AI 教您說族語。\n🎤 玩挑戰：點擊「發音挑戰」測試您的準確度。\n✍️ 教 AI：點擊「提出修正」教我說話，我會立刻唸給您聽！\n\n"
"🏆 **達成任務**：解鎖 24 階稱號與「黑金數位證書」🏅。現在就開始吧！"
)
messages_to_send.append(TextSendMessage(text=sop_text))
line_bot_api.reply_message(event.reply_token, messages_to_send)
elif text in ["學習記錄", "查詢記錄", "學習紀錄"]:
res = supabase.table("users").select("count, streak, join_date").eq("user_id", user_id).execute()
if res.data:
user_data = res.data[0]
total_count = user_data.get('count', 0)
streak = user_data.get('streak', 0)
join_date = user_data.get('join_date', datetime.now(tw_tz).strftime('%Y-%m-%d'))
else:
total_count, streak, join_date = 0, 0, datetime.now(tw_tz).strftime('%Y-%m-%d')
fb_res = supabase.table("feedback_v2").select("id", count="exact").eq("user_id", user_id).execute()
feedback_count = fb_res.count or 0
if total_count < 5: title = "🌱 族語探索者"
elif total_count < 20: title = "🌿 語彙拾穗者"
elif total_count < 50: title = "🗣️ 語感培育員"
elif total_count < 100: title = "🔥 族語實踐家"
elif total_count < 200: title = "🤝 文化搭橋人"
elif total_count < 500: title = "🧭 族語引路人"
elif total_count < 1000: title = "👑 族語傳承者"
else: title = "🌅 復振點燈人"
lang_res = supabase.table("language_stats").select("tribe, count").eq("user_id", user_id).order("count", desc=True).execute()
lang_list_str = ""
for lang in lang_res.data:
l_name, l_count = lang['tribe'], lang['count']
percentage = int((l_count / total_count) * 100) if total_count > 0 else 0
bar_count = percentage // 20
bar = "▓" * bar_count + "░" * (5 - bar_count)
lang_list_str += f" • {l_name}語: {l_count} 次 ({percentage}%) {bar}\n"
cat_res = supabase.table("category_stats").select("category, count").eq("user_id", user_id).order("count", desc=True).limit(3).execute()
cat_str = "\n".join([f" • {cat['category']}: {cat['count']} 次" for cat in cat_res.data]) if cat_res.data else " (尚無分類數據)"
special_badge = "\n🌍 【解鎖成就：多語學習達人】" if len(lang_res.data) >= 3 else ""
stats_text = (
f"📊 【您的族語學習護照】\n━━━━━━━━━━━━━━\n"
f"📅 開始學習：{join_date}\n🏅 當前階級：{title}\n🔥 連續學習：{streak} 天\n"
f"🗣️ 累計發言：{total_count} 次{special_badge}\n✍️ 貢獻紀錄：{feedback_count} 筆建議回饋\n━━━━━━━━━━━━━━\n"
f"🗺️ 語言分佈：\n{lang_list_str}\n📌 最常聊的話題：\n{cat_str}\n\nSa'icelen! Kisamulja! Tamasaza! Kmbiyax!（加油！）"
)
reply_msgs = [TextSendMessage(text=stats_text)]
if total_count > 0 and total_count % 3 == 0:
survey_buttons = QuickReply(items=[
QuickReplyButton(action=MessageAction(label="👍 好用", text="回饋：好用")),
QuickReplyButton(action=MessageAction(label="😐 普通", text="回饋：普通")),
QuickReplyButton(action=MessageAction(label="✍️ 我有具體建議", text="回饋：我有具體建議"))
])
reply_msgs.append(TextSendMessage(text="原語會ILRDF關心您！這幾次的練習感覺如何？點選下方按鈕告訴我們吧！", quick_reply=survey_buttons))
line_bot_api.reply_message(event.reply_token, reply_msgs)
else:
# 💡 一般聊天模式：翻譯並準備語音下載連結！(Supabase 雲端版)
res = supabase.table("users").select("tribe, streak, last_active_date").eq("user_id", user_id).execute()
if not res.data or not res.data[0].get('tribe'):
show_tribe_menu(event, 0)
return
row = res.data[0]
tribe = row.get('tribe')
current_streak = row.get('streak') or 0
last_date = row.get('last_active_date')
config = TRIBE_CONFIG[tribe]
today_str = datetime.now(tw_tz).strftime('%Y-%m-%d')
yesterday_str = (datetime.now(tw_tz) - timedelta(days=1)).strftime('%Y-%m-%d')
try:
import re
if re.search(r'[\u4e00-\u9fa5]', text):
zh_in = text
back_code = get_clean_value(mt_client.predict(ethnicity=config["mt"], api_name="/lambda_1"))
native_in = get_clean_value(mt_client.predict(text=zh_in, src_lang="zho_Hant", tgt_lang=back_code, api_name="/translate_1"))
else:
native_in = format_native_text(text, tribe)
go_code = get_clean_value(mt_client.predict(ethnicity=config["mt"], api_name="/lambda"))
zh_in = get_clean_value(mt_client.predict(text=native_in, src_lang=go_code, tgt_lang="zho_Hant", api_name="/translate"))
native_in = format_native_text(native_in, tribe)
# 💡 雲端雙引擎：先查 Supabase 快取記憶體！
cache_res = supabase.table("chat_cache").select("ai_zh, ai_native").eq("tribe", tribe).eq("user_zh", zh_in).execute()
if cache_res.data:
ai_zh, ai_native = cache_res.data[0]['ai_zh'], cache_res.data[0]['ai_native']
else:
ai_zh = get_ai_response(user_id, zh_in, tribe)
back_code = get_clean_value(mt_client.predict(ethnicity=config["mt"], api_name="/lambda_1"))
ai_native = get_clean_value(mt_client.predict(text=ai_zh, src_lang="zho_Hant", tgt_lang=back_code, api_name="/translate_1"))
ai_native = format_native_text(ai_native, tribe)
# 存入雲端快取
supabase.table("chat_cache").insert({"tribe": tribe, "user_zh": zh_in, "ai_zh": ai_zh, "ai_native": ai_native}).execute()
last_bot_reply[user_id] = {"native": ai_native, "zh": ai_zh}
if last_date == yesterday_str: new_streak = current_streak + 1
elif last_date == today_str: new_streak = current_streak
else: new_streak = 1
topic = categorize_topic(zh_in)
# 💡 更新雲端統計數據
lang_res = supabase.table("language_stats").select("count").eq("user_id", user_id).eq("tribe", tribe).execute()
if lang_res.data:
supabase.table("language_stats").update({"count": lang_res.data[0]['count'] + 1}).eq("user_id", user_id).eq("tribe", tribe).execute()
else:
supabase.table("language_stats").insert({"user_id": user_id, "tribe": tribe, "count": 1}).execute()
supabase.table("phrases").insert({"user_id": user_id, "native": ai_native, "zh": ai_zh}).execute()
cat_res = supabase.table("category_stats").select("count").eq("user_id", user_id).eq("category", topic).execute()
if cat_res.data:
supabase.table("category_stats").update({"count": cat_res.data[0]['count'] + 1}).eq("user_id", user_id).eq("category", topic).execute()
else:
supabase.table("category_stats").insert({"user_id": user_id, "category": topic, "count": 1}).execute()
user_last_msg[user_id] = {"input": native_in, "output": ai_native}
try:
# 執行長注意：這裡呼叫的成就引擎目前仍接收 DB_FILE
awards = update_user_stats_and_get_awards(user_id, DB_FILE, action="chat")
except Exception as e:
print(f"成就引擎錯誤: {e}")
awards = []
chat_msgs = [FlexSendMessage(alt_text="AI 對話", contents=create_chat_card(tribe, native_in, ai_zh, ai_native))]
for award in awards:
# 若抓不到 user_name 預設顯示"您"
chat_msgs.append(FlexSendMessage(alt_text=f"🎊 解鎖新成就：{award.get('title', '新稱號')}！", contents=create_achievement_flex(award, "您")))
line_bot_api.reply_message(event.reply_token, chat_msgs)
except Exception as e:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"AI 稍後回來: {str(e)}"))
@handler.add(MessageEvent, message=ImageMessage)
def handle_image(event):
user_id = event.source.user_id
show_loading_animation(user_id)
res = supabase.table("users").select("tribe, streak, last_active_date").eq("user_id", user_id).execute()
if not res.data or not res.data[0].get('tribe'):
line_bot_api.reply_message(event.reply_token, TextSendMessage(text="💡 您好，請先選擇一個族語，我才能告訴您照片裡的東西怎麼說喔！"))
show_tribe_menu(event, 0)
return
row = res.data[0]
tribe = row.get('tribe')
current_streak = row.get('streak') or 0
last_date = row.get('last_active_date')
config = TRIBE_CONFIG[tribe]
msg_id = event.message.id
image_path = f"static/{msg_id}.jpg"
today_str = datetime.now(tw_tz).strftime('%Y-%m-%d')
yesterday_str = (datetime.now(tw_tz) - timedelta(days=1)).strftime('%Y-%m-%d')
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)
img = PIL.Image.open(image_path)
prompt = (
f"請仔細觀察這張照片，用『一句簡短的中文直述句』描述照片中最主要的物品或場景。\n"
f"【重要規則】：\n1. 必須是單純的直述句，絕對不要包含主觀情緒、語氣詞或感嘆詞。\n"
f"2. 句尾請一律用「句號」結束，絕對不要使用驚嘆號。\n"
f"3. 句子結構越簡單、越客觀越好，以利於系統進行原住民族語的機器翻譯。"
)
response = model.generate_content([prompt, img])
ai_zh = response.text.strip()
back_code = get_clean_value(mt_client.predict(ethnicity=config["mt"], api_name="/lambda_1"))
ai_native = get_clean_value(mt_client.predict(text=ai_zh, src_lang="zho_Hant", tgt_lang=back_code, api_name="/translate_1"))
ai_native = format_native_text(ai_native, tribe)
last_bot_reply[user_id] = {"native": ai_native, "zh": ai_zh}
topic = "看圖學族語"
# 寫入雲端數據
lang_res = supabase.table("language_stats").select("count").eq("user_id", user_id).eq("tribe", tribe).execute()
if lang_res.data:
supabase.table("language_stats").update({"count": lang_res.data[0]['count'] + 1}).eq("user_id", user_id).eq("tribe", tribe).execute()
else:
supabase.table("language_stats").insert({"user_id": user_id, "tribe": tribe, "count": 1}).execute()
supabase.table("phrases").insert({"user_id": user_id, "native": ai_native, "zh": ai_zh}).execute()
cat_res = supabase.table("category_stats").select("count").eq("user_id", user_id).eq("category", topic).execute()
if cat_res.data:
supabase.table("category_stats").update({"count": cat_res.data[0]['count'] + 1}).eq("user_id", user_id).eq("category", topic).execute()
else:
supabase.table("category_stats").insert({"user_id": user_id, "category": topic, "count": 1}).execute()
user_last_msg[user_id] = {"input": f"📸 傳送圖片辨識", "output": ai_native}
line_bot_api.reply_message(event.reply_token, [
TextSendMessage(text="👀 我看懂這張照片了！"),
FlexSendMessage(alt_text="看圖學族語", contents=create_image_chat_card(tribe, ai_zh, ai_native))
])
except Exception as e:
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"AI 視力模糊了，稍後回來: {str(e)}"))
@handler.add(MessageEvent, message=AudioMessage)
def handle_audio(event):
user_id = event.source.user_id
show_loading_animation(user_id)
mode = user_modes.get(user_id, "chat")
res = supabase.table("users").select("tribe, streak, last_active_date").eq("user_id", user_id).execute()
if not res.data or not res.data[0].get('tribe'):
show_tribe_menu(event, 0)
return
row = res.data[0]
tribe = row.get('tribe')
current_streak = row.get('streak') or 0
last_date = row.get('last_active_date')
config = TRIBE_CONFIG[tribe]
msg_id = event.message.id
today_str = datetime.now(tw_tz).strftime('%Y-%m-%d')
yesterday_str = (datetime.now(tw_tz) - timedelta(days=1)).strftime('%Y-%m-%d')
try:
content = line_bot_api.get_message_content(msg_id)
with open(f"static/{msg_id}.m4a", "wb") as f: f.write(content.content)
audio = AudioSegment.from_file(f"static/{msg_id}.m4a")
audio.export(f"static/{msg_id}.wav", format="wav")
native_in = asr_client.predict(dialect_id=config["asr"], audio_data=handle_file(f"static/{msg_id}.wav"), api_name="/automatic_speech_recognition")
if mode == "challenge":
target = challenge_target.get(user_id, {})
result = eval_pronunciation(tribe, native_in, target.get("native", ""), target.get("zh", ""))
line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="評分結果", contents=create_score_card(tribe, result["score"], result["feedback"])))
return
go_code = get_clean_value(mt_client.predict(ethnicity=config["mt"], api_name="/lambda"))
zh_in = get_clean_value(mt_client.predict(text=native_in, src_lang=go_code, tgt_lang="zho_Hant", api_name="/translate"))
# 💡 雲端雙引擎：語音也查快取
cache_res = supabase.table("chat_cache").select("ai_zh, ai_native").eq("tribe", tribe).eq("user_zh", zh_in).execute()
if cache_res.data:
ai_zh, ai_native = cache_res.data[0]['ai_zh'], cache_res.data[0]['ai_native']
else:
ai_zh = get_ai_response(user_id, zh_in, tribe)
back_code = get_clean_value(mt_client.predict(ethnicity=config["mt"], api_name="/lambda_1"))
ai_native = get_clean_value(mt_client.predict(text=ai_zh, src_lang="zho_Hant", tgt_lang=back_code, api_name="/translate_1"))
ai_native = format_native_text(ai_native, tribe)
supabase.table("chat_cache").insert({"tribe": tribe, "user_zh": zh_in, "ai_zh": ai_zh, "ai_native": ai_native}).execute()
last_bot_reply[user_id] = {"native": ai_native, "zh": ai_zh}
topic = categorize_topic(zh_in)
# 寫入雲端數據
lang_res = supabase.table("language_stats").select("count").eq("user_id", user_id).eq("tribe", tribe).execute()
if lang_res.data:
supabase.table("language_stats").update({"count": lang_res.data[0]['count'] + 1}).eq("user_id", user_id).eq("tribe", tribe).execute()
else:
supabase.table("language_stats").insert({"user_id": user_id, "tribe": tribe, "count": 1}).execute()
supabase.table("phrases").insert({"user_id": user_id, "native": ai_native, "zh": ai_zh}).execute()
cat_res = supabase.table("category_stats").select("count").eq("user_id", user_id).eq("category", topic).execute()
if cat_res.data:
supabase.table("category_stats").update({"count": cat_res.data[0]['count'] + 1}).eq("user_id", user_id).eq("category", topic).execute()
else:
supabase.table("category_stats").insert({"user_id": user_id, "category": topic, "count": 1}).execute()
user_last_msg[user_id] = {"input": f"🎤 語音輸入 ({native_in})", "output": ai_native}
line_bot_api.reply_message(event.reply_token, FlexSendMessage(alt_text="AI 對話", contents=create_chat_card(tribe, native_in, ai_zh, ai_native)))
except Exception as e:
user_modes[user_id] = "chat"
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=f"AI 稍後回來: {str(e)}"))
# 💡 終極整合：新手引導 (當使用者加入好友或解除封鎖時觸發)
@handler.add(FollowEvent)
def handle_follow(event):
user_id = event.source.user_id
today_str = datetime.now(tw_tz).strftime('%Y-%m-%d')
# 💾 資料庫隱身掛號 (Supabase 雲端版)
try:
res = supabase.table("users").select("user_id").eq("user_id", user_id).execute()
if not res.data:
supabase.table("users").insert({"user_id": user_id, "join_date": today_str, "count": 0, "streak": 0}).execute()
except Exception as e:
print(f"Follow 事件註冊失敗: {e}")
welcome_text = (
"🎉 Embiyax su hug? 歡迎來到【原語會ILRDF 族語AI學習夥伴 (Beta測試版)】！\n\n"
"我是您的專屬 AI 族語導師，系統已全面升級「三大體驗維度」，陪您輕鬆學 16 族語：\n\n"
"🌸【AI 隨身家教】\n打字、語音都能雙向翻譯！上傳照片還能「看圖學族語」，並提供專屬語音下載。\n\n"
"🏆【實戰與榮譽殿堂】\n勇敢玩「發音挑戰」！持續學習將為您解鎖 24 階隱藏稱號，並頒發刻有您名字的「黑金數位證書」。\n\n"
"🤝【族語共學策略】\n發現錯誤？提出修正後，立刻聆聽您的專屬語音回饋！輸入「#建議」還能將心聲直達專屬信箱。\n\n"
"⚠️ 【Beta測試版溫馨提醒】：本系統目前為 Beta 測試階段，請多多利用「提出修正」功能教導它喔！\n\n"
"👇 準備好了嗎？請先點擊下方選單「🔄 切換族語」，或直接跟我說一句話，馬上領取您的第一張迎賓獎狀吧！"
)
line_bot_api.reply_message(event.reply_token, TextSendMessage(text=welcome_text))
@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"ILRDF_族語發音_{filename}",
media_type='application/octet-stream'
)
return {"error": "檔案不存在"}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=7860)