370 lines
16 KiB
Java
370 lines
16 KiB
Java
package com.labelsys.backend.service;
|
||
|
||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
|
||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||
import com.labelsys.backend.common.ResultCode;
|
||
import com.labelsys.backend.common.exception.BusinessException;
|
||
import com.labelsys.backend.context.LoginUser;
|
||
import com.labelsys.backend.dto.common.PageResult;
|
||
import com.labelsys.backend.dto.request.AnnotationResultHistoryPageQuery;
|
||
import com.labelsys.backend.dto.response.AnnotationResultHistoryDetailResponse;
|
||
import com.labelsys.backend.dto.response.AnnotationResultHistoryResponse;
|
||
import com.labelsys.backend.dto.response.FileContentResponse;
|
||
import com.labelsys.backend.dto.response.MergeReviewResultResponse;
|
||
import com.labelsys.backend.entity.AnnotationResult;
|
||
import com.labelsys.backend.entity.AnnotationResultHistory;
|
||
import com.labelsys.backend.enums.UserPosition;
|
||
import com.labelsys.backend.enums.UserRole;
|
||
import com.labelsys.backend.mapper.AnnotationResultHistoryMapper;
|
||
import com.labelsys.backend.mapper.AnnotationResultMapper;
|
||
import com.labelsys.backend.util.IdGenerator;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.beans.factory.annotation.Value;
|
||
import org.springframework.stereotype.Service;
|
||
import org.springframework.transaction.annotation.Transactional;
|
||
|
||
import java.nio.charset.StandardCharsets;
|
||
import java.time.Duration;
|
||
import java.time.LocalDateTime;
|
||
import java.util.List;
|
||
|
||
import static org.springframework.util.StringUtils.hasText;
|
||
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class AnnotationResultArchiveService {
|
||
|
||
private static final String MANUAL_ARCHIVE_REASON = "MANUAL_REVIEW";
|
||
|
||
private final AnnotationResultMapper annotationResultMapper;
|
||
private final AnnotationResultHistoryMapper annotationResultHistoryMapper;
|
||
private final ObjectStorageService objectStorageService;
|
||
private final ObjectMapper objectMapper;
|
||
private final DataPermissionService dataPermissionService;
|
||
|
||
@Value("${labelsys.annotation.auto-archive-timeout:PT2H}")
|
||
private Duration autoArchiveTimeout;
|
||
|
||
public PageResult<AnnotationResultHistoryResponse> pageHistory(LoginUser currentUser,
|
||
AnnotationResultHistoryPageQuery query) {
|
||
try {
|
||
List<String> allowedRoles = dataPermissionService.getAllowedRoles(currentUser);
|
||
boolean shouldFilterByUserId = dataPermissionService.shouldFilterByUserId(currentUser);
|
||
|
||
var wrapper = new LambdaQueryWrapper<AnnotationResultHistory>()
|
||
.eq(AnnotationResultHistory::getCompanyId, currentUser.companyId())
|
||
.eq(query.taskId() != null, AnnotationResultHistory::getTaskId, query.taskId())
|
||
.eq(query.resourceId() != null, AnnotationResultHistory::getResourceId, query.resourceId())
|
||
.orderByDesc(AnnotationResultHistory::getCreatedAt);
|
||
|
||
if (shouldFilterByUserId) {
|
||
wrapper.eq(AnnotationResultHistory::getCreatorId, currentUser.userId());
|
||
} else if (!allowedRoles.isEmpty()) {
|
||
wrapper.in(AnnotationResultHistory::getCreatorRole, allowedRoles);
|
||
}
|
||
|
||
var page = new Page<AnnotationResultHistory>(query.pageNo(), query.pageSize());
|
||
var resultPage = annotationResultHistoryMapper.selectPage(page, wrapper);
|
||
|
||
// 分页查询不加载 qa 内容
|
||
var records = resultPage.getRecords().stream()
|
||
.map(this::toResponse)
|
||
.toList();
|
||
|
||
return new PageResult<>(records, resultPage.getTotal(), (int) resultPage.getCurrent(),
|
||
(int) resultPage.getSize());
|
||
} catch (Exception e) {
|
||
log.error("pageHistory failed, companyId={}, userId={}, error={}",
|
||
currentUser.companyId(), currentUser.userId(), e.getMessage(), e);
|
||
throw e;
|
||
}
|
||
}
|
||
|
||
public AnnotationResultHistoryDetailResponse getHistory(LoginUser currentUser, Long historyId) {
|
||
try {
|
||
AnnotationResultHistory history = annotationResultHistoryMapper.selectById(historyId);
|
||
if (history == null || !history.getCompanyId().equals(currentUser.companyId())) {
|
||
throw new BusinessException(ResultCode.NOT_FOUND, "历史记录不存在");
|
||
}
|
||
assertHistoryPermission(currentUser, history);
|
||
|
||
// 详情查询加载 QA 内容
|
||
QaContent qaContent = loadQaContent(history.getQaContentFilePath());
|
||
return toDetailResponse(history, qaContent);
|
||
} catch (Exception e) {
|
||
log.error("getHistory failed, companyId={}, userId={}, historyId={}, error={}",
|
||
currentUser.companyId(), currentUser.userId(), historyId, e.getMessage(), e);
|
||
throw e;
|
||
}
|
||
}
|
||
|
||
private void assertHistoryPermission(LoginUser currentUser, AnnotationResultHistory history) {
|
||
if (!dataPermissionService.canAccessCreator(currentUser, history.getCreatorId(),
|
||
UserRole.valueOf(history.getCreatorRole()))) {
|
||
throw new BusinessException(ResultCode.FORBIDDEN, "无权访问该归档记录");
|
||
}
|
||
}
|
||
|
||
private AnnotationResultHistoryResponse toResponse(AnnotationResultHistory history) {
|
||
return new AnnotationResultHistoryResponse(
|
||
history.getId(),
|
||
history.getTaskId(),
|
||
history.getTaskName(),
|
||
history.getResourceId(),
|
||
history.getResourceName(),
|
||
history.getQaContentFilePath(),
|
||
history.getArchiveReason(),
|
||
history.getArchivedBy(),
|
||
history.getArchivedAt(),
|
||
history.getCreatedAt(),
|
||
history.getReviewerId(),
|
||
history.getReviewerName()
|
||
);
|
||
}
|
||
|
||
private AnnotationResultHistoryDetailResponse toDetailResponse(AnnotationResultHistory history,
|
||
QaContent qaContent) {
|
||
// 转换 QA 内容
|
||
AnnotationResultHistoryDetailResponse.QaContentDto qaContentDto = null;
|
||
if (qaContent != null && qaContent.records() != null) {
|
||
qaContentDto = new AnnotationResultHistoryDetailResponse.QaContentDto(
|
||
qaContent.records().stream()
|
||
.map(r -> new AnnotationResultHistoryDetailResponse.QaRecordDto(
|
||
r.id(),
|
||
r.batchId(),
|
||
r.question(),
|
||
r.answer(),
|
||
r.requiresReview(),
|
||
r.sourceSegments() != null ?
|
||
new AnnotationResultHistoryDetailResponse.SourceSegmentsDto(
|
||
r.sourceSegments().segment(),
|
||
r.sourceSegments().chunkIndex(),
|
||
r.sourceSegments().chunkTitle(),
|
||
r.sourceSegments().chunkContent()) :
|
||
null,
|
||
r.questionCategory(),
|
||
r.scores() != null ? new AnnotationResultHistoryDetailResponse.ScoresDto(
|
||
r.scores().similarity(),
|
||
r.scores().confidence1(),
|
||
r.scores().confidence2(),
|
||
r.scores().hallucination(),
|
||
r.scores().trust()) : null,
|
||
r.reviewComment()))
|
||
.toList()
|
||
);
|
||
}
|
||
|
||
return new AnnotationResultHistoryDetailResponse(
|
||
history.getId(),
|
||
history.getTaskId(),
|
||
history.getTaskName(),
|
||
history.getResourceId(),
|
||
history.getResourceName(),
|
||
history.getQaContentFilePath(),
|
||
qaContentDto,
|
||
history.getArchiveReason(),
|
||
history.getArchivedBy(),
|
||
history.getArchivedAt(),
|
||
history.getCreatedAt(),
|
||
history.getReviewerId(),
|
||
history.getReviewerName()
|
||
);
|
||
}
|
||
|
||
@Transactional
|
||
public int autoArchiveEligibleResults() {
|
||
try {
|
||
LocalDateTime cutoff = LocalDateTime.now().minus(autoArchiveTimeout);
|
||
List<AnnotationResult> results = annotationResultMapper.selectList(
|
||
new LambdaQueryWrapper<AnnotationResult>()
|
||
.eq(AnnotationResult::getIsDeleted, false)
|
||
.eq(AnnotationResult::getRequiresManualReview, false)
|
||
.lt(AnnotationResult::getCreatedAt, cutoff));
|
||
int archivedCount = 0;
|
||
for (AnnotationResult result : results) {
|
||
if (archiveRuntimeResult(result, null, "AUTO_ARCHIVE") != null) {
|
||
archivedCount++;
|
||
}
|
||
}
|
||
return archivedCount;
|
||
} catch (Exception e) {
|
||
log.error("autoArchiveEligibleResults failed, error={}", e.getMessage(), e);
|
||
throw e;
|
||
}
|
||
}
|
||
|
||
private void assertReviewer(LoginUser currentUser) {
|
||
if (currentUser.position() != UserPosition.REVIEWER && currentUser.position() != UserPosition.ADMIN) {
|
||
throw new BusinessException(ResultCode.FORBIDDEN, "当前用户没有审核权限");
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 归档运行态标注结果到历史表
|
||
* 从对象存储读取 qa.json 内容进行归档
|
||
*/
|
||
private MergeReviewResultResponse archiveRuntimeResult(AnnotationResult result,
|
||
Long reviewerId,
|
||
String archiveReason) {
|
||
LocalDateTime archivedAt = LocalDateTime.now();
|
||
|
||
int updated = annotationResultMapper.markArchived(
|
||
result.getId(),
|
||
result.getCompanyId(),
|
||
reviewerId);
|
||
if (updated == 0) {
|
||
return null;
|
||
}
|
||
|
||
AnnotationResultHistory history = AnnotationResultHistory.builder()
|
||
.id(IdGenerator.nextId())
|
||
.companyId(result.getCompanyId())
|
||
.creatorId(result.getCreatorId())
|
||
.creatorRole(result.getCreatorRole())
|
||
.sourceResultId(result.getId())
|
||
.taskId(result.getTaskId())
|
||
.taskName(result.getTaskName())
|
||
.resourceId(result.getResourceId())
|
||
.resourceName(result.getResourceName())
|
||
//.qaContentJson(qaContentJson) // 使用从对象存储读取的内容
|
||
.qaContentFilePath(result.getQaContentFilePath())
|
||
.archiveReason(archiveReason)
|
||
.archivedBy(reviewerId)
|
||
.archivedAt(archivedAt)
|
||
.createdAt(archivedAt)
|
||
.reviewerId(null)
|
||
.reviewerName("auto")
|
||
.build();
|
||
annotationResultHistoryMapper.insert(history);
|
||
return new MergeReviewResultResponse(result.getId(), history.getId(), archiveReason, archivedAt);
|
||
}
|
||
|
||
/**
|
||
* 加载 QA 内容
|
||
*/
|
||
private QaContent loadQaContent(String filePath) {
|
||
try {
|
||
if (!hasText(filePath)) {
|
||
log.warn("QA content file path is empty");
|
||
return new QaContent(null, null, List.of(), null);
|
||
}
|
||
String bucketName = extractBucketName(filePath);
|
||
String objectKey = extractObjectKey(filePath);
|
||
byte[] content = objectStorageService.download(bucketName, objectKey);
|
||
String jsonContent = new String(content, StandardCharsets.UTF_8);
|
||
return objectMapper.readValue(jsonContent, new com.fasterxml.jackson.core.type.TypeReference<QaContent>() {
|
||
});
|
||
} catch (Exception e) {
|
||
log.warn("Failed to load QA content, returning empty content. filePath={}, error={}", filePath,
|
||
e.getMessage());
|
||
return new QaContent(null, null, List.of(), null);
|
||
}
|
||
}
|
||
|
||
// 内部类：qa.json 结构
|
||
private record QaContent(
|
||
Long taskId,
|
||
Long resourceId,
|
||
List<QaRecord> records,
|
||
Metadata metadata
|
||
) {
|
||
private record QaRecord(String id, Long batchId, String question, String answer,
|
||
Boolean requiresReview, SourceSegments sourceSegments,
|
||
String questionCategory, Scores scores, String reviewComment) {
|
||
}
|
||
|
||
private record SourceSegments(String segment, Integer chunkIndex, String chunkTitle, String chunkContent) {
|
||
}
|
||
|
||
private record Scores(Double similarity, Double confidence1, Double confidence2,
|
||
Double hallucination, Double trust) {
|
||
}
|
||
|
||
private record Metadata(String createdAt, String updatedAt) {
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 从对象存储读取 qa.json 内容
|
||
*/
|
||
private String loadQaContentJson(AnnotationResult result) {
|
||
try {
|
||
String filePath = result.getQaContentFilePath();
|
||
if (filePath == null || filePath.isEmpty()) {
|
||
log.warn("qa_content_file_path is null or empty, resultId={}", result.getId());
|
||
return "{}";
|
||
}
|
||
|
||
// 解析文件路径，提取 bucket 和 object key
|
||
String bucketName = extractBucketName(filePath);
|
||
String objectKey = extractObjectKey(filePath);
|
||
|
||
// 从对象存储下载文件内容
|
||
byte[] content = objectStorageService.download(bucketName, objectKey);
|
||
return new String(content, StandardCharsets.UTF_8);
|
||
} catch (Exception e) {
|
||
log.error("Failed to load qa content from object storage, resultId={}, filePath={}",
|
||
result.getId(), result.getQaContentFilePath(), e);
|
||
// 如果读取失败，返回空 JSON
|
||
return "{}";
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 从文件路径中提取 bucket 名称
|
||
* 例如：annotation-results/2/qa/801.json -> annotation-results
|
||
*/
|
||
private String extractBucketName(String filePath) {
|
||
int firstSlash = filePath.indexOf('/');
|
||
return firstSlash > 0 ? filePath.substring(0, firstSlash) : filePath;
|
||
}
|
||
|
||
/**
|
||
* 从文件路径中提取 object key
|
||
* 例如：annotation-results/2/qa/801.json -> 2/qa/801.json
|
||
*/
|
||
private String extractObjectKey(String filePath) {
|
||
int firstSlash = filePath.indexOf('/');
|
||
return firstSlash > 0 ? filePath.substring(firstSlash + 1) : "";
|
||
}
|
||
|
||
/**
|
||
* 加载归档记录的文件内容
|
||
*
|
||
* @param currentUser 当前用户
|
||
* @param historyId 历史记录ID
|
||
* @return 文件内容响应
|
||
*/
|
||
public FileContentResponse loadFileContent(LoginUser currentUser, Long historyId) {
|
||
try {
|
||
AnnotationResultHistory history = annotationResultHistoryMapper.selectById(historyId);
|
||
if (history == null || !history.getCompanyId().equals(currentUser.companyId())) {
|
||
throw new BusinessException(ResultCode.NOT_FOUND, "历史记录不存在");
|
||
}
|
||
//assertHistoryPermission(currentUser, history);
|
||
|
||
String filePath = history.getQaContentFilePath();
|
||
if (filePath == null || filePath.isEmpty()) {
|
||
throw new BusinessException(ResultCode.ERROR, "文件路径为空");
|
||
}
|
||
|
||
String bucketName = extractBucketName(filePath);
|
||
String objectKey = extractObjectKey(filePath);
|
||
byte[] content = objectStorageService.download(bucketName, objectKey);
|
||
String contentStr = new String(content, StandardCharsets.UTF_8);
|
||
|
||
return new FileContentResponse(filePath, contentStr, content.length);
|
||
} catch (BusinessException e) {
|
||
throw e;
|
||
} catch (Exception e) {
|
||
log.error("loadFileContent failed, companyId={}, userId={}, historyId={}, error={}",
|
||
currentUser.companyId(), currentUser.userId(), historyId, e.getMessage(), e);
|
||
throw new BusinessException(ResultCode.ERROR, "加载文件内容失败");
|
||
}
|
||
}
|
||
|
||
}
|