AIyaodianzhushou/backend-java/src/main/java/com/pharmacopoeia/controller/ChatController.java

package com.pharmacopoeia.controller;

import com.pharmacopoeia.config.QwenProperties;
import com.pharmacopoeia.dto.ChatRequest;
import com.pharmacopoeia.dto.FeedbackRequest;
import com.pharmacopoeia.dto.ImageChatRequest;
import com.pharmacopoeia.dto.MultimodalChatRequest;
import com.pharmacopoeia.service.*;
import com.pharmacopoeia.util.IpUtils;
import jakarta.servlet.http.HttpServletRequest;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;

import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.web.multipart.MultipartFile;

import java.util.*;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/api/v1/chat")
@Slf4j
public class ChatController {

    // 复用 PromptService.SECTION_DISPLAY 统一权威映射，避免两处重复定义导致不一致

    private final RetrieverService retrieverService;
    private final LLMService llmService;
    private final PromptService promptService;
    private final ChatPersistenceService persistenceService;
    private final RerankerService rerankerService;
    private final QACacheService qaCache;
    private final BrandRecommendService brandRecommendService;
    private final JdbcTemplate jdbc;
    private final QwenProperties props;
    private final HttpServletRequest request;
    private final AnalyticsService analyticsService;
    private final DrugService drugService;

    public ChatController(RetrieverService rs, LLMService ls, PromptService ps,
                          ChatPersistenceService cps, RerankerService rrs,
                          QACacheService qaCache,
                          BrandRecommendService brandRecommendService,
                          JdbcTemplate jdbc, QwenProperties props,
                          HttpServletRequest request,
                          AnalyticsService analyticsService,
                          DrugService drugService) {
        this.retrieverService = rs;
        this.llmService = ls;
        this.promptService = ps;
        this.persistenceService = cps;
        this.rerankerService = rrs;
        this.qaCache = qaCache;
        this.brandRecommendService = brandRecommendService;
        this.jdbc = jdbc;
        this.props = props;
        this.request = request;
        this.analyticsService = analyticsService;
        this.drugService = drugService;
    }

    @PostMapping("/ask")
    public ResponseEntity<Map<String, Object>> chatAsk(@RequestBody ChatRequest request) {
        final String userKey = getCurrentUserKey();
        String query = request.getMessage();
        String normalized = qaCache.normalize(query);
        String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                ? request.getConversationId()
                : UUID.randomUUID().toString();

        // 检查全局缓存（24 小时有效，不区分用户）
        Map<String, Object> cached = qaCache.get(normalized);
        if (cached != null) {
            String answer = (String) cached.get("answer");
            String intent = (String) cached.getOrDefault("intent", "");
            @SuppressWarnings("unchecked")
            List<Map<String, Object>> sources = (List<Map<String, Object>>) cached.getOrDefault("sources", List.of());
            List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, answer);
            persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
            persistenceService.saveMessage(userKey, cid, "assistant", answer, intent, sources, brandRecs);
            return ResponseEntity.ok(Map.of(
                    "answer", answer,
                    "sources", sources,
                    "brand_recommendations", brandRecs,
                    "intent", intent,
                    "conversation_id", cid,
                    "cached", true
            ));
        }

        // 未命中缓存：尝试抢占处理权，避免并发重复调 LLM
        if (qaCache.tryMarkPending(normalized)) {
            Map<String, Object> waited = qaCache.waitForCache(normalized);
            if (waited != null) {
                String answer = (String) waited.get("answer");
                String intent = (String) waited.getOrDefault("intent", "");
                @SuppressWarnings("unchecked")
                List<Map<String, Object>> sources = (List<Map<String, Object>>) waited.getOrDefault("sources", List.of());
                List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, answer);
                persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
                persistenceService.saveMessage(userKey, cid, "assistant", answer, intent, sources, brandRecs);
                return ResponseEntity.ok(Map.of(
                        "answer", answer, "sources", sources,
                        "brand_recommendations", brandRecs,
                        "intent", intent,
                        "conversation_id", cid, "cached", true
                ));
            }
        }

        String intent = retrieverService.classifyIntent(query);
        List<Map<String, Object>> docs;
        String llmAnswer;
        List<Map<String, Object>> sources;
        try {
            docs = retrieverService.search(query, intent, 20);
            docs = rerankerService.rerank(docs, query, 5);

            // 统一：LLM 回答 + 原文对照
            List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
            llmAnswer = cleanAnswer(llmService.chat(messages));
            sources = buildSources(docs);
        } catch (Exception e) {
            // 异常时清除 PENDING 标记，避免后续同问题请求被锁死
            qaCache.removePending(normalized);
            throw e;
        }
        String answer = llmAnswer;
        List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, answer);
        persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
        persistenceService.saveMessage(userKey, cid, "assistant", answer, intent, sources, brandRecs);

        // 写入全局缓存
        qaCache.put(normalized, answer, intent, sources);

        return ResponseEntity.ok(Map.of(
                "answer", answer,
                "sources", sources,
                "brand_recommendations", brandRecs,
                "intent", intent,
                "conversation_id", cid
        ));
    }

    @PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> chatStream(@RequestBody ChatRequest request, HttpServletRequest servletRequest) {
        final String userKey = getCurrentUserKey();
        String query = request.getMessage();
        String normalized = qaCache.normalize(query);
        final String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                ? request.getConversationId()
                : UUID.randomUUID().toString();

        // 检查全局缓存
        Map<String, Object> cached = qaCache.get(normalized);
        if (cached != null) {
            return streamCached(cached, cid, query, userKey);
        }

        // 未命中缓存：尝试抢占处理权，避免并发重复调 LLM
        if (qaCache.tryMarkPending(normalized)) {
            Map<String, Object> waited = qaCache.waitForCache(normalized);
            if (waited != null) {
                return streamCached(waited, cid, query, userKey);
            }
        }

        final String intent = retrieverService.classifyIntent(query);
        final long t0 = System.currentTimeMillis();
        // request_id：请求体优先（DTO 已带 requestId 字段），header 兜底
        final String chatRequestId = request.getRequestId() != null && !request.getRequestId().isBlank()
                ? request.getRequestId() : servletRequest.getHeader("X-Request-Id");
        final String chatIp = IpUtils.getClientIp(servletRequest);
        final String chatUa = servletRequest.getHeader("User-Agent");
        final java.util.Map<String, Long> timings = new java.util.concurrent.ConcurrentHashMap<>();
        log.info("[chatStream] intent={}, query={}", intent, query.substring(0, Math.min(50, query.length())));

        // L1 知识库直取：药名命中 → 取该药栏目原文 + 一句结论（思考关），跳过全量 RAG 与思考
        if ("drug_query".equals(intent)) {
            Flux<ServerSentEvent<String>> kbFlux = tryKbDirect(query, intent, userKey, cid,
                    chatRequestId, chatIp, chatUa, t0);
            if (kbFlux != null) return kbFlux;
        }

        // L2：后台独立订阅 + Sinks 桥接。前端断开后后台继续把 LLM 跑完并存库，
        // 用户下次进来 /recent-messages 仍能拿到。计时移到 pipeline 的 doFinally（真实完成）。
        final java.util.concurrent.atomic.AtomicBoolean clientGone = new java.util.concurrent.atomic.AtomicBoolean(false);
        final Sinks.Many<ServerSentEvent<String>> sink = Sinks.many().multicast().onBackpressureBuffer();
        sink.tryEmitNext(ServerSentEvent.<String>builder().event("intent").data(intent).build());
        sink.tryEmitNext(ServerSentEvent.<String>builder().event("status").data("Retrieving...").build());

        Mono<Void> pipeline = retrieverService.searchReactive(query, intent, 20)
                .map(docs -> rerankerService.rerank(docs, query, 5))
                .flatMapMany(docs -> {
                    long t2 = System.currentTimeMillis();
                    log.info("[chatStream] search+rerank done, docs={}, elapsed={}ms", docs.size(), t2 - t0);
                    timings.put("search_rerank_ms", t2 - t0);

                    final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
                    final long t3 = System.currentTimeMillis();
                    log.info("[chatStream] prompt built, elapsed={}ms, chars={}",
                            t3 - t0, messages.stream().mapToInt(m -> m.get("content").length()).sum());
                    timings.put("prompt_ms", t3 - t0);

                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("status")
                            .data("Matched " + docs.size() + " records, generating...").build());

                    var fullAnswerBuf = new StringBuilder();
                    return llmService.chatStream(messages)
                            .doOnNext(token -> {
                                fullAnswerBuf.append(token);
                                if (!clientGone.get()) {
                                    sink.tryEmitNext(ServerSentEvent.<String>builder().data(token).build());
                                }
                            })
                            .doOnComplete(() -> {
                                String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
                                final List<Map<String, Object>> sources = buildSources(docs);
                                final List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, finalAnswer);
                                long t4 = System.currentTimeMillis();
                                log.info("[chatStream] chatStream done, llmElapsed={}ms, total={}ms", t4 - t3, t4 - t0);
                                timings.put("llm_ms", t4 - t3);
                                timings.put("total_internal_ms", t4 - t0);
                                persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
                                persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources, brandRecs);
                                qaCache.put(normalized, finalAnswer, intent, sources);
                                if (!clientGone.get()) {
                                    sink.tryEmitNext(buildBrandRecommendEvent(brandRecs));
                                    String meta;
                                    try {
                                        meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
                                                "intent", intent, "sources", sources, "conversation_id", cid
                                        ));
                                    } catch (Exception e) {
                                        meta = "{}";
                                    }
                                    sink.tryEmitNext(ServerSentEvent.<String>builder().event("meta").data(meta).build());
                                    sink.tryEmitComplete();
                                }
                            });
                })
                .then()
                .doOnError(e -> {
                    long t4 = System.currentTimeMillis();
                    log.error("[chatStream] error, totalElapsed={}ms, error={}", t4 - t0, e.getMessage());
                    qaCache.removePending(normalized);
                    if (!clientGone.get()) {
                        sink.tryEmitError(e);
                    }
                })
                .doFinally(sig -> recordChatTiming(userKey, chatRequestId, chatIp, chatUa, "stream", t0, sig, timings));

        // 后台跑完（不绑 SSE 连接），subscribeOn boundedElastic 不占 servlet/Netty 线程
        pipeline.subscribeOn(Schedulers.boundedElastic())
                .subscribe(v -> {}, e -> log.error("[chatStream] 后台 pipeline 出错: cid={}, error={}", cid, e.getMessage()));

        return sink.asFlux().doFinally(sig -> {
            if (sig == SignalType.CANCEL) {
                clientGone.set(true);
                log.info("[chatStream] 客户端断开，后台继续生成并存库: cid={}, request_id={}", cid, chatRequestId);
            }
        });
    }

    /** 将缓存命中结果以流式 SSE 形式返回 */
    private Flux<ServerSentEvent<String>> streamCached(Map<String, Object> cached, String cid,
                                                        String query,
                                                       String userKey) {
        String answer = (String) cached.get("answer");
        String intent = (String) cached.getOrDefault("intent", "");
        @SuppressWarnings("unchecked")
        List<Map<String, Object>> sources = (List<Map<String, Object>>) cached.getOrDefault("sources", List.of());

        persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
        List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, answer);
        persistenceService.saveMessage(userKey, cid, "assistant", answer, intent, sources, brandRecs);

        return Flux.create(sink -> {
            sink.next(ServerSentEvent.<String>builder().event("intent").data(intent).build());
            sink.next(ServerSentEvent.<String>builder().event("status").data("命中缓存，直接返回...").build());
            // 将缓存答案按段落拆分发送，模拟流式体验
            String[] chunks = answer.split("(?<=\\n)");
            for (String chunk : chunks) {
                sink.next(ServerSentEvent.<String>builder().data(chunk).build());
            }
            sink.next(buildBrandRecommendEvent(brandRecs));
            try {
                String meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
                        "intent", intent, "sources", sources, "conversation_id", cid, "cached", true
                ));
                sink.next(ServerSentEvent.<String>builder().event("meta").data(meta).build());
            } catch (Exception ignored) {}
            sink.complete();
        });
    }

    // ============================================================
    // 图片对话 API（Qwen VL 分析 + OCR → RAG 检索 → 联网搜索）
    // ============================================================

    @PostMapping("/ask-image")
    public ResponseEntity<Map<String, Object>> chatAskImage(@RequestBody ImageChatRequest request) {
        final String userKey = getCurrentUserKey();
        String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                ? request.getConversationId()
                : UUID.randomUUID().toString();

        // Step 1: Qwen VL 分析图片 + OCR 提取文字
        String ocrText = llmService.analyzeImage(
                request.getImageBase64(), request.getMimeType(),
                "请分析这张图片，提取其中所有文字信息（OCR），特别是药品名称、成分、用法用量等关键药学信息。简要输出即可。");

        // Step 2: 拼接查询 → RAG 检索
        String query = (!request.getMessage().isBlank())
                ? request.getMessage() + "\n\n（图片OCR提取内容：" + ocrText + "）"
                : ocrText;

        String intent = retrieverService.classifyIntent(query);
        List<Map<String, Object>> docs = retrieverService.search(query, intent, 20);
        docs = rerankerService.rerank(docs, query, 5);

        // Step 3: 构建 Prompt（含图片分析上下文）+ 联网搜索
        List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
        String imageContext = "\n\n【图片分析结果】\n" + ocrText + "\n";
        messages.getFirst().put("content", messages.getFirst().get("content") + imageContext);

        String answer = cleanAnswer(llmService.chat(messages, true));

        List<Map<String, Object>> sources = buildSources(docs);
        List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, answer);
        persistenceService.saveMessage(userKey, cid, "user",
                request.getMessage().isBlank() ? "[图片]" : "[图片] " + request.getMessage(),
                intent, null);
        persistenceService.saveMessage(userKey, cid, "assistant", answer, intent, sources, brandRecs);

        return ResponseEntity.ok(Map.of(
                "answer", answer,
                "sources", sources,
                "brand_recommendations", brandRecs,
                "intent", intent,
                "conversation_id", cid
        ));
    }

    @PostMapping(value = "/stream-image", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> chatStreamImage(@RequestBody ImageChatRequest request) {
        final String userKey = getCurrentUserKey();
        final String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                ? request.getConversationId()
                : UUID.randomUUID().toString();

        // 校验图片数据
        if (request.getImageBase64() == null || request.getImageBase64().isBlank()) {
            return Flux.just(
                    ServerSentEvent.<String>builder().event("status").data("图片数据为空，请重新上传").build()
            );
        }

        // 先发状态，不等图片分析完成
        return Flux.just(
                ServerSentEvent.<String>builder().event("status").data("正在分析图片（OCR 文字识别）...").build()
        ).concatWith(
            llmService.analyzeImageStream(request.getImageBase64(), request.getMimeType(),
                    "请分析这张图片，提取其中所有文字信息（OCR），特别是药品名称、成分、用法用量等。简要输出。")
                    .collectList()
                    .flatMapMany(tokens -> {
                        String ocrText = String.join("", tokens);

                        String query = (!request.getMessage().isBlank())
                                ? request.getMessage() + "\n\n（图片OCR提取内容：" + ocrText + "）"
                                : ocrText;
                        final String intent = retrieverService.classifyIntent(query);

                        return Flux.just(
                                ServerSentEvent.<String>builder().event("status").data("图片分析完成，正在检索药典知识库...").build(),
                                ServerSentEvent.<String>builder().event("intent").data(intent).build()
                        ).concatWith(retrieverService.searchReactive(query, intent, 20)
                                .map(docs -> rerankerService.rerank(docs, query, 5))
                                .flatMapMany(docs -> {
                                    final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
                                    String imageContext = "\n\n【图片分析结果】\n" + ocrText + "\n";
                                    messages.getFirst().put("content", messages.getFirst().get("content") + imageContext);

                                    var fullAnswerBuf = new StringBuilder();

                                    Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages, true)
                                            .map(token -> {
                                                fullAnswerBuf.append(token);
                                                return ServerSentEvent.<String>builder().data(token).build();
                                            });

                                    Flux<ServerSentEvent<String>> tailFlux = Flux.defer(() -> {
                                        String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
                                        final List<Map<String, Object>> sources = buildSources(docs);
                                        final List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, finalAnswer);
                                        String meta;
                                        try {
                                            meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
                                                    "intent", intent,
                                                    "sources", sources,
                                                    "conversation_id", cid,
                                                    "ocr_text", ocrText
                                            ));
                                        } catch (Exception e) {
                                            meta = "{}";
                                        }
                                        persistenceService.saveMessage(userKey, cid, "user",
                                                request.getMessage().isBlank() ? "[图片]" : "[图片] " + request.getMessage(),
                                                intent, null);
                                        persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources, brandRecs);
                                        return Flux.just(
                                                buildBrandRecommendEvent(brandRecs),
                                                ServerSentEvent.<String>builder().event("meta").data(meta).build()
                                        );
                                    });

                                    Flux<ServerSentEvent<String>> headFlux = Flux.just(
                                            ServerSentEvent.<String>builder().event("status")
                                                    .data("已匹配 " + docs.size() + " 条药典资料，生成回答中（已启用联网搜索）...").build()
                                    );

                                    return Flux.concat(headFlux, contentFlux, tailFlux);
                                }));
                    })
        );
    }

    // ============================================================
    // 统一多模态对话 API（文本 + 图片 + 视频）
    // ============================================================

    @PostMapping("/ask-multimodal")
    public ResponseEntity<Map<String, Object>> chatAskMultimodal(@RequestBody MultimodalChatRequest request) {
        final String userKey = getCurrentUserKey();
        String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                ? request.getConversationId()
                : UUID.randomUUID().toString();

        // Step 1: 媒体分析（如果有附件）
        String ocrText = "";
        String mediaLabel = "";
        if (request.getMediaBase64() != null && !request.getMediaBase64().isBlank()
                && request.getMediaType() != null && !request.getMediaType().isBlank()) {
            mediaLabel = "video".equals(request.getMediaType()) ? "视频" : "图片";
            ocrText = llmService.analyzeMedia(
                    request.getMediaBase64(), request.getMediaType(),
                    request.getMediaMime(), "");
        }

        // Step 2: 拼接查询
        String query = request.getMessage() != null ? request.getMessage().trim() : "";
        if (!query.isEmpty() && !ocrText.isEmpty()) {
            query = query + "\n\n（" + mediaLabel + "OCR提取内容：" + ocrText + "）";
        } else if (!ocrText.isEmpty()) {
            query = ocrText;
        } else if (query.isEmpty()) {
            query = "请介绍一下自己";
        }

        // Step 3: RAG 检索
        String intent = retrieverService.classifyIntent(query);
        List<Map<String, Object>> docs = retrieverService.search(query, intent, 20);
        docs = rerankerService.rerank(docs, query, 5);

        // Step 4: 构建 Prompt + 联网搜索
        List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
        if (!ocrText.isEmpty()) {
            messages.getFirst().put("content",
                    messages.getFirst().get("content") + "\n\n【" + mediaLabel + "分析结果】\n" + ocrText + "\n");
        }

        boolean enableSearch = !ocrText.isEmpty() || props.isEnableWebSearch();
        String answer = cleanAnswer(llmService.chat(messages, enableSearch));

        List<Map<String, Object>> sources = buildSources(docs);
        List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, answer);
        String userMsg = !request.getMessage().isBlank() ? request.getMessage()
                : !ocrText.isEmpty() ? "[" + mediaLabel + "]" : request.getMessage();
        persistenceService.saveMessage(userKey, cid, "user", userMsg, intent, null);
        persistenceService.saveMessage(userKey, cid, "assistant", answer, intent, sources, brandRecs);

        return ResponseEntity.ok(Map.of(
                "answer", answer,
                "sources", sources,
                "brand_recommendations", brandRecs,
                "intent", intent,
                "conversation_id", cid
        ));
    }

    @PostMapping(value = "/stream-multimodal", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<String>> chatStreamMultimodal(@RequestBody MultimodalChatRequest request) {
        final String userKey = getCurrentUserKey();
        final String cid = request.getConversationId() != null && !request.getConversationId().isBlank()
                ? request.getConversationId()
                : UUID.randomUUID().toString();
        final boolean hasMedia = request.getMediaBase64() != null && !request.getMediaBase64().isBlank()
                && request.getMediaType() != null && !request.getMediaType().isBlank();
        final String mediaLabel = hasMedia && "video".equals(request.getMediaType()) ? "视频" : "图片";

        if (hasMedia) {
            // 先发 OCR section header
            return Flux.just(
                    ServerSentEvent.<String>builder().event("status")
                            .data("🔍 正在分析" + mediaLabel + "...").build(),
                    ServerSentEvent.<String>builder().data("【📷 " + mediaLabel + "分析】\n\n").build()
            ).concatWith(
                    // 用 collectList 收集流式结果，同时攒 OCR 文本，全程在管道内无阻塞
                    llmService.analyzeMediaStream(request.getMediaBase64(), request.getMediaType(),
                            request.getMediaMime(), "")
                            .flatMap(token -> Flux.just(
                                    ServerSentEvent.<String>builder().data(token).build()  // 前端实时看到
                            ))
                            .concatWith(Flux.just(ServerSentEvent.<String>builder().data("\n\n").build()))
                            // collectList 后再 flatMapMany 接 RAG——Reactor 取消时自动 abort
                            .collectList()
                            .flatMapMany(events -> {
                                // 从已发送的事件中拼回 OCR 文本（不额外调 API）
                                StringBuilder ocrBuilder = new StringBuilder();
                                for (ServerSentEvent<String> e : events) {
                                    String d = e.data();
                                    if (d != null && !"\n\n".equals(d)) {
                                        ocrBuilder.append(d);
                                    }
                                }
                                String mediaOcr = ocrBuilder.toString().trim();
                                // 不重复发送 OCR 事件（已经通过 analyzeMediaStream 实时发送过了）
                                return buildRagPipeline(cid,
                                        request.getMessage(), mediaOcr, userKey, mediaLabel);
                            })
            );
        } else {
            return buildRagPipeline(cid, request.getMessage(), "", userKey, "");
        }
    }

    /** 构建 RAG → 百炼流式管道（纯 Reactor，零裸 subscribe） */
    private Flux<ServerSentEvent<String>> buildRagPipeline(String cid, String rawMsg,
                                                          String ocrText, String userKey,
                                                          String mediaLabel) {
        String rq = rawMsg != null ? rawMsg.trim() : "";
        if (!rq.isEmpty() && ocrText != null && !ocrText.isEmpty()) {
            rq = rq + "\n\n（" + mediaLabel + "OCR提取内容：" + ocrText + "）";
        } else if (ocrText != null && !ocrText.isEmpty()) {
            rq = ocrText;
        } else if (rq.isEmpty()) {
            rq = "请介绍一下自己";
        }
        final String query = rq;
        final String intent = retrieverService.classifyIntent(query);
        final boolean enableSearch = (ocrText != null && !ocrText.isEmpty()) || props.isEnableWebSearch();

        // 如果等待检索，先发状态
        Flux<ServerSentEvent<String>> prefixFlux = (ocrText != null && !ocrText.isEmpty())
                ? Flux.just(ServerSentEvent.<String>builder().event("status")
                        .data("📚 检索药典知识库...").build())
                : Flux.just(ServerSentEvent.<String>builder().event("intent").data(intent).build());

        Flux<ServerSentEvent<String>> intentFlux = (ocrText != null && !ocrText.isEmpty())
                ? Flux.just(ServerSentEvent.<String>builder().event("intent").data(intent).build())
                : Flux.empty();

        return prefixFlux.concatWith(intentFlux)
                .concatWith(retrieverService.searchReactive(query, intent, 20)
                        .map(docs -> rerankerService.rerank(docs, query, 5))
                        .flatMapMany(docs -> {
                            final List<Map<String, String>> messages = promptService.buildPrompt(query, docs, intent);
                            if (ocrText != null && !ocrText.isEmpty()) {
                                messages.getFirst().put("content",
                                        messages.getFirst().get("content") + "\n\n【" + mediaLabel + "分析结果】\n" + ocrText + "\n");
                            }

                            var fullAnswerBuf = new StringBuilder();

                            Flux<ServerSentEvent<String>> contentFlux = llmService.chatStream(messages, enableSearch)
                                    .map(token -> {
                                        fullAnswerBuf.append(token);
                                        return ServerSentEvent.<String>builder().data(token).build();
                                    });

                            Flux<ServerSentEvent<String>> tailFlux = Flux.defer(() -> {
                                final List<Map<String, Object>> sources = buildSources(docs);
                                String finalAnswer = cleanAnswer(fullAnswerBuf.toString());
                                final List<Map<String, Object>> brandRecs = brandRecommendService.match(sources, finalAnswer);
                                String meta;
                                try {
                                    meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
                                            "intent", intent, "sources", sources, "conversation_id", cid,
                                            "ocr_text", ocrText != null ? ocrText : ""));
                                } catch (Exception e) { meta = "{}"; }
                                String userMsg = (rawMsg != null && !rawMsg.isBlank()) ? rawMsg
                                        : (ocrText != null && !ocrText.isEmpty()) ? "[" + mediaLabel + "]" : "";
                                persistenceService.saveMessage(userKey, cid, "user", userMsg, intent, null);
                                persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, sources, brandRecs);
                                return Flux.just(
                                        buildBrandRecommendEvent(brandRecs),
                                        ServerSentEvent.<String>builder().event("meta").data(meta).build()
                                );
                            });

                            Flux<ServerSentEvent<String>> headFlux = Flux.just(
                                    ServerSentEvent.<String>builder().data("\n【📚 药典参考回答】\n\n").build(),
                                    ServerSentEvent.<String>builder().event("status")
                                            .data("已匹配 " + docs.size() + " 条药典资料，生成回答中"
                                                    + (enableSearch ? "（已启用联网搜索）" : "") + "...").build()
                            );

                            return Flux.concat(headFlux, contentFlux, tailFlux);
                        }));
    }

    // ============================================================
    // 文件上传 API（multipart → base64 → 复用已有对话管线）
    // ============================================================

    @PostMapping("/upload-image")
    public ResponseEntity<Map<String, Object>> uploadImage(
            @RequestParam("file") MultipartFile file,
            @RequestParam(defaultValue = "") String message,
            @RequestParam(defaultValue = "") String conversationId) {

        // 校验 MIME 类型
        Set<String> allowed = Set.of("image/jpeg", "image/png", "image/webp", "image/bmp");
        String contentType = file.getContentType();
        if (contentType == null || !allowed.contains(contentType)) {
            throw new IllegalArgumentException(
                    "不支持的图片格式: " + contentType + "，支持 jpg/png/webp/bmp");
        }

        // 校验大小 ≤ 10MB
        if (file.getSize() > 10 * 1024 * 1024) {
            throw new IllegalArgumentException("图片大小不能超过 10MB");
        }

        // 转 base64 → 委托给 ask-image
        String base64;
        try {
            base64 = Base64.getEncoder().encodeToString(file.getBytes());
        } catch (Exception e) {
            throw new RuntimeException("读取上传文件失败", e);
        }

        ImageChatRequest req = new ImageChatRequest();
        req.setImageBase64(base64);
        req.setMimeType(contentType);
        req.setMessage(message);
        req.setConversationId(
                conversationId.isBlank() ? UUID.randomUUID().toString() : conversationId);
        return chatAskImage(req);
    }

    @PostMapping("/upload-media")
    public ResponseEntity<Map<String, Object>> uploadMedia(
            @RequestParam("file") MultipartFile file,
            @RequestParam(defaultValue = "") String message,
            @RequestParam(defaultValue = "") String conversationId) {

        String contentType = file.getContentType();
        if (contentType == null) {
            throw new IllegalArgumentException("无法识别的媒体类型");
        }

        String mediaType;
        long maxSize;
        if (contentType.startsWith("image/")) {
            mediaType = "image";
            maxSize = 10 * 1024 * 1024;  // 10MB
        } else if (contentType.startsWith("video/")) {
            mediaType = "video";
            maxSize = 50 * 1024 * 1024;  // 50MB
        } else {
            throw new IllegalArgumentException(
                    "不支持的媒体格式: " + contentType + "，支持 jpg/png/webp/bmp/mp4/mov/avi/webm");
        }

        if (file.getSize() > maxSize) {
            throw new IllegalArgumentException(
                    "文件大小不能超过 " + (maxSize / 1024 / 1024) + "MB");
        }

        String base64;
        try {
            base64 = Base64.getEncoder().encodeToString(file.getBytes());
        } catch (Exception e) {
            throw new RuntimeException("读取上传文件失败", e);
        }

        MultimodalChatRequest req = new MultimodalChatRequest();
        req.setMessage(message);
        req.setMediaType(mediaType);
        req.setMediaBase64(base64);
        req.setMediaMime(contentType);
        req.setConversationId(
                conversationId.isBlank() ? UUID.randomUUID().toString() : conversationId);
        return chatAskMultimodal(req);
    }

    @GetMapping("/history")
    public ResponseEntity<Map<String, Object>> getHistory(
            @RequestParam(defaultValue = "1") int page,
            @RequestParam(defaultValue = "20") int pageSize) {
        final String userKey = getCurrentUserKey();
        // getHistory 现在直接返回包含 items/page/page_size/total/total_pages 的 Map
        var result = persistenceService.getHistory(userKey, page, pageSize);
        return ResponseEntity.ok(result);
    }

    @GetMapping("/history/{cid}")
    public ResponseEntity<Map<String, Object>> getConversationDetail(@PathVariable String cid) {
        var msgs = persistenceService.getConversationDetail(cid);
        return ResponseEntity.ok(Map.of("conversation_id", cid, "messages", msgs));
    }

    /** 返回最近 N 条消息，供前端恢复对话（微信 WebView 等 IndexedDB 不可用场景） */
    @GetMapping("/recent-messages")
    public ResponseEntity<Map<String, Object>> getRecentMessages(
            @RequestParam(defaultValue = "50") int limit) {
        final String userKey = getCurrentUserKey();
        var msgs = persistenceService.getRecentMessages(userKey, Math.min(limit, 200));
        return ResponseEntity.ok(Map.of("messages", msgs));
    }

    @PostMapping("/feedback")
    public ResponseEntity<Map<String, Object>> submitFeedback(@RequestBody FeedbackRequest request) {
        persistenceService.updateFeedback(request.getMessageId(), request.getFeedback());
        return ResponseEntity.ok(Map.of("status", "ok"));
    }

    @GetMapping("/admin/conversations")
    public ResponseEntity<Map<String, Object>> adminListConversations(
            @RequestParam(defaultValue = "1") int page,
            @RequestParam(defaultValue = "20") int pageSize,
            @RequestParam(required = false) String keyword) {
        int offset = (page - 1) * pageSize;

        StringBuilder sql = new StringBuilder("""
            SELECT DISTINCT ON (c.conversation_id)
                   c.conversation_id, c.title, c.created_at,
                   m.content AS last_msg, m.role
            FROM conversations c
            JOIN messages m ON m.conversation_id = c.conversation_id
            """);

        List<Object> params = new ArrayList<>();

        if (keyword != null && !keyword.isBlank()) {
            sql.append("WHERE m.content ILIKE ? ");
            params.add("%" + keyword + "%");
        }

        sql.append("""
            ORDER BY c.conversation_id, m.created_at DESC
            LIMIT ? OFFSET ?
            """);
        params.add(pageSize);
        params.add(offset);

        List<Map<String, Object>> items = jdbc.queryForList(
                sql.toString(), params.toArray());

        return ResponseEntity.ok(Map.of(
                "items", items,
                "page", page,
                "page_size", pageSize
        ));
    }

    private ServerSentEvent<String> buildBrandRecommendEvent(List<Map<String, Object>> recs) {
        try {
            String data = new com.fasterxml.jackson.databind.ObjectMapper()
                    .writeValueAsString(Map.of("recommendations", recs != null ? recs : List.of()));
            return ServerSentEvent.<String>builder().event("brand_recommend").data(data).build();
        } catch (Exception e) {
            return ServerSentEvent.<String>builder().event("brand_recommend")
                    .data("{\"recommendations\":[]}").build();
        }
    }

    private List<Map<String, Object>> buildSources(List<Map<String, Object>> docs) {
        Set<String> rawSeen = new HashSet<>();
        Set<String> seen = new HashSet<>();
        return docs.stream()
                // 第一层：按原始 name|section 去重，消除同一栏目的多个分块
                .filter(d -> {
                    String key = d.getOrDefault("name", "") + "|" + d.getOrDefault("section", "");
                    return rawSeen.add(key);
                })
                .map(d -> {
                    String content = (String) d.getOrDefault("content", "");
                    String drugName = (String) d.getOrDefault("name", "");
                    String storedSection = (String) d.getOrDefault("section", "");
                    String sourceVersion = (String) d.getOrDefault("source_version", "");
                    String sourceVolume = (String) d.getOrDefault("source_volume", "");
                    String category = (String) d.getOrDefault("category", "");

                    // 优先用 DB 元数据，回退到内容解析
                    if (drugName == null || drugName.isEmpty()) {
                        drugName = extractDrugName(content);
                    }
                    String sectionDisplay = PromptService.SECTION_DISPLAY.getOrDefault(storedSection, storedSection);
                    if (sectionDisplay == null || sectionDisplay.isEmpty()) {
                        sectionDisplay = realSection(content, storedSection);
                    }
                    // 把"正文"替换为更友好的显示名
                    if ("正文".equals(sectionDisplay)) {
                        sectionDisplay = "药典正文";
                    }

                    // 构建完整来源引用
                    String fullSource = getFullSource(sourceVersion, sourceVolume);

                    content = content.replaceAll("\\s*来源：.*$", "");
                    content = content.replaceAll("[\\r\\n]+", " ").trim();
                    String excerpt = content.length() > 500 ? content.substring(0, 500) + "…" : content;
                    return Map.of(
                            "drug_id", d.getOrDefault("drug_id", ""),
                            "name", drugName,
                            "section", sectionDisplay,
                            "category", category != null ? category : "",
                            "source", fullSource,
                            "excerpt", excerpt
                    );
                })
                .filter(Objects::nonNull)
                // 第二层：按显示名去重，避免别名映射（如"功能""主治"→"功能与主治"）导致重复
                .filter(m -> {
                    String key = m.get("name") + "|" + m.get("section");
                    return seen.add(key);
                })
                // 每种药最多 5 个栏目，总数最多 8 条
                .collect(Collectors.groupingBy(m -> (String) m.get("name"), LinkedHashMap::new, Collectors.toList()))
                .values().stream()
                .flatMap(list -> list.stream().limit(5))
                .limit(8)
                .collect(Collectors.toList());
    }

    @NotNull
    private static String getFullSource(String sourceVersion, String sourceVolume) {
        StringBuilder sourceBuilder = new StringBuilder();
        if (!sourceVersion.isEmpty()) {
            sourceBuilder.append(sourceVersion);
        }
        if (!sourceVolume.isEmpty()) {
            if (!sourceBuilder.isEmpty()) {
                sourceBuilder.append(" ");
            }
            sourceBuilder.append(sourceVolume);
        }
        return sourceBuilder.toString();
    }

    private String extractDrugName(String content) {
        if (content == null) {
            return "";
        }
        int start = content.indexOf("【");
        int end = content.indexOf(" - ");
        if (start >= 0 && end > start) {
            return content.substring(start + 1, end);
        }
        return content.length() > 20 ? content.substring(0, 20) : content;
    }

    /** 从 content 文本中提取真实 section（兜底"正文"） */
    private String realSection(String content, String storedSection) {
        if (!"正文".equals(storedSection) || content == null) {
            return storedSection;
        }
        int sep = content.indexOf(" - ");
        if (sep < 0) {
            return storedSection;
        }
        int end = content.indexOf("】", sep);
        if (end > sep) {
            return content.substring(sep + 3, end).trim();
        }
        return storedSection;
    }

    /** 从 SecurityContext 获取当前用户标识（JWT subject），未登录则用 IP 隔离 */
    private String getCurrentUserKey() {
        var auth = SecurityContextHolder.getContext().getAuthentication();
        if (auth != null && auth.isAuthenticated() && !"anonymousUser".equals(auth.getPrincipal())) {
            return auth.getName();
        }
        // 未登录用 IP 隔离，避免不同手机会话串了
        String ip = IpUtils.getClientIp(request);
        return "ip:" + ip;
    }

    private String cleanAnswer(String text) {
        if (text == null) {
            return "";
        }
        // 去除多余空白行（保留单个换行），修复 Qwen 常见格式问题
        return text
                .replace("\r\n", "\n")
                .replaceAll("\\n{3,}", "\n\n")
                .trim();
    }

    /**
     * L1 知识库直取：drug_query 且识别到药名时，取该药栏目原文 + 一句结论（思考关）流式返回。
     * 命中返回 SSE Flux；未命中（识别不到药名/库里无该药/无栏目）返回 null，由调用方走 L2 兜底。
     */
    private Flux<ServerSentEvent<String>> tryKbDirect(String query, String intent, String userKey, String cid,
                                                      String requestId, String ip, String ua, long t0) {
        String drugName = retrieverService.extractDrugName(query);
        if (drugName == null || drugName.isBlank()) return null;
        var drug = drugService.findDrugByName(drugName);
        if (drug == null) return null;
        var sectionsRaw = drug.getSections();
        if (sectionsRaw == null || sectionsRaw.isEmpty()) return null;

        // 归一化栏目：把 sectionsRaw 的 key 经 SECTION_DISPLAY 映射到标准显示名
        java.util.Map<String, String> displayToContent = new java.util.LinkedHashMap<>();
        for (var e : sectionsRaw.entrySet()) {
            if (e.getValue() == null) continue;
            String disp = PromptService.SECTION_DISPLAY.getOrDefault(e.getKey(), e.getKey());
            String text = String.valueOf(e.getValue());
            if (!text.isBlank()) displayToContent.put(disp, text);
        }
        if (displayToContent.isEmpty()) return null;

        // 栏目选择：query 命中某栏目 → 单栏；否则默认集
        java.util.List<String> order = new java.util.ArrayList<>();
        String hit = detectSection(query, displayToContent.keySet());
        if (hit != null) {
            order.add(hit);
        } else {
            for (String d : new String[]{"正文", "性状", "类别", "制剂", "贮藏",
                    "功能与主治", "用法与用量", "不良反应", "禁忌", "注意事项"}) {
                if (displayToContent.containsKey(d)) order.add(d);
            }
        }
        if (order.isEmpty()) return null;

        final String sourceLabel = getFullSource(
                drug.getSourceVersion() == null ? "" : drug.getSourceVersion(),
                drug.getSourceVolume() == null ? "" : drug.getSourceVolume());
        final java.util.List<String> sectionOrder = java.util.Collections.unmodifiableList(order);
        final java.util.Map<String, String> sections = java.util.Collections.unmodifiableMap(displayToContent);

        // 拼 sources + 各栏目原文（供结论 prompt 与最终答案）
        StringBuilder sectionsText = new StringBuilder();
        java.util.List<Map<String, Object>> sources = new java.util.ArrayList<>();
        for (String sec : sectionOrder) {
            String content = sections.get(sec);
            sectionsText.append("【").append(sec).append("】\n").append(content).append("\n");
            Map<String, Object> s = new java.util.LinkedHashMap<>();
            s.put("drug_id", drug.getDrugId() == null ? "" : drug.getDrugId());
            s.put("name", drug.getName() == null ? "" : drug.getName());
            s.put("section", sec);
            s.put("source", sourceLabel);
            s.put("excerpt", content.length() > 400 ? content.substring(0, 400) + "…" : content);
            sources.add(s);
        }

        // 结论 prompt：小 LLM 调用，思考关
        List<Map<String, String>> conclusionMessages = new java.util.ArrayList<>();
        Map<String, String> sysMsg = new java.util.HashMap<>();
        sysMsg.put("role", "system");
        sysMsg.put("content", "你是一名药师。根据给定药典栏目原文，用一句话（不超过80字）概括该药关键信息作为【结论】，只输出结论文本，不得编造栏目外的信息。");
        Map<String, String> userMsg = new java.util.HashMap<>();
        userMsg.put("role", "user");
        userMsg.put("content", "药品：" + drug.getName() + "\n" + sectionsText);
        conclusionMessages.add(sysMsg);
        conclusionMessages.add(userMsg);

        final java.util.List<Map<String, Object>> srcList = java.util.Collections.unmodifiableList(sources);
        final String drugNameResolved = drug.getName();

        return Flux.<ServerSentEvent<String>>create(sink -> {
            sink.next(ServerSentEvent.<String>builder().event("intent").data(intent).build());
            sink.next(ServerSentEvent.<String>builder().event("status").data("已命中药典知识库，直取中...").build());

            // 结论（思考关，~1-2s）
            String conclusion;
            try {
                conclusion = llmService.chat(conclusionMessages, false, false);
            } catch (Exception ex) {
                log.warn("[kbDirect] 结论生成失败，跳过结论: {}", ex.getMessage());
                conclusion = "";
            }

            StringBuilder answer = new StringBuilder();
            if (conclusion != null && !conclusion.isBlank()) {
                answer.append("【结论】\n").append(conclusion.trim()).append("\n\n");
            }
            for (String sec : sectionOrder) {
                answer.append("【").append(sec).append("】\n")
                        .append(sections.get(sec)).append("\n")
                        .append("（来源：").append(sourceLabel).append("）\n\n");
            }
            answer.append("【来源明细】\n").append(sourceLabel);
            String finalAnswer = answer.toString();

            // 按段落流式发送
            String[] chunks = finalAnswer.split("(?<=\\n)");
            for (String chunk : chunks) {
                sink.next(ServerSentEvent.<String>builder().data(chunk).build());
            }

            // 品牌推荐
            List<Map<String, Object>> brandRecs = brandRecommendService.match(srcList, finalAnswer);
            sink.next(buildBrandRecommendEvent(brandRecs));

            // meta
            String meta;
            try {
                meta = new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(Map.of(
                        "intent", intent, "sources", srcList, "conversation_id", cid, "source", "kb_direct"
                ));
            } catch (Exception e) {
                meta = "{}";
            }
            sink.next(ServerSentEvent.<String>builder().event("meta").data(meta).build());

            // 持久化
            persistenceService.saveMessage(userKey, cid, "user", query, intent, null);
            persistenceService.saveMessage(userKey, cid, "assistant", finalAnswer, intent, srcList, brandRecs);
            sink.complete();
        }).doFinally(sig -> recordChatTiming(userKey, requestId, ip, ua, "kb_direct", t0, sig, null));
    }

    /** 扫描 query 中是否出现某栏目别名（SECTION_DISPLAY 的 key），返回其标准显示名（且需在可用栏目中） */
    private String detectSection(String query, java.util.Collection<String> available) {
        if (query == null) return null;
        for (var e : PromptService.SECTION_DISPLAY.entrySet()) {
            if (query.contains(e.getKey()) && available.contains(e.getValue())) {
                return e.getValue();
            }
        }
        return null;
    }

    /** 流式问答结束时上报一次"入参→出参"耗时（含分段），用 request_id 与前端 chat_complete 关联 */
    private void recordChatTiming(String userKey, String requestId, String ip, String ua,
                                  String endpoint, long t0, SignalType sig,
                                  java.util.Map<String, Long> timings) {
        try {
            long total = System.currentTimeMillis() - t0;
            String status;
            if (sig == null) {
                status = "unknown";
            } else {
                status = switch (sig) {
                    case ON_COMPLETE -> "ok";
                    case ON_ERROR -> "error";
                    case CANCEL -> "cancelled";
                    default -> sig.name().toLowerCase();
                };
            }
            java.util.Map<String, Object> ed = new java.util.LinkedHashMap<>();
            ed.put("endpoint", endpoint);
            ed.put("request_id", requestId != null ? requestId : "");
            ed.put("total_ms", total);
            ed.put("status", status);
            if (timings != null && !timings.isEmpty()) {
                ed.putAll(timings);
            }
            analyticsService.saveEvent(userKey, "chat_stream_server", ed, "", "", ip, ua, "AI药典");
        } catch (Exception ignored) {
            // 埋点失败不影响主流程
        }
    }
}