package tech.easyflow.ai.documentimport.task;

import com.alibaba.fastjson2.JSON;
import com.easyagents.rag.core.RagDefaults;
import com.easyagents.rag.ingestion.model.AnalysisResult;
import com.easyagents.rag.ingestion.model.StrategyConfig;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.springframework.stereotype.Service;
import tech.easyflow.ai.document.support.DocumentInputStreamSupport;
import tech.easyflow.ai.documentimport.DocumentImportDtos;
import tech.easyflow.ai.documentimport.DocumentImportKeys;
import tech.easyflow.ai.easyagents.CustomMultipartFile;
import tech.easyflow.ai.entity.DocumentChunk;
import tech.easyflow.common.filestorage.FileStorageService;
import tech.easyflow.common.util.StringUtil;

import javax.annotation.Resource;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.PushbackInputStream;
import java.io.Reader;
import java.io.UncheckedIOException;
import java.math.BigInteger;
import java.nio.charset.CharacterCodingException;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.nio.charset.CodingErrorAction;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.HexFormat;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;

/**
 * CSV 表格流式解析与行分片服务。
 *
 * <p>原文件在解析阶段按逻辑记录迭代，行数据以最多 4 MiB 的 JSON 分片写入对象存储，
 * 清单最后写入作为提交标记。后续调整 {@code rowsPerChunk} 时直接复用表格快照。</p>
 *
 * @author Codex
 * @since 2026-08-04
 */
@Service
public class CsvTableSnapshotService {

    public static final String FAILURE_INVALID_ENCODING = "csv_invalid_encoding";
    public static final String FAILURE_MALFORMED = "csv_malformed";
    public static final String FAILURE_COLUMN_MISMATCH = "csv_column_mismatch";
    public static final String FAILURE_STRUCTURE_LIMIT = "csv_structure_limit_exceeded";
    public static final String FAILURE_SNAPSHOT_CORRUPTED = "csv_snapshot_corrupted";
    public static final String FAILURE_SNAPSHOT_STORAGE = "csv_snapshot_storage_failed";

    private static final String SNAPSHOT_VERSION = "csv-table-v1";
    private static final Charset GB18030 = Charset.forName("GB18030");
    private static final int MAX_COLUMNS = 1_024;
    private static final int MAX_FIELD_CHARS = 1_048_576;
    private static final int MAX_RECORD_UTF8_BYTES = 4 * 1024 * 1024;
    private static final long MAX_PART_BYTES = 4L * 1024L * 1024L;
    private static final long TARGET_PART_BYTES = 3L * 1024L * 1024L;
    private static final long MAX_MANIFEST_BYTES = 4L * 1024L * 1024L;
    private static final int MAX_PARTS = 4_096;
    private static final int MAX_CHUNKS = 200_000;
    private static final int PREVIEW_CHUNK_LIMIT = 50;

    @Resource(name = "default")
    private FileStorageService storageService;

    @Resource
    private DocumentImportChunkSnapshotService chunkSnapshotService;

    private final TabularRowWindowChunkBuilder chunkBuilder =
        new TabularRowWindowChunkBuilder();

    /**
     * 流式解析 CSV 并生成可复用表格快照。
     *
     * @param sourcePath 原文件对象存储路径
     * @param knowledgeId 知识库 ID
     * @param documentId 文档 ID
     * @param generation 当前任务执行代次
     * @return 解析结果
     */
    public CsvParseResult parse(
        String sourcePath,
        BigInteger knowledgeId,
        BigInteger documentId,
        String generation) {
        if (!StringUtil.hasText(sourcePath)
            || knowledgeId == null
            || documentId == null) {
            throw new CsvImportException(
                FAILURE_SNAPSHOT_STORAGE, "CSV 解析缺少文件归属信息");
        }
        String safeGeneration = StringUtil.hasText(generation)
            ? generation.replaceAll("[^A-Za-z0-9_-]", "")
            : UUID.randomUUID().toString();
        ParseAttempt utf8Attempt = new ParseAttempt(
            sourcePath, knowledgeId, documentId, safeGeneration, StandardCharsets.UTF_8);
        try {
            return utf8Attempt.execute();
        } catch (CsvImportException error) {
            if (!utf8Attempt.isEncodingFailure() || utf8Attempt.hasBom()) {
                throw error;
            }
            ParseAttempt gb18030Attempt = new ParseAttempt(
                sourcePath, knowledgeId, documentId, safeGeneration + "-gb18030", GB18030);
            try {
                return gb18030Attempt.execute();
            } catch (CsvImportException fallbackError) {
                if (gb18030Attempt.isEncodingFailure()) {
                    throw new CsvImportException(
                        FAILURE_INVALID_ENCODING,
                        "CSV 编码无法识别，请转换为 UTF-8、UTF-16 或 GB18030 后重试");
                }
                throw fallbackError;
            }
        }
    }

    /**
     * 根据表格快照和行窗口策略生成分块快照。
     *
     * @param document 文档实体
     * @param tableSnapshotPath 表格快照清单路径
     * @param strategyConfig 分块策略
     * @return 只携带预览页和分块快照路径的有界会话
     */
    public DocumentImportDtos.PreviewSession buildChunkSnapshot(
        tech.easyflow.ai.entity.Document document,
        String tableSnapshotPath,
        StrategyConfig strategyConfig) {
        CsvTableManifest manifest = loadManifest(tableSnapshotPath);
        if (!document.getId().equals(manifest.getDocumentId())
            || !document.getCollectionId().equals(manifest.getKnowledgeId())) {
            throw new CsvImportException(
                FAILURE_SNAPSHOT_CORRUPTED, "CSV 表格快照与当前文档不匹配");
        }
        int rowsPerChunk = strategyConfig == null
            || strategyConfig.getRowsPerChunk() == null
            ? 10
            : strategyConfig.getRowsPerChunk();
        if (rowsPerChunk < 1 || rowsPerChunk > 200) {
            throw new CsvImportException(
                FAILURE_STRUCTURE_LIMIT, "CSV 每分块行数必须在 1 到 200 之间");
        }

        DocumentImportDtos.PreviewSession session =
            buildSessionHeader(document, manifest, strategyConfig);
        List<DocumentChunk> previewChunks = new ArrayList<DocumentChunk>();
        int sorting = 1;
        int totalChunks = 0;
        List<TabularRowWindowChunkBuilder.TabularRow> window =
            new ArrayList<TabularRowWindowChunkBuilder.TabularRow>(rowsPerChunk);
        long windowChars = chunkBuilder.estimateHeaderChars(
            document.getTitle(), manifest.getNormalizedHeaders());
        long windowTokens = chunkBuilder.estimateHeaderTokens(
            document.getTitle(), manifest.getNormalizedHeaders());
        long headerChars = windowChars;
        long headerTokens = windowTokens;
        try (DocumentImportChunkSnapshotService.SnapshotWriter writer =
                 chunkSnapshotService.createWriter(session)) {
            for (CsvTablePart part : manifest.getParts()) {
                for (CsvRow row : readRows(part)) {
                    TabularRowWindowChunkBuilder.TabularRow tabularRow =
                        new TabularRowWindowChunkBuilder.TabularRow(
                            row.getRowNumber(), row.getValues());
                    int rowChars = chunkBuilder.estimateRowChars(
                        manifest.getNormalizedHeaders(), tabularRow);
                    int rowTokens = chunkBuilder.estimateRowTokens(
                        manifest.getNormalizedHeaders(), tabularRow);
                    if (!window.isEmpty()
                        && exceedsWindowLimit(
                            windowChars + rowChars,
                            windowTokens + rowTokens)) {
                        List<DocumentChunk> chunks = appendWindow(
                            writer, previewChunks, document, manifest,
                            window, sorting);
                        sorting += chunks.size();
                        totalChunks += chunks.size();
                        assertChunkLimit(totalChunks);
                        window.clear();
                        windowChars = headerChars;
                        windowTokens = headerTokens;
                    }
                    window.add(tabularRow);
                    windowChars += rowChars;
                    windowTokens += rowTokens;
                    if (window.size() >= rowsPerChunk
                        || exceedsWindowLimit(windowChars, windowTokens)) {
                        List<DocumentChunk> chunks = appendWindow(
                            writer, previewChunks, document, manifest,
                            window, sorting);
                        sorting += chunks.size();
                        totalChunks += chunks.size();
                        assertChunkLimit(totalChunks);
                        window.clear();
                        windowChars = headerChars;
                        windowTokens = headerTokens;
                    }
                }
            }
            if (!window.isEmpty() || manifest.getTotalRows() == 0) {
                List<DocumentChunk> chunks = appendWindow(
                    writer, previewChunks, document, manifest,
                    window, sorting);
                totalChunks += chunks.size();
                assertChunkLimit(totalChunks);
            }
            String chunkSnapshotPath = writer.finish();
            session.setChunkSnapshotPath(chunkSnapshotPath);
            session.setTotalChunks(totalChunks);
            session.setDocumentChunks(previewChunks);
            return session;
        } catch (CsvImportException error) {
            throw error;
        } catch (RuntimeException error) {
            throw new CsvImportException(
                FAILURE_SNAPSHOT_STORAGE,
                "CSV 分块快照保存失败，请重试");
        }
    }

    /**
     * 判断当前窗口是否达到字符或 Token 安全上限。
     *
     * @param chars 渲染字符估算
     * @param tokens Token 估算
     * @return 是否达到上限
     */
    private boolean exceedsWindowLimit(
        long chars,
        long tokens) {
        return chars > TabularRowWindowChunkBuilder.MAX_CHUNK_CONTENT_CHARS
            || tokens > RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT;
    }

    /**
     * 构建并写入一个有界表格行窗口。
     *
     * @param writer 分块快照写入器
     * @param previewChunks 有界预览集合
     * @param document 文档
     * @param manifest CSV 表格清单
     * @param window 当前行窗口
     * @param sorting 起始排序号
     * @return 已写入分块
     */
    private List<DocumentChunk> appendWindow(
        DocumentImportChunkSnapshotService.SnapshotWriter writer,
        List<DocumentChunk> previewChunks,
        tech.easyflow.ai.entity.Document document,
        CsvTableManifest manifest,
        List<TabularRowWindowChunkBuilder.TabularRow> window,
        int sorting) {
        List<DocumentChunk> chunks = chunkBuilder.build(
            document.getId(), document.getCollectionId(),
            document.getTitle(), manifest.getNormalizedHeaders(),
            window, sorting, "TABLE_ROW");
        writer.append(chunks);
        addPreview(previewChunks, chunks);
        return chunks;
    }

    /**
     * 删除 CSV 表格快照及其行分片。
     *
     * @param manifestPath 清单路径
     */
    public void delete(String manifestPath) {
        if (!StringUtil.hasText(manifestPath)) {
            return;
        }
        deleteParts(manifestPath);
        deleteManifest(manifestPath);
    }

    /**
     * 删除 CSV 行分片并保留清单。
     *
     * <p>任一分片删除失败时立即抛出异常，清单继续作为精确重试依据。</p>
     *
     * @param manifestPath CSV 表格清单路径
     */
    public void deleteParts(String manifestPath) {
        if (!StringUtil.hasText(manifestPath)) {
            return;
        }
        CsvTableManifest manifest = loadManifest(manifestPath);
        for (CsvTablePart part : manifest.getParts()) {
            if (StringUtil.hasText(part.getPath())) {
                storageService.delete(part.getPath());
            }
        }
    }

    /**
     * 删除 CSV 表格清单。
     *
     * @param manifestPath CSV 表格清单路径
     */
    public void deleteManifest(String manifestPath) {
        if (StringUtil.hasText(manifestPath)) {
            storageService.delete(manifestPath);
        }
    }

    /**
     * 构建有界分块会话头。
     *
     * @param document 文档
     * @param manifest CSV 清单
     * @param strategyConfig 策略
     * @return 会话头
     */
    private DocumentImportDtos.PreviewSession buildSessionHeader(
        tech.easyflow.ai.entity.Document document,
        CsvTableManifest manifest,
        StrategyConfig strategyConfig) {
        AnalysisResult analysis = new AnalysisResult();
        analysis.setSourceFormat("csv");
        analysis.setNormalizedContent(null);
        analysis.setRecommendedStrategyCode("TABLE_ROW");
        analysis.setRecommendedStrategyLabel("按表头 / 行窗口");
        analysis.setConfidence(1.0D);
        analysis.getFeatures().put("sourceFormat", "csv");
        analysis.getFeatures().put("rowCount", manifest.getTotalRows());
        analysis.getFeatures().put("columnCount", manifest.getColumnCount());
        analysis.getFeatures().put("encoding", manifest.getEncoding());
        analysis.getFeatures().put(
            "rowsPerChunk",
            strategyConfig == null ? 10 : strategyConfig.getRowsPerChunk());

        DocumentImportDtos.PreviewSession session =
            new DocumentImportDtos.PreviewSession();
        session.setKnowledgeId(document.getCollectionId());
        session.setDocumentId(document.getId());
        session.setFilePath(document.getDocumentPath());
        session.setFileName(document.getTitle());
        session.setSourceFormat("csv");
        session.setStrategyConfig(strategyConfig);
        session.setAnalysis(analysis);
        session.setDocument(null);
        session.setDocumentChunks(new ArrayList<DocumentChunk>());
        session.setPreviewChunks(new ArrayList<>());
        session.setCreatedAt(new Date());
        return session;
    }

    /**
     * 追加有限数量的预览分块。
     *
     * @param previewChunks 预览集合
     * @param chunks 新分块
     */
    private void addPreview(
        List<DocumentChunk> previewChunks,
        List<DocumentChunk> chunks) {
        int remaining = PREVIEW_CHUNK_LIMIT - previewChunks.size();
        if (remaining <= 0) {
            return;
        }
        previewChunks.addAll(
            chunks.subList(0, Math.min(remaining, chunks.size())));
    }

    /**
     * 校验分块扩张上限。
     *
     * @param totalChunks 当前总分块数
     */
    private void assertChunkLimit(int totalChunks) {
        if (totalChunks > MAX_CHUNKS) {
            throw new CsvImportException(
                FAILURE_STRUCTURE_LIMIT,
                "CSV 生成分块过多，请增大每分块行数后重试");
        }
    }

    /**
     * 读取并校验 CSV 表格清单。
     *
     * @param manifestPath 清单路径
     * @return 清单
     */
    private CsvTableManifest loadManifest(String manifestPath) {
        if (!StringUtil.hasText(manifestPath)) {
            throw new CsvImportException(
                FAILURE_SNAPSHOT_CORRUPTED, "CSV 表格快照不存在，请重新解析");
        }
        try (InputStream inputStream = storageService.readStream(manifestPath)) {
            byte[] bytes = DocumentInputStreamSupport.readBytes(
                inputStream, MAX_MANIFEST_BYTES);
            CsvTableManifest manifest = JSON.parseObject(bytes, CsvTableManifest.class);
            validateManifest(manifest);
            return manifest;
        } catch (IOException | RuntimeException error) {
            if (error instanceof CsvImportException csvError) {
                throw csvError;
            }
            throw new CsvImportException(
                FAILURE_SNAPSHOT_CORRUPTED, "CSV 表格快照读取失败，请重新解析");
        }
    }

    /**
     * 校验清单字段和计数。
     *
     * @param manifest 清单
     */
    private void validateManifest(CsvTableManifest manifest) {
        if (manifest == null
            || !SNAPSHOT_VERSION.equals(manifest.getVersion())
            || manifest.getKnowledgeId() == null
            || manifest.getDocumentId() == null
            || manifest.getNormalizedHeaders() == null
            || manifest.getNormalizedHeaders().isEmpty()
            || manifest.getParts() == null) {
            throw new CsvImportException(
                FAILURE_SNAPSHOT_CORRUPTED, "CSV 表格快照清单损坏");
        }
        long rows = 0L;
        for (CsvTablePart part : manifest.getParts()) {
            if (part == null || !StringUtil.hasText(part.getPath())
                || part.getRowCount() <= 0
                || part.getByteLength() <= 0
                || !StringUtil.hasText(part.getSha256())) {
                throw new CsvImportException(
                    FAILURE_SNAPSHOT_CORRUPTED, "CSV 表格快照清单损坏");
            }
            rows += part.getRowCount();
        }
        if (rows != manifest.getTotalRows()) {
            throw new CsvImportException(
                FAILURE_SNAPSHOT_CORRUPTED, "CSV 表格快照行数不一致");
        }
    }

    /**
     * 读取并校验一个 CSV 行分片。
     *
     * @param part 分片清单
     * @return 行数据
     */
    private List<CsvRow> readRows(CsvTablePart part) {
        try (InputStream inputStream = storageService.readStream(part.getPath())) {
            byte[] bytes = DocumentInputStreamSupport.readBytes(
                inputStream, MAX_PART_BYTES);
            if (bytes.length != part.getByteLength()
                || !sha256(bytes).equals(part.getSha256())) {
                throw new CsvImportException(
                    FAILURE_SNAPSHOT_CORRUPTED, "CSV 行分片校验失败");
            }
            List<CsvRow> rows = JSON.parseArray(
                new String(bytes, StandardCharsets.UTF_8), CsvRow.class);
            if (rows == null || rows.size() != part.getRowCount()) {
                throw new CsvImportException(
                    FAILURE_SNAPSHOT_CORRUPTED, "CSV 行分片内容损坏");
            }
            return rows;
        } catch (IOException error) {
            throw new CsvImportException(
                FAILURE_SNAPSHOT_CORRUPTED, "CSV 行分片读取失败");
        }
    }

    /**
     * 计算字节数组的 SHA-256。
     *
     * @param bytes 字节数组
     * @return 十六进制摘要
     */
    private String sha256(byte[] bytes) {
        try {
            return HexFormat.of().formatHex(
                MessageDigest.getInstance("SHA-256").digest(bytes));
        } catch (NoSuchAlgorithmException error) {
            throw new IllegalStateException("当前运行环境不支持 SHA-256", error);
        }
    }

    /**
     * 判断异常链中是否包含字符解码失败。
     *
     * @param error 异常
     * @return 是否为编码失败
     */
    private boolean isCharacterCodingFailure(Throwable error) {
        Throwable cursor = error;
        while (cursor != null) {
            if (cursor instanceof CharacterCodingException) {
                return true;
            }
            cursor = cursor.getCause();
        }
        return false;
    }

    /**
     * 单次编码解析尝试。
     */
    private final class ParseAttempt {

        private final String sourcePath;
        private final BigInteger knowledgeId;
        private final BigInteger documentId;
        private final String generation;
        private final Charset fallbackCharset;
        private final List<String> storedPaths = new ArrayList<String>();
        private boolean encodingFailure;
        private boolean bom;

        /**
         * 创建一次解析尝试。
         *
         * @param sourcePath 原文件路径
         * @param knowledgeId 知识库 ID
         * @param documentId 文档 ID
         * @param generation 代次
         * @param fallbackCharset 无 BOM 时使用的编码
         */
        private ParseAttempt(
            String sourcePath,
            BigInteger knowledgeId,
            BigInteger documentId,
            String generation,
            Charset fallbackCharset) {
            this.sourcePath = sourcePath;
            this.knowledgeId = knowledgeId;
            this.documentId = documentId;
            this.generation = generation;
            this.fallbackCharset = fallbackCharset;
        }

        /**
         * 执行解析。
         *
         * @return 解析结果
         */
        private CsvParseResult execute() {
            CsvTableManifest manifest = new CsvTableManifest();
            manifest.setVersion(SNAPSHOT_VERSION);
            manifest.setKnowledgeId(knowledgeId);
            manifest.setDocumentId(documentId);
            manifest.setCreatedAt(new Date());
            String prefix = "knowledge-import-csv/"
                + knowledgeId + "/" + documentId + "/" + generation;
            List<CsvRow> partRows = new ArrayList<CsvRow>();
            long estimatedPartBytes = 2L;
            try (InputStream rawStream = storageService.readStream(sourcePath);
                 DecodedInput decodedInput = openDecodedInput(rawStream, fallbackCharset);
                 Reader reader = decodedInput.getReader();
                 CSVParser parser = CSVParser.parse(reader, csvFormat())) {
                bom = decodedInput.hasBom();
                manifest.setEncoding(decodedInput.getCharset().name());
                boolean headerRead = false;
                long totalRows = 0L;
                for (CSVRecord record : parser) {
                    if (!headerRead) {
                        List<String> originalHeaders = readValues(record);
                        validateHeader(originalHeaders);
                        manifest.setOriginalHeaders(originalHeaders);
                        manifest.setNormalizedHeaders(normalizeHeaders(originalHeaders));
                        manifest.setColumnCount(originalHeaders.size());
                        headerRead = true;
                        continue;
                    }
                    if (record.size() != manifest.getColumnCount()) {
                        throw new CsvImportException(
                            FAILURE_COLUMN_MISMATCH,
                            "CSV 第 " + record.getRecordNumber()
                                + " 条记录列数为 " + record.size()
                                + "，与表头 " + manifest.getColumnCount() + " 列不一致");
                    }
                    List<String> values = readValues(record);
                    validateRecord(values, record.getRecordNumber());
                    CsvRow row = new CsvRow();
                    row.setRowNumber(toSafeRowNumber(record.getRecordNumber()));
                    row.setValues(values);
                    long estimatedRowBytes = estimateUtf8Bytes(values) + 64L;
                    if (!partRows.isEmpty()
                        && estimatedPartBytes + estimatedRowBytes > TARGET_PART_BYTES) {
                        flushRows(prefix, manifest, partRows);
                        partRows.clear();
                        estimatedPartBytes = 2L;
                    }
                    partRows.add(row);
                    estimatedPartBytes += estimatedRowBytes;
                    totalRows++;
                }
                if (!headerRead) {
                    throw new CsvImportException(
                        FAILURE_MALFORMED, "CSV 文件为空或缺少表头");
                }
                if (!partRows.isEmpty()) {
                    flushRows(prefix, manifest, partRows);
                }
                manifest.setTotalRows(totalRows);
                String manifestPath = writeManifest(prefix, manifest);
                storedPaths.clear();
                return new CsvParseResult(
                    manifestPath,
                    manifest.getEncoding(),
                    manifest.getTotalRows(),
                    manifest.getColumnCount(),
                    manifest.getNormalizedHeaders());
            } catch (CsvImportException error) {
                cleanup();
                throw error;
            } catch (UncheckedIOException | IOException error) {
                cleanup();
                encodingFailure = isCharacterCodingFailure(error);
                if (encodingFailure) {
                    throw new CsvImportException(
                        FAILURE_INVALID_ENCODING, "CSV 编码解析失败");
                }
                throw new CsvImportException(
                    FAILURE_MALFORMED, "CSV 格式不合法：" + safeMessage(error));
            } catch (RuntimeException error) {
                cleanup();
                encodingFailure = isCharacterCodingFailure(error);
                if (encodingFailure) {
                    throw new CsvImportException(
                        FAILURE_INVALID_ENCODING, "CSV 编码解析失败");
                }
                throw new CsvImportException(
                    FAILURE_SNAPSHOT_STORAGE, "CSV 快照写入失败，请重试");
            }
        }

        /**
         * 返回本次是否因编码失败。
         *
         * @return 是否编码失败
         */
        private boolean isEncodingFailure() {
            return encodingFailure;
        }

        /**
         * 返回原文件是否带 BOM。
         *
         * @return 是否带 BOM
         */
        private boolean hasBom() {
            return bom;
        }

        /**
         * 写入一批行数据。
         *
         * @param prefix 存储目录
         * @param manifest 清单
         * @param rows 行数据
         */
        private void flushRows(
            String prefix,
            CsvTableManifest manifest,
            List<CsvRow> rows) {
            byte[] payload = JSON.toJSONBytes(rows);
            if (payload.length > MAX_PART_BYTES && rows.size() > 1) {
                int midpoint = rows.size() / 2;
                flushRows(prefix, manifest, new ArrayList<CsvRow>(rows.subList(0, midpoint)));
                flushRows(prefix, manifest, new ArrayList<CsvRow>(rows.subList(midpoint, rows.size())));
                return;
            }
            if (payload.length > MAX_PART_BYTES) {
                throw new CsvImportException(
                    FAILURE_STRUCTURE_LIMIT, "CSV 单条记录超过 4 MiB 上限");
            }
            if (manifest.getParts().size() >= MAX_PARTS) {
                throw new CsvImportException(
                    FAILURE_STRUCTURE_LIMIT, "CSV 行分片数量超过系统上限");
            }
            int partNumber = manifest.getParts().size() + 1;
            String fileName = String.format(Locale.ROOT, "rows-%06d.json", partNumber);
            String path = storageService.save(
                new CustomMultipartFile(payload, fileName, fileName, "application/json"),
                prefix);
            if (!StringUtil.hasText(path)) {
                throw new CsvImportException(
                    FAILURE_SNAPSHOT_STORAGE, "CSV 行分片保存失败");
            }
            storedPaths.add(path);
            CsvTablePart part = new CsvTablePart();
            part.setPath(path);
            part.setRowCount(rows.size());
            part.setByteLength(payload.length);
            part.setSha256(sha256(payload));
            manifest.getParts().add(part);
        }

        /**
         * 最后写入表格清单。
         *
         * @param prefix 存储目录
         * @param manifest 清单
         * @return 清单路径
         */
        private String writeManifest(
            String prefix,
            CsvTableManifest manifest) {
            byte[] payload = JSON.toJSONBytes(manifest);
            if (payload.length > MAX_MANIFEST_BYTES) {
                throw new CsvImportException(
                    FAILURE_STRUCTURE_LIMIT, "CSV 表格快照清单超过 4 MiB 上限");
            }
            String fileName = documentId + "-csv-manifest.json";
            String path = storageService.save(
                new CustomMultipartFile(payload, fileName, fileName, "application/json"),
                prefix);
            if (!StringUtil.hasText(path)) {
                throw new CsvImportException(
                    FAILURE_SNAPSHOT_STORAGE, "CSV 表格快照清单保存失败");
            }
            return path;
        }

        /**
         * 清理未提交的行分片。
         */
        private void cleanup() {
            for (String path : storedPaths) {
                storageService.delete(path);
            }
            storedPaths.clear();
        }
    }

    /**
     * 构造严格 RFC4180 格式。
     *
     * @return CSV 格式
     */
    private CSVFormat csvFormat() {
        return CSVFormat.RFC4180.builder()
            .setIgnoreEmptyLines(true)
            .setLenientEof(false)
            .setTrailingData(false)
            .get();
    }

    /**
     * 打开带 BOM 检测和严格解码的字符流。
     *
     * @param rawStream 原始流
     * @param fallbackCharset 无 BOM 编码
     * @return 解码输入
     * @throws IOException 读取 BOM 失败
     */
    private DecodedInput openDecodedInput(
        InputStream rawStream,
        Charset fallbackCharset) throws IOException {
        PushbackInputStream pushback = new PushbackInputStream(rawStream, 3);
        byte[] prefix = new byte[3];
        int count = pushback.read(prefix);
        Charset charset = fallbackCharset;
        int bomLength = 0;
        if (count >= 3
            && (prefix[0] & 0xFF) == 0xEF
            && (prefix[1] & 0xFF) == 0xBB
            && (prefix[2] & 0xFF) == 0xBF) {
            charset = StandardCharsets.UTF_8;
            bomLength = 3;
        } else if (count >= 2
            && (prefix[0] & 0xFF) == 0xFF
            && (prefix[1] & 0xFF) == 0xFE) {
            charset = StandardCharsets.UTF_16LE;
            bomLength = 2;
        } else if (count >= 2
            && (prefix[0] & 0xFF) == 0xFE
            && (prefix[1] & 0xFF) == 0xFF) {
            charset = StandardCharsets.UTF_16BE;
            bomLength = 2;
        }
        if (count > bomLength) {
            pushback.unread(prefix, bomLength, count - bomLength);
        }
        CharsetDecoder decoder = charset.newDecoder()
            .onMalformedInput(CodingErrorAction.REPORT)
            .onUnmappableCharacter(CodingErrorAction.REPORT);
        return new DecodedInput(
            new InputStreamReader(pushback, decoder), charset, bomLength > 0);
    }

    /**
     * 读取一条 CSV 记录的全部值。
     *
     * @param record CSV 记录
     * @return 字段值
     */
    private List<String> readValues(CSVRecord record) {
        List<String> values = new ArrayList<String>(record.size());
        for (int index = 0; index < record.size(); index++) {
            values.add(record.get(index));
        }
        return values;
    }

    /**
     * 校验表头结构。
     *
     * @param headers 原始表头
     */
    private void validateHeader(List<String> headers) {
        if (headers == null || headers.isEmpty()) {
            throw new CsvImportException(FAILURE_MALFORMED, "CSV 缺少表头");
        }
        if (headers.size() > MAX_COLUMNS) {
            throw new CsvImportException(
                FAILURE_STRUCTURE_LIMIT,
                "CSV 列数超过 " + MAX_COLUMNS + " 列上限");
        }
        validateRecord(headers, 1L);
    }

    /**
     * 校验记录字段和 UTF-8 展开大小。
     *
     * @param values 字段
     * @param recordNumber 逻辑记录号
     */
    private void validateRecord(
        List<String> values,
        long recordNumber) {
        for (String value : values) {
            if (value != null && value.indexOf('\0') >= 0) {
                throw new CsvImportException(
                    FAILURE_MALFORMED,
                    "CSV 第 " + recordNumber + " 条记录包含 NUL 字节，文件可能不是文本 CSV");
            }
            if (value != null && value.length() > MAX_FIELD_CHARS) {
                throw new CsvImportException(
                    FAILURE_STRUCTURE_LIMIT,
                    "CSV 第 " + recordNumber + " 条记录存在超过 1 Mi 字符的字段");
            }
        }
        if (estimateUtf8Bytes(values) > MAX_RECORD_UTF8_BYTES) {
            throw new CsvImportException(
                FAILURE_STRUCTURE_LIMIT,
                "CSV 第 " + recordNumber + " 条记录超过 4 MiB 上限");
        }
    }

    /**
     * 规范化空表头和重复表头。
     *
     * @param originalHeaders 原始表头
     * @return 稳定规范化表头
     */
    private List<String> normalizeHeaders(List<String> originalHeaders) {
        List<String> normalized = new ArrayList<String>(originalHeaders.size());
        Map<String, Integer> occurrences = new HashMap<String, Integer>();
        for (int index = 0; index < originalHeaders.size(); index++) {
            String raw = originalHeaders.get(index);
            String base = StringUtil.hasText(raw) ? raw.trim() : "列_" + (index + 1);
            int count = occurrences.getOrDefault(base, 0) + 1;
            occurrences.put(base, count);
            normalized.add(count == 1 ? base : base + "_" + count);
        }
        return normalized;
    }

    /**
     * 估算字段按 UTF-8 输出后的字节数。
     *
     * @param values 字段值
     * @return UTF-8 字节数
     */
    private long estimateUtf8Bytes(List<String> values) {
        long total = 0L;
        for (String value : values) {
            if (value != null) {
                total += value.getBytes(StandardCharsets.UTF_8).length;
            }
            total++;
        }
        return total;
    }

    /**
     * 将长记录号转换为有界整数。
     *
     * @param recordNumber 记录号
     * @return 整数记录号
     */
    private int toSafeRowNumber(long recordNumber) {
        if (recordNumber > Integer.MAX_VALUE) {
            throw new CsvImportException(
                FAILURE_STRUCTURE_LIMIT, "CSV 记录数量超过系统上限");
        }
        return (int) recordNumber;
    }

    /**
     * 返回安全异常摘要。
     *
     * @param error 异常
     * @return 摘要
     */
    private String safeMessage(Throwable error) {
        String message = error.getMessage();
        return StringUtil.hasText(message) ? message : "无法解析逻辑记录";
    }

    /**
     * 解码输入包装。
     */
    private static final class DecodedInput implements AutoCloseable {

        private final Reader reader;
        private final Charset charset;
        private final boolean bom;

        /**
         * 创建解码输入。
         *
         * @param reader 字符流
         * @param charset 字符集
         * @param bom 是否带 BOM
         */
        private DecodedInput(Reader reader, Charset charset, boolean bom) {
            this.reader = reader;
            this.charset = charset;
            this.bom = bom;
        }

        /**
         * 返回字符流。
         *
         * @return 字符流
         */
        private Reader getReader() {
            return reader;
        }

        /**
         * 返回字符集。
         *
         * @return 字符集
         */
        private Charset getCharset() {
            return charset;
        }

        /**
         * 返回是否带 BOM。
         *
         * @return 是否带 BOM
         */
        private boolean hasBom() {
            return bom;
        }

        /**
         * 关闭字符流。
         *
         * @throws IOException 关闭失败
         */
        @Override
        public void close() throws IOException {
            reader.close();
        }
    }

    /**
     * CSV 解析结果。
     */
    public static class CsvParseResult {

        private final String manifestPath;
        private final String encoding;
        private final long rowCount;
        private final int columnCount;
        private final List<String> headers;

        /**
         * 创建解析结果。
         *
         * @param manifestPath 清单路径
         * @param encoding 编码
         * @param rowCount 数据行数
         * @param columnCount 列数
         * @param headers 规范化表头
         */
        public CsvParseResult(
            String manifestPath,
            String encoding,
            long rowCount,
            int columnCount,
            List<String> headers) {
            this.manifestPath = manifestPath;
            this.encoding = encoding;
            this.rowCount = rowCount;
            this.columnCount = columnCount;
            this.headers = headers == null
                ? Collections.<String>emptyList()
                : new ArrayList<String>(headers);
        }

        /**
         * 返回清单路径。
         *
         * @return 清单路径
         */
        public String getManifestPath() {
            return manifestPath;
        }

        /**
         * 返回编码。
         *
         * @return 编码
         */
        public String getEncoding() {
            return encoding;
        }

        /**
         * 返回数据行数。
         *
         * @return 数据行数
         */
        public long getRowCount() {
            return rowCount;
        }

        /**
         * 返回列数。
         *
         * @return 列数
         */
        public int getColumnCount() {
            return columnCount;
        }

        /**
         * 返回规范化表头。
         *
         * @return 表头副本
         */
        public List<String> getHeaders() {
            return new ArrayList<String>(headers);
        }

        /**
         * 构造写入文档正文的有界摘要。
         *
         * @return 摘要
         */
        public String buildSummary() {
            String headerSummary = String.join("、", headers);
            if (headerSummary.length() > 2_000) {
                headerSummary = headerSummary.substring(0, 2_000) + "…";
            }
            return "CSV 表格，共 " + columnCount + " 列、" + rowCount
                + " 行数据。列：" + headerSummary;
        }

        /**
         * 构造有界解析元信息。
         *
         * @return 元信息
         */
        public Map<String, Object> toMetadata() {
            Map<String, Object> metadata = new LinkedHashMap<String, Object>();
            metadata.put("sourceFormat", "csv");
            metadata.put(DocumentImportKeys.KEY_DOCUMENT_CSV_TABLE_SNAPSHOT_PATH, manifestPath);
            metadata.put(DocumentImportKeys.KEY_DOCUMENT_CSV_ENCODING, encoding);
            metadata.put(DocumentImportKeys.KEY_DOCUMENT_CSV_ROW_COUNT, rowCount);
            metadata.put(DocumentImportKeys.KEY_DOCUMENT_CSV_COLUMN_COUNT, columnCount);
            return metadata;
        }
    }

    /**
     * CSV 表格快照清单。
     */
    public static class CsvTableManifest {

        private String version;
        private BigInteger knowledgeId;
        private BigInteger documentId;
        private String encoding;
        private List<String> originalHeaders = new ArrayList<String>();
        private List<String> normalizedHeaders = new ArrayList<String>();
        private int columnCount;
        private long totalRows;
        private List<CsvTablePart> parts = new ArrayList<CsvTablePart>();
        private Date createdAt;

        /**
         * 返回清单版本。
         *
         * @return 版本
         */
        public String getVersion() {
            return version;
        }

        /**
         * 设置清单版本。
         *
         * @param version 版本
         */
        public void setVersion(String version) {
            this.version = version;
        }

        /**
         * 返回知识库 ID。
         *
         * @return 知识库 ID
         */
        public BigInteger getKnowledgeId() {
            return knowledgeId;
        }

        /**
         * 设置知识库 ID。
         *
         * @param knowledgeId 知识库 ID
         */
        public void setKnowledgeId(BigInteger knowledgeId) {
            this.knowledgeId = knowledgeId;
        }

        /**
         * 返回文档 ID。
         *
         * @return 文档 ID
         */
        public BigInteger getDocumentId() {
            return documentId;
        }

        /**
         * 设置文档 ID。
         *
         * @param documentId 文档 ID
         */
        public void setDocumentId(BigInteger documentId) {
            this.documentId = documentId;
        }

        /**
         * 返回源文件编码。
         *
         * @return 编码名称
         */
        public String getEncoding() {
            return encoding;
        }

        /**
         * 设置源文件编码。
         *
         * @param encoding 编码名称
         */
        public void setEncoding(String encoding) {
            this.encoding = encoding;
        }

        /**
         * 返回原始表头。
         *
         * @return 原始表头
         */
        public List<String> getOriginalHeaders() {
            return originalHeaders;
        }

        /**
         * 设置原始表头。
         *
         * @param originalHeaders 原始表头
         */
        public void setOriginalHeaders(List<String> originalHeaders) {
            this.originalHeaders = originalHeaders;
        }

        /**
         * 返回规范化表头。
         *
         * @return 规范化表头
         */
        public List<String> getNormalizedHeaders() {
            return normalizedHeaders;
        }

        /**
         * 设置规范化表头。
         *
         * @param normalizedHeaders 规范化表头
         */
        public void setNormalizedHeaders(List<String> normalizedHeaders) {
            this.normalizedHeaders = normalizedHeaders;
        }

        /**
         * 返回列数。
         *
         * @return 列数
         */
        public int getColumnCount() {
            return columnCount;
        }

        /**
         * 设置列数。
         *
         * @param columnCount 列数
         */
        public void setColumnCount(int columnCount) {
            this.columnCount = columnCount;
        }

        /**
         * 返回数据行数。
         *
         * @return 数据行数
         */
        public long getTotalRows() {
            return totalRows;
        }

        /**
         * 设置数据行数。
         *
         * @param totalRows 数据行数
         */
        public void setTotalRows(long totalRows) {
            this.totalRows = totalRows;
        }

        /**
         * 返回行分片清单。
         *
         * @return 行分片清单
         */
        public List<CsvTablePart> getParts() {
            return parts;
        }

        /**
         * 设置行分片清单。
         *
         * @param parts 行分片清单
         */
        public void setParts(List<CsvTablePart> parts) {
            this.parts = parts;
        }

        /**
         * 返回创建时间。
         *
         * @return 创建时间
         */
        public Date getCreatedAt() {
            return createdAt;
        }

        /**
         * 设置创建时间。
         *
         * @param createdAt 创建时间
         */
        public void setCreatedAt(Date createdAt) {
            this.createdAt = createdAt;
        }
    }

    /**
     * CSV 行分片元信息。
     */
    public static class CsvTablePart {

        private String path;
        private int rowCount;
        private long byteLength;
        private String sha256;

        /**
         * 返回存储路径。
         *
         * @return 存储路径
         */
        public String getPath() {
            return path;
        }

        /**
         * 设置存储路径。
         *
         * @param path 存储路径
         */
        public void setPath(String path) {
            this.path = path;
        }

        /**
         * 返回分片行数。
         *
         * @return 行数
         */
        public int getRowCount() {
            return rowCount;
        }

        /**
         * 设置分片行数。
         *
         * @param rowCount 行数
         */
        public void setRowCount(int rowCount) {
            this.rowCount = rowCount;
        }

        /**
         * 返回分片字节数。
         *
         * @return 字节数
         */
        public long getByteLength() {
            return byteLength;
        }

        /**
         * 设置分片字节数。
         *
         * @param byteLength 字节数
         */
        public void setByteLength(long byteLength) {
            this.byteLength = byteLength;
        }

        /**
         * 返回分片 SHA-256。
         *
         * @return SHA-256
         */
        public String getSha256() {
            return sha256;
        }

        /**
         * 设置分片 SHA-256。
         *
         * @param sha256 SHA-256
         */
        public void setSha256(String sha256) {
            this.sha256 = sha256;
        }
    }

    /**
     * CSV 逻辑数据行。
     */
    public static class CsvRow {

        private int rowNumber;
        private List<String> values = new ArrayList<String>();

        /**
         * 返回逻辑记录号。
         *
         * @return 逻辑记录号
         */
        public int getRowNumber() {
            return rowNumber;
        }

        /**
         * 设置逻辑记录号。
         *
         * @param rowNumber 逻辑记录号
         */
        public void setRowNumber(int rowNumber) {
            this.rowNumber = rowNumber;
        }

        /**
         * 返回字段值。
         *
         * @return 字段值
         */
        public List<String> getValues() {
            return values;
        }

        /**
         * 设置字段值。
         *
         * @param values 字段值
         */
        public void setValues(List<String> values) {
            this.values = values;
        }
    }
}
