TranslationApplicationService.java

package com.yumu.noveltranslator.application.service;

import com.alibaba.fastjson2.JSON;
import com.yumu.noveltranslator.port.dto.translation.SelectionTranslateResponse;
import com.yumu.noveltranslator.port.dto.translation.ReaderTranslateResponse;
import com.yumu.noveltranslator.port.dto.translation.SelectionTranslationRequest;
import com.yumu.noveltranslator.port.dto.translation.ReaderTranslateRequest;
import com.yumu.noveltranslator.port.dto.translation.WebpageTranslateRequest;
import com.yumu.noveltranslator.domain.model.User;
import com.yumu.noveltranslator.enums.TranslationMode;
import com.yumu.noveltranslator.domain.service.TranslationPipeline;
import com.yumu.noveltranslator.application.service.RagTranslationApplicationService;
import com.yumu.noveltranslator.domain.service.TranslationPostProcessingService;
import com.yumu.noveltranslator.domain.service.EntityConsistencyService;
import com.yumu.noveltranslator.domain.service.QuotaService;
import com.yumu.noveltranslator.domain.util.EngineAliasRegistry;
import com.yumu.noveltranslator.port.out.TranslationCachePort;
import com.yumu.noveltranslator.port.out.TranslationClientPort;
import com.yumu.noveltranslator.port.out.TeamTranslationPort;
import com.yumu.noveltranslator.util.CacheKeyUtil;
import com.yumu.noveltranslator.util.SseEmitterUtil;
import com.yumu.noveltranslator.util.TextCleaningUtil;
import com.yumu.noveltranslator.util.TextSegmentationUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 翻译服务类
 * 提供三种翻译模式：选中文本翻译、阅读器翻译、网页翻译
 *
 * 优化特性：
 * - 三级缓存：内存缓存 → 数据库缓存 → 翻译服务
 * - 改进的缓存 Key：使用 MD5 避免碰撞
 * - 全局线程池复用
 * - 批量翻译优化
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class TranslationApplicationService implements com.yumu.noveltranslator.port.in.TranslatePort {

    private static final String DEFAULT_TARGET_LANG = "zh";
    private static final String DEFAULT_ENGINE = "auto";

    /**
     * 语言代码 → 语言名称映射（用于团队翻译 prompt）
     * Python 侧 prompt 使用自然语言名称（如 "请将以下中文文本翻译为日文"）
     */
    private static final Map<String, String> LANGUAGE_NAME_MAP = Map.ofEntries(
            Map.entry("zh", "中文"),
            Map.entry("cn", "中文"),
            Map.entry("en", "英文"),
            Map.entry("ja", "日文"),
            Map.entry("jp", "日文"),
            Map.entry("ko", "韩文"),
            Map.entry("fr", "法文"),
            Map.entry("de", "德文"),
            Map.entry("es", "西班牙文"),
            Map.entry("pt", "葡萄牙文"),
            Map.entry("ru", "俄文"),
            Map.entry("it", "意大利文"),
            Map.entry("ar", "阿拉伯文"),
            Map.entry("th", "泰文"),
            Map.entry("vi", "越南文")
    );

    /**
     * 将语言代码转换为自然语言名称
     * 例如 "zh" → "中文", "ja" → "日文"
     */
    private static String toLanguageName(String code) {
        if (code == null || "auto".equalsIgnoreCase(code)) {
            return "未知";
        }
        return LANGUAGE_NAME_MAP.getOrDefault(code.toLowerCase(), code);
    }

    // 依赖注入
    private final TranslationClientPort translationClientPort;
    private final TranslationCachePort cachePort;
    private final RagTranslationApplicationService ragTranslationService;
    private final com.yumu.noveltranslator.domain.service.EntityConsistencyService entityConsistencyService;
    private final TranslationPostProcessingService postProcessingService;
    private final TeamTranslationPort teamTranslationPort;
    private final com.yumu.noveltranslator.domain.service.QuotaService quotaService;

    /**
     * 选中文本翻译
     * @param req 翻译请求
     * @return 翻译响应
     */
    public SelectionTranslateResponse selectionTranslate(Long userId, SelectionTranslationRequest req) {
        String combined = req.getText();
        if (combined == null || combined.trim().isEmpty()) {
            return new SelectionTranslateResponse(false, req.getEngine(), "内容为空");
        }

        TranslationMode mode = EngineAliasRegistry.normalizeToMode(req.getEngine());
        String target = req.getTargetLang() == null ? DEFAULT_TARGET_LANG : req.getTargetLang();

        // 检查字符配额
        if (userId != null) {
            String userLevel = com.yumu.noveltranslator.util.SecurityUtil.getCurrentUserLevel().orElse(null);
            if (userLevel != null) {
                String modeName = req.getMode() != null ? req.getMode() : "fast";
                if (!quotaService.tryConsumeChars(userId, userLevel, combined.length(), modeName)) {
                    return new SelectionTranslateResponse(false, mode.getName(), "字符配额不足，请升级档位或等待下月重置");
                }
            }
        }

        try {
            TranslationPipeline pipeline = new TranslationPipeline(
                    cachePort, ragTranslationService, entityConsistencyService,
                    translationClientPort, postProcessingService, userId, null);
            String result = pipeline.executeFast(combined, target, mode);
            if (result == null || result.trim().isEmpty()) {
                if (userId != null) {
                    String modeName = req.getMode() != null ? req.getMode() : "fast";
                    quotaService.refundChars(userId, combined.length(), modeName);
                }
                return new SelectionTranslateResponse(false, mode.getName(), "翻译失败：返回结果为空");
            }
            // 净化翻译结果，防止恶意 HTML/脚本注入（XSS 防护）
            String sanitized = TextCleaningUtil.sanitizeHtml(result);
            return new SelectionTranslateResponse(true, mode.getName(), sanitized);
        } catch (Exception e) {
            log.error("选中文本翻译失败：{}", e.getMessage(), e);
            if (userId != null) {
                String modeName = req.getMode() != null ? req.getMode() : "fast";
                quotaService.refundChars(userId, combined.length(), modeName);
            }
            return new SelectionTranslateResponse(false, mode.getName(), "翻译失败：服务器内部错误");
        }
    }

    /**
     * 阅读器翻译 - 支持并行翻译长文本
     * 默认使用 MTranServer 进行翻译，html 模式启用
     * @param req 翻译请求
     * @return 翻译响应
     */
    public ReaderTranslateResponse readerTranslate(Long userId, ReaderTranslateRequest req) {
        String content = req.getContent();
        if (content == null || content.trim().isEmpty()) {
            return new ReaderTranslateResponse(false, req.getEngine(), "没有收到内容");
        }

        String target = req.getTargetLang() == null ? DEFAULT_TARGET_LANG : req.getTargetLang();
        TranslationMode mode = EngineAliasRegistry.normalizeToMode(req.getEngine());

        // 检查字符配额
        if (userId != null) {
            String userLevel = com.yumu.noveltranslator.util.SecurityUtil.getCurrentUserLevel().orElse(null);
            if (userLevel != null) {
                String modeName = req.getMode() != null ? req.getMode() : "fast";
                if (!quotaService.tryConsumeChars(userId, userLevel, content.length(), modeName)) {
                    return new ReaderTranslateResponse(false, mode.getName(), "字符配额不足，请升级档位或等待下月重置");
                }
            }
        }

        // 文本分段
        List<String> segments;
        try {
            segments = TextSegmentationUtil.segmentByTextEngine(content, mode.getName());
        } catch (Exception e) {
            log.error("阅读器文本分段失败：{}", e.getMessage(), e);
            if (userId != null) {
                String modeName = req.getMode() != null ? req.getMode() : "fast";
                quotaService.refundChars(userId, content.length(), modeName);
            }
            return new ReaderTranslateResponse(false, mode.getName(), "文本分段失败：服务器内部错误");
        }
        if (segments.isEmpty()) {
            if (userId != null) {
                String modeName = req.getMode() != null ? req.getMode() : "fast";
                quotaService.refundChars(userId, content.length(), modeName);
            }
            return new ReaderTranslateResponse(false, mode.getName(), "段落为空");
        }

        // 并行翻译所有段落（复用全局线程池）
        // 阅读器模式默认使用 MTranServer，html=true
        try {
            List<String> translatedSegments = translateSegmentsInParallel(segments, target, mode, true, userId);
            String rawResult = String.join("", translatedSegments);
            String combinedResult = TextCleaningUtil.sanitizeHtml(rawResult);
            log.info("阅读器翻译完成，原文长度：{}，译文长度：{}", content.length(), combinedResult.length());
            return new ReaderTranslateResponse(true, mode.getName(), combinedResult);
        } catch (Exception e) {
            log.error("阅读器翻译失败：{}", e.getMessage(), e);
            if (userId != null) {
                String modeName = req.getMode() != null ? req.getMode() : "fast";
                quotaService.refundChars(userId, content.length(), modeName);
            }
            return new ReaderTranslateResponse(false, mode.getName(), "翻译失败：服务器内部错误");
        }
    }

    /**
     * 并行翻译文本段落（使用虚拟线程并发翻译）
     *
     * 优化策略：
     * 1. 先尝试从缓存获取每个段落
     * 2. 对缓存未命中的段落使用虚拟线程并发翻译
     * 3. 翻译结果写回缓存
     *
     * @param segments 待翻译的文本段落
     * @param target 目标语言
     * @param engine 翻译引擎
     * @param html 是否启用 HTML 翻译模式（仅对 MTranServer 有效）
     */
    private List<String> translateSegmentsInParallel(List<String> segments, String target, TranslationMode mode,
                                                      boolean html, Long userId) {
        List<String> results = new ArrayList<>(segments.size());
        List<int[]> indexesToTranslate = new ArrayList<>();

        // 1. 先尝试从缓存获取（分层缓存）
        for (int i = 0; i < segments.size(); i++) {
            String segment = segments.get(i);
            String cacheKey = CacheKeyUtil.buildCacheKey(segment, target);
            String cached = cachePort.getCacheByMode(cacheKey, mode.getName()).orElse(null);

            if (cached != null) {
                results.add(cached);
                log.debug("阅读器缓存命中：索引 {} mode={}", i, mode.getName());
            } else {
                results.add(null); // 占位，稍后填充
                indexesToTranslate.add(new int[]{i});
            }
        }

        // 2. 使用 Pipeline 快速模式并发翻译缓存未命中的段落
        if (!indexesToTranslate.isEmpty()) {
            TranslationPipeline pipeline = new TranslationPipeline(
                    cachePort, ragTranslationService, entityConsistencyService,
                    translationClientPort, postProcessingService, userId, null);

            List<Runnable> tasks = new ArrayList<>();
            List<String> tempResults = new ArrayList<>(indexesToTranslate.size());

            for (int j = 0; j < indexesToTranslate.size(); j++) {
                tempResults.add(null);
            }

            for (int j = 0; j < indexesToTranslate.size(); j++) {
                int i = indexesToTranslate.get(j)[0];
                final int index = i;
                final int taskIndex = j;
                String segment = segments.get(i);

                tasks.add(() -> {
                    try {
                        String translation = pipeline.executeFast(segment, target, mode, html);
                        if (translation != null && !translation.trim().isEmpty()) {
                            tempResults.set(taskIndex, translation);
                        } else {
                            tempResults.set(taskIndex, segment);
                        }
                    } catch (Exception e) {
                        log.error("翻译段落失败（使用原文兜底）：索引 {}, 错误：{}", index, e.getMessage());
                        tempResults.set(taskIndex, segment);
                    }
                });
            }

            // 使用虚拟线程并发执行所有翻译任务
            List<Thread> threads = new ArrayList<>();
            for (Runnable task : tasks) {
                Thread thread = Thread.startVirtualThread(task);
                threads.add(thread);
            }

            // 等待所有线程完成
            for (Thread thread : threads) {
                try {
                    thread.join();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("等待翻译线程被中断");
                }
            }

            // 填充翻译结果
            for (int j = 0; j < indexesToTranslate.size(); j++) {
                int i = indexesToTranslate.get(j)[0];
                String translation = tempResults.get(j);
                if (translation != null) {
                    results.set(i, translation);
                }
            }
        }

        // 3. 清理 null 值
        for (int i = 0; i < results.size(); i++) {
            if (results.get(i) == null) {
                results.set(i, segments.get(i));
                log.warn("发现未处理的 null 值，使用原文填充：索引 {}", i);
            }
        }

        return results;
    }

    /**
     * 网页翻译流式版本 - 并发翻译（虚拟线程 + 信号量限流）
     */
    public SseEmitter webpageTranslateStream(Long userId, WebpageTranslateRequest req) {
        SseEmitter emitter = SseEmitterUtil.createSseEmitter(300000L); // 5 分钟超时

        // 计算总字符数
        var items = req.getTextRegistry();
        int totalChars = items.stream()
                .mapToInt(item -> item.getOriginal() != null ? item.getOriginal().length() : 0)
                .sum();

        // 检查字符配额
        String quotaMode = (req.getFastMode() != null && !req.getFastMode()) ? "expert" : "fast";

        if (userId != null) {
            String userLevel = com.yumu.noveltranslator.util.SecurityUtil.getCurrentUserLevel().orElse(null);
            if (userLevel != null) {
                if (!quotaService.tryConsumeChars(userId, userLevel, totalChars, quotaMode)) {
                    SseEmitterUtil.sendError(emitter, "字符配额不足，请升级档位或等待下月重置");
                    SseEmitterUtil.complete(emitter);
                    return emitter;
                }
            }
        }

        Thread.startVirtualThread(() -> {
            try {
                String target = req.getTargetLang() == null ? DEFAULT_TARGET_LANG : req.getTargetLang();
                TranslationMode mode = EngineAliasRegistry.normalizeToMode(req.getEngine());
                int totalCount = items.size();

                log.info("[SSE流式翻译] 开始处理，文本数量: {}, 引擎: {}, fastMode: {}", totalCount, req.getEngine(), req.getFastMode());

                TranslationPipeline pipeline = new TranslationPipeline(
                        cachePort, ragTranslationService, entityConsistencyService,
                        translationClientPort, postProcessingService, userId, null);

                log.info("[SSE流式翻译] Pipeline 初始化完成");

                // 限制并发数，避免超过 DeepSeek 限流（10 req/s）和用户级别并发限制
                int maxConcurrency = 5;
                var semaphore = new java.util.concurrent.Semaphore(maxConcurrency);
                var barrier = new java.util.concurrent.CountDownLatch(totalCount);

                for (WebpageTranslateRequest.TextItem item : items) {
                    // 用 final 局部变量捕获当前迭代的 item，防止 lambda 闭包在多线程环境下共享循环变量
                    final WebpageTranslateRequest.TextItem currentItem = item;
                    Thread.startVirtualThread(() -> {
                        try {
                            semaphore.acquire();
                            String id = currentItem.getId() == null ? "" : currentItem.getId();
                            String original = currentItem.getOriginal() == null ? "" : currentItem.getOriginal();
                            String cleanText = TextCleaningUtil.cleanText(original);

                            // fastMode=true（默认）使用 MTranServer，fastMode=false（专家模式）使用 DeepSeek
                            boolean fastMode = req.getFastMode() == null || req.getFastMode();
                            String extracted = pipeline.executeFast(cleanText, target, mode, !fastMode);

                            if (extracted == null || extracted.trim().isEmpty()) {
                                log.warn("翻译失败，使用原文兜底：ID={}", id);
                                extracted = original;
                            }

                            Map<String, Object> eventData = new HashMap<>(3);
                            // 净化翻译结果，防止恶意 HTML/脚本注入（XSS 防护）
                            String sanitized = TextCleaningUtil.sanitizeHtml(extracted);
                            eventData.put("textId", id);
                            eventData.put("original", original);
                            eventData.put("translation", sanitized);
                            SseEmitterUtil.sendData(emitter, JSON.toJSONString(eventData));
                        } catch (Exception e) {
                            log.error("翻译失败 - ID: {}, 错误: {}", currentItem.getId(), e.getMessage());
                            Map<String, Object> errorData = new HashMap<>(3);
                            errorData.put("textId", currentItem.getId());
                            String orig = currentItem.getOriginal() == null ? "" : currentItem.getOriginal();
                            errorData.put("original", orig);
                            errorData.put("translation", TextCleaningUtil.sanitizeHtml(orig));
                            SseEmitterUtil.sendData(emitter, JSON.toJSONString(errorData));
                        } finally {
                            semaphore.release();
                            barrier.countDown();
                        }
                    });
                }

                // 心跳线程：每 30 秒发送一次心跳，使用注册表检测 emitter 是否仍然活跃
                String emitterId = SseEmitterUtil.registerEmitter(emitter);
                Thread.startVirtualThread(() -> {
                    while (SseEmitterUtil.isEmitterActive(emitterId)) {
                        try {
                            Thread.sleep(30_000);
                        } catch (InterruptedException e) {
                            Thread.currentThread().interrupt();
                            break;
                        }
                        SseEmitterUtil.sendHeartbeat(emitter);
                    }
                });

                log.info("[SSE流式翻译] 所有翻译线程已启动，等待完成 ({} 个文本)", totalCount);

                // 等待所有翻译完成
                barrier.await();

                log.info("[SSE流式翻译] 所有翻译已完成，发送 [DONE] 信号");

                SseEmitterUtil.sendDone(emitter);
                SseEmitterUtil.complete(emitter);

            } catch (Exception e) {
                // 翻译失败，触发心跳停止
                log.error("网页翻译失败：{}", e.getMessage(), e);
                if (userId != null) {
                    quotaService.refundChars(userId, totalChars, quotaMode);
                }
                SseEmitterUtil.sendError(emitter, "翻译失败：服务器内部错误");
                SseEmitterUtil.complete(emitter);
            }
        });

        return emitter;
    }

    /**
     * 文本流式翻译（SSE）— 适用于长文本的单段流式输出
     * 按段落分段，逐段 SSE 输出
     */
    public SseEmitter streamTextTranslate(Long userId, SelectionTranslationRequest req) {
        log.info("[STREAM-TRACE] Service entry: userId={}, engine={}, target={}, mode={}, textLen={}",
                userId, req.getEngine(), req.getTargetLang(), req.getMode(), req.getText() != null ? req.getText().length() : 0);

        SseEmitter emitter = SseEmitterUtil.createSseEmitter(300000L);

        String text = req.getText();
        if (text == null || text.trim().isEmpty()) {
            log.info("[STREAM-TRACE] Text empty, returning error");
            SseEmitterUtil.sendError(emitter, "文本不能为空");
            SseEmitterUtil.complete(emitter);
            return emitter;
        }

        TranslationMode mode = EngineAliasRegistry.normalizeToMode(req.getEngine());
        String target = req.getTargetLang() == null ? DEFAULT_TARGET_LANG : req.getTargetLang();
        String modeString = req.getMode() != null ? req.getMode() : "fast";
        log.info("[STREAM-TRACE] Normalized: mode={}, target={}", mode.getName(), target);

        // 检查字符配额（从 SecurityContext 获取 userLevel）
        if (userId != null) {
            String userLevel = com.yumu.noveltranslator.util.SecurityUtil.getCurrentUserLevel().orElse(null);
            if (userLevel != null) {
                if (!quotaService.tryConsumeChars(userId, userLevel, text.length(), modeString)) {
                    SseEmitterUtil.sendError(emitter, "字符配额不足，请升级档位或等待下月重置");
                    SseEmitterUtil.complete(emitter);
                    return emitter;
                }
            }
        }

        Thread.startVirtualThread(() -> {
            try {
                List<String> segments = TextSegmentationUtil.segmentByTextEngine(text, mode.getName());
                if (segments.isEmpty()) {
                    segments = List.of(text);
                }

                TranslationPipeline pipeline = new TranslationPipeline(
                        cachePort, ragTranslationService, entityConsistencyService,
                        translationClientPort, postProcessingService, userId, null);

                // 先用 \n\n+ 将全文拆分为逻辑段落
                String[] logicalParagraphs = text.split("\n\n+");
                log.info("[STREAM-TRACE] 全文拆分为 {} 个逻辑段落, 总行数 ~{}", logicalParagraphs.length,
                        text.split("\n", -1).length);

                StringBuilder fullResult = new StringBuilder();

                for (int i = 0; i < logicalParagraphs.length; i++) {
                    SseEmitterUtil.sendHeartbeat(emitter);

                    String para = logicalParagraphs[i];
                    if (para.isBlank()) {
                        fullResult.append("\n\n");
                        continue;
                    }

                    // 段落内按 \n 逐行翻译，保持原文单行换行结构
                    String[] lines = para.split("\n", -1);
                    StringBuilder paraResult = new StringBuilder();

                    log.info("[流式翻译] 段落包含 {} 行", lines.length);

                    for (int j = 0; j < lines.length; j++) {
                        String line = lines[j];
                        if (line.isEmpty()) {
                            paraResult.append("\n");
                            continue;
                        }

                        String translated;
                        String pipelineMode = "fast".equals(modeString) ? "executeFast" : "execute";
                        log.info("[STREAM-TRACE] Pipeline 委托: pipelineMode={}, lineLen={}, para={}, line={}",
                                pipelineMode, line.length(), i, j);
                        if ("fast".equals(modeString)) {
                            translated = pipeline.executeFast(line, target, mode);
                        } else {
                            translated = pipeline.execute(line, target, mode);
                        }
                        log.info("[STREAM-TRACE] Pipeline 返回: translatedLen={}", translated != null ? translated.length() : 0);

                        if (translated == null || translated.isBlank()) {
                            log.info("[STREAM-TRACE] Pipeline 返回空，使用原文兜底");
                            translated = line;
                        }

                        if (j > 0) paraResult.append("\n");
                        paraResult.append(translated);
                    }

                    log.info("[流式翻译] 段落结果行数: {} (与原文一致)", paraResult.toString().split("\n", -1).length);

                    if (i > 0) fullResult.append("\n\n");
                    fullResult.append(paraResult);

                    // 流式发送当前段落结果（含段间分隔符）
                    // §NL§ = 单换行, §NL§§NL§ = 段落分隔
                    String chunk = TextCleaningUtil.sanitizeHtml(paraResult.toString()).replace("\n", "§NL§");
                    if (i > 0) {
                        chunk = "§NL§§NL§" + chunk;
                    }
                    SseEmitterUtil.sendData(emitter, chunk);
                }

                log.info("[STREAM-TRACE] 流式翻译完成, 原文长度={}, 译文长度={}", text.length(), fullResult.length());
                SseEmitterUtil.sendDone(emitter);
                SseEmitterUtil.complete(emitter);

            } catch (Exception e) {
                log.error("文本流式翻译失败：{}", e.getMessage(), e);
                Long currentUserId = userId;
                if (currentUserId != null) {
                    quotaService.refundChars(currentUserId, text.length(), modeString);
                }
                SseEmitterUtil.sendError(emitter, "翻译失败：服务器内部错误");
                SseEmitterUtil.complete(emitter);
            }
        });

        return emitter;
    }

    /**
     * 获取缓存统计信息
     */
    public Map<String, Object> getCacheStats() {
        return cachePort.getCacheStats();
    }
}