From c1376a2e68bb04b32e1ab2d60de69fbb0965a9a6 Mon Sep 17 00:00:00 2001 From: luke Date: Fri, 20 Jun 2025 11:15:05 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=8C=E6=88=90=E5=AF=B9=E8=B1=A1=E5=AD=98?= =?UTF-8?q?=E5=82=A8=E5=88=A0=E9=99=A4=E6=8E=A5=E5=8F=A3=EF=BC=8C=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E9=83=A8=E5=88=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Readme.md | 1 + .../application/service/DocAppService.java | 23 +++ .../service/DocAppServiceImpl.java | 76 ++++++++++ .../repository/iface/OSRecordRepository.java | 14 ++ .../doc/service/iface/FileDomainService.java | 14 ++ .../impl/AbstractBaseFileImporter.java | 23 +-- .../service/impl/FileDomainServiceImpl.java | 11 ++ .../cache/iface/FileCacheService.java | 3 + .../cache/impl/RedisFileCacheServiceImpl.java | 17 +++ .../converter/FileDtoConverter.java | 11 ++ .../north/controller/FileWriteController.java | 22 ++- .../doc/OSRecordRepositoryImpl.java | 19 +++ .../south/es/FileElasticsearchGateway.java | 134 ++++++++++++++++++ .../south/{ => minio}/MinioOSGatewayImpl.java | 36 ++++- .../{ => minio}/ObjectStorageGateway.java | 5 +- .../infrastructure/util/LocalFileUtil.java | 75 ++++++++++ 16 files changed, 465 insertions(+), 19 deletions(-) create mode 100644 src/main/java/com/knowledge/base/infrastructure/south/es/FileElasticsearchGateway.java rename src/main/java/com/knowledge/base/infrastructure/south/{ => minio}/MinioOSGatewayImpl.java (70%) rename src/main/java/com/knowledge/base/infrastructure/south/{ => minio}/ObjectStorageGateway.java (82%) create mode 100644 src/main/java/com/knowledge/base/infrastructure/util/LocalFileUtil.java diff --git a/Readme.md b/Readme.md index d006cf8..44ea155 100644 --- a/Readme.md +++ b/Readme.md @@ -9,6 +9,7 @@ | 1.0.3 | 2025-06-16 | Luke.Ye | 新增文档上传至对象存储接口 | | 1.0.4 | 2025-06-17 | Luke.Ye | 上传文档记录入库 | | 1.0.5 | 2025-06-19 | Luke.Ye | 文档导入ES逻辑优化 | +| 1.0.6 | 2025-06-20 | Luke.Ye | 完成对象存储删除接口，优化部分代码 | diff --git a/src/main/java/com/knowledge/base/application/service/DocAppService.java b/src/main/java/com/knowledge/base/application/service/DocAppService.java index 4c0c22d..fed3096 100644 --- a/src/main/java/com/knowledge/base/application/service/DocAppService.java +++ b/src/main/java/com/knowledge/base/application/service/DocAppService.java @@ -3,6 +3,8 @@ package com.knowledge.base.application.service; import com.knowledge.base.domain.common.model.PageResult; import com.knowledge.base.infrastructure.north.dto.doc.OSRecordDTO; +import java.util.List; + /** * @author Luke.ye * @date 2025/6/16 19:13 @@ -14,6 +16,27 @@ public interface DocAppService { */ PageResult pageQueryOSRecord(int page, int size, String uploader); + /** + * 批量获取上传记录 + * @param recordIds + * @return + */ + List batchQueryOSRecord(List recordIds); + + + + + /** + * 批量删除上传记录 + * + * @param clearAll 是否清空所有 + * true: 清空（ES/缓存/MySQL/对象存储/本地存储）中所有相关的数据； + * false：仅清除 MySQL+对象存储 中的数据 + * @param recordIds 本地存储路径 + * @return + */ + boolean batchRemoveOSFiles(boolean clearAll, List recordIds); + /** * 保存上传记录 */ diff --git a/src/main/java/com/knowledge/base/application/service/DocAppServiceImpl.java b/src/main/java/com/knowledge/base/application/service/DocAppServiceImpl.java index 4384469..4c4af29 100644 --- a/src/main/java/com/knowledge/base/application/service/DocAppServiceImpl.java +++ b/src/main/java/com/knowledge/base/application/service/DocAppServiceImpl.java @@ -1,22 +1,42 @@ package com.knowledge.base.application.service; +import cn.hutool.core.collection.CollectionUtil; import com.google.common.collect.Lists; import com.knowledge.base.application.exceptions.AppException; import com.knowledge.base.domain.common.model.PageResult; import com.knowledge.base.domain.doc.model.OSRecordDO; import com.knowledge.base.domain.doc.service.iface.FileDomainService; +import com.knowledge.base.infrastructure.cache.iface.FileCacheService; +import com.knowledge.base.infrastructure.config.ObjectStorageProperties; import com.knowledge.base.infrastructure.converter.FileDtoConverter; import com.knowledge.base.infrastructure.north.dto.doc.OSRecordDTO; +import com.knowledge.base.infrastructure.south.es.FileElasticsearchGateway; +import com.knowledge.base.infrastructure.south.minio.ObjectStorageGateway; +import com.knowledge.base.infrastructure.util.LocalFileUtil; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Component; import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Optional; +import java.util.stream.Collectors; @Component @RequiredArgsConstructor +@Slf4j public class DocAppServiceImpl implements DocAppService{ private final FileDomainService fileDomainService; + private final ObjectStorageProperties osProperties; + + private final FileElasticsearchGateway esGateway; + + private final FileCacheService fileCacheService; + + private final ObjectStorageGateway osGateway; + @Override public PageResult pageQueryOSRecord(int page, int size, String uploader) { if(page < 1 || size < 1) { @@ -35,6 +55,62 @@ public class DocAppServiceImpl implements DocAppService{ return new PageResult<>(dtoList, osRecordDOPageResult.getTotal()); } + @Override + public List batchQueryOSRecord(List recordIds) { + if(CollectionUtil.isEmpty(recordIds)) { + throw new AppException("参数recordIds不能为空"); + } + List osRecordDOList = null; + try { + osRecordDOList = fileDomainService.batchQueryOSRecord(recordIds); + } catch (Exception e) { + throw new AppException("批量获取对象存储记录失败，请检查recordIds", e); + } + return FileDtoConverter.toDTOs(osRecordDOList); + } + + @Override + public boolean batchRemoveOSFiles(boolean clearAll, List recordIds) { + if(CollectionUtil.isEmpty(recordIds)) { + throw new AppException("参数recordIds不能为空"); + } + + + try { + List osRecordDTOS = Optional.ofNullable(batchQueryOSRecord(recordIds)).orElse(Lists.newArrayList()); + List filesLocalRelaPath = osRecordDTOS.stream().filter(Objects::nonNull) + .map(OSRecordDTO::getLocalRelaFilePath).collect(Collectors.toList()); + + Map> bucketToObjectsMap = osRecordDTOS.stream() + .collect(Collectors.groupingBy( + OSRecordDTO::getBucketName, + Collectors.mapping(OSRecordDTO::getObjectPath, Collectors.toList()) + )); + // 删除关联数据库记录 + fileDomainService.batchRemoveOSFiles(recordIds); + // 删除对象存储中的文件 + bucketToObjectsMap.forEach((bucket, objPaths) -> { + try { + osGateway.deleteObjects(bucket, objPaths); + } catch (Exception e) { + log.error("删除对象存储中的文件失败. bucket: {}", bucket, e); + log.error("待删除路径如下: {}", objPaths); + } + }); + if(clearAll) { + // 删除本地文件 + LocalFileUtil.deleteFilesByRelativePath(osProperties.getLocalSearchablePathPrefix(), filesLocalRelaPath); + // 删除ES + esGateway.deleteByFilepaths(filesLocalRelaPath); + // 删除缓存 + fileCacheService.removeMetaCacheBatch(filesLocalRelaPath); + } + return true; + } catch (Exception e) { + throw new AppException("批量删除对象存储记录失败，请检查recordIds", e); + } + } + @Override public void saveOSUplodRecord(OSRecordDTO dto) { try { diff --git a/src/main/java/com/knowledge/base/domain/doc/repository/iface/OSRecordRepository.java b/src/main/java/com/knowledge/base/domain/doc/repository/iface/OSRecordRepository.java index a676d26..449785a 100644 --- a/src/main/java/com/knowledge/base/domain/doc/repository/iface/OSRecordRepository.java +++ b/src/main/java/com/knowledge/base/domain/doc/repository/iface/OSRecordRepository.java @@ -3,6 +3,7 @@ package com.knowledge.base.domain.doc.repository.iface; import com.knowledge.base.domain.common.model.PageResult; import com.knowledge.base.domain.doc.repository.po.OSRecord; +import java.util.List; import java.util.Optional; public interface OSRecordRepository { @@ -23,4 +24,17 @@ public interface OSRecordRepository { * @return */ Optional getLatestRecordByRelaPath(String localRelaFilePath); + + /** + * 批量获取 + * @param recordIds + * @return + */ + List batchQueryOSRecord(List recordIds); + + /** + * 批量删除 + * @param recordIds + */ + boolean batchRemoveOSFiles(List recordIds); } diff --git a/src/main/java/com/knowledge/base/domain/doc/service/iface/FileDomainService.java b/src/main/java/com/knowledge/base/domain/doc/service/iface/FileDomainService.java index 2cf2777..8f93e34 100644 --- a/src/main/java/com/knowledge/base/domain/doc/service/iface/FileDomainService.java +++ b/src/main/java/com/knowledge/base/domain/doc/service/iface/FileDomainService.java @@ -31,4 +31,18 @@ public interface FileDomainService { * @return */ Optional getLatestRecordByRelaPath(String localRelaFilePath); + + /** + * 批量获取上传记录 + * @param recordIds + * @return + */ + List batchQueryOSRecord(List recordIds); + + /** + * 批量删除上传记录 + * @param recordIds + * @return + */ + boolean batchRemoveOSFiles(List recordIds); } diff --git a/src/main/java/com/knowledge/base/domain/doc/service/impl/AbstractBaseFileImporter.java b/src/main/java/com/knowledge/base/domain/doc/service/impl/AbstractBaseFileImporter.java index 92f6d4b..a0ec845 100644 --- a/src/main/java/com/knowledge/base/domain/doc/service/impl/AbstractBaseFileImporter.java +++ b/src/main/java/com/knowledge/base/domain/doc/service/impl/AbstractBaseFileImporter.java @@ -13,23 +13,16 @@ import com.knowledge.base.domain.doc.service.iface.FileDomainService; import com.knowledge.base.infrastructure.cache.iface.FileCacheService; import com.knowledge.base.infrastructure.config.ConstantConfig; import com.knowledge.base.infrastructure.config.ThreadPoolConfig; +import com.knowledge.base.infrastructure.south.es.FileElasticsearchGateway; import com.knowledge.base.infrastructure.util.*; -import org.elasticsearch.action.delete.DeleteRequest; -import org.elasticsearch.action.get.GetRequest; import org.elasticsearch.action.get.GetResponse; -import org.elasticsearch.action.index.IndexRequest; -import org.elasticsearch.client.RequestOptions; -import org.elasticsearch.client.RestHighLevelClient; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import java.io.IOException; import java.nio.file.*; -import java.util.HashMap; -import java.util.Map; -import java.util.Optional; -import java.util.Set; +import java.util.*; public abstract class AbstractBaseFileImporter implements DocumentImporter { @@ -37,7 +30,7 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter { private static final String INDEX_NAME = "documents"; @Autowired - protected RestHighLevelClient esClient; + private FileElasticsearchGateway esGateway; @Autowired private FileCacheService fileCacheService; @@ -129,9 +122,8 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter { // === Step 4: 检查 ES 是否已有未变动版本 === String docId = SafeIdUtil.encode(localRelaFilePath); - GetRequest getRequest = new GetRequest(INDEX_NAME, docId); - if (esClient.exists(getRequest, RequestOptions.DEFAULT)) { - GetResponse existing = esClient.get(getRequest, RequestOptions.DEFAULT); + GetResponse existing = esGateway.getIfExists(INDEX_NAME, docId); + if(Objects.nonNull(existing)) { Map existingSource = existing.getSourceAsMap(); Object esMtime = existingSource.get("mtime"); if (esMtime != null && Long.parseLong(esMtime.toString()) == localMTime) { @@ -139,7 +131,7 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter { fileCacheService.cacheMeta(localRelaFilePath, transToMetaJson(existingSource), ConstantConfig.FILE_META_CACHE_EXPIRED_MINUTES); return true; } - esClient.delete(new DeleteRequest(INDEX_NAME, docId), RequestOptions.DEFAULT); + esGateway.deleteDoc(INDEX_NAME, docId); logger.info("[ES] 已删除旧版本文件: {}", localRelaFilePath); } @@ -149,8 +141,7 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter { Map doc = buildDocument(fileNameWithSuffix, localRelaFilePath, content, localMTime, extInfo); // === Step 6: 写入 Elasticsearch 并更新缓存 === - IndexRequest request = new IndexRequest(INDEX_NAME).id(docId).source(doc); - esClient.index(request, RequestOptions.DEFAULT); + esGateway.saveDoc(INDEX_NAME, docId, doc); logger.info("导入成功: {}", localRelaFilePath); fileCacheService.cacheMeta(localRelaFilePath, transToMetaJson(doc), ConstantConfig.FILE_META_CACHE_EXPIRED_MINUTES); diff --git a/src/main/java/com/knowledge/base/domain/doc/service/impl/FileDomainServiceImpl.java b/src/main/java/com/knowledge/base/domain/doc/service/impl/FileDomainServiceImpl.java index d06ea58..061d248 100644 --- a/src/main/java/com/knowledge/base/domain/doc/service/impl/FileDomainServiceImpl.java +++ b/src/main/java/com/knowledge/base/domain/doc/service/impl/FileDomainServiceImpl.java @@ -77,4 +77,15 @@ public class FileDomainServiceImpl implements FileDomainService { } return Optional.of(BeanConvertUtil.convert(osRecordOpt.get(), OSRecordDO.class)); } + + @Override + public List batchQueryOSRecord(List recordIds) { + return BeanConvertUtil.convertList(osRecordRepository.batchQueryOSRecord(recordIds), OSRecordDO.class); + } + + @Override + public boolean batchRemoveOSFiles(List recordIds) { + osRecordRepository.batchRemoveOSFiles(recordIds); + return true; + } } diff --git a/src/main/java/com/knowledge/base/infrastructure/cache/iface/FileCacheService.java b/src/main/java/com/knowledge/base/infrastructure/cache/iface/FileCacheService.java index 9ec0480..04c2be7 100644 --- a/src/main/java/com/knowledge/base/infrastructure/cache/iface/FileCacheService.java +++ b/src/main/java/com/knowledge/base/infrastructure/cache/iface/FileCacheService.java @@ -1,5 +1,6 @@ package com.knowledge.base.infrastructure.cache.iface; +import java.util.Collection; import java.util.Optional; /** @@ -31,6 +32,8 @@ public interface FileCacheService { */ default void removeMetaCache(String fnWithRelativePath) {}; + default public void removeMetaCacheBatch(Collection fileRelativePaths) {} + /** * 清空所有缓存（注意：某些实现可能未实现） */ diff --git a/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisFileCacheServiceImpl.java b/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisFileCacheServiceImpl.java index c8e1ab9..850c3f0 100644 --- a/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisFileCacheServiceImpl.java +++ b/src/main/java/com/knowledge/base/infrastructure/cache/impl/RedisFileCacheServiceImpl.java @@ -1,5 +1,6 @@ package com.knowledge.base.infrastructure.cache.impl; +import cn.hutool.core.collection.CollectionUtil; import com.knowledge.base.infrastructure.cache.iface.FileCacheService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -7,6 +8,7 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.data.redis.core.StringRedisTemplate; import org.springframework.stereotype.Service; +import java.util.Collection; import java.util.HashSet; import java.util.Optional; import java.util.Set; @@ -48,6 +50,21 @@ public class RedisFileCacheServiceImpl implements FileCacheService { redisTemplate.delete(key("meta", fnWithRelativePath)); } + @Override + public void removeMetaCacheBatch(Collection fileRelativePaths) { + if (CollectionUtil.isEmpty(fileRelativePaths)) { + return; + } + Set redisKeys = new HashSet<>(); + for (String path : fileRelativePaths) { + redisKeys.add(key("meta", path)); + } + + redisTemplate.delete(redisKeys); + log.info("批量删除 Redis 文件缓存，数量: {}", redisKeys.size()); + } + + @Override public void clearAll() { log.warn("正在清空 Redis 文件缓存，前缀: {}", FILE_CACHE_PREFIX); diff --git a/src/main/java/com/knowledge/base/infrastructure/converter/FileDtoConverter.java b/src/main/java/com/knowledge/base/infrastructure/converter/FileDtoConverter.java index 384ad63..b0db640 100644 --- a/src/main/java/com/knowledge/base/infrastructure/converter/FileDtoConverter.java +++ b/src/main/java/com/knowledge/base/infrastructure/converter/FileDtoConverter.java @@ -1,5 +1,7 @@ package com.knowledge.base.infrastructure.converter; +import cn.hutool.core.collection.CollectionUtil; +import com.google.common.collect.Lists; import com.knowledge.base.domain.doc.model.FileDO; import com.knowledge.base.domain.doc.model.OSRecordDO; import com.knowledge.base.infrastructure.north.dto.doc.FileDTO; @@ -8,6 +10,7 @@ import org.springframework.beans.BeanUtils; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; +import java.util.List; public class FileDtoConverter { private static final DateTimeFormatter dtf = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @@ -66,4 +69,12 @@ public class FileDtoConverter { return dto; } + public static List toDTOs(List doObjList) { + if(CollectionUtil.isEmpty(doObjList)) { + return Lists.newArrayList(); + } + List dtoList = Lists.newArrayList(); + doObjList.forEach(doObj -> dtoList.add(toDTO(doObj))); + return dtoList; + } } diff --git a/src/main/java/com/knowledge/base/infrastructure/north/controller/FileWriteController.java b/src/main/java/com/knowledge/base/infrastructure/north/controller/FileWriteController.java index 7f6cd23..40b2b14 100644 --- a/src/main/java/com/knowledge/base/infrastructure/north/controller/FileWriteController.java +++ b/src/main/java/com/knowledge/base/infrastructure/north/controller/FileWriteController.java @@ -14,7 +14,7 @@ import com.knowledge.base.infrastructure.config.DynamicConfig; import com.knowledge.base.infrastructure.config.ObjectStorageProperties; import com.knowledge.base.infrastructure.north.dto.doc.OSRecordDTO; import com.knowledge.base.infrastructure.north.dto.user.UserDTO; -import com.knowledge.base.infrastructure.south.ObjectStorageGateway; +import com.knowledge.base.infrastructure.south.minio.ObjectStorageGateway; import com.knowledge.base.infrastructure.util.DateUtil; import com.knowledge.base.infrastructure.util.ThreadPoolUtil; import lombok.RequiredArgsConstructor; @@ -166,6 +166,26 @@ public class FileWriteController { } } + @PostMapping("/batch-delete-os-upload") + public ResponseEntity batchDeleteOSUpload(@CookieValue(value = ConstantConfig.COOKIE_KEY, required = false) String cookieToken, + @RequestHeader(value = "Authorization", required = false) String token, + @RequestBody Map body) { + // TODO 越权校验 + Object idsObj = body.get("recordIds"); + if (!(idsObj instanceof List)) { + return ResponseEntity.badRequest().body("recordIds 必须为数组"); + } + @SuppressWarnings("unchecked") + List idObjects = (List) idsObj; + List recordIds = idObjects.stream() + .filter(Objects::nonNull) + .map(Object::toString) + .map(Long::parseLong) + .collect(Collectors.toList()); + boolean removeRes = docAppService.batchRemoveOSFiles(true, recordIds); + return ResponseEntity.ok(Map.of("result", removeRes)); + } + private void handleAsyncRecord(String token, String bucket, String originFileNameWithSuffix, String suffix, String s3FullPath, String url, byte[] fileBytes, boolean searchable) { try { diff --git a/src/main/java/com/knowledge/base/infrastructure/repository/persistence/doc/OSRecordRepositoryImpl.java b/src/main/java/com/knowledge/base/infrastructure/repository/persistence/doc/OSRecordRepositoryImpl.java index 4d37aa4..10d4046 100644 --- a/src/main/java/com/knowledge/base/infrastructure/repository/persistence/doc/OSRecordRepositoryImpl.java +++ b/src/main/java/com/knowledge/base/infrastructure/repository/persistence/doc/OSRecordRepositoryImpl.java @@ -1,9 +1,11 @@ package com.knowledge.base.infrastructure.repository.persistence.doc; +import cn.hutool.core.collection.CollectionUtil; import cn.hutool.core.util.StrUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; +import com.google.common.collect.Lists; import com.knowledge.base.domain.common.model.PageResult; import com.knowledge.base.domain.doc.repository.iface.OSRecordRepository; import com.knowledge.base.domain.doc.repository.po.OSRecord; @@ -11,6 +13,7 @@ import com.knowledge.base.infrastructure.repository.mapper.doc.OSRecordMapper; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Repository; +import java.util.List; import java.util.Optional; @Repository @@ -46,4 +49,20 @@ public class OSRecordRepositoryImpl implements OSRecordRepository { .orderByDesc("upload_time")); return Optional.ofNullable(fileLatestRecord); } + + @Override + public List batchQueryOSRecord(List recordIds) { + if(CollectionUtil.isEmpty(recordIds)) { + return Lists.newArrayList(); + } + return osRecordMapper.selectBatchIds(recordIds); + } + + @Override + public boolean batchRemoveOSFiles(List recordIds) { + if(CollectionUtil.isEmpty(recordIds)) { + return true; + } + return osRecordMapper.deleteBatchIds(recordIds) > 0; + } } diff --git a/src/main/java/com/knowledge/base/infrastructure/south/es/FileElasticsearchGateway.java b/src/main/java/com/knowledge/base/infrastructure/south/es/FileElasticsearchGateway.java new file mode 100644 index 0000000..9faac1c --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/south/es/FileElasticsearchGateway.java @@ -0,0 +1,134 @@ +package com.knowledge.base.infrastructure.south.es; + +import com.knowledge.base.infrastructure.util.SafeIdUtil; +import lombok.extern.slf4j.Slf4j; +import org.elasticsearch.action.bulk.BulkRequest; +import org.elasticsearch.action.bulk.BulkResponse; +import org.elasticsearch.action.delete.DeleteRequest; +import org.elasticsearch.action.get.GetRequest; +import org.elasticsearch.action.get.GetResponse; +import org.elasticsearch.action.index.IndexRequest; +import org.elasticsearch.client.RequestOptions; +import org.elasticsearch.client.RestHighLevelClient; +import org.elasticsearch.index.query.TermQueryBuilder; +import org.elasticsearch.search.SearchHit; +import org.elasticsearch.search.builder.SearchSourceBuilder; +import org.elasticsearch.action.search.SearchRequest; +import org.elasticsearch.action.search.SearchResponse; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.io.IOException; +import java.util.List; +import java.util.Map; + +/** + * @author Luke.ye + * @date 2025/6/20 10:20 + */ +@Slf4j +@Component +public class FileElasticsearchGateway { + + private static final String INDEX_NAME = "documents"; + + @Autowired + private RestHighLevelClient esClient; + + /** + * 根据 filepath 批量删除文档 + * + * @param filepaths 相对路径列表 + */ + public void deleteByFilepaths(List filepaths) { + if (filepaths == null || filepaths.isEmpty()) { + return; + } + + try { + BulkRequest bulkDelete = new BulkRequest(); + for (String filepath : filepaths) { + String docId = SafeIdUtil.encode(filepath); // 和导入逻辑保持一致 + bulkDelete.add(new DeleteRequest(INDEX_NAME, docId)); + } + + BulkResponse response = esClient.bulk(bulkDelete, RequestOptions.DEFAULT); + if (response.hasFailures()) { + log.warn("部分删除失败: {}", response.buildFailureMessage()); + } else { + log.info("已成功删除 {} 个文档", filepaths.size()); + } + } catch (IOException e) { + log.error("批量删除文件失败", e); + } + } + + /** + * 根据字段删除（例如：filename/url 等） + * 可扩展调用者传字段名与值 + */ + public void deleteByField(String field, String value) { + try { + SearchSourceBuilder builder = new SearchSourceBuilder(); + builder.query(new TermQueryBuilder(field, value)).size(1000); // 单次最多匹配1000条 + SearchRequest searchRequest = new SearchRequest(INDEX_NAME).source(builder); + + SearchResponse response = esClient.search(searchRequest, RequestOptions.DEFAULT); + BulkRequest bulkRequest = new BulkRequest(); + for (SearchHit hit : response.getHits().getHits()) { + bulkRequest.add(new DeleteRequest(INDEX_NAME, hit.getId())); + } + + if (bulkRequest.numberOfActions() > 0) { + BulkResponse result = esClient.bulk(bulkRequest, RequestOptions.DEFAULT); + if (result.hasFailures()) { + log.warn("字段 {} 删除失败: {}", field, result.buildFailureMessage()); + } + } + + } catch (IOException e) { + log.error("deleteByField 出错", e); + } + } + + + /** + * 判断文档是否存在并获取内容 + */ + public GetResponse getIfExists(String index, String docId) { + try { + GetRequest getRequest = new GetRequest(index, docId); + if (!esClient.exists(getRequest, RequestOptions.DEFAULT)) { + return null; + } + return esClient.get(getRequest, RequestOptions.DEFAULT); + } catch (Exception e) { + log.error("ES 获取文档失败: index={}, docId={}", index, docId, e); + return null; + } + } + + /** + * 删除文档 + */ + public void deleteDoc(String index, String docId) { + try { + esClient.delete(new DeleteRequest(index, docId), RequestOptions.DEFAULT); + } catch (Exception e) { + log.warn("ES 删除文档失败: index={}, docId={}", index, docId, e); + } + } + + /** + * 写入/更新文档 + */ + public void saveDoc(String index, String docId, Map source) { + try { + IndexRequest request = new IndexRequest(index).id(docId).source(source); + esClient.index(request, RequestOptions.DEFAULT); + } catch (Exception e) { + log.error("ES 写入文档失败: index={}, docId={}", index, docId, e); + } + } +} + diff --git a/src/main/java/com/knowledge/base/infrastructure/south/MinioOSGatewayImpl.java b/src/main/java/com/knowledge/base/infrastructure/south/minio/MinioOSGatewayImpl.java similarity index 70% rename from src/main/java/com/knowledge/base/infrastructure/south/MinioOSGatewayImpl.java rename to src/main/java/com/knowledge/base/infrastructure/south/minio/MinioOSGatewayImpl.java index 8297296..badbba6 100644 --- a/src/main/java/com/knowledge/base/infrastructure/south/MinioOSGatewayImpl.java +++ b/src/main/java/com/knowledge/base/infrastructure/south/minio/MinioOSGatewayImpl.java @@ -1,8 +1,10 @@ -package com.knowledge.base.infrastructure.south; +package com.knowledge.base.infrastructure.south.minio; import com.knowledge.base.infrastructure.config.ObjectStorageProperties; import io.minio.*; import io.minio.http.Method; +import io.minio.messages.DeleteError; +import io.minio.messages.DeleteObject; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; @@ -10,6 +12,7 @@ import org.springframework.web.multipart.MultipartFile; import java.util.List; import java.util.Optional; +import java.util.stream.Collectors; @Component @Slf4j @@ -76,6 +79,37 @@ public class MinioOSGatewayImpl implements ObjectStorageGateway { } } + @Override + public void deleteObjects(String bucket, List objectPaths) { + if (objectPaths == null || objectPaths.isEmpty()) { + log.info("无需删除，objectPath 列表为空"); + return; + } + + try { + Iterable> results = minioClient.removeObjects( + RemoveObjectsArgs.builder() + .bucket(bucket) + .objects(objectPaths.stream() + .map(e -> new DeleteObject(e)) + .collect(Collectors.toList())) + .build() + ); + + for (Result result : results) { + DeleteError error = result.get(); + log.warn("删除失败: object={}, message={}", error.objectName(), error.message()); + } + + log.info("MinIO 批量删除完成，共 {} 个对象", objectPaths.size()); + + } catch (Exception e) { + log.error("MinIO 批量删除出错", e); + throw new RuntimeException("MinIO 批量删除失败", e); + } + } + + /** * 判断 bucket 是否公开（默认策略：通过配置控制） */ diff --git a/src/main/java/com/knowledge/base/infrastructure/south/ObjectStorageGateway.java b/src/main/java/com/knowledge/base/infrastructure/south/minio/ObjectStorageGateway.java similarity index 82% rename from src/main/java/com/knowledge/base/infrastructure/south/ObjectStorageGateway.java rename to src/main/java/com/knowledge/base/infrastructure/south/minio/ObjectStorageGateway.java index e786542..91d5e95 100644 --- a/src/main/java/com/knowledge/base/infrastructure/south/ObjectStorageGateway.java +++ b/src/main/java/com/knowledge/base/infrastructure/south/minio/ObjectStorageGateway.java @@ -1,7 +1,9 @@ -package com.knowledge.base.infrastructure.south; +package com.knowledge.base.infrastructure.south.minio; import org.springframework.web.multipart.MultipartFile; +import java.util.List; + public interface ObjectStorageGateway { /** @@ -21,4 +23,5 @@ public interface ObjectStorageGateway { */ String generateUrl(String bucket, String objectPath) throws Exception; + void deleteObjects(String bucket, List objectPaths); } diff --git a/src/main/java/com/knowledge/base/infrastructure/util/LocalFileUtil.java b/src/main/java/com/knowledge/base/infrastructure/util/LocalFileUtil.java new file mode 100644 index 0000000..1310843 --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/util/LocalFileUtil.java @@ -0,0 +1,75 @@ +package com.knowledge.base.infrastructure.util; + +import lombok.extern.slf4j.Slf4j; + +import java.io.IOException; +import java.nio.file.*; +import java.util.List; + + +/** + * @author Luke.ye + * @date 2025/6/20 10:45 + */ +@Slf4j +public class LocalFileUtil { + + /** + * 批量删除文件（根据 basePath + 相对路径 拼接） + * + * @param localPathPrefix 本地搜索路径前缀 + * @param relativePaths 文件相对路径列表 + * @return 删除成功数量 + */ + public static int deleteFilesByRelativePath(String localPathPrefix, List relativePaths) { + if (relativePaths == null || relativePaths.isEmpty()) { + log.info("无需删除文件，相对路径为空"); + return 0; + } + + int successCount = 0; + + for (String relativePath : relativePaths) { + Path path = Paths.get(localPathPrefix, relativePath).toAbsolutePath().normalize(); + try { + if (Files.exists(path) && Files.isRegularFile(path)) { + Files.delete(path); + log.info("已删除本地文件: {}", path); + successCount++; + } else { + log.warn("文件不存在或不是常规文件，跳过: {}", path); + } + } catch (IOException e) { + log.error("删除本地文件失败: {}", path, e); + } + } + + log.info("批量删除本地文件完成，共 {} 个请求，成功删除 {} 个", relativePaths.size(), successCount); + return successCount; + } + + /** + * 写入文件到指定目录下（自动创建目录） + * + * @param localPathPrefix 本地存储根路径（如 /data/searchable/yyyy-MM-dd） + * @param relativeDir 目标子目录路径（如 "a/b"） + * @param filename 文件名（如 "xx.pdf"） + * @param content 文件内容字节 + * @return 最终写入的绝对路径 + * @throws IOException 写入失败时抛出 + */ + public static Path writeFile(String localPathPrefix, String relativeDir, String filename, byte[] content) throws IOException { + Path dirPath = Paths.get(localPathPrefix, relativeDir).toAbsolutePath().normalize(); + if (!Files.exists(dirPath)) { + Files.createDirectories(dirPath); + log.info("创建本地目录: {}", dirPath); + } + + Path targetFilePath = dirPath.resolve(filename); + Files.write(targetFilePath, content); + log.info("写入本地文件成功: {}", targetFilePath); + + return targetFilePath; + } +} +