package com.njuzr.eaibackend.service.impl; import cn.xfyun.api.IseClient; import cn.xfyun.model.response.ise.IseResponseData; import cn.xfyun.service.ise.AbstractIseWebSocketListener; import com.alibaba.fastjson2.JSONObject; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; import com.njuzr.eaibackend.config.DoubaoConfig; import com.njuzr.eaibackend.config.IseClientFactory; import com.njuzr.eaibackend.exception.MyException; import com.njuzr.eaibackend.po.SpeakingAIDialogue; import com.njuzr.eaibackend.service.MediaAnalysisService; import com.njuzr.eaibackend.utils.XmlParasUtil; import com.njuzr.eaibackend.vo.VideoAnalysisVo; import com.volcengine.ark.runtime.model.completion.chat.ChatCompletionRequest; import com.volcengine.ark.runtime.model.completion.chat.ChatMessage; import com.volcengine.ark.runtime.model.completion.chat.ChatMessageRole; import lombok.extern.slf4j.Slf4j; import okhttp3.*; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.mongodb.core.query.Update; import org.springframework.http.HttpStatus; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; import org.springframework.stereotype.Service; import java.io.IOException; import java.io.InputStream; import java.net.URL; import java.net.URLConnection; import java.security.SignatureException; import java.util.*; import java.util.concurrent.*; /** * @author Liululin * @date 2025/3/13 - 15:04 */ @Slf4j @Service public class MediaAnalysisServiceImpl implements MediaAnalysisService { private static final String VIDEO_URL = "http://8.130.28.19:2002/deepface"; private static final String AUDIO_URL = "http://8.130.28.19:2003/asr"; private static final HashSet emotionSet = new HashSet<>(); private static final Base64.Decoder decoder = Base64.getDecoder(); private static final long DEFAULT_TIMEOUT_SECONDS = 30; private static final long VIDEO_ANALYSIS_TIMEOUT_SECONDS = 600; private static final String AUDIO_SCORE_HEADER = "[audio-score]"; private static final String AUDIO_SENTIMENT_HEADER = "[audio-sentiment]"; private static final String VIDEO_SENTIMENT_HEADER = "[video-sentiment]"; private static final String SPEAKING_SUGGESTION_HEADER = "[speaking-suggestion]"; static { emotionSet.add("😊"); emotionSet.add("😔"); emotionSet.add("😡"); emotionSet.add("😰"); emotionSet.add("🤢"); emotionSet.add("😮"); } @Autowired DoubaoConfig doubaoConfig; @Autowired MongoTemplate mongoTemplate; @Autowired IseClientFactory iseClientFactory; @Autowired @Qualifier("doubaoTaskExecutor") // 指定 Bean 名称 private ThreadPoolTaskExecutor doubaoTaskExecutor; // 注入线程池 private final Map> pendingSuggestionRequests = new ConcurrentHashMap<>(); private final Map> pendingVideoAnalysisRequests = new ConcurrentHashMap<>(); private final OkHttpClient httpClient = new OkHttpClient.Builder() .connectTimeout(20, TimeUnit.SECONDS) .readTimeout(600, TimeUnit.SECONDS) .writeTimeout(300, TimeUnit.SECONDS) .build(); @Override public CompletableFuture getAudioScore(Long userId, Long assignmentId, Integer index) { log.info("{} 开始获取音频评分, userId: {}, assignmentId: {}, index: {}", AUDIO_SCORE_HEADER, userId, assignmentId, index); Query query = new Query(Criteria.where("userId").is(userId).and("assignmentId").is(assignmentId)); SpeakingAIDialogue dialogue = mongoTemplate.findOne(query, SpeakingAIDialogue.class); if (dialogue == null) { log.error("{} 未找到对应的对话记录, userId: {}, assignmentId: {}", AUDIO_SCORE_HEADER, userId, assignmentId); throw new MyException(HttpStatus.BAD_REQUEST.value(), "不存在相应记录"); } if (index == -1) { index = Math.max(dialogue.getDialogues().size() - 2, 0); } SpeakingAIDialogue.DialogueEntry entry = dialogue.getDialogues().get(index); Integer finalIndex = index; return CompletableFuture.supplyAsync(() -> { try { log.debug("{} 开始处理音频评分, 音频URL: {}", AUDIO_SCORE_HEADER, entry.getAudioUrl()); URL url = new URL(entry.getAudioUrl()); URLConnection connection = url.openConnection(); InputStream inputStream = connection.getInputStream(); IseClient client = iseClientFactory.getIseClient(entry.getContent()); CompletableFuture future = new CompletableFuture<>(); client.send(inputStream, new AbstractIseWebSocketListener() { @Override public void onSuccess(WebSocket webSocket, IseResponseData iseResponseData) { try { String decodedData = new String(decoder.decode(iseResponseData.getData().getData())); XmlParasUtil.paras(decodedData, entry); updateDialogueEntry(userId, assignmentId, finalIndex, entry); log.info("{} 音频评分处理成功, userId: {}, assignmentId: {}, index: {}", AUDIO_SCORE_HEADER, userId, assignmentId, finalIndex); future.complete(entry); } catch (Exception e) { log.error("{} 音频评分数据处理异常", AUDIO_SCORE_HEADER, e); future.completeExceptionally(e); } finally { client.closeWebsocket(); } } @Override public void onFail(WebSocket webSocket, Throwable throwable, Response response) { try { log.error("{} ISE服务调用失败, 响应: {}, 错误: {}", AUDIO_SCORE_HEADER, response, throwable.getMessage()); future.completeExceptionally(throwable); } finally { client.closeWebsocket(); } } }); return future.get(); } catch (IOException | SignatureException e) { log.error("{} 音频评分处理IO异常", AUDIO_SCORE_HEADER, e); throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "ISEService调用失败:" + e.getMessage()); } catch (InterruptedException | ExecutionException e) { log.error("{} 音频评分处理中断异常", AUDIO_SCORE_HEADER, e); Thread.currentThread().interrupt(); throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "音频评分处理中断"); } }, doubaoTaskExecutor); } @Async("doubaoTaskExecutor") public CompletableFuture getAudioSentimentAsync(Long userId, Long assignmentId, Integer index) { return CompletableFuture.supplyAsync(() -> getAudioSentiment(userId, assignmentId, index), doubaoTaskExecutor); } public String getAudioSentiment(Long userId, Long assignmentId, Integer index) { log.info("{} 开始获取音频情感分析, userId: {}, assignmentId: {}, index: {}", AUDIO_SENTIMENT_HEADER, userId, assignmentId, index); Query query = new Query(Criteria.where("userId").is(userId).and("assignmentId").is(assignmentId)); SpeakingAIDialogue dialogue = mongoTemplate.findOne(query, SpeakingAIDialogue.class); if (dialogue == null) { log.error("{} 未找到对话记录, userId: {}, assignmentId: {}", AUDIO_SENTIMENT_HEADER, userId, assignmentId); throw new MyException(HttpStatus.BAD_REQUEST.value(), "不存在相应记录"); } if (index == -1) { index = Math.max(dialogue.getDialogues().size() - 2, 0); log.debug("{} 自动计算index: {}", AUDIO_SENTIMENT_HEADER, index); } SpeakingAIDialogue.DialogueEntry entry = dialogue.getDialogues().get(index); JSONObject json = new JSONObject(); json.put("url", entry.getAudioUrl()); RequestBody body = RequestBody.create(MediaType.get("application/json; charset=utf-8"), json.toString()); Request request = new Request.Builder() .url(AUDIO_URL) .post(body) .build(); try (Response response = httpClient.newCall(request).execute()) { if (!response.isSuccessful()) { log.error("{} 音频情感分析请求失败, 状态码: {}, 响应: {}", AUDIO_SENTIMENT_HEADER, response.code(), response.body().string()); throw new IOException("Unexpected code " + response); } String res = response.body().string(); log.debug("{} 音频情感分析原始响应: {}", AUDIO_SENTIMENT_HEADER, res); int endIndex = res.lastIndexOf("\""); String emotion = res.substring(endIndex - 2, endIndex); if (!emotionSet.contains(emotion)) { emotion = "😐"; log.warn("{} 未识别的表情符号, 使用默认表情", AUDIO_SENTIMENT_HEADER); } entry.setAudioSentiment(emotion); updateDialogueEntry(userId, assignmentId, index, entry); log.info("{} 音频情感分析完成, 结果: {}", AUDIO_SENTIMENT_HEADER, emotion); return emotion; } catch (IOException e) { log.error("{} 音频情感分析IO异常", AUDIO_SENTIMENT_HEADER, e); throw new MyException(HttpStatus.BAD_REQUEST.value(), "情感识别失败: " + e.getMessage()); } } @Override public String getAudioSentimentWithTimeout(Long userId, Long assignmentId, Integer index) { log.info("{} 开始带超时的音频情感分析, userId: {}, assignmentId: {}, index: {}", AUDIO_SENTIMENT_HEADER, userId, assignmentId, index); CompletableFuture future = getAudioSentimentAsync(userId, assignmentId, index); try { return future.get(DEFAULT_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (TimeoutException e) { log.error("{} 音频情感分析请求超时", AUDIO_SENTIMENT_HEADER, e); throw new MyException(HttpStatus.REQUEST_TIMEOUT.value(), "请求超时"); } catch (Exception e) { log.error("{} 音频情感分析异常", AUDIO_SENTIMENT_HEADER, e); throw new MyException(HttpStatus.BAD_REQUEST.value(), "情感识别失败: " + e.getMessage()); } } @Async("doubaoTaskExecutor") public CompletableFuture getSpeakingSuggestionAsync(Long userId, Long assignmentId) { return CompletableFuture.supplyAsync(() -> getSpeakingSuggestion(userId, assignmentId), doubaoTaskExecutor); } public String getSpeakingSuggestion(Long userId, Long assignmentId) { String requestKey = userId + "-" + assignmentId + "-suggestion"; log.info("{} 开始获取口语建议, userId: {}, assignmentId: {}", SPEAKING_SUGGESTION_HEADER, userId, assignmentId); // 检查是否有相同的请求正在进行 CompletableFuture existingFuture = pendingSuggestionRequests.get(requestKey); if (existingFuture != null) { log.debug("{} 发现重复请求, 等待已有请求完成", SPEAKING_SUGGESTION_HEADER); try { return existingFuture.get(); } catch (Exception e) { pendingSuggestionRequests.remove(requestKey); log.error("{} 等待重复请求时发生异常", SPEAKING_SUGGESTION_HEADER, e); throw new MyException(HttpStatus.BAD_REQUEST.value(), "请求次数过多,请稍后再试"); } } SpeakingAIDialogue existingDialogue = mongoTemplate.findOne( Query.query(Criteria.where("assignmentId").is(assignmentId).and("userId").is(userId)), SpeakingAIDialogue.class ); if (existingDialogue == null) { log.error("{} 未找到对话记录, userId: {}, assignmentId: {}", SPEAKING_SUGGESTION_HEADER, userId, assignmentId); throw new MyException(HttpStatus.BAD_REQUEST.value(), "不存在相应记录"); } if (existingDialogue.getSuggestion() != null && existingDialogue.getSuggestion().length() != 0) { log.debug("{} 返回已存在的建议", SPEAKING_SUGGESTION_HEADER); return existingDialogue.getSuggestion(); } // 创建新的Future并放入缓存 CompletableFuture future = CompletableFuture.supplyAsync(() -> { try { String message = buildSuggestionMessage(existingDialogue); log.debug("{} 构建的建议请求消息: {}", SPEAKING_SUGGESTION_HEADER, message); ChatMessage chatMessage = ChatMessage.builder() .role(ChatMessageRole.USER) .content(message) .build(); List list = Collections.singletonList(chatMessage); ChatCompletionRequest chatCompletionRequest = ChatCompletionRequest.builder() .model(doubaoConfig.getEndpointId()) .messages(list) .build(); String response = (String) doubaoConfig.arkService() .createChatCompletion(chatCompletionRequest) .getChoices().get(0).getMessage().getContent(); log.info("{} 获取到口语建议响应: {}", SPEAKING_SUGGESTION_HEADER, response); existingDialogue.setSuggestion(response); mongoTemplate.save(existingDialogue); return response; } catch (Exception e) { log.error("{} 获取口语建议异常", SPEAKING_SUGGESTION_HEADER, e); throw new RuntimeException("获取建议失败: " + e.getMessage()); } finally { pendingSuggestionRequests.remove(requestKey); } }, doubaoTaskExecutor); pendingSuggestionRequests.put(requestKey, future); try { return future.get(); } catch (Exception e) { pendingSuggestionRequests.remove(requestKey); log.error("{} 处理口语建议时异常", SPEAKING_SUGGESTION_HEADER, e); throw new MyException(HttpStatus.BAD_REQUEST.value(), "获取建议失败: " + e.getMessage()); } } @Override public String getSpeakingSuggestionWithTimeout(Long userId, Long assignmentId) { log.info("{} 开始带超时的口语建议获取, userId: {}, assignmentId: {}", SPEAKING_SUGGESTION_HEADER, userId, assignmentId); CompletableFuture future = getSpeakingSuggestionAsync(userId, assignmentId); try { return future.get(DEFAULT_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (TimeoutException e) { log.error("{} 获取口语建议请求超时", SPEAKING_SUGGESTION_HEADER, e); throw new MyException(HttpStatus.REQUEST_TIMEOUT.value(), "获取建议请求超时"); } catch (Exception e) { log.error("{} 获取口语建议异常", SPEAKING_SUGGESTION_HEADER, e); throw new MyException(HttpStatus.BAD_REQUEST.value(), "获取建议失败: " + e.getMessage()); } } public VideoAnalysisVo getVideoSentiment(Long userId, Long assignmentId) { String requestKey = userId + "-" + assignmentId + "-video-sentiment"; log.info("{} 开始获取视频情感分析, userId: {}, assignmentId: {}", VIDEO_SENTIMENT_HEADER, userId, assignmentId); // 检查是否有相同的请求正在进行 CompletableFuture existingFuture = pendingVideoAnalysisRequests.get(requestKey); if (existingFuture != null) { log.debug("{} 发现重复视频分析请求, 等待已有请求完成", VIDEO_SENTIMENT_HEADER); try { return existingFuture.get(VIDEO_ANALYSIS_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (Exception e) { pendingVideoAnalysisRequests.remove(requestKey); log.error("{} 等待重复视频分析请求时异常", VIDEO_SENTIMENT_HEADER, e); throw new MyException(HttpStatus.BAD_REQUEST.value(), "请求次数过多,请稍后再试"); } } Query query = new Query(Criteria.where("userId").is(userId).and("assignmentId").is(assignmentId)); SpeakingAIDialogue dialogue = mongoTemplate.findOne(query, SpeakingAIDialogue.class); if (dialogue == null) { log.error("{} 未找到对话记录, userId: {}, assignmentId: {}", VIDEO_SENTIMENT_HEADER, userId, assignmentId); throw new MyException(HttpStatus.BAD_REQUEST.value(), "不存在相应记录"); } if (dialogue.getVideoAnalysis() != null && dialogue.getVideoAnalysis().getNewAudioUrl() != null) { log.debug("{} 返回已存在的视频分析结果", VIDEO_SENTIMENT_HEADER); return dialogue.getVideoAnalysis(); } // 提交任务到线程池 CompletableFuture future = CompletableFuture.supplyAsync(() -> { try { VideoAnalysisVo result = sendRequestToDeepface(dialogue.getAudioSigment(), dialogue); log.info("{} 视频情感分析完成, 结果: {}", VIDEO_SENTIMENT_HEADER, result); return result; } catch (IOException e) { log.error("{} 视频情感分析IO异常", VIDEO_SENTIMENT_HEADER, e); throw new RuntimeException("视频情感识别失败: " + e.getMessage()); } finally { pendingVideoAnalysisRequests.remove(requestKey); } }, doubaoTaskExecutor); pendingVideoAnalysisRequests.put(requestKey, future); try { return future.get(VIDEO_ANALYSIS_TIMEOUT_SECONDS, TimeUnit.SECONDS); } catch (TimeoutException e) { log.error("{} 视频情感分析请求超时", VIDEO_SENTIMENT_HEADER, e); throw new MyException(HttpStatus.REQUEST_TIMEOUT.value(), "请求超时"); } catch (Exception e) { pendingVideoAnalysisRequests.remove(requestKey); log.error("{} 视频情感分析异常", VIDEO_SENTIMENT_HEADER, e); throw new MyException(HttpStatus.BAD_REQUEST.value(), "情感识别失败: " + e.getMessage()); } } private VideoAnalysisVo sendRequestToDeepface(List audioSegment, SpeakingAIDialogue dialogue) throws IOException { log.debug("{} 开始发送视频分析请求, 音频片段数量: {}", VIDEO_SENTIMENT_HEADER, audioSegment.size()); JSONObject json = new JSONObject(); json.put("video_urls", audioSegment); RequestBody body = RequestBody.create(MediaType.get("application/json; charset=utf-8"), json.toString()); Request request = new Request.Builder() .url(VIDEO_URL) .post(body) .build(); try (Response response = httpClient.newCall(request).execute()) { if (!response.isSuccessful()) { String errorBody = response.body() != null ? response.body().string() : "empty body"; log.error("{} 视频分析请求失败, 状态码: {}, 响应: {}", VIDEO_SENTIMENT_HEADER, response.code(), errorBody); throw new IOException("Unexpected code " + response); } String res = response.body().string(); log.debug("{} 视频分析原始响应: {}", VIDEO_SENTIMENT_HEADER, res); VideoAnalysisVo videoAnalysisVo = parseResponse(res); dialogue.setVideoAnalysis(videoAnalysisVo); mongoTemplate.save(dialogue); return videoAnalysisVo; } } private VideoAnalysisVo parseResponse(String response) throws JsonProcessingException { try { ObjectMapper objectMapper = new ObjectMapper(); Map jsonMap = objectMapper.readValue(response, Map.class); VideoAnalysisVo videoAnalysisVo = new VideoAnalysisVo(); videoAnalysisVo.setNewAudioUrl((String) jsonMap.get("merged_video_path")); Map emotionPercentageDict = (Map) jsonMap.get("emotion_percentage_dict"); List emotionPercentageList = new ArrayList<>(); for (Map.Entry entry : emotionPercentageDict.entrySet()) { VideoAnalysisVo.EmotionPercentage emotionPercentage = new VideoAnalysisVo.EmotionPercentage(); emotionPercentage.setEmotion(entry.getKey()); emotionPercentage.setPercentage(entry.getValue().floatValue()); emotionPercentageList.add(emotionPercentage); } videoAnalysisVo.setEmotionPercentageList(emotionPercentageList); return videoAnalysisVo; } catch (Exception e) { log.error("{} 解析视频分析响应异常, 响应内容: {}", VIDEO_SENTIMENT_HEADER, response, e); throw e; } } private void updateDialogueEntry(Long userId, Long assignmentId, int index, SpeakingAIDialogue.DialogueEntry updatedEntry) { try { Query query = new Query(Criteria.where("userId").is(userId).and("assignmentId").is(assignmentId)); Update update = new Update().set("dialogues." + index, updatedEntry); mongoTemplate.updateFirst(query, update, SpeakingAIDialogue.class); log.debug("{} 成功更新对话条目, userId: {}, assignmentId: {}, index: {}", AUDIO_SCORE_HEADER, userId, assignmentId, index); } catch (Exception e) { log.error("{} 更新对话条目失败", AUDIO_SCORE_HEADER, e); throw new MyException(HttpStatus.INTERNAL_SERVER_ERROR.value(), "更新对话条目失败"); } } private String buildSuggestionMessage(SpeakingAIDialogue dialogue) { return "你是一位英语助教老师,下面是一位学生口语练习时的口语记录和口语得分," + dialogue + "\n请严格按照以下格式返回详细且实用的口语和交流建议:" + "1. 语法错误:[评估句法复杂性与准确性,提出修改建议,提供修改后的样本,提供提升建议]\n" + "2. 主题相关性:[分析对话内容与主题的关联程度,是否紧扣主题,有无偏离主题的情况,提出改进方向]\n" + "3. 逻辑表现情况:[判断对话逻辑是否清晰,过渡是否自然,给出优化逻辑的建议]\n" + "4. 词汇量:[评估说话人的词汇量、词汇复杂性、词汇准确性、词汇搭配,提供修改建议,提供修改后的样本,提供提升建议]\n" + "5. 语用能力评估和反馈:[根据语境、身份等评估说话人话语在内容和语言表达上的恰当性,提供修改建议,提供修改后的样本,提出提升建议]" + "6. 发音准确性:" + "- 具体表现:详细描述学生的发音准确性,例如哪些单词或音标发音正确(如元音、辅音、连读等),哪些发音存在问题(如常见的 /θ/、/ð/、/r/ 等音标)。" + "- 改进建议:提供具体的练习方法,例如:" + " - 针对错误发音,推荐使用音标练习工具(如 IPA 图表)或模仿母语者的发音视频。" + " - 练习单词时,建议使用“慢速-正常速”对比法,逐步纠正发音。" + " - 对于连读和弱读问题,推荐练习常见的连读规则(如 \"want to\" → \"wanna\")。" + "7. 对话流利度:" + "- 具体表现:评估学生的流利度,例如是否存在长时间停顿、重复、自我纠正或语法错误导致的卡顿。" + "- 改进建议:提供实用的提升方法" + "8. 语音情绪分析:" + "- 具体表现:分析学生在对话中的情绪表达,例如是否表现出自信、紧张、平淡或热情。" + "- 改进建议:提供情绪表达的优化建议,例如:" + " - 如果学生语气平淡,建议练习通过语调变化(如升调、降调)增强表达力。" + " - 如果学生表现出紧张,建议通过深呼吸和放松练习缓解压力,同时多进行模拟对话以增强信心。" + " - 鼓励学生在对话中加入情感词汇(如 \"I'm really excited about...\" 或 \"I feel a bit unsure about...\")以增强情感表达。"; } }