Update app.py

app.py

@@ -1,184 +1,142 @@
 import os
-import json
-import time
-import base64
-import sqlite3
 import tempfile
 import requests
-from bs4 import BeautifulSoup
-
+import gradio as gr
 import torch
 import whisper
 from gtts import gTTS
 from pydub import AudioSegment
-import numpy as np
 from sentence_transformers import SentenceTransformer, util
 from transformers import AutoTokenizer, AutoModelForCausalLM
+from bs4 import BeautifulSoup
+from duckduckgo_search import DDGS
+import time
 
-import gradio as gr
-
-# =====================
-# 初始化模型
-# =====================
+# ------------------------
+# 模型初始化
+# ------------------------
 device = "cuda" if torch.cuda.is_available() else "cpu"
 whisper_model = whisper.load_model("base", device=device)
-embedding_model = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
-tokenizer = AutoTokenizer.from_pretrained("lmsys/mini-cpm-1b-sft")
-gen_model = AutoModelForCausalLM.from_pretrained("lmsys/mini-cpm-1b-sft").to(device)
+embedder = SentenceTransformer("all-MiniLM-L6-v2")
+tokenizer = AutoTokenizer.from_pretrained("gpt2")
+lm_model = AutoModelForCausalLM.from_pretrained("gpt2").to(device)
 
-# =====================
-# 向量資料庫 (SQLite)
-# =====================
-DB_PATH = "vector_db.sqlite"
-
-def init_db():
-    conn = sqlite3.connect(DB_PATH)
-    c = conn.cursor()
-    c.execute("""
-        CREATE TABLE IF NOT EXISTS documents (
-            id INTEGER PRIMARY KEY AUTOINCREMENT,
-            content TEXT,
-            embedding BLOB
-        )
-    """)
-    conn.commit()
-    conn.close()
-
-def add_document(content):
-    embedding = embedding_model.encode(content)
-    conn = sqlite3.connect(DB_PATH)
-    c = conn.cursor()
-    c.execute("INSERT INTO documents (content, embedding) VALUES (?, ?)", 
-              (content, embedding.tobytes()))
-    conn.commit()
-    conn.close()
-
-def search_similar(query, top_k=3):
-    query_vec = embedding_model.encode(query)
-    conn = sqlite3.connect(DB_PATH)
-    c = conn.cursor()
-    c.execute("SELECT content, embedding FROM documents")
-    rows = c.fetchall()
-    conn.close()
-
-    if not rows:
-        return []
-
-    contents, embeddings = zip(*rows)
-    embeddings = [np.frombuffer(e, dtype=np.float32) for e in embeddings]
-    cos_scores = util.cos_sim(query_vec, embeddings)[0].cpu().numpy()
-
-    ranked = sorted(zip(contents, cos_scores), key=lambda x: x[1], reverse=True)
-    return [r[0] for r in ranked[:top_k]]
-
-# =====================
-# 爬蟲模組
-# =====================
+# ------------------------
+# 信任網站（含行政院人事行政總處行事曆）
+# ------------------------
 TRUSTED_SITES = [
     "https://www.dgpa.gov.tw",  # 行政院人事行政總處 行事曆
-    "https://www.cna.com.tw",   # 中央通訊社
-    "https://www.stust.edu.tw"  # 南台科技大學
-    "https://www.moi.gov.tw",
-    "https://www.taiwan.net.tw",
-    "https://publicholidays.tw",
+    "https://zh.wikipedia.org",  # 中文維基百科
 ]
 
-def crawl_trusted_sites(query):
-    """先從信任網站搜尋"""
-    results = []
-    headers = {"User-Agent": "Mozilla/5.0"}
-    for site in TRUSTED_SITES:
-        try:
-            resp = requests.get(site, headers=headers, timeout=10)
-            if resp.status_code == 200:
-                soup = BeautifulSoup(resp.text, "html.parser")
-                text = " ".join([p.get_text(strip=True) for p in soup.find_all("p")])
-                if query in text:
-                    results.append(f"[可信來源] {site}: {text[:200]}...")
-        except Exception:
-            continue
-    return results
-
-def crawl_general_web(query):
-    """一般網站搜尋"""
+# ------------------------
+# 文字轉語音
+# ------------------------
+def text_to_speech(text):
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") as fp:
+        tts = gTTS(text=text, lang="zh")
+        tts.save(fp.name)
+        return fp.name
+
+# ------------------------
+# 爬蟲 - 行政院人事行政總處 行事曆
+# ------------------------
+def crawl_official_calendar(query):
     try:
-        url = f"https://www.google.com/search?q={query}"
-        headers = {"User-Agent": "Mozilla/5.0"}
-        resp = requests.get(url, headers=headers, timeout=10)
-        if resp.status_code == 200:
-            return [f"[一般搜尋結果] {url}"]
+        url = "https://www.dgpa.gov.tw/holidaycalendar"
+        resp = requests.get(url, timeout=10)
+        soup = BeautifulSoup(resp.text, "html.parser")
+        rows = soup.find_all("tr")
+        for row in rows:
+            if query in row.text:
+                return row.text.strip()
+    except Exception as e:
+        return None
+    return None
+
+# ------------------------
+# 網路搜尋 (DuckDuckGo)
+# ------------------------
+def web_search(query):
+    try:
+        ddgs = DDGS()
+        results = ddgs.text(query, max_results=3)
+        answer = ""
+        for r in results:
+            answer += f"{r['title']} - {r['body']}\n"
+        return answer if answer else "查無資料。"
     except Exception:
-        return []
-    return []
-
-# =====================
-# 問答邏輯
-# =====================
-def generate_answer(query, context=""):
-    inputs = tokenizer(query + " " + context, return_tensors="pt").to(device)
-    outputs = gen_model.generate(**inputs, max_new_tokens=128)
+        return "網路搜尋失敗。"
+
+# ------------------------
+# AI 回答
+# ------------------------
+def ai_answer(question):
+    inputs = tokenizer(question, return_tensors="pt").to(device)
+    outputs = lm_model.generate(inputs["input_ids"], max_length=150, do_sample=True, top_k=50)
     return tokenizer.decode(outputs[0], skip_special_tokens=True)
 
-def qa_pipeline(query):
-    start_time = time.time()
-
-    # 1. 南台相關 → AI
+# ------------------------
+# 主邏輯
+# ------------------------
+def qa_system(audio, text_input):
+    query = ""
+    if text_input:  # 文字優先
+        query = text_input.strip()
+    elif audio is not None:
+        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmpfile:
+            audio.export(tmpfile.name, format="wav")
+            result = whisper_model.transcribe(tmpfile.name)
+            query = result["text"]
+    else:
+        return "請提供語音或文字輸入。", None, ""
+
+    # 判斷是否包含「南台/南臺」
     if "南台" in query or "南臺" in query:
-        context = " ".join(search_similar(query))
-        if time.time() - start_time > 120:
-            web_results = crawl_general_web(query)
-            context += " ".join(web_results)
-        return generate_answer(query, context)
-
-    # 2. 一般問題 → 先查可信網站 (60 秒內)
-    if time.time() - start_time <= 60:
-        trusted_results = crawl_trusted_sites(query)
-        if trusted_results:
-            return "\n".join(trusted_results)
-
-    # 3. 超過 60 秒 → 一般網路搜尋
-    web_results = crawl_general_web(query)
-    if web_results:
-        return "\n".join(web_results)
-
-    return "抱歉，目前無法找到相關資訊。"
-
-# =====================
+        # 使用 AI 回答
+        start_time = time.time()
+        answer = ai_answer(query)
+        elapsed = time.time() - start_time
+        if elapsed > 120:  # 超過 120 秒 fallback
+            web_ans = web_search(query)
+            answer += f"\n（補充搜尋結果）{web_ans}"
+    else:
+        # 節日 or 一般問題，先查可信網站
+        start_time = time.time()
+        answer = crawl_official_calendar(query)
+        if not answer:
+            elapsed = time.time() - start_time
+            if elapsed > 60:
+                answer = web_search(query)
+            else:
+                answer = "查無相關資料（可信網站）。"
+
+    # 生成語音回覆
+    audio_file = text_to_speech(answer)
+    return query, answer, audio_file
+
+# ------------------------
 # Gradio 介面
-# =====================
-def chatbot_interface(audio=None, text=None):
-    query = ""
-    if audio:
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".wav") as tmp:
-            tmp.write(audio)
-            tmp.flush()
-            result = whisper_model.transcribe(tmp.name)
-            query = result["text"].strip()
-    if text:
-        query = text.strip()
-
-    if not query:
-        return "請提供語音或文字輸入。"
-
-    answer = qa_pipeline(query)
-
-    # TTS
-    tts = gTTS(answer, lang="zh")
-    tts_file = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3")
-    tts.save(tts_file.name)
-    return answer, tts_file.name
-
+# ------------------------
 with gr.Blocks() as demo:
-    gr.Markdown("# 🎤 南臺科技大學 智慧語音助理")
-    with gr.Row():
-        audio_input = gr.Audio(source="microphone", type="filepath", label="語音輸入 (可選)")
-        text_input = gr.Textbox(label="文字輸入 (可選)", placeholder="請輸入問題...")
-    submit_btn = gr.Button("送出")
-    output_text = gr.Textbox(label="回答")
-    output_audio = gr.Audio(label="語音回答", type="filepath")
+    gr.Markdown("# 南臺科技大學語音文字問答系統")
+    gr.Markdown("你可以輸入文字或錄音提問，系統會找出答案，並用語音回覆。")
 
-    submit_btn.click(fn=chatbot_interface, inputs=[audio_input, text_input], outputs=[output_text, output_audio])
+    with gr.Row():
+        with gr.Column():
+            text_input = gr.Textbox(label="文字輸入（可選）", placeholder="請輸入你的問題...")
+            audio_input = gr.Audio(sources=["microphone", "upload"], type="filepath", label="請上傳語音檔或錄音")
+            submit_btn = gr.Button("Submit")
+        with gr.Column():
+            text_output = gr.Textbox(label="語音辨識文字")
+            answer_output = gr.Textbox(label="AI 回答")
+            audio_output = gr.Audio(label="語音回覆", type="filepath")
+
+    submit_btn.click(
+        qa_system,
+        inputs=[audio_input, text_input],
+        outputs=[text_output, answer_output, audio_output]
+    )
 
 if __name__ == "__main__":
-    init_db()
-    demo.launch(server_name="0.0.0.0", server_port=7860)
+    demo.launch()