Update app.py

app.py

@@ -1,221 +1,195 @@
 import os
 import time
-import uuid
-from datetime import timedelta
+import smtplib
+from email.mime.text import MIMEText
 from pydub import AudioSegment
 from openai import OpenAI
 import gradio as gr
 
 # ========================
-# 🔐 環境變數設定
+# 🔐 設定區
 # ========================
 PASSWORD = os.getenv("APP_PASSWORD", "defaultpass")
-MAX_SIZE = 25 * 1024 * 1024  # 25 MB 限制
+MAX_SIZE = 25 * 1024 * 1024
 client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+ALERT_EMAIL = os.getenv("ALERT_EMAIL")
+ALERT_PASS = os.getenv("ALERT_PASS")
 
 # ========================
-# ⚔️ 防暴力破解機制
+# 📧 寄信通知
 # ========================
-MAX_FAILED_IN_WINDOW = 10           # 24小時內最多10次
+def send_alert_email(session_id, reason):
+    if not ALERT_EMAIL or not ALERT_PASS:
+        return
+    msg = MIMEText(f"Session {session_id} 已被鎖定。\n原因：{reason}\n時間：{time.ctime()}", "plain", "utf-8")
+    msg["Subject"] = "⚠️ 語音轉錄系統警報"
+    msg["From"] = ALERT_EMAIL
+    msg["To"] = ALERT_EMAIL
+    try:
+        with smtplib.SMTP_SSL("smtp.gmail.com", 465) as server:
+            server.login(ALERT_EMAIL, ALERT_PASS)
+            server.sendmail(ALERT_EMAIL, ALERT_EMAIL, msg.as_string())
+        print(f"✅ 已寄出警報郵件至 {ALERT_EMAIL}")
+    except Exception as e:
+        print(f"❌ 寄信失敗：{e}")
+
+# ========================
+# ⚔️ 防暴力破解
+# ========================
+MAX_FAILED_IN_WINDOW = 10
 WINDOW_SECONDS = 24 * 3600
-LOCK_DURATION_SECONDS = 24 * 3600   # 鎖24小時
-SHORT_BURST_LIMIT = 5               # 一分鐘內最多5次
+LOCK_DURATION_SECONDS = 24 * 3600
+SHORT_BURST_LIMIT = 5
 SHORT_BURST_SECONDS = 60
-attempts = {}   # {session_id: [timestamps]}
-locked = {}     # {session_id: unlock_time}
 
+attempts = {}
+locked = {}
 
-def _now():
-    return int(time.time())
+def _now(): return int(time.time())
 
-
-def prune_old_attempts(session_id):
-    if session_id not in attempts:
-        return
+def prune_old_attempts(sid):
     cutoff = _now() - WINDOW_SECONDS
-    attempts[session_id] = [t for t in attempts[session_id] if t >= cutoff]
-    if not attempts[session_id]:
-        del attempts[session_id]
-
-
-def check_lock(session_id):
-    if session_id in locked:
-        unlock = locked[session_id]
-        if _now() < unlock:
-            remain = unlock - _now()
+    if sid in attempts:
+        attempts[sid] = [t for t in attempts[sid] if t >= cutoff]
+        if not attempts[sid]:
+            del attempts[sid]
+
+def check_lock(sid):
+    if sid in locked:
+        if _now() < locked[sid]:
+            remain = locked[sid] - _now()
             return True, f"🔒 已被鎖定，請 {remain // 60} 分鐘後再試。"
         else:
-            del locked[session_id]
-            attempts.pop(session_id, None)
-    prune_old_attempts(session_id)
-    cnt = len(attempts.get(session_id, []))
+            locked.pop(sid, None)
+            attempts.pop(sid, None)
+    prune_old_attempts(sid)
+    cnt = len(attempts.get(sid, []))
     if cnt >= MAX_FAILED_IN_WINDOW:
-        locked[session_id] = _now() + LOCK_DURATION_SECONDS
+        locked[sid] = _now() + LOCK_DURATION_SECONDS
+        send_alert_email(sid, f"密碼錯誤 {cnt} 次，鎖定24小時")
         return True, f"🔒 嘗試過多，已鎖定 24 小時。"
     return False, ""
 
-
-def record_failed_attempt(session_id):
+def record_failed_attempt(sid):
     now = _now()
-    attempts.setdefault(session_id, []).append(now)
-    prune_old_attempts(session_id)
-    cnt = len(attempts.get(session_id, []))
+    attempts.setdefault(sid, []).append(now)
+    prune_old_attempts(sid)
     recent_cutoff = now - SHORT_BURST_SECONDS
-    recent = [t for t in attempts[session_id] if t >= recent_cutoff]
+    recent = [t for t in attempts[sid] if t >= recent_cutoff]
     if len(recent) >= SHORT_BURST_LIMIT:
-        locked[session_id] = now + 300  # 鎖5分鐘
-        return cnt, "⚠️ 多次快速嘗試，暫時鎖定5分鐘。"
-    return cnt, ""
-
+        locked[sid] = now + 300
+        send_alert_email(sid, "短時間內多次錯誤，鎖定5分鐘")
+        return len(attempts[sid]), "⚠️ 多次快速嘗試，暫時鎖定5分鐘。"
+    return len(attempts[sid]), ""
 
-def clear_attempts(session_id):
-    attempts.pop(session_id, None)
-    locked.pop(session_id, None)
+def clear_attempts(sid):
+    attempts.pop(sid, None)
+    locked.pop(sid, None)
 
 # ========================
-# 🔊 音訊分割
+# 🎧 音訊轉錄
 # ========================
-def split_audio_if_needed(input_path):
-    size = os.path.getsize(input_path)
+def split_audio_if_needed(path):
+    size = os.path.getsize(path)
     if size <= MAX_SIZE:
-        return [input_path]
-    audio = AudioSegment.from_file(input_path)
-    num_chunks = int(size / MAX_SIZE) + 1
-    chunk_length = len(audio) / num_chunks
-    chunk_files = []
-    for i in range(num_chunks):
-        start = int(i * chunk_length)
-        end = int((i + 1) * chunk_length)
+        return [path]
+    audio = AudioSegment.from_file(path)
+    num = int(size / MAX_SIZE) + 1
+    chunk_ms = len(audio) / num
+    files = []
+    for i in range(num):
+        start, end = int(i * chunk_ms), int((i + 1) * chunk_ms)
         chunk = audio[start:end]
-        chunk_filename = f"chunk_{i+1}.wav"
-        chunk.export(chunk_filename, format="wav")
-        chunk_files.append(chunk_filename)
-    return chunk_files
-
-# ========================
-# 🎧 轉錄與摘要
-# ========================
-def transcribe_core(file, model_choice):
-    chunks = split_audio_if_needed(file)
-    transcripts = []
-    for idx, f in enumerate(chunks, 1):
-        with open(f, "rb") as audio_file:
-            text = client.audio.transcriptions.create(
-                model=model_choice,
-                file=audio_file,
-                response_format="text"
-            )
-        transcripts.append(text)
-    full_text = "\n".join(transcripts)
-    response = client.chat.completions.create(
+        fn = f"chunk_{i+1}.wav"
+        chunk.export(fn, format="wav")
+        files.append(fn)
+    return files
+
+def transcribe_core(path, model):
+    chunks = split_audio_if_needed(path)
+    txts = []
+    for f in chunks:
+        with open(f, "rb") as af:
+            res = client.audio.transcriptions.create(model=model, file=af, response_format="text")
+        txts.append(res)
+    full = "\n".join(txts)
+    res = client.chat.completions.create(
         model="gpt-4o-mini",
-        messages=[
-            {"role": "system", "content": "你是一位精準且擅長摘要的助手。"},
-            {"role": "user", "content": "請用繁體中文摘要以下內容：\n" + full_text}
-        ],
+        messages=[{"role":"user","content":f"請用繁體中文摘要以下內容：\n{full}"}],
         temperature=0.4,
     )
-    summary = response.choices[0].message.content.strip()
-    return full_text, summary
-
+    summ = res.choices[0].message.content.strip()
+    return full, summ
 
+# ========================
+# 💬 主流程
+# ========================
 def transcribe_with_password(session_id, password, file, model_choice):
     locked_flag, msg = check_lock(session_id)
     if locked_flag:
-        return msg, "", "", ""
+        return msg, "", ""
     if password != PASSWORD:
         cnt, msg2 = record_failed_attempt(session_id)
-        if msg2:
-            return msg2, "", "", ""
-        return f"密碼錯誤（第 {cnt} 次）", "", "", ""
+        return msg2 or f"密碼錯誤（第 {cnt} 次）", "", ""
     if not file:
-        return "請上傳音訊檔。", "", "", ""
+        return "請上傳音訊檔。", "", ""
     clear_attempts(session_id)
-    full_text, summary = transcribe_core(file, model_choice)
-    return "✅ 成功轉錄與摘要完成", full_text, summary, ""
+    full, summ = transcribe_core(file, model_choice)
+    return "✅ 轉錄完成", full, summ
 
-# ========================
-# 💬 進一步問 AI
-# ========================
-def ask_about_transcript(full_text, user_question):
+def ask_about_transcript(full_text, q):
     if not full_text.strip():
-        return "⚠️ 尚未有轉錄內容。"
-    if not user_question.strip():
-        return "請輸入問題。"
-    prompt = f"以下是轉錄內容：\n{full_text}\n\n使用者問：{user_question}\n請用繁體中文回答。"
-    response = client.chat.completions.create(
-        model="gpt-4o-mini",
-        messages=[{"role": "user", "content": prompt}],
-        temperature=0.6,
-    )
-    return response.choices[0].message.content.strip()
+        return "⚠️ 尚未有轉錄內容"
+    if not q.strip():
+        return "請輸入問題"
+    prompt = f"以下是轉錄內容：\n{full_text}\n\n問題：{q}\n請用繁體中文回答。"
+    res = client.chat.completions.create(
+        model="gpt-4o-mini", messages=[{"role":"user","content":prompt}], temperature=0.6)
+    return res.choices[0].message.content.strip()
 
 # ========================
 # 🌐 Gradio 介面
 # ========================
 with gr.Blocks(theme=gr.themes.Soft()) as demo:
-    gr.Markdown("## 🎧 語音轉錄與摘要工具（含防暴力破解）")
-
-    session_id_box = gr.Textbox(value="", visible=False)
-    init_session_js = """
-    () => {
-        let sid = localStorage.getItem('my_session_id');
-        if (!sid) {
-            sid = crypto.randomUUID ? crypto.randomUUID() :
-                  (Date.now().toString(36) + Math.random().toString(36).slice(2));
-            localStorage.setItem('my_session_id', sid);
-        }
-        return sid;
-    }
-    """
-    session_id_box.load(_js=init_session_js)
+    gr.Markdown("## 🎧 語音轉錄與摘要工具（防暴力破解＋郵件警報）")
+
+    session_state = gr.State(value=None)
 
     with gr.Row():
         password_input = gr.Textbox(label="輸入密碼", type="password")
-        model_choice = gr.Dropdown(
-            choices=["whisper-1", "gpt-4o-mini-transcribe"],
-            value="whisper-1",
-            label="選擇轉錄模型"
-        )
+        model_choice = gr.Dropdown(["whisper-1", "gpt-4o-mini-transcribe"], value="whisper-1", label="選擇模型")
 
     audio_input = gr.Audio(type="filepath", label="上傳音訊 (.m4a, .aac, .wav)")
     transcribe_btn = gr.Button("開始轉錄與摘要 🚀")
-
     status_box = gr.Textbox(label="狀態", interactive=False)
-    with gr.Row():
-        transcript_box = gr.Textbox(label="完整轉錄文字", lines=10)
-        copy_transcript = gr.Button("📋 複製")
-
-    with gr.Row():
-        summary_box = gr.Textbox(label="摘要結果", lines=10)
-        copy_summary = gr.Button("📋 複製")
+    transcript_box = gr.Textbox(label="完整轉錄文字", lines=10)
+    copy_transcript = gr.Button("📋 複製")
+    summary_box = gr.Textbox(label="摘要結果", lines=10)
+    copy_summary = gr.Button("📋 複製")
 
     with gr.Accordion("💬 進一步問 AI", open=False):
-        user_question = gr.Textbox(label="輸入你的問題", lines=2)
+        user_q = gr.Textbox(label="輸入問題", lines=2)
         ask_btn = gr.Button("詢問 AI 🤔")
         ai_reply = gr.Textbox(label="AI 回覆", lines=6)
         copy_reply = gr.Button("📋 複製")
 
-    transcribe_btn.click(
-        fn=transcribe_with_password,
-        inputs=[session_id_box, password_input, audio_input, model_choice],
-        outputs=[status_box, transcript_box, summary_box, gr.Textbox(visible=False)]
-    )
+    def init_session():
+        import uuid
+        return str(uuid.uuid4())
+    demo.load(init_session, None, session_state)
 
-    ask_btn.click(
-        fn=ask_about_transcript,
-        inputs=[transcript_box, user_question],
-        outputs=[ai_reply]
+    transcribe_btn.click(
+        transcribe_with_password,
+        [session_state, password_input, audio_input, model_choice],
+        [status_box, transcript_box, summary_box],
     )
+    ask_btn.click(ask_about_transcript, [transcript_box, user_q], [ai_reply])
 
     copy_js = """
     async (text) => {
-        try {
-            await navigator.clipboard.writeText(text);
-            alert("✅ 已複製到剪貼簿！");
-        } catch (err) {
-            alert("❌ 複製失敗：" + err);
-        }
+      try { await navigator.clipboard.writeText(text);
+            alert("✅ 已複製到剪貼簿！"); }
+      catch(e){ alert("❌ 複製失敗：" + e); }
     }
     """
     copy_transcript.click(None, transcript_box, None, _js=copy_js)