feat:常规任务通知定时任务

“mitao”

2026-03-02 7a394e5bf750464ef5cdfe4034847fe354fcfe20

feat:常规任务通知定时任务

 xp-robotaxi-account-boot/src/main/java/com/xiaopeng/account/controller/XpOperationsUserController.java

@@ -174,6 +174,19 @@
        return R.ok(listAll);
    }

    /**
     * 获取所有启用状态的运维用户（内部接口）
     * @return 启用状态的用户列表
     */
    @ApiOperation(value = "获取所有启用状态的运维用户", tags = {"内部接口"})
    @GetMapping("/inner/xpOperationsUser/listAllEnabled")
    public R<List<XpOperationsUser>> listAllEnabled() {
        List<XpOperationsUser> users = xpOperationsUserService.lambdaQuery()
                .eq(XpOperationsUser::getStatus, true)
                .list();
        return R.ok(users);
    }

    @ApiOperation(value = "运维人员管理新增", tags = {"管理后台-运维执行"})
    @PostMapping("/api/v1/operationsUser/add")
    public R<String> add(@RequestBody XpOperationsUserAddDTO dto) {

 xp-robotaxi-account-boot/src/main/java/com/xiaopeng/account/service/impl/XpOperationsUserServiceImpl.java

@@ -163,11 +163,15 @@
        }

        // 8. 查询角色信息
        List<XpOperationsUserWithRoleVO> userWithRoles = xpOperationsRoleMapper.selectUsersWithRoleByIds(Collections.singletonList(userId));
        List<XpOperationsUserWithRoleVO> userWithRoles = this.selectUsersWithRoleByIds(Collections.singletonList(userId));
        XpOperationsUserWithRoleVO operationsUserWithRoleVO = userWithRoles.get(0);
        if (CollUtil.isNotEmpty(operationsUserWithRoleVO.getRoles())) {
            List<XpOperationsUserProfileVO.RoleInfoVO> roleVOs = operationsUserWithRoleVO.getRoles().stream()
                    .map(item -> BeanUtil.copyProperties(item, XpOperationsUserProfileVO.RoleInfoVO.class))
                    .map(item -> {
                        XpOperationsUserProfileVO.RoleInfoVO roleInfoVO = BeanUtil.copyProperties(item, XpOperationsUserProfileVO.RoleInfoVO.class);
                        roleInfoVO.setRoleId(item.getId());
                        return roleInfoVO;
                    })
                    .collect(Collectors.toList());
            vo.setRoles(roleVOs);
        }

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/factory/account/XpOperationUserClientFallbackFactory.java

@@ -58,6 +58,12 @@
                return R.fail("根据角色ID获取运维人员列表失败:" + cause.getMessage());
            }

            @Override
            public R<List<XpOperationsUser>> listAllEnabled() {
                log.error("获取启用状态运维用户列表失败，原因：{}", cause.getMessage());
                return R.ok(new ArrayList<>());
            }

        };
    }
}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/feignClient/account/XpOperationUserClient.java

@@ -71,4 +71,11 @@
     */
    @GetMapping("/inner/xp-operations-role/getXpOperationsRoleByRoleId")
    R<List<XpOperationsRole>> getXpOperationsRoleByRoleId(@RequestParam("roleId") Long roleId);

    /**
     * 获取所有启用状态的运维用户
     * @return 启用状态的用户列表
     */
    @GetMapping("/inner/xpOperationsUser/listAllEnabled")
    R<List<XpOperationsUser>> listAllEnabled();
}

 xp-robotaxi-api/src/main/java/com/xiaopeng/api/model/maintenance/XpTask.java

@@ -87,6 +87,10 @@
    @ApiModelProperty(value = "任务时限结束时间")
    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss", timezone = "GMT+8")
    private LocalDateTime endTime;

    @ApiModelProperty(value = "是否已发送过期通知 0=未发送 1=已发送")
    private Integer expiredNoticeSent;

    @ApiModelProperty(value = "状态 0=待到达 1=待处理 2=已完成 3=已取消")
    private Integer status;


 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/constant/TaskScheduleConstants.java

@@ -88,6 +88,16 @@
    public static final Integer TASK_STATUS_CANCELLED = 3;

    /**
     * 任务过期通知发送状态 - 未发送
     */
    public static final Integer TASK_EXPIRED_NOTICE_NOT_SENT = 0;

    /**
     * 任务过期通知发送状态 - 已发送
     */
    public static final Integer TASK_EXPIRED_NOTICE_SENT = 1;

    /**
     * 任务规则角色类型 - 车辆日常检测任务接收角色
     */
    public static final Integer RULE_ROLE_TYPE_INSPECTION = 1;

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/mapper/XpTaskMapper.java

@@ -1,6 +1,5 @@
package com.xiaopeng.maintenance.mapper;


import com.baomidou.mybatisplus.core.mapper.BaseMapper;
import com.xiaopeng.api.model.maintenance.XpTask;
import com.xiaopeng.common.core.web.page.PageInfo;

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/schedule/InsuranceInspectionScheduleTask.java

@@ -344,10 +344,8 @@
            // 设置车辆信息
            task.setVehicleId(vehicle.getId());

            // 设置任务时限
            if (taskRule.getProcessingTimeLimit() != null) {
                task.setEndTime(LocalDateTime.now().plusMinutes(taskRule.getProcessingTimeLimit()));
            }
            // 设置任务时限 常规任务限时3天
            task.setEndTime(LocalDateTime.now().plusDays(3));

            // 设置任务状态为待到达
            task.setStatus(TaskScheduleConstants.TASK_STATUS_PENDING);
@@ -359,7 +357,6 @@
            // 设置创建信息
            task.setReportTime(LocalDateTime.now());
            task.setCreateBy(TaskScheduleConstants.SCHEDULE_CREATOR);

            tasksToSave.add(task);
        }


 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/schedule/RoutinePreparationScheduleTask.java

@@ -364,10 +364,8 @@
            // 设置车辆信息
            task.setVehicleId(vehicle.getId());

            // 设置任务时限
            if (taskRule.getProcessingTimeLimit() != null) {
                task.setEndTime(LocalDateTime.now().plusMinutes(taskRule.getProcessingTimeLimit()));
            }
            // 设置任务时限 常规任务限时3天
            task.setEndTime(LocalDateTime.now().plusDays(3));

            // 设置任务状态为待到达
            task.setStatus(TaskScheduleConstants.TASK_STATUS_PENDING);

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/schedule/RoutineTaskExpireCheckScheduleTask.java

New file
@@ -0,0 +1,279 @@
package com.xiaopeng.maintenance.schedule;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.xiaopeng.api.dto.pushMessage.AliPushMessageDTO;
import com.xiaopeng.api.feignClient.account.XpOperationUserClient;
import com.xiaopeng.api.feignClient.account.XpOperationUserNotificationClient;
import com.xiaopeng.api.feignClient.pushMessage.XpAliPushMessageClient;
import com.xiaopeng.api.model.account.XpOperationsUser;
import com.xiaopeng.api.model.account.XpOperationsUserNotification;
import com.xiaopeng.api.model.maintenance.XpTask;
import com.xiaopeng.common.core.constant.AliPushTypeConstant;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.enums.NotificationTypeEnum;
import com.xiaopeng.common.core.enums.TaskTypeEnum;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.maintenance.constant.TaskScheduleConstants;
import com.xiaopeng.maintenance.mapper.XpTaskMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 常规任务过期检查定时任务
 *
 * @author mitao
 * @since 2026-03-02
 */
@Component
public class RoutineTaskExpireCheckScheduleTask {

    private static final Logger log = LoggerFactory.getLogger(RoutineTaskExpireCheckScheduleTask.class);

    private static final String TASK_NAME = "常规任务过期检查定时任务";

    private static final String PUSH_TITLE = "日常任务过期提示";

    private static final String CONTENT_TEMPLATE = "你有一条%s任务已超过3天未处理，系统已按过期处理；";

    @Autowired
    private XpTaskMapper taskMapper;

    @Autowired
    private XpOperationUserClient xpOperationUserClient;

    @Autowired
    private XpOperationUserNotificationClient xpOperationUserNotificationClient;

    @Autowired
    private XpAliPushMessageClient xpAliPushMessageClient;

    /**
     * 每小时执行一次
     */
    @Scheduled(cron = "0 0 * * * ?")
    public void executeRoutineTaskExpireCheck() {
        log.info("========== 开始执行{} ==========", TASK_NAME);
        try {
            List<XpTask> tasks = queryExpiredRoutineTasks();
            if (CollectionUtils.isEmpty(tasks)) {
                log.info("未找到超过3天未处理的常规任务，跳过执行");
                return;
            }

            Map<Long, Set<Long>> taskUserIdsMap = buildTaskUserIdsMap(tasks);
            if (taskUserIdsMap.isEmpty()) {
                log.warn("过期任务未配置有效处理人，跳过执行");
                return;
            }

            Set<Long> allUserIds = taskUserIdsMap.values().stream()
                    .flatMap(Set::stream)
                    .collect(Collectors.toSet());
            Map<Long, XpOperationsUser> userMap = fetchOperationUsers(allUserIds);
            if (CollectionUtils.isEmpty(userMap)) {
                log.warn("未获取到运维人员信息，跳过执行");
                return;
            }

            int notifiedTaskCount = 0;
            int notificationCount = 0;
            int pushCount = 0;

            for (XpTask task : tasks) {
                try {
                    Set<Long> userIds = taskUserIdsMap.get(task.getId());
                    if (CollectionUtils.isEmpty(userIds)) {
                        continue;
                    }

                    List<XpOperationsUser> validUsers = userIds.stream()
                            .map(userMap::get)
                            .filter(Objects::nonNull)
                            .collect(Collectors.toList());
                    if (CollectionUtils.isEmpty(validUsers)) {
                        log.warn("任务ID={} 未匹配到有效运维人员，跳过通知", task.getId());
                        continue;
                    }

                    int affectedRows = markExpiredNoticeSent(task.getId());
                    if (affectedRows <= 0) {
                        continue;
                    }

                    String content = buildContent(task.getTaskTypeId());
                    String bodyJson = buildPushBodyJson(content, task);

                    List<XpOperationsUserNotification> notifications = new ArrayList<>();
                    for (XpOperationsUser user : validUsers) {
                        XpOperationsUserNotification notification = new XpOperationsUserNotification();
                        notification.setNotificationType(NotificationTypeEnum.TASK_EXPIRED.getCode());
                        notification.setNotificationContent(content);
                        notification.setOperationsUserId(user.getId());
                        notification.setDataId(task.getId());
                        notification.setReadStatus(0);
                        notification.setNotificationExt(bodyJson);
                        notifications.add(notification);

                        if (pushTaskExpiredNotification(user, bodyJson)) {
                            pushCount++;
                        }
                    }

                    if (!CollectionUtils.isEmpty(notifications)) {
                        xpOperationUserNotificationClient.addNotificationBatch(notifications);
                        notificationCount += notifications.size();
                    }

                    notifiedTaskCount++;
                } catch (Exception e) {
                    log.error("处理过期任务通知失败，任务ID={}", task.getId(), e);
                }
            }

            log.info("========== {} 执行完成，任务数: {}，通知记录数: {}，推送次数: {} ==========",
                    TASK_NAME, notifiedTaskCount, notificationCount, pushCount);
        } catch (Exception e) {
            log.error("{}执行失败", TASK_NAME, e);
        }
    }

    private List<XpTask> queryExpiredRoutineTasks() {
        return taskMapper.selectList(new LambdaQueryWrapper<XpTask>()
                .eq(XpTask::getTaskCategory, TaskScheduleConstants.TASK_CATEGORY_ROUTINE)
                .eq(XpTask::getStatus, TaskScheduleConstants.TASK_STATUS_PENDING)
                .lt(XpTask::getEndTime, LocalDateTime.now())
                .eq(XpTask::getExpiredNoticeSent, TaskScheduleConstants.TASK_EXPIRED_NOTICE_NOT_SENT));
    }

    private int markExpiredNoticeSent(Long taskId) {
        return taskMapper.update(null, new LambdaUpdateWrapper<XpTask>()
                .set(XpTask::getExpiredNoticeSent, TaskScheduleConstants.TASK_EXPIRED_NOTICE_SENT)
                .eq(XpTask::getId, taskId)
                .eq(XpTask::getTaskCategory, TaskScheduleConstants.TASK_CATEGORY_ROUTINE)
                .eq(XpTask::getStatus, TaskScheduleConstants.TASK_STATUS_PENDING)
                .lt(XpTask::getEndTime, LocalDateTime.now())
                .eq(XpTask::getExpiredNoticeSent, TaskScheduleConstants.TASK_EXPIRED_NOTICE_NOT_SENT));
    }

    private Map<Long, Set<Long>> buildTaskUserIdsMap(List<XpTask> tasks) {
        Map<Long, Set<Long>> taskUserIdsMap = new HashMap<>();
        for (XpTask task : tasks) {
            if (task == null || task.getId() == null) {
                continue;
            }
            Set<Long> userIds = parseUserIds(task.getHandleUserId());
            if (!CollectionUtils.isEmpty(userIds)) {
                taskUserIdsMap.put(task.getId(), userIds);
            }
        }
        return taskUserIdsMap;
    }

    private Map<Long, XpOperationsUser> fetchOperationUsers(Set<Long> userIds) {
        if (CollectionUtils.isEmpty(userIds)) {
            return Collections.emptyMap();
        }
        try {
            R<List<XpOperationsUser>> result = xpOperationUserClient.listByIds(new ArrayList<>(userIds));
            if (result == null || CollectionUtils.isEmpty(result.getData())) {
                return Collections.emptyMap();
            }
            return result.getData().stream()
                    .collect(Collectors.toMap(XpOperationsUser::getId, user -> user, (a, b) -> a));
        } catch (Exception e) {
            log.error("获取运维人员信息失败", e);
            return Collections.emptyMap();
        }
    }

    private Set<Long> parseUserIds(String handleUserId) {
        if (StringUtils.isBlank(handleUserId)) {
            return Collections.emptySet();
        }
        String[] parts = handleUserId.split(",");
        Set<Long> userIds = new HashSet<>();
        for (String part : parts) {
            if (StringUtils.isBlank(part)) {
                continue;
            }
            try {
                userIds.add(Long.valueOf(part.trim()));
            } catch (NumberFormatException e) {
                log.warn("任务接收人ID非法，handleUserId={}", handleUserId);
            }
        }
        return userIds;
    }

    private String buildContent(String taskTypeId) {
        String taskTypeName = resolveTaskTypeName(taskTypeId);
        return String.format(CONTENT_TEMPLATE, taskTypeName);
    }

    private String resolveTaskTypeName(String taskTypeId) {
        Integer taskTypeCode = parseTaskTypeCode(taskTypeId);
        if (taskTypeCode == null) {
            return "常规";
        }
        String taskTypeName = TaskTypeEnum.getEnumDescByCode(taskTypeCode);
        return StringUtils.isBlank(taskTypeName) ? "常规" : taskTypeName;
    }

    private Integer parseTaskTypeCode(String taskTypeId) {
        if (StringUtils.isBlank(taskTypeId)) {
            return null;
        }
        try {
            return Integer.valueOf(taskTypeId.trim());
        } catch (NumberFormatException e) {
            return null;
        }
    }

    private String buildPushBodyJson(String content, XpTask task) {
        Map<String, Object> param = new HashMap<>();
        Integer taskTypeCode = parseTaskTypeCode(task.getTaskTypeId());
        param.put("content", content);
        param.put("strategy", NotificationTypeEnum.TASK_EXPIRED.getCode());
        param.put("type", taskTypeCode == null ? 0 : taskTypeCode);
        param.put("id", task.getId() == null ? "0" : task.getId().toString());
        return JSONObject.toJSONString(param);
    }

    private boolean pushTaskExpiredNotification(XpOperationsUser user, String bodyJson) {
        if (user.getDeviceType() == null || StringUtils.isBlank(user.getPhoneNumber())) {
            return false;
        }
        AliPushMessageDTO pushDTO = AliPushMessageDTO.builder()
                .deviceType(user.getDeviceType())
                .account(user.getPhoneNumber())
                .title(PUSH_TITLE)
                .body(bodyJson)
                .pushType(AliPushTypeConstant.MESSAGE)
                .build();
        try {
            xpAliPushMessageClient.pushMessage(pushDTO);
            return true;
        } catch (Exception e) {
            log.error("过期任务推送失败，运维用户ID={}", user.getId(), e);
            return false;
        }
    }
}

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/schedule/RoutineTaskNotificationScheduleTask.java

New file
@@ -0,0 +1,249 @@
package com.xiaopeng.maintenance.schedule;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.xiaopeng.api.dto.pushMessage.AliPushMessageDTO;
import com.xiaopeng.api.feignClient.account.XpOperationUserClient;
import com.xiaopeng.api.feignClient.account.XpOperationUserNotificationClient;
import com.xiaopeng.api.feignClient.pushMessage.XpAliPushMessageClient;
import com.xiaopeng.api.model.account.XpOperationsUser;
import com.xiaopeng.api.model.account.XpOperationsUserNotification;
import com.xiaopeng.api.model.maintenance.XpTask;
import com.xiaopeng.common.core.constant.AliPushTypeConstant;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.enums.NotificationTypeEnum;
import com.xiaopeng.common.core.utils.StringUtils;
import com.xiaopeng.maintenance.constant.TaskScheduleConstants;
import com.xiaopeng.maintenance.mapper.XpTaskMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

/**
 * 常规任务通知定时任务
 * <p>
 * 每天 9:00 / 12:00 / 18:00 执行，查询常规待处理任务，
 * 通知相关运维人员（handleUserId）
 * </p>
 *
 * @author mitao
 * @since 2026-03-02
 */
@Component
public class RoutineTaskNotificationScheduleTask {

    private static final Logger log = LoggerFactory.getLogger(RoutineTaskNotificationScheduleTask.class);

    /**
     * 任务名称
     */
    private static final String TASK_NAME = "常规任务通知定时任务";

    /**
     * 推送标题
     */
    private static final String PUSH_TITLE = "常规任务提醒";

    /**
     * 通知内容模板
     */
    private static final String CONTENT_TEMPLATE = "你目前剩余%s条待办任务，请及时处理！";

    @Autowired
    private XpTaskMapper taskMapper;

    @Autowired
    private XpOperationUserClient xpOperationUserClient;

    @Autowired
    private XpOperationUserNotificationClient xpOperationUserNotificationClient;

    @Autowired
    private XpAliPushMessageClient xpAliPushMessageClient;

    /**
     * 每天 9:00 / 12:00 / 18:00 执行
     */
    @Scheduled(cron = "0 0 9,12,18 * * ?")
    public void executeRoutineTaskNotification() {
        log.info("========== 开始执行{} ==========", TASK_NAME);

        try {
            List<XpTask> tasks = queryPendingRoutineTasks();
            if (CollectionUtils.isEmpty(tasks)) {
                log.info("未找到待处理常规任务，跳过通知");
                return;
            }

            Map<Long, Integer> userTaskCountMap = buildUserTaskCountMap(tasks);
            if (userTaskCountMap.isEmpty()) {
                log.warn("任务接收人为空，跳过通知");
                return;
            }

            Set<Long> allUserIds = userTaskCountMap.keySet();

            Map<Long, XpOperationsUser> userMap = fetchOperationUsers(allUserIds);
            if (CollectionUtils.isEmpty(userMap)) {
                log.warn("未获取到运维人员信息，跳过通知");
                return;
            }

            List<XpOperationsUserNotification> notifications = new ArrayList<>();
            int pushCount = 0;

            for (Map.Entry<Long, Integer> entry : userTaskCountMap.entrySet()) {
                Long userId = entry.getKey();
                Integer taskCount = entry.getValue();
                if (taskCount == null || taskCount <= 0) {
                    continue;
                }

                XpOperationsUser user = userMap.get(userId);
                if (user == null) {
                    continue;
                }

                String content = buildContent(taskCount);
                String bodyJson = buildPushBodyJson(content);

                XpOperationsUserNotification notification = new XpOperationsUserNotification();
                notification.setNotificationType(NotificationTypeEnum.TASK_NEW.getCode());
                notification.setNotificationContent(content);
                notification.setOperationsUserId(userId);
                notification.setDataId(0L);
                notification.setReadStatus(0);
                notification.setNotificationExt(bodyJson);
                notifications.add(notification);

                if (pushTaskNotification(user, bodyJson)) {
                    pushCount++;
                }
            }

            if (!CollectionUtils.isEmpty(notifications)) {
                xpOperationUserNotificationClient.addNotificationBatch(notifications);
                log.info("常规任务通知记录保存成功，共 {} 条", notifications.size());
            }

            log.info("========== {} 执行完成，推送次数: {} ==========", TASK_NAME, pushCount);
        } catch (Exception e) {
            log.error("{}执行失败", TASK_NAME, e);
        }
    }

    /**
     * 查询待处理常规任务
     */
    private List<XpTask> queryPendingRoutineTasks() {
        return taskMapper.selectList(new LambdaQueryWrapper<XpTask>()
                .eq(XpTask::getTaskCategory, TaskScheduleConstants.TASK_CATEGORY_ROUTINE)
                .eq(XpTask::getStatus, TaskScheduleConstants.TASK_STATUS_PENDING));
    }

    /**
     * 构建用户待办数量映射
     */
    private Map<Long, Integer> buildUserTaskCountMap(List<XpTask> tasks) {
        Map<Long, Integer> userTaskCountMap = new HashMap<>();
        for (XpTask task : tasks) {
            Set<Long> userIds = parseUserIds(task.getHandleUserId());
            if (!CollectionUtils.isEmpty(userIds)) {
                for (Long userId : userIds) {
                    userTaskCountMap.put(userId, userTaskCountMap.getOrDefault(userId, 0) + 1);
                }
            }
        }
        return userTaskCountMap;
    }

    /**
     * 获取运维人员信息
     */
    private Map<Long, XpOperationsUser> fetchOperationUsers(Set<Long> userIds) {
        if (CollectionUtils.isEmpty(userIds)) {
            return Collections.emptyMap();
        }
        try {
            R<List<XpOperationsUser>> result = xpOperationUserClient.listByIds(new ArrayList<>(userIds));
            if (result == null || CollectionUtils.isEmpty(result.getData())) {
                return Collections.emptyMap();
            }
            return result.getData().stream()
                    .collect(Collectors.toMap(XpOperationsUser::getId, user -> user, (a, b) -> a));
        } catch (Exception e) {
            log.error("获取运维人员信息失败", e);
            return Collections.emptyMap();
        }
    }

    /**
     * 解析 handleUserId
     */
    private Set<Long> parseUserIds(String handleUserId) {
        if (StringUtils.isBlank(handleUserId)) {
            return Collections.emptySet();
        }
        String[] parts = handleUserId.split(",");
        Set<Long> userIds = new HashSet<>();
        for (String part : parts) {
            if (StringUtils.isBlank(part)) {
                continue;
            }
            try {
                userIds.add(Long.valueOf(part.trim()));
            } catch (NumberFormatException e) {
                log.warn("任务接收人ID非法，handleUserId={}", handleUserId);
            }
        }
        return userIds;
    }

    /**
     * 构建通知内容
     */
    private String buildContent(Integer taskCount) {
        return String.format(CONTENT_TEMPLATE, String.valueOf(taskCount));
    }

    /**
     * 构建推送 body JSON
     */
    private String buildPushBodyJson(String content) {
        Map<String, Object> param = new HashMap<>();
        param.put("content", content);
        param.put("strategy", NotificationTypeEnum.TASK_NEW.getCode());
        param.put("type", 0);
        param.put("id", 0);
        return JSONObject.toJSONString(param);
    }

    /**
     * 推送任务提醒
     */
    private boolean pushTaskNotification(XpOperationsUser user, String bodyJson) {
        if (user.getDeviceType() == null || StringUtils.isBlank(user.getPhoneNumber())) {
            return false;
        }
        AliPushMessageDTO pushDTO = AliPushMessageDTO.builder()
                .deviceType(user.getDeviceType())
                .account(user.getPhoneNumber())
                .title(PUSH_TITLE)
                .body(bodyJson)
                .pushType(AliPushTypeConstant.MESSAGE)
                .build();
        try {
            xpAliPushMessageClient.pushMessage(pushDTO);
            return true;
        } catch (Exception e) {
            log.error("常规任务推送失败，运维用户ID={}", user.getId(), e);
            return false;
        }
    }
}

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/schedule/VehicleCleanScheduleTask.java

@@ -364,10 +364,8 @@
            // 设置车辆信息
            task.setVehicleId(vehicle.getId());

            // 设置任务时限
            if (taskRule.getProcessingTimeLimit() != null) {
                task.setEndTime(LocalDateTime.now().plusMinutes(taskRule.getProcessingTimeLimit()));
            }
            // 设置任务时限 常规任务限时3天
            task.setEndTime(LocalDateTime.now().plusDays(3));

            // 设置任务状态为待到达
            task.setStatus(TaskScheduleConstants.TASK_STATUS_PENDING);

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/schedule/VehicleInspectionScheduleTask.java

@@ -324,10 +324,8 @@
            // 设置车辆信息
            task.setVehicleId(vehicle.getId());

            // 设置任务时限
            if (taskRule.getProcessingTimeLimit() != null) {
                task.setEndTime(LocalDateTime.now().plusMinutes(taskRule.getProcessingTimeLimit()));
            }
            // 设置任务时限 常规任务限时3天
            task.setEndTime(LocalDateTime.now().plusDays(3));

            // 设置任务状态为待到达
            task.setStatus(TaskScheduleConstants.TASK_STATUS_PENDING);

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/schedule/VehicleMaintenanceScheduleTask.java

@@ -350,10 +350,8 @@
            // 设置车辆信息
            task.setVehicleId(vehicle.getId());

            // 设置任务时限
            if (taskRule.getProcessingTimeLimit() != null) {
                task.setEndTime(LocalDateTime.now().plusMinutes(taskRule.getProcessingTimeLimit()));
            }
            // 设置任务时限 常规任务限时3天
            task.setEndTime(LocalDateTime.now().plusDays(3));

            // 设置任务状态为待到达
            task.setStatus(TaskScheduleConstants.TASK_STATUS_PENDING);

 xp-robotaxi-maintenance-boot/src/main/java/com/xiaopeng/maintenance/service/impl/XpTaskServiceImpl.java

@@ -41,6 +41,7 @@
import com.xiaopeng.maintenance.service.IXpTaskRuleService;
import com.xiaopeng.maintenance.service.XpTaskService;
import com.xiaopeng.maintenance.utils.UUIDUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
@@ -60,6 +61,8 @@
import java.time.LocalTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@@ -71,6 +74,7 @@
 * @author WuGuanFengYue
 * @since 2026-02-03
 */
@Slf4j
@Service
public class XpTaskServiceImpl extends ServiceImpl<XpTaskMapper, XpTask> implements XpTaskService {
    @Autowired
@@ -223,33 +227,84 @@
        xpTask.setEndTime(endTime);
        this.save(xpTask);

        // 发送消息通知
        List<Long> userIds = Arrays.stream(xpTask.getHandleUserId().split(",")).map(Long::valueOf).collect(Collectors.toList());
        if (CollUtil.isNotEmpty(userIds)) {
            R<List<XpOperationsUser>> xpOperationUserResult = xpOperationUserClient.listByIds(userIds);
            List<XpOperationsUser> data = xpOperationUserResult.getData();
            if (CollUtil.isNotEmpty(data)) {
        // 异步发送消息通知
        String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
        sendAccidentNotificationAsync(xpTask, plateNumber, taskRule.getProcessingTimeLimit());
    }

                String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
                // 车辆${车牌号}在${车辆位置信息}发生了紧急事故，请在${到达现场限时}内到达现场！${APP链接}
                String content = String.format("车辆%s在%s发生了紧急事故，请在%d分钟内到达现场！", plateNumber, xpTask.getVehicleLocation(), taskRule.getProcessingTimeLimit());
    /**
     * 异步发送事故通知
     * @param xpTask 任务信息
     * @param plateNumber 车牌号
     * @param processingTimeLimit 处理时限（分钟）
     */
    private void sendAccidentNotificationAsync(XpTask xpTask, String plateNumber, Integer processingTimeLimit) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()
        );

        executor.execute(() -> {
            try {
                List<Long> userIds = Arrays.stream(xpTask.getHandleUserId().split(",")).map(Long::valueOf).collect(Collectors.toList());
                if (CollUtil.isEmpty(userIds)) {
                    return;
                }

                R<List<XpOperationsUser>> xpOperationUserResult = xpOperationUserClient.listByIds(userIds);
                List<XpOperationsUser> data = xpOperationUserResult.getData();
                if (CollUtil.isEmpty(data)) {
                    return;
                }

                // 构建通知内容
                String content = String.format("车辆%s在%s发生了紧急事故，请在%d分钟内到达现场！", plateNumber, xpTask.getVehicleLocation(), processingTimeLimit);
                String bodyJson = buildAccidentPushBodyJson(content, xpTask);

                // 构建通知记录（包含 notificationExt）
                List<XpOperationsUserNotification> notifications = data.stream().map(item -> {
                    XpOperationsUserNotification notification = new XpOperationsUserNotification();
                    notification.setOperationsUserId(item.getId());
                    notification.setNotificationType(NotificationTypeEnum.ACCIDENT_NEW.getCode());
                    notification.setNotificationContent(content);
                    notification.setDataId(xpTask.getId());
                    notification.setReadStatus(0);
                    notification.setCreateTime(LocalDateTime.now());
                    notification.setNotificationExt(bodyJson);
                    return notification;
                }).collect(Collectors.toList());

                // 批量保存通知记录
                xpOperationUserNotificationClient.addNotificationBatch(notifications);
                // 发送阿里云推送
                Map<Integer, List<XpOperationsUser>> operationUserMap = data.stream().collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
                for (Map.Entry<Integer, List<XpOperationsUser>> integerListEntry : operationUserMap.entrySet()) {
                    List<XpOperationsUser> value = integerListEntry.getValue();
                    pushAliMessage(value, content, xpTask);
                log.info("事故通知记录保存成功，共 {} 条", notifications.size());

                // 按设备类型分组发送阿里云推送
                Map<Integer, List<XpOperationsUser>> operationUserMap = data.stream()
                        .filter(u -> u.getDeviceType() != null && u.getPhoneNumber() != null)
                        .collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
                for (Map.Entry<Integer, List<XpOperationsUser>> entry : operationUserMap.entrySet()) {
                    pushAliMessage(entry.getValue(), content, xpTask);
                }
            } catch (Exception e) {
                log.error("发送事故通知失败，任务ID={}", xpTask.getId(), e);
            }
        }
        });

        executor.shutdown();
    }

    /**
     * 构建事故推送 body JSON
     * @param content 消息内容
     * @param xpTask 任务信息
     * @return JSON 字符串
     */
    private String buildAccidentPushBodyJson(String content, XpTask xpTask) {
        Map<String, Object> param = new HashMap<>();
        param.put("content", content);
        param.put("strategy", NotificationTypeEnum.ACCIDENT_NEW.getCode());
        param.put("type", TaskTypeNotificationEnum.ACCIDENT_TASK.getCode());
        param.put("id", xpTask.getId().toString());
        return JSONObject.toJSONString(param);
    }

    /**
@@ -1070,33 +1125,9 @@
        xpTask.setEndTime(endTime);
        this.save(xpTask);

        // 发送消息通知
        List<Long> userIds = Arrays.stream(xpTask.getHandleUserId().split(",")).map(Long::valueOf).collect(Collectors.toList());
        if (CollUtil.isNotEmpty(userIds)) {
            R<List<XpOperationsUser>> xpOperationUserResult = xpOperationUserClient.listByIds(userIds);
            List<XpOperationsUser> data = xpOperationUserResult.getData();
            if (CollUtil.isNotEmpty(data)) {

                String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
                // 车辆${车牌号}在${车辆位置信息}发生了紧急事故，请在${到达现场限时}内到达现场！${APP链接}
                String content = String.format("车辆%s在%s发生了紧急事故，请在%d分钟内到达现场！", plateNumber, xpTask.getVehicleLocation(), taskRule.getProcessingTimeLimit());
                List<XpOperationsUserNotification> notifications = data.stream().map(item -> {
                    XpOperationsUserNotification notification = new XpOperationsUserNotification();
                    notification.setOperationsUserId(item.getId());
                    notification.setNotificationContent(content);
                    notification.setReadStatus(0);
                    notification.setCreateTime(LocalDateTime.now());
                    return notification;
                }).collect(Collectors.toList());
                xpOperationUserNotificationClient.addNotificationBatch(notifications);
                // 发送阿里云推送
                Map<Integer, List<XpOperationsUser>> operationUserMap = data.stream().collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
                for (Map.Entry<Integer, List<XpOperationsUser>> integerListEntry : operationUserMap.entrySet()) {
                    List<XpOperationsUser> value = integerListEntry.getValue();
                    pushAliMessage(value, content, xpTask);
                }
            }
        }
        // 异步发送消息通知
        String plateNumber = Objects.nonNull(vehicle) ? vehicle.getPlateNumber() : "";
        sendAccidentNotificationAsync(xpTask, plateNumber, taskRule.getProcessingTimeLimit());
    }

    @Override

 xp-robotaxi-maintenance-boot/src/main/resources/mapper/maintenance/XpTaskMapper.xml

@@ -20,6 +20,8 @@
        <result column="assignee_time" property="assigneeTime"/>
        <result column="report_user_id" property="reportUserId"/>
        <result column="report_time" property="reportTime"/>
        <result column="end_time" property="endTime"/>
        <result column="expired_notice_sent" property="expiredNoticeSent"/>
        <result column="status" property="status"/>
        <result column="is_deleted" property="deleted"/>
        <result column="create_time" property="createTime"/>
@@ -31,7 +33,7 @@
    <!-- 通用查询结果列 -->
    <sql id="Base_Column_List">
        id
        , task_code, task_category,end_time, task_type_id,task_image, task_urgency_level, task_description, vehicle_location, vehicle_id, vehicle_lon, vehicle_lat, handle_user_id, handle_time, assignee_user_id, assignee_time, report_user_id, report_time, status, is_deleted, create_time, update_time, create_by, update_by
        , task_code, task_category, end_time, expired_notice_sent, task_type_id, task_image, task_urgency_level, task_description, vehicle_location, vehicle_id, vehicle_lon, vehicle_lat, handle_user_id, handle_time, assignee_user_id, assignee_time, report_user_id, report_time, status, is_deleted, create_time, update_time, create_by, update_by
    </sql>
    <select id="workbench" resultType="com.xiaopeng.maintenance.model.vo.WorkBenchVO">
        select t1.*,

 xp-robotaxi-operation-boot/src/main/java/com/xiaopeng/operation/service/impl/XpAgreementServiceImpl.java

@@ -2,16 +2,35 @@


import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xiaopeng.api.dto.pushMessage.AliPushMessageDTO;
import com.xiaopeng.api.feignClient.account.XpOperationUserClient;
import com.xiaopeng.api.feignClient.account.XpOperationUserNotificationClient;
import com.xiaopeng.api.feignClient.pushMessage.XpAliPushMessageClient;
import com.xiaopeng.api.model.account.XpOperationsUser;
import com.xiaopeng.api.model.account.XpOperationsUserNotification;
import com.xiaopeng.api.model.operation.XpAgreement;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.enums.NotificationTypeEnum;
import com.xiaopeng.operation.dto.XpAgreementDTO;
import com.xiaopeng.operation.mapper.XpAgreementMapper;
import com.xiaopeng.api.model.operation.XpAgreement;
import com.xiaopeng.operation.service.XpAgreementService;
import com.xiaopeng.operation.vo.XpAgreementVo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * <p>
@@ -21,8 +40,16 @@
 * @author liu
 * @since 2026-02-03
 */
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = @Lazy)
public class XpAgreementServiceImpl extends ServiceImpl<XpAgreementMapper, XpAgreement> implements XpAgreementService {

    private final XpOperationUserClient xpOperationUserClient;

    private final XpOperationUserNotificationClient xpOperationUserNotificationClient;

    private final XpAliPushMessageClient xpAliPushMessageClient;

    @Override
    public XpAgreementVo getContentByType(Integer type) {
@@ -77,5 +104,311 @@
            XpAgreement xpAgreement = BeanUtil.copyProperties(dto, XpAgreement.class);
            this.save(xpAgreement);
        }

        // 判断是否需要发送通知（用户服务协议=1 或 平台隐私政策=2）
        if (dto.getContentType() != null && (dto.getContentType() == 1 || dto.getContentType() == 2)) {
            // 发送协议更新提示通知（type=9）
            sendAgreementUpdateNotification(dto.getContentType());
            // 发送协议授权通知（type=10）
            sendAgreementAuthNotification(dto.getContentType());
        }
    }

    /**
     * 发送协议更新通知
     * @param contentType 协议类型
     */
    private void sendAgreementUpdateNotification(Integer contentType) {
        // 异步处理，避免阻塞主流程
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()
        );

        executor.execute(() -> {
            try {
                // 1. 获取所有启用状态的运维用户
                R<List<XpOperationsUser>> result = xpOperationUserClient.listAllEnabled();
                if (result == null || result.getCode() != 200 || result.getData() == null || result.getData().isEmpty()) {
                    log.warn("没有需要通知的启用状态用户");
                    return;
                }

                List<XpOperationsUser> enabledUsers = result.getData();
                log.info("协议更新通知，共需通知 {} 位用户", enabledUsers.size());

                // 2. 构建通知内容
                String agreementTypeName = getAgreementTypeName(contentType);
                String currentDate = LocalDate.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
                String message = String.format("尊敬的用户，你好！%s已于%s更新了内容，可前往个人中心-协议与政策查看详情", agreementTypeName, currentDate);
                String bodyJson = buildAgreementPushBodyJson(message);

                // 3. 构建并保存通知记录
                List<XpOperationsUserNotification> notifications = enabledUsers.stream()
                        .map(user -> buildAgreementNotification(message, bodyJson, user))
                        .collect(Collectors.toList());

                // 批量保存通知记录
                xpOperationUserNotificationClient.addNotificationBatch(notifications);
                log.info("协议更新通知记录保存成功，共 {} 条", notifications.size());

                // 4. 按设备类型分组并批量推送
                pushMessageByDeviceType(enabledUsers, message);

            } catch (Exception e) {
                log.error("发送协议更新通知失败", e);
            }
        });

        executor.shutdown();
    }

    /**
     * 获取协议类型名称
     * @param contentType 协议类型
     * @return 协议类型名称
     */
    private String getAgreementTypeName(Integer contentType) {
        if (contentType == 1) {
            return "用户服务协议";
        } else if (contentType == 2) {
            return "平台隐私政策";
        }
        return "协议";
    }

    /**
     * 构建推送 body JSON
     * @param message 消息内容
     * @return JSON 字符串
     */
    private String buildAgreementPushBodyJson(String message) {
        JSONObject body = new JSONObject();
        body.put("content", message);
        body.put("strategy", NotificationTypeEnum.POLICY_CONTENT_CHANGED.getCode()); // 9
        body.put("type", 0);
        body.put("id", 0);
        return body.toJSONString();
    }

    /**
     * 构建通知实体
     * @param message 消息内容
     * @param bodyJson 推送body内容
     * @param user 运维用户
     * @return 通知实体
     */
    private XpOperationsUserNotification buildAgreementNotification(String message, String bodyJson, XpOperationsUser user) {
        XpOperationsUserNotification notification = new XpOperationsUserNotification();
        notification.setNotificationType(NotificationTypeEnum.POLICY_CONTENT_CHANGED.getCode()); // 9
        notification.setNotificationContent(message);
        notification.setOperationsUserId(user.getId());
        notification.setDataId(0L);
        notification.setReadStatus(0); // 未读
        notification.setNotificationExt(bodyJson);
        return notification;
    }

    /**
     * 按设备类型分组推送消息
     * @param users 用户列表
     * @param message 消息内容
     */
    private void pushMessageByDeviceType(List<XpOperationsUser> users, String message) {
        // 按设备类型分组 (1=Android, 2=iOS)
        Map<Integer, List<XpOperationsUser>> usersByDeviceType = users.stream()
                .filter(u -> u.getDeviceType() != null && u.getPhoneNumber() != null)
                .collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));

        // 分别推送 Android 和 iOS
        for (Map.Entry<Integer, List<XpOperationsUser>> entry : usersByDeviceType.entrySet()) {
            Integer deviceType = entry.getKey();
            List<XpOperationsUser> deviceUsers = entry.getValue();

            // 按 1000 个手机号一批进行推送
            List<List<XpOperationsUser>> batches = partition(deviceUsers, 1000);

            for (List<XpOperationsUser> batch : batches) {
                String accounts = batch.stream()
                        .map(XpOperationsUser::getPhoneNumber)
                        .collect(Collectors.joining(","));

                AliPushMessageDTO pushDTO = buildPushMessageDTO(deviceType, accounts, message);

                try {
                    xpAliPushMessageClient.pushMessage(pushDTO);
                    log.info("协议更新推送成功，设备类型={}，用户数={}",
                            deviceType == 1 ? "Android" : "iOS", batch.size());
                } catch (Exception e) {
                    log.error("协议更新推送失败，设备类型={}", deviceType, e);
                }
            }
        }
    }

    /**
     * 构建推送消息 DTO
     * @param deviceType 设备类型
     * @param accounts 手机号（逗号分隔）
     * @param message 消息内容
     * @return 推送 DTO
     */
    private AliPushMessageDTO buildPushMessageDTO(Integer deviceType, String accounts, String message) {
        AliPushMessageDTO dto = new AliPushMessageDTO();
        dto.setDeviceType(deviceType);
        dto.setAccount(accounts);
        dto.setTitle("协议政策内容更新提示");
        dto.setBody(buildAgreementPushBodyJson(message));
        dto.setPushType("NOTICE");
        return dto;
    }

    /**
     * 将列表按指定大小分批
     * @param list 原始列表
     * @param batchSize 批次大小
     * @return 分批后的列表
     */
    private <T> List<List<T>> partition(List<T> list, int batchSize) {
        List<List<T>> batches = new ArrayList<>();
        for (int i = 0; i < list.size(); i += batchSize) {
            batches.add(list.subList(i, Math.min(i + batchSize, list.size())));
        }
        return batches;
    }

    /**
     * 发送协议更新授权通知
     * @param contentType 协议类型
     */
    private void sendAgreementAuthNotification(Integer contentType) {
        // 异步处理，避免阻塞主流程
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()
        );

        executor.execute(() -> {
            try {
                // 1. 获取所有启用状态的运维用户
                R<List<XpOperationsUser>> result = xpOperationUserClient.listAllEnabled();
                if (result == null || result.getCode() != 200 || result.getData() == null || result.getData().isEmpty()) {
                    log.warn("没有需要通知的启用状态用户");
                    return;
                }

                List<XpOperationsUser> enabledUsers = result.getData();
                log.info("协议授权通知，共需通知 {} 位用户", enabledUsers.size());

                // 2. 构建通知内容
                String agreementTypeName = getAgreementTypeName(contentType);
                String currentDate = LocalDate.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
                String message = String.format("尊敬的用户，你好！%s已于%s更新了内容，需要你前往授权", agreementTypeName, currentDate);
                String bodyJson = buildAuthPushBodyJson(message);

                // 3. 构建并保存通知记录
                List<XpOperationsUserNotification> notifications = enabledUsers.stream()
                        .map(user -> buildAuthNotification(message, bodyJson, user))
                        .collect(Collectors.toList());

                // 批量保存通知记录
                xpOperationUserNotificationClient.addNotificationBatch(notifications);
                log.info("协议授权通知记录保存成功，共 {} 条", notifications.size());

                // 4. 按设备类型分组并批量推送
                pushAuthMessageByDeviceType(enabledUsers, message);

            } catch (Exception e) {
                log.error("发送协议授权通知失败", e);
            }
        });

        executor.shutdown();
    }

    /**
     * 构建授权推送 body JSON
     * @param message 消息内容
     * @return JSON 字符串
     */
    private String buildAuthPushBodyJson(String message) {
        JSONObject body = new JSONObject();
        body.put("content", message);
        body.put("strategy", NotificationTypeEnum.AGREEMENT_CHANGED_NEED_AUTH.getCode()); // 10
        body.put("type", 0);
        body.put("id", 0);
        return body.toJSONString();
    }

    /**
     * 构建授权通知实体
     * @param message 消息内容
     * @param bodyJson 推送body内容
     * @param user 运维用户
     * @return 通知实体
     */
    private XpOperationsUserNotification buildAuthNotification(String message, String bodyJson, XpOperationsUser user) {
        XpOperationsUserNotification notification = new XpOperationsUserNotification();
        notification.setNotificationType(NotificationTypeEnum.AGREEMENT_CHANGED_NEED_AUTH.getCode()); // 10
        notification.setNotificationContent(message);
        notification.setOperationsUserId(user.getId());
        notification.setDataId(0L);
        notification.setReadStatus(0); // 未读
        notification.setNotificationExt(bodyJson);
        return notification;
    }

    /**
     * 按设备类型分组推送授权消息
     * @param users 用户列表
     * @param message 消息内容
     */
    private void pushAuthMessageByDeviceType(List<XpOperationsUser> users, String message) {
        // 按设备类型分组 (1=Android, 2=iOS)
        Map<Integer, List<XpOperationsUser>> usersByDeviceType = users.stream()
                .filter(u -> u.getDeviceType() != null && u.getPhoneNumber() != null)
                .collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));

        // 分别推送 Android 和 iOS
        for (Map.Entry<Integer, List<XpOperationsUser>> entry : usersByDeviceType.entrySet()) {
            Integer deviceType = entry.getKey();
            List<XpOperationsUser> deviceUsers = entry.getValue();

            // 按 1000 个手机号一批进行推送
            List<List<XpOperationsUser>> batches = partition(deviceUsers, 1000);

            for (List<XpOperationsUser> batch : batches) {
                String accounts = batch.stream()
                        .map(XpOperationsUser::getPhoneNumber)
                        .collect(Collectors.joining(","));

                AliPushMessageDTO pushDTO = buildAuthPushMessageDTO(deviceType, accounts, message);

                try {
                    xpAliPushMessageClient.pushMessage(pushDTO);
                    log.info("协议授权推送成功，设备类型={}，用户数={}",
                            deviceType == 1 ? "Android" : "iOS", batch.size());
                } catch (Exception e) {
                    log.error("协议授权推送失败，设备类型={}", deviceType, e);
                }
            }
        }
    }

    /**
     * 构建授权推送消息 DTO
     * @param deviceType 设备类型
     * @param accounts 手机号（逗号分隔）
     * @param message 消息内容
     * @return 推送 DTO
     */
    private AliPushMessageDTO buildAuthPushMessageDTO(Integer deviceType, String accounts, String message) {
        AliPushMessageDTO dto = new AliPushMessageDTO();
        dto.setDeviceType(deviceType);
        dto.setAccount(accounts);
        dto.setTitle("协议政策内容更新授权");
        dto.setBody(buildAuthPushBodyJson(message));
        dto.setPushType("NOTICE");
        return dto;
    }
}

 xp-robotaxi-operation-boot/src/main/java/com/xiaopeng/operation/service/impl/XpAppVersionServiceImpl.java

@@ -1,9 +1,18 @@
package com.xiaopeng.operation.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.xiaopeng.api.dto.pushMessage.AliPushMessageDTO;
import com.xiaopeng.api.feignClient.account.XpOperationUserClient;
import com.xiaopeng.api.feignClient.account.XpOperationUserNotificationClient;
import com.xiaopeng.api.feignClient.pushMessage.XpAliPushMessageClient;
import com.xiaopeng.api.model.account.XpOperationsUser;
import com.xiaopeng.api.model.account.XpOperationsUserNotification;
import com.xiaopeng.api.model.operation.XpAppVersion;
import com.xiaopeng.common.core.domain.R;
import com.xiaopeng.common.core.enums.NotificationTypeEnum;
import com.xiaopeng.common.core.exception.ServiceException;
import com.xiaopeng.common.core.web.page.BasePage;
import com.xiaopeng.common.core.web.page.PageInfo;
@@ -11,10 +20,20 @@
import com.xiaopeng.operation.mapper.XpAppVersionMapper;
import com.xiaopeng.operation.service.IXpAppVersionService;
import com.xiaopeng.operation.vo.XpAppVersionVO;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * App版本管理 服务实现类
@@ -24,7 +43,14 @@
 */
@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = @Lazy)
public class XpAppVersionServiceImpl extends ServiceImpl<XpAppVersionMapper, XpAppVersion> implements IXpAppVersionService {

    private final XpOperationUserClient xpOperationUserClient;

    private final XpOperationUserNotificationClient xpOperationUserNotificationClient;

    private final XpAliPushMessageClient xpAliPushMessageClient;

    @Override
    public XpAppVersionVO getLatestVersion() {
@@ -68,5 +94,155 @@
        appVersion.setCreateBy(SecurityUtils.getLoginUser().getSysUser().getNickName());
        appVersion.setVersionCode(latestVersionCode + 1); // 版本号自增
        save(appVersion);

        // 异步发送版本更新通知
        sendVersionUpdateNotification(appVersion);
    }

    /**
     * 发送版本更新通知
     * @param appVersion 版本信息
     */
    private void sendVersionUpdateNotification(XpAppVersion appVersion) {
        // 异步处理，避免阻塞主流程
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<Runnable>()
        );

        executor.execute(() -> {
            try {
                // 1. 获取所有启用状态的运维用户
                R<List<XpOperationsUser>> result = xpOperationUserClient.listAllEnabled();
                if (result == null || result.getCode() != 200 || result.getData() == null || result.getData().isEmpty()) {
                    log.warn("没有需要通知的启用状态用户");
                    return;
                }

                List<XpOperationsUser> enabledUsers = result.getData();
                log.info("版本更新通知，共需通知 {} 位用户", enabledUsers.size());

                // 2. 构建通知内容
                String message = "新版本上线啦！请前往更新。";
                String bodyJson = buildPushBodyJson(message);

                // 3. 构建并保存通知记录
                List<XpOperationsUserNotification> notifications = enabledUsers.stream()
                        .map(user -> buildNotification(message, bodyJson, user))
                        .collect(Collectors.toList());

                // 批量保存通知记录
                xpOperationUserNotificationClient.addNotificationBatch(notifications);
                log.info("版本更新通知记录保存成功，共 {} 条", notifications.size());

                // 4. 按设备类型分组并批量推送
                pushMessageByDeviceType(enabledUsers, message);

            } catch (Exception e) {
                log.error("发送版本更新通知失败", e);
            }
        });

        executor.shutdown();
    }

    /**
     * 构建推送 body JSON
     * @param message 消息内容
     * @return JSON 字符串
     */
    private String buildPushBodyJson(String message) {
        JSONObject body = new JSONObject();
        body.put("content", message);
        body.put("strategy", NotificationTypeEnum.VERSION_UPDATE_TIP.getCode()); // 8
        body.put("type", 0);
        body.put("id", 0);
        return body.toJSONString();
    }

    /**
     * 构建通知实体
     * @param message 消息内容
     * @param bodyJson 推送body内容
     * @param user 运维用户
     * @return 通知实体
     */
    private XpOperationsUserNotification buildNotification(String message, String bodyJson, XpOperationsUser user) {
        XpOperationsUserNotification notification = new XpOperationsUserNotification();
        notification.setNotificationType(NotificationTypeEnum.VERSION_UPDATE_TIP.getCode()); // 8
        notification.setNotificationContent(message);
        notification.setOperationsUserId(user.getId());
        notification.setDataId(0L); // 版本更新通知 dataId 设置为 0
        notification.setReadStatus(0); // 未读
        notification.setNotificationExt(bodyJson);
        return notification;
    }

    /**
     * 按设备类型分组推送消息
     * @param users 用户列表
     * @param message 消息内容
     */
    private void pushMessageByDeviceType(List<XpOperationsUser> users, String message) {
        // 按设备类型分组 (1=Android, 2=iOS)
        Map<Integer, List<XpOperationsUser>> usersByDeviceType = users.stream()
                .filter(u -> u.getDeviceType() != null && u.getPhoneNumber() != null)
                .collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));

        // 分别推送 Android 和 iOS
        for (Map.Entry<Integer, List<XpOperationsUser>> entry : usersByDeviceType.entrySet()) {
            Integer deviceType = entry.getKey();
            List<XpOperationsUser> deviceUsers = entry.getValue();

            // 按 1000 个手机号一批进行推送
            List<List<XpOperationsUser>> batches = partition(deviceUsers, 1000);

            for (List<XpOperationsUser> batch : batches) {
                String accounts = batch.stream()
                        .map(XpOperationsUser::getPhoneNumber)
                        .collect(Collectors.joining(","));

                AliPushMessageDTO pushDTO = buildPushMessageDTO(deviceType, accounts, message);

                try {
                    xpAliPushMessageClient.pushMessage(pushDTO);
                    log.info("版本更新推送成功，设备类型={}，用户数={}",
                            deviceType == 1 ? "Android" : "iOS", batch.size());
                } catch (Exception e) {
                    log.error("版本更新推送失败，设备类型={}", deviceType, e);
                }
            }
        }
    }

    /**
     * 构建推送消息 DTO
     * @param deviceType 设备类型
     * @param accounts 手机号（逗号分隔）
     * @param message 消息内容
     * @return 推送 DTO
     */
    private AliPushMessageDTO buildPushMessageDTO(Integer deviceType, String accounts, String message) {
        AliPushMessageDTO dto = new AliPushMessageDTO();
        dto.setDeviceType(deviceType);
        dto.setAccount(accounts);
        dto.setTitle("APP版本更新提示");
        dto.setBody(buildPushBodyJson(message));
        dto.setPushType("NOTICE");
        return dto;
    }

    /**
     * 将列表按指定大小分批
     * @param list 原始列表
     * @param batchSize 批次大小
     * @return 分批后的列表
     */
    private <T> List<List<T>> partition(List<T> list, int batchSize) {
        List<List<T>> batches = new ArrayList<>();
        for (int i = 0; i < list.size(); i += batchSize) {
            batches.add(list.subList(i, Math.min(i + batchSize, list.size())));
        }
        return batches;
    }
}