server.py

# ClareVoice FastAPI server: React UI + same backend as app.py (Weaviate + FAISS).
# Run: uvicorn server:app --host 0.0.0.0 --port 7860

import os
import re
import time
import concurrent.futures
from collections import defaultdict
from typing import Dict, List, Any, Optional

from fastapi import FastAPI, UploadFile, File, Form
from fastapi.responses import FileResponse, JSONResponse, Response
from fastapi.staticfiles import StaticFiles
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

from config import (
    DEFAULT_MODEL,
    DEFAULT_COURSE_TOPICS,
    USE_WEAVIATE_DIRECT,
    GENAI_COURSES_SPACE,
    WEAVIATE_URL,
    WEAVIATE_API_KEY,
    WEAVIATE_COLLECTION,
)
from clare_core import (
    detect_language,
    chat_with_clare,
    update_weaknesses_from_message,
    update_cognitive_state_from_message,
    render_session_status,
    export_conversation,
    summarize_conversation,
)
from rag_engine import build_rag_chunks_from_file, retrieve_relevant_chunks
from syllabus_utils import extract_course_topics_from_file
from tts_podcast import (
    text_to_speech,
    build_podcast_script_from_history,
    build_podcast_script_from_summary,
    generate_podcast_audio,
)

MODULE10_PATH = os.path.join(os.path.dirname(__file__), "module10_responsible_ai.pdf")
MODULE10_DOC_TYPE = "Literature Review / Paper"

# Preload Module 10 (same as app.py)
preloaded_topics: List[str] = []
preloaded_chunks: List[Dict] = []
if os.path.exists(MODULE10_PATH):
    try:
        class _FileObj:
            name = MODULE10_PATH
        preloaded_topics = extract_course_topics_from_file(_FileObj(), MODULE10_DOC_TYPE) or []
        preloaded_chunks = build_rag_chunks_from_file(MODULE10_PATH, MODULE10_DOC_TYPE) or []
        print("[server] Module 10 preloaded.")
    except Exception as e:
        print(f"[server] Module 10 preload failed: {e}")
if not preloaded_topics:
    preloaded_topics = list(DEFAULT_COURSE_TOPICS)

_WEAVIATE_EMBED_MODEL = None

def _get_weaviate_embed_model():
    """使用 HF 免费 sentence-transformers（与建索引时一致）。"""
    global _WEAVIATE_EMBED_MODEL
    if _WEAVIATE_EMBED_MODEL is None:
        from llama_index.embeddings.huggingface import HuggingFaceEmbedding
        _WEAVIATE_EMBED_MODEL = HuggingFaceEmbedding(
            model_name="sentence-transformers/all-MiniLM-L6-v2"
        )
    return _WEAVIATE_EMBED_MODEL

def _retrieve_from_weaviate(question: str, top_k: int = 5, timeout_sec: float = 45.0) -> str:
    if not USE_WEAVIATE_DIRECT or len(question.strip()) < 5:
        return ""
    def _call():
        try:
            import weaviate
            from weaviate.classes.init import Auth
            from llama_index.core import Settings, VectorStoreIndex
            from llama_index.vector_stores.weaviate import WeaviateVectorStore
            Settings.embed_model = _get_weaviate_embed_model()
            client = weaviate.connect_to_weaviate_cloud(
                cluster_url=WEAVIATE_URL,
                auth_credentials=Auth.api_key(WEAVIATE_API_KEY),
            )
            try:
                if not client.is_ready():
                    return ""
                vs = WeaviateVectorStore(weaviate_client=client, index_name=WEAVIATE_COLLECTION)
                index = VectorStoreIndex.from_vector_store(vs)
                nodes = index.as_retriever(similarity_top_k=top_k).retrieve(question)
                return "\n\n---\n\n".join(n.get_content() for n in nodes) if nodes else ""
            finally:
                client.close()
        except Exception as e:
            print(f"[weaviate] retrieve failed: {repr(e)}")
            return ""
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
            return ex.submit(_call).result(timeout=timeout_sec)
    except concurrent.futures.TimeoutError:
        print(f"[weaviate] timeout after {timeout_sec}s")
        return ""

def _retrieve_from_genai_courses(question: str, top_k: int = 5, timeout_sec: float = 25.0) -> str:
    if not GENAI_COURSES_SPACE or len(question.strip()) < 5:
        return ""
    def _call():
        try:
            from gradio_client import Client
            c = Client(GENAI_COURSES_SPACE)
            return (c.predict(question, api_name="/retrieve") or "").strip()
        except Exception as e:
            print(f"[genai_courses] failed: {repr(e)}")
            return ""
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as ex:
            return ex.submit(_call).result(timeout=timeout_sec)
    except concurrent.futures.TimeoutError:
        return ""

def format_references(rag_chunks: List[Dict], max_files: int = 2, max_sections_per_file: int = 3) -> str:
    if not rag_chunks:
        return "\n".join(["**References:**", "- (No RAG context used. Answer is based on the model's general knowledge.)"])
    chunks = list(rag_chunks)
    chunks.sort(key=lambda c: float(c.get("_rag_score", 0.0)), reverse=True)
    refs_by_file: Dict[str, List[str]] = defaultdict(list)
    for chunk in chunks:
        file_name = chunk.get("source_file") or "module10_responsible_ai.pdf"
        section = chunk.get("section") or "Related section"
        score = chunk.get("_rag_score")
        score_str = f" (score={float(score):.2f})" if score is not None else ""
        entry = section + score_str
        if entry not in refs_by_file[file_name]:
            refs_by_file[file_name].append(entry)
    if not refs_by_file:
        return "\n".join(["**References:**", "- (No RAG context used.)"])
    lines = ["**References (RAG context used):**"]
    for i, (file_name, sections) in enumerate(refs_by_file.items()):
        if i >= max_files:
            break
        lines.append(f"- *{file_name}* — {'; '.join(sections[:max_sections_per_file])}")
    return "\n".join(lines)

def is_academic_query(message: str) -> bool:
    if not message or not message.strip():
        return False
    m = " ".join(message.strip().lower().split())
    smalltalk = {"hi", "hello", "hey", "thanks", "thank", "ok", "okay", "bye", "goodbye", "haha", "lol"}
    tokens = m.split()
    if "?" not in m and all(t in smalltalk for t in tokens):
        return False
    meta = ["who are you", "what are you", "what is your name", "what can you do", "what is clare"]
    if any(p in m for p in meta):
        return False
    if len(tokens) <= 2 and "?" not in m:
        return False
    return True

MODULE10_DOC_TYPE = "Literature Review / Paper"
MICRO_QUIZ_INSTRUCTION = (
    "We are running a short micro-quiz session based ONLY on **Module 10 – "
    "Responsible AI (Alto, 2024, Chapter 12)** and the pre-loaded materials.\n\n"
    "Step 1 – Before asking any content question:\n"
    "• First ask me which quiz style I prefer right now:\n"
    "  - (1) Multiple-choice questions\n"
    "  - (2) Short-answer / open-ended questions\n"
    "• Ask me explicitly: \"Which quiz style do you prefer now: 1) Multiple-choice or 2) Short-answer? "
    "Please reply with 1 or 2.\"\n"
    "• Do NOT start a content question until I have answered 1 or 2.\n\n"
    "Step 2 – After I choose the style:\n"
    "• If I choose 1 (multiple-choice):\n"
    "  - Ask ONE multiple-choice question at a time, based on Module 10 concepts.\n"
    "  - Provide 3–4 options (A, B, C, D) and make only one option clearly correct.\n"
    "• If I choose 2 (short-answer):\n"
    "  - Ask ONE short-answer question at a time, also based on Module 10 concepts.\n"
    "  - Do NOT show the answer when you ask the question.\n\n"
    "Step 3 – For each answer I give:\n"
    "• Grade my answer (correct / partially correct / incorrect).\n"
    "• Give a brief explanation and the correct answer.\n"
    "• Then ask if I want another question of the SAME style.\n\n"
    "Please start by asking me which quiz style I prefer (1 = multiple-choice, 2 = short-answer). "
    "Do not ask any content question before I choose."
)

# ----------------------------
# Session store (in-memory)
# ----------------------------
SESSIONS: Dict[str, Dict[str, Any]] = {}

def _get_session(user_id: str) -> Dict[str, Any]:
    if user_id not in SESSIONS:
        SESSIONS[user_id] = {
            "user_id": user_id,
            "name": "",
            "history": [],
            "weaknesses": [],
            "cognitive_state": {"confusion": 0, "mastery": 0},
            "course_outline": list(preloaded_topics) if preloaded_topics else list(DEFAULT_COURSE_TOPICS),
            "rag_chunks": list(preloaded_chunks) if preloaded_chunks else [],
            "model_name": DEFAULT_MODEL,
            "uploaded_files": [],
            "profile_bio": "",
            "init_answers": {},
            "init_dismiss_until": 0,
        }
    return SESSIONS[user_id]

# ----------------------------
# App
# ----------------------------
app = FastAPI(title="ClareVoice API")
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

WEB_DIR = os.path.join(os.path.dirname(__file__), "web", "build")
WEB_INDEX = os.path.join(WEB_DIR, "index.html")
WEB_ASSETS = os.path.join(WEB_DIR, "assets")

if os.path.isdir(WEB_ASSETS):
    app.mount("/assets", StaticFiles(directory=WEB_ASSETS), name="assets")
if os.path.isdir(WEB_DIR):
    app.mount("/static", StaticFiles(directory=WEB_DIR), name="static")

# ----------------------------
# Request models
# ----------------------------
class LoginReq(BaseModel):
    name: str
    user_id: str

class ChatReq(BaseModel):
    user_id: str
    message: str
    learning_mode: str = "Concept Explainer"
    language_preference: str = "Auto"
    doc_type: str = "Syllabus"

class QuizStartReq(BaseModel):
    user_id: str
    language_preference: str = "Auto"
    doc_type: str = MODULE10_DOC_TYPE
    learning_mode: str = "quiz"

class ExportReq(BaseModel):
    user_id: str
    learning_mode: str

class SummaryReq(BaseModel):
    user_id: str
    learning_mode: str
    language_preference: str = "Auto"

class TtsReq(BaseModel):
    user_id: str
    text: str
    voice: Optional[str] = "nova"

class PodcastReq(BaseModel):
    user_id: str
    source: str = "summary"
    voice: Optional[str] = "nova"

class FeedbackReq(BaseModel):
    user_id: str
    rating: str
    run_id: Optional[str] = None
    assistant_message_id: Optional[str] = None
    assistant_text: str = ""
    user_text: Optional[str] = None
    comment: Optional[str] = None
    refs: Optional[List] = None
    tags: Optional[List] = None
    timestamp_ms: Optional[int] = None
    learning_mode: Optional[str] = None
    doc_type: Optional[str] = None

# ----------------------------
# Routes
# ----------------------------
@app.get("/")
def index():
    if os.path.exists(WEB_INDEX):
        return FileResponse(WEB_INDEX)
    return JSONResponse({"detail": "web/build not found. Build frontend first."}, status_code=500)

@app.post("/api/login")
def login(req: LoginReq):
    user_id = (req.user_id or "").strip()
    name = (req.name or "").strip()
    if not user_id or not name:
        return JSONResponse({"ok": False, "error": "Missing name/user_id"}, status_code=400)
    sess = _get_session(user_id)
    sess["name"] = name
    return {"ok": True, "user": {"name": name, "user_id": user_id}}

@app.post("/api/chat")
def chat(req: ChatReq):
    user_id = (req.user_id or "").strip()
    msg = (req.message or "").strip()
    if not user_id:
        return JSONResponse({"error": "Missing user_id"}, status_code=400)

    sess = _get_session(user_id)
    if not msg:
        return {
            "reply": "",
            "session_status_md": render_session_status(
                req.learning_mode, sess["weaknesses"], sess["cognitive_state"]
            ),
            "refs": [],
            "latency_ms": 0.0,
            "run_id": None,
        }

    t0 = time.time()
    resolved_lang = detect_language(msg, req.language_preference)
    sess["weaknesses"] = update_weaknesses_from_message(msg, sess["weaknesses"])
    sess["cognitive_state"] = update_cognitive_state_from_message(msg, sess["cognitive_state"])

    rag_context_text = ""
    rag_used_chunks: List[Dict] = []
    if is_academic_query(msg):
        rag_context_text, rag_used_chunks = retrieve_relevant_chunks(msg, sess["rag_chunks"] or [])
        course_chunks = ""
        course_source = ""
        if USE_WEAVIATE_DIRECT:
            course_chunks = _retrieve_from_weaviate(msg)
            course_source = "Weaviate Cloud (GENAI COURSES)"
        elif GENAI_COURSES_SPACE:
            course_chunks = _retrieve_from_genai_courses(msg)
            course_source = "GenAICoursesDB"
        if course_chunks and course_source:
            rag_context_text = (rag_context_text or "") + "\n\n[来自 GENAI 课程知识库]\n\n" + course_chunks
            rag_used_chunks = list(rag_used_chunks or []) + [
                {"source_file": course_source, "section": "retrieve (GENAI COURSES dataset)", "_rag_score": 1.0}
            ]

    try:
        answer, new_history = chat_with_clare(
            message=msg,
            history=sess["history"],
            model_name=sess["model_name"],
            language_preference=resolved_lang,
            learning_mode=req.learning_mode,
            doc_type=req.doc_type,
            course_outline=sess["course_outline"],
            weaknesses=sess["weaknesses"],
            cognitive_state=sess["cognitive_state"],
            rag_context=rag_context_text,
        )
    except Exception as e:
        print(f"[chat] error: {repr(e)}")
        return JSONResponse({"error": f"chat failed: {repr(e)}"}, status_code=500)

    sess["history"] = new_history
    total_ms = (time.time() - t0) * 1000.0

    ref_text = format_references(rag_used_chunks) if is_academic_query(msg) else ""
    if ref_text and new_history:
        last_u, last_a = new_history[-1]
        if "References (RAG context used):" not in (last_a or ""):
            answer = f"{last_a or ''}\n\n{ref_text}"

    refs = [{"source_file": c.get("source_file"), "section": c.get("section")} for c in (rag_used_chunks or [])]
    if not refs:
        refs = [{"source_file": "No RAG", "section": "Answer based on model general knowledge."}]

    return {
        "reply": answer,
        "session_status_md": render_session_status(req.learning_mode, sess["weaknesses"], sess["cognitive_state"]),
        "refs": refs,
        "latency_ms": total_ms,
        "run_id": None,
    }

@app.post("/api/quiz/start")
def quiz_start(req: QuizStartReq):
    user_id = (req.user_id or "").strip()
    if not user_id:
        return JSONResponse({"error": "Missing user_id"}, status_code=400)
    sess = _get_session(user_id)
    resolved_lang = detect_language(MICRO_QUIZ_INSTRUCTION, req.language_preference)
    quiz_ctx_text, _ = retrieve_relevant_chunks("Module 10 quiz", sess["rag_chunks"] or [])
    try:
        answer, new_history = chat_with_clare(
            message=MICRO_QUIZ_INSTRUCTION,
            history=sess["history"],
            model_name=sess["model_name"],
            language_preference=resolved_lang,
            learning_mode=req.learning_mode,
            doc_type=req.doc_type,
            course_outline=sess["course_outline"],
            weaknesses=sess["weaknesses"],
            cognitive_state=sess["cognitive_state"],
            rag_context=quiz_ctx_text,
        )
    except Exception as e:
        print(f"[quiz] error: {repr(e)}")
        return JSONResponse({"error": str(e)}, status_code=500)
    sess["history"] = new_history
    return {
        "reply": answer,
        "session_status_md": render_session_status(req.learning_mode, sess["weaknesses"], sess["cognitive_state"]),
        "refs": [],
        "latency_ms": 0.0,
        "run_id": None,
    }

@app.post("/api/upload")
async def upload(
    user_id: str = Form(...),
    doc_type: str = Form(...),
    file: UploadFile = File(...),
):
    user_id = (user_id or "").strip()
    doc_type = (doc_type or "").strip()
    if not user_id:
        return JSONResponse({"ok": False, "error": "Missing user_id"}, status_code=400)
    if not file or not file.filename:
        return JSONResponse({"ok": False, "error": "Missing file"}, status_code=400)

    sess = _get_session(user_id)
    safe_name = os.path.basename(file.filename).replace("..", "_")
    tmp_path = os.path.join("/tmp", safe_name)
    content = await file.read()
    with open(tmp_path, "wb") as f:
        f.write(content)

    if doc_type == "Syllabus":
        class _F:
            pass
        fo = _F()
        fo.name = tmp_path
        try:
            sess["course_outline"] = extract_course_topics_from_file(fo, doc_type)
        except Exception as e:
            print(f"[upload] syllabus parse error: {repr(e)}")

    try:
        new_chunks = build_rag_chunks_from_file(tmp_path, doc_type) or []
        sess["rag_chunks"] = (sess["rag_chunks"] or []) + new_chunks
    except Exception as e:
        print(f"[upload] rag build error: {repr(e)}")
        new_chunks = []
    sess.setdefault("uploaded_files", []).append({
        "filename": safe_name, "doc_type": doc_type, "added_chunks": len(new_chunks), "ts": int(time.time()),
    })
    return {"ok": True, "added_chunks": len(new_chunks), "status_md": f"✅ Loaded base reading + uploaded {doc_type} file."}

@app.post("/api/feedback")
def api_feedback(req: FeedbackReq):
    user_id = (req.user_id or "").strip()
    if not user_id:
        return JSONResponse({"ok": False, "error": "Missing user_id"}, status_code=400)
    if (req.rating or "").strip().lower() not in ("helpful", "not_helpful"):
        return JSONResponse({"ok": False, "error": "Invalid rating"}, status_code=400)
    return {"ok": True}

@app.post("/api/export")
def api_export(req: ExportReq):
    user_id = (req.user_id or "").strip()
    if not user_id:
        return JSONResponse({"error": "Missing user_id"}, status_code=400)
    sess = _get_session(user_id)
    md = export_conversation(
        sess["history"], sess["course_outline"], req.learning_mode,
        sess["weaknesses"], sess["cognitive_state"],
    )
    return {"markdown": md}

@app.post("/api/summary")
def api_summary(req: SummaryReq):
    user_id = (req.user_id or "").strip()
    if not user_id:
        return JSONResponse({"error": "Missing user_id"}, status_code=400)
    sess = _get_session(user_id)
    lang = (req.language_preference or "Auto").strip()
    if lang == "简体中文":
        lang = "中文"
    md = summarize_conversation(
        sess["history"], sess["course_outline"], sess["weaknesses"],
        sess["cognitive_state"], sess["model_name"], lang,
    )
    return {"markdown": md}

@app.post("/api/tts")
def api_tts(req: TtsReq):
    user_id = (req.user_id or "").strip()
    if not user_id:
        return JSONResponse({"error": "Missing user_id"}, status_code=400)
    text = (req.text or "").strip()
    if not text:
        return JSONResponse({"error": "Missing text"}, status_code=400)
    try:
        audio_bytes = text_to_speech(text, voice=req.voice or "nova")
    except Exception as e:
        print(f"[tts] error: {repr(e)}")
        return JSONResponse({"error": str(e)}, status_code=500)
    if not audio_bytes:
        return JSONResponse({"error": "No audio generated"}, status_code=500)
    return Response(content=audio_bytes, media_type="audio/mpeg")

@app.post("/api/podcast")
def api_podcast(req: PodcastReq):
    user_id = (req.user_id or "").strip()
    if not user_id:
        return JSONResponse({"error": "Missing user_id"}, status_code=400)
    sess = _get_session(user_id)
    source = (req.source or "summary").lower()
    voice = req.voice or "nova"
    try:
        if source == "conversation":
            script = build_podcast_script_from_history(sess["history"])
        else:
            md = summarize_conversation(
                sess["history"], sess["course_outline"], sess["weaknesses"],
                sess["cognitive_state"], sess["model_name"], "Auto",
            )
            script = build_podcast_script_from_summary(md)
        audio_bytes = generate_podcast_audio(script, voice=voice)
    except Exception as e:
        print(f"[podcast] error: {repr(e)}")
        return JSONResponse({"error": str(e)}, status_code=500)
    if not audio_bytes:
        return JSONResponse({"error": "No audio generated"}, status_code=500)
    return Response(content=audio_bytes, media_type="audio/mpeg")

@app.get("/api/memoryline")
def memoryline(user_id: str):
    _ = _get_session((user_id or "").strip())
    return {"next_review_label": "T+7", "progress_pct": 0.4}

@app.get("/api/profile/status")
def profile_status(user_id: str):
    user_id = (user_id or "").strip()
    if not user_id:
        return JSONResponse({"error": "Missing user_id"}, status_code=400)
    sess = _get_session(user_id)
    bio = (sess.get("profile_bio") or "").strip()
    need_init = len(bio) <= 50 and (int(time.time()) >= int(sess.get("init_dismiss_until") or 0))
    return {"need_init": need_init, "bio_length": len(bio)}

@app.get("/health")
def health():
    return {"status": "ok"}

@app.get("/{full_path:path}")
def spa_fallback(full_path: str):
    if full_path.startswith("api/") or full_path.startswith("assets/") or full_path.startswith("static/"):
        return JSONResponse({"detail": "Not Found"}, status_code=404)
    if os.path.exists(WEB_INDEX):
        return FileResponse(WEB_INDEX)
    return JSONResponse({"detail": "web/build not found"}, status_code=500)