ChenHaiAi/chenhai-ai/src/main/java/com/chenhai/chenhaiai/service/AnalysisStreamService.java

// AnalysisStreamService.java
package com.chenhai.chenhaiai.service;

import com.chenhai.chenhaiai.entity.WeekPlanResponse;
import com.chenhai.chenhaiai.utils.TextFormatUtils;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.ai.chat.client.ChatClient;
import org.springframework.stereotype.Service;
import reactor.core.publisher.FluxSink;

import java.util.Map;
import java.util.Optional;

/**
 * 流式分析服务
 */
@Service
public class AnalysisStreamService {

    private final ObjectMapper objectMapper = new ObjectMapper();

    /**
     * 发送流式分析结果
     */
    public void sendStreamAnalysis(
            FluxSink<String> sink,
            String analysisContent,
            ChatClient chatClient,
            String prompt) {

        try {
            // 使用 StringBuilder 收集完整响应
            StringBuilder fullResponse = new StringBuilder();

            chatClient.prompt()
                    .user(prompt)
                    .stream()
                    .content()
                    .subscribe(
                            chunk -> {
                                // 收集所有chunk
                                fullResponse.append(chunk);
                            },
                            error -> {
                                sink.next(TextFormatUtils.formatMessage("error",
                                        "分析失败: " + error.getMessage()));
                                sink.complete();
                            },
                            () -> {
                                try {
                                    // 在流式响应完成后，按模块分割发送
                                    String completeResponse = fullResponse.toString();
                                    sendFormattedModules(sink, completeResponse);
                                } catch (Exception e) {
                                    sink.next(TextFormatUtils.formatMessage("error",
                                            "格式化输出失败: " + e.getMessage()));
                                    sink.complete();
                                }
                            }
                    );

        } catch (Exception e) {
            sink.next(TextFormatUtils.formatMessage("error", "流式分析失败: " + e.getMessage()));
            sink.complete();
        }
    }

    /**
     * 发送格式化模块
     */
    private void sendFormattedModules(FluxSink<String> sink, String analysisContent) {
        // 使用工具类分割模块
        String[] modules = TextFormatUtils.splitAnalysisModules(analysisContent);

        for (String module : modules) {
            String trimmedModule = module.trim();
            if (!trimmedModule.isEmpty()) {
                // 发送完整的模块
                sink.next(TextFormatUtils.formatMessage("content", trimmedModule));

                // 每个模块之间稍微延迟，让前端有时间渲染
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }

        // 发送完成消息
        sink.next(TextFormatUtils.formatMessage("complete", "分析完成"));
        sink.complete();
    }

    /**
     * 准备分析提示词
     */
    public String prepareAnalysisPrompt(String promptTemplate, WeekPlanResponse fullData) {
        try {
            String jsonData = objectMapper.writeValueAsString(fullData);
            return promptTemplate.replace("{jsonData}", jsonData);
        } catch (Exception e) {
            throw new RuntimeException("准备分析提示词失败", e);
        }
    }

    /**
     * 获取数据概览消息
     */
    public String getDataSummary(WeekPlanResponse fullData) {
        int planCount = fullData.getPlanDetails() != null ? fullData.getPlanDetails().size() : 0;
        int dailyCount = fullData.getDailyPapers() != null ? fullData.getDailyPapers().size() : 0;

        return String.format("获取到 %d 个计划任务和 %d 条日报记录", planCount, dailyCount);
    }
}