修复酒店回调队列问题

This commit is contained in:
zjf committed 2026-10-15 17:05:50 +08:00
1 parent 57f7f0d287
commit 846820b2db
8 files changed
+327 -37

No files matched your search

@@ -44,13 +44,22 @@ public class SmHotelCallbackTaskController {
}
@SysLog("酒店回调任务重试")
@Operation(summary = "失败/跳过任务重新入队")
@Operation(summary = "失败/跳过/待处理任务重新入队")
@PostMapping("/retry/{id}")
@PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_retry')")
public R<Boolean> retry(@PathVariable Long id) {
return callbackTaskService.retry(id);
}
@SysLog("酒店回调待处理任务批量重新入队")
@Operation(summary = "把滞留 PENDING(及超时 RUNNING)重新推入 Redis 队列")
@PostMapping("/requeuePending")
@PreAuthorize("@pms.hasPermission('supermarket_smHotelCallback_retry')")
public R<Integer> requeuePending() {
int n = callbackTaskService.requeueStalePending();
return R.ok(n, "已重新入队 " + n + " 条(含超时 RUNNING 恢复)");
}
@SysLog("删除酒店回调任务")
@Operation(summary = "删除单条回调任务")
@DeleteMapping("/{id}")
@@ -8,6 +8,7 @@ import com.bsd.pjcloud.common.log.annotation.SysLog;
import com.bsd.pjcloud.common.security.annotation.Inner;
import com.bsd.pjcloud.supermarket.integration.downstream.DownstreamInboundAuth;
import com.bsd.pjcloud.supermarket.service.SmHotelHuanqiOrderService;
import com.bsd.pjcloud.supermarket.service.SmHotelOrderService;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import io.swagger.v3.oas.annotations.tags.Tag;
@@ -34,10 +35,41 @@ public class SmHotelHuanqiOrderController {
private final SmHotelHuanqiOrderService huanqiOrderService;
private final SmHotelOrderService hotelOrderService;
/**
* 环启(B)向本系统(A)推送订单。响应体必须是 {code:200},与文档约定一致。
* 配置到环启侧 order_push_url。
* 销售前二次校验(环启文档无此接口,我方补充):下单前查本地酒店/产品/上架/日历库存/可选验价。
* 请环启在成交前调用;返回 data.bookable=true 再售卖。
*/
@Inner(value = false)
@Operation(summary = "下游下单前可订校验")
@PostMapping("/third/party/api/order/check")
public Map<String, Object> checkBookable(HttpServletRequest request, @RequestBody String rawBody) {
Map<String, Object> resp = new LinkedHashMap<>(4);
try {
String appKey = header(request, "appKey");
inboundAuth.verifyForHuanqi(appKey, header(request, "timestamp"), header(request, "nonce"),
header(request, "sign"), rawBody);
JSONObject payload = JSONUtil.parseObj(rawBody);
R<Map<String, Object>> r = hotelOrderService.checkBookableForDownstream(payload);
if (r != null && r.getCode() == 0) {
resp.put("code", 200);
resp.put("msg", "success");
resp.put("data", r.getData());
return resp;
}
resp.put("code", 500);
resp.put("msg", r == null ? "校验失败" : r.getMsg());
return resp;
}
catch (Exception e) {
log.warn("|环启订单预检|失败|{}", e.getMessage());
resp.put("code", 500);
resp.put("msg", StrUtil.blankToDefault(e.getMessage(), "校验失败"));
return resp;
}
}
@Inner(value = false)
@Operation(summary = "接收环启订单推送(push/update/cancel)")
@PostMapping("/third/party/api/order/push")
@@ -2,6 +2,7 @@ package com.bsd.pjcloud.supermarket.integration.shangyou;
import com.bsd.pjcloud.supermarket.config.ShangyouHotelProperties;
import com.bsd.pjcloud.supermarket.service.SmHotelCallbackTaskService;
import com.bsd.pjcloud.supermarket.service.SmHotelChannelConfigService;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
@@ -18,14 +19,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
/**
* 消费 Redis 回调队列:BRPOP 高优队列优先于普通队列。
* <p>
* 另起看门狗线程:即使 Worker 卡在某次上游调用里,也能把超时 RUNNING 重置入队,避免整队假死。
* 线程始终拉起;是否真正消费看渠道运行时 {@code callbackQueueEnabled}(库表可覆盖 YAML)。
* 看门狗定期把滞留 PENDING / 超时 RUNNING 重新入队,避免「库里全是待处理、Redis 已空」假死。
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class ShangyouCallbackQueueWorker implements ApplicationRunner {
private final ShangyouHotelProperties properties;
private final SmHotelChannelConfigService channelConfigService;
private final StringRedisTemplate redisTemplate;
@@ -33,21 +35,27 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
private final AtomicBoolean running = new AtomicBoolean(true);
/** 上次消费循环是否处于启用态;从关→开时补一次 PENDING 入队 */
private final AtomicBoolean lastEnabled = new AtomicBoolean(false);
private Thread worker;
private Thread watchdog;
private ShangyouHotelProperties props() {
return channelConfigService.resolveShangyouRuntime();
}
@Override
public void run(ApplicationArguments args) {
if (!properties.isCallbackQueueEnabled()) {
log.info("|回调队列|Worker 未启用(shangyou-hotel.callback-queue-enabled=false)");
return;
boolean enabled = props().isCallbackQueueEnabled();
lastEnabled.set(enabled);
if (!enabled) {
log.warn("|回调队列|当前 callback-queue-enabled=false,Worker 已启动但暂停消费;"
+ "库表/YAML 打开后自动恢复。PENDING 任务不会执行,直到开启。");
}
try {
callbackTaskService.requeueStalePending();
}
catch (Exception e) {
log.warn("|回调队列|恢复 PENDING 失败(可稍后人工重试)", e);
else {
safeRequeuePending();
}
worker = new Thread(this::loop, "shangyou-callback-queue");
worker.setDaemon(true);
@@ -57,7 +65,7 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
watchdog.setDaemon(true);
watchdog.start();
log.info("|回调队列|Worker+Watchdog 已启动|high={}|normal={}|",
log.info("|回调队列|Worker+Watchdog 已启动|enabled={}|high={}|normal={}|", enabled,
ShangyouHotelApi.REDIS_CALLBACK_QUEUE_HIGH, ShangyouHotelApi.REDIS_CALLBACK_QUEUE);
}
@@ -65,6 +73,19 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
int consecutiveFail = 0;
while (running.get()) {
try {
ShangyouHotelProperties p = props();
boolean enabled = p.isCallbackQueueEnabled();
if (!enabled) {
if (lastEnabled.getAndSet(false)) {
log.warn("|回调队列|已暂停消费(callback-queue-enabled=false)");
}
Thread.sleep(3000L);
continue;
}
if (!lastEnabled.getAndSet(true)) {
log.info("|回调队列|已恢复消费,补入队滞留 PENDING");
safeRequeuePending();
}
// 底层 BRPOP 多 key:先高优再普通(ListOperations 无多 key 重载)
String taskId = brpopPreferHigh(5);
if (taskId == null || taskId.isBlank() || "null".equalsIgnoreCase(taskId)) {
@@ -79,7 +100,7 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
continue;
}
boolean ok = callbackTaskService.processQueuedTask(id);
int interval = Math.max(0, properties.getCallbackTaskIntervalMs());
int interval = Math.max(0, p.getCallbackTaskIntervalMs());
if (interval > 0) {
Thread.sleep(interval);
}
@@ -88,9 +109,9 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
}
else {
consecutiveFail++;
int pauseAfter = Math.max(1, properties.getCallbackFailPauseAfter());
int pauseAfter = Math.max(1, p.getCallbackFailPauseAfter());
if (consecutiveFail >= pauseAfter) {
int pauseMs = Math.max(0, properties.getCallbackFailPauseMs());
int pauseMs = Math.max(0, p.getCallbackFailPauseMs());
log.warn("|回调队列|连续失败 {} 次,冷却 {}ms 后再继续|", consecutiveFail, pauseMs);
if (pauseMs > 0) {
Thread.sleep(pauseMs);
@@ -119,6 +140,15 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
}
}
private void safeRequeuePending() {
try {
callbackTaskService.requeueStalePending();
}
catch (Exception e) {
log.warn("|回调队列|恢复 PENDING 失败(可稍后人工重试)", e);
}
}
/**
* Redis BRPOP:按 key 顺序优先弹出高优队列。
* 当前 Spring Data Redis 的 {@code ListOperations#rightPop} 不支持多 key,走 connection 原生命令。
@@ -141,7 +171,7 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
return taskId == null ? null : taskId.trim();
}
/** 不依赖 Worker 空闲:定时扫卡死的 RUNNING */
/** 不依赖 Worker 空闲:扫超时 RUNNING + 滞留 PENDING(Redis 丢队列时也能救回来) */
private void watchdogLoop() {
while (running.get()) {
try {
@@ -149,7 +179,12 @@ public class ShangyouCallbackQueueWorker implements ApplicationRunner {
if (!running.get()) {
break;
}
if (!props().isCallbackQueueEnabled()) {
continue;
}
callbackTaskService.recoverStaleRunning();
// 只补「超过约 3 分钟仍 PENDING」的孤儿任务,避免把排队中的任务反复 leftPush
callbackTaskService.requeueOrphanPending();
}
catch (InterruptedException ie) {
Thread.currentThread().interrupt();
@@ -41,9 +41,15 @@ public interface SmHotelCallbackTaskService {
/** 消费一条队列任务(Worker 调用);返回 true=本次成功/跳过,false=失败或待再试 */
boolean processQueuedTask(Long taskId);
/** 启动时把滞留 PENDING / 超时 RUNNING 重新入队 */
/** 启动/人工:把全部 PENDING + 超时 RUNNING 重新入队 */
int requeueStalePending();
/**
* 看门狗:只把「超过若干分钟仍为 PENDING」的任务补入队(避免 Redis 丢队列后假死;
* 不会把刚入队、尚在排队的任务反复 leftPush)。
*/
int requeueOrphanPending();
/** 仅把超时仍为 RUNNING 的任务重置入队(Worker 空闲时定期调用) */
int recoverStaleRunning();
}
@@ -107,4 +107,10 @@ public interface SmHotelOrderService {
* 已有 platformNo 则幂等跳过。
*/
R<Boolean> bookUpstreamForExisting(Long orderId);
/**
* 下游销售前二次校验(查本地酒店/产品/上架/日历库存/可选验价)。
* 入参建议:extHotelId、extProductId、checkIn、checkOut、roomCnt、price(可选)。
*/
R<Map<String, Object>> checkBookableForDownstream(Map<String, Object> body);
}
@@ -14,6 +14,7 @@ import com.bsd.pjcloud.supermarket.integration.shangyou.ShangyouHotelClient;
import com.bsd.pjcloud.supermarket.mapper.SmHotelCallbackTaskMapper;
import com.bsd.pjcloud.supermarket.mapper.SmHotelMapper;
import com.bsd.pjcloud.supermarket.service.SmHotelCallbackTaskService;
import com.bsd.pjcloud.supermarket.service.SmHotelChannelConfigService;
import com.bsd.pjcloud.supermarket.service.SmHotelMidService;
import com.bsd.pjcloud.supermarket.service.SmHotelOrderService;
import com.bsd.pjcloud.common.core.util.R;
@@ -48,7 +49,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
public static final String STATUS_SKIPPED = "SKIPPED";
private final ShangyouHotelProperties properties;
private final SmHotelChannelConfigService channelConfigService;
private final ShangyouHotelClient client;
@@ -62,6 +63,11 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
private final StringRedisTemplate redisTemplate;
/** 与 Worker 一致:读渠道库表覆盖后的运行时,不要只读 YAML 默认值 */
private ShangyouHotelProperties props() {
return channelConfigService.resolveShangyouRuntime();
}
@Override
public Map<String, Object> enqueueFromNotify(String notifyType, Map<String, Object> body, String time,
String sign) {
@@ -75,18 +81,25 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
return failResp("401", "验签失败");
}
Long tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId();
ShangyouHotelProperties p = props();
Long tenantId = p.getCallbackTenantId() == null ? 1L : p.getCallbackTenantId();
try {
TenantBroker.runAs(tenantId, tid -> {
SmHotelCallbackTaskEntity task = buildTask(notifyType, body, time, tid);
taskMapper.insert(task);
if (properties.isCallbackQueueEnabled()) {
if (p.isCallbackQueueEnabled()) {
enqueueTaskId(task.getId(), task.getNotifyType());
}
log.info("|上游酒店通知|已入队|taskId={}|type={}|priority={}|hotelCode={}|hotelId={}|status={}|",
else {
task.setResultMsg("已落库,但 callback-queue-enabled=false,未入 Redis;开启后会自动补入队");
taskMapper.updateById(task);
log.warn("|上游酒店通知|未入 Redis(队列关闭)|taskId={}|type={}|", task.getId(),
task.getNotifyType());
}
log.info("|上游酒店通知|已入队|taskId={}|type={}|priority={}|hotelCode={}|hotelId={}|status={}|queueOn={}|",
task.getId(), task.getNotifyType(),
isHighPriority(task.getNotifyType()) ? "HIGH" : "NORMAL", task.getHotelCode(),
task.getHotelId(), task.getStatus());
task.getHotelId(), task.getStatus(), p.isCallbackQueueEnabled());
});
}
catch (Exception e) {
@@ -126,8 +139,12 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
if (task == null) {
return R.failed("任务不存在");
}
if (STATUS_RUNNING.equals(task.getStatus()) || STATUS_PENDING.equals(task.getStatus())) {
return R.failed("任务处理中或已在队列,请稍候");
// RUNNING 禁止;PENDING 允许补入 Redis(常见:队列关过 / Redis 丢过,页面一直「待处理」)
if (STATUS_RUNNING.equals(task.getStatus())) {
return R.failed("任务处理中,请稍候");
}
if (!props().isCallbackQueueEnabled()) {
return R.failed("回调队列未启用(callback-queue-enabled=false),请先在渠道对接配置打开");
}
if (task.getHotelId() == null && StrUtil.isNotBlank(task.getHotelCode())) {
SmHotelEntity hotel = resolveWritableHotelByCode(task.getHotelCode(), SmHotelDataSource.SHANGYOU);
@@ -230,29 +247,47 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
if (taskId == null) {
return true;
}
Long tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId();
ShangyouHotelProperties p = props();
Long tenantId = p.getCallbackTenantId() == null ? 1L : p.getCallbackTenantId();
Boolean ok = TenantBroker.applyAs(tenantId, tid -> doProcess(taskId));
return Boolean.TRUE.equals(ok);
}
@Override
public int requeueStalePending() {
Long tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId();
return requeuePendingInternal(null);
}
@Override
public int requeueOrphanPending() {
// 超过 3 分钟仍 PENDING,视为 Redis 未消费到(丢队列 / 当时未入队)
return requeuePendingInternal(LocalDateTime.now().minusMinutes(3));
}
/**
* @param olderThan 非空时仅入队 updateTime &lt; olderThan 的 PENDING;空则入队全部 PENDING(最多 200)
*/
private int requeuePendingInternal(LocalDateTime olderThan) {
ShangyouHotelProperties p = props();
if (!p.isCallbackQueueEnabled()) {
return 0;
}
Long tenantId = p.getCallbackTenantId() == null ? 1L : p.getCallbackTenantId();
return TenantBroker.applyAs(tenantId, tid -> {
int n = 0;
// 1) PENDING:按创建时间 FIFO;ORDER_STATUS 进高优队列
var pending = taskMapper.selectList(Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
var qw = Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
.eq(SmHotelCallbackTaskEntity::getStatus, STATUS_PENDING)
.lt(olderThan != null, SmHotelCallbackTaskEntity::getUpdateTime, olderThan)
.orderByAsc(SmHotelCallbackTaskEntity::getCreateTime)
.last("limit 200"));
.last("limit 200");
var pending = taskMapper.selectList(qw);
for (SmHotelCallbackTaskEntity t : pending) {
enqueueTaskId(t.getId(), t.getNotifyType());
n++;
}
// 2) 卡住的 RUNNING
n += resetStaleRunningLocked();
if (n > 0) {
log.info("|回调队列|重启恢复入队|pendingAndStaleRunning={}|", n);
log.info("|回调队列|补入队|pendingFilterOlderThan={}|count={}|", olderThan, n);
}
return n;
});
@@ -260,7 +295,8 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
@Override
public int recoverStaleRunning() {
Long tenantId = properties.getCallbackTenantId() == null ? 1L : properties.getCallbackTenantId();
ShangyouHotelProperties p = props();
Long tenantId = p.getCallbackTenantId() == null ? 1L : p.getCallbackTenantId();
return TenantBroker.applyAs(tenantId, tid -> resetStaleRunningLocked());
}
@@ -268,7 +304,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
private int resetStaleRunningLocked() {
// 单日区间拉取正常应秒级完成;2 分钟仍 RUNNING 视为卡死(Token 冷却/进程挂起等)
LocalDateTime staleBefore = LocalDateTime.now().minusMinutes(2);
int max = Math.max(1, properties.getCallbackMaxAttempts());
int max = Math.max(1, props().getCallbackMaxAttempts());
var running = taskMapper.selectList(Wrappers.<SmHotelCallbackTaskEntity>lambdaQuery()
.eq(SmHotelCallbackTaskEntity::getStatus, STATUS_RUNNING)
.and(w -> w.lt(SmHotelCallbackTaskEntity::getStartProcessTime, staleBefore)
@@ -313,7 +349,7 @@ public class SmHotelCallbackTaskServiceImpl implements SmHotelCallbackTaskServic
return true;
}
int attempts = task.getAttemptCount() == null ? 0 : task.getAttemptCount();
int max = Math.max(1, properties.getCallbackMaxAttempts());
int max = Math.max(1, props().getCallbackMaxAttempts());
task.setStatus(STATUS_RUNNING);
task.setAttemptCount(attempts + 1);
task.setStartProcessTime(LocalDateTime.now());
@@ -176,7 +176,29 @@ public class SmHotelHuanqiOrderServiceImpl implements SmHotelHuanqiOrderService
scheduleCancel(tenantId, orderId);
}
else if (channelStatus != null && channelStatus >= 1) {
scheduleBook(tenantId, orderId);
// 推单履约前再验本地可订,避免环启已售、我方无库存/下架导致扯皮
Map<String, Object> checkBody = new LinkedHashMap<>();
checkBody.put("extHotelId", order.getHotelCode());
checkBody.put("extProductId", order.getProductCode());
checkBody.put("extApartmentId", order.getRoomTypeCode());
checkBody.put("checkIn", order.getCheckInDate() == null ? null : order.getCheckInDate().toString());
checkBody.put("checkOut", order.getCheckOutDate() == null ? null : order.getCheckOutDate().toString());
checkBody.put("roomCnt", order.getCount() == null ? 1 : order.getCount());
// 推单履约预检不带 price:环启 price 是结算价,与本地日历卖价口径不同,避免误拦
R<Map<String, Object>> check = hotelOrderService.checkBookableForDownstream(checkBody);
boolean bookable = check != null && check.getCode() == 0 && check.getData() != null
&& Boolean.TRUE.equals(check.getData().get("bookable"));
if (!bookable) {
Object reasons = check == null || check.getData() == null ? check == null ? "预检失败" : check.getMsg()
: check.getData().get("reasons");
order.setOrderStatus("NEED_MANUAL");
order.setFailMsg(StrUtil.maxLength("推单履约前预检不可订:" + reasons, 500));
orderMapper.updateById(order);
log.warn("|环启订单|预检不可订,转人工|orderNo={}|reasons={}|", orderNo, reasons);
}
else {
scheduleBook(tenantId, orderId);
}
}
Map<String, Object> ok = new LinkedHashMap<>(6);
@@ -42,6 +42,7 @@ import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -557,6 +558,149 @@ public class SmHotelOrderServiceImpl extends ServiceImpl<SmHotelOrderMapper, SmH
return R.ok(Boolean.TRUE);
}
@Override
public R<Map<String, Object>> checkBookableForDownstream(Map<String, Object> body) {
Map<String, Object> data = new LinkedHashMap<>(16);
List<String> reasons = new ArrayList<>();
if (body == null || body.isEmpty()) {
return R.failed("报文为空");
}
String extHotelId = StrUtil.trim(str(body.get("extHotelId")));
if (StrUtil.isBlank(extHotelId)) {
extHotelId = StrUtil.trim(str(body.get("hotelCode")));
}
String extProductId = StrUtil.trim(str(body.get("extProductId")));
if (StrUtil.isBlank(extProductId)) {
extProductId = StrUtil.trim(str(body.get("productCode")));
}
String extApartmentId = StrUtil.trim(str(body.get("extApartmentId")));
if (StrUtil.isBlank(extApartmentId)) {
extApartmentId = StrUtil.trim(str(body.get("roomTypeCode")));
}
LocalDate checkIn = parseDate(body.get("checkIn") != null ? body.get("checkIn") : body.get("checkInDate"));
LocalDate checkOut = parseDate(body.get("checkOut") != null ? body.get("checkOut") : body.get("checkOutDate"));
int roomCnt = 1;
Object cntObj = body.get("roomCnt") != null ? body.get("roomCnt") : body.get("count");
if (cntObj != null) {
try {
roomCnt = Math.max(1, Integer.parseInt(String.valueOf(cntObj)));
}
catch (Exception ignored) {
roomCnt = 1;
}
}
BigDecimal expectPrice = toDecimal(body.get("price") != null ? body.get("price") : body.get("totalAmount"));
data.put("extHotelId", extHotelId);
data.put("extProductId", extProductId);
data.put("extApartmentId", extApartmentId);
data.put("checkIn", checkIn == null ? null : checkIn.toString());
data.put("checkOut", checkOut == null ? null : checkOut.toString());
data.put("roomCnt", roomCnt);
if (StrUtil.isBlank(extHotelId)) {
reasons.add("extHotelId 不能为空");
}
if (StrUtil.isBlank(extProductId)) {
reasons.add("extProductId 不能为空");
}
if (checkIn == null || checkOut == null || !checkIn.isBefore(checkOut)) {
reasons.add("checkIn/checkOut 无效");
}
if (!reasons.isEmpty()) {
data.put("bookable", false);
data.put("reasons", reasons);
return R.ok(data);
}
SmHotelEntity hotel = hotelMapper.selectOne(Wrappers.<SmHotelEntity>lambdaQuery()
.eq(SmHotelEntity::getHotelCode, extHotelId)
.last("limit 1"));
if (hotel == null) {
reasons.add("本地无此酒店(请先 hotel/sync 建档):" + extHotelId);
data.put("bookable", false);
data.put("reasons", reasons);
return R.ok(data);
}
data.put("hotelId", hotel.getId());
data.put("hotelName", hotel.getHotelName());
data.put("hotelDataSource", hotel.getSheetName());
data.put("upstreamStatus", hotel.getUpstreamStatus());
if ("OFFLINE".equalsIgnoreCase(hotel.getUpstreamStatus())) {
reasons.add("酒店已下架");
}
SmHotelProductEntity product = productMapper.selectOne(Wrappers.<SmHotelProductEntity>lambdaQuery()
.eq(SmHotelProductEntity::getHotelId, hotel.getId())
.eq(SmHotelProductEntity::getProductCode, extProductId)
.last("limit 1"));
if (product == null) {
reasons.add("本地无此产品:" + extProductId);
data.put("productExists", false);
data.put("bookable", false);
data.put("reasons", reasons);
return R.ok(data);
}
data.put("productExists", true);
data.put("productName", product.getProductName());
if (StrUtil.isNotBlank(extApartmentId) && StrUtil.isNotBlank(product.getRoomTypeCode())
&& !extApartmentId.equals(product.getRoomTypeCode())) {
reasons.add("房型不匹配:报文=" + extApartmentId + ",产品所属=" + product.getRoomTypeCode());
}
// 日历库存 + 本地卖价合计
BigDecimal localSell = BigDecimal.ZERO;
List<Map<String, Object>> days = new ArrayList<>();
boolean localDs = SmHotelDataSource.isLocal(hotel.getSheetName());
String invMsg = precheckLocalInventory(product.getId(), checkIn, checkOut, roomCnt);
if (invMsg != null) {
reasons.add(invMsg);
}
for (LocalDate d = checkIn; d.isBefore(checkOut); d = d.plusDays(1)) {
SmHotelDailyEntity row = dailyMapper.selectOne(Wrappers.<SmHotelDailyEntity>lambdaQuery()
.eq(SmHotelDailyEntity::getProductId, product.getId())
.eq(SmHotelDailyEntity::getStayDate, d)
.last("limit 1"));
Map<String, Object> one = new LinkedHashMap<>(8);
one.put("date", d.toString());
if (row == null) {
one.put("exists", false);
}
else {
int stock = row.getStock() == null ? 0 : row.getStock();
int sold = row.getSold() == null ? 0 : row.getSold();
int remain = Math.max(0, stock - sold);
BigDecimal sell = row.getSellPrice() != null ? row.getSellPrice()
: (product.getSellPrice() != null ? product.getSellPrice() : BigDecimal.ZERO);
localSell = localSell.add(sell.multiply(BigDecimal.valueOf(roomCnt)));
one.put("exists", true);
one.put("dailyStatus", row.getDailyStatus());
one.put("remain", remain);
one.put("sellPrice", sell);
}
days.add(one);
}
data.put("days", days);
data.put("localSellAmount", localSell);
if (expectPrice != null) {
data.put("expectPrice", expectPrice);
BigDecimal diff = localSell.subtract(expectPrice).abs();
boolean match = diff.compareTo(new BigDecimal("0.50")) <= 0;
data.put("priceMatch", match);
data.put("priceDiff", diff);
if (!match) {
reasons.add("结算价不一致:本地合计 " + localSell + ",报文 " + expectPrice + "(容差 0.50 元)");
}
}
// 第三方来源:额外提示上游可定需在真正下单时再 DoubleCheck
data.put("needUpstreamDoubleCheck", !localDs);
data.put("bookable", reasons.isEmpty());
data.put("reasons", reasons);
return R.ok(data);
}
@Override
@Transactional(rollbackFor = Exception.class)
public R<JSONObject> createByAdmin(Map<String, Object> body) {