This commit is contained in:
dzc committed 2026-10-30 16:08:26 +08:00
commit 1c845b57d2
10 files changed
+1032 -101

No files matched your search

@@ -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;
}
@@ -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;
}
@@ -31,7 +31,7 @@ public class SmHotelSyncLogEntity extends Model<SmHotelSyncLogEntity> {
@Schema(description = "渠道/数据来源")
private String dataSource;
@Schema(description = "RUNNING / SUCCESS / FAILED")
@Schema(description = "RUNNING / PAUSED / SUCCESS / FAILED")
private String status;
private String stage;
@@ -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;
}
@@ -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<SmHotelDownstreamGapExcel> exportGaps(@RequestParam Long syncLogId) {
return pushLogService.exportGapsBySyncLogId(syncLogId);
}
@Operation(summary = "详情")
@GetMapping("/{id}")
@PreAuthorize("@pms.hasPermission('supermarket_smHotelDownstreamPushLog_view')")
@@ -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<Boolean> abortSyncLog(@PathVariable Long id) {
return syncService.abortSyncLog(id);
}
@SysLog("暂停酒店同步任务")
@Operation(summary = "暂停下游全量推/全量详情(批次结束后落盘,可继续)")
@PostMapping("/mid/syncLog/{id}/pause")
@PreAuthorize("@pms.hasPermission('supermarket_smHotel_sync')")
public R<Boolean> 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<SmHotelSyncLogEntity> resumeSyncLog(@PathVariable Long id) {
return syncService.resumeSyncLog(id);
}
@Operation(summary = "导出同步日志失败明细")
@ResponseExcel
@GetMapping("/mid/syncLog/{id}/exportFail")
@PreAuthorize("@pms.hasPermission('supermarket_smHotel_sync')")
public List<SmHotelSyncFailExcel> exportSyncFail(@PathVariable Long id) {
return syncService.exportFailDetail(id);
}
@SysLog("清理酒店同步卡死任务")
@Operation(summary = "清理长时间无进度的 RUNNING 同步日志")
@PostMapping("/mid/syncLog/abortStale")
@@ -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<SmHotelDownstreamPushLogEntity> {
Page<SmHotelDownstreamPushLogEntity> pageLogs(Page<SmHotelDownstreamPushLogEntity> page,
@@ -21,4 +24,10 @@ public interface SmHotelDownstreamPushLogService extends IService<SmHotelDownstr
/** 按创建时间区间删除(至少指定一端) */
R<Integer> removeByCreateTimeRange(String createTimeBegin, String createTimeEnd);
/**
* 按某次下游全量推同步日志导出完善清单(不限行数):
* ① 圣诺 getInfo 无店/失败;② getInfo 成功但房型/产品编码匹配不上。
*/
List<SmHotelDownstreamGapExcel> exportGapsBySyncLogId(Long syncLogId);
}
@@ -56,13 +56,29 @@ public interface SmHotelSyncService {
java.util.List<Long> filterDownstreamPushHotelIds(String partnerCode, java.util.List<Long> hotelIds);
/**
* 根据某次同步日志的 failDetail,只重试失败酒店(支持 DETAIL_PULL / PRICE_ROLL)。
* 根据某次同步日志的 failDetail,只重试失败酒店(支持 DETAIL_PULL / PRICE_ROLL / DOWNSTREAM_PUSH)。
* 会新开一条 RETRY 日志,可反复:完善数据 → 再点重试,直到失败明细清空。
*/
R<SmHotelSyncLogEntity> startRetryFailedAsync(Long sourceLogId);
/** 人工终止一条仍显示进行中的同步日志(死任务) */
/** 人工终止一条仍显示进行中或已暂停的同步日志(保留已落库失败明细) */
R<Boolean> abortSyncLog(Long id);
/**
* 暂停进行中的长任务(下游全量推/全量详情等)。当前批次结束后落盘进度与失败明细,可稍后「继续」。
* 发布代码前请先暂停,避免进程重启丢进度。
*/
R<Boolean> pauseSyncLog(Long id);
/**
* 继续已暂停任务(同 logId 接着跑,不丢已成功/失败计数与明细)。
* 当前支持 DOWNSTREAM_PUSH / DETAIL_PULL。
*/
R<SmHotelSyncLogEntity> resumeSyncLog(Long id);
/** 导出失败明细(failDetail 解析为行) */
java.util.List<com.bsd.pjcloud.supermarket.dto.SmHotelSyncFailExcel> exportFailDetail(Long id);
/**
* 将长时间无进度更新的 RUNNING 日志标为失败,避免挡后续任务。
* @return 清理条数
@@ -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<SmHotelDownstreamPushLogMapper, SmHotelDownstreamPushLogEntity>
implements SmHotelDownstreamPushLogService {
private final SmHotelSyncLogMapper syncLogMapper;
private final SmHotelMapper hotelMapper;
@Override
public Page<SmHotelDownstreamPushLogEntity> pageLogs(Page<SmHotelDownstreamPushLogEntity> page,
SmHotelDownstreamPushLogEntity query) {
@@ -108,6 +128,208 @@ public class SmHotelDownstreamPushLogServiceImpl
return R.ok(n, "已删除 " + n + " 条");
}
@Override
public List<SmHotelDownstreamGapExcel> 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<SmHotelDownstreamGapExcel> out = new ArrayList<>();
Set<String> dedupe = new HashSet<>();
Map<Long, SmHotelEntity> hotelMap = new HashMap<>();
// 分批扫本次同步相关流水(不限行数):只要 getInfo 无店 + 房型/产品匹配缺口
long lastId = 0L;
while (true) {
List<SmHotelDownstreamPushLogEntity> batch = baseMapper.selectList(Wrappers
.<SmHotelDownstreamPushLogEntity>lambdaQuery()
.eq(SmHotelDownstreamPushLogEntity::getSyncLogId, syncLogId)
.gt(SmHotelDownstreamPushLogEntity::getId, lastId)
.orderByAsc(SmHotelDownstreamPushLogEntity::getId)
.last("limit 500"));
if (CollUtil.isEmpty(batch)) {
break;
}
Set<Long> 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<Long, SmHotelEntity> loadHotels(Set<Long> hotelIds) {
Map<Long, SmHotelEntity> map = new HashMap<>();
if (CollUtil.isEmpty(hotelIds)) {
return map;
}
List<Long> ids = new ArrayList<>(hotelIds);
int batch = 500;
for (int i = 0; i < ids.size(); i += batch) {
List<Long> slice = ids.subList(i, Math.min(i + batch, ids.size()));
List<SmHotelEntity> rows = hotelMapper.selectList(Wrappers.<SmHotelEntity>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<SmHotelDownstreamGapExcel> out, Set<String> 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<SmHotelDownstreamGapExcel> out, Set<String> dedupe, Long syncLogId,
String partner, String failDetail, Map<Long, SmHotelEntity> hotelMap) {
if (StrUtil.isBlank(failDetail)) {
return;
}
Map<String, SmHotelEntity> 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<String> 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.<SmHotelEntity>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<SmHotelDownstreamGapExcel> out, Set<String> 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)) {
@@ -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<Long> 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<String> codes = parseFailHotelCodes(src.getFailDetail());
if (CollUtil.isEmpty(codes)) {
return R.failed("该日志无失败明细可重试(请确认 failDetail 有 hotelCode|名称|原因)");
return R.failed("该日志无失败明细可重试(请确认 failDetail 有 hotelCode|名称|原因;可先导出失败明细核对)");
}
abortStaleRunningSyncLogs();
Long running = syncLogMapper.selectCount(Wrappers.<SmHotelSyncLogEntity>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<Long> 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<Long> 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<Boolean> 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<SmHotelSyncLogEntity> 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.<SmHotelSyncLogEntity>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<Long> 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<Long> ids = List.copyOf(hotelIds == null ? List.of() : hotelIds);
final int baseIdx = resumeIdx;
final List<String> 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<Long> 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<String> 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<SmHotelSyncFailExcel> exportFailDetail(Long id) {
List<SmHotelSyncFailExcel> 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<Long> hotelIds) {
/**
* 按范围映射向指定下游推送本地酒店(默认酒店级并行)。
* @param hotelIds 本段待推列表(续跑时已去掉前面已处理)
* @param baseProcessed 续跑前已处理条数(用于进度分母与检查点)
* @param totalAll 全量目标家数(进度分母)
*/
private void runDownstreamScopePush(Long logId, String partnerCode, List<Long> hotelIds, int baseProcessed,
int baseOk, int baseFail, int baseSkip, List<String> 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<String> 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<String> 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<Map<String, Object>> 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<Long> slice = hotelIds.subList(i, Math.min(i + batchSize, hotelIds.size()));
List<Callable<Void>> 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<Future<Void>> futures = pushPool.invokeAll(tasks);
for (Future<Void> 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<String> 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<Map<String, Object>> 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<String> 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<Long> filterDownstreamPushHotelIds(String partnerCode, List<Long> 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<String> 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<String> 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<String> 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<String> 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<String> parseFailLines(String failDetail) {
List<String> 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<String> 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<String> 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.<SmHotelEntity>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<String> 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<SmHotelEntity> rows = hotelMapper.selectList(Wrappers.<SmHotelEntity>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<String> parseFailHotelCodes(String failDetail) {
List<String> codes = new ArrayList<>();
if (StrUtil.isBlank(failDetail)) {
@@ -1213,6 +1717,64 @@ public class SmHotelSyncServiceImpl implements SmHotelSyncService {
return codes;
}
/**
* 失败明细码 → 本地酒店 ID:先按 hotelCode 匹配,未命中且为纯数字再按主键 ID 匹配。
*/
private List<Long> resolveHotelIdsFromFailCodes(List<String> codes) {
java.util.LinkedHashSet<Long> ids = new java.util.LinkedHashSet<>();
if (CollUtil.isEmpty(codes)) {
return new ArrayList<>();
}
java.util.LinkedHashSet<String> unmatched = new java.util.LinkedHashSet<>(codes);
int batch = 200;
List<String> codeList = new ArrayList<>(codes);
for (int i = 0; i < codeList.size(); i += batch) {
List<String> slice = codeList.subList(i, Math.min(i + batch, codeList.size()));
List<SmHotelEntity> part = hotelMapper.selectList(Wrappers.<SmHotelEntity>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<Long> 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<Long> slice = byId.subList(i, Math.min(i + batch, byId.size()));
List<SmHotelEntity> part = hotelMapper.selectList(Wrappers.<SmHotelEntity>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<String> failLines, int fail) {
private static void applyFailDetailOnly(SmHotelSyncLogEntity done, List<String> 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<String> 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));
}