feat: 支持知识库 CSV 大文件导入

- 增加 CSV 流式解析、表格语义分块和分页预览

- 增加快照两阶段清理、失败重试和格式校验

- 补充批量入口、管理端交互和回归测试
This commit is contained in:
2026-08-07 13:11:59 +08:00
parent 13dec6c216
commit 0d14f1c165
32 changed files with 5672 additions and 156 deletions

View File

@@ -84,6 +84,10 @@
<groupId>org.jsoup</groupId>
<artifactId>jsoup</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-csv</artifactId>
</dependency>
<dependency>
<groupId>org.commonmark</groupId>
<artifactId>commonmark-ext-gfm-tables</artifactId>

View File

@@ -95,6 +95,9 @@ public final class DocumentImportDtos {
private BigInteger knowledgeId;
private BigInteger documentId;
private List<PreviewFileRequest> files = new ArrayList<PreviewFileRequest>();
private String previewSessionId;
private Integer pageNo;
private Integer pageSize;
public BigInteger getKnowledgeId() {
return knowledgeId;
@@ -119,6 +122,60 @@ public final class DocumentImportDtos {
public void setFiles(List<PreviewFileRequest> files) {
this.files = files;
}
/**
* 返回已有预览会话 ID。
*
* @return 预览会话 ID
*/
public String getPreviewSessionId() {
return previewSessionId;
}
/**
* 设置已有预览会话 ID用于只读取指定分页而不重新分块。
*
* @param previewSessionId 预览会话 ID
*/
public void setPreviewSessionId(String previewSessionId) {
this.previewSessionId = previewSessionId;
}
/**
* 返回预览页码。
*
* @return 从 1 开始的页码
*/
public Integer getPageNo() {
return pageNo;
}
/**
* 设置预览页码。
*
* @param pageNo 从 1 开始的页码
*/
public void setPageNo(Integer pageNo) {
this.pageNo = pageNo;
}
/**
* 返回单页分块数。
*
* @return 单页分块数
*/
public Integer getPageSize() {
return pageSize;
}
/**
* 设置单页分块数。
*
* @param pageSize 单页分块数
*/
public void setPageSize(Integer pageSize) {
this.pageSize = pageSize;
}
}
public static class CommitRequest implements Serializable {
@@ -419,6 +476,8 @@ public final class DocumentImportDtos {
private AnalysisResult analysis;
private Integer totalChunks;
private Integer totalWarnings;
private Integer pageNo;
private Integer pageSize;
private List<PreviewChunkResult> chunks = new ArrayList<PreviewChunkResult>();
public String getPreviewSessionId() {
@@ -493,6 +552,42 @@ public final class DocumentImportDtos {
this.totalWarnings = totalWarnings;
}
/**
* 返回当前预览页码。
*
* @return 从 1 开始的页码
*/
public Integer getPageNo() {
return pageNo;
}
/**
* 设置当前预览页码。
*
* @param pageNo 从 1 开始的页码
*/
public void setPageNo(Integer pageNo) {
this.pageNo = pageNo;
}
/**
* 返回当前预览页大小。
*
* @return 单页分块数
*/
public Integer getPageSize() {
return pageSize;
}
/**
* 设置当前预览页大小。
*
* @param pageSize 单页分块数
*/
public void setPageSize(Integer pageSize) {
this.pageSize = pageSize;
}
public List<PreviewChunkResult> getChunks() {
return chunks;
}
@@ -640,6 +735,8 @@ public final class DocumentImportDtos {
private Document document;
private List<DocumentChunk> documentChunks = new ArrayList<DocumentChunk>();
private List<RagChunk> previewChunks = new ArrayList<RagChunk>();
private String chunkSnapshotPath;
private Integer totalChunks;
private Date createdAt;
public String getSessionId() {
@@ -730,6 +827,42 @@ public final class DocumentImportDtos {
this.previewChunks = previewChunks;
}
/**
* 返回持久化分块快照清单路径。
*
* @return 快照清单路径
*/
public String getChunkSnapshotPath() {
return chunkSnapshotPath;
}
/**
* 设置持久化分块快照清单路径。
*
* @param chunkSnapshotPath 快照清单路径
*/
public void setChunkSnapshotPath(String chunkSnapshotPath) {
this.chunkSnapshotPath = chunkSnapshotPath;
}
/**
* 返回完整分块总数。
*
* @return 分块总数
*/
public Integer getTotalChunks() {
return totalChunks;
}
/**
* 设置完整分块总数。
*
* @param totalChunks 分块总数
*/
public void setTotalChunks(Integer totalChunks) {
this.totalChunks = totalChunks;
}
public Date getCreatedAt() {
return createdAt;
}

View File

@@ -39,6 +39,10 @@ public final class DocumentImportKeys {
public static final String KEY_DOCUMENT_ROW_END = "rowEnd";
public static final String KEY_DOCUMENT_IMAGE_REFS = "imageRefs";
public static final String KEY_DOCUMENT_PARSE_ARTIFACT_SUMMARY = "parseArtifactSummary";
public static final String KEY_DOCUMENT_CSV_TABLE_SNAPSHOT_PATH = "parse.csvTableSnapshotPath";
public static final String KEY_DOCUMENT_CSV_ENCODING = "parse.csvEncoding";
public static final String KEY_DOCUMENT_CSV_ROW_COUNT = "parse.csvRowCount";
public static final String KEY_DOCUMENT_CSV_COLUMN_COUNT = "parse.csvColumnCount";
public static final String KEY_DOCUMENT_IMPORT_MODE = "import.mode";
public static final String KEY_DOCUMENT_IMPORT_BATCH_ID = "import.batchId";
public static final String KEY_DOCUMENT_IMPORT_BATCH_ITEM_ID = "import.batchItemId";

View File

@@ -5,6 +5,7 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import tech.easyflow.common.web.exceptions.BusinessException;
import java.math.BigInteger;
import java.time.Duration;
import java.util.concurrent.TimeUnit;
import java.util.UUID;
@@ -27,6 +28,34 @@ public class DocumentImportPreviewService {
return sessionId;
}
/**
* 为同一文档替换预览会话,返回被替换的旧会话供调用方清理对象存储工件。
*
* @param session 新预览会话
* @return 旧会话;不存在时返回 {@code null}
*/
public DocumentImportDtos.PreviewSession replaceForDocument(
DocumentImportDtos.PreviewSession session) {
String latestKey = buildLatestDocumentKey(
session.getKnowledgeId(), session.getDocumentId());
Object previousSessionId = defaultCache.get(latestKey);
DocumentImportDtos.PreviewSession previous = null;
if (previousSessionId instanceof String id) {
Object cached = defaultCache.get(buildKey(id));
if (cached instanceof DocumentImportDtos.PreviewSession oldSession) {
previous = oldSession;
}
}
String sessionId = put(session);
defaultCache.put(
latestKey, sessionId, SESSION_TTL.toMinutes(), TimeUnit.MINUTES);
if (previousSessionId instanceof String id
&& !id.equals(sessionId)) {
defaultCache.remove(buildKey(id));
}
return previous;
}
public DocumentImportDtos.PreviewSession getRequired(String sessionId) {
Object cached = defaultCache.get(buildKey(sessionId));
if (!(cached instanceof DocumentImportDtos.PreviewSession)) {
@@ -36,10 +65,33 @@ public class DocumentImportPreviewService {
}
public void remove(String sessionId) {
Object cached = defaultCache.get(buildKey(sessionId));
defaultCache.remove(buildKey(sessionId));
if (cached instanceof DocumentImportDtos.PreviewSession session) {
String latestKey = buildLatestDocumentKey(
session.getKnowledgeId(), session.getDocumentId());
Object latestSessionId = defaultCache.get(latestKey);
if (sessionId.equals(latestSessionId)) {
defaultCache.remove(latestKey);
}
}
}
private String buildKey(String sessionId) {
return DocumentImportKeys.CACHE_KEY_PREFIX + sessionId;
}
/**
* 构造文档当前预览会话索引键。
*
* @param knowledgeId 知识库 ID
* @param documentId 文档 ID
* @return 缓存键
*/
private String buildLatestDocumentKey(
BigInteger knowledgeId,
BigInteger documentId) {
return DocumentImportKeys.CACHE_KEY_PREFIX
+ "latest:" + knowledgeId + ":" + documentId;
}
}

View File

@@ -0,0 +1,34 @@
package tech.easyflow.ai.documentimport.task;
import tech.easyflow.common.web.exceptions.BusinessException;
/**
* CSV 导入异常,携带可持久化的稳定失败码。
*
* @author Codex
* @since 2026-08-04
*/
public class CsvImportException extends BusinessException {
private final String failureCode;
/**
* 创建 CSV 导入异常。
*
* @param failureCode 稳定失败码
* @param message 用户可见错误信息
*/
public CsvImportException(String failureCode, String message) {
super(message);
this.failureCode = failureCode;
}
/**
* 返回稳定失败码。
*
* @return 失败码
*/
public String getFailureCode() {
return failureCode;
}
}

View File

@@ -57,7 +57,7 @@ public class DocumentImportBatchAppService {
private static final Logger LOG = LoggerFactory.getLogger(DocumentImportBatchAppService.class);
private static final Set<String> 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;

View File

@@ -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;
/**
* 自动导入分块快照持久化服务。
*
* <p>快照写入对象存储,向量化任务只保存稳定路径,避免依赖短期预览缓存。</p>
* <p>V2 快照使用“有界分片 + 最后写入清单”的提交协议,索引任务可以按批读取,
* 同时保留对历史 V1 单 JSON 快照的读取兼容。</p>
*
* @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<DocumentChunk> 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<DocumentChunk>(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));
}
/**
* 从稳定存储完整恢复预览会话。
*
* <p>该方法用于兼容现有调用和小型预览。大文件索引应使用
* {@link #forEachBatch(String, int, Consumer)}。</p>
*
* @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<DocumentChunk> chunks = new ArrayList<DocumentChunk>();
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<DocumentChunk> loadPage(String path, int offset, int limit) {
if (offset < 0 || limit <= 0) {
throw new BusinessException("分块快照分页参数不合法");
}
SnapshotReadResult readResult = readManifestOrLegacy(path);
if (readResult.legacySession != null) {
List<DocumentChunk> chunks = readResult.legacySession.getDocumentChunks();
if (chunks == null || offset >= chunks.size()) {
return new ArrayList<DocumentChunk>();
}
int end = Math.min(offset + limit, chunks.size());
return new ArrayList<DocumentChunk>(chunks.subList(offset, end));
}
List<DocumentChunk> page = new ArrayList<DocumentChunk>(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<DocumentChunk> 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<List<DocumentChunk>> 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);
}
/**
* 删除快照分片并保留清单。
*
* <p>任一分片删除失败时立即抛出异常,清单继续作为精确重试依据。</p>
*
* @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<DocumentChunk>());
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<DocumentChunk> 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<DocumentChunk> chunks,
int batchSize,
Consumer<List<DocumentChunk>> 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<DocumentChunk>(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<DocumentChunk> chunks) {
if (committed) {
throw new IllegalStateException("分块快照已提交");
}
if (chunks == null || chunks.isEmpty()) {
return;
}
writeBoundedPart(new ArrayList<DocumentChunk>(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<DocumentChunk> chunks) {
byte[] payload = JSON.toJSONBytes(chunks);
if (payload.length > MAX_PART_BYTES && chunks.size() > 1) {
int midpoint = chunks.size() / 2;
writeBoundedPart(new ArrayList<DocumentChunk>(chunks.subList(0, midpoint)));
writeBoundedPart(new ArrayList<DocumentChunk>(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<SnapshotPart> parts = new ArrayList<SnapshotPart>();
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<SnapshotPart> getParts() {
return parts;
}
/**
* 设置分片清单。
*
* @param parts 分片清单
*/
public void setParts(List<SnapshotPart> parts) {
this.parts = parts == null
? new ArrayList<SnapshotPart>()
: 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<DocumentChunk> chunks;
private final long byteLength;
/**
* 创建分片载荷。
*
* @param chunks 分块
* @param byteLength 字节数
*/
private PartPayload(List<DocumentChunk> chunks, long byteLength) {
this.chunks = chunks;
this.byteLength = byteLength;
}
}
}

View File

@@ -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<String> SUPPORTED_EXTENSIONS =
Set.of("txt", "pdf", "docx", "md", "pptx", "xlsx", "csv");
/**
* 禁止实例化格式策略工具类。
*/
private DocumentImportFormatPolicy() {
}
/**
* 返回只读的支持格式集合。
*
* @return 支持的文件扩展名
*/
public static Set<String> supportedExtensions() {
return SUPPORTED_EXTENSIONS;
}
/**
* 判断扩展名是否属于知识库导入支持范围。
*
* @param extension 已转换为小写的文件扩展名
* @return 支持时返回 {@code true}
*/
public static boolean isSupported(String extension) {
return SUPPORTED_EXTENSIONS.contains(extension);
}
}

View File

@@ -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();
}
}

View File

@@ -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;
/**
* 文档导入中间快照可靠清理服务。
*
* <p>先持久化精确清单路径,再按“分片、清单”两个阶段删除。阶段推进先于
* 清单删除落库,确保进程在任一位置中断后都能安全继续。</p>
*
* @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<DocumentImportSnapshotCleanup> 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)));
}
}

View File

@@ -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<DocumentChunk> 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<DocumentChunk> storedChunks = new ArrayList<DocumentChunk>();
try {
DocumentCollection knowledge = assertDocumentCollection(task.getKnowledgeId());
DocumentImportDtos.PreviewSession session = resolveIndexPreviewSession(
knowledge,
document,
asString(task.getPayloadJson().get("previewSessionId")),
asString(task.getPayloadJson().get("chunkSnapshotPath"))
);
List<DocumentChunk> 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<DocumentChunk> batch = new ArrayList<DocumentChunk>(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<BigInteger> uniqueChunkIds =
new HashSet<BigInteger>(Math.min(totalChunks, 65_536));
int[] completedChunks = new int[]{0};
java.util.function.Consumer<List<DocumentChunk>> 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<DocumentChunk> 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<DocumentChunk>(
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<String, Object>(parsedResult.getMetadata()));
copyCsvParseMetadata(options, parsedResult.getMetadata());
}
if (parsedResult.getWarnings() != null && !parsedResult.getWarnings().isEmpty()) {
options.put(DocumentImportKeys.KEY_DOCUMENT_PARSE_WARNINGS, new ArrayList<String>(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<Map<String, Object>>(sheetRows.subList(1, sheetRows.size()))
: new ArrayList<Map<String, Object>>();
if (dataRows.isEmpty()) {
chunks.add(buildXlsxWindowChunk(document, sorting++, sheetName, sheet, headerRow,
new ArrayList<Map<String, Object>>(), sheetImages));
List<DocumentChunk> windowChunks = buildXlsxWindowChunks(
document, sorting, sheetName, sheet, headerRow,
new ArrayList<Map<String, Object>>(), 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<Map<String, Object>> windowRows = new ArrayList<Map<String, Object>>(dataRows.subList(start, end));
chunks.add(buildXlsxWindowChunk(document, sorting++, sheetName, sheet, headerRow, windowRows, sheetImages));
List<DocumentChunk> 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<String, Object> sheetSummary,
Map<String, Object> headerRow,
List<Map<String, Object>> windowRows,
List<Map<String, Object>> sheetImages) {
/**
* 构建 XLSX 行窗口分块。无图片窗口与 CSV 共用表格构建器,图片窗口保留现有 OCR 元信息。
*
* @param document 文档
* @param sorting 起始排序号
* @param sheetName Sheet 名称
* @param sheetSummary Sheet 摘要
* @param headerRow 表头行
* @param windowRows 数据行
* @param sheetImages Sheet 图片
* @return 一个或多个分块
*/
private List<DocumentChunk> buildXlsxWindowChunks(
tech.easyflow.ai.entity.Document document,
int sorting,
String sheetName,
Map<String, Object> sheetSummary,
Map<String, Object> headerRow,
List<Map<String, Object>> windowRows,
List<Map<String, Object>> 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<Map<String, Object>> selectedImages = selectWindowImages(sheetImages, rowStart, rowEnd, windowRows.isEmpty());
if (selectedImages.isEmpty()) {
List<Map<String, Object>> allRows =
new ArrayList<Map<String, Object>>();
allRows.add(headerRow);
allRows.addAll(windowRows);
int maxCol = resolveMaxColumnCount(allRows);
if (maxCol > 0) {
List<String> headers =
resolveTabularRowValues(headerRow, maxCol, true);
List<TabularRowWindowChunkBuilder.TabularRow> rows =
new ArrayList<TabularRowWindowChunkBuilder.TabularRow>();
for (Map<String, Object> row : windowRows) {
rows.add(new TabularRowWindowChunkBuilder.TabularRow(
asInteger(row.get("rowIndex"), 0) + 1,
resolveTabularRowValues(row, maxCol, false)));
}
List<DocumentChunk> chunks = tabularChunkBuilder.build(
document.getId(), document.getCollectionId(), sheetName,
headers, rows, sorting, RagChunkTypes.SECTION);
for (DocumentChunk chunk : chunks) {
Map<String, Object> options =
new LinkedHashMap<String, Object>(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<String>());
chunk.setOptions(options);
}
return chunks;
}
}
String renderMarkdown = buildXlsxChunkRenderMarkdown(sheetName, headerRow, windowRows, selectedImages);
String llmContent = buildXlsxChunkLlmContent(sheetName, headerRow, windowRows, selectedImages);
Map<String, Object> options = new LinkedHashMap<String, Object>();
@@ -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<String> resolveMarkdownRowValues(Map<String, Object> row,
int maxCol,
boolean headerRow) {
List<String> rawValues =
resolveTabularRowValues(row, maxCol, headerRow);
List<String> values = new ArrayList<String>(rawValues.size());
for (String value : rawValues) {
values.add(escapeMarkdownCell(value));
}
return values;
}
/**
* 将 XLSX 行转换为通用表格单元格值。
*
* @param row XLSX 行
* @param maxCol 最大列数
* @param headerRow 是否为表头
* @return 固定列数的原始值
*/
private List<String> resolveTabularRowValues(
Map<String, Object> row,
int maxCol,
boolean headerRow) {
List<String> values = new ArrayList<String>(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<DocumentChunk> 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<DocumentChunk> 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<DocumentChunk> chunks = session.getDocumentChunks();
if (chunks == null || offset >= chunks.size()) {
return new ArrayList<DocumentChunk>();
}
int end = Math.min(offset + pageSize, chunks.size());
return new ArrayList<DocumentChunk>(
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 向量化完成后登记中间表格快照清理。
*
* <p>清理记录持久化后由当前线程立即尝试,失败时由后台调度器继续重试。</p>
*
* @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<String, Object> options,
Map<String, Object> 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<String, Object> target,
Map<String, Object> 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<DocumentChunk> chunks) {
Set<BigInteger> uniqueIds = new HashSet<BigInteger>(chunks.size());
assertUniqueChunkIds(chunks, uniqueIds);
}
/**
* 校验批次分块 ID并把已见 ID 保留到跨分片集合。
*
* @param chunks 当前批次
* @param uniqueIds 已见 ID
*/
private void assertUniqueChunkIds(
List<DocumentChunk> chunks,
Set<BigInteger> 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<DocumentChunk> toChunkIdMarkers(List<DocumentChunk> chunks) {
List<DocumentChunk> markers = new ArrayList<DocumentChunk>(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());

View File

@@ -60,7 +60,7 @@ public class KnowledgeImportBatchFacade {
private static final Duration INCOMPLETE_SUBMISSION_TIMEOUT =
Duration.ofMinutes(30);
private static final Set<String> SUPPORTED_EXTENSIONS =
Set.of("txt", "md", "pdf", "docx", "pptx", "xlsx");
DocumentImportFormatPolicy.supportedExtensions();
private final DocumentImportBatchAppService batchAppService;
private final DocumentImportBatchTracker batchTracker;

View File

@@ -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;
/**
* 表格行窗口分块构建器。
*
* <p>CSV 与 XLSX 统一复用“表头 + 连续数据行”的语义,并对超长内容做有界续片,
* 保持单个分块满足下游模型的 Token 硬限制。</p>
*
* @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<DocumentChunk> build(
BigInteger documentId,
BigInteger collectionId,
String sourceLabel,
List<String> headers,
List<TabularRow> rows,
int startingSorting,
String chunkType) {
if (headers == null || headers.isEmpty()) {
throw new IllegalArgumentException("表格分块缺少表头");
}
List<TabularRow> safeRows = rows == null
? Collections.<TabularRow>emptyList()
: rows;
List<RenderedPart> parts =
renderParts(sourceLabel, headers, safeRows);
List<DocumentChunk> chunks =
new ArrayList<DocumentChunk>(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<String, Object> options = new LinkedHashMap<String, Object>();
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<String>());
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<String> 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<String> 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<String> headers,
TabularRow row) {
validateRow(headers, row);
List<String> values =
normalizeValues(row.getValues(), headers.size());
return boundedInt(5L + estimateCellsChars(values));
}
/**
* 估算单行追加到 Markdown 表格后的 Token 数。
*
* @param headers 表头
* @param row 表格数据行
* @return 保守 Token 数
*/
public int estimateRowTokens(
List<String> headers,
TabularRow row) {
validateRow(headers, row);
List<String> values =
normalizeValues(row.getValues(), headers.size());
return boundedInt(5L + estimateCellsTokens(values));
}
/**
* 按表格语义生成有界 Markdown 续片。
*
* @param sourceLabel 表格名称
* @param headers 表头
* @param rows 数据行
* @return 有界续片
*/
private List<RenderedPart> renderParts(
String sourceLabel,
List<String> headers,
List<TabularRow> 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<RenderedPart> parts = new ArrayList<RenderedPart>();
List<TabularRow> currentRows = new ArrayList<TabularRow>();
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<RenderedPart> parts,
String sourceLabel,
List<String> headers,
List<TabularRow> 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<RenderedPart> renderOversizedRow(
String sourceLabel,
List<String> headers,
TabularRow row) {
List<String> values =
normalizeValues(row.getValues(), headers.size());
List<RenderedPart> parts = new ArrayList<RenderedPart>();
List<String> partHeaders = new ArrayList<String>();
List<String> partValues = new ArrayList<String>();
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<String> candidateHeaders =
new ArrayList<String>(partHeaders);
candidateHeaders.add(header);
List<String> candidateValues =
new ArrayList<String>(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<String> headers,
List<String> 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<RenderedPart> parts,
String sourceLabel,
List<String> headers,
List<String> 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<RenderedPart> 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<RenderedPart> parts = new ArrayList<RenderedPart>();
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<String> headers,
List<String> 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<String> headers,
List<TabularRow> 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<String> 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<String> normalizeValues(List<String> values, int columnCount) {
List<String> normalized = new ArrayList<String>(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("<br/>");
index += index + 1 < safeValue.length()
&& safeValue.charAt(index + 1) == '\n' ? 2 : 1;
continue;
}
if (current == '\n') {
builder.append("<br/>");
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<String> 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<String> 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<String> headers) {
if (headers == null || headers.isEmpty()) {
throw new IllegalArgumentException("表格分块缺少表头");
}
}
/**
* 校验数据行。
*
* @param headers 表头
* @param row 数据行
*/
private void validateRow(
List<String> 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<String> values;
/**
* 创建表格数据行。
*
* @param rowNumber 一基逻辑行号
* @param values 单元格值
*/
public TabularRow(int rowNumber, List<String> values) {
this.rowNumber = rowNumber;
this.values = values == null
? Collections.<String>emptyList()
: new ArrayList<String>(values);
}
/**
* 返回逻辑行号。
*
* @return 一基逻辑行号
*/
public int getRowNumber() {
return rowNumber;
}
/**
* 返回单元格值。
*
* @return 不可变语义的单元格值副本
*/
public List<String> getValues() {
return new ArrayList<String>(values);
}
}
}

View File

@@ -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;
/**
* 文档导入中间快照清理记录。
*
* <p>记录精确清单路径、清理阶段和租约,使对象存储删除失败后可以跨进程重试。</p>
*
* @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;
}
}

View File

@@ -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<DocumentImportSnapshotCleanup> {
/**
* 幂等登记一个快照清理请求。
*
* @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<DocumentImportSnapshotCleanup> 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);
}

View File

@@ -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<Document> {
/**
* 在路径仍指向已清理 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);
}

View File

@@ -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<DocumentMapper, Document> i
@Autowired
private DocumentImportPreviewService documentImportPreviewService;
@Autowired
private DocumentImportSnapshotCleanupService snapshotCleanupService;
@Autowired
private KnowledgeDocumentImportTaskAppService importTaskAppService;
@@ -240,7 +244,18 @@ public class DocumentServiceImpl extends ServiceImpl<DocumentMapper, Document> 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());
// 主记录必须最后删除;否则接口返回成功后文档仍会出现在列表中。

View File

@@ -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<String> readSupportedExtensions(Class<?> type)
throws Exception {
Field field = type.getDeclaredField("SUPPORTED_EXTENSIONS");
field.setAccessible(true);
return (Set<String>) field.get(null);
}
}

View File

@@ -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("第一行<br/>第二行"));
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<DocumentChunk> 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<DocumentChunk> 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);
}
}

View File

@@ -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<byte[]> storedBytes = new AtomicReference<byte[]>();
FileStorageService storageService = Mockito.mock(FileStorageService.class);
Mockito.when(storageService.save(
Mockito.any(MultipartFile.class),
Mockito.anyString()
)).thenAnswer(invocation -> {
MultipartFile file = invocation.getArgument(0);
storedBytes.set(file.getBytes());
return storedPath;
});
Mockito.when(storageService.readStream(storedPath))
.thenAnswer(invocation -> new ByteArrayInputStream(storedBytes.get()));
InMemoryFileStorageService storageService =
new InMemoryFileStorageService();
DocumentImportChunkSnapshotService service = new DocumentImportChunkSnapshotService();
Field storageField = DocumentImportChunkSnapshotService.class
@@ -59,10 +44,108 @@ public class DocumentImportChunkSnapshotServiceTest {
String path = service.save(session);
DocumentImportDtos.PreviewSession restored = service.load(path);
Assert.assertEquals(storedPath, path);
Assert.assertTrue(path.endsWith("-manifest.json"));
Assert.assertEquals(session.getKnowledgeId(), restored.getKnowledgeId());
Assert.assertEquals(session.getDocumentId(), restored.getDocumentId());
Assert.assertEquals(1, restored.getDocumentChunks().size());
Assert.assertEquals("稳定分块", restored.getDocumentChunks().get(0).getContent());
}
/**
* 验证 V2 快照可分页并按小批顺序消费,删除时同时清理清单和分片。
*
* @throws Exception 反射注入异常
*/
@Test
public void shouldReadV2SnapshotByPageAndBatch() throws Exception {
InMemoryFileStorageService storageService =
new InMemoryFileStorageService();
DocumentImportChunkSnapshotService service =
new DocumentImportChunkSnapshotService();
Field storageField = DocumentImportChunkSnapshotService.class
.getDeclaredField("storageService");
storageField.setAccessible(true);
storageField.set(service, storageService);
DocumentImportDtos.PreviewSession session =
new DocumentImportDtos.PreviewSession();
session.setKnowledgeId(BigInteger.valueOf(7));
session.setDocumentId(BigInteger.valueOf(9));
List<DocumentChunk> chunks = new ArrayList<DocumentChunk>();
for (int index = 0; index < 5; index++) {
DocumentChunk chunk = new DocumentChunk();
chunk.setId(BigInteger.valueOf(100 + index));
chunk.setDocumentId(BigInteger.valueOf(9));
chunk.setDocumentCollectionId(BigInteger.valueOf(7));
chunk.setContent("chunk-" + index);
chunks.add(chunk);
}
session.setDocumentChunks(chunks);
String path = service.save(session);
List<DocumentChunk> page = service.loadPage(path, 2, 2);
List<String> consumed = new ArrayList<String>();
service.forEachBatch(path, 2, batch -> {
Assert.assertTrue(batch.size() <= 2);
for (DocumentChunk chunk : batch) {
consumed.add(chunk.getContent());
}
});
Assert.assertEquals(List.of("chunk-2", "chunk-3"),
page.stream().map(DocumentChunk::getContent).toList());
Assert.assertEquals(
List.of("chunk-0", "chunk-1", "chunk-2", "chunk-3", "chunk-4"),
consumed);
Assert.assertTrue(storageService.size() >= 2);
service.delete(path);
Assert.assertEquals(0, storageService.size());
}
/**
* 验证分片删除失败时保留清单,使后续重试仍能获取精确对象集合。
*
* @throws Exception 反射注入异常
*/
@Test
public void shouldKeepManifestWhenPartDeletionFails() throws Exception {
InMemoryFileStorageService storageService =
new InMemoryFileStorageService();
DocumentImportChunkSnapshotService service =
new DocumentImportChunkSnapshotService();
Field storageField = DocumentImportChunkSnapshotService.class
.getDeclaredField("storageService");
storageField.setAccessible(true);
storageField.set(service, storageService);
DocumentChunk chunk = new DocumentChunk();
chunk.setId(BigInteger.valueOf(21));
chunk.setDocumentId(BigInteger.valueOf(9));
chunk.setDocumentCollectionId(BigInteger.valueOf(7));
chunk.setContent("等待可靠清理");
DocumentImportDtos.PreviewSession session =
new DocumentImportDtos.PreviewSession();
session.setKnowledgeId(BigInteger.valueOf(7));
session.setDocumentId(BigInteger.valueOf(9));
session.setDocumentChunks(List.of(chunk));
String manifestPath = service.save(session);
String partPath = storageService.paths().stream()
.filter(path -> !path.equals(manifestPath))
.findFirst()
.orElseThrow();
storageService.failDelete(partPath, 1);
try {
service.delete(manifestPath);
Assert.fail("分片删除失败时应抛出异常");
} catch (IllegalStateException expected) {
Assert.assertTrue(expected.getMessage().contains("模拟删除失败"));
}
Assert.assertTrue(storageService.contains(manifestPath));
Assert.assertTrue(storageService.contains(partPath));
service.delete(manifestPath);
Assert.assertEquals(0, storageService.size());
}
}

View File

@@ -0,0 +1,237 @@
package tech.easyflow.ai.documentimport.task;
import com.mybatisflex.core.query.QueryWrapper;
import org.junit.Test;
import org.mockito.Mockito;
import tech.easyflow.ai.entity.DocumentImportSnapshotCleanup;
import tech.easyflow.ai.mapper.DocumentImportSnapshotCleanupMapper;
import tech.easyflow.ai.mapper.DocumentMapper;
import java.lang.reflect.Field;
import java.math.BigInteger;
import java.util.Date;
import java.util.concurrent.atomic.AtomicReference;
/**
* {@link DocumentImportSnapshotCleanupService} 可靠清理测试。
*/
public class DocumentImportSnapshotCleanupServiceTest {
/**
* 验证分片和清单按持久化阶段顺序删除。
*
* @throws Exception 反射注入失败
*/
@Test
public void shouldPersistPhaseBeforeDeletingManifest()
throws Exception {
CleanupHarness harness = createHarness();
harness.service.scheduleChunkSnapshot(
"snapshot/chunk-manifest.json");
Mockito.verify(harness.chunkSnapshotService).deleteParts(
"snapshot/chunk-manifest.json");
Mockito.verify(harness.cleanupMapper).advanceToManifest(
Mockito.any(BigInteger.class),
Mockito.anyString(),
Mockito.any(Date.class));
Mockito.verify(harness.chunkSnapshotService).deleteManifest(
"snapshot/chunk-manifest.json");
Mockito.verify(harness.cleanupMapper).deleteCompleted(
Mockito.any(BigInteger.class),
Mockito.anyString());
Mockito.verify(harness.cleanupMapper, Mockito.never())
.releaseForRetry(
Mockito.any(BigInteger.class),
Mockito.anyString(),
Mockito.any(Date.class),
Mockito.anyString(),
Mockito.any(Date.class));
}
/**
* 验证分片删除失败后保留记录并安排重试。
*
* @throws Exception 反射注入失败
*/
@Test
public void shouldReleaseFailedCleanupForRetry()
throws Exception {
CleanupHarness harness = createHarness();
Mockito.doThrow(new IllegalStateException("对象存储暂不可用"))
.when(harness.chunkSnapshotService)
.deleteParts("snapshot/failing-manifest.json");
harness.service.scheduleChunkSnapshot(
"snapshot/failing-manifest.json");
Mockito.verify(harness.cleanupMapper).releaseForRetry(
Mockito.any(BigInteger.class),
Mockito.anyString(),
Mockito.any(Date.class),
Mockito.contains("对象存储暂不可用"),
Mockito.any(Date.class));
Mockito.verify(harness.chunkSnapshotService, Mockito.never())
.deleteManifest(Mockito.anyString());
Mockito.verify(harness.cleanupMapper, Mockito.never())
.deleteCompleted(
Mockito.any(BigInteger.class),
Mockito.anyString());
}
/**
* 验证 CSV 清单删除后原子清理文档中的旧快照指针。
*
* @throws Exception 反射注入失败
*/
@Test
public void shouldClearCsvPointerAfterManifestDeletion()
throws Exception {
CleanupHarness harness = createHarness();
BigInteger knowledgeId = BigInteger.valueOf(7);
BigInteger documentId = BigInteger.valueOf(9);
harness.service.scheduleCsvTableSnapshot(
knowledgeId,
documentId,
"snapshot/csv-manifest.json");
Mockito.verify(harness.csvTableSnapshotService).deleteParts(
"snapshot/csv-manifest.json");
Mockito.verify(harness.csvTableSnapshotService).deleteManifest(
"snapshot/csv-manifest.json");
Mockito.verify(harness.documentMapper).clearCsvSnapshotPath(
Mockito.eq(documentId),
Mockito.eq("snapshot/csv-manifest.json"),
Mockito.any(Date.class));
}
/**
* 创建带内存状态的清理服务测试夹具。
*
* @return 测试夹具
* @throws Exception 反射注入失败
*/
@SuppressWarnings("unchecked")
private CleanupHarness createHarness() throws Exception {
DocumentImportSnapshotCleanupService service =
new DocumentImportSnapshotCleanupService();
DocumentImportSnapshotCleanupMapper cleanupMapper =
Mockito.mock(DocumentImportSnapshotCleanupMapper.class);
DocumentImportChunkSnapshotService chunkSnapshotService =
Mockito.mock(DocumentImportChunkSnapshotService.class);
CsvTableSnapshotService csvTableSnapshotService =
Mockito.mock(CsvTableSnapshotService.class);
DocumentMapper documentMapper =
Mockito.mock(DocumentMapper.class);
AtomicReference<DocumentImportSnapshotCleanup> recordRef =
new AtomicReference<DocumentImportSnapshotCleanup>();
Mockito.when(cleanupMapper.insertIgnore(
Mockito.any(DocumentImportSnapshotCleanup.class)))
.thenAnswer(invocation -> {
recordRef.set(invocation.getArgument(0));
return 1;
});
Mockito.when(cleanupMapper.selectOneByQuery(
Mockito.any(QueryWrapper.class)))
.thenAnswer(invocation -> recordRef.get());
Mockito.when(cleanupMapper.claim(
Mockito.any(BigInteger.class),
Mockito.anyString(),
Mockito.any(Date.class),
Mockito.any(Date.class)))
.thenAnswer(invocation -> {
DocumentImportSnapshotCleanup record = recordRef.get();
record.setExecutionToken(invocation.getArgument(1));
record.setAttemptCount(
(record.getAttemptCount() == null
? 0
: record.getAttemptCount()) + 1);
return 1;
});
Mockito.when(cleanupMapper.advanceToManifest(
Mockito.any(BigInteger.class),
Mockito.anyString(),
Mockito.any(Date.class)))
.thenAnswer(invocation -> {
recordRef.get().setPhase("MANIFEST_PENDING");
return 1;
});
Mockito.when(cleanupMapper.deleteCompleted(
Mockito.any(BigInteger.class),
Mockito.anyString())).thenReturn(1);
Mockito.when(cleanupMapper.releaseForRetry(
Mockito.any(BigInteger.class),
Mockito.anyString(),
Mockito.any(Date.class),
Mockito.anyString(),
Mockito.any(Date.class))).thenReturn(1);
setField(service, "cleanupMapper", cleanupMapper);
setField(
service, "chunkSnapshotService", chunkSnapshotService);
setField(
service, "csvTableSnapshotService", csvTableSnapshotService);
setField(service, "documentMapper", documentMapper);
return new CleanupHarness(
service,
cleanupMapper,
chunkSnapshotService,
csvTableSnapshotService,
documentMapper);
}
/**
* 反射设置测试字段。
*
* @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);
}
/**
* 快照清理测试依赖集合。
*/
private static final class CleanupHarness {
private final DocumentImportSnapshotCleanupService service;
private final DocumentImportSnapshotCleanupMapper cleanupMapper;
private final DocumentImportChunkSnapshotService
chunkSnapshotService;
private final CsvTableSnapshotService csvTableSnapshotService;
private final DocumentMapper documentMapper;
/**
* 创建测试依赖集合。
*
* @param service 清理服务
* @param cleanupMapper 清理 Mapper
* @param chunkSnapshotService 分块快照服务
* @param csvTableSnapshotService CSV 快照服务
* @param documentMapper 文档 Mapper
*/
private CleanupHarness(
DocumentImportSnapshotCleanupService service,
DocumentImportSnapshotCleanupMapper cleanupMapper,
DocumentImportChunkSnapshotService chunkSnapshotService,
CsvTableSnapshotService csvTableSnapshotService,
DocumentMapper documentMapper) {
this.service = service;
this.cleanupMapper = cleanupMapper;
this.chunkSnapshotService = chunkSnapshotService;
this.csvTableSnapshotService = csvTableSnapshotService;
this.documentMapper = documentMapper;
}
}
}

View File

@@ -0,0 +1,148 @@
package tech.easyflow.ai.documentimport.task;
import org.springframework.web.multipart.MultipartFile;
import tech.easyflow.common.filestorage.FileStorageService;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 文档导入测试使用的内存文件存储。
*/
final class InMemoryFileStorageService implements FileStorageService {
private final AtomicInteger sequence = new AtomicInteger();
private final Map<String, byte[]> files = new LinkedHashMap<String, byte[]>();
private String failingDeletePath;
private int remainingDeleteFailures;
/**
* 保存到默认测试目录。
*
* @param file 上传文件
* @return 测试路径
*/
@Override
public String save(MultipartFile file) {
return save(file, "default");
}
/**
* 保存到指定测试目录。
*
* @param file 上传文件
* @param prePath 测试目录
* @return 测试路径
*/
@Override
public String save(MultipartFile file, String prePath) {
try {
String path = prePath + "/" + sequence.incrementAndGet()
+ "-" + file.getOriginalFilename();
files.put(path, file.getBytes());
return path;
} catch (IOException error) {
throw new IllegalStateException("测试文件保存失败", error);
}
}
/**
* 删除测试对象。
*
* @param path 测试路径
*/
@Override
public void delete(String path) {
if (remainingDeleteFailures > 0
&& path != null
&& path.equals(failingDeletePath)) {
remainingDeleteFailures--;
throw new IllegalStateException(
"模拟删除失败: " + path);
}
files.remove(path);
}
/**
* 打开测试对象。
*
* @param path 测试路径
* @return 输入流
* @throws IOException 对象不存在
*/
@Override
public InputStream readStream(String path) throws IOException {
byte[] bytes = files.get(path);
if (bytes == null) {
throw new IOException("测试对象不存在: " + path);
}
return new ByteArrayInputStream(bytes);
}
/**
* 返回测试对象大小。
*
* @param path 测试路径
* @return 字节数
*/
@Override
public long getFileSize(String path) {
byte[] bytes = files.get(path);
return bytes == null ? -1L : bytes.length;
}
/**
* 直接放入一个源文件。
*
* @param path 测试路径
* @param bytes 文件内容
*/
void put(String path, byte[] bytes) {
files.put(path, bytes);
}
/**
* 返回当前对象数量。
*
* @return 对象数量
*/
int size() {
return files.size();
}
/**
* 判断测试对象是否存在。
*
* @param path 测试路径
* @return 是否存在
*/
boolean contains(String path) {
return files.containsKey(path);
}
/**
* 返回当前全部测试路径快照。
*
* @return 路径集合
*/
Set<String> paths() {
return new LinkedHashSet<String>(files.keySet());
}
/**
* 配置指定路径接下来若干次删除失败。
*
* @param path 测试路径
* @param failureCount 失败次数
*/
void failDelete(String path, int failureCount) {
failingDeletePath = path;
remainingDeleteFailures = Math.max(0, failureCount);
}
}

View File

@@ -24,6 +24,7 @@ import tech.easyflow.ai.document.exception.DocumentParseBridgeException;
import tech.easyflow.ai.document.model.DocumentParseArtifacts;
import tech.easyflow.ai.document.model.DocumentParsedResult;
import tech.easyflow.ai.document.model.DocumentSourceRef;
import tech.easyflow.ai.documentimport.DocumentImportDtos;
import tech.easyflow.ai.documentimport.DocumentImportKeys;
import tech.easyflow.ai.entity.DocumentChunk;
import tech.easyflow.ai.entity.DocumentCollection;
@@ -68,6 +69,46 @@ import java.util.concurrent.atomic.AtomicReference;
*/
public class KnowledgeDocumentImportTaskAppServiceTest {
/**
* 验证预览翻页直接读取分片快照,避免恢复完整会话。
*
* @throws Exception 反射调用失败
*/
@Test
public void loadPreviewPageShouldReadOnlyRequestedSnapshotRange()
throws Exception {
KnowledgeDocumentImportTaskAppService service =
new KnowledgeDocumentImportTaskAppService();
DocumentImportChunkSnapshotService snapshotService =
Mockito.mock(DocumentImportChunkSnapshotService.class);
setField(
service,
"documentImportChunkSnapshotService",
snapshotService);
DocumentChunk chunk = new DocumentChunk();
chunk.setId(BigInteger.valueOf(9001));
Mockito.when(snapshotService.loadPage("snapshot.json", 20, 20))
.thenReturn(List.of(chunk));
DocumentImportDtos.PreviewSession session =
new DocumentImportDtos.PreviewSession();
session.setChunkSnapshotPath("snapshot.json");
Method method = KnowledgeDocumentImportTaskAppService.class
.getDeclaredMethod(
"loadPreviewPage",
DocumentImportDtos.PreviewSession.class,
int.class,
int.class);
method.setAccessible(true);
@SuppressWarnings("unchecked")
List<DocumentChunk> result = (List<DocumentChunk>) method.invoke(
service, session, 2, 20);
Assert.assertEquals(List.of(chunk), result);
Mockito.verify(snapshotService).loadPage("snapshot.json", 20, 20);
}
/**
* 验证待处理任务重新投递只更新必要字段,避免自定义查询结果覆盖非空列。
*