shudao-main/shudao-chat-py/routers/chat.py

from fastapi import APIRouter, Depends, Request
from fastapi.responses import StreamingResponse, JSONResponse
from sqlalchemy.orm import Session
from pydantic import BaseModel
from typing import Optional
from database import get_db, SessionLocal
from models.chat import AIConversation, AIMessage
from models.total import RecommendQuestion
from utils.config import settings
from utils.logger import logger
from services.qwen_service import qwen_service
from services.deepseek_service import deepseek_service
from utils.prompt_loader import load_prompt
from utils.thinking_summary import split_thinking_and_answer, summarize_thinking_content
import time
import json
import httpx
import re

router = APIRouter()


def _build_conversation_preview(content: str, limit: int = 50) -> str:
    content = (content or "").strip()
    if len(content) <= limit:
        return content
    return content[:limit] + "..."


def _to_frontend_timestamp(timestamp: Optional[int]) -> Optional[int]:
    if not timestamp:
        return None
    return timestamp if timestamp >= 10**12 else timestamp * 1000


def _build_conversation_title(conversation: AIConversation) -> str:
    if conversation.business_type == 3 and (conversation.exam_name or "").strip():
        return conversation.exam_name.strip()
    return _build_conversation_preview(conversation.content or "", limit=30)


def _extract_json_object_from_index(source: str, start_idx: int) -> str:
    if start_idx < 0 or start_idx >= len(source) or source[start_idx] != "{":
        return ""

    depth = 0
    in_string = False
    escaped = False

    for idx in range(start_idx, len(source)):
        ch = source[idx]

        if escaped:
            escaped = False
            continue

        if in_string:
            if ch == "\\":
                escaped = True
            elif ch == '"':
                in_string = False
            continue

        if ch == '"':
            in_string = True
            continue

        if ch == "{":
            depth += 1
        elif ch == "}":
            depth -= 1
            if depth == 0:
                return source[start_idx: idx + 1]

    return ""


def _extract_balanced_json_objects(text: str) -> list[str]:
    source = (text or "").strip()
    if not source:
        return []

    objects = []
    seen = set()
    for idx, ch in enumerate(source):
        if ch != "{":
            continue
        candidate = _extract_json_object_from_index(source, idx)
        if candidate and candidate not in seen:
            objects.append(candidate)
            seen.add(candidate)

    return objects


def _extract_trailing_json_candidates(text: str) -> list[str]:
    source = (text or "").strip()
    if not source:
        return []

    candidates = []
    seen = set()
    line_start_indexes = [
        match.start()
        for match in re.finditer(r"(?m)^[ \t]*\{", source)
    ]

    for start_idx in reversed(line_start_indexes):
        candidate = source[start_idx:].strip()
        if candidate and candidate not in seen:
            candidates.append(candidate)
            seen.add(candidate)

    return candidates


def _looks_like_exam_payload(payload: object) -> bool:
    if not isinstance(payload, dict):
        return False

    questions = payload.get("questions")
    return any(
        key in payload
        for key in (
            "singleChoice",
            "single_choice",
            "单选题",
            "judge",
            "判断题",
            "multiple",
            "multiple_choice",
            "multipleChoice",
            "多选题",
            "short",
            "short_answer",
            "shortAnswer",
            "简答题",
        )
    ) or (
        isinstance(questions, dict)
        and any(
            key in questions
            for key in (
                "singleChoice",
                "single_choice",
                "单选题",
                "judge",
                "判断题",
                "multiple",
                "multiple_choice",
                "multipleChoice",
                "多选题",
                "short",
                "short_answer",
                "shortAnswer",
                "简答题",
            )
        )
    )


def _score_exam_payload_candidate(payload: object) -> int:
    if not isinstance(payload, dict):
        return 0

    score = 0
    questions = payload.get("questions") if isinstance(
        payload.get("questions"), dict) else {}

    strong_keys = (
        "singleChoice",
        "single_choice",
        "单选题",
        "judge",
        "判断题",
        "multiple",
        "multiple_choice",
        "multipleChoice",
        "多选题",
        "short",
        "short_answer",
        "shortAnswer",
        "简答题",
    )
    weak_keys = (
        "title",
        "exam_name",
        "examTitle",
        "试卷标题",
        "总分",
        "totalScore",
        "totalQuestions",
    )

    score += sum(10 for key in strong_keys if key in payload)
    score += sum(8 for key in strong_keys if key in questions)
    score += sum(2 for key in weak_keys if key in payload)

    section_candidates = []
    for _, value in payload.items():
        if isinstance(value, dict):
            section_candidates.append(value)
    section_candidates.extend(
        value for value in questions.values() if isinstance(value, dict))

    for section in section_candidates:
        if "questions" in section and isinstance(section.get("questions"), list):
            score += 6
            question_list = section.get("questions") or []
            if question_list and isinstance(question_list[0], dict):
                first_question = question_list[0]
                if any(k in first_question for k in ("text", "question_text", "question", "title", "content", "题干", "题目")):
                    score += 4
                if "options" in first_question:
                    score += 3
                if any(k in first_question for k in ("answer", "answers", "correct_answer", "correct_answers", "答案", "正确答案")):
                    score += 3
                if any(k in first_question for k in ("analysis", "explanation", "解析")):
                    score += 2
        if any(k in section for k in ("count", "question_count", "数量")):
            score += 2
        if any(k in section for k in ("scorePerQuestion", "score_per_question", "每题分值")):
            score += 1

    return score


def _escape_inner_quotes_in_json(text: str) -> str:
    chars = []
    in_string = False
    escaped = False

    for idx, ch in enumerate(text):
        if not in_string:
            chars.append(ch)
            if ch == '"':
                in_string = True
                escaped = False
            continue

        if escaped:
            chars.append(ch)
            escaped = False
            continue

        if ch == "\\":
            chars.append(ch)
            escaped = True
            continue

        if ch == '"':
            next_non_space = ""
            for next_idx in range(idx + 1, len(text)):
                if not text[next_idx].isspace():
                    next_non_space = text[next_idx]
                    break

            if next_non_space in {",", "}", "]", ":"}:
                chars.append(ch)
                in_string = False
            else:
                chars.append('\\"')
            continue

        chars.append(ch)

    return "".join(chars)


def _try_parse_exam_json(candidate: str) -> Optional[dict]:
    text = (candidate or "").strip()
    if not text:
        return None

    text = (
        text.replace("\ufeff", "")
        .replace("```json", "")
        .replace("```JSON", "")
        .replace("```", "")
        .replace("“", '"')
        .replace("”", '"')
    ).strip()

    try:
        parsed = json.loads(text)
    except Exception:
        repaired_text = _escape_inner_quotes_in_json(text)
        repaired_text = re.sub(r",\s*([}\]])", r"\1", repaired_text)
        try:
            parsed = json.loads(repaired_text)
        except Exception:
            return None

    return parsed if _looks_like_exam_payload(parsed) else None


def _sanitize_exam_response(raw_response: str) -> str:
    """考试工坊只向前端/数据库透传可 JSON.parse 的试卷 JSON。"""
    raw_text = (raw_response or "").strip()
    if not raw_text:
        return ""

    _, answer = split_thinking_and_answer(raw_text)
    for candidate in (answer, raw_text):
        parsed = _try_parse_exam_json(candidate)
        if parsed:
            return json.dumps(parsed, ensure_ascii=False)

    parsed_candidates = []
    for candidate in _extract_balanced_json_objects(raw_text):
        parsed = _try_parse_exam_json(candidate)
        if parsed:
            parsed_candidates.append((parsed, candidate))

    for candidate in _extract_trailing_json_candidates(raw_text):
        parsed = _try_parse_exam_json(candidate)
        if parsed:
            parsed_candidates.append((parsed, candidate))

    if parsed_candidates:
        parsed_candidates.sort(
            key=lambda item: (
                _score_exam_payload_candidate(item[0]),
                len(json.dumps(item[0], ensure_ascii=False)),
            ),
            reverse=True,
        )
        best_payload, best_raw_candidate = parsed_candidates[0]
        if _score_exam_payload_candidate(best_payload) > 0:
            return json.dumps(best_payload, ensure_ascii=False)
        logger.warning(
            "[exam] 已提取到JSON对象但试卷特征较弱，选择最大候选兜底: score=%s snippet=%s",
            _score_exam_payload_candidate(best_payload),
            (best_raw_candidate or "")[:200],
        )
        return json.dumps(best_payload, ensure_ascii=False)

    logger.warning("[exam] 未能从模型响应中提取试卷 JSON，保留原始响应供前端兜底解析")
    return raw_text


def _normalize_related_question(question: str) -> str:
    if not isinstance(question, str):
        return ""

    text = question.strip().strip('"').strip("'")
    text = re.sub(r"^[0-9]+[\.\)\]、]\s*", "", text)
    text = re.sub(r"^[-*]\s*", "", text)
    return text.strip()


def _is_placeholder_related_question(question: str) -> bool:
    normalized = _normalize_related_question(question).lower()
    if not normalized:
        return True

    placeholder_patterns = (
        r"^q\s*\d+$",
        r"^question\s*\d+$",
        r"^questions?\s*\d+$",
        r"^问题\s*\d+$",
        r"^相关问题\s*\d+$",
        r"^推荐问题\s*\d+$",
        r"^更多相关问题$",
        r"^更多问题$",
    )

    return any(re.fullmatch(pattern, normalized) for pattern in placeholder_patterns)


def _contains_chinese(text: str) -> bool:
    return any("\u4e00" <= char <= "\u9fff" for char in text or "")


def _is_invalid_related_question(question: str) -> bool:
    normalized = _normalize_related_question(question)
    if (
        not normalized
        or len(normalized) < 4
        or _is_placeholder_related_question(normalized)
        or not _contains_chinese(normalized)
    ):
        return True

    lowered = normalized.lower()
    blocked_keywords = (
        "thinking process",
        "analyze the request",
        "role:",
        "**role",
        "professional question recommendation",
        "infrastructure construction technology",
        "output format",
        "json",
        "prompt",
        "system",
        "assistant",
        "角色定义",
        "任务目标",
        "输入内容",
        "生成要求",
        "输出格式",
        "开始生成",
    )

    return any(keyword in lowered for keyword in blocked_keywords)


def _extract_related_question_topic(content: str) -> str:
    if not content:
        return "当前话题"

    text = re.sub(r"<[^>]+>", " ", str(content))
    text = re.sub(r"\s+", " ", text).strip()
    text = re.sub(
        r"^(好的[！!，, ]*|我理解您提出的问题[，, ]*|这个问题[，, ]*|总的来说[：:，, ]*)+",
        "",
        text,
    )

    pattern = re.search(
        r"(?:主要围绕|围绕|关于|针对|聚焦)([^。！？\n，,；;]{4,32})",
        text,
    )
    if pattern:
        topic = pattern.group(1).strip("“”\"' ：:，,")
        if topic:
            return topic

    sentence = re.split(r"[。！？\n]", text, maxsplit=1)[0].strip("“”\"' ：:，,")
    if sentence:
        return sentence[:24]

    return "当前话题"


def _build_related_question_fallbacks(content: str) -> list[str]:
    topic = _extract_related_question_topic(content)
    return [
        f"{topic}在现场实施时需要重点关注哪些风险点？",
        f"{topic}相关的方案编制、审批和验收要求有哪些？",
        f"针对{topic}，日常检查和监测应抓住哪些关键指标？",
    ]


def _finalize_related_questions(questions: list, content: str, limit: int = 3) -> list[str]:
    cleaned_questions = []
    seen = set()

    for question in questions or []:
        normalized = _normalize_related_question(question)
        lowered = normalized.lower()
        if (
            _is_invalid_related_question(normalized)
            or lowered in seen
        ):
            continue
        cleaned_questions.append(normalized)
        seen.add(lowered)
        if len(cleaned_questions) == limit:
            return cleaned_questions

    for fallback in _build_related_question_fallbacks(content):
        lowered = fallback.lower()
        if lowered in seen:
            continue
        cleaned_questions.append(fallback)
        seen.add(lowered)
        if len(cleaned_questions) == limit:
            break

    return cleaned_questions[:limit]


def _refresh_conversation_snapshot(db: Session, conversation_id: int, user_id: int) -> None:
    latest_message = (
        db.query(AIMessage)
        .filter(
            AIMessage.ai_conversation_id == conversation_id,
            AIMessage.user_id == user_id,
            AIMessage.is_deleted == 0,
        )
        .order_by(AIMessage.id.desc())
        .first()
    )

    if not latest_message:
        db.query(AIConversation).filter(
            AIConversation.id == conversation_id,
            AIConversation.user_id == user_id,
        ).update({"is_deleted": 1, "updated_at": int(time.time())})
        return

    latest_user_message = (
        db.query(AIMessage)
        .filter(
            AIMessage.ai_conversation_id == conversation_id,
            AIMessage.user_id == user_id,
            AIMessage.type == "user",
            AIMessage.is_deleted == 0,
        )
        .order_by(AIMessage.id.desc())
        .first()
    )

    preview_source = (
        latest_user_message.content
        if latest_user_message and latest_user_message.content
        else latest_message.content
    )
    preview_content = _build_conversation_preview(
        preview_source or "", limit=100)

    db.query(AIConversation).filter(
        AIConversation.id == conversation_id,
        AIConversation.user_id == user_id,
    ).update(
        {
            "content": preview_content or " ",
            "updated_at": int(time.time()),
        }
    )


# ─────────────────────────────────────────────────────────────────────────
# 辅助函数
# ─────────────────────────────────────────────────────────────────────────

async def _rag_search(message: str, top_k: int = 5) -> str:
    """调用 search API 做 RAG 检索，返回上下文文本"""
    try:
        search_cfg = getattr(settings, 'search', None)
        if not search_cfg or not hasattr(search_cfg, 'api_url'):
            return ""
        search_url = search_cfg.api_url
        if not search_url:
            return ""
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.post(
                search_url,
                json={"query": message, "n_results": top_k},
            )
            if resp.status_code == 200:
                data = resp.json()
                docs = data.get("results") or data.get("documents") or []
                return "\n\n".join(
                    d.get("content") or d.get("text") or str(d)
                    for d in docs[:top_k]
                    if d.get("content") or d.get("text")
                )
    except Exception as e:
        logger.warning(f"[RAG] 检索失败（可忽略）: {e}")
    return ""


SAFETY_TRAINING_PLAN_SYSTEM_PROMPT = """
你是安全培训需求整理助手。请把用户的自然语言输入整理成安全培训PPT大纲生成任务。

规则：
1. 只输出一个 JSON 对象，不要输出 Markdown、解释或额外文字。
2. 即使用户说“通知”“材料”“文档”，也必须理解为安全培训模块中的 PPT 大纲需求，不要切换到其他文档生成任务。
3. 如果字段缺失，请根据安全培训场景合理补全，但不要编造具体制度编号、人员姓名或不存在的事实。
4. template 字段用于选择大纲模板，默认填“标准安全培训PPT大纲”。
5. content_focus 至少给出 3 个要点。

JSON 字段：
{
  "topic": "培训主题",
  "template": "模板名称",
  "content_focus": ["内容要点1", "内容要点2", "内容要点3"],
  "audience": "参训对象",
  "time": "培训时间",
  "location": "培训地点",
  "goal": "培训目标",
  "notes": "其他要求",
  "normalized_request": "归一化后的安全培训PPT大纲生成需求"
}
"""


def _extract_tag_value(message: str, tag: str) -> str:
    match = re.search(fr"<{tag}>(.*?)</{tag}>", message or "", re.DOTALL)
    return match.group(1).strip() if match else ""


def _strip_document_tags(message: str) -> str:
    text = message or ""
    for tag in ("word", "filename", "filesize"):
        text = re.sub(fr"<{tag}>.*?</{tag}>", " ", text, flags=re.DOTALL)
    return re.sub(r"\s+", " ", text).strip()


def _extract_safety_training_request_payload(message: str) -> dict:
    return {
        "document_content": _extract_tag_value(message, "word"),
        "filename": _extract_tag_value(message, "filename"),
        "filesize": _extract_tag_value(message, "filesize"),
        "request": _strip_document_tags(message),
    }


def _clean_safety_training_topic(message: str) -> str:
    request_text = _extract_safety_training_request_payload(message)["request"]
    first_clause = re.split(r"[，。；;,\n]", request_text, maxsplit=1)[0].strip()
    topic = first_clause or request_text or "安全培训"
    for token in ("请", "帮我", "帮忙", "生成", "制作", "输出", "一份", "一个", "一下", "PPT大纲", "ppt大纲", "大纲", "通知", "文档", "材料"):
        topic = topic.replace(token, "")
    topic = re.sub(r"\s+", "", topic).strip(" ：:，,。；;")
    if not topic:
        topic = "安全培训"
    if "培训" not in topic:
        topic = f"{topic}安全培训"
    return topic


def _parse_json_object(text: str) -> dict:
    if not text:
        return {}
    cleaned = re.sub(r"```(?:json)?\s*", "", str(text)
                     ).replace("```", "").strip()
    match = re.search(r"\{.*\}", cleaned, re.DOTALL)
    if not match:
        return {}
    try:
        parsed = json.loads(match.group(0))
        return parsed if isinstance(parsed, dict) else {}
    except json.JSONDecodeError:
        return {}


def _build_fallback_safety_training_plan(message: str) -> dict:
    topic = _clean_safety_training_topic(message)
    payload = _extract_safety_training_request_payload(message)
    return {
        "topic": topic,
        "template": "标准安全培训PPT大纲",
        "content_focus": ["安全生产责任", "现场风险识别", "安全意识提升", "培训纪律与行为规范"],
        "audience": "参训员工",
        "time": "",
        "location": "",
        "goal": "提升参训人员安全意识和施工现场风险防控能力",
        "notes": payload["request"],
        "normalized_request": f"围绕{topic}生成安全培训PPT大纲",
    }


def _normalize_safety_training_plan(message: str, raw_plan: dict) -> dict:
    plan = _build_fallback_safety_training_plan(message)
    if not isinstance(raw_plan, dict):
        return plan

    for key in ("topic", "template", "audience", "time", "location", "goal", "notes", "normalized_request"):
        value = raw_plan.get(key)
        if isinstance(value, str) and value.strip():
            plan[key] = value.strip()

    focus = raw_plan.get("content_focus")
    if isinstance(focus, list):
        normalized_focus = [str(item).strip()
                            for item in focus if str(item).strip()]
        if normalized_focus:
            plan["content_focus"] = normalized_focus
    elif isinstance(focus, str) and focus.strip():
        plan["content_focus"] = [item.strip()
                                 for item in re.split(r"[、,，;\n]", focus) if item.strip()]

    if "培训" not in plan["topic"]:
        plan["topic"] = f"{plan['topic']}安全培训"
    if "PPT大纲" not in plan["template"]:
        plan["template"] = f"{plan['template']}PPT大纲"
    return plan


def _build_safety_training_generation_message(message: str, plan: dict) -> str:
    payload = _extract_safety_training_request_payload(message)
    focus_text = "、".join(plan.get("content_focus") or [])
    lines = [
        "输出类型：安全培训PPT大纲",
        "请基于以下结构化需求生成安全培训PPT大纲，不要生成通知正文，不要切换到其他文档生成任务。",
        f"主题：{plan.get('topic') or '安全培训'}",
        f"模板：{plan.get('template') or '标准安全培训PPT大纲'}",
        f"内容要点：{focus_text or '安全生产责任、风险识别、应急处置、安全意识提升'}",
        f"参训对象：{plan.get('audience') or '参训员工'}",
        f"培训时间：{plan.get('time') or '未指定'}",
        f"培训地点：{plan.get('location') or '未指定'}",
        f"培训目标：{plan.get('goal') or '提升参训人员安全意识和风险防控能力'}",
        f"其他要求：{plan.get('notes') or '无'}",
        f"归一化需求：{plan.get('normalized_request') or ''}",
        f"原始需求：{payload['request'] or message}",
    ]
    if payload["filename"] or payload["document_content"]:
        lines.extend([
            f"上传文档名称：{payload['filename'] or '未命名文档'}",
            f"上传文档大小：{payload['filesize'] or '未知'}",
            "上传文档内容：",
            payload["document_content"] or "无",
        ])
    return "\n".join(lines)


async def _infer_safety_training_plan(message: str) -> dict:
    payload = _extract_safety_training_request_payload(message)
    planning_input = payload["request"] or message
    if payload["document_content"]:
        planning_input = (
            f"{planning_input}\n\n"
            f"上传文档名称：{payload['filename'] or '未命名文档'}\n"
            f"上传文档内容摘要：{payload['document_content'][:3000]}"
        )

    try:
        response = await qwen_service.chat([
            {"role": "system", "content": SAFETY_TRAINING_PLAN_SYSTEM_PROMPT},
            {"role": "user", "content": planning_input},
        ])
        return _normalize_safety_training_plan(message, _parse_json_object(response))
    except Exception as e:
        logger.warning(
            f"[safety_training] 需求整理失败，使用兜底结构: {type(e).__name__}: {e}")
        return _build_fallback_safety_training_plan(message)


def _clean_ai_writing_response(content: str) -> str:
    text = str(content or "").strip()
    if not text:
        return ""

    text = re.sub(r"```(?:html)?\s*", "", text,
                  flags=re.IGNORECASE).replace("```", "").strip()

    body_match = re.search(
        r"<body[^>]*>(.*?)</body>", text, re.IGNORECASE | re.DOTALL)
    if body_match:
        text = body_match.group(1).strip()

    first_content_tag = re.search(
        r"<(?:article|section|main|div|h[1-6]|p|table|ul|ol)\b",
        text,
        re.IGNORECASE,
    )
    if first_content_tag and text[:first_content_tag.start()].strip():
        text = text[first_content_tag.start():]

    cleanup_patterns = (
        r"<!DOCTYPE[^>]*>",
        r"<html[^>]*>",
        r"</html>",
        r"<head[^>]*>.*?</head>",
        r"<body[^>]*>",
        r"</body>",
        r"<style[^>]*>.*?</style>",
        r"<script[^>]*>.*?</script>",
        r"<meta[^>]*>",
        r"<title[^>]*>.*?</title>",
    )
    for pattern in cleanup_patterns:
        text = re.sub(pattern, "", text, flags=re.IGNORECASE | re.DOTALL)

    return text.strip()


async def _generate_ai_writing_response(message: str) -> str:
    rag_context = await _rag_search(message, top_k=10)
    system_content = load_prompt(
        "document_writing",
        userMessage=message,
        contextJSON=rag_context if rag_context else "暂无相关知识库内容",
    )

    messages = [
        {"role": "system", "content": system_content},
        {
            "role": "user",
            "content": (
                "请根据上面的写作规范和我的原始需求，直接生成可放入富文本编辑器的公文正文 HTML 片段。"
                "不要输出道歉、解释、DOCTYPE、html、head、body、style 或 script 标签。\n\n"
                f"原始需求：\n{message}"
            ),
        },
    ]

    raw_response = await deepseek_service.chat(messages)
    raw_thinking, raw_answer = split_thinking_and_answer(raw_response)
    answer_text = _clean_ai_writing_response(raw_answer or raw_response)
    if raw_thinking:
        thinking_summary = await summarize_thinking_content(
            user_question=message,
            raw_thinking=raw_thinking,
            final_answer=answer_text,
            chat_service=deepseek_service,
            context="document_writing",
        )
        return (
            f"思考过程：\n{thinking_summary}\n\n回答：\n{answer_text}"
            if thinking_summary
            else answer_text
        )
    return answer_text


async def _generate_ppt_outline_response(message: str) -> str:
    training_plan = await _infer_safety_training_plan(message)
    generation_message = _build_safety_training_generation_message(
        message, training_plan)
    rag_context = await _rag_search(generation_message, top_k=10)
    system_content = load_prompt(
        "ppt_outline",
        userMessage=generation_message,
        contextJSON=rag_context if rag_context else "暂无相关知识库内容",
    )

    messages = [
        {"role": "system", "content": system_content},
        {"role": "user", "content": "请直接输出安全培训PPT大纲正文，从标题开始，不要解释提示词或安全规则。"},
    ]

    raw_response = await qwen_service.chat(messages)
    raw_thinking, raw_answer = split_thinking_and_answer(raw_response)
    answer_text = raw_answer or raw_response
    if raw_thinking:
        thinking_summary = await summarize_thinking_content(
            user_question=message,
            raw_thinking=raw_thinking,
            final_answer=answer_text,
            chat_service=qwen_service,
            context="ppt_outline",
        )
        return (
            f"思考过程：\n{thinking_summary}\n\n回答：\n{answer_text}"
            if thinking_summary
            else answer_text
        )
    return answer_text


def _persist_message_pair(db: Session, conv_id: int, user, user_content: str, ai_content: str):
    now_ts = int(time.time())
    user_message = AIMessage(
        ai_conversation_id=conv_id,
        user_id=user.user_id,
        type="user",
        content=user_content,
        created_at=now_ts,
        updated_at=now_ts,
        is_deleted=0,
    )
    db.add(user_message)
    db.commit()
    db.refresh(user_message)

    ai_message = AIMessage(
        ai_conversation_id=conv_id,
        user_id=user.user_id,
        type="ai",
        content=ai_content,
        prev_user_id=user_message.id,
        created_at=now_ts,
        updated_at=now_ts,
        is_deleted=0,
    )
    db.add(ai_message)
    db.commit()
    db.refresh(ai_message)
    return user_message, ai_message


def _build_history_messages(conv_id: int, limit: int = 10) -> list:
    """从数据库读取最近对话历史，构建 messages 列表"""
    db = SessionLocal()
    try:
        msgs = (
            db.query(AIMessage)
            .filter(AIMessage.ai_conversation_id == conv_id, AIMessage.is_deleted == 0)
            .order_by(AIMessage.id.desc())
            .limit(limit)
            .all()
        )
        msgs.reverse()
        result = []
        for m in msgs:
            role = "user" if m.type == "user" else "assistant"
            if m.content:
                result.append({"role": role, "content": m.content})
        return result
    finally:
        db.close()


# ─────────────────────────────────────────────────────────────────────────
# 非流式接口
# ─────────────────────────────────────────────────────────────────────────

class SendMessageRequest(BaseModel):
    message: str
    conversation_id: Optional[int] = None
    ai_conversation_id: Optional[int] = None
    business_type: int = 0  # 0=AI问答, 1=PPT大纲, 2=AI写作, 3=考试工坊
    exam_name: str = ""
    ai_message_id: int = 0


@router.post("/send_deepseek_message")
async def send_deepseek_message(
    request: Request,
    data: SendMessageRequest,
    db: Session = Depends(get_db),
):
    """
    发送消息（非流式）
    支持多种业务类型：
    - 0: AI问答（意图识别 + RAG）
    - 1: PPT大纲生成
    - 2: AI写作
    - 3: 考试工坊
    """
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    try:
        message = data.message.strip()
        if not message:
            return {"statusCode": 400, "msg": "消息不能为空"}

        conversation_id = data.conversation_id or data.ai_conversation_id

        # 创建或获取对话
        if not conversation_id:
            conversation = AIConversation(
                user_id=user.user_id,
                content=message[:100],
                business_type=data.business_type,
                exam_name=data.exam_name if data.business_type == 3 else "",
                created_at=int(time.time()),
                updated_at=int(time.time()),
                is_deleted=0,
            )
            db.add(conversation)
            db.commit()
            db.refresh(conversation)
            conv_id = conversation.id
        else:
            conv_id = conversation_id
            db.query(AIConversation).filter(
                AIConversation.id == conv_id,
                AIConversation.user_id == user.user_id,
                AIConversation.is_deleted == 0,
            ).update({
                "content": message[:100],
                "business_type": data.business_type,
                "exam_name": data.exam_name if data.business_type == 3 else "",
                "updated_at": int(time.time()),
            })
            db.commit()

        response_text = ""
        ai_message_id = 0

        if data.business_type == 0:
            # AI问答：意图识别 + RAG
            try:
                intent_result = await qwen_service.intent_recognition(message)
                intent_type = ""
                if isinstance(intent_result, dict):
                    intent_type = (
                        intent_result.get("intent_type") or intent_result.get(
                            "intent") or ""
                    ).lower()

                rag_context = ""
                if intent_type in ("query_knowledge_base", "知识库查询", "技术咨询"):
                    rag_context = await _rag_search(message, top_k=10)

                # 使用prompt加载器加载最终回答prompt
                system_content = load_prompt(
                    "final_answer",
                    userMessage=message,
                    contextJSON=rag_context if rag_context else "暂无相关知识库内容"
                )

                messages = [
                    {"role": "user", "content": system_content},
                ]

                qwen_response = await qwen_service.chat(messages)
                raw_thinking, raw_answer = split_thinking_and_answer(
                    qwen_response)
                answer_source = raw_answer or qwen_response

                # 兼容模型直接返回 JSON 的场景
                answer_text = answer_source
                try:
                    if isinstance(answer_source, str) and answer_source.strip().startswith("{"):
                        response_json = json.loads(answer_source)
                        answer_text = response_json.get(
                            "natural_language_answer", answer_source
                        )
                except Exception:
                    answer_text = answer_source

                if raw_thinking:
                    thinking_summary = await summarize_thinking_content(
                        user_question=message,
                        raw_thinking=raw_thinking,
                        final_answer=answer_text,
                        chat_service=qwen_service,
                        context="send_message",
                    )
                    response_text = (
                        f"思考过程：\n{thinking_summary}\n\n回答：\n{answer_text}"
                        if thinking_summary
                        else answer_text
                    )
                else:
                    response_text = answer_text
            except Exception as e:
                error_detail = str(e).strip() if str(
                    e).strip() else f"未知错误({type(e).__name__})"
                logger.error(
                    f"[send_deepseek_message] AI问答异常: {type(e).__name__}: {error_detail}")
                response_text = f"处理失败: {error_detail}"

        elif data.business_type == 1:
            # PPT大纲生成
            try:
                response_text = await _generate_ppt_outline_response(message)
                _, ai_message = _persist_message_pair(
                    db=db,
                    conv_id=conv_id,
                    user=user,
                    user_content=message,
                    ai_content=response_text,
                )
                ai_message_id = ai_message.id
                _refresh_conversation_snapshot(db, conv_id, user.user_id)
                db.commit()
                return {
                    "statusCode": 200,
                    "msg": "success",
                    "data": {
                        "conversation_id": conv_id,
                        "ai_conversation_id": conv_id,
                        "response": response_text,
                        "reply": response_text,
                        "content": response_text,
                        "message": response_text,
                        "ai_message_id": ai_message_id,
                        "user_id": user.user_id,
                        "business_type": data.business_type,
                    },
                }
            except Exception as e:
                error_detail = str(e).strip() if str(
                    e).strip() else f"未知错误({type(e).__name__})"
                logger.error(
                    f"[send_deepseek_message] PPT大纲生成异常: {type(e).__name__}: {error_detail}")
                response_text = f"处理失败: {error_detail}"

        elif data.business_type == 2:
            # AI写作
            try:
                response_text = await _generate_ai_writing_response(message)
                _, ai_message = _persist_message_pair(
                    db=db,
                    conv_id=conv_id,
                    user=user,
                    user_content=message,
                    ai_content=response_text,
                )
                ai_message_id = ai_message.id
                _refresh_conversation_snapshot(db, conv_id, user.user_id)
                db.commit()
                return {
                    "statusCode": 200,
                    "msg": "success",
                    "data": {
                        "conversation_id": conv_id,
                        "ai_conversation_id": conv_id,
                        "response": response_text,
                        "reply": response_text,
                        "content": response_text,
                        "message": response_text,
                        "ai_message_id": ai_message_id,
                        "user_id": user.user_id,
                        "business_type": data.business_type,
                    },
                }
            except Exception as e:
                error_detail = str(e).strip() if str(
                    e).strip() else f"未知错误({type(e).__name__})"
                logger.error(
                    f"[send_deepseek_message] AI写作异常: {type(e).__name__}: {error_detail}")
                response_text = f"处理失败: {error_detail}"

        elif data.business_type == 3:
            # 考试工坊：生成题目
            try:
                system_content = (
                    "你是一个专业的考试题目生成助手，专注于路桥隧轨施工安全领域。\n"
                    "请根据用户需求生成专业的考试题目，包括单选题、多选题、判断题、简答题等。\n"
                    "用户消息中已经包含考试标题、题型要求和出题依据内容，必须以其中的出题依据内容为核心生成题目，不能脱离依据内容自由发挥。\n"
                    "题干、选项、答案和解析都要与出题依据内容中的知识点、专业术语、操作流程、规范要求或培训主题直接相关。\n"
                    "输出必须是可直接 JSON.parse 的纯 JSON，不要包含 markdown 代码块、解释文字或额外前后缀。\n"
                    "JSON 顶层结构必须包含 singleChoice、judge、multiple、short 四个字段。\n"
                    "singleChoice.questions 和 multiple.questions 中每道题必须包含 text、options、answer、analysis。\n"
                    "options 必须是数组，元素格式为 {\"key\":\"A\",\"text\":\"具体选项内容\"}，禁止输出“选项A”这类占位文本。\n"
                    "judge.questions 中每道题必须包含 text、answer、analysis。\n"
                    "short.questions 中每道题必须包含 text、outline，其中 outline 至少包含 keyFactors。\n"
                    "所有题目内容、选项内容、答案和解析都要结合用户给出的工程类型、题型数量、分值和课件内容具体生成。"
                )

                messages = [
                    {"role": "system", "content": system_content},
                    {"role": "user", "content": message},
                ]

                raw_response_text = await qwen_service.chat(messages)
                response_text = _sanitize_exam_response(raw_response_text)

                now_ts = int(time.time())
                user_message = AIMessage(
                    ai_conversation_id=conv_id,
                    user_id=user.user_id,
                    type="user",
                    content=message,
                    created_at=now_ts,
                    updated_at=now_ts,
                    is_deleted=0,
                )
                db.add(user_message)
                db.commit()
                db.refresh(user_message)

                ai_message = AIMessage(
                    ai_conversation_id=conv_id,
                    user_id=user.user_id,
                    type="ai",
                    content=response_text,
                    prev_user_id=user_message.id,
                    created_at=now_ts,
                    updated_at=now_ts,
                    is_deleted=0,
                )
                db.add(ai_message)
                db.commit()

                _refresh_conversation_snapshot(db, conv_id, user.user_id)
                db.commit()

                if data.exam_name:
                    db.query(AIConversation).filter(AIConversation.id == conv_id).update(
                        {"exam_name": data.exam_name,
                            "updated_at": int(time.time())}
                    )
                    db.commit()
            except Exception as e:
                error_detail = str(e).strip() if str(
                    e).strip() else f"未知错误({type(e).__name__})"
                logger.error(
                    f"[send_deepseek_message] 考试工坊异常: {type(e).__name__}: {error_detail}")
                response_text = f"处理失败: {error_detail}"

        else:
            return {"statusCode": 400, "msg": f"不支持的业务类型: {data.business_type}"}

        return {
            "statusCode": 200,
            "msg": "success",
            "data": {
                "conversation_id": conv_id,
                "ai_conversation_id": conv_id,
                "response": response_text,
                "reply": response_text,
                "content": response_text,
                "message": response_text,
                "user_id": user.user_id,
                "business_type": data.business_type,
            },
        }
    except Exception as e:
        logger.error(f"[send_deepseek_message] 异常: {e}")
        return {"statusCode": 500, "msg": f"处理失败: {str(e)}"}


@router.get("/get_history_record")
async def get_history_record(
    request: Request,
    ai_conversation_id: int = 0,
    business_type: Optional[int] = None,
    db: Session = Depends(get_db),
):
    """兼容前端的历史记录查询：ai_conversation_id=0 返回对话列表，否则返回消息详情。"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    if ai_conversation_id > 0:
        messages = (
            db.query(AIMessage)
            .filter(
                AIMessage.ai_conversation_id == ai_conversation_id,
                AIMessage.user_id == user.user_id,
                AIMessage.is_deleted == 0,
            )
            .order_by(AIMessage.id.asc())
            .all()
        )

        return {
            "statusCode": 200,
            "msg": "success",
            "total": len(messages),
            "data": [
                {
                    "id": message.id,
                    "ai_conversation_id": message.ai_conversation_id,
                    "user_id": message.user_id,
                    "type": message.type,
                    "content": message.content,
                    "user_feedback": message.user_feedback,
                    "prev_user_id": message.prev_user_id,
                    "search_source": message.search_source or "",
                    "guess_you_want": message.guess_you_want or "",
                    "created_at": _to_frontend_timestamp(message.created_at),
                    "updated_at": _to_frontend_timestamp(message.updated_at),
                }
                for message in messages
            ],
        }

    conversations_query = db.query(AIConversation).filter(
        AIConversation.user_id == user.user_id,
        AIConversation.is_deleted == 0,
    )

    if business_type is not None:
        conversations_query = conversations_query.filter(
            AIConversation.business_type == business_type
        )

    total = conversations_query.count()
    conversations = (
        conversations_query
        .order_by(AIConversation.updated_at.desc(), AIConversation.id.desc())
        .limit(50)
        .all()
    )

    return {
        "statusCode": 200,
        "msg": "success",
        "total": total,
        "data": [
            {
                "id": conv.id,
                "title": _build_conversation_title(conv),
                "content": conv.content or "",
                "business_type": conv.business_type,
                "exam_name": conv.exam_name or "",
                "created_at": _to_frontend_timestamp(conv.created_at),
                "updated_at": _to_frontend_timestamp(conv.updated_at),
            }
            for conv in conversations
        ],
    }


class DeleteConversationRequest(BaseModel):
    ai_conversation_id: int = 0
    ai_message_id: int = 0


@router.post("/delete_conversation")
async def delete_conversation(
    request: Request, data: DeleteConversationRequest, db: Session = Depends(get_db)
):
    """
    删除对话（软删除）
    同时软删除对话记录和所有关联的消息
    """
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    now_ts = int(time.time())

    if data.ai_message_id:
        ai_message = (
            db.query(AIMessage)
            .filter(
                AIMessage.id == data.ai_message_id,
                AIMessage.user_id == user.user_id,
                AIMessage.type == "ai",
                AIMessage.is_deleted == 0,
            )
            .first()
        )
        if not ai_message:
            return {"statusCode": 404, "msg": "消息不存在"}

        db.query(AIMessage).filter(
            AIMessage.id == ai_message.id,
            AIMessage.user_id == user.user_id,
        ).update({"is_deleted": 1, "updated_at": now_ts})

        if ai_message.prev_user_id:
            db.query(AIMessage).filter(
                AIMessage.id == ai_message.prev_user_id,
                AIMessage.user_id == user.user_id,
                AIMessage.ai_conversation_id == ai_message.ai_conversation_id,
            ).update({"is_deleted": 1, "updated_at": now_ts})

        _refresh_conversation_snapshot(
            db, ai_message.ai_conversation_id, user.user_id)
        db.commit()
        return {"statusCode": 200, "msg": "删除成功"}

    if not data.ai_conversation_id:
        return {"statusCode": 400, "msg": "缺少删除参数"}

    db.query(AIConversation).filter(
        AIConversation.id == data.ai_conversation_id,
        AIConversation.user_id == user.user_id,
    ).update({"is_deleted": 1, "updated_at": now_ts})

    db.query(AIMessage).filter(
        AIMessage.ai_conversation_id == data.ai_conversation_id,
        AIMessage.user_id == user.user_id,
    ).update({"is_deleted": 1, "updated_at": now_ts})

    db.commit()
    return {"statusCode": 200, "msg": "删除成功"}


class DeleteHistoryRequest(BaseModel):
    ai_conversation_id: int


@router.post("/delete_history_record")
async def delete_history_record(
    request: Request, data: DeleteHistoryRequest, db: Session = Depends(get_db)
):
    """删除历史记录（软删除）"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}
    db.query(AIConversation).filter(
        AIConversation.id == data.ai_conversation_id,
        AIConversation.user_id == user.user_id,
    ).update({"is_deleted": 1, "updated_at": int(time.time())})
    db.commit()
    return {"statusCode": 200, "msg": "删除成功"}


# ─────────────────────────────────────────────────────────────────────────
# 流式接口 /stream/chat（无 DB，意图识别 + RAG）
# ─────────────────────────────────────────────────────────────────────────

class StreamChatRequest(BaseModel):
    message: str
    model: str = ""


@router.post("/stream/chat")
async def stream_chat(request: Request, data: StreamChatRequest):
    """流式聊天（SSE，不写 DB）"""
    message = data.message.strip()
    if not message:
        return JSONResponse(content={"statusCode": 400, "msg": "消息不能为空"})

    async def event_generator():
        intent_type = ""
        try:
            intent_result = await qwen_service.intent_recognition(message)
            if isinstance(intent_result, dict):
                intent_type = (
                    intent_result.get("intent_type") or intent_result.get(
                        "intent") or ""
                ).lower()
        except Exception as ie:
            logger.warning(f"[stream/chat] 意图识别异常: {ie}")

        rag_context = ""
        if intent_type in ("query_knowledge_base", "知识库查询", "技术咨询"):
            rag_context = await _rag_search(message)

        # 使用prompt加载器加载最终回答prompt
        system_content = load_prompt(
            "final_answer",
            userMessage=message,
            contextJSON=rag_context if rag_context else "暂无相关知识库内容"
        )

        messages = [
            {"role": "user", "content": system_content},
        ]

        try:
            buffer = ""
            pre_answer = ""
            thinking_buf = ""
            in_think = False
            thinking_done = False
            max_input_chars = getattr(
                settings.thinking_summary, "max_input_chars", 1500)

            async for chunk in qwen_service.stream_chat(messages):
                buffer += chunk

                while buffer:
                    lower = buffer.lower()
                    if not thinking_done:
                        if not in_think:
                            start_idx = lower.find("<think>")
                            if start_idx == -1:
                                yield f"data: {json.dumps({'content': buffer}, ensure_ascii=False)}\n\n"
                                buffer = ""
                                break

                            pre_answer += buffer[:start_idx]
                            buffer = buffer[start_idx + len("<think>"):]
                            in_think = True
                            continue

                        end_idx = lower.find("</think>")
                        if end_idx == -1:
                            if max_input_chars and len(thinking_buf) < max_input_chars:
                                thinking_buf += buffer[: max_input_chars -
                                                       len(thinking_buf)]
                            buffer = ""
                            break

                        if max_input_chars and len(thinking_buf) < max_input_chars:
                            thinking_part = buffer[:end_idx]
                            thinking_buf += thinking_part[: max_input_chars - len(
                                thinking_buf)]

                        buffer = buffer[end_idx + len("</think>"):]
                        in_think = False
                        thinking_done = True

                        thinking_summary = await summarize_thinking_content(
                            user_question=message,
                            raw_thinking=thinking_buf,
                            final_answer="",
                            chat_service=qwen_service,
                            context="stream_chat",
                        )
                        if thinking_summary:
                            prefix = f"思考过程：\n{thinking_summary}\n\n回答：\n"
                            yield f"data: {json.dumps({'content': prefix}, ensure_ascii=False)}\n\n"

                        answer_chunk = (pre_answer + buffer).lstrip()
                        if answer_chunk:
                            yield f"data: {json.dumps({'content': answer_chunk}, ensure_ascii=False)}\n\n"

                        pre_answer = ""
                        buffer = ""
                        break

                    yield f"data: {json.dumps({'content': buffer}, ensure_ascii=False)}\n\n"
                    buffer = ""

            # 流结束但未遇到 </think>：仅尝试生成要点（不回退输出 raw thinking）
            if in_think and not thinking_done and thinking_buf:
                thinking_summary = await summarize_thinking_content(
                    user_question=message,
                    raw_thinking=thinking_buf,
                    final_answer="",
                    chat_service=qwen_service,
                    context="stream_chat_eof",
                )
                if thinking_summary:
                    prefix = f"思考过程：\n{thinking_summary}\n\n回答：\n"
                    yield f"data: {json.dumps({'content': prefix}, ensure_ascii=False)}\n\n"
                if pre_answer:
                    yield f"data: {json.dumps({'content': pre_answer}, ensure_ascii=False)}\n\n"
        except Exception as e:
            logger.error(f"[stream/chat] 流式输出异常: {e}")
            yield f"data: {json.dumps({'error': str(e)}, ensure_ascii=False)}\n\n"
        finally:
            yield "data: [DONE]\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")


# ─────────────────────────────────────────────────────────────────────────
# 流式接口 /stream/chat-with-db（前端主聊天接口）
# ─────────────────────────────────────────────────────────────────────────

class StreamChatWithDBRequest(BaseModel):
    message: str
    ai_conversation_id: int = 0
    business_type: int = 0
    exam_name: str = ""
    ai_message_id: int = 0
    online_search_content: str = ""


@router.post("/stream/chat-with-db")
async def stream_chat_with_db(request: Request, data: StreamChatWithDBRequest):
    """
    带 DB 操作的流式聊天（SSE）
    流程：
    1. 创建/获取对话
    2. 插入用户消息和 AI 占位消息
    3. 发送 initial 事件
    4. RAG 检索
    5. 构建历史上下文
    6. 流式输出
    7. 更新 AI 消息内容
    """
    user = request.state.user
    if not user:
        return JSONResponse(content={"statusCode": 401, "msg": "未授权"})

    message = data.message.strip()
    if not message:
        return JSONResponse(content={"statusCode": 400, "msg": "消息不能为空"})

    async def event_generator():
        db = SessionLocal()
        try:
            # 1. 创建或获取对话
            if data.ai_conversation_id == 0:
                conversation = AIConversation(
                    user_id=user.user_id,
                    content=_build_conversation_preview(message, limit=100),
                    business_type=data.business_type,
                    exam_name=data.exam_name,
                    created_at=int(time.time()),
                    updated_at=int(time.time()),
                    is_deleted=0,
                )
                db.add(conversation)
                db.commit()
                db.refresh(conversation)
                conv_id = conversation.id
            else:
                existing_conversation = (
                    db.query(AIConversation)
                    .filter(
                        AIConversation.id == data.ai_conversation_id,
                        AIConversation.user_id == user.user_id,
                        AIConversation.is_deleted == 0,
                    )
                    .first()
                )

                if existing_conversation:
                    conv_id = existing_conversation.id
                    db.query(AIConversation).filter(
                        AIConversation.id == conv_id,
                        AIConversation.user_id == user.user_id,
                    ).update(
                        {
                            "content": _build_conversation_preview(message, limit=100),
                            "business_type": data.business_type,
                            "exam_name": data.exam_name if data.business_type == 3 else "",
                            "updated_at": int(time.time()),
                        }
                    )
                    db.commit()
                else:
                    conversation = AIConversation(
                        user_id=user.user_id,
                        content=_build_conversation_preview(
                            message, limit=100),
                        business_type=data.business_type,
                        exam_name=data.exam_name if data.business_type == 3 else "",
                        created_at=int(time.time()),
                        updated_at=int(time.time()),
                        is_deleted=0,
                    )
                    db.add(conversation)
                    db.commit()
                    db.refresh(conversation)
                    conv_id = conversation.id

            # 2. 插入用户消息
            user_msg = AIMessage(
                ai_conversation_id=conv_id,
                user_id=user.user_id,
                type="user",
                content=message,
                created_at=int(time.time()),
                updated_at=int(time.time()),
                is_deleted=0,
            )
            db.add(user_msg)
            db.commit()
            db.refresh(user_msg)

            # 3. 插入 AI 占位消息
            ai_msg = AIMessage(
                ai_conversation_id=conv_id,
                user_id=user.user_id,
                type="ai",
                content="",
                prev_user_id=user_msg.id,
                created_at=int(time.time()),
                updated_at=int(time.time()),
                is_deleted=0,
            )
            db.add(ai_msg)
            db.commit()
            db.refresh(ai_msg)

            # 4. 发送 initial 事件
            yield f"data: {json.dumps({'type': 'initial', 'ai_conversation_id': conv_id, 'ai_message_id': ai_msg.id}, ensure_ascii=False)}\n\n"

            # 5. RAG search
            rag_context = await _rag_search(message, top_k=10)

            if data.business_type in (1, 2):
                # PPT outline / AI writing: use dedicated prompt
                prompt_name = "ppt_outline" if data.business_type == 1 else "document_writing"
                system_content = load_prompt(
                    prompt_name,
                    userMessage=message,
                    contextJSON=rag_context if rag_context else "?????????"
                )

                messages = [
                    {"role": "user", "content": system_content},
                ]
            else:
                # 6. History context (last 4 items, 2 turns)
                history_msgs = (
                    db.query(AIMessage)
                    .filter(
                        AIMessage.ai_conversation_id == conv_id,
                        AIMessage.id < ai_msg.id,
                        AIMessage.is_deleted == 0,
                    )
                    .order_by(AIMessage.updated_at.desc())
                    .limit(4)
                    .all()
                )
                history_msgs.reverse()

                history_context = ""
                for msg in history_msgs:
                    role = "??" if msg.type == "user" else "??"
                    history_context += f"{role}: {msg.content}\n\n"

                # 7. Build final prompt
                context_parts = []
                if rag_context:
                    context_parts.append(f"??????\n{rag_context}")
                if data.online_search_content:
                    context_parts.append(
                        f"???????\n{data.online_search_content}")

                context_json = "\n\n".join(
                    context_parts) if context_parts else "?????????"

                system_content = load_prompt(
                    "final_answer",
                    userMessage=message,
                    contextJSON=context_json,
                    historyContext=history_context if history_context else ""
                )

                messages = [
                    {"role": "user", "content": system_content},
                ]

            # 8. 流式输出并收集完整回复
            full_response = ""
            try:
                summary_enabled = getattr(
                    settings.thinking_summary, "enabled", True)
                max_input_chars = getattr(
                    settings.thinking_summary, "max_input_chars", 1500)

                buffer = ""
                pre_answer = ""
                thinking_buf = ""
                in_think = False
                thinking_done = False

                async for chunk in qwen_service.stream_chat(messages):
                    if not summary_enabled:
                        escaped_chunk = chunk.replace("\n", "\\n")
                        full_response += chunk
                        yield f"data: {escaped_chunk}\n\n"
                        continue

                    buffer += chunk
                    while buffer:
                        lower = buffer.lower()
                        if not thinking_done:
                            if not in_think:
                                start_idx = lower.find("<think>")
                                if start_idx == -1:
                                    escaped_text = buffer.replace("\n", "\\n")
                                    full_response += buffer
                                    yield f"data: {escaped_text}\n\n"
                                    buffer = ""
                                    break

                                pre_answer += buffer[:start_idx]
                                buffer = buffer[start_idx + len("<think>"):]
                                in_think = True
                                continue

                            end_idx = lower.find("</think>")
                            if end_idx == -1:
                                if max_input_chars and len(thinking_buf) < max_input_chars:
                                    thinking_buf += buffer[: max_input_chars -
                                                           len(thinking_buf)]
                                buffer = ""
                                break

                            if max_input_chars and len(thinking_buf) < max_input_chars:
                                thinking_part = buffer[:end_idx]
                                thinking_buf += thinking_part[: max_input_chars - len(
                                    thinking_buf)]

                            buffer = buffer[end_idx + len("</think>"):]
                            in_think = False
                            thinking_done = True

                            thinking_summary = await summarize_thinking_content(
                                user_question=message,
                                raw_thinking=thinking_buf,
                                final_answer="",
                                chat_service=qwen_service,
                                context="stream_chat_with_db",
                            )
                            if thinking_summary:
                                prefix = f"思考过程：\n{thinking_summary}\n\n回答：\n"
                                full_response += prefix
                                escaped_prefix = prefix.replace('\n', '\\n')
                                yield f"data: {escaped_prefix}\n\n"

                            answer_chunk = (pre_answer + buffer).lstrip()
                            if answer_chunk:
                                full_response += answer_chunk
                                escaped_answer = answer_chunk.replace(
                                    '\n', '\\n')
                                yield f"data: {escaped_answer}\n\n"

                            pre_answer = ""
                            buffer = ""
                            break

                        escaped_text = buffer.replace("\n", "\\n")
                        full_response += buffer
                        yield f"data: {escaped_text}\n\n"
                        buffer = ""
            except Exception as e:
                logger.error(f"[stream/chat-with-db] 流式输出异常: {e}")
                yield f"data: {json.dumps({'error': str(e)}, ensure_ascii=False)}\n\n"

            # 流结束但未遇到 </think>：仅尝试生成要点（不回退输出 raw thinking）
            if summary_enabled and in_think and not thinking_done and thinking_buf:
                thinking_summary = await summarize_thinking_content(
                    user_question=message,
                    raw_thinking=thinking_buf,
                    final_answer="",
                    chat_service=qwen_service,
                    context="stream_chat_with_db_eof",
                )
                if thinking_summary:
                    prefix = f"思考过程：\n{thinking_summary}\n\n回答：\n"
                    full_response += prefix
                    escaped_prefix = prefix.replace('\n', '\\n')
                    yield f"data: {escaped_prefix}\n\n"
                if pre_answer:
                    full_response += pre_answer
                    escaped_pre_answer = pre_answer.replace('\n', '\\n')
                    yield f"data: {escaped_pre_answer}\n\n"

            # 9. 更新 AI 消息内容
            if full_response:
                now_ts = int(time.time())
                db.query(AIMessage).filter(AIMessage.id == ai_msg.id).update(
                    {"content": full_response, "updated_at": now_ts}
                )
                db.query(AIConversation).filter(
                    AIConversation.id == conv_id,
                    AIConversation.user_id == user.user_id,
                ).update(
                    {
                        "content": _build_conversation_preview(message, limit=100),
                        "business_type": data.business_type,
                        "exam_name": data.exam_name if data.business_type == 3 else "",
                        "updated_at": now_ts,
                    }
                )
                db.commit()

            # 10. 结束标记
            yield "data: [DONE]\n\n"

        except Exception as e:
            logger.error(f"[stream/chat-with-db] 处理异常: {e}")
            yield f"data: {json.dumps({'error': str(e)}, ensure_ascii=False)}\n\n"
        finally:
            db.close()

    return StreamingResponse(event_generator(), media_type="text/event-stream")


# ─────────────────────────────────────────────────────────────────────────
# 猜你想问
# ─────────────────────────────────────────────────────────────────────────

class GuessYouWantRequest(BaseModel):
    ai_message_id: int


@router.post("/guess_you_want")
async def guess_you_want(
    request: Request,
    data: GuessYouWantRequest,
    db: Session = Depends(get_db),
):
    """生成"猜你想问"的3个关联问题，保存到 AIMessage.guess_you_want"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    try:
        ai_msg = (
            db.query(AIMessage)
            .filter(AIMessage.id == data.ai_message_id, AIMessage.is_deleted == 0)
            .first()
        )
        if not ai_msg:
            return {"statusCode": 404, "msg": "消息不存在"}

        # 使用prompt加载器加载猜你想问prompt
        system_content = load_prompt(
            "guess_questions",
            currentContent=ai_msg.content[:500]
        )

        messages = [
            {"role": "user", "content": system_content},
        ]

        response = await qwen_service.chat(messages)

        try:
            # 尝试从响应中提取 JSON
            json_match = re.search(
                r'\{[^{}]*"questions"[^{}]*\}', response, re.DOTALL)
            if json_match:
                response_json = json.loads(json_match.group())
            else:
                response_json = json.loads(response)
            questions = response_json.get("questions", [])
        except Exception:
            lines = [l.strip() for l in response.split("\n") if l.strip()]
            questions = []
            for line in lines:
                clean = line.lstrip("0123456789.-、 ").strip()
                if clean and len(clean) > 5:
                    questions.append(clean)
            if not questions:
                questions = ["该话题的具体应用场景？", "有哪些注意事项？", "相关案例分析？"]

        questions = _finalize_related_questions(
            questions, ai_msg.content, limit=3)

        guess_json = json.dumps({"questions": questions}, ensure_ascii=False)

        db.query(AIMessage).filter(AIMessage.id == data.ai_message_id).update(
            {"guess_you_want": guess_json, "updated_at": int(time.time())}
        )
        db.commit()

        return {
            "statusCode": 200,
            "msg": "success",
            "data": {"ai_message_id": data.ai_message_id, "questions": questions},
        }

    except Exception as e:
        logger.error(f"[guess_you_want] 处理异常: {e}")
        return {"statusCode": 500, "msg": f"处理失败: {str(e)}"}


# ─────────────────────────────────────────────────────────────────────────
# 在线搜索（Dify 工作流集成）
# ─────────────────────────────────────────────────────────────────────────

@router.get("/online_search")
async def online_search(question: str, request: Request, db: Session = Depends(get_db)):
    """
    在线搜索
    流程：Qwen 提炼关键词 → Dify 工作流 → 返回摘要
    """
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    try:
        keywords = await qwen_service.extract_keywords(question)

        dify_config = getattr(settings, "dify", None)
        if not dify_config or not getattr(dify_config, "workflow_url", None):
            return {"statusCode": 500, "msg": "Dify 配置未设置"}

        headers = {
            "Authorization": f"Bearer {dify_config.auth_token}",
            "Content-Type": "application/json",
        }
        payload = {
            "workflow_id": dify_config.workflow_id,
            "inputs": {
                "keywords": keywords,
                "num": 5,  # 搜索结果数量
                "max_text_len": 4000  # 最大文本长度
            },
            "response_mode": "blocking",
            "user": getattr(user, "account", str(user.user_id)),
        }

        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(dify_config.workflow_url, headers=headers, json=payload)
            if resp.status_code != 200:
                logger.error(
                    f"[online_search] Dify 调用失败: {resp.status_code}, 响应: {resp.text}")
                return {"statusCode": 500, "msg": f"搜索服务异常: {resp.status_code}"}
            result = resp.json()
            search_text = result.get("data", {}).get(
                "outputs", {}).get("text", "")

        return {
            "statusCode": 200,
            "msg": "success",
            "data": {"keywords": keywords, "result": search_text},
        }

    except Exception as e:
        logger.error(f"[online_search] 处理异常: {e}")
        return {"statusCode": 500, "msg": f"搜索失败: {str(e)}"}


class SaveOnlineSearchResultRequest(BaseModel):
    ai_message_id: int
    search_result: str


@router.post("/save_online_search_result")
async def save_online_search_result(
    request: Request,
    data: SaveOnlineSearchResultRequest,
    db: Session = Depends(get_db),
):
    """保存联网搜索结果到 AIMessage.search_source"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    try:
        db.query(AIMessage).filter(AIMessage.id == data.ai_message_id).update(
            {"search_source": data.search_result,
                "updated_at": int(time.time())}
        )
        db.commit()
        return {"statusCode": 200, "msg": "保存成功"}
    except Exception as e:
        logger.error(f"[save_online_search_result] 处理异常: {e}")
        return {"statusCode": 500, "msg": f"保存失败: {str(e)}"}


# ─────────────────────────────────────────────────────────────────────────
# 意图识别独立接口
# ─────────────────────────────────────────────────────────────────────────

class IntentRecognitionRequest(BaseModel):
    message: str
    save_to_db: bool = False
    ai_conversation_id: int = 0


@router.post("/intent_recognition")
async def intent_recognition(
    request: Request,
    data: IntentRecognitionRequest,
    db: Session = Depends(get_db),
):
    """独立意图识别接口；若为 greeting/faq 且 save_to_db=True 则直接存 DB"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    try:
        intent_result = await qwen_service.intent_recognition(data.message)
        intent_type = ""
        response_text = ""
        if isinstance(intent_result, dict):
            intent_type = (
                intent_result.get("intent_type") or intent_result.get(
                    "intent") or ""
            ).lower()
            response_text = intent_result.get("response", "")

        if data.save_to_db and intent_type in ("greeting", "问候", "faq", "常见问题"):
            if data.ai_conversation_id == 0:
                conversation = AIConversation(
                    user_id=user.user_id,
                    content=data.message[:100],
                    business_type=0,
                    created_at=int(time.time()),
                    updated_at=int(time.time()),
                    is_deleted=0,
                )
                db.add(conversation)
                db.commit()
                db.refresh(conversation)
                conv_id = conversation.id
            else:
                conv_id = data.ai_conversation_id

            user_msg = AIMessage(
                ai_conversation_id=conv_id,
                user_id=user.user_id,
                type="user",
                content=data.message,
                created_at=int(time.time()),
                updated_at=int(time.time()),
                is_deleted=0,
            )
            db.add(user_msg)
            db.commit()

            ai_msg = AIMessage(
                ai_conversation_id=conv_id,
                user_id=user.user_id,
                type="ai",
                content=response_text,
                prev_user_id=user_msg.id,
                created_at=int(time.time()),
                updated_at=int(time.time()),
                is_deleted=0,
            )
            db.add(ai_msg)
            db.commit()
            db.refresh(ai_msg)

            return {
                "statusCode": 200,
                "msg": "success",
                "data": {
                    "intent_type": intent_type,
                    "response": response_text,
                    "ai_conversation_id": conv_id,
                    "ai_message_id": ai_msg.id,
                    "saved_to_db": True,
                },
            }

        return {
            "statusCode": 200,
            "msg": "success",
            "data": {
                "intent_type": intent_type,
                "response": response_text,
                "saved_to_db": False,
            },
        }

    except Exception as e:
        logger.error(f"[intent_recognition] 处理异常: {e}")
        return {"statusCode": 500, "msg": f"处理失败: {str(e)}"}


# ─────────────────────────────────────────────────────────────────────────
# 获取用户推荐问题（模糊查询 QA / RecommendQuestion 表）
# ─────────────────────────────────────────────────────────────────────────

@router.get("/get_user_recommend_question")
async def get_user_recommend_question(
    keyword: str = "",
    limit: int = 10,
    db: Session = Depends(get_db),
):
    """获取推荐问题（支持模糊查询）"""
    try:
        query = db.query(RecommendQuestion).filter(
            RecommendQuestion.is_deleted == 0)
        if keyword:
            query = query.filter(
                RecommendQuestion.question.like(f"%{keyword}%"))
        questions = query.order_by(
            RecommendQuestion.id.desc()).limit(limit).all()

        return {
            "statusCode": 200,
            "msg": "success",
            "data": [
                {"id": q.id, "question": q.question, "created_at": q.created_at}
                for q in questions
            ],
        }

    except Exception as e:
        logger.error(f"[get_user_recommend_question] 处理异常: {e}")
        return {"statusCode": 500, "msg": f"查询失败: {str(e)}"}


# ─────────────────────────────────────────────────────────────────────────
# PPT 大纲 / 文档编辑保存
# ─────────────────────────────────────────────────────────────────────────

class SavePPTOutlineRequest(BaseModel):
    ai_message_id: int
    content: str


@router.post("/save_ppt_outline")
async def save_ppt_outline(
    request: Request,
    data: SavePPTOutlineRequest,
    db: Session = Depends(get_db),
):
    """更新 AIMessage.content 保存 PPT 大纲内容"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    try:
        db.query(AIMessage).filter(AIMessage.id == data.ai_message_id).update(
            {"content": data.content, "updated_at": int(time.time())}
        )
        db.commit()
        return {"statusCode": 200, "msg": "保存成功"}
    except Exception as e:
        logger.error(f"[save_ppt_outline] 处理异常: {e}")
        return {"statusCode": 500, "msg": f"保存失败: {str(e)}"}


class SaveEditDocumentRequest(BaseModel):
    ai_message_id: int
    content: str


@router.post("/save_edit_document")
async def save_edit_document(
    request: Request,
    data: SaveEditDocumentRequest,
    db: Session = Depends(get_db),
):
    """更新 ai 类型 AIMessage.content（AI写作编辑保存）"""
    user = request.state.user
    if not user:
        return {"statusCode": 401, "msg": "未授权"}

    try:
        db.query(AIMessage).filter(
            AIMessage.id == data.ai_message_id,
            AIMessage.type == "ai",
        ).update({"content": data.content, "updated_at": int(time.time())})
        db.commit()
        return {"statusCode": 200, "msg": "保存成功"}
    except Exception as e:
        logger.error(f"[save_edit_document] 处理异常: {e}")
        return {"statusCode": 500, "msg": f"保存失败: {str(e)}"}