MZC/KnowMathBot/backend/app/routers/digital_human.py

"""
数字人路由：多轮问答、题库例题检索、DeepSeek 调用、可选 TTS 与语音资源管理。
"""
from pathlib import Path
import base64
import os
import tempfile
import random
import re
from typing import Annotated
from uuid import UUID

from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
from fastapi.responses import FileResponse
import httpx
from sqlmodel import Session, select

from app.core.config import settings
from app.db import get_session
from app.deps import get_current_user, require_role
from app.models import ChatMessage, ChatSession, Exercise, User, UserRole
from app.schemas import AskRequest, AskResponse, VoiceAssetItem
from app.utils.exercise_constants import RESOLUTION_EMPTY_PLACEHOLDER
from app.utils.exercise_knowledge import exercise_matches_knowledge_point

router = APIRouter(prefix="/digital-human", tags=["digital-human"])
VOICE_EXTS = {".wav", ".mp3", ".m4a", ".ogg"}


def _voice_assets_dir() -> Path:
    p = Path(settings.voice_assets_dir).expanduser()
    p.mkdir(parents=True, exist_ok=True)
    return p


def _safe_voice_name(name: str) -> str:
    base = Path(name).name.strip()
    if not base:
        raise HTTPException(status_code=400, detail="文件名不能为空")
    ext = Path(base).suffix.lower()
    if ext not in VOICE_EXTS:
        raise HTTPException(status_code=400, detail="仅支持 wav/mp3/m4a/ogg")
    return base


def _resolve_local_tts_paths() -> tuple[Path | None, Path | None, str | None]:
    ckpt_raw = settings.local_tts_ckpt_path.strip()
    pth_raw = settings.local_tts_pth_path.strip()

    ckpt = Path(ckpt_raw).expanduser() if ckpt_raw else None
    pth = Path(pth_raw).expanduser() if pth_raw else None

    base_dir = Path(settings.local_tts_model_dir).expanduser()
    base_dir.mkdir(parents=True, exist_ok=True)

    if ckpt is None:
        candidates = [p for p in base_dir.rglob("*.ckpt") if p.is_file()]
        if candidates:
            ckpt = sorted(candidates, key=lambda x: x.stat().st_mtime, reverse=True)[0]
    if pth is None:
        candidates = [p for p in base_dir.rglob("*.pth") if p.is_file()]
        if candidates:
            pth = sorted(candidates, key=lambda x: x.stat().st_mtime, reverse=True)[0]

    hint = None
    if ckpt is None or pth is None:
        hint = f"请将模型放到 {base_dir}，或在 .env 设置 LOCAL_TTS_CKPT_PATH / LOCAL_TTS_PTH_PATH"
    return ckpt, pth, hint


def _is_deepseek_selected(model_name: str | None, model_provider: str | None) -> bool:
    provider = (model_provider or "").strip().lower()
    if provider:
        return provider == "deepseek"
    if not model_name:
        return False
    return "deepseek" in model_name.lower()


def _build_default_answer() -> str:
    return (
        "我先给出一个解题思路：把题目条件转成方程/不等式，再根据知识点逐步推导。"
        "如果你把具体题目发完整（含图形/已知量/求什么），我可以一步步讲解。"
    )


def _extract_example_limit(question: str) -> int:
    """
    粗略解析用户想要的例题数量：优先识别数字，其次识别“两/二”。
    """
    q = question or ""
    m = re.search(r"([1-9]|10)\s*(道|题|个)?", q)
    if m:
        return int(m.group(1))
    if "两" in q or "二" in q:
        return 2
    return 3


def _question_wants_examples(question: str) -> bool:
    """
    判断用户是否在聊天里提出“要例题/举例/给例子”之类诉求。
    """
    q = (question or "").strip()
    if not q:
        return False
    keywords = [
        "例题",
        "例子",
        "举例",
        "示例",
        "给我题",
        "给我几道",
        "来几道",
        "来几题",
        "给我几题",
        "给出几道",
        "练习题",
        "习题",
        "来一题",
        "来一道",
        "出一题",
        "出几道",
        "做几道",
        "题库",
        "抽题",
        "换几道题",
    ]
    return any(k in q for k in keywords)


def _pick_weighted_without_replacement(candidates: list[Exercise], limit: int) -> list[Exercise]:
    """
    按 weight 做不放回加权抽样；weight 高的题更容易被抽到。
    """
    remaining = list(candidates)
    picked: list[Exercise] = []
    limit = max(1, int(limit))
    while remaining and len(picked) < limit:
        weights: list[float] = []
        for ex in remaining:
            w = int(getattr(ex, "weight", 0) or 0)
            weights.append(float(w + 1))  # 避免全为 0 时无法抽样
        total = sum(weights)
        if total <= 0:
            chosen = remaining[0]
        else:
            r = random.random() * total
            acc = 0.0
            chosen = remaining[-1]
            for ex, w in zip(remaining, weights):
                acc += w
                if r <= acc:
                    chosen = ex
                    break
        remaining = [ex for ex in remaining if ex.id != chosen.id]
        picked.append(chosen)
    return picked


def _build_examples_answer(
    question: str,
    knowledge_code: str | None,
    session: Session,
) -> tuple[str, list[str]] | None:
    """
    若用户请求例题，则从题库抽取同一 knowledge_code 的题并输出题干+解析/答案。
    返回 None 表示不触发该逻辑或题库无匹配题目。
    """
    if not knowledge_code or not knowledge_code.strip():
        return None
    if not _question_wants_examples(question):
        return None

    limit = _extract_example_limit(question)
    code = knowledge_code.strip()

    all_exs = session.exec(select(Exercise)).all()
    candidates = [ex for ex in all_exs if exercise_matches_knowledge_point(ex, code)]
    if not candidates:
        return None

    picked = _pick_weighted_without_replacement(candidates, limit)

    sources: list[str] = []
    parts: list[str] = []
    parts.append(f"根据你选择的知识点（`{code}`），我从题库整理了 {len(picked)} 道例题与解析：")
    for i, ex in enumerate(picked, start=1):
        title = (ex.title or "").strip() or f"例题 {i}"
        body = (ex.body or "").strip()
        correct_answer = (ex.correct_answer or "").strip() if getattr(ex, "correct_answer", None) else ""
        resolution = ((ex.resolution or "").strip() if getattr(ex, "resolution", None) else "") or RESOLUTION_EMPTY_PLACEHOLDER
        sources.append(str(ex.id))

        parts.append(f"\n{i}. {title}\n题干：{body}")
        if correct_answer:
            parts.append(f"参考答案：{correct_answer}")
        parts.append(f"解析：{resolution}")

    return "\n".join(parts).strip(), sources


def _generate_answer_with_deepseek(question: str, knowledge_code: str | None, model_id: str | None) -> str:
    api_key = settings.deepseek_api_key.strip()
    if not api_key:
        return "DeepSeek 未配置 API Key（请在 backend/.env 设置 DEEPSEEK_API_KEY），当前使用占位回复。"

    prompt_context = f"当前知识点编码：{knowledge_code}" if knowledge_code else "当前无指定知识点编码"
    messages = [
        {
            "role": "system",
            "content": (
                "你是一个面向中学生数学辅导的助教。回答要准确、分步骤、尽量简洁。"
                "数学公式请使用 LaTeX：行内用 \\( ... \\)，独立成行用 \\[ ... \\] 或 $$ ... $$；"
                "不要用 Markdown 代码块（```）包裹公式。"
                "避免滥用 Markdown 小标题与加粗，以自然段与编号步骤为主，便于阅读。"
            ),
        },
        {"role": "system", "content": prompt_context},
        {"role": "user", "content": question},
    ]
    payload = {
        "model": (model_id or "").strip() or settings.deepseek_model,
        "messages": messages,
        "temperature": 0.5,
    }
    base_url = settings.deepseek_base_url.rstrip("/")
    url = f"{base_url}/chat/completions"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    try:
        with httpx.Client(timeout=settings.llm_request_timeout_seconds) as client:
            res = client.post(url, json=payload, headers=headers)
            res.raise_for_status()
            data = res.json()
            return data["choices"][0]["message"]["content"].strip()
    except Exception as e:
        return f"DeepSeek 调用失败（{e}），当前使用占位回复。"


def _probe_local_tts_model() -> str:
    if not settings.local_tts_enabled:
        return "未启用本地TTS模型"

    ckpt, pth, hint = _resolve_local_tts_paths()
    if ckpt is None:
        return f"本地TTS不可用：未找到ckpt文件。{hint or ''}".strip()
    if pth is None:
        return f"本地TTS不可用：未找到pth文件。{hint or ''}".strip()
    if not ckpt.exists():
        return f"本地TTS不可用：找不到ckpt文件（{ckpt}）"
    if not pth.exists():
        return f"本地TTS不可用：找不到pth文件（{pth}）"

    try:
        import torch
    except ImportError:
        return "本地TTS不可用：未安装 torch（可选依赖）"

    try:
        _ = torch.load(ckpt, map_location="cpu")
    except Exception as e:
        return f"本地TTS不可用：ckpt加载失败（{e}）"

    try:
        _ = torch.load(pth, map_location="cpu", weights_only=False)
    except Exception as e:
        return f"本地TTS不可用：pth加载失败（{e}）"

    return f"本地TTS模型加载检查通过（ckpt={ckpt.name}, pth={pth.name}；已接入检测，尚未执行真实语音合成）"


def _normalize_tts_api_url(raw_url: str) -> str:
    url = (raw_url or "").strip().rstrip("/")
    if not url:
        return ""
    if url.endswith("/tts"):
        return url
    return f"{url}/tts"


def _synthesize_with_http_tts(text: str) -> tuple[str | None, str | None, str]:
    raw_url = settings.local_tts_api_url
    url = _normalize_tts_api_url(raw_url)
    if not url:
        return None, None, "本地TTS HTTP未配置：请在 .env 设置 LOCAL_TTS_API_URL"

    payload = {
        "text": text,
        "text_lang": settings.local_tts_text_lang,
    }
    if settings.local_tts_ref_wav_path.strip():
        payload["ref_audio_path"] = settings.local_tts_ref_wav_path.strip()
    if settings.local_tts_prompt_text.strip():
        payload["prompt_text"] = settings.local_tts_prompt_text.strip()
    if settings.local_tts_prompt_lang.strip():
        payload["prompt_lang"] = settings.local_tts_prompt_lang.strip()

    try:
        with httpx.Client(timeout=settings.local_tts_api_timeout_seconds) as client:
            res = client.post(url, json=payload)
            res.raise_for_status()
    except Exception as e:
        return None, None, f"本地TTS HTTP调用失败（{e}）"

    content_type = (res.headers.get("content-type") or "").split(";")[0].strip().lower()
    if content_type.startswith("audio/"):
        try:
            b64 = base64.b64encode(res.content).decode("ascii")
            return b64, content_type, f"已使用本地TTS HTTP接口合成语音（{url}）"
        except Exception as e:
            return None, None, f"本地TTS HTTP音频解析失败（{e}）"

    try:
        data = res.json()
    except Exception:
        return None, None, f"本地TTS HTTP返回非音频且非JSON（content-type={content_type or 'unknown'}）"

    audio_b64 = None
    audio_mime = "audio/wav"
    if isinstance(data, dict):
        if isinstance(data.get("audio_base64"), str):
            audio_b64 = data["audio_base64"]
            audio_mime = data.get("audio_mime") or audio_mime
        elif isinstance(data.get("audio"), str):
            audio_b64 = data["audio"]
            audio_mime = data.get("mime") or audio_mime
        elif isinstance(data.get("data"), dict) and isinstance(data["data"].get("audio_base64"), str):
            audio_b64 = data["data"]["audio_base64"]
            audio_mime = data["data"].get("audio_mime") or audio_mime

    if not audio_b64:
        return None, None, "本地TTS HTTP返回JSON但未找到 audio_base64/audio 字段"
    return audio_b64, audio_mime, f"已使用本地TTS HTTP接口合成语音（{url}）"


def _synthesize_audio_base64(text: str) -> tuple[str | None, str | None, str]:
    provider = settings.local_tts_provider.strip().lower()
    http_fail: str | None = None
    # GPT-SoVITS HTTP 模式不需要本地 ckpt/pth；原先先探测权重会导致未放模型时永远不请求 HTTP。
    if settings.local_tts_enabled and provider in {"gpt_sovits_http", "http", "api"}:
        audio_b64, audio_mime, status = _synthesize_with_http_tts(text)
        if audio_b64:
            return audio_b64, audio_mime, status
        http_fail = status

    probe = _probe_local_tts_model()
    if "加载检查通过" not in probe:
        if http_fail:
            return None, None, f"{http_fail}；{probe}"
        return None, None, probe

    if http_fail:
        probe = f"{probe}；{http_fail}；已回退到后端系统语音。"

    try:
        import pyttsx3  # type: ignore
    except Exception:
        return None, None, f"{probe}；未安装pyttsx3，已回退到前端浏览器语音。"

    tmp_path = ""
    try:
        fd, tmp_path = tempfile.mkstemp(suffix=".wav")
        os.close(fd)
        engine = pyttsx3.init()
        engine.save_to_file(text, tmp_path)
        engine.runAndWait()
        with open(tmp_path, "rb") as f:
            payload = base64.b64encode(f.read()).decode("ascii")
        return payload, "audio/wav", "本地模型检查通过；当前使用后端系统语音做兼容播放。"
    except Exception as e:
        return None, None, f"{probe}；后端语音生成失败（{e}），已回退到前端浏览器语音。"
    finally:
        if tmp_path and os.path.exists(tmp_path):
            try:
                os.remove(tmp_path)
            except Exception:
                pass


@router.get("/voice-assets", response_model=list[VoiceAssetItem])
def list_voice_assets():
    base = _voice_assets_dir()
    items: list[VoiceAssetItem] = []
    for fp in sorted(base.glob("*")):
        if not fp.is_file() or fp.suffix.lower() not in VOICE_EXTS:
            continue
        items.append(
            VoiceAssetItem(
                name=fp.name,
                size=fp.stat().st_size,
                url=f"{settings.api_v1_prefix}/digital-human/voice-assets/{fp.name}",
            )
        )
    return items


@router.get("/voice-assets/{name}")
def get_voice_asset(name: str):
    safe = _safe_voice_name(name)
    fp = _voice_assets_dir() / safe
    if not fp.exists():
        raise HTTPException(status_code=404, detail="语音文件不存在")
    media_type = {
        ".wav": "audio/wav",
        ".mp3": "audio/mpeg",
        ".m4a": "audio/mp4",
        ".ogg": "audio/ogg",
    }.get(fp.suffix.lower(), "application/octet-stream")
    return FileResponse(path=str(fp), media_type=media_type, filename=fp.name)


@router.post("/voice-assets", response_model=VoiceAssetItem)
def upload_voice_asset(
    file: Annotated[UploadFile, File(...)],
    _teacher: Annotated[User, Depends(require_role(UserRole.teacher))],
):
    safe = _safe_voice_name(file.filename or "")
    data = file.file.read()
    if not data:
        raise HTTPException(status_code=400, detail="上传文件为空")
    target = _voice_assets_dir() / safe
    target.write_bytes(data)
    return VoiceAssetItem(name=target.name, size=target.stat().st_size, url=f"{settings.api_v1_prefix}/digital-human/voice-assets/{target.name}")


@router.delete("/voice-assets/{name}", response_model=dict[str, bool])
def delete_voice_asset(
    name: str,
    _teacher: Annotated[User, Depends(require_role(UserRole.teacher))],
):
    safe = _safe_voice_name(name)
    fp = _voice_assets_dir() / safe
    if not fp.exists():
        raise HTTPException(status_code=404, detail="语音文件不存在")
    fp.unlink()
    return {"ok": True}


@router.post("/ask", response_model=AskResponse)
def ask(payload: AskRequest, session: Session = Depends(get_session), current_user: User = Depends(get_current_user)):
    if payload.session_id:
        chat_session = session.get(ChatSession, payload.session_id)
        if not chat_session or chat_session.user_id != current_user.id:
            chat_session = None
    else:
        chat_session = None

    if not chat_session:
        chat_session = ChatSession(user_id=current_user.id, topic=payload.knowledge_code)
        session.add(chat_session)
        session.commit()
        session.refresh(chat_session)

    session.add(ChatMessage(session_id=chat_session.id, sender="user", content=payload.question))

    examples_candidate = _build_examples_answer(payload.question, payload.knowledge_code, session)
    if examples_candidate:
        answer, sources = examples_candidate
    else:
        sources = []
        answer = _build_default_answer()
        if _is_deepseek_selected(payload.model_name, payload.model_provider):
            candidate = _generate_answer_with_deepseek(payload.question, payload.knowledge_code, payload.model_id)
            answer = candidate if candidate else answer

    session.add(ChatMessage(session_id=chat_session.id, sender="assistant", content=answer))
    session.commit()

    audio_base64, audio_mime, tts_status = _synthesize_audio_base64(answer)
    return AskResponse(
        session_id=UUID(str(chat_session.id)),
        answer=answer,
        sources=sources,
        tts_status=tts_status,
        audio_base64=audio_base64,
        audio_mime=audio_mime,
    )