diff --git a/frontend/src/views/Calendar.vue b/frontend/src/views/Calendar.vue index 69249d2..b278f85 100644 --- a/frontend/src/views/Calendar.vue +++ b/frontend/src/views/Calendar.vue @@ -401,7 +401,13 @@ async function openDetail(year: number, month: number, date: number) { } } -/** 跳转到收支记录页并按当日筛选 */ +/** + * 跳转到收支记录页并按当日筛选 + * + * 日历入口以「日期」为唯一上下文，除日期范围与计入统计外， + * 需清空上次在记录页/统计面板残留的类型、分类、渠道、备注、金额、标签筛选， + * 避免只看到当天被筛掉后的部分记录（甚至空列表）。 + */ function goToTransactions() { const [y, m, d] = detailDate.value.split('-').map(Number) if (!y || !m || !d) return @@ -411,6 +417,13 @@ function goToTransactions() { new Date(y, m - 1, d, 23, 59, 59).getTime(), ], filterIncludeInStats: statsScopePref(), + filterType: [], + filterCategory: [], + filterChannel: [], + filterRemark: '', + filterMinAmount: null, + filterMaxAmount: null, + filterTag: [], }) detailShow.value = false router.push('/transactions') diff --git a/src/main/java/cn/rosercode/ledgerly/client/DeepSeekClient.java b/src/main/java/cn/rosercode/ledgerly/client/DeepSeekClient.java index 1eb6188..dd55b6f 100644 --- a/src/main/java/cn/rosercode/ledgerly/client/DeepSeekClient.java +++ b/src/main/java/cn/rosercode/ledgerly/client/DeepSeekClient.java @@ -17,6 +17,7 @@ import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.http.client.JdkClientHttpRequestFactory; import org.springframework.stereotype.Component; +import org.springframework.web.client.ResourceAccessException; import org.springframework.web.client.RestClient; import org.springframework.web.client.RestClientException; import org.springframework.web.client.RestClientResponseException; @@ -55,6 +56,9 @@ public class DeepSeekClient { /** Chat Completions 接口相对路径 */ private static final String CHAT_COMPLETIONS_PATH = "/chat/completions"; + /** 外部调用瞬态失败的最大重试次数（不含首次请求） */ + private static final int MAX_RETRIES = 2; + /** 按 baseUrl + apiKey + 超时 缓存的 HTTP 客户端，避免每次调用重复构建 */ private final Map clientCache = new ConcurrentHashMap<>(); @@ -131,7 +135,10 @@ public class DeepSeekClient { } /** - * 执行 Chat Completions 请求并解析内容与 token 用量。 + * 执行 Chat Completions 请求，遇到瞬态错误时按指数退避重试。 + * + *

LLM 属高延迟外部依赖，网络抖动、连接重置、502/503、429 等瞬态失败率不低， + * 这里对可恢复错误重试 {@link #MAX_RETRIES} 次（1s、2s 退避）。

* * @param client HTTP 客户端 * @param baseUrl API 基础地址 @@ -140,18 +147,74 @@ public class DeepSeekClient { * @return 模型输出内容与 token 用量 */ private DeepSeekResponse execute(RestClient client, String baseUrl, String model, Map payload) { - long start = System.currentTimeMillis(); - JsonNode response; - try { - response = client.post() - .uri(normalizeBaseUrl(baseUrl) + CHAT_COMPLETIONS_PATH) - .contentType(MediaType.APPLICATION_JSON) - .body(payload) - .retrieve() - .body(JsonNode.class); - } catch (RestClientException e) { - throw toBusinessException(model, e); + int maxAttempts = MAX_RETRIES + 1; + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + try { + return doExecute(client, baseUrl, model, payload); + } catch (RestClientException e) { + if (attempt >= maxAttempts || !isRetryable(e)) { + throw toBusinessException(model, e); + } + long backoff = (long) Math.pow(2, attempt - 1) * 1000L; + log.warn("调用 DeepSeek 失败，第 {} 次请求将重试: model={}, 退避={}ms, error={}", + 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 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 ? "" : response.path("choices").path(0).path("message").path("content").asText(""); diff --git a/src/main/java/cn/rosercode/ledgerly/config/WebConfig.java b/src/main/java/cn/rosercode/ledgerly/config/WebConfig.java index 9b6e2a2..7135f32 100644 --- a/src/main/java/cn/rosercode/ledgerly/config/WebConfig.java +++ b/src/main/java/cn/rosercode/ledgerly/config/WebConfig.java @@ -62,14 +62,21 @@ public class WebConfig implements WebMvcConfigurer { /** * 配置跨域策略。 - *

允许所有来源的跨域请求，支持常用 HTTP 方法，并允许携带凭据。

+ *

仅放开本机与局域网来源，避免任意站点携带凭据发起跨域请求；支持常用 HTTP 方法并允许携带凭据。 + * 生产部署前后端同源，无需跨域；开发场景为 5173/5174，局域网访问为 192.168.*、10.*、172.* 网段。

* * @param registry 跨域配置注册器 */ @Override public void addCorsMappings(CorsRegistry registry) { 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") .allowedHeaders("*") .allowCredentials(true) diff --git a/src/main/java/cn/rosercode/ledgerly/importer/BillArchiveExtractor.java b/src/main/java/cn/rosercode/ledgerly/importer/BillArchiveExtractor.java index 7092efe..3bdf148 100644 --- a/src/main/java/cn/rosercode/ledgerly/importer/BillArchiveExtractor.java +++ b/src/main/java/cn/rosercode/ledgerly/importer/BillArchiveExtractor.java @@ -7,6 +7,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.List; +import lombok.extern.slf4j.Slf4j; import net.lingala.zip4j.ZipFile; import net.lingala.zip4j.exception.ZipException; import net.lingala.zip4j.model.FileHeader; @@ -24,6 +25,7 @@ import cn.rosercode.ledgerly.common.BusinessException; * @date 2026-09-18 23:26:58 */ @Component +@Slf4j public class BillArchiveExtractor { /** 压缩包已加密、需要用户输入密码的业务错误码 */ @@ -81,6 +83,7 @@ public class BillArchiveExtractor { } } catch (IOException e) { // 读取失败按非账单压缩包处理，交由原解析器给出更准确的错误 + log.debug("读取压缩包条目失败，按非账单压缩包处理: {}", e.getMessage()); return null; } finally { deleteQuietly(temp); @@ -158,8 +161,9 @@ public class BillArchiveExtractor { } try { Files.deleteIfExists(temp); - } catch (IOException ignored) { - // 删除失败不影响解析结果 + } catch (IOException e) { + // 删除失败不影响解析结果，但需记录以便定位临时文件泄漏 + log.warn("临时文件删除失败: {}", temp, e); } } diff --git a/src/main/java/cn/rosercode/ledgerly/service/AiImageRecordStreamService.java b/src/main/java/cn/rosercode/ledgerly/service/AiImageRecordStreamService.java index 303c74c..d68c56c 100644 --- a/src/main/java/cn/rosercode/ledgerly/service/AiImageRecordStreamService.java +++ b/src/main/java/cn/rosercode/ledgerly/service/AiImageRecordStreamService.java @@ -4,6 +4,9 @@ import java.io.IOException; import java.util.List; import java.util.concurrent.ExecutorService; 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 com.fasterxml.jackson.core.JsonProcessingException; @@ -28,16 +31,32 @@ import cn.rosercode.ledgerly.vo.AiRecordImageResultVO; @Service 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 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.setDaemon(true); 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) { this.aiImageRecordService = aiImageRecordService; this.objectMapper = objectMapper; @@ -52,17 +71,18 @@ public class AiImageRecordStreamService { * @return SSE 发射器 */ public SseEmitter stream(Long userId, List images, String hint) { - SseEmitter emitter = new SseEmitter(0L); + SseEmitter emitter = new SseEmitter(EMITTER_TIMEOUT_MS); executor.submit(() -> run(emitter, userId, images, hint)); return emitter; } /** - * 应用关闭时停止线程池。 + * 应用关闭时停止线程池与心跳调度器。 */ @PreDestroy public void shutdown() { executor.shutdownNow(); + heartbeatScheduler.shutdownNow(); } /** @@ -74,6 +94,10 @@ public class AiImageRecordStreamService { * @param hint 用户补充说明 */ private void run(SseEmitter emitter, Long userId, List images, String hint) { + // 视觉模型调用可能长达数分钟，期间定期发送心跳避免连接被中间件断开 + ScheduledFuture heartbeat = heartbeatScheduler.scheduleAtFixedRate( + () -> sendHeartbeat(emitter), + HEARTBEAT_INTERVAL_SECONDS, HEARTBEAT_INTERVAL_SECONDS, TimeUnit.SECONDS); try { Consumer progress = line -> trySend(emitter, "progress", line); AiRecordImageResultVO result = aiImageRecordService.parse(userId, images, hint, progress); @@ -82,10 +106,26 @@ public class AiImageRecordStreamService { log.warn("图片记账流式识别失败: userId={}, error={}", userId, e.getMessage()); trySend(emitter, "failed", "{\"message\":\"" + escape(e.getMessage()) + "\"}"); } finally { + heartbeat.cancel(false); complete(emitter); } } + /** + * 发送 SSE 注释行作为心跳。 + * + *

前端 SSE 客户端会跳过以冒号开头的注释行，不产生业务事件，仅用于保持连接存活。

+ * + * @param emitter SSE 发射器 + */ + private void sendHeartbeat(SseEmitter emitter) { + try { + emitter.send(SseEmitter.event().comment("")); + } catch (IOException | IllegalStateException e) { + // 连接已断开，忽略；识别任务结束时统一 complete + } + } + /** * 推送一个 SSE 事件，客户端已断开时静默失败。 * diff --git a/src/main/java/cn/rosercode/ledgerly/service/impl/BillImportServiceImpl.java b/src/main/java/cn/rosercode/ledgerly/service/impl/BillImportServiceImpl.java index 60a3fff..a7589e8 100644 --- a/src/main/java/cn/rosercode/ledgerly/service/impl/BillImportServiceImpl.java +++ b/src/main/java/cn/rosercode/ledgerly/service/impl/BillImportServiceImpl.java @@ -4,7 +4,6 @@ import java.io.IOException; import java.math.BigDecimal; import java.util.ArrayList; import java.util.EnumMap; -import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashSet; 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.Tag; import cn.rosercode.ledgerly.entity.Transaction; -import cn.rosercode.ledgerly.entity.TransactionTag; import cn.rosercode.ledgerly.enums.BillType; import cn.rosercode.ledgerly.enums.TransactionType; 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.TagMapper; import cn.rosercode.ledgerly.mapper.TransactionMapper; -import cn.rosercode.ledgerly.mapper.TransactionTagMapper; import cn.rosercode.ledgerly.service.BillImportService; import cn.rosercode.ledgerly.service.RuleMatchContext; import cn.rosercode.ledgerly.service.RuleMatchResult; @@ -60,7 +57,7 @@ public class BillImportServiceImpl implements BillImportService { private final TransactionMapper transactionMapper; private final CategoryMapper categoryMapper; private final TagMapper tagMapper; - private final TransactionTagMapper transactionTagMapper; + private final BillImportWriteService billImportWriteService; private final RuleService ruleService; private final VoConverter voConverter; private final BillArchiveExtractor archiveExtractor; @@ -70,7 +67,7 @@ public class BillImportServiceImpl implements BillImportService { TransactionMapper transactionMapper, CategoryMapper categoryMapper, TagMapper tagMapper, - TransactionTagMapper transactionTagMapper, + BillImportWriteService billImportWriteService, RuleService ruleService, VoConverter voConverter, BillArchiveExtractor archiveExtractor, @@ -82,7 +79,7 @@ public class BillImportServiceImpl implements BillImportService { this.transactionMapper = transactionMapper; this.categoryMapper = categoryMapper; this.tagMapper = tagMapper; - this.transactionTagMapper = transactionTagMapper; + this.billImportWriteService = billImportWriteService; this.ruleService = ruleService; this.voConverter = voConverter; this.archiveExtractor = archiveExtractor; @@ -262,13 +259,10 @@ public class BillImportServiceImpl implements BillImportService { } } - // 批量插入交易记录，失败时回退逐条以定位问题行 - List inserted = insertTransactions(userId, pending, messages); + // 在独立事务中批量插入交易与标签关联，避免批量失败污染当前事务 + List inserted = billImportWriteService.write(userId, pending, pendingTagIds, messages); failed += pending.size() - inserted.size(); - // 批量插入标签关联（批量插入不回填主键，按交易号回查ID） - insertTransactionTags(userId, pending, pendingTagIds, inserted, messages); - BillImportResultVO result = new BillImportResultVO(); result.setTotal(items.size()); result.setImported(inserted.size()); @@ -280,123 +274,9 @@ public class BillImportServiceImpl implements BillImportService { return result; } - /** 批量插入的分批大小 */ + /** 批量查询的分批大小 */ private static final int BATCH_SIZE = 500; - /** - * 分批批量插入交易记录；某一批失败时回退逐条插入以定位并跳过问题行。 - * - * @param userId 用户ID - * @param pending 待插入的交易列表 - * @param messages 失败信息收集列表 - * @return 成功插入的交易列表 - */ - private List insertTransactions(Long userId, List pending, List messages) { - List inserted = new ArrayList<>(); - for (int i = 0; i < pending.size(); i += BATCH_SIZE) { - List 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(); - } - - /** - * 批量插入交易与标签的关联关系。 - * - *

批量插入不会回填自增主键，故对未回填主键的交易按交易号批量回查ID。

- * - * @param userId 用户ID - * @param pending 全部待插入交易（含失败行，用于与标签列表对齐） - * @param pendingTagIds 各交易对应的标签ID集合，下标与 pending 对应 - * @param inserted 成功插入的交易列表 - * @param messages 失败信息收集列表 - */ - private void insertTransactionTags(Long userId, List pending, List> pendingTagIds, - List inserted, List messages) { - // 未回填主键的交易按交易号批量回查ID - List missingExternalNos = inserted.stream() - .filter(transaction -> transaction.getId() == null) - .map(Transaction::getExternalNo) - .filter(StringUtils::hasText) - .distinct() - .collect(Collectors.toList()); - Map idByExternalNo = new HashMap<>(); - for (int i = 0; i < missingExternalNos.size(); i += BATCH_SIZE) { - List batch = missingExternalNos.subList(i, Math.min(i + BATCH_SIZE, missingExternalNos.size())); - List list = transactionMapper.selectList(new LambdaQueryWrapper() - .select(Transaction::getId, Transaction::getExternalNo) - .eq(Transaction::getUserId, userId) - .in(Transaction::getExternalNo, batch)); - for (Transaction transaction : list) { - idByExternalNo.put(transaction.getExternalNo(), transaction.getId()); - } - } - - List 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 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()) { return existing; } - List list = transactionMapper.selectList(new LambdaQueryWrapper() - .select(Transaction::getExternalNo) - .eq(Transaction::getUserId, userId) - .in(Transaction::getExternalNo, externalNos)); - for (Transaction t : list) { - if (t.getExternalNo() != null) { - existing.add(t.getExternalNo()); + // 分批查询，避免 IN 子句参数过多超出数据库限制或性能急剧下降 + for (int i = 0; i < externalNos.size(); i += BATCH_SIZE) { + List batch = externalNos.subList(i, Math.min(i + BATCH_SIZE, externalNos.size())); + List list = transactionMapper.selectList(new LambdaQueryWrapper() + .select(Transaction::getExternalNo) + .eq(Transaction::getUserId, userId) + .in(Transaction::getExternalNo, batch)); + for (Transaction t : list) { + if (t.getExternalNo() != null) { + existing.add(t.getExternalNo()); + } } } return existing; diff --git a/src/main/java/cn/rosercode/ledgerly/service/impl/BillImportWriteService.java b/src/main/java/cn/rosercode/ledgerly/service/impl/BillImportWriteService.java new file mode 100644 index 0000000..f9647bf --- /dev/null +++ b/src/main/java/cn/rosercode/ledgerly/service/impl/BillImportWriteService.java @@ -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; + +/** + * 账单导入的写入组件。 + * + *

将「批量插入交易 → 失败回退逐条 → 建立标签关联」放入独立的新事务 + * （{@link Propagation#REQUIRES_NEW}）执行：批量插入失败时不会污染调用方事务， + * 使单行失败可被跳过而不影响其余记录。

+ * + * @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 write(Long userId, List pending, List> pendingTagIds, + List messages) { + List inserted = insertTransactions(userId, pending, messages); + insertTransactionTags(userId, pending, pendingTagIds, inserted, messages); + return inserted; + } + + /** + * 分批批量插入交易记录；某一批失败时回退逐条插入以定位并跳过问题行。 + * + * @param userId 用户ID + * @param pending 待插入的交易列表 + * @param messages 失败信息收集列表 + * @return 成功插入的交易列表 + */ + private List insertTransactions(Long userId, List pending, List messages) { + List inserted = new ArrayList<>(); + for (int i = 0; i < pending.size(); i += BATCH_SIZE) { + List 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; + } + + /** + * 批量插入交易与标签的关联关系。 + * + *

批量插入不会回填自增主键，故对未回填主键的交易按交易号批量回查ID。

+ * + * @param userId 用户ID + * @param pending 全部待插入交易（含失败行，用于与标签列表对齐） + * @param pendingTagIds 各交易对应的标签ID集合，下标与 pending 对应 + * @param inserted 成功插入的交易列表 + * @param messages 失败信息收集列表 + */ + private void insertTransactionTags(Long userId, List pending, List> pendingTagIds, + List inserted, List messages) { + // 未回填主键的交易按交易号批量回查ID + List missingExternalNos = inserted.stream() + .filter(transaction -> transaction.getId() == null) + .map(Transaction::getExternalNo) + .filter(StringUtils::hasText) + .distinct() + .collect(Collectors.toList()); + Map idByExternalNo = new HashMap<>(); + for (int i = 0; i < missingExternalNos.size(); i += BATCH_SIZE) { + List batch = missingExternalNos.subList(i, Math.min(i + BATCH_SIZE, missingExternalNos.size())); + List list = transactionMapper.selectList(new LambdaQueryWrapper() + .select(Transaction::getId, Transaction::getExternalNo) + .eq(Transaction::getUserId, userId) + .in(Transaction::getExternalNo, batch)); + for (Transaction transaction : list) { + idByExternalNo.put(transaction.getExternalNo(), transaction.getId()); + } + } + + List 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 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(); + } +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index b9bbc00..7568b0b 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -28,7 +28,8 @@ mybatis-plus: log-impl: org.apache.ibatis.logging.slf4j.Slf4jImpl jwt: - secret: finance-app-secret-key-2026-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6 + # 生产环境通过环境变量 JWT_SECRET 覆盖；未配置时回退内置默认值，避免个人自托管场景重启后登录态失效 + secret: ${JWT_SECRET:finance-app-secret-key-2026-a1b2c3d4e5f6g7h8i9j0k1l2m3n4o5p6} expiration: 86400000 springdoc: