增加一个定时任务。如果拣货出库过程中，相同料箱号，存在(199 ，200)的任务 并且同时存在 101,196的任务 则101和196的任务会 自动...

chen.lin

昨天 ad5fef776eb9d66a0fcfc2ddf89698c5642e0eec

 rsf-server/src/main/java/com/vincent/rsf/server/manager/schedules/TaskSchedules.java

@@ -104,10 +104,11 @@


    /**
     * @param
     * @return
     * 完成入库，更新库位明细、组托状态，并在此统一执行 9.1 入/出库结果上报云仓。
     * 与 RCS 回调形成闭环：RCS 上报任务结束后仅将任务状态置为 COMPLETE_IN（见 WcsServiceImpl.receiveExMsg），
     * 本定时任务扫描 COMPLETE_IN 并执行 complateInTask（库位、组托、9.1 上报云仓）。
     *
     * @author Ryan
     * @description 完成入库，更新库存
     * @time 2025/4/2 12:37
     */
    @Scheduled(cron = "0/3 * * * * ?")
@@ -122,10 +123,12 @@
    }

    /**
     * 完成出库任务，更新库位/出库单，并在此统一执行 9.1 入/出库结果上报云仓。
     * 与 RCS 回调形成闭环：RCS 上报 END 后仅将出库任务状态置为 COMPLETE_OUT（见 WcsServiceImpl.receiveExMsg），
     * 本定时任务扫描 COMPLETE_OUT 并执行 completeTask（扣库位、更新出库单、9.1 上报云仓）。
     *
     * @author Ryan
     * @date 2025/5/20
     * @description: 完成出库任务，更新库存
     * @version 1.0
     */
    @Scheduled(cron = "0/5 * * * * ?  ")
    @Transactional(rollbackFor = Exception.class)
@@ -251,7 +254,7 @@
    @Scheduled(cron = "0/35 * * * * ?  ")
    @Transactional(rollbackFor = Exception.class)
    public void pubTaskToWcs() {
        log.info("定时任务开始执行：任务下发到RCS");
        log.debug("定时任务开始执行：任务下发到RCS");
        Long loginUserId = SystemAuthUtils.getLoginUserId();
        List<Integer> list = Arrays.asList(
                 TaskType.TASK_TYPE_LOC_MOVE.type
@@ -271,7 +274,7 @@
                .in(Task::getTaskType, list)
                .in(Task::getTaskStatus, integers)
                .orderByDesc(Task::getSort));
        log.info("查询到待下发任务数量：{}", tasks.size());
        log.debug("查询到待下发任务数量：{}", tasks.size());
        if (tasks.isEmpty()) {
            log.debug("没有待下发的任务，定时任务结束");
            return;
@@ -283,6 +286,28 @@
                return;
            }
        }
        // 同料箱号规则：若 101 任务所在料箱号下已存在 196/198/199/200 任务，则不向 RCS 发送该 101 任务（/api/open/bus/submit）
        List<Integer> higherStatuses = Arrays.asList(TaskStsType.AWAIT.id, TaskStsType.COMPLETE_OUT.id, TaskStsType.WAVE_SEED.id, TaskStsType.UPDATED_OUT.id);
        List<Task> higherTasks = taskService.list(new LambdaQueryWrapper<Task>()
                .in(Task::getTaskStatus, higherStatuses)
                .isNotNull(Task::getBarcode)
                .ne(Task::getBarcode, ""));
        Set<String> barcodesWithHigher = higherTasks.stream().map(Task::getBarcode).filter(StringUtils::isNotBlank).collect(Collectors.toSet());
        final Set<String> skipBarcodes = barcodesWithHigher;
        List<Task> toPublish = tasks.stream()
                .filter(t -> {
                    if (TaskStsType.GENERATE_OUT.id.equals(t.getTaskStatus()) && StringUtils.isNotBlank(t.getBarcode()) && skipBarcodes.contains(t.getBarcode())) {
                        log.debug("同料箱号{}下已存在196/198/199/200任务，跳过101任务下发：taskId={}", t.getBarcode(), t.getId());
                        return false;
                    }
                    return true;
                })
                .collect(Collectors.toList());
        if (toPublish.isEmpty()) {
            log.debug("过滤后无待下发任务，定时任务结束");
            return;
        }
        tasks = toPublish;
//        for (Task task : tasks) {
//            /**移库不做站点操作*/
//            if (!task.getTaskType().equals(TaskType.TASK_TYPE_LOC_MOVE.type)) {
@@ -295,9 +320,44 @@
//            }
//        }
        /**下发普通站点任务，报错回滚，不再往下执行*/
        log.info("开始下发{}个任务到RCS", tasks.size());
        log.debug("开始下发{}个任务到RCS", tasks.size());
        taskService.pubTaskToWcs(tasks);
        log.info("定时任务执行完成：任务下发到RCS");
        log.debug("定时任务执行完成：任务下发到RCS");
    }

    /**
     * 拣货出库同料箱号状态同步：相同料箱号下若同时存在(199、200)任务与(101、196)任务，则将 101、196 自动更新为 199。
     * 执行周期与任务下发一致，便于在下发前完成状态同步。
     */
    @Scheduled(cron = "0/35 * * * * ?  ")
    @Transactional(rollbackFor = Exception.class)
    public void syncBarcodeTaskStatusTo199() {
        List<Integer> statuses = Arrays.asList(TaskStsType.GENERATE_OUT.id, TaskStsType.AWAIT.id, TaskStsType.WAVE_SEED.id, TaskStsType.UPDATED_OUT.id);
        List<Task> candidates = taskService.list(new LambdaQueryWrapper<Task>()
                .in(Task::getTaskStatus, statuses)
                .isNotNull(Task::getBarcode)
                .ne(Task::getBarcode, ""));
        if (candidates.isEmpty()) return;
        Map<String, Set<Integer>> statusByBarcode = new HashMap<>();
        for (Task t : candidates) {
            statusByBarcode.computeIfAbsent(t.getBarcode(), k -> new HashSet<>()).add(t.getTaskStatus());
        }
        List<Integer> to199 = Arrays.asList(TaskStsType.GENERATE_OUT.id, TaskStsType.AWAIT.id);
        List<Integer> has199Or200 = Arrays.asList(TaskStsType.WAVE_SEED.id, TaskStsType.UPDATED_OUT.id);
        for (Map.Entry<String, Set<Integer>> e : statusByBarcode.entrySet()) {
            Set<Integer> set = e.getValue();
            boolean hasHigh = set.stream().anyMatch(has199Or200::contains);
            boolean hasLow = set.stream().anyMatch(to199::contains);
            if (!hasHigh || !hasLow) continue;
            String barcode = e.getKey();
            boolean updated = taskService.update(new LambdaUpdateWrapper<Task>()
                    .eq(Task::getBarcode, barcode)
                    .in(Task::getTaskStatus, to199)
                    .set(Task::getTaskStatus, TaskStsType.WAVE_SEED.id));
            if (updated) {
                log.info("同料箱号{}下存在199/200且存在101/196，已将101/196任务自动更新为199", barcode);
            }
        }
    }

    /**
@@ -372,7 +432,7 @@

    /**
     * 每五秒校验深库位是否为空，如果浅库位有货，将浅库位移至深库位
     *///TODO 生成移库任务前，需要检查是否有任务前往当前托盘
     *///TODO 生成移库任务前，需要检查是否有任务前往当前料箱
//    @Scheduled(cron = "0/35 * * * * ?  ")
//    @Transactional(rollbackFor = Exception.class)
//    public void shallocToDeep() throws Exception {
@@ -587,96 +647,120 @@
        }

        tasks.forEach(task -> {
            TaskLog taskLog = new TaskLog();
            BeanUtils.copyProperties(task, taskLog);
            taskLog.setTaskId(task.getId()).setId(null);
            if (!taskLogService.save(taskLog)) {
                throw new CoolException("任务历史档保存失败！！");
            }
            List<TaskItem> taskItems = taskItemService.list(new LambdaQueryWrapper<TaskItem>().eq(TaskItem::getTaskId, task.getId()));

            if (task.getTaskType().equals(TaskType.TASK_TYPE_IN.type)) {
                for (TaskItem taskItem : taskItems) {
                    if (Objects.isNull(taskItem.getOrderId())) {
                        continue;
                    }
                    WkOrder order = asnOrderService.getById(taskItem.getOrderId());
                    if (Objects.isNull(order)) {
                        continue;
                    }
                    //入库单任务明细上报
                    WkOrderItem wkOrderItem = asnOrderItemService.getOne(new LambdaQueryWrapper<WkOrderItem>()
                            .eq(WkOrderItem::getOrderId, order.getId())
                            .eq(WkOrderItem::getFieldsIndex, taskItem.getFieldsIndex()));
                    if (Objects.isNull(wkOrderItem)) {
                        throw new CoolException("数据错误，单据明细不存在或已完成！！");
                    }
                    /**入库单明细上报*/
                    reportMsgService.reportOrderItem(wkOrderItem);
            // 只对出库 200 做同箱码检查：同箱码下若有 101/196/198/199 等（未到 200）则跳过，等全部 200 才一次处理；拣料入库 100 不受影响
            List<Task> toProcess = Collections.singletonList(task);
            if (TaskStsType.UPDATED_OUT.id.equals(task.getTaskStatus()) && TaskType.TASK_TYPE_PICK_AGAIN_OUT.type.equals(task.getTaskType()) && StringUtils.isNotBlank(task.getBarcode())) {
                long not200 = taskService.count(new LambdaQueryWrapper<Task>()
                        .eq(Task::getBarcode, task.getBarcode())
                        .ne(Task::getTaskStatus, TaskStsType.UPDATED_OUT.id));
                if (not200 > 0) {
                    return; // 同箱码尚有 101/196/198/199 等非 200 任务，不处理，继续等待
                }
            } else if ((task.getTaskType() >= TaskType.TASK_TYPE_OUT.type && task.getTaskType() <= TaskType.TASK_TYPE_EMPITY_OUT.type)
                    || task.getTaskType().equals(TaskType.TASK_TYPE_PICK_IN.type)) {
                /**判断单据是否完成**/
                // 只有波次类型的任务才需要查询波次关联单
                if (task.getResource() != null && task.getResource().equals(TaskResouceType.TASK_RESOUCE_WAVE_TYPE.val)) {
                // 同箱码已全部 200：一次性处理该箱码下所有 200 拣料出库（合计扣减、更新库存、生成一张拣料入库单、更新库位状态）
                List<Task> all200 = taskService.list(new LambdaQueryWrapper<Task>()
                        .eq(Task::getBarcode, task.getBarcode())
                        .eq(Task::getTaskType, TaskType.TASK_TYPE_PICK_AGAIN_OUT.type)
                        .eq(Task::getTaskStatus, TaskStsType.UPDATED_OUT.id)
                        .orderByAsc(Task::getId));
                if (!all200.isEmpty()) {
                    taskService.processPickOutBarcodeAll200(all200);
                    toProcess = all200;
                }
            }
            for (Task t : toProcess) {
                TaskLog taskLog = new TaskLog();
                BeanUtils.copyProperties(t, taskLog);
                taskLog.setTaskId(t.getId()).setId(null);
                if (!taskLogService.save(taskLog)) {
                    throw new CoolException("任务历史档保存失败！！");
                }
                List<TaskItem> taskItems = taskItemService.list(new LambdaQueryWrapper<TaskItem>().eq(TaskItem::getTaskId, t.getId()));

            // 上报ERP暂时注释（/rsf-open-api/erp/report/order）
            // if (task.getTaskType().equals(TaskType.TASK_TYPE_IN.type)) {
            //     for (TaskItem taskItem : taskItems) {
            //         if (Objects.isNull(taskItem.getOrderId())) {
            //             continue;
            //         }
            //         WkOrder order = asnOrderService.getById(taskItem.getOrderId());
            //         if (Objects.isNull(order)) {
            //             continue;
            //         }
            //         // 入库单任务明细上报：优先按 orderItemId（单据明细ID）查，否则按 orderId+matnrId(+fieldsIndex) 查，确保能找到单据明细
            //         WkOrderItem wkOrderItem = null;
            //         if (taskItem.getOrderItemId() != null) {
            //             wkOrderItem = asnOrderItemService.getById(taskItem.getOrderItemId());
            //         }
            //         if (wkOrderItem == null) {
            //             LambdaQueryWrapper<WkOrderItem> qw = new LambdaQueryWrapper<WkOrderItem>()
            //                     .eq(WkOrderItem::getOrderId, order.getId())
            //                     .eq(WkOrderItem::getMatnrId, taskItem.getMatnrId());
            //             if (StringUtils.isNotBlank(taskItem.getFieldsIndex())) {
            //                 qw.eq(WkOrderItem::getFieldsIndex, taskItem.getFieldsIndex());
            //             }
            //             wkOrderItem = asnOrderItemService.getOne(qw);
            //         }
            //         if (Objects.isNull(wkOrderItem)) {
            //             logger.warn("任务历史档处理：单据明细不存在或已完成，跳过上报 - taskId={}, orderId={}, orderItemId={}, matnrId={}, fieldsIndex={}", task.getId(), order.getId(), taskItem.getOrderItemId(), taskItem.getMatnrId(), taskItem.getFieldsIndex());
            //             continue;
            //         }
            //         /**入库单明细上报*/
            //         reportMsgService.reportOrderItem(wkOrderItem);
            //     }
            // } else
            if (t.getTaskType().equals(TaskType.TASK_TYPE_IN.type)) {
                // 入库类型仅转历史，不上报ERP（已注释）
            } else if ((t.getTaskType() >= TaskType.TASK_TYPE_OUT.type && t.getTaskType() <= TaskType.TASK_TYPE_EMPITY_OUT.type)
                    || t.getTaskType().equals(TaskType.TASK_TYPE_PICK_IN.type)) {
                /**判断单据是否完成：波次下发、按单下发（点击下发任务）完成后均将出库单置为完结*/
                Set<Long> orderIdsToDone = new HashSet<>();
                if (t.getResource() != null && t.getResource().equals(TaskResouceType.TASK_RESOUCE_WAVE_TYPE.val)) {
                    Set<Long> longSet = taskItems.stream()
                            .map(TaskItem::getSourceId)
                            .filter(Objects::nonNull)
                            .collect(Collectors.toSet());
                    
                    if (longSet.isEmpty()) {
                        logger.warn("任务{}的任务明细中没有有效的sourceId，跳过波次关联单查询。任务编码：{}，任务类型：{}", 
                                task.getId(), task.getTaskCode(), task.getTaskType());
                    } else {
                    if (!longSet.isEmpty()) {
                        List<WaveOrderRela> waveOrderRelas = waveOrderRelaService.list(new LambdaQueryWrapper<WaveOrderRela>()
                                .in(WaveOrderRela::getWaveId, longSet));
                        if (Cools.isEmpty(waveOrderRelas)) {
                            logger.warn("任务{}的波次对应关联单未找到，可能是数据不一致或任务不是通过波次创建的。任务编码：{}，sourceIds：{}", 
                                    task.getId(), task.getTaskCode(), longSet);
                        } else {
                            Set<Long> orderIds = waveOrderRelas.stream().map(WaveOrderRela::getOrderId).collect(Collectors.toSet());
                            List<WkOrder> wkOrders = asnOrderService.listByIds(orderIds);
                            if (wkOrders.isEmpty()) {
                                logger.warn("任务{}的关联单据不存在。任务编码：{}，orderIds：{}", 
                                        task.getId(), task.getTaskCode(), orderIds);
                            } else {
                                Config allowChang = configService.getOne(new LambdaQueryWrapper<Config>().eq(Config::getFlag, GlobalConfigCode.ALLOW_OVER_CHANGE));

                                wkOrders.forEach(order -> {
                                    //判断是否允许超收，不允许超收添加拒收判断
                                    if (!Objects.isNull(allowChang)) {
                                        if (!Boolean.parseBoolean(allowChang.getVal())) {
                                            if (order.getAnfme().compareTo(order.getQty()) == 0) {
                                                order.setExceStatus(AsnExceStatus.OUT_STOCK_STATUS_TASK_DONE.val);
                                                if (!asnOrderService.updateById(order)) {
                                                    logger.error("出库单更新状态失败。订单ID：{}，订单编码：{}", order.getId(), order.getCode());
                                                }
                                            }
                                        } else {
                                            if (order.getAnfme().compareTo(order.getQty()) <= 0) {
                                                order.setExceStatus(AsnExceStatus.OUT_STOCK_STATUS_TASK_DONE.val);
                                                if (!asnOrderService.updateById(order)) {
                                                    logger.error("出库单更新状态失败。订单ID：{}，订单编码：{}", order.getId(), order.getCode());
                                                }
                                            }
                                        }
                                    }
                                    //检查单据是否完成

                                });
                            }
                        if (!Cools.isEmpty(waveOrderRelas)) {
                            orderIdsToDone.addAll(waveOrderRelas.stream().map(WaveOrderRela::getOrderId).collect(Collectors.toSet()));
                        }
                    }
                } else {
                    logger.debug("任务{}不是波次类型任务（资源类型：{}），跳过波次关联单查询。任务编码：{}", 
                            task.getId(), task.getResource(), task.getTaskCode());
                } else if (t.getResource() != null && t.getResource().equals(TaskResouceType.TASK_RESOUCE_ORDER_TYPE.val)) {
                    // 按单下发：任务明细 sourceId 为出库单ID
                    Set<Long> ids = taskItems.stream()
                            .map(TaskItem::getSourceId)
                            .filter(Objects::nonNull)
                            .collect(Collectors.toSet());
                    orderIdsToDone.addAll(ids);
                }
                if (!orderIdsToDone.isEmpty()) {
                    List<WkOrder> wkOrders = asnOrderService.listByIds(orderIdsToDone);
                    if (!wkOrders.isEmpty()) {
                        Config allowChang = configService.getOne(new LambdaQueryWrapper<Config>().eq(Config::getFlag, GlobalConfigCode.ALLOW_OVER_CHANGE));
                        wkOrders.forEach(order -> {
                            if (order.getAnfme() == null) return;
                            boolean canDone = Boolean.TRUE.equals(allowChang != null && Boolean.parseBoolean(allowChang.getVal()))
                                    ? (order.getQty() != null && order.getAnfme().compareTo(order.getQty()) <= 0)
                                    : (order.getQty() != null && order.getAnfme().compareTo(order.getQty()) == 0);
                            if (canDone) {
                                order.setExceStatus(AsnExceStatus.OUT_STOCK_STATUS_TASK_DONE.val);
                                if (order.getQty() == null || order.getQty().compareTo(0.0) == 0) {
                                    order.setQty(order.getWorkQty() != null ? order.getWorkQty() : 0.0);
                                }
                                if (!asnOrderService.updateById(order)) {
                                    logger.error("出库单更新状态失败。订单ID：{}，订单编码：{}", order.getId(), order.getCode());
                                }
                            }
                        });
                    }
                }
                
                //出库单上报RCS修改库位状态
                try {
                    reportStationStatus(task);
                    reportStationStatus(t);
                } catch (Exception e) {
                    logger.error("任务{}上报RCS修改库位状态失败。任务编码：{}", task.getId(), task.getTaskCode(), e);
                    logger.error("任务{}上报RCS修改库位状态失败。任务编码：{}", t.getId(), t.getTaskCode(), e);
                    // 不抛出异常，避免中断定时任务
                }
            }
@@ -686,16 +770,16 @@
                TaskItemLog itemLog = new TaskItemLog();
                BeanUtils.copyProperties(item, itemLog);
                itemLog.setId(null)
                        .setTaskId(task.getId())
                        .setTaskId(t.getId())
                        .setLogId(taskLog.getId())
                        .setTaskItemId(item.getId());
                itemLogs.add(itemLog);
            }

            locItemWorkingService.remove(new LambdaQueryWrapper<LocItemWorking>().eq(LocItemWorking::getTaskId, task.getId()));
            locItemWorkingService.remove(new LambdaQueryWrapper<LocItemWorking>().eq(LocItemWorking::getTaskId, t.getId()));


            if (!taskService.removeById(task.getId())) {
            if (!taskService.removeById(t.getId())) {
                throw new CoolException("原始任务删除失败！！");
            }

@@ -703,11 +787,12 @@
                if (!taskItemLogService.saveBatch(itemLogs)) {
                    throw new CoolException("任务明细历史档保存失败！！");
                }
                if (!taskItemService.remove(new LambdaQueryWrapper<TaskItem>().eq(TaskItem::getTaskId, task.getId()))) {
                if (!taskItemService.remove(new LambdaQueryWrapper<TaskItem>().eq(TaskItem::getTaskId, t.getId()))) {
                    throw new CoolException("原始任务明细删除失败！！");
                }
            }

            }
        });
    }