营销系统-手动推送

pig-marketing/pig-marketing-api/src/main/java/com/pig4cloud/pig/marketing/api/dto/MktMgmtHandPushQueryDTO.java

@@ -0,0 +1,36 @@
+package com.pig4cloud.pig.marketing.api.dto;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotNull;
+
+/**
+ * 手动推送查询DTO
+ * 
+ * @author pig4cloud
+ * @date 2025-01-20
+ * @description 手动推送记录查询参数
+ */
+@Data
+@Schema(description = "手动推送查询DTO")
+public class MktMgmtHandPushQueryDTO {
+
+    @Schema(description = "推送开始日期", example = "2025-01-20")
+    private String startTime;;
+
+	@Schema(description = "推送结束日期", example = "2025-01-20")
+	private String endTime;;
+
+    @Schema(description = "页码", example = "1")
+    @NotNull(message = "页码不能为空")
+    @Min(value = 1, message = "页码必须大于0")
+    private Integer pageNum = 1;
+
+    @Schema(description = "每页大小", example = "10")
+    @NotNull(message = "每页大小不能为空")
+    @Min(value = 1, message = "每页大小必须大于0")
+    private Integer pageSize = 10;
+}
+

pig-marketing/pig-marketing-api/src/main/java/com/pig4cloud/pig/marketing/api/dto/MktMgmtHandPushSaveDTO.java

@@ -0,0 +1,37 @@
+package com.pig4cloud.pig.marketing.api.dto;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+
+/**
+ * 手动推送保存DTO
+ * 
+ * @author pig4cloud
+ * @date 2025-01-20
+ * @description 手动推送记录保存参数
+ */
+@Data
+@Schema(description = "手动推送保存DTO")
+public class MktMgmtHandPushSaveDTO {
+
+    @Schema(description = "推送内容", required = true)
+    @NotBlank(message = "推送内容不能为空")
+    private String pushContent;
+
+    @Schema(description = "推送IP")
+    private String pushIP;
+
+    @Schema(description = "推送域名")
+    private String pushDomain;
+
+    @Schema(description = "规则ID", required = true)
+    @NotNull(message = "规则ID不能为空")
+    private Long ruleId;
+
+    @Schema(description = "推送频率")
+    private String pushFrequency;
+}
+

pig-marketing/pig-marketing-api/src/main/java/com/pig4cloud/pig/marketing/api/entity/MktMgmtPushRecord.java

@@ -61,6 +61,18 @@ public class MktMgmtPushRecord extends Model<MktMgmtPushRecord> {
 	@Schema(description = "推送方式")
 	private String pushAction;
 
+	/**
+	 * 延时推送时间
+	 */
+	@Schema(description = "延时推送")
+	private Integer delayPush;
+
+	/**
+	 * 推送类别（0-自动 1-手动）
+	 */
+	@Schema(description = "推送类别（0-自动 1-手动）")
+	private Integer autoPush;
+
 	/**
 	 * 推送类型
 	 */
@@ -79,6 +91,24 @@ public class MktMgmtPushRecord extends Model<MktMgmtPushRecord> {
 	@Schema(description = "推送详情")
 	private String pushDetail;
 
+	/**
+	 * 规则ID
+	 */
+	@Schema(description = "规则ID")
+	private Long ruleId;
+
+	/**
+	 * 推送IP
+	 */
+	@Schema(description = "推送IP")
+	private String pushIP;
+
+	/**
+	 * 推送域名
+	 */
+	@Schema(description = "推送域名")
+	private String pushDomain;
+
 	/**
 	 * 删除标记
 	 */

pig-marketing/pig-marketing-api/src/main/java/com/pig4cloud/pig/marketing/api/service/MktMgmtHandPushService.java

@@ -0,0 +1,33 @@
+package com.pig4cloud.pig.marketing.api.service;
+
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.pig4cloud.pig.marketing.api.dto.MktMgmtHandPushQueryDTO;
+import com.pig4cloud.pig.marketing.api.dto.MktMgmtPushRecordSaveDTO;
+import com.pig4cloud.pig.marketing.api.vo.rule.push.HandPushVO;
+
+/**
+ * 手动推送服务接口
+ * 
+ * @author pig4cloud
+ * @date 2025-01-20
+ * @description 手动推送相关业务接口
+ */
+public interface MktMgmtHandPushService {
+
+    /**
+     * 分页查询手动推送记录
+     * 
+     * @param queryDTO 查询参数
+     * @return 分页结果
+     */
+    Page<HandPushVO> pageQuery(MktMgmtHandPushQueryDTO queryDTO);
+
+    /**
+     * 手动推送
+     * 
+     * @param saveDTO 推送参数
+     * @return 推送结果
+     */
+    String handPush(MktMgmtPushRecordSaveDTO saveDTO);
+}
+

pig-marketing/pig-marketing-api/src/main/java/com/pig4cloud/pig/marketing/api/vo/mongo/OnlineUserVO.java

@@ -0,0 +1,30 @@
+package com.pig4cloud.pig.marketing.api.vo.mongo;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.time.LocalDateTime;
+
+/**
+ * @author: wcl
+ * @date: 2025-08-31
+ * @description: 在线用户VO
+ */
+@Data
+@Schema(description = "在线用户信息")
+public class OnlineUserVO implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    @Schema(description = "客户端ID", example = "client_123")
+    private String clientID;
+
+    @Schema(description = "消息内容", example = "用户活动数据")
+    private String msgData;
+	
+    @Schema(description = "最后活跃时间", example = "2025-08-31 10:30:00")
+    private LocalDateTime lastActiveTime;
+
+}
+

pig-marketing/pig-marketing-api/src/main/java/com/pig4cloud/pig/marketing/api/vo/rule/push/HandPushVO.java

@@ -0,0 +1,90 @@
+package com.pig4cloud.pig.marketing.api.vo.rule.push;
+
+
+import com.baomidou.mybatisplus.annotation.FieldFill;
+import com.baomidou.mybatisplus.annotation.TableField;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+/**
+ * @author wcl
+ * @date 2025/9/3 15:27
+ * @description: 手动推送VO
+ */
+@Data
+@Schema(description = "手动推送VO")
+public class HandPushVO {
+
+	/**
+	 * 推送内容
+	 */
+	@Schema(description = "推送内容")
+	private String pushContent;
+
+	/**
+	 * 推送频率
+	 */
+	@Schema(description = "推送频率")
+	private String pushFrequency;
+
+	/**
+	 * 推送状态
+	 */
+	@Schema(description = "推送状态")
+	private Boolean pushStatus;
+
+	/**
+	 * 推送方式
+	 */
+	@Schema(description = "推送方式")
+	private String pushAction;
+
+	/**
+	 * 延时推送时间
+	 */
+	@Schema(description = "延时推送")
+	private Integer delayPush;
+
+	/**
+	 * 推送类型
+	 */
+	@Schema(description = "推送类型")
+	private Boolean pushType;
+
+
+	/**
+	 * 推送详情
+	 */
+	@Schema(description = "推送详情")
+	private String pushDetail;
+
+	/**
+	 * 推送IP
+	 */
+	@Schema(description = "推送IP")
+	private String pushIP;
+
+	/**
+	 * 推送域名
+	 */
+	@Schema(description = "推送域名")
+	private String pushDomain;
+
+
+	/**
+	 * 创建时间
+	 */
+	@TableField(fill = FieldFill.INSERT)
+	@Schema(description = "创建时间")
+	private LocalDateTime createTime;
+
+	/**
+	 * 更新时间
+	 */
+	@TableField(fill = FieldFill.UPDATE)
+	@Schema(description = "更新时间")
+	private LocalDateTime updateTime;
+
+}

pig-marketing/pig-marketing-biz/src/main/java/com/pig4cloud/pig/marketing/controller/MktMgmtHandPushController.java

@@ -0,0 +1,73 @@
+package com.pig4cloud.pig.marketing.controller;
+
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.pig4cloud.pig.common.core.util.R;
+import com.pig4cloud.pig.marketing.api.dto.MktMgmtHandPushQueryDTO;
+import com.pig4cloud.pig.marketing.api.dto.MktMgmtPushRecordSaveDTO;
+import com.pig4cloud.pig.marketing.api.service.MktMgmtHandPushService;
+import com.pig4cloud.pig.marketing.api.vo.rule.push.HandPushVO;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.*;
+
+import jakarta.validation.Valid;
+
+/**
+ * 手动推送管理控制器
+ * 
+ * @author pig4cloud
+ * @date 2025-01-20
+ * @description 手动推送相关接口
+ */
+@Slf4j
+@RestController
+@RequestMapping("/handPush")
+@RequiredArgsConstructor
+@Tag(name = "手动推送管理", description = "手动推送相关接口")
+public class MktMgmtHandPushController {
+
+    private final MktMgmtHandPushService mktMgmtHandPushService;
+
+    /**
+     * 分页查询手动推送记录
+     * 
+     * @param queryDTO 查询参数
+     * @return 分页结果
+     */
+    @GetMapping("/page")
+    @Operation(summary = "分页查询手动推送记录", description = "根据条件分页查询手动推送记录")
+    public R<Page<HandPushVO>> pageQuery(@Valid MktMgmtHandPushQueryDTO queryDTO) {
+        try {
+            Page<HandPushVO> result = mktMgmtHandPushService.pageQuery(queryDTO);
+            return R.ok(result);
+        } catch (Exception e) {
+            log.error("分页查询手动推送记录异常", e);
+            return R.failed("查询失败：" + e.getMessage());
+        }
+    }
+
+    /**
+     * 手动推送
+     * 
+     * @param saveDTO 推送参数
+     * @return 推送结果
+     */
+    @PostMapping("/push")
+    @Operation(summary = "手动推送", description = "根据规则进行手动推送")
+    public R<String> handPush(@Valid @RequestBody MktMgmtPushRecordSaveDTO saveDTO) {
+        try {
+            String result = mktMgmtHandPushService.handPush(saveDTO);
+            if (result.contains("成功")) {
+                return R.ok(result);
+            } else {
+                return R.failed(result);
+            }
+        } catch (Exception e) {
+            log.error("手动推送异常", e);
+            return R.failed("推送失败：" + e.getMessage());
+        }
+    }
+}
+

pig-marketing/pig-marketing-biz/src/main/java/com/pig4cloud/pig/marketing/controller/TcpDataController.java

@@ -9,6 +9,7 @@ import com.pig4cloud.pig.marketing.api.dto.mongo.SaveDeviceInfoDTO;
 import com.pig4cloud.pig.marketing.api.dto.mongo.SaveTcpMessageDTO;
 import com.pig4cloud.pig.marketing.api.entity.mongo.Device;
 import com.pig4cloud.pig.marketing.api.entity.mongo.Message;
+import com.pig4cloud.pig.marketing.api.vo.mongo.OnlineUserVO;
 import com.pig4cloud.pig.marketing.api.vo.mongo.PageDeviceInfoVO;
 import com.pig4cloud.pig.marketing.api.vo.mongo.UserStatisticsVO;
 import com.pig4cloud.pig.marketing.service.TcpDataService;
@@ -23,6 +24,8 @@ import org.springdoc.core.annotations.ParameterObject;
 import org.springframework.http.HttpHeaders;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.List;
+
 /**
  * @author: lwh
  * @date: 2025-08-27
@@ -92,4 +95,17 @@ public class TcpDataController {
 			return R.failed("统计失败：" + e.getMessage());
 		}
 	}
+
+	@GetMapping("/online/users")
+	@Operation(summary = "查询在线用户列表", description = "查询最近10分钟内有活动的用户列表")
+	public R<List<OnlineUserVO>> getOnlineUsers() {
+		log.info("开始查询在线用户列表");
+		try {
+			List<OnlineUserVO> onlineUsers = tcpDataService.getOnlineUsers();
+			return R.ok(onlineUsers);
+		} catch (Exception e) {
+			log.error("查询在线用户列表异常", e);
+			return R.failed("查询失败：" + e.getMessage());
+		}
+	}
 }

pig-marketing/pig-marketing-biz/src/main/java/com/pig4cloud/pig/marketing/service/TcpDataService.java

@@ -10,6 +10,7 @@ import com.pig4cloud.pig.marketing.api.entity.mongo.Device;
 import com.pig4cloud.pig.marketing.api.entity.mongo.Message;
 import com.pig4cloud.pig.marketing.api.vo.mongo.PageDeviceInfoVO;
 import com.pig4cloud.pig.marketing.api.vo.mongo.UserStatisticsVO;
+import com.pig4cloud.pig.marketing.api.vo.mongo.OnlineUserVO;
 
 /**
  * @author: lwh
@@ -68,4 +69,10 @@ public interface TcpDataService {
 	 * @return 用户统计信息
 	 */
 	UserStatisticsVO getUserStatistics();
+
+	/**
+	 * 查询在线用户列表
+	 * @return 在线用户列表
+	 */
+	java.util.List<OnlineUserVO> getOnlineUsers();
 }

pig-marketing/pig-marketing-biz/src/main/java/com/pig4cloud/pig/marketing/service/impl/MktMgmtHandPushServiceImpl.java

@@ -0,0 +1,287 @@
+package com.pig4cloud.pig.marketing.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.pig4cloud.pig.marketing.api.dto.MktMgmtHandPushQueryDTO;
+import com.pig4cloud.pig.marketing.api.dto.MktMgmtPushRecordSaveDTO;
+import com.pig4cloud.pig.marketing.api.entity.MktMgmtPushRecord;
+import com.pig4cloud.pig.marketing.api.service.MktMgmtHandPushService;
+import com.pig4cloud.pig.marketing.api.dto.config.SaveGlobalRuleDTO;
+import com.pig4cloud.pig.marketing.service.MarketingConfigService;
+import com.pig4cloud.pig.marketing.api.vo.rule.push.HandPushVO;
+import com.pig4cloud.pig.marketing.mapper.MktMgmtPushRecordMapper;
+import com.pig4cloud.pig.marketing.util.PushFrequencyUtil;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * 手动推送服务实现类
+ * 
+ * @author pig4cloud
+ * @date 2025-01-20
+ * @description 手动推送相关业务实现
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class MktMgmtHandPushServiceImpl implements MktMgmtHandPushService {
+
+    private final MktMgmtPushRecordMapper mktMgmtPushRecordMapper;
+    private final MarketingConfigService marketingConfigService;
+    private final RedisTemplate<String, Object> redisTemplate;
+
+    @Override
+    public Page<HandPushVO> pageQuery(MktMgmtHandPushQueryDTO queryDTO) {
+        Page<MktMgmtPushRecord> page = new Page<>(queryDTO.getPageNum(), queryDTO.getPageSize());
+        
+        LambdaQueryWrapper<MktMgmtPushRecord> queryWrapper = new LambdaQueryWrapper<>();
+        queryWrapper.eq(MktMgmtPushRecord::getDelFlag, "0");
+        queryWrapper.eq(MktMgmtPushRecord::getAutoPush, 1);
+        
+        // 根据时间范围查询
+        if (StringUtils.hasText(queryDTO.getStartTime())) {
+            LocalDateTime startTime = LocalDateTime.parse(queryDTO.getStartTime() + " 00:00:00", 
+                DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
+            queryWrapper.ge(MktMgmtPushRecord::getCreateTime, startTime);
+        }
+        
+        if (StringUtils.hasText(queryDTO.getEndTime())) {
+            LocalDateTime endTime = LocalDateTime.parse(queryDTO.getEndTime() + " 23:59:59", 
+                DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
+            queryWrapper.le(MktMgmtPushRecord::getCreateTime, endTime);
+        }
+        
+        queryWrapper.orderByDesc(MktMgmtPushRecord::getCreateTime);
+        
+        Page<MktMgmtPushRecord> recordPage = mktMgmtPushRecordMapper.selectPage(page, queryWrapper);
+        
+        // 转换为HandPushVO
+        Page<HandPushVO> voPage = new Page<>(recordPage.getCurrent(), recordPage.getSize(), recordPage.getTotal());
+        List<HandPushVO> voList = recordPage.getRecords().stream()
+            .map(this::convertToHandPushVO)
+            .collect(Collectors.toList());
+        voPage.setRecords(voList);
+        
+        return voPage;
+    }
+
+    @Override
+    public String handPush(MktMgmtPushRecordSaveDTO saveDTO) {
+        try {
+            log.info("开始手动推送，推送内容：{}", saveDTO.getPushContent());
+            
+            // 1. 获取全局手动推送规则
+            SaveGlobalRuleDTO globalRule = marketingConfigService.getGlobalRule();
+            if (globalRule == null) {
+                return "获取全局手动推送规则失败";
+            }
+            
+            // 2. 校验IP
+            if (!validateGlobalRuleIP(globalRule.getIp(), saveDTO.getPushIP())) {
+                return "推送IP【" + saveDTO.getPushIP() + "】不在允许的IP范围内";
+            }
+            
+            // 3. 校验域名
+            if (!validateGlobalRuleDomain(globalRule.getDomain(), saveDTO.getPushDomain())) {
+                return "推送域名【" + saveDTO.getPushDomain() + "】不在允许的域名范围内";
+            }
+            
+            // 4. 校验推送频率
+            if (!PushFrequencyUtil.checkPushFrequency(globalRule.getPushFrequency(), 0L, saveDTO.getPushContent(), redisTemplate)) {
+                return "推送频率检查未通过，推送频率：" + globalRule.getPushFrequency();
+            }
+            
+            // 5. 构建推送记录
+            MktMgmtPushRecord pushRecord = buildHandPushRecord(saveDTO, globalRule);
+            
+            // 6. 保存推送记录
+            int result = mktMgmtPushRecordMapper.insert(pushRecord);
+            if (result > 0) {
+                log.info("手动推送成功，推送内容：{}", saveDTO.getPushContent());
+                return "手动推送成功";
+            } else {
+                return "手动推送失败，数据库保存异常";
+            }
+            
+        } catch (Exception e) {
+            log.error("手动推送异常，推送内容：{}", saveDTO.getPushContent(), e);
+            return "手动推送异常：" + e.getMessage();
+        }
+    }
+    
+    /**
+     * 校验全局规则IP
+     * 
+     * @param allowedIPs 允许的IP列表
+     * @param pushIP 推送IP
+     * @return 是否匹配
+     */
+    private boolean validateGlobalRuleIP(List<String> allowedIPs, String pushIP) {
+        if (allowedIPs == null || allowedIPs.isEmpty()) {
+            return true; // 如果没有配置IP限制，则允许
+        }
+        
+        if (pushIP == null || pushIP.trim().isEmpty()) {
+            return false; // 推送IP为空，不允许
+        }
+        
+        for (String allowedIP : allowedIPs) {
+            if (allowedIP.contains("/")) {
+                // IP段格式：192.168.10.101/192.168.10.110
+                String[] ipRange = allowedIP.split("/");
+                if (ipRange.length == 2) {
+                    String startIP = ipRange[0].trim();
+                    String endIP = ipRange[1].trim();
+                    if (isIPInRange(pushIP, startIP, endIP)) {
+                        return true;
+                    }
+                }
+            } else {
+                // 单个IP格式：192.168.10.102
+                if (pushIP.equals(allowedIP.trim())) {
+                    return true;
+                }
+            }
+        }
+        
+        return false;
+    }
+    
+    /**
+     * 校验全局规则域名
+     * 
+     * @param allowedDomains 允许的域名列表
+     * @param pushDomain 推送域名
+     * @return 是否匹配
+     */
+    private boolean validateGlobalRuleDomain(List<String> allowedDomains, String pushDomain) {
+        if (allowedDomains == null || allowedDomains.isEmpty()) {
+            return true; // 如果没有配置域名限制，则允许
+        }
+        
+        if (pushDomain == null || pushDomain.trim().isEmpty()) {
+            return false; // 推送域名为空，不允许
+        }
+        
+        for (String allowedDomain : allowedDomains) {
+            if (pushDomain.equals(allowedDomain.trim())) {
+                return true;
+            }
+        }
+        
+        return false;
+    }
+    
+    /**
+     * 判断IP是否在指定范围内
+     * 
+     * @param ip 要检查的IP
+     * @param startIP 起始IP
+     * @param endIP 结束IP
+     * @return 是否在范围内
+     */
+    private boolean isIPInRange(String ip, String startIP, String endIP) {
+        try {
+            long ipLong = ipToLong(ip);
+            long startIPLong = ipToLong(startIP);
+            long endIPLong = ipToLong(endIP);
+            return ipLong >= startIPLong && ipLong <= endIPLong;
+        } catch (Exception e) {
+            log.warn("IP范围检查异常，IP：{}，起始IP：{}，结束IP：{}", ip, startIP, endIP, e);
+            return false;
+        }
+    }
+    
+    /**
+     * 将IP地址转换为Long类型
+     * 
+     * @param ip IP地址
+     * @return Long值
+     */
+    private long ipToLong(String ip) {
+        String[] parts = ip.split("\\.");
+        if (parts.length != 4) {
+            throw new IllegalArgumentException("Invalid IP address: " + ip);
+        }
+        
+        long result = 0;
+        for (int i = 0; i < 4; i++) {
+            int part = Integer.parseInt(parts[i]);
+            if (part < 0 || part > 255) {
+                throw new IllegalArgumentException("Invalid IP address: " + ip);
+            }
+            result = result * 256 + part;
+        }
+        return result;
+    }
+    
+    /**
+     * 构建手动推送记录
+     * 
+     * @param saveDTO 保存参数
+     * @param globalRule 全局规则信息
+     * @return 推送记录
+     */
+    private MktMgmtPushRecord buildHandPushRecord(MktMgmtPushRecordSaveDTO saveDTO, SaveGlobalRuleDTO globalRule) {
+        MktMgmtPushRecord pushRecord = new MktMgmtPushRecord();
+        
+        // 基本信息
+        pushRecord.setRuleId(0L); // 全局规则ID为0
+        pushRecord.setRuleName("全局手动推送规则");
+        pushRecord.setPushContent(globalRule.getPushContent());
+        pushRecord.setPushIP(saveDTO.getPushIP());
+        pushRecord.setPushDomain(saveDTO.getPushDomain());
+        pushRecord.setAutoPush(1);
+        pushRecord.setPushStatus(true);
+        pushRecord.setCreateTime(LocalDateTime.now());
+        pushRecord.setUpdateTime(LocalDateTime.now());
+        
+        // 构建触发条件JSON
+        Map<String, Object> triggerCondition = new HashMap<>();
+        triggerCondition.put("type", "manual");
+        triggerCondition.put("ruleId", 0L);
+        triggerCondition.put("ruleName", "全局手动推送规则");
+        triggerCondition.put("pushFrequency", globalRule.getPushFrequency());
+        triggerCondition.put("allowedIPs", globalRule.getIp());
+        triggerCondition.put("allowedDomains", globalRule.getDomain());
+        pushRecord.setTriggerCondition(triggerCondition.toString());
+        
+        // 构建推送详情JSON
+        Map<String, Object> pushDetail = new HashMap<>();
+        pushDetail.put("pushType", "manual");
+        pushDetail.put("pushAction", globalRule.getAction());
+        pushDetail.put("pushContent", saveDTO.getPushContent());
+        pushDetail.put("pushIP", saveDTO.getPushIP());
+        pushDetail.put("pushDomain", saveDTO.getPushDomain());
+        pushDetail.put("pushTime", LocalDateTime.now().toString());
+        pushDetail.put("delayPush", globalRule.getDelayPush());
+        pushRecord.setPushDetail(pushDetail.toString());
+        
+        return pushRecord;
+    }
+    
+
+    
+    /**
+     * 转换为HandPushVO
+     * 
+     * @param record 推送记录
+     * @return HandPushVO
+     */
+    private HandPushVO convertToHandPushVO(MktMgmtPushRecord record) {
+        HandPushVO vo = new HandPushVO();
+        BeanUtils.copyProperties(record, vo);
+        return vo;
+    }
+}

pig-marketing/pig-marketing-biz/src/main/java/com/pig4cloud/pig/marketing/service/impl/MktMgmtPushRecordServiceImpl.java

@@ -7,9 +7,11 @@ import com.pig4cloud.pig.marketing.api.dto.MktMgmtPushRecordSaveDTO;
 import com.pig4cloud.pig.marketing.api.entity.*;
 import com.pig4cloud.pig.marketing.mapper.*;
 import com.pig4cloud.pig.marketing.service.MktMgmtPushRecordService;
+import com.pig4cloud.pig.marketing.util.PushFrequencyUtil;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.jetbrains.annotations.NotNull;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 
@@ -31,9 +33,9 @@ public class MktMgmtPushRecordServiceImpl implements MktMgmtPushRecordService {
 	private final MktMgmtPushRecordMapper mktMgmtPushRecordMapper;
 	private final MktMgmtKeywordMapper mktMgmtKeywordMapper;
 	private final MktMgmtRuleMapper mktMgmtRuleMapper;
-
 	private final MktMgmtRuleIpMapper mktMgmtRuleIpMapper;
 	private final MktMgmtRuleDomainMapper mktMgmtRuleDomainMapper;
+	private final RedisTemplate<String, Object> redisTemplate;
 
 	@Override
 	public Page<MktMgmtPushRecord> pageQuery(MktMgmtPushRecordQueryDTO queryDTO) {
@@ -50,6 +52,7 @@ public class MktMgmtPushRecordServiceImpl implements MktMgmtPushRecordService {
 		}
 
 		queryWrapper.eq(MktMgmtPushRecord::getDelFlag, "0");
+		queryWrapper.eq(MktMgmtPushRecord::getAutoPush, 0);
 		
 		queryWrapper.orderByDesc(MktMgmtPushRecord::getCreateTime);
 		
@@ -100,10 +103,19 @@ public class MktMgmtPushRecordServiceImpl implements MktMgmtPushRecordService {
 				ValidationResult validationResult = validateRuleIpAndDomainWithDetails(ruleId, saveDTO.getPushIP(), saveDTO.getPushDomain());
 				
 				if (validationResult.isValid()) {
-					selectedRule = rule;
-					selectedRuleId = ruleId;
-					log.info("找到匹配的规则：{}，匹配次数：{}", rule.getRuleName(), matchCount);
-					break;
+					// 验证推送频率
+					boolean shouldPush = PushFrequencyUtil.checkPushFrequency(rule.getPushFrequency(), rule.getId(), saveDTO.getPushContent(), redisTemplate);
+					if (shouldPush) {
+						selectedRule = rule;
+						selectedRuleId = ruleId;
+						log.info("找到匹配的规则：{}，匹配次数：{}，推送频率验证通过", rule.getRuleName(), matchCount);
+						break;
+					} else {
+						errorDetails.append("规则").append(ruleId).append("(").append(rule.getRuleName()).append(")");
+						errorDetails.append("匹配").append(matchCount).append("个关键字[").append(String.join(",", matchedKeywords)).append("]");
+						errorDetails.append("但推送频率验证失败；");
+						log.debug("规则{}的推送频率验证失败，尝试下一个规则", ruleId);
+					}
 				} else {
 					errorDetails.append("规则").append(ruleId).append("(").append(rule.getRuleName()).append(")");
 					errorDetails.append("匹配").append(matchCount).append("个关键字[").append(String.join(",", matchedKeywords)).append("]");
@@ -159,6 +171,8 @@ public class MktMgmtPushRecordServiceImpl implements MktMgmtPushRecordService {
 		record.setPushDetail(pushDetailJson);
 		record.setPushAction(selectedRule.getAction());
 		record.setPushType(selectedRule.getPushType());
+		record.setDelayPush(selectedRule.getDelayPush());
+		record.setAutoPush(0);
 		record.setDelFlag("0");
 		return record;
 	}
@@ -190,6 +204,7 @@ public class MktMgmtPushRecordServiceImpl implements MktMgmtPushRecordService {
 		// 1. 查询所有未删除的关键字，按规则ID分组
 		LambdaQueryWrapper<MktMgmtKeyword> queryWrapper = new LambdaQueryWrapper<>();
 		queryWrapper.eq(MktMgmtKeyword::getDelFlag, "0");
+		queryWrapper.ne(MktMgmtKeyword::getRuleId, "0");
 		List<MktMgmtKeyword> allKeywords = mktMgmtKeywordMapper.selectList(queryWrapper);
 		
 		// 按规则ID分组
@@ -408,4 +423,6 @@ public class MktMgmtPushRecordServiceImpl implements MktMgmtPushRecordService {
 		
 		return new ValidationResult(false, "推送域名 " + pushDomain + " 不匹配任何规则域名配置：" + configuredDomains);
 	}
+	
+
 }

pig-marketing/pig-marketing-biz/src/main/java/com/pig4cloud/pig/marketing/service/impl/TcpDataServiceImpl.java

@@ -9,8 +9,10 @@ import com.pig4cloud.pig.marketing.api.dto.mongo.SaveDeviceInfoDTO;
 import com.pig4cloud.pig.marketing.api.dto.mongo.SaveTcpMessageDTO;
 import com.pig4cloud.pig.marketing.api.entity.mongo.Device;
 import com.pig4cloud.pig.marketing.api.entity.mongo.Message;
+import com.pig4cloud.pig.marketing.api.service.MktMgmtHandPushService;
 import com.pig4cloud.pig.marketing.api.vo.mongo.PageDeviceInfoVO;
 import com.pig4cloud.pig.marketing.api.vo.mongo.UserStatisticsVO;
+import com.pig4cloud.pig.marketing.api.vo.mongo.OnlineUserVO;
 import com.pig4cloud.pig.marketing.config.UserStatisticsConfig;
 import com.pig4cloud.pig.marketing.repository.MessageRepository;
 import com.pig4cloud.pig.marketing.service.MktMgmtPushRecordService;
@@ -32,8 +34,11 @@ import org.springframework.stereotype.Service;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 /**
  * @author: lwh
@@ -55,6 +60,8 @@ import java.util.Objects;
 		@Autowired
 		private MktMgmtPushRecordService mktMgmtPushRecordService;
 
+		private final MktMgmtHandPushService mktMgmtHandPushService;
+
 	// 定义时间格式器
 	private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
 
@@ -205,14 +212,26 @@ import java.util.Objects;
 		message.setMsgData(reqDto.getMsgData());
 		message.setReportTime(LocalDateTime.now().format(DATE_TIME_FORMATTER));
 		Message save = messageRepository.save(message);
+		MktMgmtPushRecordSaveDTO saveRecordDTO = new MktMgmtPushRecordSaveDTO();
+		MktMgmtPushRecordSaveDTO handDTO = new MktMgmtPushRecordSaveDTO();
+		String s = "";
+		String hand = "";
 		if (save.getId() != null) {
-			MktMgmtPushRecordSaveDTO saveRecordDTO = new MktMgmtPushRecordSaveDTO();
 			saveRecordDTO.setPushContent(reqDto.getMsgData());
 			saveRecordDTO.setPushIP(reqDto.getClientIP());
 			saveRecordDTO.setPushDomain(reqDto.getClientDomain());
-			String s = mktMgmtPushRecordService.saveRecord(saveRecordDTO);
-			log.info("保存推送记录：{}", s);
+
+			handDTO.setPushContent(reqDto.getMsgData());
+			handDTO.setPushIP(reqDto.getClientIP());
+			handDTO.setPushDomain(reqDto.getClientDomain());
+
+			s = mktMgmtPushRecordService.saveRecord(saveRecordDTO);
+			hand = mktMgmtHandPushService.handPush(handDTO);
 		}
+
+
+		log.info("保存推送记录：{}", s);
+		log.info("保存手动推送记录：{}", hand);
 		return save.getId();
 	}
 
@@ -418,4 +437,76 @@ import java.util.Objects;
 		log.debug("活跃用户数统计完成，时间范围：{} 至现在，去重后的clientID数量：{}", activeStartTimeStr, count);
 		return count;
 	}
+
+	/**
+	 * 查询在线用户列表
+	 * @return 在线用户列表
+	 */
+	@Override
+	public List<OnlineUserVO> getOnlineUsers() {
+		log.info("开始查询在线用户列表");
+		
+		try {
+			// 计算最近10分钟的时间范围
+			LocalDateTime tenMinutesAgo = LocalDateTime.now().minusMinutes(userStatisticsConfig.getActiveTimeRangeMinutes());
+			String tenMinutesAgoStr = tenMinutesAgo.format(DATE_TIME_FORMATTER);
+			
+			// 创建查询条件：时间在最近10分钟内
+			Criteria criteria = Criteria.where("reportTime").gte(tenMinutesAgoStr);
+			Query query = new Query(criteria);
+			
+			// 获取所有符合条件的消息
+			List<Message> messages = mongoTemplate.find(query, Message.class);
+			
+			// 按clientID分组，获取每个clientID的最新消息
+			Map<String, Message> latestMessagesByClient = messages.stream()
+				.filter(msg -> msg.getClientID() != null && !msg.getClientID().trim().isEmpty())
+				.collect(Collectors.groupingBy(
+					Message::getClientID,
+					Collectors.collectingAndThen(
+						Collectors.maxBy(Comparator.comparing(Message::getReportTime)),
+						opt -> opt.orElse(null)
+					)
+				));
+			
+			// 转换为OnlineUserVO列表
+			List<OnlineUserVO> onlineUsers = latestMessagesByClient.values().stream()
+				.filter(Objects::nonNull)
+				.map(this::convertToOnlineUserVO)
+				.sorted(Comparator.comparing(OnlineUserVO::getLastActiveTime).reversed())
+				.collect(Collectors.toList());
+			
+			log.info("在线用户列表查询完成，共找到 {} 个在线用户", onlineUsers.size());
+			return onlineUsers;
+			
+		} catch (Exception e) {
+			log.error("查询在线用户列表异常", e);
+			return new ArrayList<>();
+		}
+	}
+	
+	/**
+	 * 将Message转换为OnlineUserVO
+	 * @param message 消息对象
+	 * @return 在线用户VO
+	 */
+	private OnlineUserVO convertToOnlineUserVO(Message message) {
+		OnlineUserVO onlineUserVO = new OnlineUserVO();
+		onlineUserVO.setClientID(message.getClientID());
+
+		// 设置最后活跃时间
+		if (message.getReportTime() != null) {
+			try {
+				LocalDateTime reportDateTime = LocalDateTime.parse(message.getReportTime(), DATE_TIME_FORMATTER);
+				onlineUserVO.setLastActiveTime(reportDateTime);
+			} catch (Exception e) {
+				log.warn("解析时间失败，clientID: {}, reportTime: {}", message.getClientID(), message.getReportTime());
+				onlineUserVO.setLastActiveTime(LocalDateTime.now());
+			}
+		} else {
+			onlineUserVO.setLastActiveTime(LocalDateTime.now());
+		}
+		
+		return onlineUserVO;
+	}
 }

pig-marketing/pig-marketing-biz/src/main/java/com/pig4cloud/pig/marketing/util/PushFrequencyUtil.java

@@ -0,0 +1,111 @@
+package com.pig4cloud.pig.marketing.util;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 推送频率工具类
+ * 
+ * @author pig4cloud
+ * @date 2025-01-20
+ * @description 提供推送频率检查的公共方法
+ */
+@Slf4j
+public class PushFrequencyUtil {
+
+    /**
+     * 检查推送频率
+     * 
+     * @param pushFrequency 推送频率字符串
+     * @param ruleId 规则ID
+     * @param pushContent 推送内容
+     * @param redisTemplate Redis模板
+     * @return 是否应该推送
+     */
+    public static boolean checkPushFrequency(String pushFrequency, Long ruleId, String pushContent, RedisTemplate<String, Object> redisTemplate) {
+        if (pushFrequency == null || pushFrequency.trim().isEmpty()) {
+            log.debug("规则{}的推送频率为空，默认推送", ruleId);
+            return true;
+        }
+        
+        try {
+            BigDecimal frequency = new BigDecimal(pushFrequency).setScale(2, RoundingMode.HALF_UP);
+            
+            if (frequency.compareTo(BigDecimal.ONE) < 0) {
+                // 小于1则按概率推送
+                double random = Math.random();
+                boolean shouldPush = random <= frequency.doubleValue();
+                log.debug("规则{}推送频率{}小于1，{}%几率推送，结果：{}", ruleId, frequency, frequency.multiply(new BigDecimal(100)), shouldPush);
+                return shouldPush;
+            } else if (frequency.compareTo(BigDecimal.ONE) == 0) {
+                // 等于1则每次都推送
+                log.debug("规则{}推送频率为1，每次都推送", ruleId);
+                return true;
+            } else {
+                // 大于1则向上取整转换为整数周期
+                int cycle = frequency.setScale(0, RoundingMode.CEILING).intValue();
+                log.debug("规则{}推送频率{}大于1，向上取整为周期{}", ruleId, frequency, cycle);
+                
+                // 使用Redis管理周期计数
+                return checkPushCycleWithRedis(ruleId, pushContent, cycle, redisTemplate);
+            }
+        } catch (NumberFormatException e) {
+            log.warn("规则{}的推送频率格式错误：{}，默认推送", ruleId, pushFrequency);
+            return true;
+        }
+    }
+    
+    /**
+     * 使用Redis检查推送周期
+     * 
+     * @param ruleId 规则ID
+     * @param pushContent 推送内容
+     * @param cycle 周期
+     * @param redisTemplate Redis模板
+     * @return 是否应该推送
+     */
+    private static boolean checkPushCycleWithRedis(Long ruleId, String pushContent, int cycle, RedisTemplate<String, Object> redisTemplate) {
+        if (cycle < 1) {
+            return false;
+        }
+        
+        String uuid = UUID.randomUUID().toString();
+        
+        // 构建Redis key，使用规则ID和推送内容作为唯一标识
+        String countKey = String.format("mkt:push:count:%d:%s", ruleId, uuid);
+        
+        try {
+            // Redis原子自增计数
+            Long currentCount = redisTemplate.opsForValue().increment(countKey);
+            if (currentCount == null) {
+                log.warn("Redis自增失败，规则ID：{}，推送内容：{}", ruleId, pushContent);
+                return false;
+            }
+            
+            // 首次访问设置过期时间（24小时）
+            if (currentCount == 1) {
+                redisTemplate.expire(countKey, 24, TimeUnit.HOURS);
+            }
+            
+            // 达到周期次数时返回true并重置计数
+            if (currentCount % cycle == 0) {
+                redisTemplate.delete(countKey);
+                log.debug("规则{}达到推送周期{}，重置计数", ruleId, cycle);
+                return true;
+            }
+            
+            log.debug("规则{}当前计数{}，周期{}，未达到推送条件", ruleId, currentCount, cycle);
+            return false;
+            
+        } catch (Exception e) {
+            log.error("Redis操作异常，规则ID：{}，推送内容：{}", ruleId, pushContent, e);
+            // Redis异常时默认推送，避免影响业务
+            return true;
+        }
+    }
+}