package com.yumu.noveltranslator.application.service;

import com.yumu.noveltranslator.port.dto.translation.TranslationResultResponse;
import com.yumu.noveltranslator.port.dto.entity.TaskStatusResponse;
import com.yumu.noveltranslator.port.dto.entity.TranslationHistoryResponse;
import com.yumu.noveltranslator.domain.model.Document;
import com.yumu.noveltranslator.domain.model.Glossary;
import com.yumu.noveltranslator.domain.model.TranslationHistory;
import com.yumu.noveltranslator.domain.model.TranslationTask;
import com.yumu.noveltranslator.enums.TranslationMode;
import com.yumu.noveltranslator.enums.TranslationStatus;
import com.yumu.noveltranslator.port.out.TranslationRepositoryPort;
import com.yumu.noveltranslator.port.out.DocumentRepositoryPort;
import com.yumu.noveltranslator.port.out.GlossaryRepositoryPort;
import com.yumu.noveltranslator.port.out.TranslationCachePort;
import com.yumu.noveltranslator.port.out.TranslationClientPort;
import com.yumu.noveltranslator.domain.service.TranslationStateMachine;
import com.yumu.noveltranslator.domain.service.TranslationPipeline;
import com.yumu.noveltranslator.application.service.RagTranslationApplicationService;
import com.yumu.noveltranslator.domain.service.EntityConsistencyService;
import com.yumu.noveltranslator.domain.service.TranslationPostProcessingService;
import com.yumu.noveltranslator.util.ExternalResponseUtil;
import com.yumu.noveltranslator.port.in.StreamTranslateEventConsumer;
import com.yumu.noveltranslator.port.dto.translation.StreamTranslateEvent;
import lombok.RequiredArgsConstructor;
import org.springframework.scheduling.annotation.Scheduled;
import jakarta.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;

import com.yumu.noveltranslator.port.out.UserRepositoryPort;

/**
 * 翻译任务服务
 */
@Service
@RequiredArgsConstructor
public class TranslationTaskApplicationService implements com.yumu.noveltranslator.port.in.TranslationTaskPort {
    private static final Logger log = LoggerFactory.getLogger(TranslationTaskApplicationService.class);

    private final TranslationRepositoryPort translationPort;
    private final DocumentRepositoryPort documentPort;
    private final GlossaryRepositoryPort glossaryPort;
    private final UserRepositoryPort userRepositoryPort;
    private final TranslationStateMachine stateMachine;
    private final TranslationClientPort translationClientPort;
    private final TranslationCachePort cachePort;
    private final RagTranslationApplicationService ragTranslationService;
    private final EntityConsistencyService entityConsistencyService;
    private final TranslationPostProcessingService postProcessingService;

    /**
     * 创建文档翻译任务
     */
    public TranslationTask createDocumentTask(Long userId, Document doc) {
        String taskId = generateTaskId();

        TranslationTask task = new TranslationTask();
        task.setTaskId(taskId);
        task.setUserId(userId);
        task.setType("document");
        task.setDocumentId(doc.getId());
        task.setSourceLang(doc.getSourceLang());
        task.setTargetLang(doc.getTargetLang());
        task.setMode(doc.getMode());
        task.setEngine("google");
        task.setStatus(TranslationStatus.PENDING.getValue());
        task.setProgress(0);
        task.setCreateTime(LocalDateTime.now());

        translationPort.saveTask(task);

        // 关联文档与任务
        doc.setTaskId(taskId);
        documentPort.update(doc);

        return task;
    }

    /**
     * 启动文档翻译
     */
    public void startDocumentTranslation(TranslationTask task, Document doc) {
        if (task == null || doc == null) {
            return;
        }
        if (!TranslationStatus.PENDING.getValue().equals(task.getStatus()) && !TranslationStatus.FAILED.getValue().equals(task.getStatus())) {
            return;
        }

        // 更新文档状态为处理中
        doc.setStatus(TranslationStatus.PROCESSING.getValue());
        doc.setUpdateTime(LocalDateTime.now());
        documentPort.update(doc);

        updateTaskProgress(task, TranslationStatus.PROCESSING, 10, null);

        executeDocumentTranslation(task, doc);
    }

    /**
     * 异步执行文档翻译（使用虚拟线程）
     * 接入三级缓存 + RAG + 实体一致性
     */
    private void executeDocumentTranslation(TranslationTask task, Document doc) {
        Thread.startVirtualThread(() -> {
            try {
                // 读取文件内容
                String content = readDocumentContent(doc.getPath(), doc.getFileType());
                if (content == null || content.trim().isEmpty()) {
                    updateTaskProgress(task, TranslationStatus.FAILED, 0, "文件内容为空");
                    return;
                }

                // 按段落分割翻译（每 3000 字符一批）
                updateTaskProgress(task, TranslationStatus.PROCESSING, 20, null);
                StringBuilder translatedContent = new StringBuilder();
                String[] paragraphs = content.split("(?<=\n)");
                int total = paragraphs.length;
                int batchStart = 0;
                String targetLang = task.getTargetLang();
                String engine = task.getEngine();
                Long userId = task.getUserId();
                String docId = "doc_" + doc.getId();

                // 加载用户术语表
                List<Glossary> glossaryTerms = loadGlossaryTermsForUser(userId, content);

                // 查询用户等级用于限流和配额
                String userLevel = userRepositoryPort.findById(userId)
                        .map(com.yumu.noveltranslator.domain.model.User::getUserLevel)
                        .orElse(null);

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

                while (batchStart < total) {
                    StringBuilder batch = new StringBuilder();
                    int batchEnd = batchStart;
                    while (batchEnd < total && batch.length() + paragraphs[batchEnd].length() <= 1500) {
                        batch.append(paragraphs[batchEnd]);
                        batchEnd++;
                    }

                    if (batch.length() == 0) {
                        batch.append(paragraphs[batchStart]);
                        batchEnd = batchStart + 1;
                    }

                    String batchText = batch.toString();
                    String translated;
                    try {
                        if ("fast".equals(doc.getMode())) {
                            translated = pipeline.executeFast(batchText, targetLang, TranslationMode.FAST);
                        } else {
                            translated = pipeline.execute(batchText, targetLang, TranslationMode.EXPERT);
                        }
                    } catch (Exception e) {
                        log.warn("翻译批次失败 [{}/{}]，保留原文: {}", batchStart, total, e.getMessage());
                        translated = batchText; // fallback to original text
                    }
                    if (translated != null && !translated.isEmpty()) {
                        // 补回原文批次中的换行符以保持格式对齐
                        if (!translated.endsWith("\n") && !translated.endsWith("\r")) {
                            String trailingNewline = getTrailingNewline(batchText);
                            if (!trailingNewline.isEmpty()) {
                                translated += trailingNewline;
                            }
                        }
                        translatedContent.append(translated);
                    }
                    batchStart = batchEnd;

                    int progress = 20 + (int) ((batchStart * 80.0) / total);
                    updateTaskProgress(task, TranslationStatus.PROCESSING, progress, null);
                }

                // 保存翻译结果到文件
                String translatedPath = ExternalResponseUtil.buildTranslatedPath(doc.getPath());
                Files.write(Paths.get(translatedPath), translatedContent.toString().getBytes(java.nio.charset.StandardCharsets.UTF_8));

                // 更新文档状态为已完成
                doc.setStatus(TranslationStatus.COMPLETED.getValue());
                doc.setCompletedTime(LocalDateTime.now());
                doc.setUpdateTime(LocalDateTime.now());
                documentPort.update(doc);

                updateTaskProgress(task, TranslationStatus.COMPLETED, 100, null);
                saveTranslationHistory(task, content, translatedContent.toString());

            } catch (Exception e) {
                log.error("文档翻译失败: {}", e.getMessage(), e);
                updateTaskProgress(task, TranslationStatus.FAILED, 0, "翻译失败：" + e.getMessage());
            }
        });
    }

    /**
     * 读取文档内容
     */
    private String readDocumentContent(String filePath, String fileType) throws IOException {
        Path path = Paths.get(filePath);
        if (!Files.exists(path)) {
            throw new IOException("文件不存在: " + filePath);
        }

        if ("txt".equals(fileType)) {
            return Files.readString(path, java.nio.charset.StandardCharsets.UTF_8);
        } else {
            // DOCX/EPUB/PDF 暂不支持，返回原始文本
            throw new IOException("暂不支持 " + fileType.toUpperCase() + " 格式，仅支持 TXT 文件");
        }
    }

    @Transactional
    protected void updateTaskProgress(TranslationTask task, TranslationStatus status, int progress, String errorMessage) {
        // 如果任务已被取消，不再更新
        TranslationTask current = translationPort.findTaskByTaskId(task.getTaskId()).orElse(null);
        if (current != null && TranslationStatus.FAILED.getValue().equals(current.getStatus()) && "用户取消任务".equals(current.getErrorMessage())) {
            return;
        }
        task.setStatus(status.getValue());
        task.setProgress(progress);
        if (status == TranslationStatus.COMPLETED) {
            task.setCompletedTime(LocalDateTime.now());
        }
        if (errorMessage != null) {
            task.setErrorMessage(errorMessage);
        }
        translationPort.updateTask(task);
    }

    /**
     * 保存翻译历史
     */
    private void saveTranslationHistory(TranslationTask task, String sourceText, String targetText) {
        TranslationHistory history = new TranslationHistory();
        history.setUserId(task.getUserId());
        history.setTaskId(task.getTaskId());
        history.setType(task.getType());
        history.setDocumentId(task.getDocumentId());
        history.setSourceLang(task.getSourceLang());
        history.setTargetLang(task.getTargetLang());
        history.setSourceText(sourceText != null && sourceText.length() > 500
                ? sourceText.substring(0, 500)
                : sourceText);

        // 解析翻译响应，提取实际翻译内容
        String translatedContent = ExternalResponseUtil.extractDataField(targetText);
        history.setTargetText(translatedContent != null && translatedContent.length() > 500
                ? translatedContent.substring(0, 500)
                : translatedContent);
        history.setEngine(task.getEngine());
        history.setCreateTime(LocalDateTime.now());

        translationPort.saveHistory(history);
    }

    /**
     * 加载用户术语表中在原文中出现的词条
     */
    private List<Glossary> loadGlossaryTermsForUser(Long userId, String sourceText) {
        try {
            List<Glossary> allTerms = glossaryPort.findActiveGlossaryByUserId(userId);

            // 过滤只保留在原文中出现的术语
            return allTerms.stream()
                    .filter(term -> sourceText.contains(term.getSourceWord()))
                    .toList();
        } catch (Exception e) {
            log.warn("加载术语表失败: {}", e.getMessage());
            return List.of();
        }
    }

    /**
     * 根据任务 ID 获取任务
     */
    public TranslationTask getTaskByTaskId(String taskId) {
        return translationPort.findTaskByTaskId(taskId).orElse(null);
    }

    /**
     * 根据文档 ID 获取任务
     */
    public TranslationTask getTaskByDocumentId(Long docId) {
        return translationPort.findTaskByDocumentId(docId).orElse(null);
    }

    /**
     * 取消翻译任务
     */
    public boolean cancelTask(String taskId, Long userId) {
        TranslationTask task = translationPort.findTaskByTaskId(taskId).orElse(null);
        if (task != null && task.getUserId().equals(userId)) {
            if (TranslationStatus.PENDING.getValue().equals(task.getStatus()) || TranslationStatus.PROCESSING.getValue().equals(task.getStatus())) {
                task.setStatus(TranslationStatus.FAILED.getValue());
                task.setErrorMessage("用户取消任务");
                translationPort.updateTask(task);
                // 同步更新 Document 表状态
                if (task.getDocumentId() != null) {
                    Document doc = documentPort.findById(task.getDocumentId()).orElse(null);
                    if (doc != null) {
                        doc.setStatus(TranslationStatus.FAILED.getValue());
                        doc.setErrorMessage("用户取消任务");
                        doc.setUpdateTime(LocalDateTime.now());
                        documentPort.update(doc);
                    }
                }
                return true;
            }
        }
        return false;
    }

    /**
     * 删除翻译历史记录（逻辑删除）
     */
    public boolean deleteHistory(String taskId, Long userId) {
        TranslationHistory history = translationPort.findHistoryByTaskId(taskId).orElse(null);
        if (history != null && history.getUserId().equals(userId)) {
            translationPort.deleteHistory(history.getId());
            return true;
        }
        return false;
    }

    /**
     * 获取翻译结果
     */
    public TranslationResultResponse getTranslationResult(String taskId) {
        TranslationTask task = translationPort.findTaskByTaskId(taskId).orElse(null);
        if (task == null) {
            return null;
        }

        TranslationResultResponse response = new TranslationResultResponse();
        response.setTaskId(taskId);
        response.setStatus(task.getStatus());
        response.setSourceLang(task.getSourceLang());
        response.setTargetLang(task.getTargetLang());

        String status = task.getStatus();
        if (TranslationStatus.COMPLETED.getValue().equals(status) || TranslationStatus.PROCESSING.getValue().equals(status) || TranslationStatus.PENDING.getValue().equals(status) || TranslationStatus.FAILED.getValue().equals(status)) {
            if (TranslationStatus.COMPLETED.getValue().equals(status)) {
                response.setCompletedTime(task.getCompletedTime() != null
                        ? task.getCompletedTime().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
                        : null);
            }

            // 如果是文本翻译，从历史中获取结果
            if ("text".equals(task.getType())) {
                TranslationHistory history = translationPort.findHistoryByTaskId(task.getTaskId()).orElse(null);
                if (history != null) {
                    response.setTranslatedText(history.getTargetText());
                    response.setSourceContent(history.getSourceText());
                }
            } else if ("document".equals(task.getType())) {
                // 文档翻译，读取文件内容
                if (task.getDocumentId() != null) {
                    Document doc = documentPort.findById(task.getDocumentId()).orElse(null);
                    if (doc != null) {
                        response.setTranslatedFilePath(doc.getPath());
                        // 读取原文内容（任何状态都尝试读取）
                        try {
                            Path path = Paths.get(doc.getPath());
                            if (Files.exists(path)) {
                                String content = Files.readString(path, java.nio.charset.StandardCharsets.UTF_8);
                                response.setSourceContent(content);
                            }
                        } catch (Exception e) {
                            log.warn("读取原文文件失败: {}", e.getMessage());
                        }
                        // 已完成的文档翻译，读取翻译文件内容
                        if (TranslationStatus.COMPLETED.getValue().equals(status)) {
                            String translatedPath = ExternalResponseUtil.buildTranslatedPath(doc.getPath());
                            try {
                                Path path = Paths.get(translatedPath);
                                if (Files.exists(path)) {
                                    // 文件内容已经是纯文本翻译结果，不要再次调用 extractTranslatedContent
                                    response.setTranslatedText(Files.readString(path, java.nio.charset.StandardCharsets.UTF_8));
                                }
                            } catch (Exception e) {
                                log.warn("读取翻译文件失败: {}", e.getMessage());
                            }
                        }
                    }
                }
            }
        }

        return response;
    }

    /**
     * 下载翻译结果（返回文件路径）
     */
    public String getDownloadPath(String taskId, Long userId) {
        TranslationTask task = translationPort.findTaskByTaskId(taskId).orElse(null);
        if (task == null || !task.getUserId().equals(userId)) {
            return null;
        }

        if (TranslationStatus.COMPLETED.getValue().equals(task.getStatus()) && "document".equals(task.getType())) {
            if (task.getDocumentId() != null) {
                Document doc = documentPort.findById(task.getDocumentId()).orElse(null);
                if (doc != null && Files.exists(Paths.get(doc.getPath()))) {
                    return doc.getPath();
                }
            }
        }

        return null;
    }

    /**
     * 获取翻译历史列表（包含进行中的任务）
     */
    public List<TranslationHistory> getTranslationHistory(Long userId, int page, int pageSize, String status) {
        int offset = (page - 1) * pageSize;

        if ("completed".equals(status)) {
            // Completed records are in translation_history table
            return translationPort.findHistoryByUserId(userId, offset, pageSize);
        }

        if ("processing".equals(status)) {
            // Processing records are in translation_task table with pending/processing status
            return translationPort.findTasksByUserIdAndStatuses(userId, List.of("pending", "processing"), offset, pageSize)
                    .stream()
                    .map(this::taskToHistory)
                    .toList();
        }

        if ("failed".equals(status)) {
            // Failed records are in translation_task table with failed status
            return translationPort.findTasksByUserIdAndStatuses(userId, List.of("failed"), offset, pageSize)
                    .stream()
                    .map(this::taskToHistory)
                    .toList();
        }

        // No filter: merge in-progress tasks and completed histories
        List<TranslationHistory> histories = new ArrayList<>();

        List<TranslationTask> inProgressTasks = translationPort.findTasksByUserIdAndStatus(userId, offset, pageSize);
        for (TranslationTask task : inProgressTasks) {
            histories.add(taskToHistory(task));
        }

        List<TranslationHistory> completedHistories = translationPort.findHistoryByUserId(userId, offset, pageSize);
        histories.addAll(completedHistories);

        // Deduplicate by taskId, sort by createTime desc
        return histories.stream()
                .collect(java.util.stream.Collectors.toMap(
                        TranslationHistory::getTaskId,
                        h -> h,
                        (existing, replacement) -> existing
                ))
                .values()
                .stream()
                .sorted((a, b) -> {
                    if (a.getCreateTime() == null) return 1;
                    if (b.getCreateTime() == null) return -1;
                    return b.getCreateTime().compareTo(a.getCreateTime());
                })
                .limit(pageSize)
                .toList();
    }

    /**
     * 统计翻译历史总数（按状态过滤）
     */
    @Override
    public int countTranslationHistory(Long userId, String status) {
        if ("completed".equals(status)) {
            return translationPort.countHistoryByUserId(userId);
        }
        if ("processing".equals(status)) {
            return translationPort.countTasksByUserIdAndStatuses(userId, List.of("pending", "processing"));
        }
        if ("failed".equals(status)) {
            return translationPort.countTasksByUserIdAndStatuses(userId, List.of("failed"));
        }
        // No filter: count both in-progress tasks and completed histories
        int taskCount = translationPort.countTasksByUserIdAndStatuses(userId, List.of("pending", "processing"));
        int historyCount = translationPort.countHistoryByUserId(userId);
        return taskCount + historyCount;
    }

    /**
     * 生成任务 ID
     */
    private String generateTaskId() {
        return "task_" + UUID.randomUUID().toString().replace("-", "").substring(0, 16);
    }

    /**
     * 转换为 TaskStatusResponse
     */
    public TaskStatusResponse toTaskStatusResponse(TranslationTask task) {
        if (task == null) {
            return null;
        }

        TaskStatusResponse response = new TaskStatusResponse();
        response.setTaskId(task.getTaskId());
        response.setType(task.getType());
        response.setStatus(task.getStatus());
        response.setProgress(task.getProgress());
        response.setSourceLang(task.getSourceLang());
        response.setTargetLang(task.getTargetLang());
        response.setCreateTime(task.getCreateTime() != null
                ? task.getCreateTime().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
                : null);
        response.setCompletedTime(task.getCompletedTime() != null
                ? task.getCompletedTime().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
                : null);
        response.setErrorMessage(task.getErrorMessage());

        return response;
    }

    /**
     * 将 TranslationTask 转为 TranslationHistory（用于进行中/失败任务的历史列表展示）
     */
    private TranslationHistory taskToHistory(TranslationTask task) {
        TranslationHistory history = new TranslationHistory();
        history.setUserId(task.getUserId());
        history.setTaskId(task.getTaskId());
        history.setType(task.getType());
        history.setDocumentId(task.getDocumentId());
        history.setSourceLang(task.getSourceLang());
        history.setTargetLang(task.getTargetLang());
        history.setSourceText(null);
        history.setTargetText(null);
        history.setEngine(task.getEngine());
        history.setCreateTime(task.getCreateTime());
        return history;
    }

    /**
     * 转换为 TranslationHistoryResponse
     */
    public TranslationHistoryResponse toHistoryResponse(TranslationHistory history) {
        if (history == null) {
            return null;
        }

        TranslationHistoryResponse response = new TranslationHistoryResponse();
        response.setId(history.getId());
        response.setTaskId(history.getTaskId());
        response.setType(history.getType());
        response.setSourceLang(history.getSourceLang());
        response.setTargetLang(history.getTargetLang());
        response.setSourceTextPreview(history.getSourceText() != null
                ? history.getSourceText().substring(0, Math.min(100, history.getSourceText().length()))
                : null);
        response.setTargetTextPreview(history.getTargetText() != null
                ? history.getTargetText().substring(0, Math.min(100, history.getTargetText().length()))
                : null);
        response.setCreateTime(history.getCreateTime() != null
                ? history.getCreateTime().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME)
                : null);

        // 查询关联任务状态
        if (history.getTaskId() != null) {
            TranslationTask task = translationPort.findTaskByTaskId(history.getTaskId()).orElse(null);
            response.setStatus(task != null ? task.getStatus() : TranslationStatus.COMPLETED.getValue());
        } else {
            response.setStatus(TranslationStatus.COMPLETED.getValue());
        }

        // 查询关联文档名称
        if (history.getDocumentId() != null) {
            Document doc = documentPort.findById(history.getDocumentId()).orElse(null);
            if (doc != null) {
                response.setDocumentName(doc.getName());
            }
        }

        // fallback：如果documentName为空，尝试从任务关联的文档获取
        if (response.getDocumentName() == null && history.getTaskId() != null) {
            TranslationTask task = translationPort.findTaskByTaskId(history.getTaskId()).orElse(null);
            if (task != null && task.getDocumentId() != null) {
                Document doc = documentPort.findById(task.getDocumentId()).orElse(null);
                if (doc != null) {
                    response.setDocumentName(doc.getName());
                }
            }
        }

        // 最终fallback：使用任务类型作为名称
        if (response.getDocumentName() == null) {
            response.setDocumentName("document".equals(history.getType()) ? "文档翻译" : "文本翻译");
        }

        return response;
    }

    /**
     * 流式文档翻译（基于已上传文档）
     * 读取磁盘文件 → 分段翻译 → 通过 eventConsumer 推送事件
     */
    public void streamTranslateDocumentById(Long docId, String targetLang, String mode, StreamTranslateEventConsumer eventConsumer) {
        Thread.startVirtualThread(() -> {
            try {
                Document doc = documentPort.findById(docId).orElse(null);
                if (doc == null) {
                    eventConsumer.accept(StreamTranslateEvent.error("文档不存在"));
                    return;
                }

                TranslationTask task = createDocumentTask(doc.getUserId(), doc);
                doc.setStatus(TranslationStatus.PROCESSING.getValue());
                doc.setUpdateTime(LocalDateTime.now());
                documentPort.update(doc);
                updateTaskProgress(task, TranslationStatus.PROCESSING, 10, null);

                String content = readDocumentContent(doc.getPath(), doc.getFileType());
                if (content == null || content.trim().isEmpty()) {
                    updateTaskProgress(task, TranslationStatus.FAILED, 0, "文件内容为空");
                    eventConsumer.accept(StreamTranslateEvent.error("文件内容为空"));
                    eventConsumer.accept(StreamTranslateEvent.done());
                    return;
                }

                String[] paragraphs = content.split("(?<=\n)");
                int total = paragraphs.length;
                StringBuilder translatedContent = new StringBuilder();

                for (int i = 0; i < total; i++) {
                    TranslationTask currentTask = translationPort.findTaskByTaskId(task.getTaskId()).orElse(null);
                    if (currentTask != null && TranslationStatus.FAILED.getValue().equals(currentTask.getStatus())) {
                        log.info("翻译任务已被取消，提前退出 [taskId={}]", task.getTaskId());
                        eventConsumer.accept(StreamTranslateEvent.done());
                        return;
                    }

                    String paragraph = paragraphs[i];
                    if (paragraph.trim().isEmpty()) {
                        translatedContent.append(paragraph);
                        continue;
                    }

                    String textId = "seg_" + i;
                    try {
                        String translation = translateParagraph(paragraph, targetLang, mode, doc.getUserId(), "doc_" + docId);
                        if (translation == null || translation.isEmpty()) {
                            log.warn("翻译结果为空，保留原文");
                            translation = paragraph;
                        }
                        translation = restoreTrailingNewline(translation, paragraph);
                        translatedContent.append(translation);

                        eventConsumer.accept(StreamTranslateEvent.chunk(textId, paragraph, translation, (int) (((i + 1) * 100.0) / total)));
                    } catch (Exception e) {
                        log.warn("段落翻译失败 [{}]: {}", i, e.getMessage());
                        translatedContent.append(paragraph);
                        eventConsumer.accept(StreamTranslateEvent.chunk(textId, paragraph, paragraph, (int) (((i + 1) * 100.0) / total)));
                        // error flag handled in event data, but here we just continue
                    }

                    int progress = 20 + (int) (((i + 1) * 80.0) / total);
                    updateTaskProgress(task, TranslationStatus.PROCESSING, progress, null);
                }

                String translatedPath = ExternalResponseUtil.buildTranslatedPath(doc.getPath());
                Files.write(Paths.get(translatedPath), translatedContent.toString().getBytes(java.nio.charset.StandardCharsets.UTF_8));

                doc.setStatus(TranslationStatus.COMPLETED.getValue());
                doc.setCompletedTime(LocalDateTime.now());
                doc.setUpdateTime(LocalDateTime.now());
                documentPort.update(doc);

                updateTaskProgress(task, TranslationStatus.COMPLETED, 100, null);
                saveTranslationHistory(task, content, translatedContent.toString());

                eventConsumer.accept(StreamTranslateEvent.done());

            } catch (Exception e) {
                log.error("流式文档翻译失败: {}", e.getMessage(), e);
                try {
                    eventConsumer.accept(StreamTranslateEvent.error("翻译失败：" + e.getMessage()));
                    eventConsumer.accept(StreamTranslateEvent.done());
                } catch (IOException ex) {
                    log.error("发送错误事件失败", ex);
                }
            }
        });
    }

    /**
     * 流式文档翻译（基于上传文件）
     * 读取文件 → 分段翻译 → 通过 eventConsumer 推送事件
     */
    public void streamTranslateDocument(byte[] fileContent, String fileName, String sourceLang, String targetLang, String mode, StreamTranslateEventConsumer eventConsumer) {
        Thread.startVirtualThread(() -> {
            try {
                String fileType = fileName != null ? fileName.substring(fileName.lastIndexOf('.') + 1).toLowerCase() : "";
                String content = new String(fileContent, java.nio.charset.StandardCharsets.UTF_8);
                if (!"txt".equals(fileType)) {
                    eventConsumer.accept(StreamTranslateEvent.error("暂不支持 " + fileType.toUpperCase() + " 格式，仅支持 TXT 文件"));
                    eventConsumer.accept(StreamTranslateEvent.done());
                    return;
                }
                if (content == null || content.trim().isEmpty()) {
                    eventConsumer.accept(StreamTranslateEvent.error("文件内容为空"));
                    eventConsumer.accept(StreamTranslateEvent.done());
                    return;
                }

                String[] paragraphs = content.split("(?<=\n)");
                int total = paragraphs.length;

                for (int i = 0; i < total; i++) {
                    String paragraph = paragraphs[i];
                    if (paragraph.trim().isEmpty()) {
                        continue;
                    }

                    String textId = "seg_" + i;
                    try {
                        String translation = translateParagraph(paragraph, targetLang, mode, null, null);
                        if (translation == null || translation.isEmpty()) {
                            log.warn("翻译结果为空，保留原文");
                            translation = paragraph;
                        }
                        translation = restoreTrailingNewline(translation, paragraph);

                        eventConsumer.accept(StreamTranslateEvent.chunk(textId, paragraph, translation, (int) (((i + 1) * 100.0) / total)));
                    } catch (Exception e) {
                        log.warn("段落翻译失败 [{}]: {}", i, e.getMessage());
                        eventConsumer.accept(StreamTranslateEvent.chunk(textId, paragraph, paragraph, (int) (((i + 1) * 100.0) / total)));
                    }
                }

                eventConsumer.accept(StreamTranslateEvent.done());

            } catch (Exception e) {
                log.error("流式文档翻译失败: {}", e.getMessage(), e);
                try {
                    eventConsumer.accept(StreamTranslateEvent.error("翻译失败：" + e.getMessage()));
                    eventConsumer.accept(StreamTranslateEvent.done());
                } catch (IOException ex) {
                    log.error("发送错误事件失败", ex);
                }
            }
        });
    }

    /**
     * 翻译单个段落（两个流式方法共用）
     */
    private String translateParagraph(String paragraph, String targetLang, String mode, Long userId, String docId) throws Exception {
        String userLevel = userRepositoryPort.findById(userId)
                .map(com.yumu.noveltranslator.domain.model.User::getUserLevel)
                .orElse(null);
        TranslationPipeline pipeline = new TranslationPipeline(
                cachePort, ragTranslationService, entityConsistencyService,
                translationClientPort, postProcessingService, userId, userLevel, docId);
        String result;
        if ("expert".equals(mode)) {
            result = pipeline.execute(paragraph, targetLang, TranslationMode.EXPERT);
        } else {
            result = pipeline.executeFast(paragraph, targetLang, TranslationMode.FAST);
        }
        String translation = ExternalResponseUtil.extractDataField(result);
        return translation != null ? translation : paragraph;
    }

    /**
     * 恢复原文中的末尾换行符到译文中
     */
    private static String restoreTrailingNewline(String translation, String original) {
        if (!translation.endsWith("\n") && !translation.endsWith("\r")) {
            String trailing = getTrailingNewline(original);
            if (!trailing.isEmpty()) {
                translation += trailing;
            }
        }
        return translation;
    }

    /**
     * 提取字符串末尾的所有换行符
     * 用于在翻译后恢复原文的格式
     */
    private static String getTrailingNewline(String text) {
        if (text == null || text.isEmpty()) {
            return "";
        }
        int end = text.length();
        while (end > 0) {
            char c = text.charAt(end - 1);
            if (c != '\n' && c != '\r') {
                break;
            }
            end--;
        }
        return text.substring(end);
    }

    /**
     * 定时清理卡死的任务（每 5 分钟执行一次）
     * 将超过 30 分钟仍处于 PENDING/PROCESSING 状态的任务标记为 FAILED
     */
    @Scheduled(fixedRate = 300000)
    public void cleanupStuckTasks() {
        LocalDateTime cutoff = LocalDateTime.now().minusMinutes(30);
        List<TranslationTask> stuckTasks = translationPort.findTasksByStatusAndCreateTimeBefore(
                TranslationStatus.PROCESSING.getValue(), cutoff);
        for (TranslationTask task : stuckTasks) {
            log.warn("清理卡死任务: taskId={}, createTime={}", task.getTaskId(), task.getCreateTime());
            task.setStatus(TranslationStatus.FAILED.getValue());
            task.setErrorMessage("任务超时，自动标记为失败");
            translationPort.updateTask(task);
            // 同步更新文档状态
            if (task.getDocumentId() != null) {
                Document doc = documentPort.findById(task.getDocumentId()).orElse(null);
                if (doc != null && TranslationStatus.PROCESSING.getValue().equals(doc.getStatus())) {
                    doc.setStatus(TranslationStatus.FAILED.getValue());
                    doc.setErrorMessage("任务超时，自动标记为失败");
                    doc.setUpdateTime(LocalDateTime.now());
                    documentPort.update(doc);
                }
            }
        }
    }

    /**
     * 应用关闭时优雅关闭异步翻译线程
     */
    @PreDestroy
    public void shutdown() {
        log.info("TranslationTaskApplicationService 关闭中，等待异步翻译任务完成...");
        // 虚拟线程由 JVM 自动管理，无需显式关闭
        log.info("TranslationTaskApplicationService 已关闭");
    }
}
