diff --git a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelDownstreamGapExcel.java b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelDownstreamGapExcel.java new file mode 100644 index 00000000..60c9524c --- /dev/null +++ b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelDownstreamGapExcel.java @@ -0,0 +1,43 @@ +package com.bsd.pjcloud.supermarket.dto; + +import com.alibaba.excel.annotation.ExcelProperty; +import com.alibaba.excel.annotation.write.style.ColumnWidth; +import lombok.Data; + +/** + * 下游圣诺完善清单:getInfo 无店,或房型/产品编码匹配不上。 + */ +@Data +@ColumnWidth(18) +public class SmHotelDownstreamGapExcel { + + @ExcelProperty("同步日志ID") + private String syncLogId; + + @ExcelProperty("下游伙伴") + private String partnerCode; + + @ExcelProperty("本系统酒店ID") + private String hotelId; + + @ExcelProperty("酒店编码") + private String hotelCode; + + @ExcelProperty("酒店名称") + @ColumnWidth(28) + private String hotelName; + + @ExcelProperty("外部酒店ID(eid)") + private String eid; + + @ExcelProperty("问题类型") + private String issueType; + + @ExcelProperty("问题明细") + @ColumnWidth(48) + private String issueDetail; + + @ExcelProperty("处理建议") + @ColumnWidth(36) + private String suggest; +} diff --git a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelSyncFailExcel.java b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelSyncFailExcel.java new file mode 100644 index 00000000..96846d24 --- /dev/null +++ b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/dto/SmHotelSyncFailExcel.java @@ -0,0 +1,29 @@ +package com.bsd.pjcloud.supermarket.dto; + +import com.alibaba.excel.annotation.ExcelProperty; +import lombok.Data; + +/** + * 同步日志失败明细导出(hotelCode|hotelName|原因)。 + */ +@Data +public class SmHotelSyncFailExcel { + + @ExcelProperty("同步日志ID") + private Long syncLogId; + + @ExcelProperty("类型") + private String syncType; + + @ExcelProperty("渠道") + private String dataSource; + + @ExcelProperty("酒店编码") + private String hotelCode; + + @ExcelProperty("酒店名称") + private String hotelName; + + @ExcelProperty("失败原因") + private String reason; +} diff --git a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelSyncLogEntity.java b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelSyncLogEntity.java index 4980707c..d9d483cf 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelSyncLogEntity.java +++ b/pjcloud-supermarket/pjcloud-supermarket-api/src/main/java/com/bsd/pjcloud/supermarket/entity/SmHotelSyncLogEntity.java @@ -31,7 +31,7 @@ public class SmHotelSyncLogEntity extends Model { @Schema(description = "渠道/数据来源") private String dataSource; - @Schema(description = "RUNNING / SUCCESS / FAILED") + @Schema(description = "RUNNING / PAUSED / SUCCESS / FAILED") private String status; private String stage; diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/HotelDownstreamProperties.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/HotelDownstreamProperties.java index f2e6756d..917cc7a8 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/HotelDownstreamProperties.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/config/HotelDownstreamProperties.java @@ -151,4 +151,10 @@ public class HotelDownstreamProperties { /** 资源日更 cron(默认每天 02:30) */ private String resourcePushCron = "0 30 2 * * ?"; + + /** + * 下游全量推酒店并发(圣诺等 HTTP 出站)。1=串行;建议 4~8。 + * 对方限流时报错再调低。伙伴 YAML 可覆盖。 + */ + private int resourcePushConcurrency = 6; } diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java index 0b62172a..c533b82e 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelDownstreamPushLogController.java @@ -3,7 +3,9 @@ package com.bsd.pjcloud.supermarket.controller; import cn.hutool.core.util.StrUtil; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.bsd.pjcloud.common.core.util.R; +import com.bsd.pjcloud.common.excel.annotation.ResponseExcel; import com.bsd.pjcloud.common.log.annotation.SysLog; +import com.bsd.pjcloud.supermarket.dto.SmHotelDownstreamGapExcel; import com.bsd.pjcloud.supermarket.entity.SmHotelDownstreamPushLogEntity; import com.bsd.pjcloud.supermarket.service.SmHotelDownstreamInboundRetryService; import com.bsd.pjcloud.supermarket.service.SmHotelDownstreamPushLogService; @@ -17,6 +19,7 @@ import org.springframework.http.HttpHeaders; import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.web.bind.annotation.*; +import java.util.List; import java.util.Map; @RestController @@ -40,6 +43,14 @@ public class SmHotelDownstreamPushLogController { return R.ok(pushLogService.pageLogs(page, query)); } + @Operation(summary = "导出某次下游全量推完善清单:getInfo无店 + 房型/产品未匹配(不限行数)") + @ResponseExcel + @GetMapping("/exportGaps") + @PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_view')") + public List exportGaps(@RequestParam Long syncLogId) { + return pushLogService.exportGapsBySyncLogId(syncLogId); + } + @Operation(summary = "详情") @GetMapping("/{id}") @PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_view')") diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java index 0b54c537..4c4ec5b3 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/controller/SmHotelMidController.java @@ -1,7 +1,9 @@ package com.bsd.pjcloud.supermarket.controller; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; +import com.bsd.pjcloud.common.excel.annotation.ResponseExcel; import com.bsd.pjcloud.supermarket.dto.SmHotelLocalDailyBatchDTO; +import com.bsd.pjcloud.supermarket.dto.SmHotelSyncFailExcel; import com.bsd.pjcloud.supermarket.entity.SmHotelDailyEntity; import com.bsd.pjcloud.supermarket.entity.SmHotelPriceConfigEntity; import com.bsd.pjcloud.supermarket.entity.SmHotelProductEntity; @@ -226,13 +228,37 @@ public class SmHotelMidController { } @SysLog("终止酒店同步死任务") - @Operation(summary = "终止仍显示进行中的同步日志(死任务)") + @Operation(summary = "终止进行中或已暂停的同步日志(保留失败明细)") @PostMapping("/mid/syncLog/{id}/abort") @PreAuthorize("@pms.hasPermission('supermarket_smHotel_sync')") public R abortSyncLog(@PathVariable Long id) { return syncService.abortSyncLog(id); } + @SysLog("暂停酒店同步任务") + @Operation(summary = "暂停下游全量推/全量详情(批次结束后落盘,可继续)") + @PostMapping("/mid/syncLog/{id}/pause") + @PreAuthorize("@pms.hasPermission('supermarket_smHotel_sync')") + public R pauseSyncLog(@PathVariable Long id) { + return syncService.pauseSyncLog(id); + } + + @SysLog("继续酒店同步任务") + @Operation(summary = "继续已暂停的同步任务(同 logId 接着跑)") + @PostMapping("/mid/syncLog/{id}/resume") + @PreAuthorize("@pms.hasPermission('supermarket_smHotel_sync')") + public R resumeSyncLog(@PathVariable Long id) { + return syncService.resumeSyncLog(id); + } + + @Operation(summary = "导出同步日志失败明细") + @ResponseExcel + @GetMapping("/mid/syncLog/{id}/exportFail") + @PreAuthorize("@pms.hasPermission('supermarket_smHotel_sync')") + public List exportSyncFail(@PathVariable Long id) { + return syncService.exportFailDetail(id); + } + @SysLog("清理酒店同步卡死任务") @Operation(summary = "清理长时间无进度的 RUNNING 同步日志") @PostMapping("/mid/syncLog/abortStale") diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java index 14550f61..c4e173f7 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelDownstreamPushLogService.java @@ -3,8 +3,11 @@ package com.bsd.pjcloud.supermarket.service; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.baomidou.mybatisplus.extension.service.IService; import com.bsd.pjcloud.common.core.util.R; +import com.bsd.pjcloud.supermarket.dto.SmHotelDownstreamGapExcel; import com.bsd.pjcloud.supermarket.entity.SmHotelDownstreamPushLogEntity; +import java.util.List; + public interface SmHotelDownstreamPushLogService extends IService { Page pageLogs(Page page, @@ -21,4 +24,10 @@ public interface SmHotelDownstreamPushLogService extends IService removeByCreateTimeRange(String createTimeBegin, String createTimeEnd); + + /** + * 按某次下游全量推同步日志导出完善清单(不限行数): + * ① 圣诺 getInfo 无店/失败;② getInfo 成功但房型/产品编码匹配不上。 + */ + List exportGapsBySyncLogId(Long syncLogId); } diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java index fa4c3de3..a40c82b8 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/SmHotelSyncService.java @@ -56,13 +56,29 @@ public interface SmHotelSyncService { java.util.List filterDownstreamPushHotelIds(String partnerCode, java.util.List hotelIds); /** - * 根据某次同步日志的 failDetail,只重试失败酒店(支持 DETAIL_PULL / PRICE_ROLL)。 + * 根据某次同步日志的 failDetail,只重试失败酒店(支持 DETAIL_PULL / PRICE_ROLL / DOWNSTREAM_PUSH)。 + * 会新开一条 RETRY 日志,可反复:完善数据 → 再点重试,直到失败明细清空。 */ R startRetryFailedAsync(Long sourceLogId); - /** 人工终止一条仍显示进行中的同步日志(死任务) */ + /** 人工终止一条仍显示进行中或已暂停的同步日志(保留已落库失败明细) */ R abortSyncLog(Long id); + /** + * 暂停进行中的长任务(下游全量推/全量详情等)。当前批次结束后落盘进度与失败明细,可稍后「继续」。 + * 发布代码前请先暂停,避免进程重启丢进度。 + */ + R pauseSyncLog(Long id); + + /** + * 继续已暂停任务(同 logId 接着跑,不丢已成功/失败计数与明细)。 + * 当前支持 DOWNSTREAM_PUSH / DETAIL_PULL。 + */ + R resumeSyncLog(Long id); + + /** 导出失败明细(failDetail 解析为行) */ + java.util.List exportFailDetail(Long id); + /** * 将长时间无进度更新的 RUNNING 日志标为失败,避免挡后续任务。 * @return 清理条数 diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java index 0769d4a5..4e4c271f 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelDownstreamPushLogServiceImpl.java @@ -1,12 +1,21 @@ package com.bsd.pjcloud.supermarket.service.impl; +import cn.hutool.core.collection.CollUtil; import cn.hutool.core.util.StrUtil; +import cn.hutool.json.JSONArray; +import cn.hutool.json.JSONObject; +import cn.hutool.json.JSONUtil; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; import com.bsd.pjcloud.common.core.util.R; +import com.bsd.pjcloud.supermarket.dto.SmHotelDownstreamGapExcel; import com.bsd.pjcloud.supermarket.entity.SmHotelDownstreamPushLogEntity; +import com.bsd.pjcloud.supermarket.entity.SmHotelEntity; +import com.bsd.pjcloud.supermarket.entity.SmHotelSyncLogEntity; import com.bsd.pjcloud.supermarket.mapper.SmHotelDownstreamPushLogMapper; +import com.bsd.pjcloud.supermarket.mapper.SmHotelMapper; +import com.bsd.pjcloud.supermarket.mapper.SmHotelSyncLogMapper; import com.bsd.pjcloud.supermarket.service.SmHotelDownstreamPushLogService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -15,6 +24,13 @@ import org.springframework.stereotype.Service; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; @Slf4j @Service @@ -23,6 +39,10 @@ public class SmHotelDownstreamPushLogServiceImpl extends ServiceImpl implements SmHotelDownstreamPushLogService { + private final SmHotelSyncLogMapper syncLogMapper; + + private final SmHotelMapper hotelMapper; + @Override public Page pageLogs(Page page, SmHotelDownstreamPushLogEntity query) { @@ -108,6 +128,208 @@ public class SmHotelDownstreamPushLogServiceImpl return R.ok(n, "已删除 " + n + " 条"); } + @Override + public List exportGapsBySyncLogId(Long syncLogId) { + if (syncLogId == null) { + throw new IllegalArgumentException("请指定同步日志ID"); + } + SmHotelSyncLogEntity syncLog = syncLogMapper.selectById(syncLogId); + if (syncLog == null) { + throw new IllegalArgumentException("同步日志不存在:" + syncLogId); + } + String partner = StrUtil.blankToDefault(syncLog.getDataSource(), "").trim().toLowerCase(); + List out = new ArrayList<>(); + Set dedupe = new HashSet<>(); + Map hotelMap = new HashMap<>(); + + // 分批扫本次同步相关流水(不限行数):只要 getInfo 无店 + 房型/产品匹配缺口 + long lastId = 0L; + while (true) { + List batch = baseMapper.selectList(Wrappers + .lambdaQuery() + .eq(SmHotelDownstreamPushLogEntity::getSyncLogId, syncLogId) + .gt(SmHotelDownstreamPushLogEntity::getId, lastId) + .orderByAsc(SmHotelDownstreamPushLogEntity::getId) + .last("limit 500")); + if (CollUtil.isEmpty(batch)) { + break; + } + Set needHotels = new HashSet<>(); + for (SmHotelDownstreamPushLogEntity row : batch) { + lastId = row.getId() == null ? lastId : row.getId(); + if (row.getHotelId() != null && !hotelMap.containsKey(row.getHotelId())) { + needHotels.add(row.getHotelId()); + } + } + hotelMap.putAll(loadHotels(needHotels)); + for (SmHotelDownstreamPushLogEntity row : batch) { + appendMatchOrGetInfoGaps(out, dedupe, syncLogId, partner, row, hotelMap.get(row.getHotelId())); + } + } + + // 同步日志 failDetail 里 getInfo 失败行(可售推抛错会进这里) + appendGetInfoFailDetailGaps(out, dedupe, syncLogId, partner, syncLog.getFailDetail(), hotelMap); + return out; + } + + private Map loadHotels(Set hotelIds) { + Map map = new HashMap<>(); + if (CollUtil.isEmpty(hotelIds)) { + return map; + } + List ids = new ArrayList<>(hotelIds); + int batch = 500; + for (int i = 0; i < ids.size(); i += batch) { + List slice = ids.subList(i, Math.min(i + batch, ids.size())); + List rows = hotelMapper.selectList(Wrappers.lambdaQuery() + .select(SmHotelEntity::getId, SmHotelEntity::getHotelCode, SmHotelEntity::getHotelName, + SmHotelEntity::getSHotelId) + .in(SmHotelEntity::getId, slice)); + if (CollUtil.isNotEmpty(rows)) { + for (SmHotelEntity h : rows) { + map.put(h.getId(), h); + } + } + } + return map; + } + + /** 仅两类:getInfo 无店/失败;房型或产品编码匹配不上 */ + private void appendMatchOrGetInfoGaps(List out, Set dedupe, Long syncLogId, + String partner, SmHotelDownstreamPushLogEntity row, SmHotelEntity hotel) { + String body = row.getResponseBody(); + boolean matchedGap = false; + if (StrUtil.isNotBlank(body) && JSONUtil.isTypeJSON(body)) { + try { + JSONObject json = JSONUtil.parseObj(body); + JSONArray rooms = json.getJSONArray("unmatchedRooms"); + if (rooms != null) { + for (int i = 0; i < rooms.size(); i++) { + String detail = StrUtil.trim(rooms.getStr(i)); + if (StrUtil.isBlank(detail)) { + continue; + } + addGap(out, dedupe, syncLogId, partner, hotel, row, "房型未匹配", detail, + "核对本地房型编码/名称与圣诺 getInfo.rooms"); + matchedGap = true; + } + } + JSONArray plans = json.getJSONArray("planErrors"); + if (plans != null) { + for (int i = 0; i < plans.size(); i++) { + String detail = StrUtil.trim(plans.getStr(i)); + if (StrUtil.isBlank(detail)) { + continue; + } + addGap(out, dedupe, syncLogId, partner, hotel, row, "产品未匹配", detail, + "核对产品编码及所属房型是否能在圣诺侧对上"); + matchedGap = true; + } + } + } + catch (Exception ignored) { + // ignore + } + } + if (matchedGap) { + return; + } + String msg = StrUtil.blankToDefault(row.getErrorMsg(), "") + " " + + StrUtil.blankToDefault(row.getResponseBody(), ""); + if (isGetInfoMiss(msg)) { + addGap(out, dedupe, syncLogId, partner, hotel, row, "圣诺getInfo无店", + StrUtil.blankToDefault(row.getErrorMsg(), "getInfo 失败或无酒店"), + "核对 eid/sHotelId,或在圣诺侧建档后再推"); + } + } + + private void appendGetInfoFailDetailGaps(List out, Set dedupe, Long syncLogId, + String partner, String failDetail, Map hotelMap) { + if (StrUtil.isBlank(failDetail)) { + return; + } + Map byCode = new HashMap<>(); + for (SmHotelEntity h : hotelMap.values()) { + if (h != null && StrUtil.isNotBlank(h.getHotelCode())) { + byCode.put(h.getHotelCode(), h); + } + } + for (String line : StrUtil.splitTrim(failDetail, '\n')) { + if (StrUtil.isBlank(line) || !isGetInfoMiss(line)) { + continue; + } + List parts = StrUtil.splitTrim(line, '|'); + String code = parts.isEmpty() ? "" : parts.get(0); + String name = parts.size() > 1 ? parts.get(1) : ""; + String reason = parts.size() > 2 ? StrUtil.join("|", parts.subList(2, parts.size())) : line; + SmHotelEntity h = byCode.get(code); + if (h == null && StrUtil.isNotBlank(code)) { + h = hotelMapper.selectOne(Wrappers.lambdaQuery() + .select(SmHotelEntity::getId, SmHotelEntity::getHotelCode, SmHotelEntity::getHotelName, + SmHotelEntity::getSHotelId) + .eq(SmHotelEntity::getHotelCode, code) + .last("limit 1")); + } + SmHotelDownstreamGapExcel row = new SmHotelDownstreamGapExcel(); + row.setSyncLogId(String.valueOf(syncLogId)); + row.setPartnerCode(partner); + if (h != null) { + row.setHotelId(h.getId() == null ? null : String.valueOf(h.getId())); + row.setHotelCode(h.getHotelCode()); + row.setHotelName(h.getHotelName()); + row.setEid(h.getSHotelId()); + } + else { + row.setHotelCode(code); + row.setHotelName(name); + } + row.setIssueType("圣诺getInfo无店"); + row.setIssueDetail(StrUtil.maxLength(reason, 500)); + row.setSuggest("核对 eid/sHotelId,或在圣诺侧建档后再推"); + String key = StrUtil.blankToDefault(row.getHotelId(), row.getHotelCode()) + "|" + row.getIssueType() + + "|" + row.getIssueDetail(); + if (dedupe.add(key)) { + out.add(row); + } + } + } + + private static boolean isGetInfoMiss(String text) { + if (StrUtil.isBlank(text)) { + return false; + } + String t = text.toLowerCase(Locale.ROOT); + return t.contains("getinfo") || text.contains("未返回酒店") || text.contains("圣诺侧无此酒店") + || text.contains("getInfo 无 data") || text.contains("getInfo失败") || text.contains("getInfo 失败"); + } + + private void addGap(List out, Set dedupe, Long syncLogId, String partner, + SmHotelEntity hotel, SmHotelDownstreamPushLogEntity logRow, String issueType, String issueDetail, + String suggest) { + SmHotelDownstreamGapExcel row = new SmHotelDownstreamGapExcel(); + row.setSyncLogId(String.valueOf(syncLogId)); + row.setPartnerCode(StrUtil.blankToDefault(partner, + logRow == null ? null : logRow.getPartnerCode())); + if (hotel != null) { + row.setHotelId(hotel.getId() == null ? null : String.valueOf(hotel.getId())); + row.setHotelCode(hotel.getHotelCode()); + row.setHotelName(hotel.getHotelName()); + row.setEid(hotel.getSHotelId()); + } + else if (logRow != null) { + row.setHotelId(logRow.getHotelId() == null ? null : String.valueOf(logRow.getHotelId())); + row.setEid(logRow.getExtHotelId()); + } + row.setIssueType(issueType); + row.setIssueDetail(StrUtil.maxLength(issueDetail, 500)); + row.setSuggest(suggest); + String key = StrUtil.blankToDefault(row.getHotelId(), row.getHotelCode()) + "|" + issueType + "|" + + row.getIssueDetail(); + if (dedupe.add(key)) { + out.add(row); + } + } + /** 支持 yyyy-MM-dd 或 yyyy-MM-dd HH:mm:ss;日期无时分秒时 begin=00:00:00、end=23:59:59 */ private static LocalDateTime parseDateTime(String raw, boolean beginOfDay) { if (StrUtil.isBlank(raw)) { diff --git a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java index 7e59cda4..d5ad3d5a 100644 --- a/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java +++ b/pjcloud-supermarket/pjcloud-supermarket-biz/src/main/java/com/bsd/pjcloud/supermarket/service/impl/SmHotelSyncServiceImpl.java @@ -9,6 +9,7 @@ import com.bsd.pjcloud.supermarket.config.ShangyouHotelProperties; import com.bsd.pjcloud.supermarket.constant.SmHotelDataSource; import com.bsd.pjcloud.supermarket.constant.SmHotelDownstreamProtocol; import com.bsd.pjcloud.supermarket.dto.SmHotelLocalDailyBatchDTO; +import com.bsd.pjcloud.supermarket.dto.SmHotelSyncFailExcel; import com.bsd.pjcloud.supermarket.entity.SmHotelEntity; import com.bsd.pjcloud.supermarket.entity.SmHotelSyncLogEntity; import com.bsd.pjcloud.supermarket.integration.downstream.DownstreamPushLogContext; @@ -66,10 +67,15 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { private static final String STATUS_RUNNING = "RUNNING"; + private static final String STATUS_PAUSED = "PAUSED"; + private static final String STATUS_SUCCESS = "SUCCESS"; private static final String STATUS_FAILED = "FAILED"; + /** 暂停续跑检查点:idx:已处理条数 或 lid:详情拉取 lastId */ + private static final String RESUME_CKPT_KEY = "sm:hotel:sync:resume:"; + /** 同步日志超过该时间无 progress 心跳,视为死任务 */ private static final int STALE_RUNNING_MINUTES = 30; @@ -248,7 +254,8 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { final int cap = maxHotels; final String ch = channel; final Long dept = deptId; - EXECUTOR.execute(() -> TenantBroker.runAs(tid, id -> runDetailPull(logId, cap, ch, dept))); + EXECUTOR.execute(() -> TenantBroker.runAs(tid, + id -> runDetailPull(logId, cap, ch, dept, 0L, 0, 0, 0, new ArrayList<>()))); return R.ok(logEntity, "已提交全量详情拉取"); } @@ -303,7 +310,9 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { final Long logId = logEntity.getId(); final String partnerFinal = partner; final List ids = List.copyOf(hotelIds); - EXECUTOR.execute(() -> TenantBroker.runAs(tid, id -> runDownstreamScopePush(logId, partnerFinal, ids))); + final int total = hotelIds.size(); + EXECUTOR.execute(() -> TenantBroker.runAs(tid, + id -> runDownstreamScopePush(logId, partnerFinal, ids, 0, 0, 0, 0, new ArrayList<>(), total))); return R.ok(logEntity, "已提交下游全量推送,共 " + hotelIds.size() + " 家"); } @@ -319,20 +328,64 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { if (STATUS_RUNNING.equals(src.getStatus())) { return R.failed("原任务仍在执行,请结束后再重试失败项"); } + if (STATUS_PAUSED.equals(src.getStatus())) { + return R.failed("任务已暂停,请先「继续」或「终止」后再重试失败项"); + } String syncType = src.getSyncType(); - if (!TYPE_DETAIL.equals(syncType) && !TYPE_ROLL.equals(syncType)) { - return R.failed("仅支持「全量详情」或「滚价」的失败重试"); + if (!TYPE_DETAIL.equals(syncType) && !TYPE_ROLL.equals(syncType) + && !TYPE_DOWNSTREAM_PUSH.equals(syncType)) { + return R.failed("仅支持「全量详情」「滚价」「下游全量推」的失败重试"); } List codes = parseFailHotelCodes(src.getFailDetail()); if (CollUtil.isEmpty(codes)) { - return R.failed("该日志无失败明细可重试(请确认 failDetail 有 hotelCode|名称|原因)"); + return R.failed("该日志无失败明细可重试(请确认 failDetail 有 hotelCode|名称|原因;可先导出失败明细核对)"); } abortStaleRunningSyncLogs(); Long running = syncLogMapper.selectCount(Wrappers.lambdaQuery() .eq(SmHotelSyncLogEntity::getSyncType, syncType) .eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING)); if (running != null && running > 0) { - return R.failed("已有同类型任务进行中,请稍后再试(若卡住请先终止)"); + return R.failed("已有同类型任务进行中,请稍后再试(若卡住请先暂停/终止)"); + } + + // 下游全量推:按失败酒店整店再推,不扫全量范围 + if (TYPE_DOWNSTREAM_PUSH.equals(syncType)) { + String partner = SmHotelDataSource.normalize(src.getDataSource()).toLowerCase(); + if (StrUtil.isBlank(partner)) { + return R.failed("来源日志缺少下游伙伴编码"); + } + if (!channelConfigService.listEnabledDownstreamPartners().contains(partner)) { + return R.failed("下游伙伴「" + partner + "」未启用或不存在"); + } + List hotelIds = resolveHotelIdsFromFailCodes(codes); + int beforeFilter = hotelIds.size(); + hotelIds = filterShengnuoHotelsRequireEid(partner, hotelIds); + if (CollUtil.isEmpty(hotelIds)) { + return R.failed(beforeFilter > 0 + ? "失败明细对应酒店均无外部酒店ID(eid)或无法匹配本地,无法重推" + : "失败明细未能匹配到本地酒店"); + } + SmHotelSyncLogEntity logEntity = newLog(TYPE_DOWNSTREAM_PUSH, "RETRY"); + logEntity.setDataSource(partner); + logEntity.setStage("QUEUED"); + logEntity.setStageMsg("重试失败推送:来源日志 " + sourceLogId + ",共 " + hotelIds.size() + " 家 →「" + + partner + "」"); + logEntity.setProgressPct(1); + logEntity.setLocalCount((long) hotelIds.size()); + syncLogMapper.insert(logEntity); + + Long tenantId = TenantContextHolder.getTenantId(); + if (tenantId == null) { + tenantId = props().getCallbackTenantId() == null ? 1L : props().getCallbackTenantId(); + } + final Long tid = tenantId; + final Long logId = logEntity.getId(); + final String partnerFinal = partner; + final List ids = List.copyOf(hotelIds); + final int total = hotelIds.size(); + EXECUTOR.execute(() -> TenantBroker.runAs(tid, + id -> runDownstreamScopePush(logId, partnerFinal, ids, 0, 0, 0, 0, new ArrayList<>(), total))); + return R.ok(logEntity, "已提交失败重推,共 " + hotelIds.size() + " 家(完善数据后可对本次新日志再重试)"); } SmHotelSyncLogEntity logEntity = newLog(syncType, "RETRY"); @@ -374,15 +427,163 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { if (e == null) { return R.failed("同步记录不存在"); } - if (!STATUS_RUNNING.equals(e.getStatus())) { - return R.failed("仅能终止「进行中」的任务,当前状态=" + e.getStatus()); + if (!STATUS_RUNNING.equals(e.getStatus()) && !STATUS_PAUSED.equals(e.getStatus())) { + return R.failed("仅能终止「进行中/已暂停」的任务,当前状态=" + e.getStatus()); } - failLog(id, "人工终止死任务(线程已中断或进程重启后残留)", - e.getCreateTime() == null ? 0L - : Math.max(0L, java.time.Duration.between(e.getCreateTime(), LocalDateTime.now()).toMillis())); - log.warn("|酒店同步|人工终止|logId={}|type={}|stage={}|pct={}|", id, e.getSyncType(), e.getStage(), - e.getProgressPct()); - return R.ok(Boolean.TRUE, "已终止并标记失败,可重新发起同步"); + String was = e.getStatus(); + long cost = e.getCreateTime() == null ? 0L + : Math.max(0L, java.time.Duration.between(e.getCreateTime(), LocalDateTime.now()).toMillis()); + // 不调用 failLog:避免把 stageMsg 盖成「执行失败」且冲掉已落库 failDetail。 + // RUNNING 时由工作线程感知后 flush;PAUSED 无线程,直接收尾。 + e.setStatus(STATUS_FAILED); + e.setStage("FAILED"); + e.setStageMsg(StrUtil.maxLength( + "已终止(成功 " + nullToZero(e.getSuccessCount()) + " / 失败 " + nullToZero(e.getFailCount()) + + " / 跳过 " + nullToZero(e.getSkipCount()) + "),失败明细仍可查看或导出", + 500)); + e.setErrorMsg(StrUtil.maxLength( + "人工终止(原状态=" + was + ");已处理进度保留,失败明细见 failDetail", + 1000)); + e.setCostMs(cost); + e.setFinishTime(LocalDateTime.now()); + e.setUpdateTime(LocalDateTime.now()); + syncLogMapper.updateById(e); + clearResumeCkpt(id); + log.warn("|酒店同步|人工终止|logId={}|type={}|was={}|pct={}|fail={}|", id, e.getSyncType(), was, + e.getProgressPct(), e.getFailCount()); + return R.ok(Boolean.TRUE, "已终止;失败明细仍可查看/导出"); + } + + @Override + public R pauseSyncLog(Long id) { + if (id == null) { + return R.failed("日志ID不能为空"); + } + SmHotelSyncLogEntity e = syncLogMapper.selectById(id); + if (e == null) { + return R.failed("同步记录不存在"); + } + if (!STATUS_RUNNING.equals(e.getStatus())) { + return R.failed("仅能暂停「进行中」的任务,当前状态=" + e.getStatus()); + } + String type = e.getSyncType(); + if (!TYPE_DOWNSTREAM_PUSH.equals(type) && !TYPE_DETAIL.equals(type)) { + return R.failed("当前仅支持暂停「下游全量推」或「全量详情」;其它类型请用终止"); + } + e.setStatus(STATUS_PAUSED); + e.setStage("PAUSE_REQ"); + e.setStageMsg(StrUtil.maxLength( + "暂停请求已提交,当前批次结束后生效(发布代码/插队其它任务前请等状态变为「已暂停」)", 500)); + e.setUpdateTime(LocalDateTime.now()); + syncLogMapper.updateById(e); + log.info("|酒店同步|请求暂停|logId={}|type={}|pct={}|", id, type, e.getProgressPct()); + return R.ok(Boolean.TRUE, "已请求暂停,批次结束后保留进度,可点「继续」"); + } + + @Override + public R resumeSyncLog(Long id) { + if (id == null) { + return R.failed("日志ID不能为空"); + } + SmHotelSyncLogEntity e = syncLogMapper.selectById(id); + if (e == null) { + return R.failed("同步记录不存在"); + } + if (!STATUS_PAUSED.equals(e.getStatus())) { + return R.failed("仅能继续「已暂停」的任务,当前状态=" + e.getStatus()); + } + String type = e.getSyncType(); + if (!TYPE_DOWNSTREAM_PUSH.equals(type) && !TYPE_DETAIL.equals(type)) { + return R.failed("该类型不支持继续,请重新发起任务"); + } + abortStaleRunningSyncLogs(); + Long running = syncLogMapper.selectCount(Wrappers.lambdaQuery() + .eq(SmHotelSyncLogEntity::getSyncType, type) + .eq(SmHotelSyncLogEntity::getStatus, STATUS_RUNNING)); + if (running != null && running > 0) { + return R.failed("已有同类型任务进行中,请先等其结束或暂停/终止后再继续"); + } + + e.setStatus(STATUS_RUNNING); + e.setStage("RESUME"); + e.setStageMsg("继续执行中…"); + e.setFinishTime(null); + e.setUpdateTime(LocalDateTime.now()); + syncLogMapper.updateById(e); + + Long tenantId = TenantContextHolder.getTenantId(); + if (tenantId == null) { + tenantId = props().getCallbackTenantId() == null ? 1L : props().getCallbackTenantId(); + } + final Long tid = tenantId; + final Long logId = e.getId(); + if (TYPE_DOWNSTREAM_PUSH.equals(type)) { + String partner = SmHotelDataSource.normalize(e.getDataSource()).toLowerCase(); + List hotelIds = channelDeptService.listHotelIdsForDownstreamPartner(partner); + hotelIds = filterShengnuoHotelsRequireEid(partner, hotelIds); + final int baseOk = nullToZero(e.getSuccessCount()); + final int baseFail = nullToZero(e.getFailCount()); + final int baseSkip = nullToZero(e.getSkipCount()); + // Redis 检查点优先;发版丢 Redis 时用库内成功+失败+跳过条数兜底,避免从头重推 + int resumeIdx = Math.max(readResumeIdx(logId), baseOk + baseFail + baseSkip); + if (resumeIdx > 0 && CollUtil.isNotEmpty(hotelIds) && resumeIdx < hotelIds.size()) { + hotelIds = hotelIds.subList(resumeIdx, hotelIds.size()); + } + else if (resumeIdx > 0 && CollUtil.isNotEmpty(hotelIds) && resumeIdx >= hotelIds.size()) { + return R.failed("检查点已覆盖全部酒店,无需继续;可将本日志视为完成或重新发起"); + } + final String partnerFinal = partner; + final List ids = List.copyOf(hotelIds == null ? List.of() : hotelIds); + final int baseIdx = resumeIdx; + final List baseFails = parseFailLines(e.getFailDetail()); + final long totalHint = e.getLocalCount() == null ? (baseIdx + ids.size()) : e.getLocalCount(); + EXECUTOR.execute(() -> TenantBroker.runAs(tid, + x -> runDownstreamScopePush(logId, partnerFinal, ids, baseIdx, baseOk, baseFail, baseSkip, + baseFails, (int) totalHint))); + return R.ok(e, "已继续下游全量推,从第 " + (baseIdx + 1) + " 家接着跑"); + } + + // DETAIL_PULL + String channel = SmHotelDataSource.normalize(e.getDataSource()).toLowerCase(); + R deptR = channelDeptService.requireEnabledDeptId(channel); + Long deptId = deptR != null && deptR.isOk() ? deptR.getData() : null; + int maxHotels = e.getLocalCount() == null ? 0 : e.getLocalCount().intValue(); + long resumeLastId = Math.max(0L, readResumeLastId(logId)); + final int baseOk = nullToZero(e.getSuccessCount()); + final int baseFail = nullToZero(e.getFailCount()); + final int baseSkip = nullToZero(e.getSkipCount()); + final List baseFails = parseFailLines(e.getFailDetail()); + EXECUTOR.execute(() -> TenantBroker.runAs(tid, + x -> runDetailPull(logId, maxHotels, channel, deptId, resumeLastId, baseOk, baseFail, baseSkip, + baseFails))); + return R.ok(e, "已继续全量详情拉取"); + } + + @Override + public List exportFailDetail(Long id) { + List out = new ArrayList<>(); + if (id == null) { + return out; + } + SmHotelSyncLogEntity e = syncLogMapper.selectById(id); + if (e == null || StrUtil.isBlank(e.getFailDetail())) { + return out; + } + for (String line : StrUtil.splitTrim(e.getFailDetail(), '\n')) { + if (StrUtil.isBlank(line) || line.startsWith("…")) { + continue; + } + String[] parts = line.split("\\|", 3); + SmHotelSyncFailExcel row = new SmHotelSyncFailExcel(); + row.setSyncLogId(id); + row.setSyncType(e.getSyncType()); + row.setDataSource(e.getDataSource()); + row.setHotelCode(parts.length > 0 ? parts[0] : ""); + row.setHotelName(parts.length > 1 ? parts[1] : ""); + row.setReason(parts.length > 2 ? parts[2] : line); + out.add(row); + } + return out; } @Override @@ -738,91 +939,193 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { } } - /** 按范围映射向指定下游逐家推送本地酒店数据(不拉上游)。 */ - private void runDownstreamScopePush(Long logId, String partnerCode, List hotelIds) { + /** + * 按范围映射向指定下游推送本地酒店(默认酒店级并行)。 + * @param hotelIds 本段待推列表(续跑时已去掉前面已处理) + * @param baseProcessed 续跑前已处理条数(用于进度分母与检查点) + * @param totalAll 全量目标家数(进度分母) + */ + private void runDownstreamScopePush(Long logId, String partnerCode, List hotelIds, int baseProcessed, + int baseOk, int baseFail, int baseSkip, List baseFailLines, int totalAll) { long t0 = System.currentTimeMillis(); HotelDownstreamProperties dsProps = channelConfigService.resolveDownstreamRuntime(partnerCode); int sampleMax = dsProps == null ? 20 : Math.max(0, dsProps.getPushLogSampleSuccessMax()); - DownstreamPushLogContext.begin(logId, sampleMax); + int concurrency = dsProps == null ? 6 : Math.max(1, dsProps.getResourcePushConcurrency()); + int total = totalAll > 0 ? totalAll : (baseProcessed + (hotelIds == null ? 0 : hotelIds.size())); + AtomicInteger ok = new AtomicInteger(Math.max(0, baseOk)); + AtomicInteger fail = new AtomicInteger(Math.max(0, baseFail)); + AtomicInteger skip = new AtomicInteger(Math.max(0, baseSkip)); + AtomicInteger processed = new AtomicInteger(Math.max(0, baseProcessed)); + List failLines = baseFailLines == null ? new ArrayList<>() : new ArrayList<>(baseFailLines); + final Object failLock = new Object(); + ExecutorService pushPool = null; try { - int total = hotelIds == null ? 0 : hotelIds.size(); - int ok = 0; - int fail = 0; - int skip = 0; - int processed = 0; - List failLines = new ArrayList<>(); - progress(logId, 5, "PUSH", "开始向下游「" + partnerCode + "」推送,共 " + total + " 家…"); - if (CollUtil.isNotEmpty(hotelIds)) { - for (Long hotelId : hotelIds) { - processed++; - SmHotelEntity h = hotelMapper.selectById(hotelId); - if (h == null) { - skip++; - continue; - } - try { - R> r = midService.pushToDownstream(hotelId, partnerCode); - if (r != null && r.getCode() == 0) { - // 预检跳过(圣诺无 eid;其它下游无产品等):算 skip。 - // 圣诺有 eid 后无产品/无日历走关房,计成功,不再跳过。 - if (isDownstreamPushSkipped(r.getData())) { - skip++; - } - else { - ok++; - } + progressSnapshot(logId, Math.min(95, 5 + (int) (processed.get() * 90L / Math.max(1, total))), "PUSH", + (baseProcessed > 0 ? "继续向下游「" : "开始向下游「") + partnerCode + "」推送,共 " + total + + " 家,已处理 " + processed.get() + ",并发=" + concurrency + "…", + ok.get(), fail.get(), skip.get(), total, failLines); + if (CollUtil.isEmpty(hotelIds)) { + // fall through to complete + } + else if (concurrency <= 1) { + DownstreamPushLogContext.begin(logId, sampleMax); + try { + for (Long hotelId : hotelIds) { + if (handleStopIfNeeded(logId, t0, total, ok, fail, skip, processed, failLines)) { + return; } - else { - fail++; - appendFail(failLines, h, r == null ? "返回为空" : StrUtil.blankToDefault(r.getMsg(), "业务失败")); - } - } - catch (Exception e) { - fail++; - appendFail(failLines, h, e.getMessage()); - log.warn("|下游全量推送|单店失败|partner={}|hotelId={}|{}", partnerCode, hotelId, e.getMessage()); - } - int pct = total <= 0 ? 50 : Math.min(95, 5 + (int) (processed * 90L / total)); - progress(logId, pct, "PUSH", - "推送中 " + processed + "/" + total + " 成功 " + ok + " 失败 " + fail + " 跳过 " + skip - + " 当前 " + StrUtil.blankToDefault(h.getHotelName(), h.getHotelCode())); - if (isAborted(logId)) { - failLog(logId, "任务已终止", System.currentTimeMillis() - t0); - return; + pushOneDownstreamHotel(logId, partnerCode, hotelId, total, ok, fail, skip, processed, + failLines, failLock); } } + finally { + DownstreamPushLogContext.end(); + } } + else { + pushPool = Executors.newFixedThreadPool(concurrency, r -> { + Thread t = new Thread(r, "sm-hotel-ds-push"); + t.setDaemon(true); + return t; + }); + int batchSize = Math.max(concurrency * 2, 20); + for (int i = 0; i < hotelIds.size(); i += batchSize) { + if (handleStopIfNeeded(logId, t0, total, ok, fail, skip, processed, failLines)) { + return; + } + List slice = hotelIds.subList(i, Math.min(i + batchSize, hotelIds.size())); + List> tasks = new ArrayList<>(slice.size()); + for (Long hotelId : slice) { + tasks.add(() -> { + DownstreamPushLogContext.begin(logId, sampleMax); + try { + pushOneDownstreamHotel(logId, partnerCode, hotelId, total, ok, fail, skip, + processed, failLines, failLock); + } + finally { + DownstreamPushLogContext.end(); + } + return null; + }); + } + List> futures = pushPool.invokeAll(tasks); + for (Future f : futures) { + try { + f.get(); + } + catch (Exception e) { + log.warn("|下游全量推送|批次异常|{}", e.getMessage()); + } + } + // 批次末落盘,终止/暂停时失败明细不丢 + progressSnapshot(logId, + total <= 0 ? 50 : Math.min(95, 5 + (int) (processed.get() * 90L / total)), "PUSH", + "推送中 " + processed.get() + "/" + total + " 成功 " + ok.get() + " 失败 " + fail.get() + + " 跳过 " + skip.get(), + ok.get(), fail.get(), skip.get(), total, failLines); + } + } + + if (handleStopIfNeeded(logId, t0, total, ok, fail, skip, processed, failLines)) { + return; + } + SmHotelSyncLogEntity done = syncLogMapper.selectById(logId); if (done == null) { return; } done.setStatus(STATUS_SUCCESS); done.setStage("DONE"); - done.setStageMsg("下游「" + partnerCode + "」全量推送完成:成功 " + ok + " / 失败 " + fail + " / 跳过 " + skip - + "(成功含可售推与关房推;圣诺跳过仅无 eid,无产品/无日历会关房;其它下游跳过含无产品等);" - + "明细见下游推送日志(同步日志ID=" + logId + ")"); + done.setStageMsg("下游「" + partnerCode + "」全量推送完成:成功 " + ok.get() + " / 失败 " + fail.get() + + " / 跳过 " + skip.get() + "(并发=" + concurrency + + ";成功含可售推与关房推;明细见下游推送日志 syncLogId=" + logId + ")"); done.setProgressPct(100); - done.setSuccessCount(ok); - done.setFailCount(fail); - done.setSkipCount(skip); + done.setSuccessCount(ok.get()); + done.setFailCount(fail.get()); + done.setSkipCount(skip.get()); done.setLocalCount((long) total); - applyFailDetail(done, failLines, fail); + applyFailDetail(done, failLines, fail.get()); done.setCostMs(System.currentTimeMillis() - t0); done.setFinishTime(LocalDateTime.now()); done.setUpdateTime(LocalDateTime.now()); syncLogMapper.updateById(done); - log.info("|下游全量推送|完成|logId={}|partner={}|ok={}|fail={}|skip={}|costMs={}|", logId, partnerCode, ok, - fail, skip, done.getCostMs()); + clearResumeCkpt(logId); + log.info("|下游全量推送|完成|logId={}|partner={}|ok={}|fail={}|skip={}|concurrency={}|costMs={}|", logId, + partnerCode, ok.get(), fail.get(), skip.get(), concurrency, done.getCostMs()); } catch (Throwable e) { log.error("|下游全量推送|失败|logId={}|partner={}|", logId, partnerCode, e); - failLog(logId, e.getMessage(), System.currentTimeMillis() - t0); + flushThenFail(logId, e.getMessage(), System.currentTimeMillis() - t0, ok, fail, skip, processed, total, + failLines); } finally { - DownstreamPushLogContext.end(); + if (pushPool != null) { + pushPool.shutdown(); + try { + if (!pushPool.awaitTermination(60, TimeUnit.SECONDS)) { + pushPool.shutdownNow(); + } + } + catch (InterruptedException ie) { + pushPool.shutdownNow(); + Thread.currentThread().interrupt(); + } + } } } + private void pushOneDownstreamHotel(Long logId, String partnerCode, Long hotelId, int total, + AtomicInteger ok, AtomicInteger fail, AtomicInteger skip, AtomicInteger processed, + List failLines, Object failLock) { + SmHotelEntity h = hotelMapper.selectById(hotelId); + if (h == null) { + skip.incrementAndGet(); + int doneCnt = processed.incrementAndGet(); + progressDownstream(logId, total, doneCnt, ok.get(), fail.get(), skip.get(), null, failLines); + return; + } + try { + R> r = midService.pushToDownstream(hotelId, partnerCode); + if (r != null && r.getCode() == 0) { + // 预检跳过(圣诺无 eid;其它下游无产品等):算 skip。 + // 圣诺有 eid 后无产品/无日历走关房,计成功,不再跳过。 + if (isDownstreamPushSkipped(r.getData())) { + skip.incrementAndGet(); + } + else { + ok.incrementAndGet(); + } + } + else { + fail.incrementAndGet(); + synchronized (failLock) { + appendFail(failLines, h, r == null ? "返回为空" : StrUtil.blankToDefault(r.getMsg(), "业务失败")); + } + } + } + catch (Exception e) { + fail.incrementAndGet(); + synchronized (failLock) { + appendFail(failLines, h, e.getMessage()); + } + log.warn("|下游全量推送|单店失败|partner={}|hotelId={}|{}", partnerCode, hotelId, e.getMessage()); + } + int doneCnt = processed.incrementAndGet(); + progressDownstream(logId, total, doneCnt, ok.get(), fail.get(), skip.get(), h, failLines); + } + + private void progressDownstream(Long logId, int total, int processed, int ok, int fail, int skip, + SmHotelEntity h, List failLines) { + if (processed % 20 != 0 && processed < total) { + return; + } + int pct = total <= 0 ? 50 : Math.min(95, 5 + (int) (processed * 90L / total)); + String cur = h == null ? "" : (" 当前 " + StrUtil.blankToDefault(h.getHotelName(), h.getHotelCode())); + progressSnapshot(logId, pct, "PUSH", + "推送中 " + processed + "/" + total + " 成功 " + ok + " 失败 " + fail + " 跳过 " + skip + cur, ok, + fail, skip, total, failLines); + } + @Override public List filterDownstreamPushHotelIds(String partnerCode, List hotelIds) { return filterShengnuoHotelsRequireEid(partnerCode, hotelIds); @@ -877,23 +1180,220 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { return "true".equalsIgnoreCase(String.valueOf(skipped)); } - /** 人工终止后状态已非 RUNNING,后续循环应退出 */ + /** 人工终止/暂停后状态已非 RUNNING,后续循环应退出 */ private boolean isAborted(Long logId) { SmHotelSyncLogEntity cur = syncLogMapper.selectById(logId); return cur == null || !STATUS_RUNNING.equals(cur.getStatus()); } + /** @return true 表示已暂停或已终止并落盘,调用方应直接 return */ + private boolean handleStopIfNeeded(Long logId, long t0, int total, AtomicInteger ok, AtomicInteger fail, + AtomicInteger skip, AtomicInteger processed, List failLines) { + return handleStopIfNeeded(logId, t0, total, ok, fail, skip, processed, failLines, null); + } + + private boolean handleStopIfNeeded(Long logId, long t0, int total, AtomicInteger ok, AtomicInteger fail, + AtomicInteger skip, AtomicInteger processed, List failLines, Long resumeLastId) { + SmHotelSyncLogEntity cur = syncLogMapper.selectById(logId); + if (cur == null) { + return true; + } + if (STATUS_RUNNING.equals(cur.getStatus())) { + return false; + } + long cost = System.currentTimeMillis() - t0; + cur.setSuccessCount(ok.get()); + cur.setFailCount(fail.get()); + cur.setSkipCount(skip.get()); + if (total > 0) { + cur.setLocalCount((long) total); + } + applyFailDetailOnly(cur, failLines, fail.get()); + cur.setCostMs(cost); + cur.setUpdateTime(LocalDateTime.now()); + int doneCnt = processed.get(); + Integer prevPct = cur.getProgressPct(); + int pct = total <= 0 ? (prevPct == null ? 0 : prevPct) + : Math.min(95, 5 + (int) (doneCnt * 90L / total)); + cur.setProgressPct(pct); + + if (STATUS_PAUSED.equals(cur.getStatus()) || "PAUSE_REQ".equals(cur.getStage())) { + cur.setStatus(STATUS_PAUSED); + cur.setStage("PAUSED"); + cur.setStageMsg(StrUtil.maxLength( + "已暂停 " + doneCnt + "/" + total + " 成功 " + ok.get() + " 失败 " + fail.get() + " 跳过 " + + skip.get() + "。点「继续」接着跑;发版前请保持暂停以免丢进度", + 500)); + cur.setFinishTime(null); + syncLogMapper.updateById(cur); + saveResumeIdx(logId, doneCnt); + if (resumeLastId != null && resumeLastId > 0) { + saveResumeLastId(logId, resumeLastId); + } + log.info("|酒店同步|已暂停落盘|logId={}|processed={}/{}|ok={}|fail={}|", logId, doneCnt, total, ok.get(), + fail.get()); + return true; + } + + // 终止或其它非 RUNNING:保留明细 + cur.setStatus(STATUS_FAILED); + cur.setStage("FAILED"); + cur.setStageMsg(StrUtil.maxLength( + "已终止(成功 " + ok.get() + " / 失败 " + fail.get() + " / 跳过 " + skip.get() + "),失败明细仍可查看或导出", + 500)); + if (StrUtil.isBlank(cur.getErrorMsg()) || cur.getErrorMsg().contains("人工终止")) { + cur.setErrorMsg(StrUtil.maxLength( + "任务已终止;已处理 " + doneCnt + "/" + total + ",失败明细见 failDetail", 1000)); + } + cur.setFinishTime(LocalDateTime.now()); + syncLogMapper.updateById(cur); + clearResumeCkpt(logId); + log.info("|酒店同步|终止落盘|logId={}|processed={}/{}|fail={}|", logId, doneCnt, total, fail.get()); + return true; + } + + /** 运行中心跳:进度 + 计数 + 失败明细(避免终止后明细丢失) */ + private void progressSnapshot(Long logId, int pct, String stage, String msg, int ok, int fail, int skip, + int total, List failLines) { + SmHotelSyncLogEntity e = syncLogMapper.selectById(logId); + if (e == null || !STATUS_RUNNING.equals(e.getStatus())) { + return; + } + e.setProgressPct(pct); + e.setStage(stage); + e.setStageMsg(StrUtil.maxLength(msg, 500)); + e.setSuccessCount(ok); + e.setFailCount(fail); + e.setSkipCount(skip); + if (total > 0) { + e.setLocalCount((long) total); + } + applyFailDetailOnly(e, failLines, fail); + e.setUpdateTime(LocalDateTime.now()); + syncLogMapper.updateById(e); + saveResumeIdx(logId, ok + fail + skip); + } + + private void flushThenFail(Long logId, String err, long costMs, AtomicInteger ok, AtomicInteger fail, + AtomicInteger skip, AtomicInteger processed, int total, List failLines) { + SmHotelSyncLogEntity e = syncLogMapper.selectById(logId); + if (e == null) { + return; + } + e.setStatus(STATUS_FAILED); + e.setStage("FAILED"); + e.setStageMsg(StrUtil.maxLength("执行失败(已处理 " + processed.get() + (total > 0 ? "/" + total : "") + + "),失败明细仍可查看或导出", 500)); + e.setErrorMsg(StrUtil.maxLength(StrUtil.blankToDefault(err, "未知错误"), 1000)); + e.setSuccessCount(ok.get()); + e.setFailCount(fail.get()); + e.setSkipCount(skip.get()); + applyFailDetailOnly(e, failLines, fail.get()); + e.setCostMs(costMs); + e.setFinishTime(LocalDateTime.now()); + e.setUpdateTime(LocalDateTime.now()); + syncLogMapper.updateById(e); + clearResumeCkpt(logId); + } + + private static int nullToZero(Integer v) { + return v == null ? 0 : v; + } + + private static List parseFailLines(String failDetail) { + List lines = new ArrayList<>(); + if (StrUtil.isBlank(failDetail)) { + return lines; + } + for (String line : StrUtil.splitTrim(failDetail, '\n')) { + if (StrUtil.isBlank(line) || line.startsWith("…")) { + continue; + } + lines.add(line); + if (lines.size() >= 2000) { + break; + } + } + return lines; + } + + private void saveResumeIdx(Long logId, int idx) { + if (logId == null) { + return; + } + long lid = readResumeLastId(logId); + String v = "idx:" + Math.max(0, idx) + (lid > 0 ? "&lid:" + lid : ""); + redisTemplate.opsForValue().set(RESUME_CKPT_KEY + logId, v, 14, TimeUnit.DAYS); + } + + private void saveResumeLastId(Long logId, long lastId) { + if (logId == null || lastId <= 0) { + return; + } + int idx = readResumeIdx(logId); + redisTemplate.opsForValue().set(RESUME_CKPT_KEY + logId, "idx:" + idx + "&lid:" + lastId, 14, + TimeUnit.DAYS); + } + + private int readResumeIdx(Long logId) { + String v = redisTemplate.opsForValue().get(RESUME_CKPT_KEY + logId); + if (StrUtil.isBlank(v)) { + return 0; + } + try { + for (String part : v.split("&")) { + if (part.startsWith("idx:")) { + return Integer.parseInt(part.substring(4).trim()); + } + } + } + catch (Exception ignored) { + // ignore + } + return 0; + } + + private long readResumeLastId(Long logId) { + String v = redisTemplate.opsForValue().get(RESUME_CKPT_KEY + logId); + if (StrUtil.isBlank(v)) { + return 0L; + } + try { + for (String part : v.split("&")) { + if (part.startsWith("lid:")) { + return Long.parseLong(part.substring(4).trim()); + } + } + } + catch (Exception ignored) { + // ignore + } + return 0L; + } + + private void clearResumeCkpt(Long logId) { + if (logId != null) { + redisTemplate.delete(RESUME_CKPT_KEY + logId); + } + } + /** * 全量详情:酒店级并行 + 可缩房价窗,尽量吃满上游 price-api-qps。 * 本地及其它渠道酒店一律不进此任务。 */ - private void runDetailPull(Long logId, int maxHotels, String channel, Long deptId) { + private void runDetailPull(Long logId, int maxHotels, String channel, Long deptId, long resumeLastId, + int baseOk, int baseFail, int baseSkip, List baseFailLines) { long t0 = System.currentTimeMillis(); String ch = SmHotelDataSource.normalize(channel).toLowerCase(); ExecutorService pullPool = null; + AtomicInteger ok = new AtomicInteger(Math.max(0, baseOk)); + AtomicInteger fail = new AtomicInteger(Math.max(0, baseFail)); + AtomicInteger skip = new AtomicInteger(Math.max(0, baseSkip)); + AtomicInteger processed = new AtomicInteger(Math.max(0, baseOk + baseFail + baseSkip)); + List failLines = baseFailLines == null ? new ArrayList<>() : new ArrayList<>(baseFailLines); try { // 将该渠道酒店统一归属到维护的部门(便于支付商户/数据权限) - if (deptId != null && StrUtil.isNotBlank(ch)) { + if (deptId != null && StrUtil.isNotBlank(ch) && resumeLastId <= 0) { hotelMapper.update(null, Wrappers.lambdaUpdate() .set(SmHotelEntity::getDeptId, deptId) .apply(channelSheetSql(ch), ch) @@ -913,12 +1413,7 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { Integer priceDaysOverride = props().getDetailPullPriceDays() > 0 ? props().getDetailPullPriceDays() : null; int effectivePriceDays = priceDaysOverride != null ? priceDaysOverride : props().getPriceDays(); - long lastId = 0L; - AtomicInteger ok = new AtomicInteger(); - AtomicInteger fail = new AtomicInteger(); - AtomicInteger skip = new AtomicInteger(); - AtomicInteger processed = new AtomicInteger(); - List failLines = new ArrayList<>(); + long lastId = Math.max(0L, resumeLastId); final Object failLock = new Object(); Long tenantId = TenantContextHolder.getTenantId(); @@ -928,17 +1423,18 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { final Long tid = tenantId; pullPool = Executors.newFixedThreadPool(concurrency, r -> { - Thread t = new Thread(r, "sm-hotel-detail-pull"); - t.setDaemon(true); - return t; + Thread th = new Thread(r, "sm-hotel-detail-pull"); + th.setDaemon(true); + return th; }); - progress(logId, 5, "PULL", "开始拉取详情:目标约 " + target + " 家「" + ch + "」并发=" + concurrency - + " price-days=" + effectivePriceDays + "…"); + progressSnapshot(logId, 5, "PULL", + (resumeLastId > 0 ? "继续" : "开始") + "拉取详情:目标约 " + target + " 家「" + ch + "」并发=" + + concurrency + " price-days=" + effectivePriceDays + "…", + ok.get(), fail.get(), skip.get(), (int) target, failLines); while (true) { - if (isAborted(logId)) { - failLog(logId, "任务已终止", System.currentTimeMillis() - t0); + if (handleStopIfNeeded(logId, t0, (int) target, ok, fail, skip, processed, failLines, lastId)) { return; } List rows = hotelMapper.selectList(Wrappers.lambdaQuery() @@ -992,13 +1488,13 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { } } } - catch (Exception e) { + catch (Exception ex) { fail.incrementAndGet(); synchronized (failLock) { - appendFail(failLines, h, e.getMessage()); + appendFail(failLines, h, ex.getMessage()); } log.warn("|全量详情|单店失败|hotelId={}|code={}|{}", h.getId(), h.getHotelCode(), - e.getMessage()); + ex.getMessage()); } finally { if (maxHotels <= 0) { @@ -1016,9 +1512,10 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { if (doneCnt % 20 == 0 || doneCnt >= target) { int pct = target <= 0 ? 50 : Math.min(95, 5 + (int) (doneCnt * 90L / target)); - progress(logId, pct, "PULL", + progressSnapshot(logId, pct, "PULL", "拉取中 " + doneCnt + "/" + target + " 成功 " + ok.get() - + " 失败 " + fail.get() + " 并发=" + concurrency); + + " 失败 " + fail.get() + " 并发=" + concurrency, + ok.get(), fail.get(), skip.get(), (int) target, failLines); } } }); @@ -1031,11 +1528,12 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { try { f.get(); } - catch (Exception e) { - log.warn("|全量详情|批次任务异常|{}", e.getMessage()); + catch (Exception ex) { + log.warn("|全量详情|批次任务异常|{}", ex.getMessage()); } } } + saveResumeLastId(logId, lastId); if (hitCap || (maxHotels > 0 && processed.get() >= maxHotels)) { break; } @@ -1044,6 +1542,10 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { } } + if (handleStopIfNeeded(logId, t0, (int) target, ok, fail, skip, processed, failLines, lastId)) { + return; + } + SmHotelSyncLogEntity done = syncLogMapper.selectById(logId); if (done == null) { return; @@ -1061,12 +1563,14 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { done.setFinishTime(LocalDateTime.now()); done.setUpdateTime(LocalDateTime.now()); syncLogMapper.updateById(done); + clearResumeCkpt(logId); log.info("|全量详情|完成|logId={}|ok={}|fail={}|skip={}|concurrency={}|priceDays={}|costMs={}|", logId, ok.get(), fail.get(), skip.get(), concurrency, effectivePriceDays, done.getCostMs()); } catch (Throwable e) { log.error("|全量详情|失败|logId={}|", logId, e); - failLog(logId, e.getMessage(), System.currentTimeMillis() - t0); + flushThenFail(logId, e.getMessage(), System.currentTimeMillis() - t0, ok, fail, skip, processed, 0, + failLines); } finally { if (pullPool != null) { @@ -1193,7 +1697,7 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { } } - /** 从 failDetail 解析 hotelCode(每行 code|name|reason) */ + /** 从 failDetail 解析 hotelCode(每行 code|name|原因;无编码时可能是酒店数字 ID) */ private static List parseFailHotelCodes(String failDetail) { List codes = new ArrayList<>(); if (StrUtil.isBlank(failDetail)) { @@ -1213,6 +1717,64 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { return codes; } + /** + * 失败明细码 → 本地酒店 ID:先按 hotelCode 匹配,未命中且为纯数字再按主键 ID 匹配。 + */ + private List resolveHotelIdsFromFailCodes(List codes) { + java.util.LinkedHashSet ids = new java.util.LinkedHashSet<>(); + if (CollUtil.isEmpty(codes)) { + return new ArrayList<>(); + } + java.util.LinkedHashSet unmatched = new java.util.LinkedHashSet<>(codes); + int batch = 200; + List codeList = new ArrayList<>(codes); + for (int i = 0; i < codeList.size(); i += batch) { + List slice = codeList.subList(i, Math.min(i + batch, codeList.size())); + List part = hotelMapper.selectList(Wrappers.lambdaQuery() + .select(SmHotelEntity::getId, SmHotelEntity::getHotelCode) + .in(SmHotelEntity::getHotelCode, slice)); + if (CollUtil.isEmpty(part)) { + continue; + } + for (SmHotelEntity h : part) { + if (h == null || h.getId() == null) { + continue; + } + ids.add(h.getId()); + if (StrUtil.isNotBlank(h.getHotelCode())) { + unmatched.remove(h.getHotelCode()); + } + } + } + List byId = new ArrayList<>(); + for (String c : unmatched) { + if (c != null && c.matches("\\d{1,19}")) { + try { + byId.add(Long.valueOf(c)); + } + catch (NumberFormatException ignored) { + // skip + } + } + } + if (CollUtil.isNotEmpty(byId)) { + for (int i = 0; i < byId.size(); i += batch) { + List slice = byId.subList(i, Math.min(i + batch, byId.size())); + List part = hotelMapper.selectList(Wrappers.lambdaQuery() + .select(SmHotelEntity::getId) + .in(SmHotelEntity::getId, slice)); + if (CollUtil.isNotEmpty(part)) { + for (SmHotelEntity h : part) { + if (h != null && h.getId() != null) { + ids.add(h.getId()); + } + } + } + } + } + return new ArrayList<>(ids); + } + /** 从滚价日志文案解析日期;解析不到则用配置偏移 */ private LocalDate resolveRollDay(SmHotelSyncLogEntity src) { String msg = StrUtil.blankToDefault(src.getStageMsg(), "") + " " + StrUtil.blankToDefault(src.getErrorMsg(), ""); @@ -1241,7 +1803,7 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { failLines.add(code + "|" + name + "|" + StrUtil.maxLength(StrUtil.blankToDefault(reason, "未知错误"), 300)); } - private static void applyFailDetail(SmHotelSyncLogEntity done, List failLines, int fail) { + private static void applyFailDetailOnly(SmHotelSyncLogEntity done, List failLines, int fail) { if (fail <= 0 || CollUtil.isEmpty(failLines)) { return; } @@ -1250,6 +1812,13 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService { detail = detail + "\n… 另有 " + (fail - failLines.size()) + " 条未列出"; } done.setFailDetail(detail); + } + + private static void applyFailDetail(SmHotelSyncLogEntity done, List failLines, int fail) { + applyFailDetailOnly(done, failLines, fail); + if (fail <= 0 || CollUtil.isEmpty(failLines)) { + return; + } done.setErrorMsg(StrUtil.maxLength("失败 " + fail + " 家,示例: " + failLines.get(0), 1000)); done.setStageMsg(StrUtil.maxLength("完成,成功见 successCount,失败明细见 failDetail", 500)); }