package org.dromara.talk.service.impl;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;

import cn.hutool.json.JSONObject;
import okhttp3.*;
import org.dromara.talk.config.DifyConfig;
import org.dromara.talk.domain.vo.TalkAgentVo;
import org.dromara.talk.service.IDifyService;
import org.springframework.stereotype.Service;

import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Service
@RequiredArgsConstructor
public class DifyServiceImpl implements IDifyService {
    private final DifyConfig difyConfig;
    private static final String SENTENCE_DELIMITERS = "，,。.!！？?";
    private final OkHttpClient httpClient = new OkHttpClient.Builder()
        .connectTimeout(30, java.util.concurrent.TimeUnit.SECONDS)
        .readTimeout(60, java.util.concurrent.TimeUnit.SECONDS)
        .writeTimeout(30, java.util.concurrent.TimeUnit.SECONDS)
        .build();

    @Override
    public void callWorkflowStream(String traceId, String userMessage, String agentGender,
                                   List<Map<String, String>> ttsVcnList,
                                   TalkAgentVo agentConfig,
                                   Long userId, String conversationId, String customerPhone, String source,
                                   Consumer<String> onTextChunk,
                                   IDifyService.SentenceCallback onSentence) {
        long workflowStartNs = System.nanoTime();
        try {
            log.info("traceId={} currentTime={} Dify开始请求 userId={}, conversationId={}, messageLength={}",
                traceId, currentTime(), userId, conversationId, userMessage != null ? userMessage.length() : 0);

            // 构建请求体
            Map<String, Object> inputs = new HashMap<>();
            // 将 gender 数字转换为中文：0=男，1=女
            String genderChinese = "0".equals(agentGender) ? "男" : "1".equals(agentGender) ? "女" : agentGender;
            inputs.put("agentGender", genderChinese);
            inputs.put("ttsVcnList", ttsVcnList);
            inputs.put("currentVcn", agentConfig != null ? agentConfig.getTtsVcn() : null);
            inputs.put("user_id", userId != null ? String.valueOf(userId) : "");
            if (customerPhone != null && !customerPhone.isEmpty()) {
                inputs.put("customerPhone", customerPhone);
            }
            if (source != null && !source.isEmpty()) {
                inputs.put("source", source);
            }

            JSONObject requestBody = new JSONObject();
            requestBody.set("inputs", inputs);
            requestBody.set("query", userMessage);
            requestBody.set("response_mode", "streaming");
            requestBody.set("user", "user-" + userId);
            requestBody.set("send_at", System.currentTimeMillis());

            if (conversationId != null && !conversationId.isEmpty()) {
                requestBody.set("conversation_id", conversationId);
            }

            Request request = new Request.Builder()
                .url(difyConfig.getApiUrl() + "/chat-messages")
                .post(RequestBody.create(
                    requestBody.toString(),
                    MediaType.parse("application/json")))
                .addHeader("Authorization", "Bearer " + difyConfig.getApiKey())
                .addHeader("Content-Type", "application/json")
                .build();

            long httpExecuteStartNs = System.nanoTime();
            Response response = httpClient.newCall(request).execute();

            log.info("traceId={} currentTime={} Dify响应头返回 status={}, headersElapsedMs={}",
                traceId, currentTime(), response.code(), elapsedMs(httpExecuteStartNs));

            if (!response.isSuccessful()) {
                String errorBody = response.body() != null ? response.body().string() : "无响应体";
                log.error("traceId={} Dify调用失败 status={}, totalElapsedMs={}, response={}",
                    traceId, response.code(), elapsedMs(workflowStartNs), errorBody);
                throw new RuntimeException("Dify API 调用失败: " + response.code());
            }

            StringBuilder currentSentence = new StringBuilder();
            String newConversationId = null;
            boolean firstEventLogged = false;
            boolean firstMessageLogged = false;
            boolean firstSentenceLogged = false;
            int chunkCount = 0;
            int sentenceCount = 0;

            try (ResponseBody responseBody = response.body()) {
                if (responseBody == null) {
                    throw new RuntimeException("响应体为空");
                }

                java.io.BufferedReader reader = new java.io.BufferedReader(
                    new java.io.InputStreamReader(responseBody.byteStream(), java.nio.charset.StandardCharsets.UTF_8)
                );

                String line;
                while ((line = reader.readLine()) != null) {
                    if (line.startsWith("data: ")) {
                        if (!firstEventLogged) {
                            firstEventLogged = true;
                            log.info("traceId={} currentTime={} Dify首个SSE事件到达 elapsedMs={}",
                                traceId, currentTime(), elapsedMs(workflowStartNs));
                        }
                        String jsonData = line.substring(6);

                        JSONObject event;
                        try {
                            event = new JSONObject(jsonData);
                        } catch (Exception e) {
                            log.warn("解析 SSE 事件失败: {}", line, e);
                            continue;
                        }

                        String eventType = event.getStr("event");

                        if ("message".equals(eventType)) {
                            String answer = event.getStr("answer");
                            if (answer != null) {
                                chunkCount++;
                                if (!firstMessageLogged) {
                                    firstMessageLogged = true;
                                    log.info("traceId={} currentTime={} Dify首个文本分片到达 elapsedMs={}, chunkLength={}",
                                        traceId, currentTime(), elapsedMs(workflowStartNs), answer.length());
                                }
                                currentSentence.append(answer);
                                onTextChunk.accept(answer);

                                String currentText = currentSentence.toString();
                                int splitIndex;
                                while ((splitIndex = findSentenceSplitIndex(currentText)) >= 0) {
                                    String sentenceText = currentText.substring(0, splitIndex + 1).trim();
                                    if (!sentenceText.isEmpty()) {
                                        sentenceCount++;
                                        if (!firstSentenceLogged) {
                                            firstSentenceLogged = true;
                                            log.info("traceId={} currentTime={} Dify首句完成 elapsedMs={}, sentenceLength={}",
                                                traceId, currentTime(), elapsedMs(workflowStartNs), sentenceText.length());
                                        }
                                        onSentence.onSentence(sentenceText, newConversationId, false);
                                    }
                                    currentText = currentText.substring(splitIndex + 1);
                                }
                                currentSentence.setLength(0);
                                currentSentence.append(currentText);
                            }
                        } else if ("message_end".equals(eventType)) {
                            newConversationId = event.getStr("conversation_id");
                            log.info("traceId={} currentTime={} Dify收到message_end elapsedMs={}, conversationId={}, chunkCount={}, sentenceCount={}",
                                traceId, currentTime(), elapsedMs(workflowStartNs), newConversationId, chunkCount, sentenceCount);

                            // 处理最后剩余的文本（如果有）
                            if (currentSentence.length() > 0) {
                                sentenceCount++;
                                onSentence.onSentence(currentSentence.toString().trim(), newConversationId, true);
                            } else {
                                // 没有剩余文本，直接标记完成
                                onSentence.onSentence("", newConversationId, true);
                            }
                        } else if ("error".equals(eventType)) {
                            String errorMessage = event.getStr("message");
                            log.error("Dify API 返回错误: {}", errorMessage);
                            throw new RuntimeException("Dify API 错误: " + errorMessage);
                        }
                    }
                }
            }

            log.info("traceId={} currentTime={} Dify流式处理完成 totalElapsedMs={}, chunkCount={}, sentenceCount={}, finalConversationId={}",
                traceId, currentTime(), elapsedMs(workflowStartNs), chunkCount, sentenceCount, newConversationId);

        } catch (Exception e) {
            log.error("traceId={} 流式调用 Dify 工作流失败 totalElapsedMs={}", traceId, elapsedMs(workflowStartNs), e);
            throw new RuntimeException("AI 服务调用失败", e);
        }
    }

    private int findSentenceSplitIndex(String text) {
        for (int i = 0; i < text.length(); i++) {
            if (SENTENCE_DELIMITERS.indexOf(text.charAt(i)) >= 0) {
                return i;
            }
        }
        return -1;
    }

    private long elapsedMs(long startNs) {
        return java.util.concurrent.TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNs);
    }

    private String currentTime() {
        return java.time.LocalDateTime.now().toString();
    }
}
