fix: 修复代码审查安全与稳定性问题并清理日历跳转残留筛选

修复第三方代码审查报告中的安全、账单导入事务与 AI 稳定性问题,并附带修复日历页跳转收支记录时未清空残留筛选条件、可能只显示当天被筛掉后部分记录的问题。

变更内容:

- 安全性:JWT 签名密钥改为环境变量 JWT_SECRET 优先、未配置回退内置默认;CORS 由全放开收窄为 localhost 与 192.168/10/172 局域网网段
- 账单导入:新增 BillImportWriteService,以 REQUIRES_NEW 独立事务承载批量插入与失败回退,避免批量失败污染外层事务;findExistingExternalNos 改为分批查询;压缩包临时文件删除失败记录 warn、读取条目失败记录 debug
- AI 稳定性:DeepSeekClient 对超时/连接层异常/5xx/429 做指数退避重试(最多 2 次);图片记账 SSE 线程池固定为 8 线程、连接 5 分钟超时,并每 15s 发送注释行心跳
- 日历:跳转收支记录页时清空上次残留的类型/分类/渠道/备注/金额/标签筛选
This commit is contained in:
2026-10-09 14:28:43 +08:00
parent 8c4f343c34
commit 4735f0bf85
8 changed files with 343 additions and 155 deletions
+14 -1
View File
@@ -401,7 +401,13 @@ async function openDetail(year: number, month: number, date: number) {
} }
} }
/** 跳转到收支记录页并按当日筛选 */ /**
* 跳转到收支记录页并按当日筛选
*
* 日历入口以「日期」为唯一上下文,除日期范围与计入统计外,
* 需清空上次在记录页/统计面板残留的类型、分类、渠道、备注、金额、标签筛选,
* 避免只看到当天被筛掉后的部分记录(甚至空列表)。
*/
function goToTransactions() { function goToTransactions() {
const [y, m, d] = detailDate.value.split('-').map(Number) const [y, m, d] = detailDate.value.split('-').map(Number)
if (!y || !m || !d) return if (!y || !m || !d) return
@@ -411,6 +417,13 @@ function goToTransactions() {
new Date(y, m - 1, d, 23, 59, 59).getTime(), new Date(y, m - 1, d, 23, 59, 59).getTime(),
], ],
filterIncludeInStats: statsScopePref(), filterIncludeInStats: statsScopePref(),
filterType: [],
filterCategory: [],
filterChannel: [],
filterRemark: '',
filterMinAmount: null,
filterMaxAmount: null,
filterTag: [],
}) })
detailShow.value = false detailShow.value = false
router.push('/transactions') router.push('/transactions')
@@ -17,6 +17,7 @@ import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType; import org.springframework.http.MediaType;
import org.springframework.http.client.JdkClientHttpRequestFactory; import org.springframework.http.client.JdkClientHttpRequestFactory;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import org.springframework.web.client.ResourceAccessException;
import org.springframework.web.client.RestClient; import org.springframework.web.client.RestClient;
import org.springframework.web.client.RestClientException; import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestClientResponseException; import org.springframework.web.client.RestClientResponseException;
@@ -55,6 +56,9 @@ public class DeepSeekClient {
/** Chat Completions 接口相对路径 */ /** Chat Completions 接口相对路径 */
private static final String CHAT_COMPLETIONS_PATH = "/chat/completions"; private static final String CHAT_COMPLETIONS_PATH = "/chat/completions";
/** 外部调用瞬态失败的最大重试次数(不含首次请求) */
private static final int MAX_RETRIES = 2;
/** 按 baseUrl + apiKey + 超时 缓存的 HTTP 客户端,避免每次调用重复构建 */ /** 按 baseUrl + apiKey + 超时 缓存的 HTTP 客户端,避免每次调用重复构建 */
private final Map<String, RestClient> clientCache = new ConcurrentHashMap<>(); private final Map<String, RestClient> clientCache = new ConcurrentHashMap<>();
@@ -131,7 +135,10 @@ public class DeepSeekClient {
} }
/** /**
* 执行 Chat Completions 请求并解析内容与 token 用量。 * 执行 Chat Completions 请求,遇到瞬态错误时按指数退避重试。
*
* <p>LLM 属高延迟外部依赖,网络抖动、连接重置、502/503、429 等瞬态失败率不低,
* 这里对可恢复错误重试 {@link #MAX_RETRIES} 次(1s、2s 退避)。</p>
* *
* @param client HTTP 客户端 * @param client HTTP 客户端
* @param baseUrl API 基础地址 * @param baseUrl API 基础地址
@@ -140,18 +147,74 @@ public class DeepSeekClient {
* @return 模型输出内容与 token 用量 * @return 模型输出内容与 token 用量
*/ */
private DeepSeekResponse execute(RestClient client, String baseUrl, String model, Map<String, Object> payload) { private DeepSeekResponse execute(RestClient client, String baseUrl, String model, Map<String, Object> payload) {
long start = System.currentTimeMillis(); int maxAttempts = MAX_RETRIES + 1;
JsonNode response; for (int attempt = 1; attempt <= maxAttempts; attempt++) {
try { try {
response = client.post() return doExecute(client, baseUrl, model, payload);
.uri(normalizeBaseUrl(baseUrl) + CHAT_COMPLETIONS_PATH) } catch (RestClientException e) {
.contentType(MediaType.APPLICATION_JSON) if (attempt >= maxAttempts || !isRetryable(e)) {
.body(payload) throw toBusinessException(model, e);
.retrieve() }
.body(JsonNode.class); long backoff = (long) Math.pow(2, attempt - 1) * 1000L;
} catch (RestClientException e) { log.warn("调用 DeepSeek 失败,第 {} 次请求将重试: model={}, 退避={}ms, error={}",
throw toBusinessException(model, e); attempt, model, backoff, rootCauseMessage(e));
sleep(backoff);
}
} }
throw new BusinessException("调用 DeepSeek 失败");
}
/**
* 判断异常是否为可重试的瞬态错误。
*
* @param e HTTP 调用异常
* @return 可重试返回 true
*/
private boolean isRetryable(RestClientException e) {
// 超时或连接层异常(连接被重置、拒绝、DNS 解析失败等)可重试
if (isTimeout(e) || e instanceof ResourceAccessException) {
return true;
}
// 服务端错误与限流可重试,4xx 客户端错误重试无意义
if (e instanceof RestClientResponseException responseException) {
int status = responseException.getStatusCode().value();
return status >= 500 || status == 429;
}
return false;
}
/**
* 线程休眠指定毫秒;被中断时恢复中断标记并抛出业务异常。
*
* @param millis 休眠毫秒数
*/
private void sleep(long millis) {
try {
Thread.sleep(millis);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new BusinessException("调用 DeepSeek 被中断");
}
}
/**
* 实际执行一次 Chat Completions 请求并解析内容与 token 用量。
*
* @param client HTTP 客户端
* @param baseUrl API 基础地址
* @param model 模型名称(仅用于日志)
* @param payload 请求体
* @return 模型输出内容与 token 用量
*/
private DeepSeekResponse doExecute(RestClient client, String baseUrl, String model, Map<String, Object> payload) {
long start = System.currentTimeMillis();
// 此处不捕获 RestClientException,交由 execute 判断是否可重试
JsonNode response = client.post()
.uri(normalizeBaseUrl(baseUrl) + CHAT_COMPLETIONS_PATH)
.contentType(MediaType.APPLICATION_JSON)
.body(payload)
.retrieve()
.body(JsonNode.class);
String content = response == null ? "" String content = response == null ? ""
: response.path("choices").path(0).path("message").path("content").asText(""); : response.path("choices").path(0).path("message").path("content").asText("");
@@ -62,14 +62,21 @@ public class WebConfig implements WebMvcConfigurer {
/** /**
* 配置跨域策略。 * 配置跨域策略。
* <p>允许所有来源的跨域请求,支持常用 HTTP 方法,并允许携带凭据。</p> * <p>仅放开本机与局域网来源,避免任意站点携带凭据发起跨域请求;支持常用 HTTP 方法并允许携带凭据。
* 生产部署前后端同源,无需跨域;开发场景为 5173/5174,局域网访问为 192.168.*、10.*、172.* 网段。</p>
* *
* @param registry 跨域配置注册器 * @param registry 跨域配置注册器
*/ */
@Override @Override
public void addCorsMappings(CorsRegistry registry) { public void addCorsMappings(CorsRegistry registry) {
registry.addMapping("/**") registry.addMapping("/**")
.allowedOriginPatterns("*") .allowedOriginPatterns(
"http://localhost:*",
"http://127.0.0.1:*",
"http://[::1]:*",
"http://192.168.*.*:*",
"http://10.*.*.*:*",
"http://172.*.*.*:*")
.allowedMethods("GET", "POST", "PUT", "DELETE", "OPTIONS") .allowedMethods("GET", "POST", "PUT", "DELETE", "OPTIONS")
.allowedHeaders("*") .allowedHeaders("*")
.allowCredentials(true) .allowCredentials(true)
@@ -7,6 +7,7 @@ import java.nio.file.Files;
import java.nio.file.Path; import java.nio.file.Path;
import java.util.List; import java.util.List;
import lombok.extern.slf4j.Slf4j;
import net.lingala.zip4j.ZipFile; import net.lingala.zip4j.ZipFile;
import net.lingala.zip4j.exception.ZipException; import net.lingala.zip4j.exception.ZipException;
import net.lingala.zip4j.model.FileHeader; import net.lingala.zip4j.model.FileHeader;
@@ -24,6 +25,7 @@ import cn.rosercode.ledgerly.common.BusinessException;
* @date 2026-09-18 23:26:58 * @date 2026-09-18 23:26:58
*/ */
@Component @Component
@Slf4j
public class BillArchiveExtractor { public class BillArchiveExtractor {
/** 压缩包已加密、需要用户输入密码的业务错误码 */ /** 压缩包已加密、需要用户输入密码的业务错误码 */
@@ -81,6 +83,7 @@ public class BillArchiveExtractor {
} }
} catch (IOException e) { } catch (IOException e) {
// 读取失败按非账单压缩包处理,交由原解析器给出更准确的错误 // 读取失败按非账单压缩包处理,交由原解析器给出更准确的错误
log.debug("读取压缩包条目失败,按非账单压缩包处理: {}", e.getMessage());
return null; return null;
} finally { } finally {
deleteQuietly(temp); deleteQuietly(temp);
@@ -158,8 +161,9 @@ public class BillArchiveExtractor {
} }
try { try {
Files.deleteIfExists(temp); Files.deleteIfExists(temp);
} catch (IOException ignored) { } catch (IOException e) {
// 删除失败不影响解析结果 // 删除失败不影响解析结果,但需记录以便定位临时文件泄漏
log.warn("临时文件删除失败: {}", temp, e);
} }
} }
@@ -4,6 +4,9 @@ import java.io.IOException;
import java.util.List; import java.util.List;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors; import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer; import java.util.function.Consumer;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
@@ -28,16 +31,32 @@ import cn.rosercode.ledgerly.vo.AiRecordImageResultVO;
@Service @Service
public class AiImageRecordStreamService { public class AiImageRecordStreamService {
/** SSE 连接超时时间(5 分钟),避免识别任务挂起导致连接永不释放 */
private static final long EMITTER_TIMEOUT_MS = 300_000L;
/** 心跳发送间隔(秒),用于保持长连接不被中间件按空闲超时切断 */
private static final long HEARTBEAT_INTERVAL_SECONDS = 15L;
/** 识别线程池上限,避免并发上传大量图片时无限创建线程 */
private static final int EXECUTOR_THREADS = 8;
private final AiImageRecordService aiImageRecordService; private final AiImageRecordService aiImageRecordService;
private final ObjectMapper objectMapper; private final ObjectMapper objectMapper;
/** 执行识别任务的守护线程池 */ /** 执行识别任务的守护线程池(固定大小,任务超出后排队) */
private final ExecutorService executor = Executors.newCachedThreadPool(runnable -> { private final ExecutorService executor = Executors.newFixedThreadPool(EXECUTOR_THREADS, runnable -> {
Thread thread = new Thread(runnable, "ai-image-record-stream"); Thread thread = new Thread(runnable, "ai-image-record-stream");
thread.setDaemon(true); thread.setDaemon(true);
return thread; return thread;
}); });
/** 心跳调度器,定期向 SSE 连接发送注释行 */
private final ScheduledExecutorService heartbeatScheduler = Executors.newSingleThreadScheduledExecutor(runnable -> {
Thread thread = new Thread(runnable, "ai-image-record-heartbeat");
thread.setDaemon(true);
return thread;
});
public AiImageRecordStreamService(AiImageRecordService aiImageRecordService, ObjectMapper objectMapper) { public AiImageRecordStreamService(AiImageRecordService aiImageRecordService, ObjectMapper objectMapper) {
this.aiImageRecordService = aiImageRecordService; this.aiImageRecordService = aiImageRecordService;
this.objectMapper = objectMapper; this.objectMapper = objectMapper;
@@ -52,17 +71,18 @@ public class AiImageRecordStreamService {
* @return SSE 发射器 * @return SSE 发射器
*/ */
public SseEmitter stream(Long userId, List<byte[]> images, String hint) { public SseEmitter stream(Long userId, List<byte[]> images, String hint) {
SseEmitter emitter = new SseEmitter(0L); SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MS);
executor.submit(() -> run(emitter, userId, images, hint)); executor.submit(() -> run(emitter, userId, images, hint));
return emitter; return emitter;
} }
/** /**
* 应用关闭时停止线程池。 * 应用关闭时停止线程池与心跳调度器。
*/ */
@PreDestroy @PreDestroy
public void shutdown() { public void shutdown() {
executor.shutdownNow(); executor.shutdownNow();
heartbeatScheduler.shutdownNow();
} }
/** /**
@@ -74,6 +94,10 @@ public class AiImageRecordStreamService {
* @param hint 用户补充说明 * @param hint 用户补充说明
*/ */
private void run(SseEmitter emitter, Long userId, List<byte[]> images, String hint) { private void run(SseEmitter emitter, Long userId, List<byte[]> images, String hint) {
// 视觉模型调用可能长达数分钟,期间定期发送心跳避免连接被中间件断开
ScheduledFuture<?> heartbeat = heartbeatScheduler.scheduleAtFixedRate(
() -> sendHeartbeat(emitter),
HEARTBEAT_INTERVAL_SECONDS, HEARTBEAT_INTERVAL_SECONDS, TimeUnit.SECONDS);
try { try {
Consumer<String> progress = line -> trySend(emitter, "progress", line); Consumer<String> progress = line -> trySend(emitter, "progress", line);
AiRecordImageResultVO result = aiImageRecordService.parse(userId, images, hint, progress); AiRecordImageResultVO result = aiImageRecordService.parse(userId, images, hint, progress);
@@ -82,10 +106,26 @@ public class AiImageRecordStreamService {
log.warn("图片记账流式识别失败: userId={}, error={}", userId, e.getMessage()); log.warn("图片记账流式识别失败: userId={}, error={}", userId, e.getMessage());
trySend(emitter, "failed", "{\"message\":\"" + escape(e.getMessage()) + "\"}"); trySend(emitter, "failed", "{\"message\":\"" + escape(e.getMessage()) + "\"}");
} finally { } finally {
heartbeat.cancel(false);
complete(emitter); complete(emitter);
} }
} }
/**
* 发送 SSE 注释行作为心跳。
*
* <p>前端 SSE 客户端会跳过以冒号开头的注释行,不产生业务事件,仅用于保持连接存活。</p>
*
* @param emitter SSE 发射器
*/
private void sendHeartbeat(SseEmitter emitter) {
try {
emitter.send(SseEmitter.event().comment(""));
} catch (IOException | IllegalStateException e) {
// 连接已断开,忽略;识别任务结束时统一 complete
}
}
/** /**
* 推送一个 SSE 事件,客户端已断开时静默失败。 * 推送一个 SSE 事件,客户端已断开时静默失败。
* *
@@ -4,7 +4,6 @@ import java.io.IOException;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.EnumMap; import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet; import java.util.HashSet;
import java.util.LinkedHashSet; import java.util.LinkedHashSet;
import java.util.List; import java.util.List;
@@ -25,7 +24,6 @@ import cn.rosercode.ledgerly.dto.BillImportRequest;
import cn.rosercode.ledgerly.entity.Category; import cn.rosercode.ledgerly.entity.Category;
import cn.rosercode.ledgerly.entity.Tag; import cn.rosercode.ledgerly.entity.Tag;
import cn.rosercode.ledgerly.entity.Transaction; import cn.rosercode.ledgerly.entity.Transaction;
import cn.rosercode.ledgerly.entity.TransactionTag;
import cn.rosercode.ledgerly.enums.BillType; import cn.rosercode.ledgerly.enums.BillType;
import cn.rosercode.ledgerly.enums.TransactionType; import cn.rosercode.ledgerly.enums.TransactionType;
import cn.rosercode.ledgerly.importer.BillArchiveExtractor; import cn.rosercode.ledgerly.importer.BillArchiveExtractor;
@@ -34,7 +32,6 @@ import cn.rosercode.ledgerly.importer.ParsedBillRow;
import cn.rosercode.ledgerly.mapper.CategoryMapper; import cn.rosercode.ledgerly.mapper.CategoryMapper;
import cn.rosercode.ledgerly.mapper.TagMapper; import cn.rosercode.ledgerly.mapper.TagMapper;
import cn.rosercode.ledgerly.mapper.TransactionMapper; import cn.rosercode.ledgerly.mapper.TransactionMapper;
import cn.rosercode.ledgerly.mapper.TransactionTagMapper;
import cn.rosercode.ledgerly.service.BillImportService; import cn.rosercode.ledgerly.service.BillImportService;
import cn.rosercode.ledgerly.service.RuleMatchContext; import cn.rosercode.ledgerly.service.RuleMatchContext;
import cn.rosercode.ledgerly.service.RuleMatchResult; import cn.rosercode.ledgerly.service.RuleMatchResult;
@@ -60,7 +57,7 @@ public class BillImportServiceImpl implements BillImportService {
private final TransactionMapper transactionMapper; private final TransactionMapper transactionMapper;
private final CategoryMapper categoryMapper; private final CategoryMapper categoryMapper;
private final TagMapper tagMapper; private final TagMapper tagMapper;
private final TransactionTagMapper transactionTagMapper; private final BillImportWriteService billImportWriteService;
private final RuleService ruleService; private final RuleService ruleService;
private final VoConverter voConverter; private final VoConverter voConverter;
private final BillArchiveExtractor archiveExtractor; private final BillArchiveExtractor archiveExtractor;
@@ -70,7 +67,7 @@ public class BillImportServiceImpl implements BillImportService {
TransactionMapper transactionMapper, TransactionMapper transactionMapper,
CategoryMapper categoryMapper, CategoryMapper categoryMapper,
TagMapper tagMapper, TagMapper tagMapper,
TransactionTagMapper transactionTagMapper, BillImportWriteService billImportWriteService,
RuleService ruleService, RuleService ruleService,
VoConverter voConverter, VoConverter voConverter,
BillArchiveExtractor archiveExtractor, BillArchiveExtractor archiveExtractor,
@@ -82,7 +79,7 @@ public class BillImportServiceImpl implements BillImportService {
this.transactionMapper = transactionMapper; this.transactionMapper = transactionMapper;
this.categoryMapper = categoryMapper; this.categoryMapper = categoryMapper;
this.tagMapper = tagMapper; this.tagMapper = tagMapper;
this.transactionTagMapper = transactionTagMapper; this.billImportWriteService = billImportWriteService;
this.ruleService = ruleService; this.ruleService = ruleService;
this.voConverter = voConverter; this.voConverter = voConverter;
this.archiveExtractor = archiveExtractor; this.archiveExtractor = archiveExtractor;
@@ -262,13 +259,10 @@ public class BillImportServiceImpl implements BillImportService {
} }
} }
// 批量插入交易记录,失败时回退逐条以定位问题行 // 在独立事务中批量插入交易与标签关联,避免批量失败污染当前事务
List<Transaction> inserted = insertTransactions(userId, pending, messages); List<Transaction> inserted = billImportWriteService.write(userId, pending, pendingTagIds, messages);
failed += pending.size() - inserted.size(); failed += pending.size() - inserted.size();
// 批量插入标签关联(批量插入不回填主键,按交易号回查ID)
insertTransactionTags(userId, pending, pendingTagIds, inserted, messages);
BillImportResultVO result = new BillImportResultVO(); BillImportResultVO result = new BillImportResultVO();
result.setTotal(items.size()); result.setTotal(items.size());
result.setImported(inserted.size()); result.setImported(inserted.size());
@@ -280,123 +274,9 @@ public class BillImportServiceImpl implements BillImportService {
return result; return result;
} }
/** 批量插入的分批大小 */ /** 批量查询的分批大小 */
private static final int BATCH_SIZE = 500; private static final int BATCH_SIZE = 500;
/**
* 分批批量插入交易记录;某一批失败时回退逐条插入以定位并跳过问题行。
*
* @param userId 用户ID
* @param pending 待插入的交易列表
* @param messages 失败信息收集列表
* @return 成功插入的交易列表
*/
private List<Transaction> insertTransactions(Long userId, List<Transaction> pending, List<String> messages) {
List<Transaction> inserted = new ArrayList<>();
for (int i = 0; i < pending.size(); i += BATCH_SIZE) {
List<Transaction> batch = pending.subList(i, Math.min(i + BATCH_SIZE, pending.size()));
try {
transactionMapper.insertBatch(batch);
inserted.addAll(batch);
} catch (RuntimeException e) {
// 批量失败时逐条重试,跳过并记录问题行
log.warn("账单批量插入交易失败,回退逐条插入: userId={}, 批次大小={}, 原因={}",
userId, batch.size(), rootMessage(e));
for (Transaction transaction : batch) {
try {
transactionMapper.insert(transaction);
inserted.add(transaction);
} catch (RuntimeException ex) {
messages.add(transaction.getExternalNo() + ":导入失败," + ex.getMessage());
log.warn("账单交易导入失败: userId={}, externalNo={}, 原因={}",
userId, transaction.getExternalNo(), rootMessage(ex));
}
}
}
}
return inserted;
}
/**
* 获取异常链中最底层异常的消息,避免日志被 MyBatis 包装信息淹没。
*
* @param throwable 异常
* @return 根因消息
*/
private String rootMessage(Throwable throwable) {
Throwable cause = throwable;
while (cause.getCause() != null && cause.getCause() != cause) {
cause = cause.getCause();
}
return cause.getMessage() != null ? cause.getMessage() : cause.getClass().getSimpleName();
}
/**
* 批量插入交易与标签的关联关系。
*
* <p>批量插入不会回填自增主键,故对未回填主键的交易按交易号批量回查ID。</p>
*
* @param userId 用户ID
* @param pending 全部待插入交易(含失败行,用于与标签列表对齐)
* @param pendingTagIds 各交易对应的标签ID集合,下标与 pending 对应
* @param inserted 成功插入的交易列表
* @param messages 失败信息收集列表
*/
private void insertTransactionTags(Long userId, List<Transaction> pending, List<Set<Long>> pendingTagIds,
List<Transaction> inserted, List<String> messages) {
// 未回填主键的交易按交易号批量回查ID
List<String> missingExternalNos = inserted.stream()
.filter(transaction -> transaction.getId() == null)
.map(Transaction::getExternalNo)
.filter(StringUtils::hasText)
.distinct()
.collect(Collectors.toList());
Map<String, Long> idByExternalNo = new HashMap<>();
for (int i = 0; i < missingExternalNos.size(); i += BATCH_SIZE) {
List<String> batch = missingExternalNos.subList(i, Math.min(i + BATCH_SIZE, missingExternalNos.size()));
List<Transaction> list = transactionMapper.selectList(new LambdaQueryWrapper<Transaction>()
.select(Transaction::getId, Transaction::getExternalNo)
.eq(Transaction::getUserId, userId)
.in(Transaction::getExternalNo, batch));
for (Transaction transaction : list) {
idByExternalNo.put(transaction.getExternalNo(), transaction.getId());
}
}
List<TransactionTag> relations = new ArrayList<>();
for (int i = 0; i < pending.size(); i++) {
Transaction transaction = pending.get(i);
Long transactionId = transaction.getId() != null
? transaction.getId()
: idByExternalNo.get(transaction.getExternalNo());
// 插入失败或无交易号导致无法定位的记录,跳过标签关联
if (transactionId == null) {
continue;
}
for (Long tagId : pendingTagIds.get(i)) {
TransactionTag relation = new TransactionTag();
relation.setTransactionId(transactionId);
relation.setTagId(tagId);
relations.add(relation);
}
}
for (int i = 0; i < relations.size(); i += BATCH_SIZE) {
List<TransactionTag> batch = relations.subList(i, Math.min(i + BATCH_SIZE, relations.size()));
try {
transactionTagMapper.insertBatch(batch);
} catch (RuntimeException e) {
for (TransactionTag relation : batch) {
try {
transactionTagMapper.insert(relation);
} catch (RuntimeException ignored) {
// 关联失败不影响交易导入结果
}
}
}
}
}
/** /**
* 查询指定交易号中已存在的记录 * 查询指定交易号中已存在的记录
* *
@@ -409,13 +289,17 @@ public class BillImportServiceImpl implements BillImportService {
if (externalNos.isEmpty()) { if (externalNos.isEmpty()) {
return existing; return existing;
} }
List<Transaction> list = transactionMapper.selectList(new LambdaQueryWrapper<Transaction>() // 分批查询,避免 IN 子句参数过多超出数据库限制或性能急剧下降
.select(Transaction::getExternalNo) for (int i = 0; i < externalNos.size(); i += BATCH_SIZE) {
.eq(Transaction::getUserId, userId) List<String> batch = externalNos.subList(i, Math.min(i + BATCH_SIZE, externalNos.size()));
.in(Transaction::getExternalNo, externalNos)); List<Transaction> list = transactionMapper.selectList(new LambdaQueryWrapper<Transaction>()
for (Transaction t : list) { .select(Transaction::getExternalNo)
if (t.getExternalNo() != null) { .eq(Transaction::getUserId, userId)
existing.add(t.getExternalNo()); .in(Transaction::getExternalNo, batch));
for (Transaction t : list) {
if (t.getExternalNo() != null) {
existing.add(t.getExternalNo());
}
} }
} }
return existing; return existing;
@@ -0,0 +1,176 @@
package cn.rosercode.ledgerly.service.impl;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
import cn.rosercode.ledgerly.entity.Transaction;
import cn.rosercode.ledgerly.entity.TransactionTag;
import cn.rosercode.ledgerly.mapper.TransactionMapper;
import cn.rosercode.ledgerly.mapper.TransactionTagMapper;
/**
* 账单导入的写入组件。
*
* <p>将「批量插入交易 → 失败回退逐条 → 建立标签关联」放入独立的新事务
* ({@link Propagation#REQUIRES_NEW})执行:批量插入失败时不会污染调用方事务,
* 使单行失败可被跳过而不影响其余记录。</p>
*
* @date 2026-10-09 10:53:49
*/
@Service
@Slf4j
public class BillImportWriteService {
/** 批量插入的分批大小 */
private static final int BATCH_SIZE = 500;
private final TransactionMapper transactionMapper;
private final TransactionTagMapper transactionTagMapper;
public BillImportWriteService(TransactionMapper transactionMapper, TransactionTagMapper transactionTagMapper) {
this.transactionMapper = transactionMapper;
this.transactionTagMapper = transactionTagMapper;
}
/**
* 在新事务中批量写入交易记录及其标签关联。
*
* @param userId 用户ID
* @param pending 待插入的交易列表(含失败行,用于与标签列表对齐)
* @param pendingTagIds 各交易对应的标签ID集合,下标与 pending 对应
* @param messages 失败信息收集列表
* @return 成功插入的交易列表
*/
@Transactional(propagation = Propagation.REQUIRES_NEW)
public List<Transaction> write(Long userId, List<Transaction> pending, List<Set<Long>> pendingTagIds,
List<String> messages) {
List<Transaction> inserted = insertTransactions(userId, pending, messages);
insertTransactionTags(userId, pending, pendingTagIds, inserted, messages);
return inserted;
}
/**
* 分批批量插入交易记录;某一批失败时回退逐条插入以定位并跳过问题行。
*
* @param userId 用户ID
* @param pending 待插入的交易列表
* @param messages 失败信息收集列表
* @return 成功插入的交易列表
*/
private List<Transaction> insertTransactions(Long userId, List<Transaction> pending, List<String> messages) {
List<Transaction> inserted = new ArrayList<>();
for (int i = 0; i < pending.size(); i += BATCH_SIZE) {
List<Transaction> batch = pending.subList(i, Math.min(i + BATCH_SIZE, pending.size()));
try {
transactionMapper.insertBatch(batch);
inserted.addAll(batch);
} catch (RuntimeException e) {
// 批量失败时逐条重试,跳过并记录问题行
log.warn("账单批量插入交易失败,回退逐条插入: userId={}, 批次大小={}, 原因={}",
userId, batch.size(), rootMessage(e));
for (Transaction transaction : batch) {
try {
transactionMapper.insert(transaction);
inserted.add(transaction);
} catch (RuntimeException ex) {
messages.add(transaction.getExternalNo() + ":导入失败," + ex.getMessage());
log.warn("账单交易导入失败: userId={}, externalNo={}, 原因={}",
userId, transaction.getExternalNo(), rootMessage(ex));
}
}
}
}
return inserted;
}
/**
* 批量插入交易与标签的关联关系。
*
* <p>批量插入不会回填自增主键,故对未回填主键的交易按交易号批量回查ID。</p>
*
* @param userId 用户ID
* @param pending 全部待插入交易(含失败行,用于与标签列表对齐)
* @param pendingTagIds 各交易对应的标签ID集合,下标与 pending 对应
* @param inserted 成功插入的交易列表
* @param messages 失败信息收集列表
*/
private void insertTransactionTags(Long userId, List<Transaction> pending, List<Set<Long>> pendingTagIds,
List<Transaction> inserted, List<String> messages) {
// 未回填主键的交易按交易号批量回查ID
List<String> missingExternalNos = inserted.stream()
.filter(transaction -> transaction.getId() == null)
.map(Transaction::getExternalNo)
.filter(StringUtils::hasText)
.distinct()
.collect(Collectors.toList());
Map<String, Long> idByExternalNo = new HashMap<>();
for (int i = 0; i < missingExternalNos.size(); i += BATCH_SIZE) {
List<String> batch = missingExternalNos.subList(i, Math.min(i + BATCH_SIZE, missingExternalNos.size()));
List<Transaction> list = transactionMapper.selectList(new LambdaQueryWrapper<Transaction>()
.select(Transaction::getId, Transaction::getExternalNo)
.eq(Transaction::getUserId, userId)
.in(Transaction::getExternalNo, batch));
for (Transaction transaction : list) {
idByExternalNo.put(transaction.getExternalNo(), transaction.getId());
}
}
List<TransactionTag> relations = new ArrayList<>();
for (int i = 0; i < pending.size(); i++) {
Transaction transaction = pending.get(i);
Long transactionId = transaction.getId() != null
? transaction.getId()
: idByExternalNo.get(transaction.getExternalNo());
// 插入失败或无交易号导致无法定位的记录,跳过标签关联
if (transactionId == null) {
continue;
}
for (Long tagId : pendingTagIds.get(i)) {
TransactionTag relation = new TransactionTag();
relation.setTransactionId(transactionId);
relation.setTagId(tagId);
relations.add(relation);
}
}
for (int i = 0; i < relations.size(); i += BATCH_SIZE) {
List<TransactionTag> batch = relations.subList(i, Math.min(i + BATCH_SIZE, relations.size()));
try {
transactionTagMapper.insertBatch(batch);
} catch (RuntimeException e) {
for (TransactionTag relation : batch) {
try {
transactionTagMapper.insert(relation);
} catch (RuntimeException ignored) {
// 关联失败不影响交易导入结果
}
}
}
}
}
/**
* 获取异常链中最底层异常的消息,避免日志被 MyBatis 包装信息淹没。
*
* @param throwable 异常
* @return 根因消息
*/
private String rootMessage(Throwable throwable) {
Throwable cause = throwable;
while (cause.getCause() != null && cause.getCause() != cause) {
cause = cause.getCause();
}
return cause.getMessage() != null ? cause.getMessage() : cause.getClass().getSimpleName();
}
}
+2 -1
View File
@@ -28,7 +28,8 @@ mybatis-plus:
log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl
jwt: jwt:
secret: finance-app-secret-key-2026-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6 # 生产环境通过环境变量 JWT_SECRET 覆盖;未配置时回退内置默认值,避免个人自托管场景重启后登录态失效
secret: ${JWT_SECRET:finance-app-secret-key-2026-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6}
expiration: 86400000 expiration: 86400000
springdoc: springdoc: