feat(server): 短篇小说服务 SSE 代理实现
This commit is contained in:
@@ -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);
|
||||
}
|
||||
@@ -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<String, Object> 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<String, Object> 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<String, Object> 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<org.springframework.core.io.Resource> 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<String, Object> metadata = new HashMap<>();
|
||||
if (payload.get("title") != null) {
|
||||
metadata.put("title", payload.get("title"));
|
||||
}
|
||||
Map<String, String> 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;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user