SUPPORTED_EXTENSIONS =
- Set.of("txt", "pdf", "docx", "md", "pptx", "xlsx");
+ DocumentImportFormatPolicy.supportedExtensions();
private static final Duration BATCH_MUTATION_LOCK_LEASE = Duration.ofMinutes(30);
private final DocumentImportBatchService batchService;
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotService.java
index c5166d7b..251f90dc 100644
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotService.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotService.java
@@ -1,10 +1,12 @@
package tech.easyflow.ai.documentimport.task;
import com.alibaba.fastjson2.JSON;
+import com.easyagents.rag.ingestion.model.AnalysisResult;
import org.springframework.stereotype.Service;
import tech.easyflow.ai.document.support.DocumentInputStreamSupport;
import tech.easyflow.ai.documentimport.DocumentImportDtos;
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 tech.easyflow.common.web.exceptions.BusinessException;
@@ -13,11 +15,20 @@ import javax.annotation.Resource;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.UUID;
+import java.util.function.Consumer;
/**
* 自动导入分块快照持久化服务。
*
- * 快照写入对象存储，向量化任务只保存稳定路径，避免依赖短期预览缓存。
+ * V2 快照使用“有界分片 + 最后写入清单”的提交协议，索引任务可以按批读取，
+ * 同时保留对历史 V1 单 JSON 快照的读取兼容。
*
* @author Codex
* @since 2026-07-31
@@ -25,7 +36,11 @@ import java.nio.charset.StandardCharsets;
@Service
public class DocumentImportChunkSnapshotService {
- private static final long MAX_SNAPSHOT_BYTES = 256L * 1024L * 1024L;
+ private static final String SNAPSHOT_VERSION_V2 = "v2";
+ private static final int DEFAULT_PART_CHUNK_COUNT = 128;
+ private static final long MAX_MANIFEST_BYTES = 4L * 1024L * 1024L;
+ private static final long MAX_PART_BYTES = 4L * 1024L * 1024L;
+ private static final long MAX_LEGACY_SNAPSHOT_BYTES = 256L * 1024L * 1024L;
@Resource(name = "default")
private FileStorageService storageService;
@@ -34,61 +49,727 @@ public class DocumentImportChunkSnapshotService {
* 持久化预览会话及其最终分块。
*
* @param session 预览会话
- * @return 快照存储路径
+ * @return V2 清单存储路径
*/
public String save(DocumentImportDtos.PreviewSession session) {
- if (session == null || session.getKnowledgeId() == null || session.getDocumentId() == null
- || session.getDocumentChunks() == null || session.getDocumentChunks().isEmpty()) {
- throw new BusinessException("分块快照内容不完整");
+ validateSession(session, true);
+ try (SnapshotWriter writer = createWriter(session)) {
+ List chunks = session.getDocumentChunks();
+ for (int start = 0; start < chunks.size(); start += DEFAULT_PART_CHUNK_COUNT) {
+ int end = Math.min(start + DEFAULT_PART_CHUNK_COUNT, chunks.size());
+ writer.append(new ArrayList(chunks.subList(start, end)));
+ }
+ return writer.finish();
}
- String fileName = session.getDocumentId() + "-chunks.json";
- byte[] payload = JSON.toJSONBytes(session);
- if (payload.length > MAX_SNAPSHOT_BYTES) {
- throw new BusinessException("分块快照过大，请调整文档后重试");
- }
- CustomMultipartFile file = new CustomMultipartFile(
- payload, fileName, fileName, "application/json");
- String path = storageService.save(file,
- "knowledge-import-snapshots/" + session.getKnowledgeId() + "/" + session.getDocumentId());
- if (!StringUtil.hasText(path)) {
- throw new BusinessException("分块快照保存失败");
- }
- return path;
}
/**
- * 从稳定存储恢复预览会话。
+ * 创建流式快照写入器。
+ *
+ * @param session 仅需包含归属、策略和预览元信息的会话
+ * @return 快照写入器
+ */
+ public SnapshotWriter createWriter(DocumentImportDtos.PreviewSession session) {
+ validateSession(session, false);
+ return new SnapshotWriter(copyHeader(session));
+ }
+
+ /**
+ * 从稳定存储完整恢复预览会话。
+ *
+ * 该方法用于兼容现有调用和小型预览。大文件索引应使用
+ * {@link #forEachBatch(String, int, Consumer)}。
*
* @param path 快照路径
* @return 预览会话
*/
public DocumentImportDtos.PreviewSession load(String path) {
- if (!StringUtil.hasText(path)) {
- throw new BusinessException("分块快照不存在，请重试");
+ SnapshotReadResult readResult = readManifestOrLegacy(path);
+ if (readResult.legacySession != null) {
+ return readResult.legacySession;
}
- try (InputStream inputStream = storageService.readStream(path)) {
- byte[] payload = DocumentInputStreamSupport.readBytes(inputStream, MAX_SNAPSHOT_BYTES);
- String json = new String(payload, StandardCharsets.UTF_8);
- DocumentImportDtos.PreviewSession session =
- JSON.parseObject(json, DocumentImportDtos.PreviewSession.class);
- if (session == null || session.getDocumentChunks() == null
- || session.getDocumentChunks().isEmpty()) {
- throw new BusinessException("分块快照无有效内容，请重试");
+ DocumentImportDtos.PreviewSession session = copyHeader(readResult.manifest.getSession());
+ List chunks = new ArrayList();
+ long loadedBytes = 0L;
+ for (SnapshotPart part : readResult.manifest.getParts()) {
+ PartPayload payload = readPart(part);
+ loadedBytes += payload.byteLength;
+ if (loadedBytes > MAX_LEGACY_SNAPSHOT_BYTES) {
+ throw new BusinessException("分块快照过大，请使用分页或批量读取");
}
- return session;
- } catch (IOException error) {
- throw new BusinessException("分块快照读取失败，请重试");
+ chunks.addAll(payload.chunks);
+ }
+ session.setDocumentChunks(chunks);
+ session.setTotalChunks(chunks.size());
+ session.setChunkSnapshotPath(path);
+ return session;
+ }
+
+ /**
+ * 仅读取快照头部元信息。
+ *
+ * @param path 快照路径
+ * @return 不包含完整分块的预览会话
+ */
+ public DocumentImportDtos.PreviewSession loadHeader(String path) {
+ SnapshotReadResult readResult = readManifestOrLegacy(path);
+ if (readResult.legacySession != null) {
+ DocumentImportDtos.PreviewSession header = copyHeader(readResult.legacySession);
+ int total = readResult.legacySession.getDocumentChunks() == null
+ ? 0
+ : readResult.legacySession.getDocumentChunks().size();
+ header.setTotalChunks(total);
+ header.setChunkSnapshotPath(path);
+ return header;
+ }
+ DocumentImportDtos.PreviewSession header = copyHeader(readResult.manifest.getSession());
+ header.setTotalChunks(readResult.manifest.getTotalChunks());
+ header.setChunkSnapshotPath(path);
+ return header;
+ }
+
+ /**
+ * 分页读取快照分块。
+ *
+ * @param path 快照路径
+ * @param offset 零基偏移
+ * @param limit 最大返回数
+ * @return 指定页分块
+ */
+ public List loadPage(String path, int offset, int limit) {
+ if (offset < 0 || limit <= 0) {
+ throw new BusinessException("分块快照分页参数不合法");
+ }
+ SnapshotReadResult readResult = readManifestOrLegacy(path);
+ if (readResult.legacySession != null) {
+ List chunks = readResult.legacySession.getDocumentChunks();
+ if (chunks == null || offset >= chunks.size()) {
+ return new ArrayList();
+ }
+ int end = Math.min(offset + limit, chunks.size());
+ return new ArrayList(chunks.subList(offset, end));
+ }
+ List page = new ArrayList(limit);
+ int skipped = 0;
+ for (SnapshotPart part : readResult.manifest.getParts()) {
+ if (page.size() >= limit) {
+ break;
+ }
+ if (skipped + part.getChunkCount() <= offset) {
+ skipped += part.getChunkCount();
+ continue;
+ }
+ List chunks = readPart(part).chunks;
+ int localStart = Math.max(0, offset - skipped);
+ int localEnd = Math.min(chunks.size(), localStart + limit - page.size());
+ page.addAll(chunks.subList(localStart, localEnd));
+ skipped += chunks.size();
+ }
+ return page;
+ }
+
+ /**
+ * 按固定上限顺序消费快照分块。
+ *
+ * @param path 快照路径
+ * @param batchSize 单批最大分块数
+ * @param consumer 批处理函数
+ */
+ public void forEachBatch(
+ String path,
+ int batchSize,
+ Consumer> consumer) {
+ if (batchSize <= 0 || consumer == null) {
+ throw new IllegalArgumentException("快照批量消费参数不合法");
+ }
+ SnapshotReadResult readResult = readManifestOrLegacy(path);
+ if (readResult.legacySession != null) {
+ consumeInBatches(readResult.legacySession.getDocumentChunks(), batchSize, consumer);
+ return;
+ }
+ for (SnapshotPart part : readResult.manifest.getParts()) {
+ consumeInBatches(readPart(part).chunks, batchSize, consumer);
}
}
/**
- * 删除已完成向量化的分块快照。
+ * 删除快照清单及其全部分片。
*
- * @param path 快照路径
+ * @param path 快照清单或历史 V1 快照路径
*/
public void delete(String path) {
+ if (!StringUtil.hasText(path)) {
+ return;
+ }
+ deleteParts(path);
+ deleteManifest(path);
+ }
+
+ /**
+ * 删除快照分片并保留清单。
+ *
+ * 任一分片删除失败时立即抛出异常，清单继续作为精确重试依据。
+ *
+ * @param path 快照清单或历史 V1 快照路径
+ */
+ public void deleteParts(String path) {
+ if (!StringUtil.hasText(path)) {
+ return;
+ }
+ SnapshotReadResult readResult = readManifestOrLegacy(path);
+ if (readResult.manifest == null) {
+ return;
+ }
+ for (SnapshotPart part : readResult.manifest.getParts()) {
+ if (StringUtil.hasText(part.getPath())) {
+ storageService.delete(part.getPath());
+ }
+ }
+ }
+
+ /**
+ * 删除快照清单或历史 V1 单对象。
+ *
+ * @param path 快照清单或历史 V1 快照路径
+ */
+ public void deleteManifest(String path) {
if (StringUtil.hasText(path)) {
storageService.delete(path);
}
}
+
+ /**
+ * 校验预览会话的最小完整性。
+ *
+ * @param session 预览会话
+ * @param requireChunks 是否要求已包含分块
+ */
+ private void validateSession(
+ DocumentImportDtos.PreviewSession session,
+ boolean requireChunks) {
+ if (session == null || session.getKnowledgeId() == null || session.getDocumentId() == null) {
+ throw new BusinessException("分块快照归属信息不完整");
+ }
+ if (requireChunks
+ && (session.getDocumentChunks() == null || session.getDocumentChunks().isEmpty())) {
+ throw new BusinessException("分块快照内容不完整");
+ }
+ }
+
+ /**
+ * 复制有界会话头，移除完整正文和分块数组。
+ *
+ * @param source 原始会话
+ * @return 会话头
+ */
+ private DocumentImportDtos.PreviewSession copyHeader(
+ DocumentImportDtos.PreviewSession source) {
+ DocumentImportDtos.PreviewSession header = new DocumentImportDtos.PreviewSession();
+ header.setSessionId(source.getSessionId());
+ header.setKnowledgeId(source.getKnowledgeId());
+ header.setDocumentId(source.getDocumentId());
+ header.setFilePath(source.getFilePath());
+ header.setFileName(source.getFileName());
+ header.setSourceFormat(source.getSourceFormat());
+ header.setStrategyConfig(source.getStrategyConfig());
+ if (source.getAnalysis() != null) {
+ AnalysisResult analysis = JSON.parseObject(
+ JSON.toJSONBytes(source.getAnalysis()), AnalysisResult.class);
+ analysis.setNormalizedContent(null);
+ header.setAnalysis(analysis);
+ }
+ header.setDocument(null);
+ header.setDocumentChunks(new ArrayList());
+ header.setPreviewChunks(new ArrayList<>());
+ header.setChunkSnapshotPath(source.getChunkSnapshotPath());
+ header.setTotalChunks(source.getTotalChunks());
+ header.setCreatedAt(source.getCreatedAt());
+ return header;
+ }
+
+ /**
+ * 读取 V2 清单或历史 V1 会话。
+ *
+ * @param path 快照路径
+ * @return 快照读取结果
+ */
+ private SnapshotReadResult readManifestOrLegacy(String path) {
+ if (!StringUtil.hasText(path)) {
+ throw new BusinessException("分块快照不存在，请重试");
+ }
+ byte[] payload;
+ try (InputStream inputStream = storageService.readStream(path)) {
+ payload = DocumentInputStreamSupport.readBytes(
+ inputStream, MAX_LEGACY_SNAPSHOT_BYTES);
+ } catch (IOException error) {
+ throw new BusinessException("分块快照读取失败，请重试");
+ }
+ SnapshotManifest manifest = JSON.parseObject(payload, SnapshotManifest.class);
+ if (manifest != null && SNAPSHOT_VERSION_V2.equals(manifest.getVersion())) {
+ validateManifest(manifest);
+ return SnapshotReadResult.forManifest(manifest);
+ }
+ DocumentImportDtos.PreviewSession session = JSON.parseObject(
+ new String(payload, StandardCharsets.UTF_8),
+ DocumentImportDtos.PreviewSession.class);
+ if (session == null || session.getDocumentChunks() == null
+ || session.getDocumentChunks().isEmpty()) {
+ throw new BusinessException("分块快照无有效内容，请重试");
+ }
+ return SnapshotReadResult.forLegacy(session);
+ }
+
+ /**
+ * 校验 V2 清单完整性。
+ *
+ * @param manifest 快照清单
+ */
+ private void validateManifest(SnapshotManifest manifest) {
+ if (manifest.getSession() == null
+ || manifest.getSession().getKnowledgeId() == null
+ || manifest.getSession().getDocumentId() == null
+ || manifest.getTotalChunks() <= 0
+ || manifest.getParts() == null
+ || manifest.getParts().isEmpty()) {
+ throw new BusinessException("分块快照清单损坏，请重试");
+ }
+ int total = 0;
+ for (SnapshotPart part : manifest.getParts()) {
+ if (part == null || !StringUtil.hasText(part.getPath())
+ || part.getChunkCount() <= 0
+ || part.getByteLength() <= 0
+ || !StringUtil.hasText(part.getSha256())) {
+ throw new BusinessException("分块快照清单损坏，请重试");
+ }
+ total += part.getChunkCount();
+ }
+ if (total != manifest.getTotalChunks()) {
+ throw new BusinessException("分块快照清单计数不一致，请重试");
+ }
+ }
+
+ /**
+ * 读取并校验一个分块分片。
+ *
+ * @param part 分片清单
+ * @return 分片内容
+ */
+ private PartPayload readPart(SnapshotPart part) {
+ try (InputStream inputStream = storageService.readStream(part.getPath())) {
+ byte[] payload = DocumentInputStreamSupport.readBytes(inputStream, MAX_PART_BYTES);
+ if (payload.length != part.getByteLength()
+ || !sha256(payload).equals(part.getSha256())) {
+ throw new BusinessException("分块快照校验失败，请重试");
+ }
+ List chunks = JSON.parseArray(
+ new String(payload, StandardCharsets.UTF_8), DocumentChunk.class);
+ if (chunks == null || chunks.size() != part.getChunkCount()) {
+ throw new BusinessException("分块快照分片损坏，请重试");
+ }
+ return new PartPayload(chunks, payload.length);
+ } catch (IOException error) {
+ throw new BusinessException("分块快照分片读取失败，请重试");
+ }
+ }
+
+ /**
+ * 将列表按固定大小交给消费者。
+ *
+ * @param chunks 分块
+ * @param batchSize 批大小
+ * @param consumer 消费者
+ */
+ private void consumeInBatches(
+ List chunks,
+ int batchSize,
+ Consumer> consumer) {
+ if (chunks == null || chunks.isEmpty()) {
+ return;
+ }
+ for (int start = 0; start < chunks.size(); start += batchSize) {
+ int end = Math.min(start + batchSize, chunks.size());
+ consumer.accept(new ArrayList(chunks.subList(start, end)));
+ }
+ }
+
+ /**
+ * 计算 SHA-256。
+ *
+ * @param payload 字节内容
+ * @return 十六进制摘要
+ */
+ private String sha256(byte[] payload) {
+ try {
+ return HexFormat.of().formatHex(
+ MessageDigest.getInstance("SHA-256").digest(payload));
+ } catch (NoSuchAlgorithmException error) {
+ throw new IllegalStateException("当前运行环境不支持 SHA-256", error);
+ }
+ }
+
+ /**
+ * V2 快照流式写入器。
+ */
+ public final class SnapshotWriter implements AutoCloseable {
+
+ private final SnapshotManifest manifest;
+ private final String storagePrefix;
+ private boolean committed;
+
+ /**
+ * 创建写入器。
+ *
+ * @param session 会话头
+ */
+ private SnapshotWriter(DocumentImportDtos.PreviewSession session) {
+ this.manifest = new SnapshotManifest();
+ this.manifest.setVersion(SNAPSHOT_VERSION_V2);
+ this.manifest.setSession(session);
+ this.manifest.setCreatedAt(new Date());
+ this.storagePrefix = "knowledge-import-snapshots/"
+ + session.getKnowledgeId() + "/"
+ + session.getDocumentId() + "/"
+ + UUID.randomUUID();
+ }
+
+ /**
+ * 追加一批分块，超出单分片上限时自动二分。
+ *
+ * @param chunks 分块
+ */
+ public void append(List chunks) {
+ if (committed) {
+ throw new IllegalStateException("分块快照已提交");
+ }
+ if (chunks == null || chunks.isEmpty()) {
+ return;
+ }
+ writeBoundedPart(new ArrayList(chunks));
+ }
+
+ /**
+ * 最后写入清单并提交快照。
+ *
+ * @return 清单路径
+ */
+ public String finish() {
+ if (committed) {
+ throw new IllegalStateException("分块快照已提交");
+ }
+ if (manifest.getTotalChunks() <= 0 || manifest.getParts().isEmpty()) {
+ throw new BusinessException("分块快照内容不完整");
+ }
+ byte[] payload = JSON.toJSONBytes(manifest);
+ if (payload.length > MAX_MANIFEST_BYTES) {
+ throw new BusinessException("分块快照清单过大，请调整文档后重试");
+ }
+ String fileName = manifest.getSession().getDocumentId() + "-manifest.json";
+ String path = storageService.save(
+ new CustomMultipartFile(payload, fileName, fileName, "application/json"),
+ storagePrefix);
+ if (!StringUtil.hasText(path)) {
+ throw new BusinessException("分块快照清单保存失败");
+ }
+ committed = true;
+ return path;
+ }
+
+ /**
+ * 未提交时删除已经写入的分片。
+ */
+ @Override
+ public void close() {
+ if (committed) {
+ return;
+ }
+ for (SnapshotPart part : manifest.getParts()) {
+ if (StringUtil.hasText(part.getPath())) {
+ storageService.delete(part.getPath());
+ }
+ }
+ }
+
+ /**
+ * 写入一个有界分片。
+ *
+ * @param chunks 分块
+ */
+ private void writeBoundedPart(List chunks) {
+ byte[] payload = JSON.toJSONBytes(chunks);
+ if (payload.length > MAX_PART_BYTES && chunks.size() > 1) {
+ int midpoint = chunks.size() / 2;
+ writeBoundedPart(new ArrayList(chunks.subList(0, midpoint)));
+ writeBoundedPart(new ArrayList(chunks.subList(midpoint, chunks.size())));
+ return;
+ }
+ if (payload.length > MAX_PART_BYTES) {
+ throw new BusinessException("单个分块快照超过 4 MiB，请缩小分块后重试");
+ }
+ int partNumber = manifest.getParts().size() + 1;
+ String fileName = String.format("part-%06d.json", partNumber);
+ String path = storageService.save(
+ new CustomMultipartFile(payload, fileName, fileName, "application/json"),
+ storagePrefix);
+ if (!StringUtil.hasText(path)) {
+ throw new BusinessException("分块快照分片保存失败");
+ }
+ SnapshotPart part = new SnapshotPart();
+ part.setPath(path);
+ part.setChunkCount(chunks.size());
+ part.setByteLength(payload.length);
+ part.setSha256(sha256(payload));
+ manifest.getParts().add(part);
+ manifest.setTotalChunks(manifest.getTotalChunks() + chunks.size());
+ }
+ }
+
+ /**
+ * V2 快照清单。
+ */
+ public static class SnapshotManifest {
+
+ private String version;
+ private DocumentImportDtos.PreviewSession session;
+ private int totalChunks;
+ private List parts = new ArrayList();
+ private Date createdAt;
+
+ /**
+ * 返回版本。
+ *
+ * @return 版本
+ */
+ public String getVersion() {
+ return version;
+ }
+
+ /**
+ * 设置版本。
+ *
+ * @param version 版本
+ */
+ public void setVersion(String version) {
+ this.version = version;
+ }
+
+ /**
+ * 返回会话头。
+ *
+ * @return 会话头
+ */
+ public DocumentImportDtos.PreviewSession getSession() {
+ return session;
+ }
+
+ /**
+ * 设置会话头。
+ *
+ * @param session 会话头
+ */
+ public void setSession(DocumentImportDtos.PreviewSession session) {
+ this.session = session;
+ }
+
+ /**
+ * 返回总分块数。
+ *
+ * @return 总分块数
+ */
+ public int getTotalChunks() {
+ return totalChunks;
+ }
+
+ /**
+ * 设置总分块数。
+ *
+ * @param totalChunks 总分块数
+ */
+ public void setTotalChunks(int totalChunks) {
+ this.totalChunks = totalChunks;
+ }
+
+ /**
+ * 返回分片清单。
+ *
+ * @return 分片清单
+ */
+ public List getParts() {
+ return parts;
+ }
+
+ /**
+ * 设置分片清单。
+ *
+ * @param parts 分片清单
+ */
+ public void setParts(List parts) {
+ this.parts = parts == null
+ ? new ArrayList()
+ : parts;
+ }
+
+ /**
+ * 返回创建时间。
+ *
+ * @return 创建时间
+ */
+ public Date getCreatedAt() {
+ return createdAt;
+ }
+
+ /**
+ * 设置创建时间。
+ *
+ * @param createdAt 创建时间
+ */
+ public void setCreatedAt(Date createdAt) {
+ this.createdAt = createdAt;
+ }
+ }
+
+ /**
+ * V2 快照分片元信息。
+ */
+ public static class SnapshotPart {
+
+ private String path;
+ private int chunkCount;
+ 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 getChunkCount() {
+ return chunkCount;
+ }
+
+ /**
+ * 设置分块数。
+ *
+ * @param chunkCount 分块数
+ */
+ public void setChunkCount(int chunkCount) {
+ this.chunkCount = chunkCount;
+ }
+
+ /**
+ * 返回字节数。
+ *
+ * @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;
+ }
+ }
+
+ /**
+ * 内部读取结果。
+ */
+ private static final class SnapshotReadResult {
+
+ private final SnapshotManifest manifest;
+ private final DocumentImportDtos.PreviewSession legacySession;
+
+ /**
+ * 创建读取结果。
+ *
+ * @param manifest V2 清单
+ * @param legacySession V1 会话
+ */
+ private SnapshotReadResult(
+ SnapshotManifest manifest,
+ DocumentImportDtos.PreviewSession legacySession) {
+ this.manifest = manifest;
+ this.legacySession = legacySession;
+ }
+
+ /**
+ * 创建 V2 读取结果。
+ *
+ * @param manifest V2 清单
+ * @return 读取结果
+ */
+ private static SnapshotReadResult forManifest(SnapshotManifest manifest) {
+ return new SnapshotReadResult(manifest, null);
+ }
+
+ /**
+ * 创建 V1 读取结果。
+ *
+ * @param session V1 会话
+ * @return 读取结果
+ */
+ private static SnapshotReadResult forLegacy(
+ DocumentImportDtos.PreviewSession session) {
+ return new SnapshotReadResult(null, session);
+ }
+ }
+
+ /**
+ * 内部分片载荷。
+ */
+ private static final class PartPayload {
+
+ private final List chunks;
+ private final long byteLength;
+
+ /**
+ * 创建分片载荷。
+ *
+ * @param chunks 分块
+ * @param byteLength 字节数
+ */
+ private PartPayload(List chunks, long byteLength) {
+ this.chunks = chunks;
+ this.byteLength = byteLength;
+ }
+ }
}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportFormatPolicy.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportFormatPolicy.java
new file mode 100644
index 00000000..e40035de
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportFormatPolicy.java
@@ -0,0 +1,40 @@
+package tech.easyflow.ai.documentimport.task;
+
+import java.util.Set;
+
+/**
+ * 知识库文档导入格式统一策略。
+ *
+ * @author Codex
+ * @since 2026-08-04
+ */
+public final class DocumentImportFormatPolicy {
+
+ private static final Set SUPPORTED_EXTENSIONS =
+ Set.of("txt", "pdf", "docx", "md", "pptx", "xlsx", "csv");
+
+ /**
+ * 禁止实例化格式策略工具类。
+ */
+ private DocumentImportFormatPolicy() {
+ }
+
+ /**
+ * 返回只读的支持格式集合。
+ *
+ * @return 支持的文件扩展名
+ */
+ public static Set supportedExtensions() {
+ return SUPPORTED_EXTENSIONS;
+ }
+
+ /**
+ * 判断扩展名是否属于知识库导入支持范围。
+ *
+ * @param extension 已转换为小写的文件扩展名
+ * @return 支持时返回 {@code true}
+ */
+ public static boolean isSupported(String extension) {
+ return SUPPORTED_EXTENSIONS.contains(extension);
+ }
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportSnapshotCleanupMonitor.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportSnapshotCleanupMonitor.java
new file mode 100644
index 00000000..d9954674
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportSnapshotCleanupMonitor.java
@@ -0,0 +1,42 @@
+package tech.easyflow.ai.documentimport.task;
+
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import tech.easyflow.common.cache.DistributedScheduledLock;
+
+/**
+ * 文档导入中间快照清理调度器。
+ *
+ * @author Codex
+ * @since 2026-08-04
+ */
+@Component
+public class DocumentImportSnapshotCleanupMonitor {
+
+ private final DocumentImportSnapshotCleanupService cleanupService;
+
+ /**
+ * 创建快照清理调度器。
+ *
+ * @param cleanupService 快照清理服务
+ */
+ public DocumentImportSnapshotCleanupMonitor(
+ DocumentImportSnapshotCleanupService cleanupService) {
+ this.cleanupService = cleanupService;
+ }
+
+ /**
+ * 周期性重试到期的快照清理记录。
+ */
+ @Scheduled(
+ fixedDelayString =
+ "${easyflow.ai.document-import.snapshot-cleanup-interval:60s}",
+ initialDelayString =
+ "${easyflow.ai.document-import.snapshot-cleanup-initial-delay:60s}")
+ @DistributedScheduledLock(
+ key = "easyflow:schedule:document-import:snapshot-cleanup",
+ leaseSeconds = 300L)
+ public void cleanupPendingSnapshots() {
+ cleanupService.processPendingCleanups();
+ }
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportSnapshotCleanupService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportSnapshotCleanupService.java
new file mode 100644
index 00000000..f2f5e620
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportSnapshotCleanupService.java
@@ -0,0 +1,352 @@
+package tech.easyflow.ai.documentimport.task;
+
+import com.mybatisflex.core.keygen.impl.FlexIDKeyGenerator;
+import com.mybatisflex.core.query.QueryWrapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionSynchronizationManager;
+import tech.easyflow.ai.entity.DocumentImportSnapshotCleanup;
+import tech.easyflow.ai.mapper.DocumentImportSnapshotCleanupMapper;
+import tech.easyflow.ai.mapper.DocumentMapper;
+import tech.easyflow.common.util.StringUtil;
+
+import javax.annotation.Resource;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
+import java.util.Date;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
+
+/**
+ * 文档导入中间快照可靠清理服务。
+ *
+ * 先持久化精确清单路径，再按“分片、清单”两个阶段删除。阶段推进先于
+ * 清单删除落库，确保进程在任一位置中断后都能安全继续。
+ *
+ * @author Codex
+ * @since 2026-08-04
+ */
+@Service
+public class DocumentImportSnapshotCleanupService {
+
+ public static final String TYPE_CHUNK = "CHUNK";
+ public static final String TYPE_CSV_TABLE = "CSV_TABLE";
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(DocumentImportSnapshotCleanupService.class);
+ private static final String PHASE_PARTS_PENDING = "PARTS_PENDING";
+ private static final String PHASE_MANIFEST_PENDING = "MANIFEST_PENDING";
+ private static final int PROCESS_BATCH_SIZE = 50;
+ private static final int MAX_PATH_CHARS = 2_048;
+ private static final int MAX_ERROR_CHARS = 1_024;
+ private static final Duration CLAIM_LEASE = Duration.ofHours(1);
+ private static final Duration MIN_RETRY_DELAY = Duration.ofSeconds(30);
+ private static final Duration MAX_RETRY_DELAY = Duration.ofHours(1);
+
+ @Resource
+ private DocumentImportSnapshotCleanupMapper cleanupMapper;
+
+ @Resource
+ private DocumentImportChunkSnapshotService chunkSnapshotService;
+
+ @Resource
+ private CsvTableSnapshotService csvTableSnapshotService;
+
+ @Resource
+ private DocumentMapper documentMapper;
+
+ private final FlexIDKeyGenerator keyGenerator =
+ new FlexIDKeyGenerator();
+
+ /**
+ * 登记并尝试清理分块快照。
+ *
+ * @param manifestPath 分块快照清单路径
+ */
+ public void scheduleChunkSnapshot(String manifestPath) {
+ schedule(null, null, TYPE_CHUNK, manifestPath);
+ }
+
+ /**
+ * 登记并尝试清理 CSV 表格快照。
+ *
+ * @param knowledgeId 知识库 ID
+ * @param documentId 文档 ID
+ * @param manifestPath CSV 表格清单路径
+ */
+ public void scheduleCsvTableSnapshot(
+ BigInteger knowledgeId,
+ BigInteger documentId,
+ String manifestPath) {
+ schedule(
+ knowledgeId,
+ documentId,
+ TYPE_CSV_TABLE,
+ manifestPath);
+ }
+
+ /**
+ * 处理一批已到期的快照清理记录。
+ */
+ public void processPendingCleanups() {
+ Date now = new Date();
+ List records =
+ cleanupMapper.selectDueRecords(now, PROCESS_BATCH_SIZE);
+ for (DocumentImportSnapshotCleanup record : records) {
+ processRecord(record.getId());
+ }
+ }
+
+ /**
+ * 幂等登记清理记录，并在当前没有事务时立即尝试一次。
+ *
+ * @param knowledgeId 知识库 ID
+ * @param documentId 文档 ID
+ * @param snapshotType 快照类型
+ * @param manifestPath 清单路径
+ */
+ private void schedule(
+ BigInteger knowledgeId,
+ BigInteger documentId,
+ String snapshotType,
+ String manifestPath) {
+ if (!StringUtil.hasText(manifestPath)) {
+ return;
+ }
+ if (manifestPath.length() > MAX_PATH_CHARS) {
+ throw new IllegalArgumentException("快照清单路径超过 2048 字符");
+ }
+ Date now = new Date();
+ String pathHash = sha256(manifestPath);
+ DocumentImportSnapshotCleanup record =
+ new DocumentImportSnapshotCleanup();
+ record.setId(generateId(record));
+ record.setKnowledgeId(knowledgeId);
+ record.setDocumentId(documentId);
+ record.setSnapshotType(snapshotType);
+ record.setManifestPath(manifestPath);
+ record.setPathHash(pathHash);
+ record.setPhase(PHASE_PARTS_PENDING);
+ record.setAttemptCount(0);
+ record.setNextRetryAt(now);
+ record.setCreated(now);
+ record.setModified(now);
+ cleanupMapper.insertIgnore(record);
+
+ DocumentImportSnapshotCleanup persisted =
+ cleanupMapper.selectOneByQuery(
+ QueryWrapper.create()
+ .eq(
+ DocumentImportSnapshotCleanup::getSnapshotType,
+ snapshotType)
+ .eq(
+ DocumentImportSnapshotCleanup::getPathHash,
+ pathHash));
+ if (persisted == null
+ || !Objects.equals(
+ manifestPath, persisted.getManifestPath())) {
+ throw new IllegalStateException("快照清理记录登记失败");
+ }
+ // 事务中的登记必须随业务提交；后台调度器会在提交后处理。
+ if (!TransactionSynchronizationManager
+ .isActualTransactionActive()) {
+ processRecord(persisted.getId());
+ }
+ }
+
+ /**
+ * 领取并处理一条清理记录。
+ *
+ * @param recordId 清理记录 ID
+ */
+ private void processRecord(BigInteger recordId) {
+ Date now = new Date();
+ String executionToken = UUID.randomUUID().toString();
+ Date leaseUntil = new Date(
+ now.getTime() + CLAIM_LEASE.toMillis());
+ if (cleanupMapper.claim(
+ recordId, executionToken, leaseUntil, now) <= 0) {
+ return;
+ }
+
+ DocumentImportSnapshotCleanup record =
+ cleanupMapper.selectOneByQuery(
+ QueryWrapper.create()
+ .eq(DocumentImportSnapshotCleanup::getId, recordId)
+ .eq(
+ DocumentImportSnapshotCleanup::getExecutionToken,
+ executionToken));
+ if (record == null) {
+ return;
+ }
+ try {
+ if (PHASE_PARTS_PENDING.equals(record.getPhase())) {
+ deleteParts(record);
+ if (cleanupMapper.advanceToManifest(
+ recordId, executionToken, new Date()) <= 0) {
+ throw new IllegalStateException(
+ "快照清理阶段推进失败");
+ }
+ record.setPhase(PHASE_MANIFEST_PENDING);
+ }
+ if (!PHASE_MANIFEST_PENDING.equals(record.getPhase())) {
+ throw new IllegalStateException(
+ "未知快照清理阶段: " + record.getPhase());
+ }
+ deleteManifest(record);
+ clearCsvSnapshotPointer(record);
+ if (cleanupMapper.deleteCompleted(
+ recordId, executionToken) <= 0) {
+ LOG.warn(
+ "快照对象已清理，但清理记录终态删除未命中: cleanupId={}, path={}",
+ recordId, record.getManifestPath());
+ }
+ } catch (RuntimeException error) {
+ releaseForRetry(record, executionToken, error);
+ }
+ }
+
+ /**
+ * 删除记录指向的全部快照分片。
+ *
+ * @param record 清理记录
+ */
+ private void deleteParts(
+ DocumentImportSnapshotCleanup record) {
+ if (TYPE_CHUNK.equals(record.getSnapshotType())) {
+ chunkSnapshotService.deleteParts(
+ record.getManifestPath());
+ return;
+ }
+ if (TYPE_CSV_TABLE.equals(record.getSnapshotType())) {
+ csvTableSnapshotService.deleteParts(
+ record.getManifestPath());
+ return;
+ }
+ throw new IllegalStateException(
+ "未知快照类型: " + record.getSnapshotType());
+ }
+
+ /**
+ * 删除记录指向的快照清单。
+ *
+ * @param record 清理记录
+ */
+ private void deleteManifest(
+ DocumentImportSnapshotCleanup record) {
+ if (TYPE_CHUNK.equals(record.getSnapshotType())) {
+ chunkSnapshotService.deleteManifest(
+ record.getManifestPath());
+ return;
+ }
+ if (TYPE_CSV_TABLE.equals(record.getSnapshotType())) {
+ csvTableSnapshotService.deleteManifest(
+ record.getManifestPath());
+ return;
+ }
+ throw new IllegalStateException(
+ "未知快照类型: " + record.getSnapshotType());
+ }
+
+ /**
+ * CSV 清单删除完成后原子移除仍指向该清单的文档选项。
+ *
+ * @param record 清理记录
+ */
+ private void clearCsvSnapshotPointer(
+ DocumentImportSnapshotCleanup record) {
+ if (TYPE_CSV_TABLE.equals(record.getSnapshotType())
+ && record.getDocumentId() != null) {
+ documentMapper.clearCsvSnapshotPath(
+ record.getDocumentId(),
+ record.getManifestPath(),
+ new Date());
+ }
+ }
+
+ /**
+ * 释放失败记录并按指数退避安排下一次重试。
+ *
+ * @param record 清理记录
+ * @param executionToken 执行令牌
+ * @param error 清理异常
+ */
+ private void releaseForRetry(
+ DocumentImportSnapshotCleanup record,
+ String executionToken,
+ RuntimeException error) {
+ Date now = new Date();
+ int attempts = Math.max(
+ 1,
+ record.getAttemptCount() == null
+ ? 1
+ : record.getAttemptCount());
+ long multiplier = 1L << Math.min(7, attempts - 1);
+ long delayMillis = Math.min(
+ MAX_RETRY_DELAY.toMillis(),
+ MIN_RETRY_DELAY.toMillis() * multiplier);
+ String errorMessage = boundedError(error);
+ int released = cleanupMapper.releaseForRetry(
+ record.getId(),
+ executionToken,
+ new Date(now.getTime() + delayMillis),
+ errorMessage,
+ now);
+ LOG.error(
+ "文档导入快照清理失败，已安排重试: cleanupId={}, type={}, path={}, released={}",
+ record.getId(),
+ record.getSnapshotType(),
+ record.getManifestPath(),
+ released,
+ error);
+ }
+
+ /**
+ * 返回有界错误摘要。
+ *
+ * @param error 清理异常
+ * @return 有界摘要
+ */
+ private String boundedError(RuntimeException error) {
+ String message = error.getMessage();
+ String summary = error.getClass().getSimpleName()
+ + (StringUtil.hasText(message) ? ": " + message : "");
+ return summary.length() <= MAX_ERROR_CHARS
+ ? summary
+ : summary.substring(0, MAX_ERROR_CHARS);
+ }
+
+ /**
+ * 计算清单路径 SHA-256。
+ *
+ * @param value 清单路径
+ * @return 十六进制摘要
+ */
+ private String sha256(String value) {
+ try {
+ return HexFormat.of().formatHex(
+ MessageDigest.getInstance("SHA-256").digest(
+ value.getBytes(StandardCharsets.UTF_8)));
+ } catch (NoSuchAlgorithmException error) {
+ throw new IllegalStateException(
+ "当前运行环境不支持 SHA-256", error);
+ }
+ }
+
+ /**
+ * 生成清理记录主键。
+ *
+ * @param entity 清理记录
+ * @return 主键
+ */
+ private BigInteger generateId(Object entity) {
+ return new BigInteger(
+ String.valueOf(keyGenerator.generate(entity, null)));
+ }
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java
index 5d754305..c3a58941 100644
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeDocumentImportTaskAppService.java
@@ -119,6 +119,7 @@ public class KnowledgeDocumentImportTaskAppService {
private static final String KNOWLEDGE_PARSE_IMAGE_CATEGORY = "knowledge-parse";
private static final String OFFICE_PPTX_PAGE_STRATEGY = "OFFICE_PPTX_PAGE";
private static final String OFFICE_XLSX_ROW_WINDOW_STRATEGY = "OFFICE_XLSX_ROW_WINDOW";
+ private static final String TABLE_ROW_STRATEGY = "TABLE_ROW";
private static final String SEARCH_RENDER_MARKDOWN_METADATA_KEY = "renderMarkdown";
private static final String TASK_ERROR_PARSE_SERVICE_UNAVAILABLE = "parse_service_unavailable";
private static final String TASK_ERROR_PARSE_SERVICE_TIMEOUT = "parse_service_timeout";
@@ -135,6 +136,8 @@ public class KnowledgeDocumentImportTaskAppService {
private static final Pattern HTTP_SERVER_ERROR_PATTERN = Pattern.compile("\\bstatus=5\\d{2}\\b");
private static final Pattern MARKDOWN_IMAGE_PATTERN = Pattern.compile("!\\[(?:[^\\]]*)\\]\\(([^)]+)\\)");
private final FlexIDKeyGenerator flexIdKeyGenerator = new FlexIDKeyGenerator();
+ private final TabularRowWindowChunkBuilder tabularChunkBuilder =
+ new TabularRowWindowChunkBuilder();
@Resource
private DocumentMapper documentMapper;
@@ -182,6 +185,12 @@ public class KnowledgeDocumentImportTaskAppService {
@Resource
private DocumentImportChunkSnapshotService documentImportChunkSnapshotService;
+ @Resource
+ private CsvTableSnapshotService csvTableSnapshotService;
+
+ @Resource
+ private DocumentImportSnapshotCleanupService snapshotCleanupService;
+
@Resource
private RagIngestionService ragIngestionService;
@@ -713,20 +722,43 @@ public class KnowledgeDocumentImportTaskAppService {
tech.easyflow.ai.entity.Document document = requireDocumentForKnowledge(request.getDocumentId(), knowledge.getId());
ensurePreviewable(document);
- StrategyConfig requestedStrategy = null;
- if (request.getFiles() != null && !request.getFiles().isEmpty()) {
- requestedStrategy = request.getFiles().get(0).getStrategyConfig();
+ int pageNo = normalizePreviewPageNo(request.getPageNo());
+ int pageSize = normalizePreviewPageSize(request.getPageSize());
+ DocumentImportDtos.PreviewSession session;
+ String sessionId;
+ if (StringUtil.hasText(request.getPreviewSessionId())) {
+ sessionId = request.getPreviewSessionId();
+ session = documentImportPreviewService.getRequired(sessionId);
+ assertPreviewSessionOwner(knowledge, document, session);
+ } else {
+ StrategyConfig requestedStrategy = null;
+ if (request.getFiles() != null && !request.getFiles().isEmpty()) {
+ requestedStrategy = request.getFiles().get(0).getStrategyConfig();
+ }
+ session = buildPreviewSessionForDocument(
+ knowledge, document, requestedStrategy);
+ DocumentImportDtos.PreviewSession previousSession =
+ documentImportPreviewService.replaceForDocument(session);
+ sessionId = session.getSessionId();
+ if (previousSession != null
+ && StringUtil.hasText(previousSession.getChunkSnapshotPath())
+ && !Objects.equals(
+ previousSession.getChunkSnapshotPath(),
+ session.getChunkSnapshotPath())) {
+ deleteChunkSnapshotAfterCompletion(
+ previousSession.getChunkSnapshotPath());
+ }
}
- DocumentImportDtos.PreviewSession session = buildPreviewSessionForDocument(knowledge, document, requestedStrategy);
- String sessionId = documentImportPreviewService.put(session);
-
- DocumentImportDtos.PreviewFileResult item = buildPreviewFileResult(document, session, sessionId);
+ List pageChunks =
+ loadPreviewPage(session, pageNo, pageSize);
+ DocumentImportDtos.PreviewFileResult item = buildPreviewFileResult(
+ document, session, sessionId, pageNo, pageSize, pageChunks);
DocumentImportDtos.PreviewResponse response = new DocumentImportDtos.PreviewResponse();
response.setItems(List.of(item));
response.setTotalFiles(1);
- response.setTotalChunks(session.getDocumentChunks().size());
+ response.setTotalChunks(resolveSessionTotalChunks(session));
return Result.ok(response);
}
@@ -755,16 +787,22 @@ public class KnowledgeDocumentImportTaskAppService {
request.getPreviewSessionId(),
StringUtil.hasText(request.getPreviewSessionId()) ? null : existingSnapshotPath
);
- int totalChunks = session.getDocumentChunks().size();
+ int totalChunks = resolveSessionTotalChunks(session);
if (totalChunks <= 0) {
throw new BusinessException("未生成有效分块，无法开始向量化");
}
- boolean createdSnapshot = StringUtil.hasText(request.getPreviewSessionId())
- || !StringUtil.hasText(existingSnapshotPath);
- String snapshotPath = createdSnapshot
- ? documentImportChunkSnapshotService.save(session)
- : existingSnapshotPath;
+ String preparedSnapshotPath = session.getChunkSnapshotPath();
+ boolean createdSnapshot = !StringUtil.hasText(existingSnapshotPath)
+ && !StringUtil.hasText(preparedSnapshotPath);
+ String snapshotPath;
+ if (StringUtil.hasText(preparedSnapshotPath)) {
+ snapshotPath = preparedSnapshotPath;
+ } else if (createdSnapshot) {
+ snapshotPath = documentImportChunkSnapshotService.save(session);
+ } else {
+ snapshotPath = existingSnapshotPath;
+ }
mergeDocumentPreviewOptions(document, session, snapshotPath);
if (!claimDocumentIndexing(document, totalChunks)) {
if (createdSnapshot) {
@@ -993,11 +1031,13 @@ public class KnowledgeDocumentImportTaskAppService {
knowledge, document, requestedStrategy);
previewSessionId = documentImportPreviewService.put(session);
session.setSessionId(previewSessionId);
- int totalChunks = session.getDocumentChunks().size();
+ int totalChunks = resolveSessionTotalChunks(session);
if (totalChunks <= 0) {
throw new BusinessException("未生成有效分块，无法开始向量化");
}
- snapshotPath = documentImportChunkSnapshotService.save(session);
+ snapshotPath = StringUtil.hasText(session.getChunkSnapshotPath())
+ ? session.getChunkSnapshotPath()
+ : documentImportChunkSnapshotService.save(session);
if (!touchRunningTask(task) || !executionLock.renew()) {
throw new TaskOwnershipLostException(task.getId());
}
@@ -1015,8 +1055,11 @@ public class KnowledgeDocumentImportTaskAppService {
String errorMessage = truncateError(error.getMessage());
LOG.error("文档分块任务失败: taskId={}, documentId={}",
taskId, task.getDocumentId(), error);
+ String failureCode = error instanceof CsvImportException csvError
+ ? csvError.getFailureCode()
+ : TASK_ERROR_SPLIT_FAILED;
if (!selfProxy.failSplitTask(
- task, task.getDocumentId(), errorMessage, TASK_ERROR_SPLIT_FAILED)) {
+ task, task.getDocumentId(), errorMessage, failureCode)) {
LOG.warn("分块任务所有权已失效，忽略迟到失败: taskId={}", taskId);
}
} finally {
@@ -1182,37 +1225,57 @@ public class KnowledgeDocumentImportTaskAppService {
List storedChunks = new ArrayList();
try {
DocumentCollection knowledge = assertDocumentCollection(task.getKnowledgeId());
- DocumentImportDtos.PreviewSession session = resolveIndexPreviewSession(
- knowledge,
- document,
- asString(task.getPayloadJson().get("previewSessionId")),
- asString(task.getPayloadJson().get("chunkSnapshotPath"))
- );
- List chunks = session.getDocumentChunks();
- if (chunks == null || chunks.isEmpty()) {
- throw new BusinessException("预览会话无有效分块");
- }
- assertUniqueChunkIds(chunks);
-
if (!touchRunningTask(task) || !executionLock.renew()) {
throw new TaskOwnershipLostException(task.getId());
}
clearPersistedChunks(document.getId());
storeContext = prepareStoreContext(document);
- int totalChunks = chunks.size();
- int completedChunks = 0;
- for (int start = 0; start < chunks.size(); start += INDEX_BATCH_SIZE) {
- int end = Math.min(start + INDEX_BATCH_SIZE, chunks.size());
- List batch = new ArrayList(chunks.subList(start, end));
+ String chunkSnapshotPath =
+ asString(task.getPayloadJson().get("chunkSnapshotPath"));
+ DocumentImportDtos.PreviewSession session =
+ resolveIndexPreviewSession(
+ knowledge,
+ document,
+ asString(task.getPayloadJson().get("previewSessionId")),
+ chunkSnapshotPath
+ );
+ int totalChunks = resolveSessionTotalChunks(session);
+ if (totalChunks <= 0) {
+ throw new BusinessException("预览会话无有效分块");
+ }
+ StoreExecutionContext activeStoreContext = storeContext;
+ DocumentImportTask activeTask = task;
+ Set uniqueChunkIds =
+ new HashSet(Math.min(totalChunks, 65_536));
+ int[] completedChunks = new int[]{0};
+ java.util.function.Consumer> batchConsumer = batch -> {
+ assertUniqueChunkIds(batch, uniqueChunkIds);
LOG.info("文档向量化任务开始处理批次: taskId={}, documentId={}, batchStart={}, batchEnd={}, batchSize={}, totalChunks={}",
- taskId, document.getId(), start, end, batch.size(), totalChunks);
- storeDocumentChunks(storeContext, batch);
- storedChunks.addAll(batch);
+ taskId, document.getId(), completedChunks[0],
+ completedChunks[0] + batch.size(), batch.size(), totalChunks);
+ // 提前登记当前批次，确保外部存储部分成功后仍能覆盖回滚范围。
+ storedChunks.addAll(toChunkIdMarkers(batch));
+ storeDocumentChunks(activeStoreContext, batch);
persistChunkBatch(document, batch);
- completedChunks += batch.size();
- updateDocumentIndexProgress(document.getId(), totalChunks, completedChunks);
- if (!touchRunningTask(task) || !executionLock.renew()) {
- throw new TaskOwnershipLostException(task.getId());
+ completedChunks[0] += batch.size();
+ updateDocumentIndexProgress(
+ document.getId(), totalChunks, completedChunks[0]);
+ if (!touchRunningTask(activeTask) || !executionLock.renew()) {
+ throw new TaskOwnershipLostException(activeTask.getId());
+ }
+ };
+ if (StringUtil.hasText(chunkSnapshotPath)) {
+ documentImportChunkSnapshotService.forEachBatch(
+ chunkSnapshotPath, INDEX_BATCH_SIZE, batchConsumer);
+ } else {
+ List chunks = session.getDocumentChunks();
+ if (chunks == null || chunks.isEmpty()) {
+ throw new BusinessException("预览会话无有效分块");
+ }
+ for (int start = 0; start < chunks.size(); start += INDEX_BATCH_SIZE) {
+ int end = Math.min(start + INDEX_BATCH_SIZE, chunks.size());
+ batchConsumer.accept(new ArrayList(
+ chunks.subList(start, end)));
}
}
updateKnowledgeAfterStore(storeContext);
@@ -1222,6 +1285,7 @@ public class KnowledgeDocumentImportTaskAppService {
}
deleteChunkSnapshotAfterCompletion(
asString(task.getPayloadJson().get("chunkSnapshotPath")));
+ cleanupCsvTableSnapshotAfterCompletion(document);
} catch (TaskOwnershipLostException ownershipLost) {
// 旧执行失去 token 后不能清理可能已由新执行写入的索引或分块。
LOG.warn("向量化任务执行令牌已失效，停止迟到回滚: taskId={}, documentId={}",
@@ -1239,9 +1303,13 @@ public class KnowledgeDocumentImportTaskAppService {
rollbackSucceeded = rollbackStoredChunks(
taskId, document.getId(), storeContext, storedChunks);
}
+ String failureCode = e instanceof CsvImportException csvError
+ ? csvError.getFailureCode()
+ : TASK_ERROR_INDEX_FAILED;
markIndexFailed(
task, document,
- resolveIndexFailureMessage(e, rollbackSucceeded));
+ resolveIndexFailureMessage(e, rollbackSucceeded),
+ failureCode);
} finally {
closeStoreContext(storeContext);
}
@@ -1272,6 +1340,35 @@ public class KnowledgeDocumentImportTaskAppService {
String fileExt) {
LOG.info("开始同步解析文档: taskId={}, documentId={}, fileName={}, fileExt={}",
task.getId(), document.getId(), document.getTitle(), fileExt);
+ if ("csv".equals(fileExt)) {
+ CsvTableSnapshotService.CsvParseResult csvResult =
+ csvTableSnapshotService.parse(
+ document.getDocumentPath(),
+ document.getCollectionId(),
+ document.getId(),
+ task.getExecutionToken());
+ DocumentParsedResult parsedResult = new DocumentParsedResult();
+ parsedResult.setFileName(document.getTitle());
+ parsedResult.setPreferredText(csvResult.buildSummary());
+ parsedResult.setPlainText(csvResult.buildSummary());
+ parsedResult.setMetadata(csvResult.toMetadata());
+ LOG.info("CSV 流式解析完成: taskId={}, documentId={}, rows={}, columns={}, encoding={}",
+ task.getId(), document.getId(), csvResult.getRowCount(),
+ csvResult.getColumnCount(), csvResult.getEncoding());
+ boolean published = false;
+ try {
+ published = markParseSuccess(
+ task, document, parsedResult, fileExt, null);
+ } finally {
+ if (!published) {
+ snapshotCleanupService.scheduleCsvTableSnapshot(
+ document.getCollectionId(),
+ document.getId(),
+ csvResult.getManifestPath());
+ }
+ }
+ return;
+ }
String normalizedContent = readFileContent(document.getDocumentPath(), document.getTitle());
DocumentParsedResult parsedResult = new DocumentParsedResult();
parsedResult.setFileName(document.getTitle());
@@ -1283,11 +1380,22 @@ public class KnowledgeDocumentImportTaskAppService {
markParseSuccess(task, document, parsedResult, fileExt, null);
}
- private void markParseSuccess(DocumentImportTask task,
- tech.easyflow.ai.entity.Document document,
- DocumentParsedResult parsedResult,
- String sourceFormat,
- String providerTaskId) {
+ /**
+ * 发布解析结果并原子推进文档和批次状态。
+ *
+ * @param task 解析任务
+ * @param document 文档实体
+ * @param parsedResult 解析结果
+ * @param sourceFormat 来源格式
+ * @param providerTaskId 外部解析任务 ID
+ * @return 结果是否由当前任务代次成功发布
+ */
+ private boolean markParseSuccess(
+ DocumentImportTask task,
+ tech.easyflow.ai.entity.Document document,
+ DocumentParsedResult parsedResult,
+ String sourceFormat,
+ String providerTaskId) {
parsedResult = normalizeParsedImagesForKnowledgeImport(document, parsedResult);
ParsedKnowledgeContent parsedKnowledgeContent = buildParsedKnowledgeContent(document, parsedResult, sourceFormat);
if (!StringUtil.hasText(parsedKnowledgeContent.documentLlmContent)) {
@@ -1301,6 +1409,7 @@ public class KnowledgeDocumentImportTaskAppService {
}
if (parsedResult.getMetadata() != null && !parsedResult.getMetadata().isEmpty()) {
options.put(DocumentImportKeys.KEY_DOCUMENT_PARSE_METADATA, new LinkedHashMap(parsedResult.getMetadata()));
+ copyCsvParseMetadata(options, parsedResult.getMetadata());
}
if (parsedResult.getWarnings() != null && !parsedResult.getWarnings().isEmpty()) {
options.put(DocumentImportKeys.KEY_DOCUMENT_PARSE_WARNINGS, new ArrayList(parsedResult.getWarnings()));
@@ -1337,11 +1446,12 @@ public class KnowledgeDocumentImportTaskAppService {
if (!selfProxy.completeParseTask(task, document, automaticBatch)) {
LOG.warn("解析任务所有权已失效，忽略迟到结果: taskId={}, documentId={}",
task.getId(), document.getId());
- return;
+ return false;
}
LOG.info("文档解析任务完成: taskId={}, documentId={}, processStatus={}, providerTaskId={}, contentLength={}",
task.getId(), document.getId(), document.getProcessStatus(), providerTaskId,
parsedKnowledgeContent.documentLlmContent.length());
+ return true;
}
/**
@@ -1603,6 +1713,9 @@ public class KnowledgeDocumentImportTaskAppService {
Throwable current = error;
String fallbackCode = null;
while (current != null) {
+ if (current instanceof CsvImportException csvError) {
+ return csvError.getFailureCode();
+ }
if (current instanceof DocumentParseBridgeException bridgeError) {
String bridgeCode = bridgeError.getCode();
if ("unsupported_source".equals(bridgeCode)) {
@@ -2628,6 +2741,20 @@ public class KnowledgeDocumentImportTaskAppService {
throw new BusinessException("文档尚未完成解析");
}
String sourceFormat = normalizeSourceFormat(document);
+ if ("csv".equals(sourceFormat)) {
+ StrategyConfig strategyConfig =
+ resolveOfficeStrategyConfig(document, requestedStrategy, sourceFormat);
+ String tableSnapshotPath = optionAsString(
+ document.getOptions(),
+ DocumentImportKeys.KEY_DOCUMENT_CSV_TABLE_SNAPSHOT_PATH);
+ if (!StringUtil.hasText(tableSnapshotPath)) {
+ throw new CsvImportException(
+ CsvTableSnapshotService.FAILURE_SNAPSHOT_CORRUPTED,
+ "CSV 表格快照不存在，请重新解析");
+ }
+ return csvTableSnapshotService.buildChunkSnapshot(
+ document, tableSnapshotPath, strategyConfig);
+ }
if (isOfficeDocument(sourceFormat)) {
return buildOfficePreviewSession(knowledge, document, requestedStrategy, sourceFormat);
}
@@ -2742,7 +2869,9 @@ public class KnowledgeDocumentImportTaskAppService {
if (rowsPerChunk == null || rowsPerChunk <= 0) {
rowsPerChunk = 10;
}
- config.setStrategyCode(OFFICE_XLSX_ROW_WINDOW_STRATEGY);
+ config.setStrategyCode("csv".equals(sourceFormat)
+ ? TABLE_ROW_STRATEGY
+ : OFFICE_XLSX_ROW_WINDOW_STRATEGY);
config.setRowsPerChunk(rowsPerChunk);
return config;
}
@@ -2906,14 +3035,21 @@ public class KnowledgeDocumentImportTaskAppService {
? new ArrayList>(sheetRows.subList(1, sheetRows.size()))
: new ArrayList>();
if (dataRows.isEmpty()) {
- chunks.add(buildXlsxWindowChunk(document, sorting++, sheetName, sheet, headerRow,
- new ArrayList>(), sheetImages));
+ List windowChunks = buildXlsxWindowChunks(
+ document, sorting, sheetName, sheet, headerRow,
+ new ArrayList>(), sheetImages);
+ chunks.addAll(windowChunks);
+ sorting += windowChunks.size();
continue;
}
for (int start = 0; start < dataRows.size(); start += rowsPerChunk) {
int end = Math.min(start + rowsPerChunk, dataRows.size());
List> windowRows = new ArrayList>(dataRows.subList(start, end));
- chunks.add(buildXlsxWindowChunk(document, sorting++, sheetName, sheet, headerRow, windowRows, sheetImages));
+ List windowChunks = buildXlsxWindowChunks(
+ document, sorting, sheetName, sheet, headerRow,
+ windowRows, sheetImages);
+ chunks.addAll(windowChunks);
+ sorting += windowChunks.size();
}
}
if (chunks.isEmpty()) {
@@ -2929,19 +3065,70 @@ public class KnowledgeDocumentImportTaskAppService {
return chunks;
}
- private DocumentChunk buildXlsxWindowChunk(tech.easyflow.ai.entity.Document document,
- int sorting,
- String sheetName,
- Map sheetSummary,
- Map headerRow,
- List> windowRows,
- List> sheetImages) {
+ /**
+ * 构建 XLSX 行窗口分块。无图片窗口与 CSV 共用表格构建器，图片窗口保留现有 OCR 元信息。
+ *
+ * @param document 文档
+ * @param sorting 起始排序号
+ * @param sheetName Sheet 名称
+ * @param sheetSummary Sheet 摘要
+ * @param headerRow 表头行
+ * @param windowRows 数据行
+ * @param sheetImages Sheet 图片
+ * @return 一个或多个分块
+ */
+ private List buildXlsxWindowChunks(
+ tech.easyflow.ai.entity.Document document,
+ int sorting,
+ String sheetName,
+ Map sheetSummary,
+ Map headerRow,
+ List> windowRows,
+ List> sheetImages) {
int headerRowIndex = asInteger(headerRow.get("rowIndex"), 0);
int rowStart = windowRows.isEmpty() ? headerRowIndex + 1 : asInteger(windowRows.get(0).get("rowIndex"), headerRowIndex) + 1;
int rowEnd = windowRows.isEmpty()
? rowStart
: asInteger(windowRows.get(windowRows.size() - 1).get("rowIndex"), rowStart - 1) + 1;
List> selectedImages = selectWindowImages(sheetImages, rowStart, rowEnd, windowRows.isEmpty());
+ if (selectedImages.isEmpty()) {
+ List> allRows =
+ new ArrayList>();
+ allRows.add(headerRow);
+ allRows.addAll(windowRows);
+ int maxCol = resolveMaxColumnCount(allRows);
+ if (maxCol > 0) {
+ List headers =
+ resolveTabularRowValues(headerRow, maxCol, true);
+ List rows =
+ new ArrayList();
+ for (Map row : windowRows) {
+ rows.add(new TabularRowWindowChunkBuilder.TabularRow(
+ asInteger(row.get("rowIndex"), 0) + 1,
+ resolveTabularRowValues(row, maxCol, false)));
+ }
+ List chunks = tabularChunkBuilder.build(
+ document.getId(), document.getCollectionId(), sheetName,
+ headers, rows, sorting, RagChunkTypes.SECTION);
+ for (DocumentChunk chunk : chunks) {
+ Map options =
+ new LinkedHashMap(chunk.getOptions());
+ options.put(
+ DocumentImportKeys.KEY_DOCUMENT_ROW_START, rowStart);
+ options.put(
+ DocumentImportKeys.KEY_DOCUMENT_ROW_END, rowEnd);
+ options.put(
+ DocumentImportKeys.KEY_DOCUMENT_PARSE_ARTIFACT_SUMMARY,
+ buildXlsxChunkSummary(
+ sheetSummary, rowStart, rowEnd, selectedImages));
+ options.put(
+ DocumentImportKeys.KEY_DOCUMENT_IMAGE_REFS,
+ new ArrayList());
+ chunk.setOptions(options);
+ }
+ return chunks;
+ }
+ }
String renderMarkdown = buildXlsxChunkRenderMarkdown(sheetName, headerRow, windowRows, selectedImages);
String llmContent = buildXlsxChunkLlmContent(sheetName, headerRow, windowRows, selectedImages);
Map options = new LinkedHashMap();
@@ -2959,7 +3146,8 @@ public class KnowledgeDocumentImportTaskAppService {
options.put(DocumentImportKeys.KEY_DOCUMENT_ROW_END, rowEnd);
options.put(DocumentImportKeys.KEY_DOCUMENT_IMAGE_REFS, collectImageRefs(selectedImages));
options.put(DocumentImportKeys.KEY_DOCUMENT_PARSE_ARTIFACT_SUMMARY, buildXlsxChunkSummary(sheetSummary, rowStart, rowEnd, selectedImages));
- return buildOfficeChunk(document, sorting, llmContent, renderMarkdown, options);
+ return Collections.singletonList(
+ buildOfficeChunk(document, sorting, llmContent, renderMarkdown, options));
}
private DocumentChunk buildXlsxImageOnlyChunk(tech.easyflow.ai.entity.Document document,
@@ -3246,6 +3434,27 @@ public class KnowledgeDocumentImportTaskAppService {
private List resolveMarkdownRowValues(Map row,
int maxCol,
boolean headerRow) {
+ List rawValues =
+ resolveTabularRowValues(row, maxCol, headerRow);
+ List values = new ArrayList(rawValues.size());
+ for (String value : rawValues) {
+ values.add(escapeMarkdownCell(value));
+ }
+ return values;
+ }
+
+ /**
+ * 将 XLSX 行转换为通用表格单元格值。
+ *
+ * @param row XLSX 行
+ * @param maxCol 最大列数
+ * @param headerRow 是否为表头
+ * @return 固定列数的原始值
+ */
+ private List resolveTabularRowValues(
+ Map row,
+ int maxCol,
+ boolean headerRow) {
List values = new ArrayList(Collections.nCopies(maxCol, ""));
if (row == null) {
return values;
@@ -3260,7 +3469,7 @@ public class KnowledgeDocumentImportTaskAppService {
if (headerRow && !StringUtil.hasText(mergedText)) {
mergedText = "列" + (colIndex + 1);
}
- values.set(colIndex, escapeMarkdownCell(mergedText));
+ values.set(colIndex, mergedText);
}
return values;
}
@@ -3402,6 +3611,9 @@ public class KnowledgeDocumentImportTaskAppService {
if ("pptx".equals(sourceFormat) || OFFICE_PPTX_PAGE_STRATEGY.equals(strategyCode)) {
return "按页分块";
}
+ if ("csv".equals(sourceFormat) || TABLE_ROW_STRATEGY.equals(strategyCode)) {
+ return "按表头 / 行窗口";
+ }
if ("xlsx".equals(sourceFormat) || OFFICE_XLSX_ROW_WINDOW_STRATEGY.equals(strategyCode)) {
return "按 Sheet / 行窗口";
}
@@ -3414,11 +3626,18 @@ public class KnowledgeDocumentImportTaskAppService {
* @param document 文档实体
* @param session 预览会话
* @param sessionId 预览会话 ID
+ * @param pageNo 当前页码
+ * @param pageSize 当前页大小
+ * @param pageChunks 当前页分块
* @return 预览文件结果
*/
- private DocumentImportDtos.PreviewFileResult buildPreviewFileResult(tech.easyflow.ai.entity.Document document,
- DocumentImportDtos.PreviewSession session,
- String sessionId) {
+ private DocumentImportDtos.PreviewFileResult buildPreviewFileResult(
+ tech.easyflow.ai.entity.Document document,
+ DocumentImportDtos.PreviewSession session,
+ String sessionId,
+ int pageNo,
+ int pageSize,
+ List pageChunks) {
DocumentImportDtos.PreviewFileResult item = new DocumentImportDtos.PreviewFileResult();
item.setPreviewSessionId(sessionId);
item.setFilePath(document.getDocumentPath());
@@ -3427,12 +3646,77 @@ public class KnowledgeDocumentImportTaskAppService {
item.setStrategyCode(session.getStrategyConfig().getStrategyCode());
item.setStrategyLabel(resolveStrategyLabel(session.getSourceFormat(), session.getStrategyConfig()));
item.setAnalysis(session.getAnalysis());
- item.setTotalChunks(session.getDocumentChunks().size());
- item.setTotalWarnings(countChunkWarnings(session.getDocumentChunks()));
- item.setChunks(toPreviewChunkResults(session.getDocumentChunks()));
+ item.setTotalChunks(resolveSessionTotalChunks(session));
+ item.setTotalWarnings(countChunkWarnings(pageChunks));
+ item.setPageNo(pageNo);
+ item.setPageSize(pageSize);
+ item.setChunks(toPreviewChunkResults(pageChunks));
return item;
}
+ /**
+ * 从内存会话或分片快照读取一个有界预览页。
+ *
+ * @param session 预览会话
+ * @param pageNo 从 1 开始的页码
+ * @param pageSize 单页分块数
+ * @return 当前页分块
+ */
+ private List loadPreviewPage(
+ DocumentImportDtos.PreviewSession session,
+ int pageNo,
+ int pageSize) {
+ long offsetValue = (long) (pageNo - 1) * pageSize;
+ if (offsetValue > Integer.MAX_VALUE) {
+ throw new BusinessException("预览页码超出范围");
+ }
+ int offset = (int) offsetValue;
+ if (StringUtil.hasText(session.getChunkSnapshotPath())) {
+ return documentImportChunkSnapshotService.loadPage(
+ session.getChunkSnapshotPath(), offset, pageSize);
+ }
+ List chunks = session.getDocumentChunks();
+ if (chunks == null || offset >= chunks.size()) {
+ return new ArrayList();
+ }
+ int end = Math.min(offset + pageSize, chunks.size());
+ return new ArrayList(
+ chunks.subList(offset, end));
+ }
+
+ /**
+ * 规范化预览页码。
+ *
+ * @param pageNo 请求页码
+ * @return 至少为 1 的页码
+ */
+ private int normalizePreviewPageNo(Integer pageNo) {
+ if (pageNo == null) {
+ return 1;
+ }
+ if (pageNo < 1) {
+ throw new BusinessException("预览页码必须大于 0");
+ }
+ return pageNo;
+ }
+
+ /**
+ * 规范化预览单页分块数。
+ *
+ * @param pageSize 请求单页分块数
+ * @return 1 到 50 之间的单页分块数
+ */
+ private int normalizePreviewPageSize(Integer pageSize) {
+ if (pageSize == null) {
+ return 20;
+ }
+ if (pageSize < 1 || pageSize > 50) {
+ throw new BusinessException(
+ "预览每页分块数必须在 1 到 50 之间");
+ }
+ return pageSize;
+ }
+
private void mergeDocumentPreviewOptions(tech.easyflow.ai.entity.Document document,
DocumentImportDtos.PreviewSession session,
String chunkSnapshotPath) {
@@ -3460,7 +3744,7 @@ public class KnowledgeDocumentImportTaskAppService {
: optionAsString(document.getOptions(), DocumentImportKeys.KEY_DOCUMENT_CHUNK_SNAPSHOT_PATH);
if (StringUtil.hasText(persistedSnapshotPath)) {
DocumentImportDtos.PreviewSession session =
- documentImportChunkSnapshotService.load(persistedSnapshotPath);
+ documentImportChunkSnapshotService.loadHeader(persistedSnapshotPath);
assertPreviewSessionOwner(knowledge, document, session);
return session;
}
@@ -4007,8 +4291,8 @@ public class KnowledgeDocumentImportTaskAppService {
}
private void assertSupportedImportFile(String fileExt) {
- if (!Arrays.asList("pdf", "docx", "txt", "md", "pptx", "xlsx").contains(fileExt)) {
- throw new BusinessException("当前仅支持 pdf/docx/txt/md/pptx/xlsx 文档导入");
+ if (!DocumentImportFormatPolicy.isSupported(fileExt)) {
+ throw new BusinessException("当前仅支持 pdf/docx/txt/md/pptx/xlsx/csv 文档导入");
}
}
@@ -4329,7 +4613,7 @@ public class KnowledgeDocumentImportTaskAppService {
}
/**
- * 向量化完成后尽力删除持久化分块快照。
+ * 登记并尝试删除持久化分块快照。
*
* @param snapshotPath 快照路径
*/
@@ -4338,9 +4622,12 @@ public class KnowledgeDocumentImportTaskAppService {
return;
}
try {
- documentImportChunkSnapshotService.delete(snapshotPath);
+ snapshotCleanupService.scheduleChunkSnapshot(snapshotPath);
} catch (RuntimeException error) {
- LOG.error("删除已完成文档的分块快照失败: path={}", snapshotPath, error);
+ LOG.error(
+ "登记文档分块快照清理失败: path={}",
+ snapshotPath,
+ error);
}
}
@@ -4356,6 +4643,36 @@ public class KnowledgeDocumentImportTaskAppService {
runAfterCommit(() -> deleteChunkSnapshotAfterCompletion(snapshotPath));
}
+ /**
+ * CSV 向量化完成后登记中间表格快照清理。
+ *
+ * 清理记录持久化后由当前线程立即尝试，失败时由后台调度器继续重试。
+ *
+ * @param document 已完成向量化的文档
+ */
+ private void cleanupCsvTableSnapshotAfterCompletion(
+ tech.easyflow.ai.entity.Document document) {
+ if (document == null || document.getOptions() == null) {
+ return;
+ }
+ String tableSnapshotPath = optionAsString(
+ document.getOptions(),
+ DocumentImportKeys.KEY_DOCUMENT_CSV_TABLE_SNAPSHOT_PATH);
+ if (!StringUtil.hasText(tableSnapshotPath)) {
+ return;
+ }
+ try {
+ snapshotCleanupService.scheduleCsvTableSnapshot(
+ document.getCollectionId(),
+ document.getId(),
+ tableSnapshotPath);
+ } catch (RuntimeException error) {
+ LOG.error(
+ "登记已完成 CSV 文档的表格快照清理失败: documentId={}, path={}",
+ document.getId(), tableSnapshotPath, error);
+ }
+ }
+
private StrategyConfig resolveStrategyConfig(DocumentCollection knowledge,
StrategyConfig requestConfig,
AnalysisResult analysisResult) {
@@ -4653,6 +4970,63 @@ public class KnowledgeDocumentImportTaskAppService {
options.remove(DocumentImportKeys.KEY_DOCUMENT_PARSE_STATUS_MESSAGE);
}
+ /**
+ * 将 CSV 解析元信息提升到文档有界选项中，供后续分块阶段直接定位表格快照。
+ *
+ * @param options 文档选项
+ * @param metadata 解析元信息
+ */
+ private void copyCsvParseMetadata(
+ Map options,
+ Map metadata) {
+ if (options == null || metadata == null) {
+ return;
+ }
+ copyOptionIfPresent(
+ options, metadata, DocumentImportKeys.KEY_DOCUMENT_CSV_TABLE_SNAPSHOT_PATH);
+ copyOptionIfPresent(
+ options, metadata, DocumentImportKeys.KEY_DOCUMENT_CSV_ENCODING);
+ copyOptionIfPresent(
+ options, metadata, DocumentImportKeys.KEY_DOCUMENT_CSV_ROW_COUNT);
+ copyOptionIfPresent(
+ options, metadata, DocumentImportKeys.KEY_DOCUMENT_CSV_COLUMN_COUNT);
+ }
+
+ /**
+ * 复制一个存在的选项值。
+ *
+ * @param target 目标选项
+ * @param source 来源选项
+ * @param key 选项键
+ */
+ private void copyOptionIfPresent(
+ Map target,
+ Map source,
+ String key) {
+ if (source.containsKey(key) && source.get(key) != null) {
+ target.put(key, source.get(key));
+ }
+ }
+
+ /**
+ * 返回会话的真实分块总数。
+ *
+ * @param session 预览会话
+ * @return 总分块数
+ */
+ private int resolveSessionTotalChunks(
+ DocumentImportDtos.PreviewSession session) {
+ if (session == null) {
+ return 0;
+ }
+ if (session.getTotalChunks() != null) {
+ return session.getTotalChunks();
+ }
+ return session.getDocumentChunks() == null
+ ? 0
+ : session.getDocumentChunks().size();
+ }
+
private BigInteger generateId(Object entity) {
return new BigInteger(String.valueOf(flexIdKeyGenerator.generate(entity, null)));
}
@@ -4664,6 +5038,18 @@ public class KnowledgeDocumentImportTaskAppService {
*/
private void assertUniqueChunkIds(List chunks) {
Set uniqueIds = new HashSet(chunks.size());
+ assertUniqueChunkIds(chunks, uniqueIds);
+ }
+
+ /**
+ * 校验批次分块 ID，并把已见 ID 保留到跨分片集合。
+ *
+ * @param chunks 当前批次
+ * @param uniqueIds 已见 ID
+ */
+ private void assertUniqueChunkIds(
+ List chunks,
+ Set uniqueIds) {
for (DocumentChunk chunk : chunks) {
if (chunk == null || chunk.getId() == null) {
throw new IllegalStateException("文档分块缺少 ID");
@@ -4674,6 +5060,22 @@ public class KnowledgeDocumentImportTaskAppService {
}
}
+ /**
+ * 将已写入分块压缩为仅含主键的回滚标记，避免索引阶段长期持有全部正文。
+ *
+ * @param chunks 已写入分块
+ * @return 主键标记
+ */
+ private List toChunkIdMarkers(List chunks) {
+ List markers = new ArrayList(chunks.size());
+ for (DocumentChunk chunk : chunks) {
+ DocumentChunk marker = new DocumentChunk();
+ marker.setId(chunk.getId());
+ markers.add(marker);
+ }
+ return markers;
+ }
+
private BigInteger resolveOperatorId() {
try {
return BigInteger.valueOf(StpUtil.getLoginIdAsLong());
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java
index 7abe52f8..b21b6892 100644
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/KnowledgeImportBatchFacade.java
@@ -60,7 +60,7 @@ public class KnowledgeImportBatchFacade {
private static final Duration INCOMPLETE_SUBMISSION_TIMEOUT =
Duration.ofMinutes(30);
private static final Set SUPPORTED_EXTENSIONS =
- Set.of("txt", "md", "pdf", "docx", "pptx", "xlsx");
+ DocumentImportFormatPolicy.supportedExtensions();
private final DocumentImportBatchAppService batchAppService;
private final DocumentImportBatchTracker batchTracker;
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/TabularRowWindowChunkBuilder.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/TabularRowWindowChunkBuilder.java
new file mode 100644
index 00000000..ea1d3be6
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/TabularRowWindowChunkBuilder.java
@@ -0,0 +1,789 @@
+package tech.easyflow.ai.documentimport.task;
+
+import com.easyagents.rag.core.BgeM3ChunkSafety;
+import com.easyagents.rag.core.RagDefaults;
+import com.mybatisflex.core.keygen.impl.FlexIDKeyGenerator;
+import tech.easyflow.ai.documentimport.DocumentImportKeys;
+import tech.easyflow.ai.entity.DocumentChunk;
+
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 表格行窗口分块构建器。
+ *
+ * CSV 与 XLSX 统一复用“表头 + 连续数据行”的语义，并对超长内容做有界续片，
+ * 保持单个分块满足下游模型的 Token 硬限制。
+ *
+ * @author Codex
+ * @since 2026-08-04
+ */
+public class TabularRowWindowChunkBuilder {
+
+ /**
+ * 单个分块允许的最大 Java 字符数。
+ */
+ public static final int MAX_CHUNK_CONTENT_CHARS = 60_000;
+
+ private final FlexIDKeyGenerator keyGenerator = new FlexIDKeyGenerator();
+
+ /**
+ * 构建一个表格行窗口对应的分块。
+ *
+ * @param documentId 文档 ID
+ * @param collectionId 知识库 ID
+ * @param sourceLabel 表格或 Sheet 名称
+ * @param headers 规范化表头
+ * @param rows 窗口数据行
+ * @param startingSorting 起始排序号
+ * @param chunkType 分块类型
+ * @return 一个或多个有界分块
+ */
+ public List build(
+ BigInteger documentId,
+ BigInteger collectionId,
+ String sourceLabel,
+ List headers,
+ List rows,
+ int startingSorting,
+ String chunkType) {
+ if (headers == null || headers.isEmpty()) {
+ throw new IllegalArgumentException("表格分块缺少表头");
+ }
+ List safeRows = rows == null
+ ? Collections.emptyList()
+ : rows;
+ List parts =
+ renderParts(sourceLabel, headers, safeRows);
+ List chunks =
+ new ArrayList(parts.size());
+ for (int index = 0; index < parts.size(); index++) {
+ RenderedPart part = parts.get(index);
+ String content = part.content;
+ DocumentChunk chunk = new DocumentChunk();
+ chunk.setId(generateId(chunk));
+ chunk.setDocumentId(documentId);
+ chunk.setDocumentCollectionId(collectionId);
+ chunk.setSorting(startingSorting + index);
+ chunk.setContent(content);
+
+ Map options = new LinkedHashMap();
+ options.put("chunkType", chunkType);
+ options.put(
+ "sourceLabel",
+ sourceLabel + " · " + part.rowStart + "-" + part.rowEnd + " 行");
+ options.put("headingPath", Collections.singletonList(sourceLabel));
+ options.put("charCount", content.length());
+ options.put("tokenEstimate", Math.max(
+ 1, BgeM3ChunkSafety.estimateContentTokens(content)));
+ options.put("partNo", index + 1);
+ options.put("partTotal", parts.size());
+ options.put("warnings", new ArrayList());
+ options.put(DocumentImportKeys.KEY_DOCUMENT_RENDER_MARKDOWN, content);
+ options.put(DocumentImportKeys.KEY_DOCUMENT_SHEET_NAME, sourceLabel);
+ options.put(
+ DocumentImportKeys.KEY_DOCUMENT_ROW_START, part.rowStart);
+ options.put(
+ DocumentImportKeys.KEY_DOCUMENT_ROW_END, part.rowEnd);
+ chunk.setOptions(options);
+ chunks.add(chunk);
+ }
+ return chunks;
+ }
+
+ /**
+ * 估算表格标题和表头渲染后的字符数。
+ *
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @return 渲染字符数
+ */
+ public int estimateHeaderChars(
+ String sourceLabel,
+ List headers) {
+ validateHeaders(headers);
+ long length = 6L + safeText(sourceLabel).length();
+ length += estimateCellsChars(headers);
+ length += 4L + 6L * headers.size();
+ return boundedInt(length);
+ }
+
+ /**
+ * 估算表格标题和表头渲染后的 Token 数。
+ *
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @return 保守 Token 数
+ */
+ public int estimateHeaderTokens(
+ String sourceLabel,
+ List headers) {
+ validateHeaders(headers);
+ long tokens = 6L
+ + BgeM3ChunkSafety.estimateContentTokens(safeText(sourceLabel));
+ tokens += estimateCellsTokens(headers);
+ tokens += 4L + 6L * headers.size();
+ return boundedInt(tokens);
+ }
+
+ /**
+ * 估算单行追加到 Markdown 表格后的字符数。
+ *
+ * @param headers 表头
+ * @param row 表格数据行
+ * @return 渲染字符数
+ */
+ public int estimateRowChars(
+ List headers,
+ TabularRow row) {
+ validateRow(headers, row);
+ List values =
+ normalizeValues(row.getValues(), headers.size());
+ return boundedInt(5L + estimateCellsChars(values));
+ }
+
+ /**
+ * 估算单行追加到 Markdown 表格后的 Token 数。
+ *
+ * @param headers 表头
+ * @param row 表格数据行
+ * @return 保守 Token 数
+ */
+ public int estimateRowTokens(
+ List headers,
+ TabularRow row) {
+ validateRow(headers, row);
+ List values =
+ normalizeValues(row.getValues(), headers.size());
+ return boundedInt(5L + estimateCellsTokens(values));
+ }
+
+ /**
+ * 按表格语义生成有界 Markdown 续片。
+ *
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @param rows 数据行
+ * @return 有界续片
+ */
+ private List renderParts(
+ String sourceLabel,
+ List headers,
+ List rows) {
+ int headerChars = estimateHeaderChars(sourceLabel, headers);
+ int headerTokens = estimateHeaderTokens(sourceLabel, headers);
+ if (!fitsLimits(headerChars, headerTokens)) {
+ throw new IllegalArgumentException("表格标题或表头超过分块上限");
+ }
+ if (rows.isEmpty()) {
+ return Collections.singletonList(new RenderedPart(
+ renderMarkdown(sourceLabel, headers, rows), 1, 1));
+ }
+
+ List parts = new ArrayList();
+ List currentRows = new ArrayList();
+ long currentChars = headerChars;
+ long currentTokens = headerTokens;
+ for (TabularRow row : rows) {
+ int rowChars = estimateRowChars(headers, row);
+ int rowTokens = estimateRowTokens(headers, row);
+ if (!fitsLimits(
+ (long) headerChars + rowChars,
+ (long) headerTokens + rowTokens)) {
+ flushRows(parts, sourceLabel, headers, currentRows);
+ currentRows.clear();
+ currentChars = headerChars;
+ currentTokens = headerTokens;
+ parts.addAll(renderOversizedRow(
+ sourceLabel, headers, row));
+ continue;
+ }
+ if (!currentRows.isEmpty()
+ && !fitsLimits(
+ currentChars + rowChars,
+ currentTokens + rowTokens)) {
+ flushRows(parts, sourceLabel, headers, currentRows);
+ currentRows.clear();
+ currentChars = headerChars;
+ currentTokens = headerTokens;
+ }
+ currentRows.add(row);
+ currentChars += rowChars;
+ currentTokens += rowTokens;
+ }
+ flushRows(parts, sourceLabel, headers, currentRows);
+ return parts;
+ }
+
+ /**
+ * 输出一个完整行集合，续片始终包含标题和表头。
+ *
+ * @param parts 输出续片
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @param rows 数据行
+ */
+ private void flushRows(
+ List parts,
+ String sourceLabel,
+ List headers,
+ List rows) {
+ if (rows.isEmpty()) {
+ return;
+ }
+ String content = renderMarkdown(sourceLabel, headers, rows);
+ assertWithinLimits(content);
+ parts.add(new RenderedPart(
+ content,
+ rows.get(0).getRowNumber(),
+ rows.get(rows.size() - 1).getRowNumber()));
+ }
+
+ /**
+ * 将单条超长记录按列边界生成连续续片。
+ *
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @param row 超长数据行
+ * @return 语义完整的续片
+ */
+ private List renderOversizedRow(
+ String sourceLabel,
+ List headers,
+ TabularRow row) {
+ List values =
+ normalizeValues(row.getValues(), headers.size());
+ List parts = new ArrayList();
+ List partHeaders = new ArrayList();
+ List partValues = new ArrayList();
+ for (int index = 0; index < headers.size(); index++) {
+ String header = headers.get(index);
+ String value = values.get(index);
+ if (!fitsSingleRow(
+ sourceLabel,
+ Collections.singletonList(header),
+ Collections.singletonList(value),
+ row.getRowNumber())) {
+ flushColumnGroup(
+ parts, sourceLabel, partHeaders, partValues,
+ row.getRowNumber());
+ partHeaders.clear();
+ partValues.clear();
+ parts.addAll(renderOversizedCell(
+ sourceLabel, header, value, row.getRowNumber()));
+ continue;
+ }
+
+ List candidateHeaders =
+ new ArrayList(partHeaders);
+ candidateHeaders.add(header);
+ List candidateValues =
+ new ArrayList(partValues);
+ candidateValues.add(value);
+ if (!partHeaders.isEmpty()
+ && !fitsLimits(renderSingleRow(
+ sourceLabel,
+ candidateHeaders,
+ candidateValues,
+ row.getRowNumber()))) {
+ flushColumnGroup(
+ parts, sourceLabel, partHeaders, partValues,
+ row.getRowNumber());
+ partHeaders.clear();
+ partValues.clear();
+ }
+ partHeaders.add(header);
+ partValues.add(value);
+ }
+ flushColumnGroup(
+ parts, sourceLabel, partHeaders, partValues,
+ row.getRowNumber());
+ return parts;
+ }
+
+ /**
+ * 通过有界估算判断单行表格是否满足字符与 Token 上限。
+ *
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @param values 行值
+ * @param rowNumber 逻辑行号
+ * @return 是否满足
+ */
+ private boolean fitsSingleRow(
+ String sourceLabel,
+ List headers,
+ List values,
+ int rowNumber) {
+ TabularRow row = new TabularRow(rowNumber, values);
+ return fitsLimits(
+ (long) estimateHeaderChars(sourceLabel, headers)
+ + estimateRowChars(headers, row),
+ (long) estimateHeaderTokens(sourceLabel, headers)
+ + estimateRowTokens(headers, row));
+ }
+
+ /**
+ * 输出同一记录的一组完整列。
+ *
+ * @param parts 输出续片
+ * @param sourceLabel 表格名称
+ * @param headers 当前列头
+ * @param values 当前列值
+ * @param rowNumber 逻辑行号
+ */
+ private void flushColumnGroup(
+ List parts,
+ String sourceLabel,
+ List headers,
+ List values,
+ int rowNumber) {
+ if (headers.isEmpty()) {
+ return;
+ }
+ String content = renderSingleRow(
+ sourceLabel, headers, values, rowNumber);
+ assertWithinLimits(content);
+ parts.add(new RenderedPart(content, rowNumber, rowNumber));
+ }
+
+ /**
+ * 对单个超长单元格做安全续片，每个续片重复当前列名。
+ *
+ * @param sourceLabel 表格名称
+ * @param header 当前列名
+ * @param value 单元格值
+ * @param rowNumber 逻辑行号
+ * @return 单列续片
+ */
+ private List renderOversizedCell(
+ String sourceLabel,
+ String header,
+ String value,
+ int rowNumber) {
+ String normalizedValue = normalizeCellNewlines(value);
+ String emptyContent = renderSingleRow(
+ sourceLabel,
+ Collections.singletonList(header),
+ Collections.singletonList(""),
+ rowNumber);
+ if (!fitsLimits(emptyContent)) {
+ throw new IllegalArgumentException("表格列名超过分块上限");
+ }
+
+ int fixedChars = emptyContent.length();
+ int fixedTokens =
+ BgeM3ChunkSafety.estimateContentTokens(emptyContent);
+ int maxCandidateCodePoints = Math.max(
+ 1,
+ Math.min(
+ MAX_CHUNK_CONTENT_CHARS - fixedChars,
+ RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT - fixedTokens));
+ List parts = new ArrayList();
+ int offset = 0;
+ while (offset < normalizedValue.length()) {
+ int remainingCodePoints =
+ normalizedValue.codePointCount(offset, normalizedValue.length());
+ int low = 1;
+ int high = Math.min(remainingCodePoints, maxCandidateCodePoints);
+ int acceptedEnd = -1;
+ String acceptedContent = null;
+ while (low <= high) {
+ int middle = low + (high - low) / 2;
+ int end = normalizedValue.offsetByCodePoints(offset, middle);
+ String fragment = normalizedValue.substring(offset, end);
+ String content = renderSingleRow(
+ sourceLabel,
+ Collections.singletonList(header),
+ Collections.singletonList(fragment),
+ rowNumber);
+ if (fitsLimits(content)) {
+ acceptedEnd = end;
+ acceptedContent = content;
+ low = middle + 1;
+ } else {
+ high = middle - 1;
+ }
+ }
+ if (acceptedEnd <= offset || acceptedContent == null) {
+ throw new IllegalArgumentException(
+ "表格单元格无法在分块预算内安全续片");
+ }
+ parts.add(new RenderedPart(
+ acceptedContent, rowNumber, rowNumber));
+ offset = acceptedEnd;
+ }
+ return parts;
+ }
+
+ /**
+ * 统一单元格换行，避免续片边界拆开 CRLF。
+ *
+ * @param value 原始单元格值
+ * @return 统一换行后的值
+ */
+ private String normalizeCellNewlines(String value) {
+ return safeText(value)
+ .replace("\r\n", "\n")
+ .replace('\r', '\n');
+ }
+
+ /**
+ * 渲染单行 Markdown 表格。
+ *
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @param values 行值
+ * @param rowNumber 逻辑行号
+ * @return Markdown 内容
+ */
+ private String renderSingleRow(
+ String sourceLabel,
+ List headers,
+ List values,
+ int rowNumber) {
+ return renderMarkdown(
+ sourceLabel,
+ headers,
+ Collections.singletonList(new TabularRow(rowNumber, values)));
+ }
+
+ /**
+ * 将表格渲染为 Markdown。
+ *
+ * @param sourceLabel 表格名称
+ * @param headers 表头
+ * @param rows 数据行
+ * @return Markdown 内容
+ */
+ private String renderMarkdown(
+ String sourceLabel,
+ List headers,
+ List rows) {
+ StringBuilder builder = new StringBuilder();
+ builder.append("# ").append(safeText(sourceLabel)).append("\n\n| ");
+ appendCells(builder, headers);
+ builder.append(" |\n|");
+ for (int index = 0; index < headers.size(); index++) {
+ builder.append(" --- |");
+ }
+ for (TabularRow row : rows) {
+ builder.append("\n| ");
+ appendCells(builder, normalizeValues(row.getValues(), headers.size()));
+ builder.append(" |");
+ }
+ return builder.toString();
+ }
+
+ /**
+ * 追加经过 Markdown 转义的单元格。
+ *
+ * @param builder 输出缓冲区
+ * @param cells 单元格值
+ */
+ private void appendCells(StringBuilder builder, List cells) {
+ for (int index = 0; index < cells.size(); index++) {
+ if (index > 0) {
+ builder.append(" | ");
+ }
+ appendEscapedCell(builder, cells.get(index));
+ }
+ }
+
+ /**
+ * 将一行补齐到固定列数。
+ *
+ * @param values 原始值
+ * @param columnCount 列数
+ * @return 补齐后的值
+ */
+ private List normalizeValues(List values, int columnCount) {
+ List normalized = new ArrayList(columnCount);
+ for (int index = 0; index < columnCount; index++) {
+ normalized.add(values != null && index < values.size() && values.get(index) != null
+ ? values.get(index)
+ : "");
+ }
+ return normalized;
+ }
+
+ /**
+ * 转义 Markdown 表格单元格。
+ *
+ * @param value 原始值
+ * @return 转义后的值
+ */
+ private void appendEscapedCell(
+ StringBuilder builder,
+ String value) {
+ String safeValue = safeText(value);
+ for (int index = 0; index < safeValue.length();) {
+ char current = safeValue.charAt(index);
+ if (current == '\r') {
+ builder.append("
");
+ index += index + 1 < safeValue.length()
+ && safeValue.charAt(index + 1) == '\n' ? 2 : 1;
+ continue;
+ }
+ if (current == '\n') {
+ builder.append("
");
+ index++;
+ continue;
+ }
+ if (current == '\\' || current == '|') {
+ builder.append('\\');
+ }
+ int rawChars = rawUnitChars(safeValue, index);
+ builder.append(safeValue, index, index + rawChars);
+ index += rawChars;
+ }
+ }
+
+ /**
+ * 估算多个单元格完成 Markdown 转义后的字符数。
+ *
+ * @param cells 单元格
+ * @return 字符数
+ */
+ private long estimateCellsChars(List cells) {
+ long length = Math.max(0, cells.size() - 1) * 3L;
+ for (String cell : cells) {
+ String value = safeText(cell);
+ for (int index = 0; index < value.length();) {
+ char current = value.charAt(index);
+ if (current == '\r') {
+ length += 5L;
+ index += index + 1 < value.length()
+ && value.charAt(index + 1) == '\n' ? 2 : 1;
+ continue;
+ }
+ if (current == '\n') {
+ length += 5L;
+ index++;
+ continue;
+ }
+ length += current == '\\' || current == '|'
+ ? 2L
+ : rawUnitChars(value, index);
+ index += rawUnitChars(value, index);
+ if (length >= Integer.MAX_VALUE) {
+ return Integer.MAX_VALUE;
+ }
+ }
+ }
+ return length;
+ }
+
+ /**
+ * 估算多个单元格完成 Markdown 转义后的 Token 数。
+ *
+ * @param cells 单元格
+ * @return 保守 Token 数
+ */
+ private long estimateCellsTokens(List cells) {
+ long tokens = Math.max(0, cells.size() - 1) * 3L;
+ for (String cell : cells) {
+ String value = safeText(cell);
+ tokens += BgeM3ChunkSafety.estimateContentTokens(value);
+ for (int index = 0; index < value.length();) {
+ char current = value.charAt(index);
+ if (current == '\r') {
+ boolean crlf = index + 1 < value.length()
+ && value.charAt(index + 1) == '\n';
+ tokens += crlf ? 3L : 4L;
+ index += crlf ? 2 : 1;
+ continue;
+ }
+ if (current == '\n') {
+ tokens += 4L;
+ } else if (current == '\\' || current == '|') {
+ tokens++;
+ }
+ index += rawUnitChars(value, index);
+ }
+ if (tokens >= Integer.MAX_VALUE) {
+ return Integer.MAX_VALUE;
+ }
+ }
+ return tokens;
+ }
+
+ /**
+ * 计算当前 Unicode 单元占用的 Java 字符数。
+ *
+ * @param value 原始值
+ * @param index 当前偏移
+ * @return Java 字符数
+ */
+ private int rawUnitChars(String value, int index) {
+ char current = value.charAt(index);
+ if (Character.isHighSurrogate(current)
+ && index + 1 < value.length()
+ && Character.isLowSurrogate(value.charAt(index + 1))) {
+ return 2;
+ }
+ return 1;
+ }
+
+ /**
+ * 校验表头。
+ *
+ * @param headers 表头
+ */
+ private void validateHeaders(List headers) {
+ if (headers == null || headers.isEmpty()) {
+ throw new IllegalArgumentException("表格分块缺少表头");
+ }
+ }
+
+ /**
+ * 校验数据行。
+ *
+ * @param headers 表头
+ * @param row 数据行
+ */
+ private void validateRow(
+ List headers,
+ TabularRow row) {
+ validateHeaders(headers);
+ if (row == null) {
+ throw new IllegalArgumentException("表格分块缺少数据行");
+ }
+ }
+
+ /**
+ * 判断估算值是否同时满足字符与 Token 上限。
+ *
+ * @param chars 字符数
+ * @param tokens Token 数
+ * @return 是否满足
+ */
+ private boolean fitsLimits(long chars, long tokens) {
+ return chars <= MAX_CHUNK_CONTENT_CHARS
+ && tokens <= RagDefaults.BGE_M3_HARD_CHUNK_TOKEN_LIMIT;
+ }
+
+ /**
+ * 判断实际内容是否同时满足字符与 Token 上限。
+ *
+ * @param content 内容
+ * @return 是否满足
+ */
+ private boolean fitsLimits(String content) {
+ return content.length() <= MAX_CHUNK_CONTENT_CHARS
+ && BgeM3ChunkSafety.isWithinHardLimit(content);
+ }
+
+ /**
+ * 断言实际内容满足字符与 Token 上限。
+ *
+ * @param content 内容
+ */
+ private void assertWithinLimits(String content) {
+ if (!fitsLimits(content)) {
+ throw new IllegalStateException("表格续片超过分块上限");
+ }
+ }
+
+ /**
+ * 将长度安全压缩到整数范围。
+ *
+ * @param value 长度
+ * @return 整数长度
+ */
+ private int boundedInt(long value) {
+ return value >= Integer.MAX_VALUE
+ ? Integer.MAX_VALUE
+ : (int) value;
+ }
+
+ /**
+ * 返回非空文本。
+ *
+ * @param value 原始文本
+ * @return 非空文本
+ */
+ private String safeText(String value) {
+ return value == null ? "" : value;
+ }
+
+ /**
+ * 生成分块主键。
+ *
+ * @param entity 分块实体
+ * @return 主键
+ */
+ private BigInteger generateId(Object entity) {
+ return new BigInteger(String.valueOf(keyGenerator.generate(entity, null)));
+ }
+
+ /**
+ * 已完成渲染的语义续片。
+ */
+ private static final class RenderedPart {
+
+ private final String content;
+ private final int rowStart;
+ private final int rowEnd;
+
+ /**
+ * 创建语义续片。
+ *
+ * @param content Markdown 内容
+ * @param rowStart 起始行
+ * @param rowEnd 结束行
+ */
+ private RenderedPart(
+ String content,
+ int rowStart,
+ int rowEnd) {
+ this.content = content;
+ this.rowStart = rowStart;
+ this.rowEnd = rowEnd;
+ }
+ }
+
+ /**
+ * 表格逻辑数据行。
+ */
+ public static class TabularRow {
+
+ private final int rowNumber;
+ private final List values;
+
+ /**
+ * 创建表格数据行。
+ *
+ * @param rowNumber 一基逻辑行号
+ * @param values 单元格值
+ */
+ public TabularRow(int rowNumber, List values) {
+ this.rowNumber = rowNumber;
+ this.values = values == null
+ ? Collections.emptyList()
+ : new ArrayList(values);
+ }
+
+ /**
+ * 返回逻辑行号。
+ *
+ * @return 一基逻辑行号
+ */
+ public int getRowNumber() {
+ return rowNumber;
+ }
+
+ /**
+ * 返回单元格值。
+ *
+ * @return 不可变语义的单元格值副本
+ */
+ public List getValues() {
+ return new ArrayList(values);
+ }
+ }
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentImportSnapshotCleanup.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentImportSnapshotCleanup.java
new file mode 100644
index 00000000..4e14d6f3
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/DocumentImportSnapshotCleanup.java
@@ -0,0 +1,325 @@
+package tech.easyflow.ai.entity;
+
+import com.mybatisflex.annotation.Column;
+import com.mybatisflex.annotation.Id;
+import com.mybatisflex.annotation.KeyType;
+import com.mybatisflex.annotation.Table;
+import tech.easyflow.common.entity.DateEntity;
+
+import java.io.Serializable;
+import java.math.BigInteger;
+import java.util.Date;
+
+/**
+ * 文档导入中间快照清理记录。
+ *
+ * 记录精确清单路径、清理阶段和租约，使对象存储删除失败后可以跨进程重试。
+ *
+ * @author Codex
+ * @since 2026-08-04
+ */
+@Table(
+ value = "tb_document_import_snapshot_cleanup",
+ comment = "文档导入中间快照清理记录")
+public class DocumentImportSnapshotCleanup
+ extends DateEntity
+ implements Serializable {
+
+ @Id(keyType = KeyType.Generator, value = "snowFlakeId")
+ private BigInteger id;
+
+ @Column(comment = "知识库ID")
+ private BigInteger knowledgeId;
+
+ @Column(comment = "文档ID")
+ private BigInteger documentId;
+
+ @Column(comment = "快照类型")
+ private String snapshotType;
+
+ @Column(comment = "快照清单路径")
+ private String manifestPath;
+
+ @Column(comment = "快照清单路径SHA-256")
+ private String pathHash;
+
+ @Column(comment = "清理阶段")
+ private String phase;
+
+ @Column(comment = "已尝试次数")
+ private Integer attemptCount;
+
+ @Column(comment = "下次重试时间")
+ private Date nextRetryAt;
+
+ @Column(comment = "执行令牌")
+ private String executionToken;
+
+ @Column(comment = "租约截止时间")
+ private Date leaseUntil;
+
+ @Column(comment = "最近错误")
+ private String lastError;
+
+ @Column(comment = "创建时间")
+ private Date created;
+
+ @Column(comment = "修改时间")
+ private Date modified;
+
+ /**
+ * 返回主键。
+ *
+ * @return 主键
+ */
+ public BigInteger getId() {
+ return id;
+ }
+
+ /**
+ * 设置主键。
+ *
+ * @param id 主键
+ */
+ public void setId(BigInteger id) {
+ this.id = id;
+ }
+
+ /**
+ * 返回知识库 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 getSnapshotType() {
+ return snapshotType;
+ }
+
+ /**
+ * 设置快照类型。
+ *
+ * @param snapshotType 快照类型
+ */
+ public void setSnapshotType(String snapshotType) {
+ this.snapshotType = snapshotType;
+ }
+
+ /**
+ * 返回清单路径。
+ *
+ * @return 清单路径
+ */
+ public String getManifestPath() {
+ return manifestPath;
+ }
+
+ /**
+ * 设置清单路径。
+ *
+ * @param manifestPath 清单路径
+ */
+ public void setManifestPath(String manifestPath) {
+ this.manifestPath = manifestPath;
+ }
+
+ /**
+ * 返回路径摘要。
+ *
+ * @return 路径摘要
+ */
+ public String getPathHash() {
+ return pathHash;
+ }
+
+ /**
+ * 设置路径摘要。
+ *
+ * @param pathHash 路径摘要
+ */
+ public void setPathHash(String pathHash) {
+ this.pathHash = pathHash;
+ }
+
+ /**
+ * 返回清理阶段。
+ *
+ * @return 清理阶段
+ */
+ public String getPhase() {
+ return phase;
+ }
+
+ /**
+ * 设置清理阶段。
+ *
+ * @param phase 清理阶段
+ */
+ public void setPhase(String phase) {
+ this.phase = phase;
+ }
+
+ /**
+ * 返回已尝试次数。
+ *
+ * @return 已尝试次数
+ */
+ public Integer getAttemptCount() {
+ return attemptCount;
+ }
+
+ /**
+ * 设置已尝试次数。
+ *
+ * @param attemptCount 已尝试次数
+ */
+ public void setAttemptCount(Integer attemptCount) {
+ this.attemptCount = attemptCount;
+ }
+
+ /**
+ * 返回下次重试时间。
+ *
+ * @return 下次重试时间
+ */
+ public Date getNextRetryAt() {
+ return nextRetryAt;
+ }
+
+ /**
+ * 设置下次重试时间。
+ *
+ * @param nextRetryAt 下次重试时间
+ */
+ public void setNextRetryAt(Date nextRetryAt) {
+ this.nextRetryAt = nextRetryAt;
+ }
+
+ /**
+ * 返回执行令牌。
+ *
+ * @return 执行令牌
+ */
+ public String getExecutionToken() {
+ return executionToken;
+ }
+
+ /**
+ * 设置执行令牌。
+ *
+ * @param executionToken 执行令牌
+ */
+ public void setExecutionToken(String executionToken) {
+ this.executionToken = executionToken;
+ }
+
+ /**
+ * 返回租约截止时间。
+ *
+ * @return 租约截止时间
+ */
+ public Date getLeaseUntil() {
+ return leaseUntil;
+ }
+
+ /**
+ * 设置租约截止时间。
+ *
+ * @param leaseUntil 租约截止时间
+ */
+ public void setLeaseUntil(Date leaseUntil) {
+ this.leaseUntil = leaseUntil;
+ }
+
+ /**
+ * 返回最近错误。
+ *
+ * @return 最近错误
+ */
+ public String getLastError() {
+ return lastError;
+ }
+
+ /**
+ * 设置最近错误。
+ *
+ * @param lastError 最近错误
+ */
+ public void setLastError(String lastError) {
+ this.lastError = lastError;
+ }
+
+ /**
+ * 返回创建时间。
+ *
+ * @return 创建时间
+ */
+ @Override
+ public Date getCreated() {
+ return created;
+ }
+
+ /**
+ * 设置创建时间。
+ *
+ * @param created 创建时间
+ */
+ @Override
+ public void setCreated(Date created) {
+ this.created = created;
+ }
+
+ /**
+ * 返回修改时间。
+ *
+ * @return 修改时间
+ */
+ @Override
+ public Date getModified() {
+ return modified;
+ }
+
+ /**
+ * 设置修改时间。
+ *
+ * @param modified 修改时间
+ */
+ @Override
+ public void setModified(Date modified) {
+ this.modified = modified;
+ }
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportSnapshotCleanupMapper.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportSnapshotCleanupMapper.java
new file mode 100644
index 00000000..3468c5b8
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentImportSnapshotCleanupMapper.java
@@ -0,0 +1,125 @@
+package tech.easyflow.ai.mapper;
+
+import com.mybatisflex.core.BaseMapper;
+import org.apache.ibatis.annotations.Delete;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+import org.apache.ibatis.annotations.Update;
+import tech.easyflow.ai.entity.DocumentImportSnapshotCleanup;
+
+import java.math.BigInteger;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 文档导入中间快照清理记录映射层。
+ *
+ * @author Codex
+ * @since 2026-08-04
+ */
+public interface DocumentImportSnapshotCleanupMapper
+ extends BaseMapper {
+
+ /**
+ * 幂等登记一个快照清理请求。
+ *
+ * @param record 清理记录
+ * @return 新增行数
+ */
+ @Insert("INSERT IGNORE INTO tb_document_import_snapshot_cleanup "
+ + "(id, knowledge_id, document_id, snapshot_type, manifest_path, "
+ + "path_hash, phase, attempt_count, next_retry_at, created, modified) "
+ + "VALUES (#{id}, #{knowledgeId}, #{documentId}, #{snapshotType}, "
+ + "#{manifestPath}, #{pathHash}, #{phase}, #{attemptCount}, "
+ + "#{nextRetryAt}, #{created}, #{modified})")
+ int insertIgnore(DocumentImportSnapshotCleanup record);
+
+ /**
+ * 查询到期且未被有效租约持有的清理记录。
+ *
+ * @param now 当前时间
+ * @param limit 查询上限
+ * @return 待处理记录
+ */
+ @Select("SELECT * FROM tb_document_import_snapshot_cleanup "
+ + "WHERE next_retry_at <= #{now} "
+ + "AND (execution_token IS NULL OR lease_until < #{now}) "
+ + "ORDER BY next_retry_at, id LIMIT #{limit}")
+ List selectDueRecords(
+ @Param("now") Date now,
+ @Param("limit") int limit);
+
+ /**
+ * 原子领取一条清理记录。
+ *
+ * @param id 记录 ID
+ * @param executionToken 执行令牌
+ * @param leaseUntil 租约截止时间
+ * @param now 当前时间
+ * @return 更新行数
+ */
+ @Update("UPDATE tb_document_import_snapshot_cleanup SET "
+ + "execution_token=#{executionToken}, lease_until=#{leaseUntil}, "
+ + "attempt_count=attempt_count + 1, modified=#{now} "
+ + "WHERE id=#{id} AND next_retry_at <= #{now} "
+ + "AND (execution_token IS NULL OR lease_until < #{now})")
+ int claim(
+ @Param("id") BigInteger id,
+ @Param("executionToken") String executionToken,
+ @Param("leaseUntil") Date leaseUntil,
+ @Param("now") Date now);
+
+ /**
+ * 在分片全部删除后持久化推进到清单删除阶段。
+ *
+ * @param id 记录 ID
+ * @param executionToken 执行令牌
+ * @param modified 修改时间
+ * @return 更新行数
+ */
+ @Update("UPDATE tb_document_import_snapshot_cleanup SET "
+ + "phase='MANIFEST_PENDING', last_error=NULL, modified=#{modified} "
+ + "WHERE id=#{id} AND execution_token=#{executionToken} "
+ + "AND phase='PARTS_PENDING'")
+ int advanceToManifest(
+ @Param("id") BigInteger id,
+ @Param("executionToken") String executionToken,
+ @Param("modified") Date modified);
+
+ /**
+ * 释放失败记录并安排下一次重试。
+ *
+ * @param id 记录 ID
+ * @param executionToken 执行令牌
+ * @param nextRetryAt 下次重试时间
+ * @param lastError 最近错误
+ * @param modified 修改时间
+ * @return 更新行数
+ */
+ @Update("UPDATE tb_document_import_snapshot_cleanup SET "
+ + "execution_token=NULL, lease_until=NULL, "
+ + "next_retry_at=#{nextRetryAt}, last_error=#{lastError}, "
+ + "modified=#{modified} "
+ + "WHERE id=#{id} AND execution_token=#{executionToken}")
+ int releaseForRetry(
+ @Param("id") BigInteger id,
+ @Param("executionToken") String executionToken,
+ @Param("nextRetryAt") Date nextRetryAt,
+ @Param("lastError") String lastError,
+ @Param("modified") Date modified);
+
+ /**
+ * 删除当前执行令牌已完成的清理记录。
+ *
+ * @param id 记录 ID
+ * @param executionToken 执行令牌
+ * @return 删除行数
+ */
+ @Delete("DELETE FROM tb_document_import_snapshot_cleanup "
+ + "WHERE id=#{id} AND execution_token=#{executionToken} "
+ + "AND phase='MANIFEST_PENDING'")
+ int deleteCompleted(
+ @Param("id") BigInteger id,
+ @Param("executionToken") String executionToken);
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentMapper.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentMapper.java
index 8cceb84f..e6c2defe 100644
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentMapper.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/mapper/DocumentMapper.java
@@ -1,7 +1,12 @@
package tech.easyflow.ai.mapper;
-import tech.easyflow.ai.entity.Document;
import com.mybatisflex.core.BaseMapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Update;
+import tech.easyflow.ai.entity.Document;
+
+import java.math.BigInteger;
+import java.util.Date;
/**
* 映射层。
@@ -11,4 +16,37 @@ import com.mybatisflex.core.BaseMapper;
*/
public interface DocumentMapper extends BaseMapper {
+ /**
+ * 在路径仍指向已清理 CSV 快照时原子移除数据库指针。
+ *
+ * @param documentId 文档 ID
+ * @param manifestPath 已清理清单路径
+ * @param modified 修改时间
+ * @return 更新行数
+ */
+ @Update("UPDATE tb_document SET options=CASE "
+ + "WHEN JSON_UNQUOTE(JSON_EXTRACT(options, "
+ + "'$.\"parse.csvTableSnapshotPath\"'))=#{manifestPath} "
+ + "AND JSON_UNQUOTE(JSON_EXTRACT(options, "
+ + "'$.\"parse.metadata\".\"parse.csvTableSnapshotPath\"'))="
+ + "#{manifestPath} THEN JSON_REMOVE(options, "
+ + "'$.\"parse.csvTableSnapshotPath\"', "
+ + "'$.\"parse.metadata\".\"parse.csvTableSnapshotPath\"') "
+ + "WHEN JSON_UNQUOTE(JSON_EXTRACT(options, "
+ + "'$.\"parse.csvTableSnapshotPath\"'))=#{manifestPath} "
+ + "THEN JSON_REMOVE(options, "
+ + "'$.\"parse.csvTableSnapshotPath\"') "
+ + "ELSE JSON_REMOVE(options, "
+ + "'$.\"parse.metadata\".\"parse.csvTableSnapshotPath\"') END, "
+ + "modified=#{modified} "
+ + "WHERE id=#{documentId} AND JSON_VALID(options)=1 AND ("
+ + "JSON_UNQUOTE(JSON_EXTRACT(options, "
+ + "'$.\"parse.csvTableSnapshotPath\"'))=#{manifestPath} OR "
+ + "JSON_UNQUOTE(JSON_EXTRACT(options, "
+ + "'$.\"parse.metadata\".\"parse.csvTableSnapshotPath\"'))="
+ + "#{manifestPath})")
+ int clearCsvSnapshotPath(
+ @Param("documentId") BigInteger documentId,
+ @Param("manifestPath") String manifestPath,
+ @Param("modified") Date modified);
}
diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java
index d74da2cd..fbb8ab18 100644
--- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java
+++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/DocumentServiceImpl.java
@@ -34,6 +34,7 @@ import tech.easyflow.ai.config.SearcherFactory;
import tech.easyflow.ai.documentimport.DocumentImportDtos;
import tech.easyflow.ai.documentimport.DocumentImportKeys;
import tech.easyflow.ai.documentimport.DocumentImportPreviewService;
+import tech.easyflow.ai.documentimport.task.DocumentImportSnapshotCleanupService;
import tech.easyflow.ai.documentimport.task.KnowledgeDocumentImportTaskAppService;
import tech.easyflow.ai.entity.*;
import tech.easyflow.ai.enums.DocumentProcessStatus;
@@ -101,6 +102,9 @@ public class DocumentServiceImpl extends ServiceImpl i
@Autowired
private DocumentImportPreviewService documentImportPreviewService;
+ @Autowired
+ private DocumentImportSnapshotCleanupService snapshotCleanupService;
+
@Autowired
private KnowledgeDocumentImportTaskAppService importTaskAppService;
@@ -240,7 +244,18 @@ public class DocumentServiceImpl extends ServiceImpl i
: asString(oneByQuery.getOptions().get(
DocumentImportKeys.KEY_DOCUMENT_CHUNK_SNAPSHOT_PATH));
if (StringUtil.hasText(chunkSnapshotPath)) {
- storageService.delete(chunkSnapshotPath);
+ snapshotCleanupService.scheduleChunkSnapshot(
+ chunkSnapshotPath);
+ }
+ String csvTableSnapshotPath = oneByQuery.getOptions() == null
+ ? null
+ : asString(oneByQuery.getOptions().get(
+ DocumentImportKeys.KEY_DOCUMENT_CSV_TABLE_SNAPSHOT_PATH));
+ if (StringUtil.hasText(csvTableSnapshotPath)) {
+ snapshotCleanupService.scheduleCsvTableSnapshot(
+ oneByQuery.getCollectionId(),
+ oneByQuery.getId(),
+ csvTableSnapshotPath);
}
storageService.delete(oneByQuery.getDocumentPath());
// 主记录必须最后删除；否则接口返回成功后文档仍会出现在列表中。
diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/CsvImportSupportPolicyTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/CsvImportSupportPolicyTest.java
new file mode 100644
index 00000000..b6cd5477
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/CsvImportSupportPolicyTest.java
@@ -0,0 +1,67 @@
+package tech.easyflow.ai.documentimport.task;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Set;
+
+/**
+ * CSV 导入入口支持策略测试。
+ */
+public class CsvImportSupportPolicyTest {
+
+ /**
+ * 验证单文件、管理端批次和 Public API 批次三条入口都允许 CSV。
+ *
+ * @throws Exception 反射读取失败
+ */
+ @Test
+ public void shouldAllowCsvAcrossAllKnowledgeImportEntrypoints()
+ throws Exception {
+ KnowledgeDocumentImportTaskAppService taskService =
+ new KnowledgeDocumentImportTaskAppService();
+ Method assertSupported = KnowledgeDocumentImportTaskAppService.class
+ .getDeclaredMethod("assertSupportedImportFile", String.class);
+ assertSupported.setAccessible(true);
+ Method normalizeExtension =
+ KnowledgeDocumentImportTaskAppService.class.getDeclaredMethod(
+ "normalizeFileExtension", String.class, String.class);
+ normalizeExtension.setAccessible(true);
+ String lowerCaseExtension = (String) normalizeExtension.invoke(
+ taskService, "results.csv", "/knowledge/results.csv");
+ String upperCaseExtension = (String) normalizeExtension.invoke(
+ taskService, "RESULTS.CSV", "/knowledge/RESULTS.CSV");
+ Assert.assertEquals("csv", lowerCaseExtension);
+ Assert.assertEquals("csv", upperCaseExtension);
+ assertSupported.invoke(taskService, lowerCaseExtension);
+ assertSupported.invoke(taskService, upperCaseExtension);
+
+ Assert.assertTrue(readSupportedExtensions(
+ DocumentImportBatchAppService.class).contains("csv"));
+ Assert.assertTrue(readSupportedExtensions(
+ KnowledgeImportBatchFacade.class).contains("csv"));
+ Assert.assertSame(
+ DocumentImportFormatPolicy.supportedExtensions(),
+ readSupportedExtensions(DocumentImportBatchAppService.class));
+ Assert.assertSame(
+ DocumentImportFormatPolicy.supportedExtensions(),
+ readSupportedExtensions(KnowledgeImportBatchFacade.class));
+ }
+
+ /**
+ * 读取入口类的支持扩展名集合。
+ *
+ * @param type 入口类
+ * @return 扩展名集合
+ * @throws Exception 字段读取失败
+ */
+ @SuppressWarnings("unchecked")
+ private Set readSupportedExtensions(Class type)
+ throws Exception {
+ Field field = type.getDeclaredField("SUPPORTED_EXTENSIONS");
+ field.setAccessible(true);
+ return (Set) field.get(null);
+ }
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/CsvTableSnapshotServiceTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/CsvTableSnapshotServiceTest.java
new file mode 100644
index 00000000..1d007fc2
--- /dev/null
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/CsvTableSnapshotServiceTest.java
@@ -0,0 +1,361 @@
+package tech.easyflow.ai.documentimport.task;
+
+import com.easyagents.rag.core.BgeM3ChunkSafety;
+import com.easyagents.rag.ingestion.model.StrategyConfig;
+import org.junit.Assert;
+import org.junit.Test;
+import tech.easyflow.ai.documentimport.DocumentImportDtos;
+import tech.easyflow.ai.entity.DocumentChunk;
+
+import java.lang.reflect.Field;
+import java.math.BigInteger;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+/**
+ * {@link CsvTableSnapshotService} 流式解析与分块测试。
+ */
+public class CsvTableSnapshotServiceTest {
+
+ /**
+ * 验证引号逗号、跨行字段、尾空列和重复表头都能稳定解析并按行窗口分块。
+ *
+ * @throws Exception 反射注入失败
+ */
+ @Test
+ public void shouldParseRfc4180AndBuildRowWindowChunks() throws Exception {
+ InMemoryFileStorageService storage = new InMemoryFileStorageService();
+ String sourcePath = "source/demo.csv";
+ String csv = "姓名,备注,姓名,\r\n"
+ + "张三,\"第一行\n第二行\",A,\r\n"
+ + "李四,\"含,逗号\",B,\r\n";
+ storage.put(sourcePath, csv.getBytes(StandardCharsets.UTF_8));
+ CsvTableSnapshotService service = createService(storage);
+
+ CsvTableSnapshotService.CsvParseResult result = service.parse(
+ sourcePath, BigInteger.valueOf(7), BigInteger.valueOf(9), "token-1");
+
+ Assert.assertEquals("UTF-8", result.getEncoding());
+ Assert.assertEquals(2L, result.getRowCount());
+ Assert.assertEquals(4, result.getColumnCount());
+ Assert.assertEquals(
+ List.of("姓名", "备注", "姓名_2", "列_4"), result.getHeaders());
+
+ tech.easyflow.ai.entity.Document document =
+ new tech.easyflow.ai.entity.Document();
+ document.setId(BigInteger.valueOf(9));
+ document.setCollectionId(BigInteger.valueOf(7));
+ document.setTitle("demo.csv");
+ document.setDocumentPath(sourcePath);
+ StrategyConfig strategy = StrategyConfig.defaults();
+ strategy.setStrategyCode("TABLE_ROW");
+ strategy.setRowsPerChunk(1);
+
+ DocumentImportDtos.PreviewSession session =
+ service.buildChunkSnapshot(
+ document, result.getManifestPath(), strategy);
+
+ Assert.assertEquals(Integer.valueOf(2), session.getTotalChunks());
+ Assert.assertEquals(2, session.getDocumentChunks().size());
+ Assert.assertTrue(session.getDocumentChunks().get(0).getContent()
+ .contains("第一行
第二行"));
+ Assert.assertTrue(session.getDocumentChunks().get(1).getContent()
+ .contains("含,逗号"));
+ Assert.assertTrue(session.getDocumentChunks().stream()
+ .allMatch(chunk -> "TABLE_ROW".equals(
+ chunk.getOptions().get("chunkType"))));
+ }
+
+ /**
+ * 验证无 BOM 且非 UTF-8 的 CSV 会重新打开源文件并回退到 GB18030。
+ *
+ * @throws Exception 反射注入失败
+ */
+ @Test
+ public void shouldFallbackToGb18030AfterStrictUtf8Failure() throws Exception {
+ InMemoryFileStorageService storage = new InMemoryFileStorageService();
+ String sourcePath = "source/gb.csv";
+ storage.put(
+ sourcePath,
+ "名称,说明\r\n测试,中文内容\r\n".getBytes(Charset.forName("GB18030")));
+ CsvTableSnapshotService service = createService(storage);
+
+ CsvTableSnapshotService.CsvParseResult result = service.parse(
+ sourcePath, BigInteger.valueOf(7), BigInteger.valueOf(10), "token-2");
+
+ Assert.assertEquals("GB18030", result.getEncoding());
+ Assert.assertEquals(1L, result.getRowCount());
+ }
+
+ /**
+ * 验证数据行列数与表头不一致时返回稳定失败码。
+ *
+ * @throws Exception 反射注入失败
+ */
+ @Test
+ public void shouldRejectColumnMismatchWithStableFailureCode() throws Exception {
+ InMemoryFileStorageService storage = new InMemoryFileStorageService();
+ String sourcePath = "source/broken.csv";
+ storage.put(
+ sourcePath,
+ "a,b\r\n1,2,3\r\n".getBytes(StandardCharsets.UTF_8));
+ CsvTableSnapshotService service = createService(storage);
+
+ try {
+ service.parse(
+ sourcePath, BigInteger.valueOf(7), BigInteger.valueOf(11), "token-3");
+ Assert.fail("应拒绝列数不一致的 CSV");
+ } catch (CsvImportException error) {
+ Assert.assertEquals(
+ CsvTableSnapshotService.FAILURE_COLUMN_MISMATCH,
+ error.getFailureCode());
+ Assert.assertTrue(error.getMessage().contains("列数"));
+ }
+ }
+
+ /**
+ * 验证二进制伪装 CSV 中的 NUL 字节会被明确拒绝。
+ *
+ * @throws Exception 反射注入失败
+ */
+ @Test
+ public void shouldRejectNulByteAsMalformedCsv() throws Exception {
+ InMemoryFileStorageService storage = new InMemoryFileStorageService();
+ String sourcePath = "source/binary.csv";
+ storage.put(
+ sourcePath,
+ "name,value\nalice,\0binary\n".getBytes(StandardCharsets.UTF_8));
+ CsvTableSnapshotService service = createService(storage);
+
+ try {
+ service.parse(
+ sourcePath, BigInteger.valueOf(7), BigInteger.valueOf(12), "token-4");
+ Assert.fail("包含 NUL 字节的文件不应进入后续阶段");
+ } catch (CsvImportException error) {
+ Assert.assertEquals(
+ CsvTableSnapshotService.FAILURE_MALFORMED,
+ error.getFailureCode());
+ }
+ }
+
+ /**
+ * 验证超长表格正文会生成多个不超过数据库上限的续片。
+ */
+ @Test
+ public void shouldSplitOversizedWindowWithoutSilentTruncation() {
+ TabularRowWindowChunkBuilder builder =
+ new TabularRowWindowChunkBuilder();
+ String longValue = "长".repeat(
+ TabularRowWindowChunkBuilder.MAX_CHUNK_CONTENT_CHARS + 100);
+
+ List chunks = builder.build(
+ BigInteger.ONE,
+ BigInteger.valueOf(2),
+ "大表",
+ List.of("正文"),
+ List.of(new TabularRowWindowChunkBuilder.TabularRow(
+ 2, List.of(longValue))),
+ 1,
+ "TABLE_ROW");
+
+ Assert.assertTrue(chunks.size() > 1);
+ Assert.assertTrue(chunks.stream().allMatch(
+ chunk -> chunk.getContent().length()
+ <= TabularRowWindowChunkBuilder.MAX_CHUNK_CONTENT_CHARS));
+ Assert.assertTrue(chunks.stream().allMatch(
+ chunk -> BgeM3ChunkSafety.isWithinHardLimit(chunk.getContent())));
+ Assert.assertTrue(chunks.stream().allMatch(
+ chunk -> chunk.getContent().startsWith("# 大表")
+ && chunk.getContent().contains("| 正文 |")));
+ Assert.assertTrue(chunks.stream().allMatch(
+ chunk -> Integer.valueOf(2).equals(
+ chunk.getOptions().get("rowStart"))
+ && Integer.valueOf(2).equals(
+ chunk.getOptions().get("rowEnd"))));
+ long restoredChars = chunks.stream()
+ .map(DocumentChunk::getContent)
+ .flatMapToInt(String::chars)
+ .filter(value -> value == '长')
+ .count();
+ Assert.assertEquals(longValue.length(), restoredChars);
+ Assert.assertEquals(
+ chunks.size(), chunks.get(0).getOptions().get("partTotal"));
+ }
+
+ /**
+ * 验证超长记录优先按列边界续片，且每个续片都保留标题和列名。
+ */
+ @Test
+ public void shouldSplitOversizedRowAtColumnBoundaries() {
+ TabularRowWindowChunkBuilder builder =
+ new TabularRowWindowChunkBuilder();
+ String longValue = "长".repeat(20_000);
+
+ List chunks = builder.build(
+ BigInteger.ONE,
+ BigInteger.valueOf(2),
+ "列边界表",
+ List.of("左列", "正文", "右列"),
+ List.of(new TabularRowWindowChunkBuilder.TabularRow(
+ 8, List.of("LEFT_MARKER", longValue, "RIGHT_MARKER"))),
+ 1,
+ "TABLE_ROW");
+
+ Assert.assertTrue(chunks.size() > 2);
+ Assert.assertTrue(chunks.stream().allMatch(
+ chunk -> chunk.getContent().startsWith("# 列边界表")
+ && chunk.getContent().contains("\n| ")));
+ String allContent = chunks.stream()
+ .map(DocumentChunk::getContent)
+ .reduce("", String::concat);
+ Assert.assertEquals(1, occurrences(allContent, "LEFT_MARKER"));
+ Assert.assertEquals(1, occurrences(allContent, "RIGHT_MARKER"));
+ Assert.assertEquals(
+ longValue.length(),
+ allContent.chars().filter(value -> value == '长').count());
+ Assert.assertTrue(chunks.stream().allMatch(
+ chunk -> Integer.valueOf(8).equals(
+ chunk.getOptions().get("rowStart"))
+ && Integer.valueOf(8).equals(
+ chunk.getOptions().get("rowEnd"))));
+ }
+
+ /**
+ * 验证行数未满时仍会在加入下一行超过 Token 上限前闭合窗口。
+ *
+ * @throws Exception 反射注入失败
+ */
+ @Test
+ public void shouldCloseWindowBeforeNextRowExceedsTokenLimit()
+ throws Exception {
+ InMemoryFileStorageService storage =
+ new InMemoryFileStorageService();
+ String sourcePath = "source/token-window.csv";
+ String value = "中".repeat(5_000);
+ storage.put(
+ sourcePath,
+ ("正文\n" + value + "\n" + value + "\n")
+ .getBytes(StandardCharsets.UTF_8));
+ CsvTableSnapshotService service = createService(storage);
+ CsvTableSnapshotService.CsvParseResult result = service.parse(
+ sourcePath,
+ BigInteger.valueOf(7),
+ BigInteger.valueOf(13),
+ "token-window");
+ tech.easyflow.ai.entity.Document document =
+ new tech.easyflow.ai.entity.Document();
+ document.setId(BigInteger.valueOf(13));
+ document.setCollectionId(BigInteger.valueOf(7));
+ document.setTitle("token-window.csv");
+ StrategyConfig strategy = StrategyConfig.defaults();
+ strategy.setStrategyCode("TABLE_ROW");
+ strategy.setRowsPerChunk(10);
+
+ DocumentImportDtos.PreviewSession session =
+ service.buildChunkSnapshot(
+ document, result.getManifestPath(), strategy);
+
+ Assert.assertEquals(Integer.valueOf(2), session.getTotalChunks());
+ Assert.assertEquals(
+ List.of(2, 3),
+ session.getDocumentChunks().stream()
+ .map(chunk -> (Integer) chunk.getOptions().get("rowStart"))
+ .toList());
+ }
+
+ /**
+ * 验证 CSV 行分片删除失败时清单保持可重试。
+ *
+ * @throws Exception 反射注入失败
+ */
+ @Test
+ public void shouldKeepCsvManifestWhenPartDeletionFails()
+ throws Exception {
+ InMemoryFileStorageService storage =
+ new InMemoryFileStorageService();
+ String sourcePath = "source/cleanup.csv";
+ storage.put(
+ sourcePath,
+ "a,b\n1,2\n".getBytes(StandardCharsets.UTF_8));
+ CsvTableSnapshotService service = createService(storage);
+ CsvTableSnapshotService.CsvParseResult result = service.parse(
+ sourcePath,
+ BigInteger.valueOf(7),
+ BigInteger.valueOf(14),
+ "cleanup");
+ String manifestPath = result.getManifestPath();
+ String partPath = storage.paths().stream()
+ .filter(path -> !path.equals(sourcePath))
+ .filter(path -> !path.equals(manifestPath))
+ .findFirst()
+ .orElseThrow();
+ storage.failDelete(partPath, 1);
+
+ try {
+ service.delete(manifestPath);
+ Assert.fail("CSV 行分片删除失败时应抛出异常");
+ } catch (IllegalStateException expected) {
+ Assert.assertTrue(expected.getMessage().contains("模拟删除失败"));
+ }
+ Assert.assertTrue(storage.contains(manifestPath));
+ Assert.assertTrue(storage.contains(partPath));
+
+ service.delete(manifestPath);
+ Assert.assertEquals(1, storage.size());
+ Assert.assertTrue(storage.contains(sourcePath));
+ }
+
+ /**
+ * 统计文本出现次数。
+ *
+ * @param content 完整文本
+ * @param target 目标文本
+ * @return 出现次数
+ */
+ private int occurrences(String content, String target) {
+ int count = 0;
+ int offset = 0;
+ while ((offset = content.indexOf(target, offset)) >= 0) {
+ count++;
+ offset += target.length();
+ }
+ return count;
+ }
+
+ /**
+ * 创建完成依赖注入的被测服务。
+ *
+ * @param storage 内存文件存储
+ * @return CSV 服务
+ * @throws Exception 反射注入失败
+ */
+ private CsvTableSnapshotService createService(
+ InMemoryFileStorageService storage) throws Exception {
+ DocumentImportChunkSnapshotService chunkSnapshotService =
+ new DocumentImportChunkSnapshotService();
+ setField(chunkSnapshotService, "storageService", storage);
+
+ CsvTableSnapshotService service = new CsvTableSnapshotService();
+ setField(service, "storageService", storage);
+ setField(service, "chunkSnapshotService", chunkSnapshotService);
+ return service;
+ }
+
+ /**
+ * 反射设置测试字段。
+ *
+ * @param target 目标对象
+ * @param fieldName 字段名
+ * @param value 字段值
+ * @throws Exception 字段不存在或不可访问
+ */
+ private void setField(
+ Object target,
+ String fieldName,
+ Object value) throws Exception {
+ Field field = target.getClass().getDeclaredField(fieldName);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+}
diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotServiceTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotServiceTest.java
index 32d93f43..70a83b2b 100644
--- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotServiceTest.java
+++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportChunkSnapshotServiceTest.java
@@ -2,17 +2,13 @@ package tech.easyflow.ai.documentimport.task;
import org.junit.Assert;
import org.junit.Test;
-import org.mockito.Mockito;
-import org.springframework.web.multipart.MultipartFile;
import tech.easyflow.ai.documentimport.DocumentImportDtos;
import tech.easyflow.ai.entity.DocumentChunk;
-import tech.easyflow.common.filestorage.FileStorageService;
-import java.io.ByteArrayInputStream;
import java.lang.reflect.Field;
import java.math.BigInteger;
+import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.atomic.AtomicReference;
/**
* {@link DocumentImportChunkSnapshotService} 持久化恢复测试。
@@ -26,19 +22,8 @@ public class DocumentImportChunkSnapshotServiceTest {
*/
@Test
public void shouldPersistAndRestoreChunkSnapshot() throws Exception {
- String storedPath = "http://localhost/snapshots/9-chunks.json";
- AtomicReference storedBytes = new AtomicReference