From 7018f5780a030c8eb7ca2ba4ee362a9c885bf330 Mon Sep 17 00:00:00 2001 From: Peanut Date: Sun, 19 Jul 2026 12:15:50 +0800 Subject: [PATCH] =?UTF-8?q?feat(server):=20=E7=9F=AD=E7=AF=87=E5=B0=8F?= =?UTF-8?q?=E8=AF=B4=E6=9C=8D=E5=8A=A1=20SSE=20=E4=BB=A3=E7=90=86=E5=AE=9E?= =?UTF-8?q?=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../emotion/service/ShortNovelService.java | 30 +++ .../service/impl/ShortNovelServiceImpl.java | 183 ++++++++++++++++++ 2 files changed, 213 insertions(+) create mode 100644 server/src/main/java/com/emotion/service/ShortNovelService.java create mode 100644 server/src/main/java/com/emotion/service/impl/ShortNovelServiceImpl.java diff --git a/server/src/main/java/com/emotion/service/ShortNovelService.java b/server/src/main/java/com/emotion/service/ShortNovelService.java new file mode 100644 index 0000000..2913822 --- /dev/null +++ b/server/src/main/java/com/emotion/service/ShortNovelService.java @@ -0,0 +1,30 @@ +package com.emotion.service; + +import com.emotion.dto.request.ShortNovelFollowupRequest; +import com.emotion.dto.request.ShortNovelStreamRequest; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +/** + * 短篇小说外部服务代理接口 + * + * @author huazhongmin + * @date 2026-07-19 + */ +public interface ShortNovelService { + + /** + * 首次发起短篇小说生成请求 + * + * @param request 包含用户心愿文本 + * @return SSE 发射器 + */ + SseEmitter stream(ShortNovelStreamRequest request); + + /** + * 处理后续轮次（澄清回答、大纲确认/修改、重试） + * + * @param request 包含 sessionId、action、payload + * @return SSE 发射器 + */ + SseEmitter followup(ShortNovelFollowupRequest request); +} diff --git a/server/src/main/java/com/emotion/service/impl/ShortNovelServiceImpl.java b/server/src/main/java/com/emotion/service/impl/ShortNovelServiceImpl.java new file mode 100644 index 0000000..117ccb6 --- /dev/null +++ b/server/src/main/java/com/emotion/service/impl/ShortNovelServiceImpl.java @@ -0,0 +1,183 @@ +package com.emotion.service.impl; + +import com.alibaba.fastjson2.JSON; +import com.alibaba.fastjson2.JSONObject; +import com.emotion.config.ShortNovelConfig; +import com.emotion.dto.request.ShortNovelFollowupRequest; +import com.emotion.dto.request.ShortNovelStreamRequest; +import com.emotion.exception.BusinessException; +import com.emotion.service.ShortNovelService; +import com.emotion.util.UserContextHolder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.http.HttpEntity; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.stereotype.Service; +import org.springframework.web.client.RestTemplate; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.BufferedReader; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +/** + * 短篇小说外部服务代理实现 + * 使用 RestTemplate 读取外部 SSE 流并通过 SseEmitter 转发给小程序前端 + * + * @author huazhongmin + * @date 2026-07-19 + */ +@Service +public class ShortNovelServiceImpl implements ShortNovelService { + + private static final Logger log = LoggerFactory.getLogger(ShortNovelServiceImpl.class); + private static final ExecutorService EXECUTOR = Executors.newCachedThreadPool(); + + @Autowired + private ShortNovelConfig config; + + @Autowired + private EpicScriptDialogueServiceImpl epicScriptDialogueServiceImpl; + + private final RestTemplate restTemplate = new RestTemplate(); + + @Override + public SseEmitter stream(ShortNovelStreamRequest request) { + String currentUserId = UserContextHolder.getCurrentUserId(); + if (currentUserId == null) { + throw new BusinessException("用户未登录"); + } + + Map upstreamBody = new HashMap<>(); + upstreamBody.put("user_id", currentUserId); + upstreamBody.put("message_id", "web_" + System.currentTimeMillis()); + upstreamBody.put("query", request.getQuery()); + + return forwardSse("/api/novels/daily/conversation/stream", upstreamBody, currentUserId, request.getQuery()); + } + + @Override + public SseEmitter followup(ShortNovelFollowupRequest request) { + String currentUserId = UserContextHolder.getCurrentUserId(); + if (currentUserId == null) { + throw new BusinessException("用户未登录"); + } + + Map upstreamBody = new HashMap<>(); + upstreamBody.put("session_id", request.getSessionId()); + upstreamBody.put("user_id", currentUserId); + upstreamBody.put("message_id", "web_" + System.currentTimeMillis()); + upstreamBody.put("action", request.getAction()); + upstreamBody.put("payload", request.getPayload()); + + return forwardSse("/api/novels/daily/conversation/stream", upstreamBody, currentUserId, null); + } + + /** + * 通用 SSE 转发逻辑 + * + * @param path 外部服务路径 + * @param body 请求体 + * @param currentUserId 当前用户ID（从主线程捕获，避免 ThreadLocal 在异步线程中失效） + * @param originalQuery 原始用户查询（首次请求时传入，用于保存） + */ + private SseEmitter forwardSse(String path, Map body, String currentUserId, String originalQuery) { + SseEmitter emitter = new SseEmitter(config.getReadTimeout().longValue()); + + EXECUTOR.execute(() -> { + HttpHeaders headers = new HttpHeaders(); + headers.setContentType(MediaType.APPLICATION_JSON); + headers.set("X-API-Token", config.getApiToken()); + headers.set("Accept", "text/event-stream"); + + String url = config.getApiBaseUrl() + path; + + try { + ResponseEntity response = restTemplate.exchange( + url, HttpMethod.POST, + new HttpEntity<>(JSON.toJSONString(body), headers), + org.springframework.core.io.Resource.class); + + InputStream inputStream = response.getBody().getInputStream(); + BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream, StandardCharsets.UTF_8)); + + String line; + StringBuilder dataBuffer = new StringBuilder(); + + while ((line = reader.readLine()) != null) { + if (line.startsWith("data:")) { + dataBuffer.append(line.substring(5).trim()); + } else if (line.isEmpty() && dataBuffer.length() > 0) { + String dataStr = dataBuffer.toString(); + dataBuffer.setLength(0); + + if ("[DONE]".equals(dataStr)) { + continue; + } + + try { + JSONObject event = JSON.parseObject(dataStr); + String type = event.getString("type"); + + // 拦截 novel_done 事件，保存到数据库 + if ("novel_done".equals(type)) { + JSONObject payload = event.getJSONObject("payload"); + if (payload != null && originalQuery != null) { + String fullText = payload.getString("full_text"); + if (fullText != null) { + Map metadata = new HashMap<>(); + if (payload.get("title") != null) { + metadata.put("title", payload.get("title")); + } + Map saveResult = epicScriptDialogueServiceImpl.saveNovelResult( + currentUserId, + originalQuery, + fullText, + metadata); + + // 注入 scriptId 到事件中 + payload.put("scriptId", saveResult.get("scriptId")); + payload.put("conversationId", saveResult.get("conversationId")); + payload.put("currentVersionMessageId", saveResult.get("currentVersionMessageId")); + } + } + } + + // 转发事件给前端 + emitter.send(SseEmitter.event().name(type).data(event.toJSONString())); + } catch (Exception parseEx) { + log.warn("SSE 事件解析失败: {}", parseEx.getMessage()); + } + } + } + + emitter.complete(); + } catch (Exception e) { + log.error("SSE 代理异常: {}", e.getMessage(), e); + try { + JSONObject errorEvent = new JSONObject(); + errorEvent.put("type", "error"); + JSONObject errorPayload = new JSONObject(); + errorPayload.put("code", "UPSTREAM_ERROR"); + errorPayload.put("message", e.getMessage() != null ? e.getMessage() : "外部服务异常"); + errorEvent.put("payload", errorPayload); + emitter.send(SseEmitter.event().name("error").data(errorEvent.toJSONString())); + } catch (Exception ignored) { + // 忽略发送错误事件时的异常 + } + emitter.completeWithError(e); + } + }); + + return emitter; + } +}