app.py

"""
多策略 RAG 文件問答系統 v2 — ChromaDB + PDF/DOCX 版本（含 Telegram 推送）
已修正：Telegram ReadTimeout，加入 retry + 指數退避 + 分離 connect/read timeout
"""

from __future__ import annotations

import os
import re
import time
from pathlib import Path
from typing import Any

import chromadb
import gradio as gr
import numpy as np
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from docx import Document
from docx.oxml.table import CT_Tbl
from docx.oxml.text.paragraph import CT_P
from docx.table import Table
from docx.text.paragraph import Paragraph
from groq import Groq
from pypdf import PdfReader
from sentence_transformers import SentenceTransformer
from sklearn.feature_extraction.text import TfidfVectorizer


# ══════════════════════════════════════════════════════════
#  Telegram 推送設定
# ══════════════════════════════════════════════════════════
DEFAULT_TELEGRAM_CHAT_ID = "8874400558"
TELEGRAM_MAX_LEN = 4000


def _make_session() -> requests.Session:
    """建立帶有 retry 策略的 requests Session。"""
    session = requests.Session()
    retry = Retry(
        total=4,                        # 最多重試 4 次
        backoff_factor=1.5,             # 退避：1.5 / 3 / 4.5 / 6.75 秒
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"],
        raise_on_status=False,
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    return session


def send_telegram_message(text: str, chat_id: str, token: str) -> dict:
    """
    將文字訊息送到 Telegram。
    - 分離 connect timeout（10 s）與 read timeout（30 s）
    - 自動重試最多 4 次（指數退避）
    - 長訊息自動分段
    """
    if not token:
        return {"ok": False, "error": "尚未提供 Bot Token"}
    if not chat_id:
        return {"ok": False, "error": "尚未提供 Chat ID"}
    if not text:
        return {"ok": False, "error": "empty text"}

    # 先驗證 token 格式（避免明顯錯誤）
    if ":" not in token or len(token) < 20:
        return {"ok": False, "error": "Bot Token 格式不正確，應為 123456789:AAAxxxxxx"}

    url = f"https://api.telegram.org/bot{token}/sendMessage"
    session = _make_session()
    results = []

    for i in range(0, len(text), TELEGRAM_MAX_LEN):
        chunk = text[i : i + TELEGRAM_MAX_LEN]
        payload = {
            "chat_id": chat_id,
            "text": chunk,
            "parse_mode": "HTML",
        }
        last_exc = None
        # 手動額外重試迴圈（配合退避），requests Retry 已涵蓋 5xx，這裡補 timeout
        for attempt in range(4):
            try:
                resp = session.post(
                    url,
                    data=payload,
                    timeout=(10, 30),   # (connect_timeout, read_timeout)
                )
                result = resp.json()
                results.append(result)
                last_exc = None
                break  # 成功就跳出重試
            except requests.exceptions.Timeout as exc:
                last_exc = exc
                wait = 1.5 ** attempt
                time.sleep(wait)
            except requests.exceptions.ConnectionError as exc:
                last_exc = exc
                wait = 2 ** attempt
                time.sleep(wait)
            except Exception as exc:
                last_exc = exc
                break  # 非網路錯誤不重試

        if last_exc is not None:
            results.append({
                "ok": False,
                "error": f"{type(last_exc).__name__}: {last_exc} （已重試 4 次）",
            })

    return results[-1] if results else {"ok": False, "error": "no chunks sent"}


# ══════════════════════════════════════════════════════════
#  RAG 核心邏輯
# ══════════════════════════════════════════════════════════
class MultiStrategyRAG:

    STRATEGY_MAP = {
        "semantic":     "1  ChromaDB 語意搜尋",
        "tfidf":        "2  TF-IDF 關鍵詞",
        "hybrid":       "3  混合搜尋",
        "rerank":       "4  重新排序",
        "multi_query":  "5  多查詢擴展",
        "compress":     "6  上下文壓縮",
        "parent_child": "7  父子文檔",
        "hyde":         "8  假設性答案 HyDE",
    }

    def __init__(
        self,
        chroma_path: str = "./chroma_db",
        collection_name: str = "audit_rag_chunks",
        child_collection_name: str = "audit_rag_child_chunks",
    ):
        self.client: Groq | None = None

        self.embedding_model = SentenceTransformer(
            "sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2"
        )

        self.chroma_client = chromadb.PersistentClient(path=chroma_path)
        self.collection = self.chroma_client.get_or_create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"},
        )
        self.child_collection = self.chroma_client.get_or_create_collection(
            name=child_collection_name,
            metadata={"hnsw:space": "cosine"},
        )

        self.session_id: str | None = None
        self.source_name: str = ""
        self.file_type: str = ""
        self.chunks: list[str] = []
        self.child_chunks: list[str] = []
        self.tfidf_vectorizer: TfidfVectorizer | None = None
        self.tfidf_matrix = None

    def set_api_key(self, api_key: str) -> None:
        key = (api_key or "").strip()
        self.client = Groq(api_key=key) if key else None

    def load_document(self, file_path: str) -> str:
        try:
            path = Path(file_path)
            if not path.exists():
                return "✗ 載入失敗：找不到檔案"

            suffix = path.suffix.lower()
            if suffix not in (".pdf", ".docx"):
                return "✗ 目前僅支援 PDF 與 DOCX 檔案"

            self.source_name = path.name
            self.file_type = suffix.lstrip(".")
            self.session_id = (
                f"{int(time.time())}_{re.sub(r'[^a-zA-Z0-9]+', '_', path.stem)[:40]}"
            )

            if suffix == ".pdf":
                full_text, stats = self._extract_pdf(path)
            else:
                full_text, stats = self._extract_docx(path)

            if not full_text.strip():
                return "✗ 載入失敗：文件沒有可擷取文字，可能是掃描圖片檔，需先 OCR"

            self.chunks = self._split(full_text, chunk_size=800, overlap=150)
            if not self.chunks:
                return "✗ 載入失敗：切段後沒有有效內容"

            self._build_chroma_index()
            self._build_tfidf_index()
            self._build_child_index()

            return (
                f"✓ 成功載入 {self.source_name}\n"
                f"類型：{suffix.upper().lstrip('.')} · {stats}\n"
                f"{len(self.chunks)} 個主片段 · ChromaDB Session：{self.session_id}"
            )
        except Exception as exc:
            return f"✗ 載入失敗：{type(exc).__name__}: {exc}"

    def _extract_pdf(self, path: Path) -> tuple[str, str]:
        reader = PdfReader(str(path))
        parts = []
        for idx, page in enumerate(reader.pages, 1):
            text = page.extract_text() or ""
            if text.strip():
                parts.append(f"\n[PDF 第 {idx} 頁]\n{text}")
        return "\n".join(parts), f"{len(reader.pages)} 頁"

    def _extract_docx(self, path: Path) -> tuple[str, str]:
        doc = Document(str(path))
        blocks: list[str] = []
        para_count = table_count = 0

        for child in doc.element.body.iterchildren():
            if isinstance(child, CT_P):
                text = Paragraph(child, doc).text.strip()
                if text:
                    para_count += 1
                    blocks.append(text)
            elif isinstance(child, CT_Tbl):
                table_count += 1
                tbl_text = self._table_to_text(Table(child, doc))
                if tbl_text.strip():
                    blocks.append(f"\n[DOCX 表格 {table_count}]\n{tbl_text}")

        return "\n\n".join(blocks), f"{para_count} 段落 / {table_count} 表格"

    def _table_to_text(self, table: Table) -> str:
        rows = []
        for row in table.rows:
            cells = [re.sub(r"\s+", " ", c.text).strip() for c in row.cells if c.text.strip()]
            if cells:
                rows.append(" | ".join(cells))
        return "\n".join(rows)

    def _split(self, text: str, chunk_size: int, overlap: int) -> list[str]:
        clean = re.sub(r"\s+", " ", text).strip()
        step = max(1, chunk_size - overlap)
        return [
            c for start in range(0, len(clean), step)
            if (c := clean[start : start + chunk_size].strip())
        ]

    def _encode(self, texts: list[str]) -> list[list[float]]:
        return (
            self.embedding_model
            .encode(texts, convert_to_numpy=True, normalize_embeddings=True, show_progress_bar=False)
            .astype("float32")
            .tolist()
        )

    def _build_chroma_index(self) -> None:
        sid = self.session_id
        ids = [f"{sid}_chunk_{i:05d}" for i in range(len(self.chunks))]
        metas = [
            {"session_id": sid, "source": self.source_name,
             "file_type": self.file_type, "chunk_index": i}
            for i in range(len(self.chunks))
        ]
        self.collection.add(ids=ids, documents=self.chunks,
                            metadatas=metas, embeddings=self._encode(self.chunks))

    def _build_tfidf_index(self) -> None:
        self.tfidf_vectorizer = TfidfVectorizer(analyzer="char", ngram_range=(2, 4), max_features=3000)
        self.tfidf_matrix = self.tfidf_vectorizer.fit_transform(self.chunks)

    def _build_child_index(self) -> None:
        sid = self.session_id
        child_docs, child_ids, child_metas = [], [], []
        for pidx, parent in enumerate(self.chunks):
            for cidx, child in enumerate(self._split(parent, chunk_size=300, overlap=50)):
                child_docs.append(child)
                child_ids.append(f"{sid}_parent_{pidx:05d}_child_{cidx:03d}")
                child_metas.append({"session_id": sid, "source": self.source_name,
                                    "file_type": self.file_type,
                                    "parent_index": pidx, "child_index": cidx})
        self.child_chunks = child_docs
        if child_docs:
            self.child_collection.add(ids=child_ids, documents=child_docs,
                                       metadatas=child_metas, embeddings=self._encode(child_docs))

    def _where(self) -> dict[str, str]:
        return {"session_id": self.session_id or ""}

    def _chroma_search(self, query: str, k: int, child: bool = False) -> list[dict[str, Any]]:
        if not self.session_id:
            return []
        col = self.child_collection if child else self.collection
        results = col.query(
            query_embeddings=self._encode([query]),
            n_results=max(1, k),
            where=self._where(),
            include=["documents", "metadatas", "distances"],
        )
        docs = results.get("documents", [[]])[0] or []
        metas = results.get("metadatas", [[]])[0] or []
        dists = results.get("distances", [[]])[0] or []
        return [{"text": d, "metadata": m or {}, "distance": dist}
                for d, m, dist in zip(docs, metas, dists)]

    def _dedupe(self, chunks: list[str], k: int) -> list[str]:
        seen: set[str] = set()
        out: list[str] = []
        for c in chunks:
            key = c[:120]
            if key not in seen:
                seen.add(key)
                out.append(c)
            if len(out) >= k:
                break
        return out

    def _llm(self, prompt: str, max_tokens: int = 300, temperature: float = 0.3) -> str | None:
        if not self.client:
            return None
        try:
            r = self.client.chat.completions.create(
                model="llama-3.1-8b-instant",
                messages=[{"role": "user", "content": prompt}],
                max_tokens=max_tokens,
                temperature=temperature,
            )
            return r.choices[0].message.content
        except Exception:
            return None

    def s_semantic(self, query: str, k: int = 3) -> list[str]:
        return [r["text"] for r in self._chroma_search(query, k)]

    def s_tfidf(self, query: str, k: int = 3) -> list[str]:
        if self.tfidf_vectorizer is None or self.tfidf_matrix is None:
            return []
        qv = self.tfidf_vectorizer.transform([query])
        scores = (self.tfidf_matrix * qv.T).toarray().flatten()
        return [self.chunks[i] for i in scores.argsort()[-k:][::-1]]

    def s_hybrid(self, query: str, k: int = 3) -> list[str]:
        return self._dedupe(self.s_semantic(query, k * 2) + self.s_tfidf(query, k * 2), k)

    def s_rerank(self, query: str, k: int = 3) -> list[str]:
        candidates = self.s_semantic(query, k * 2)
        if not self.client:
            return candidates[:k]
        scored: list[tuple[str, float]] = []
        for chunk in candidates:
            prompt = (f"問題：{query}\n\n文本：{chunk[:500]}\n\n"
                      f"請只輸出 0 到 10 的相關度分數（僅數字）：")
            resp = self._llm(prompt, max_tokens=10, temperature=0)
            nums = re.findall(r"\d+(?:\.\d+)?", resp or "")
            scored.append((chunk, float(nums[0]) if nums else 0.0))
        scored.sort(key=lambda x: x[1], reverse=True)
        return [c for c, _ in scored[:k]]

    def s_multi_query(self, query: str, k: int = 3) -> list[str]:
        queries = [query]
        prompt = f"將以下問題改寫成 3 個角度不同的繁體中文問題，每行一題，不加編號：\n{query}"
        resp = self._llm(prompt, max_tokens=200, temperature=0.7)
        if resp:
            extras = [ln.strip("-• 1234567890.、 ") for ln in resp.splitlines() if ln.strip()]
            queries += extras[:3]
        chunks: list[str] = []
        for q in queries:
            chunks.extend(self.s_semantic(q, 2))
        return self._dedupe(chunks, k)

    def s_compress(self, query: str, k: int = 3) -> list[str]:
        chunks = self.s_semantic(query, k)
        if not self.client:
            return chunks
        compressed = []
        for chunk in chunks:
            prompt = (f"從以下文本中，提取與問題「{query}」最相關的 1-2 句，"
                      f"保留繁體中文，不要添加任何解釋：\n\n{chunk}")
            resp = self._llm(prompt, max_tokens=180, temperature=0)
            compressed.append((resp or "").strip() or chunk[:350])
        return compressed

    def s_parent_child(self, query: str, k: int = 3) -> list[str]:
        hits = self._chroma_search(query, k * 3, child=True)
        seen_parents: list[int] = []
        for h in hits:
            pidx = h.get("metadata", {}).get("parent_index")
            if isinstance(pidx, int) and pidx not in seen_parents:
                seen_parents.append(pidx)
            if len(seen_parents) >= k:
                break
        return [self.chunks[i] for i in seen_parents if 0 <= i < len(self.chunks)]

    def s_hyde(self, query: str, k: int = 3) -> list[str]:
        prompt = f"請對以下問題給出一段假設性簡短答案（繁體中文）：\n{query}"
        hypo = self._llm(prompt, max_tokens=250, temperature=0.7) or query
        return self.s_semantic(hypo, k)

    _FN = {
        "semantic":     s_semantic,
        "tfidf":        s_tfidf,
        "hybrid":       s_hybrid,
        "rerank":       s_rerank,
        "multi_query":  s_multi_query,
        "compress":     s_compress,
        "parent_child": s_parent_child,
        "hyde":         s_hyde,
    }

    def generate_answer(self, query: str, strategy_key: str, top_k: int):
        if not self.chunks:
            return "請先上傳並載入 PDF 或 DOCX 文件。", ""
        if not query.strip():
            return "請輸入問題。", ""

        fn = self._FN.get(strategy_key, self.s_semantic)
        chunks = fn(self, query, int(top_k))
        context = "\n\n—\n\n".join(chunks)

        strategy_label = self.STRATEGY_MAP.get(strategy_key, strategy_key)
        source_preview = (
            f"文件：{self.source_name}\n"
            f"策略：{strategy_label} · 片段數：{len(chunks)}\n"
            f"ChromaDB Session：{self.session_id}\n\n"
            f"{'─' * 56}\n\n{context}"
        )

        if not self.client:
            return (
                "⚠ 尚未設定 Groq API Key。\n"
                "請在左欄「Step 01」輸入您的 Groq API Key 並點擊「套用」後再提問。\n\n"
                "（檢索已完成，可在下方「查看檢索到的文本片段」確認結果）",
                source_preview,
            )

        prompt = f"""請根據以下上下文回答問題。若上下文無相關資訊，請明確說明無法從文件回答，不要自行編造。

上下文：
{context}

問題：{query}

請用繁體中文詳細回答，並以條列方式整理重點："""

        try:
            r = self.client.chat.completions.create(
                model="llama-3.1-8b-instant",
                messages=[
                    {"role": "system", "content": "你是專業的文件分析與 RAG 問答助手。"},
                    {"role": "user", "content": prompt},
                ],
                max_tokens=1024,
                temperature=0.3,
            )
            return r.choices[0].message.content, source_preview
        except Exception as exc:
            return f"生成失敗：{type(exc).__name__}: {exc}", source_preview


# ══════════════════════════════════════════════════════════
#  Gradio UI
# ══════════════════════════════════════════════════════════
STRATEGY_INFO = [
    ("semantic",     "🔍 語意搜尋",   "ChromaDB 向量相似度，最通用"),
    ("tfidf",        "📊 TF-IDF",     "字元 n-gram 關鍵詞統計"),
    ("hybrid",       "⚡ 混合搜尋",   "語意 + TF-IDF 結果合併去重"),
    ("rerank",       "🎯 重新排序",   "LLM 對候選片段二次評分"),
    ("multi_query",  "🔄 多查詢擴展", "自動生成多角度問題聯合搜尋"),
    ("compress",     "✂️ 上下文壓縮", "LLM 提取最相關句子精簡上下文"),
    ("parent_child", "📂 父子文檔",   "小片段定位 → 回傳對應大片段"),
    ("hyde",         "💡 HyDE",       "先生成假設答案再語意搜尋"),
]

STRATEGY_LABEL_TO_KEY = {label: key for key, label, _ in STRATEGY_INFO}
STRATEGY_CHOICES = [label for _, label, _ in STRATEGY_INFO]
STRATEGY_DESC_HTML = "<br>".join(f"<b>{label}</b> — {desc}" for _, label, desc in STRATEGY_INFO)

CSS = """
body, .gradio-container { background:#f5f4f1 !important; }

#hdr {
    background:#fff;
    border:1px solid #e5e0d8;
    border-radius:14px;
    padding:28px 36px;
    margin-bottom:20px;
    border-top: 4px solid #2d6a4f;
}
.hdr-eyebrow { font-size:11px; letter-spacing:2.5px; color:#2d6a4f; text-transform:uppercase; margin-bottom:6px; }
.hdr-title   { font-size:26px; font-weight:700; color:#1a1714; margin:0 0 6px; }
.hdr-sub     { font-size:14px; color:#6b5e56; }
.pill { display:inline-block; margin:10px 5px 0 0; padding:3px 10px; border-radius:16px;
        font-size:11px; background:#e8f4f0; color:#2d6a4f; border:1px solid rgba(45,106,79,.2); }
.pill-amber { background:#fdf4e3; color:#b87a1a; border-color:rgba(184,122,26,.25); }

#apikey-box {
    background: #fffbf2;
    border: 1.5px solid #f0c96a;
    border-radius: 10px;
    padding: 12px 14px;
    margin-bottom: 8px;
}
#telegram-box {
    background: #eef6ff;
    border: 1.5px solid #8ec4f0;
    border-radius: 10px;
    padding: 12px 14px;
    margin-bottom: 8px;
}
#strategy-box {
    background:#fff;
    border:1.5px solid #e5e0d8;
    border-radius:10px;
    padding:10px 12px;
    margin-bottom: 8px;
}
#strategy-box .wrap { gap:6px !important; }
#strategy-box label {
    border:1.5px solid #e5e0d8 !important;
    border-radius:8px !important;
    padding:8px 10px !important;
    margin:0 !important;
    transition: border-color .15s, background .15s;
}
#strategy-box label:hover { border-color:#2d6a4f !important; }
#strategy-desc { font-size:11px; color:#7a6e67; line-height:1.6; margin-top:6px; }

.sec-label { font-size:11px; letter-spacing:1.5px; text-transform:uppercase;
             color:#7a6e67; font-weight:700; margin:16px 0 8px; }
.card-box { background:#fff !important; border:1px solid #e5e0d8 !important;
            border-radius:12px !important; padding:16px !important; }
#ask-btn     { background:#2d6a4f !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
#apply-key-btn { background:#b87a1a !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
#send-tg-btn { background:#0088cc !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
#verify-tg-btn { background:#0099aa !important; color:#fff !important; border:0 !important; border-radius:8px !important; }
"""

HEADER_HTML = """
<div id="hdr">
  <div class="hdr-eyebrow">Intelligent Document Analysis · v2</div>
  <div class="hdr-title">多策略 RAG 文件問答系統</div>
  <div class="hdr-sub">支援 PDF / DOCX 上傳，採用 ChromaDB 持久化向量資料庫與 8 種 RAG 檢索策略，並可將結果推送至 Telegram</div>
  <div>
    <span class="pill">▸ Groq API</span>
    <span class="pill">▸ llama-3.1-8b-instant</span>
    <span class="pill pill-amber">▸ ChromaDB</span>
    <span class="pill pill-amber">▸ PDF / DOCX</span>
    <span class="pill">▸ SentenceTransformers</span>
    <span class="pill">▸ Telegram</span>
  </div>
</div>
"""

EXAMPLE_QS = [
    ["這份文件的主要內容是什麼？"],
    ["文件中提到哪些重要概念或定義？"],
    ["有哪些關鍵數據、統計資料或案例？"],
    ["文件的結論或建議是什麼？"],
    ["文件提及哪些潛在風險或挑戰？"],
]


def create_interface():
    env_key = os.getenv("GROQ_API_KEY", "").strip()
    rag = MultiStrategyRAG(chroma_path="./chroma_db")
    if env_key:
        rag.set_api_key(env_key)

    current_strategy = {"key": "semantic"}
    last_result = {"answer": "", "source": ""}

    def apply_api_key(api_key: str):
        key = (api_key or "").strip()
        rag.set_api_key(key)
        if key:
            masked = key[:8] + "****" + key[-4:] if len(key) > 12 else "****"
            return f"✓ API Key 已套用（{masked}）"
        return "⚠ API Key 已清除，無法呼叫 LLM"

    def upload_document(file):
        if file is None:
            return "⚠ 請選擇 PDF 或 DOCX 檔案"
        return rag.load_document(file.name)

    def set_strategy(label: str):
        key = STRATEGY_LABEL_TO_KEY.get(label, "semantic")
        current_strategy["key"] = key
        return f"✓ 已選擇策略：{label}"

    def ask(query, top_k):
        answer, source = rag.generate_answer(query, current_strategy["key"], int(top_k))
        last_result["answer"] = answer or ""
        last_result["source"] = source or ""
        return answer, source

    def verify_bot(token_input: str, chat_id_input: str):
        """驗證 Bot Token 與 Chat ID 是否正確，發送一則測試訊息。"""
        token = (token_input or "").strip()
        chat_id = (chat_id_input or "").strip() or DEFAULT_TELEGRAM_CHAT_ID
        if not token:
            return "⚠ 請先輸入 Bot Token"
        result = send_telegram_message("✅ Telegram 推送測試成功！RAG 系統連線正常。", chat_id=chat_id, token=token)
        if result.get("ok"):
            return "✓ 測試訊息已送達，Bot Token 與 Chat ID 正確！"
        desc = result.get("description") or result.get("error") or str(result)
        # 給出更友善的錯誤提示
        if "Unauthorized" in desc:
            return "✗ Bot Token 錯誤（Unauthorized），請確認 Token 正確"
        if "chat not found" in desc:
            return f"✗ Chat ID「{chat_id}」找不到，請確認：\n1. 個人 ID 需先對 Bot 傳訊\n2. 群組 ID 需含負號，如 -1001234567890"
        return f"✗ 驗證失敗：{desc}"

    def push_to_telegram(include_source: bool, chat_id_input: str, token_input: str):
        if not last_result["answer"]:
            return "⚠ 尚無可推送的回答，請先提問。"

        token = (token_input or "").strip()
        if not token:
            return "⚠ 請先輸入 Telegram Bot Token"

        chat_id = (chat_id_input or "").strip() or DEFAULT_TELEGRAM_CHAT_ID

        # 避免 HTML parse_mode 與純文字衝突，移除尖括號
        answer_clean = re.sub(r"[<>]", "", last_result["answer"])

        text = (
            f"📄 文件：{rag.source_name or '未知'}\n"
            f"❓ 策略：{rag.STRATEGY_MAP.get(current_strategy['key'], current_strategy['key'])}\n\n"
            f"💬 回答：\n{answer_clean}"
        )

        if include_source:
            source_clean = re.sub(r"[<>]", "", last_result["source"])
            text += f"\n\n— 檢索片段 —\n{source_clean}"

        result = send_telegram_message(text, chat_id=chat_id, token=token)
        if result.get("ok"):
            return "✓ 已成功推送至 Telegram"
        desc = result.get("description") or result.get("error") or str(result)
        if "Unauthorized" in desc:
            return "✗ 推送失敗：Bot Token 錯誤（Unauthorized）"
        if "chat not found" in desc:
            return f"✗ 推送失敗：找不到 Chat ID「{chat_id}」，請先對 Bot 傳訊或確認群組 ID"
        return f"✗ 推送失敗：{desc}"

    with gr.Blocks(
        title="多策略 RAG 文件問答 v2",
        css=CSS,
        theme=gr.themes.Base(
            primary_hue=gr.themes.colors.green,
            neutral_hue=gr.themes.colors.stone,
        ),
    ) as demo:
        gr.HTML(HEADER_HTML)

        with gr.Row(equal_height=False):
            # ── 左欄 ──────────────────────────────────
            with gr.Column(scale=1, min_width=320, elem_classes="card-box"):

                gr.HTML("<div class='sec-label'>Step 00 · Telegram 推送</div>")
                with gr.Group(elem_id="telegram-box"):
                    tg_token = gr.Textbox(
                        label="Bot Token",
                        placeholder="123456789:AAAxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx",
                        type="password",
                        lines=1,
                    )
                    tg_chat_id = gr.Textbox(
                        label="Chat ID",
                        placeholder=f"預設：{DEFAULT_TELEGRAM_CHAT_ID}（留空使用此預設值）",
                        lines=1,
                    )
                    with gr.Row():
                        verify_tg_btn = gr.Button("🔍 測試連線", elem_id="verify-tg-btn", size="sm")
                        send_tg_btn = gr.Button("📨 推送回答", elem_id="send-tg-btn", size="sm")
                    tg_include_source = gr.Checkbox(label="同時推送檢索片段內容", value=False)
                    tg_status = gr.Textbox(label="推送狀態", interactive=False, lines=2)

                gr.HTML("<div class='sec-label'>Step 01 · Groq API Key</div>")
                with gr.Group(elem_id="apikey-box"):
                    api_key_input = gr.Textbox(
                        label="",
                        placeholder="gsk_xxxxxxxxxxxxxxxxxxxxxxxx",
                        value=env_key,
                        type="password",
                        lines=1,
                        show_label=False,
                    )
                    apply_key_btn = gr.Button("套用 API Key", size="sm", elem_id="apply-key-btn")
                    api_key_status = gr.Textbox(
                        value="✓ API Key 已從環境變數載入" if env_key else "⚠ 尚未設定 API Key",
                        interactive=False,
                        lines=1,
                        label="",
                        show_label=False,
                    )

                gr.HTML("<div class='sec-label'>Step 02 · 上傳文件</div>")
                file_input = gr.File(label="PDF / DOCX", file_types=[".pdf", ".docx"])
                load_btn = gr.Button("↑ 載入文件")
                status = gr.Textbox(label="狀態", interactive=False, lines=3)

                gr.HTML("<div class='sec-label'>Step 03 · 選擇 RAG 策略</div>")
                with gr.Group(elem_id="strategy-box"):
                    strategy_radio = gr.Radio(
                        choices=STRATEGY_CHOICES,
                        value=STRATEGY_CHOICES[0],
                        label="",
                        show_label=False,
                    )
                    gr.HTML(f"<div id='strategy-desc'>{STRATEGY_DESC_HTML}</div>")
                strategy_status = gr.Textbox(
                    value=f"✓ 已選擇策略：{STRATEGY_CHOICES[0]}",
                    interactive=False,
                    lines=1,
                    label="目前策略",
                )

                gr.HTML("<div class='sec-label'>Step 04 · 搜尋參數</div>")
                topk = gr.Slider(minimum=1, maximum=10, value=3, step=1, label="Top-K 片段數量")

            # ── 右欄：問答 ────────────────────────────
            with gr.Column(scale=2, elem_classes="card-box"):
                gr.HTML("<div class='sec-label'>Step 05 · 輸入問題</div>")
                qin = gr.Textbox(
                    label="",
                    placeholder="例如：這份文件的核心論點是什麼？",
                    lines=4,
                )
                ask_btn = gr.Button("提問", variant="primary", size="lg", elem_id="ask-btn")

                gr.HTML("<div class='sec-label'>AI 回答</div>")
                ans = gr.Textbox(label="", lines=12, interactive=False)

                with gr.Accordion("▸ 查看檢索到的文本片段", open=False):
                    src = gr.Textbox(label="", lines=10, interactive=False)

                gr.Examples(examples=EXAMPLE_QS, inputs=qin, label="範例問題")

        # ── 事件綁定 ──────────────────────────────────
        apply_key_btn.click(fn=apply_api_key, inputs=[api_key_input], outputs=[api_key_status])
        api_key_input.submit(fn=apply_api_key, inputs=[api_key_input], outputs=[api_key_status])
        load_btn.click(fn=upload_document, inputs=[file_input], outputs=[status])
        strategy_radio.change(fn=set_strategy, inputs=[strategy_radio], outputs=[strategy_status])
        ask_btn.click(fn=ask, inputs=[qin, topk], outputs=[ans, src])
        qin.submit(fn=ask, inputs=[qin, topk], outputs=[ans, src])
        verify_tg_btn.click(fn=verify_bot, inputs=[tg_token, tg_chat_id], outputs=[tg_status])
        send_tg_btn.click(
            fn=push_to_telegram,
            inputs=[tg_include_source, tg_chat_id, tg_token],
            outputs=[tg_status],
        )

    return demo


if __name__ == "__main__":
    demo = create_interface()
    demo.launch(share=False, server_name="0.0.0.0")