接收业务系统调用设备服务信息

src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/Dto/CallDeviceServiceRequestInfoDto.cs

@@ -0,0 +1,15 @@
+using JiShe.ServicePro.Dto;
+
+namespace JiShe.IoT.BusinessSystemAggregation
+{
+    /// <summary>
+    /// 业务系统调用设备服务请求参数
+    /// </summary>
+    public class CallDeviceServiceRequestInfoDto: ReceiveCommandInfoDto
+    {  
+        /// <summary>
+        /// 服务名称
+        /// </summary>
+        public string ServiceName { get; set; } 
+    }
+}

src/JiShe.IoT.Application.Contracts/BusinessSystemAggregation/IBusinessSystemAggregationService.cs

@@ -1,5 +1,6 @@
 using JiShe.ServicePro;
 using JiShe.ServicePro.ApacheIoTDB.Provider.Model;
+using Microsoft.AspNetCore.Authorization;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -46,6 +47,12 @@ namespace JiShe.IoT.BusinessSystemAggregation
         /// </summary>
         /// <param name="input"></param>
         /// <returns></returns>
-        Task<HttpDataResult> BatchCreateDeviceInfoAsync(OpenApiRequest input); 
+        Task<HttpDataResult> BatchCreateDeviceInfoAsync(OpenApiRequest input);
+
+
+        /// <summary>
+        /// 接收业务系统调用设备服务信息，Msg 字段为 CallDeviceServiceRequestInfoDto 实体
+        /// </summary>
+        Task<HttpDataResult<Dictionary<string, object>>> CallDeviceServiceToOneNETForApiAsync(OpenApiRequest input);
     }
 }

src/JiShe.IoT.Application.Contracts/DeviceAggregation/Dto/CallDeviceServiceForApiInput.cs

@@ -0,0 +1,26 @@
+using JiShe.ServicePro.Core;
+using System;
+using System.Collections.Generic;
+using System.ComponentModel.DataAnnotations;
+using System.Text;
+
+namespace JiShe.IoT.DeviceAggregation
+{
+    /// <summary>
+    /// 调用设备服务入参
+    /// </summary>
+    public class CallDeviceServiceForApiInput : IdInput
+    {
+        /// <summary>
+        /// 服务名称
+        /// </summary>
+        [Required]
+        public string ServiceName { get; set; }
+
+        /// <summary>
+        /// 服务参数
+        /// </summary>
+        [Required]
+        public Dictionary<string, object> ServiceParams { get; set; }
+    }
+}

src/JiShe.IoT.Application.Contracts/DeviceAggregation/IDeviceAggregationService.cs

@@ -106,5 +106,12 @@ namespace JiShe.IoT.DeviceAggregation
         /// <param name="input"></param>
         /// <returns></returns>
         Task<bool> BindingDeviceThingModel(BindingDeviceThingModelInput input);
+
+        /// <summary>
+        /// 调用OneNET平台设备服务
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        Task<Dictionary<string, object>> CallDeviceServiceToOneNETForApiAsync(CallDeviceServiceForApiInput input);
     }
 }

src/JiShe.IoT.Application.Contracts/IoTPlatformAggregation/IIoTPlatformAggregationService.cs

@@ -1,5 +1,6 @@
 using JiShe.IoT.IoTPlatformAggregation.Dto;
 using JiShe.ServicePro.Commons;
+using JiShe.ServicePro.IoTDBManagement.TableModels;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -45,5 +46,13 @@ namespace JiShe.IoT.IoTPlatformAggregation
         /// <returns></returns>
         Task<object> UpdateIoTPlatformProductThingModelInfoAsync(UpdateIoTPlatformProductThingModelInfoInput input
             );
+
+        /// <summary>
+        /// OneNET 数据日志处理重新处理
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        Task IoTPlatformDataLogReHandlerAsync(QueryOneNETReceiveMessageInput input
+            );
     }
 }

src/JiShe.IoT.Application/BusinessSystemAggregation/BusinessSystemAggregationService.cs

@@ -96,7 +96,7 @@ namespace JiShe.IoT.BusinessSystemAggregation
                 }
 
                 List<DeviceTelemetryCommandTypeEnum> actionTypes = new List<DeviceTelemetryCommandTypeEnum>() {
-                    DeviceTelemetryCommandTypeEnum.OperateBreaker,
+                    DeviceTelemetryCommandTypeEnum.OperateService,
                     DeviceTelemetryCommandTypeEnum.RoutineOperationSet,
                 };
 
@@ -118,7 +118,7 @@ namespace JiShe.IoT.BusinessSystemAggregation
                 {
                     return HttpDataResultExtensions.Failed("设备不存在", -1041, ResponeResultEnum.Fail);
                 }
-                   
+
                 //将指令存储IoTDB数据库和Redis发布通道
                 if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET)
                 {
@@ -136,7 +136,7 @@ namespace JiShe.IoT.BusinessSystemAggregation
                         var tempPlatformThingModelInfo = platformThingModelInfo.Where(d => d.IoTPlatformRawFieldName == item.Key).FirstOrDefault();
                         if (tempPlatformThingModelInfo == null)
                         {
-                            _logger.LogError($"业务系统推送指令时设备设备{deviceInfo.DeviceAddress}平台端物模型信息不存在属性标识符{item.Key}。");
+                            _logger.LogError($"业务系统推送指令时设备{deviceInfo.DeviceAddress}平台端物模型信息不存在属性标识符{item.Key}。");
                             messageBody.Commands.RemoveAll(d => d.Key == item.Key);
                             continue;
                         }
@@ -213,14 +213,14 @@ namespace JiShe.IoT.BusinessSystemAggregation
                     return HttpDataResultExtensions.Failed<Dictionary<string, object>>("获取数据失败，设备属性值不能为空", -102, ResponeResultEnum.Fail);
                 }
 
-                var restrictionKey = string.Format(RedisConst.CacheDeviceOpenApiRestrictionKey,nameof(ReceiveGetCommandInfoAsync), messageBody.DeviceAddress);
+                var restrictionKey = string.Format(RedisConst.CacheDeviceOpenApiRestrictionKey, nameof(ReceiveGetCommandInfoAsync), messageBody.DeviceAddress);
-               var openApiDeviceRequest =  FreeRedisProvider.Instance.Get(restrictionKey);
+                var openApiDeviceRequest = FreeRedisProvider.Instance.Get(restrictionKey);
                 if (openApiDeviceRequest != null)
                 {
                     return HttpDataResultExtensions.Failed<Dictionary<string, object>>("获取数据失败，设备请求被限制", -103, ResponeResultEnum.Fail);
                 }
 
-                FreeRedisProvider.Instance.Set(restrictionKey,"1",TimeSpan.FromSeconds(5));
+                FreeRedisProvider.Instance.Set(restrictionKey, "1", TimeSpan.FromSeconds(5));
 
                 //限定来源类型必须为业务系统
                 if (messageBody.SourceType != DeviceTelemetrySourceTypeEnum.BusinessSystem)
@@ -240,7 +240,7 @@ namespace JiShe.IoT.BusinessSystemAggregation
                 {
                     return HttpDataResultExtensions.Failed<Dictionary<string, object>>("设备不存在", -106, ResponeResultEnum.Fail);
                 }
-                     
+
                 //将指令存储IoTDB数据库和Redis发布通道
                 if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET)
                 {
@@ -263,7 +263,7 @@ namespace JiShe.IoT.BusinessSystemAggregation
                             messageBody.Commands.RemoveAll(d => d.Key == item.Key);
                             continue;
                         }
-                         
+
                         if (deviceInfo.IsNeedConfigDeviceModel && deviceInfo.DeviceThingModelDataId.HasValue && item.Key.ToLowerInvariant() == ThingModelFixedTypeConst.SpecialCommand.ToLowerInvariant())
                         {
                             _logger.LogError($"业务系统推送指令时设备{deviceInfo.DeviceAddress}平台端物模型属性标识符{item.Key}是特殊指令操作，被禁止。");
@@ -278,7 +278,7 @@ namespace JiShe.IoT.BusinessSystemAggregation
                             messageBody.Commands.RemoveAll(d => d.Key == item.Key);
                             continue;
                         }
-                        commands.Add(tempPlatformThingModelInfo.IoTPlatformRawFieldName,item.Key);
+                        commands.Add(tempPlatformThingModelInfo.IoTPlatformRawFieldName, item.Key);
                     }
 
                     if (commands == null || commands.Count <= 0)
@@ -292,7 +292,7 @@ namespace JiShe.IoT.BusinessSystemAggregation
 
                     //数据写入遥测任务数据存储通道
                     await ioTDBDataChannelManageService.DeviceTelemetryTaskWriterAsync(DataChannelManage.DeviceTelemetryTaskDataChannel.Writer, (DistributedMessageCenterConst.OneNETCommandIssuedEventName, packetTaskInfo));
-                    
+
                     var queryResult = await DevicePropertyValueToOneNET(deviceInfo, new DevicePropertyValueForApiInput() { PropertyList = commands.Select(d => d.Key).ToList() });
 
                     if (queryResult == null || queryResult.Count <= 0)
@@ -616,5 +616,141 @@ namespace JiShe.IoT.BusinessSystemAggregation
                    !deviceAddress.Contains("*") &&
                    !deviceAddress.Contains("~");
         }
+
+        /// <summary>
+        /// 接收业务系统调用设备服务信息，Msg 字段为 CallDeviceServiceRequestInfoDto 实体
+        /// </summary>
+        [AllowAnonymous]
+        public async Task<HttpDataResult<Dictionary<string, object>>> CallDeviceServiceToOneNETForApiAsync(OpenApiRequest input)
+        {
+            try
+            {
+                var handleResult = HandleOpenApiRequest<CallDeviceServiceRequestInfoDto>(input, serverApplicationOptions);
+                if (handleResult.Success == false)
+                {
+                    return HttpDataResultExtensions.Failed<Dictionary<string, object>>("获取数据失败，签名校验异常", -101);
+                }
+                var messageBody = handleResult.Data;
+                if (messageBody == null || messageBody.Commands == null || messageBody.Commands.Count <= 0 || string.IsNullOrWhiteSpace(messageBody.DeviceAddress) || string.IsNullOrWhiteSpace(messageBody.ServiceName))
+                {
+                    return HttpDataResultExtensions.Failed<Dictionary<string, object>>("服务调用失败，服务调用参数不能为空", -102, ResponeResultEnum.Fail);
+                }
+
+                var restrictionKey = string.Format(RedisConst.CacheDeviceOpenApiRestrictionKey, nameof(ReceiveGetCommandInfoAsync), messageBody.DeviceAddress);
+                var openApiDeviceRequest = FreeRedisProvider.Instance.Get(restrictionKey);
+                if (openApiDeviceRequest != null)
+                {
+                    return HttpDataResultExtensions.Failed<Dictionary<string, object>>("服务调用失败，设备请求被限制", -103, ResponeResultEnum.Fail);
+                }
+
+                FreeRedisProvider.Instance.Set(restrictionKey, "1", TimeSpan.FromSeconds(5));
+
+                //限定来源类型必须为业务系统
+                if (messageBody.SourceType != DeviceTelemetrySourceTypeEnum.BusinessSystem)
+                {
+                    return HttpDataResultExtensions.Failed<Dictionary<string, object>>("服务调用失败，来源类型错误，业务系统传固定值2", -104, ResponeResultEnum.Fail);
+                }
+
+                //限定操作类型必须为设备服务调用类型
+                if (messageBody.TelemetryCommandType != DeviceTelemetryCommandTypeEnum.OperateService)
+                {
+                    return HttpDataResultExtensions.Failed<Dictionary<string, object>>("服务调用失败，指令操作类型错误", -105, ResponeResultEnum.Fail);
+                }
+
+                var deviceInfo = await deviceAppService.FindByDeviceAddressAsync(messageBody.DeviceAddress);
+
+                if (deviceInfo == null)
+                {
+                    return HttpDataResultExtensions.Failed<Dictionary<string, object>>("设备不存在", -106, ResponeResultEnum.Fail);
+                }
+
+                //将指令存储IoTDB数据库和Redis发布通道
+                if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET)
+                {
+                    //获取设备对应的平台端物模型信息，校验前端传入的属性标识集合是否存在不合法的属性标识符
+
+                    var platformThingModelInfo = await platformThingModelInfoAppService.FindByPlatformProductIdAsync(new IdInput<string>() { Id = deviceInfo.IoTPlatformProductId });
+
+                    if (platformThingModelInfo == null)
+                    {
+                        return HttpDataResultExtensions.Failed<Dictionary<string, object>>($"业务系统设备服务调时，{deviceInfo.DeviceAddress}的平台端物模型信息不存在。", -107, ResponeResultEnum.Fail);
+                    }
+
+                    Dictionary<string, string> commands = new Dictionary<string, string>();
+                    foreach (var item in messageBody.Commands)
+                    {
+                        var tempPlatformThingModelInfo = platformThingModelInfo.Where(d => d.StandardFieldName == item.Key).FirstOrDefault();
+                        if (tempPlatformThingModelInfo == null)
+                        {
+                            _logger.LogError($"业务系统设备服务调时{deviceInfo.DeviceAddress}平台端物模型信息不存在属性标识符{item.Key}。");
+                            messageBody.Commands.RemoveAll(d => d.Key == item.Key);
+                            continue;
+                        }
+
+
+                        //排除指令
+                        if (exitCommands.Contains(tempPlatformThingModelInfo.StandardFieldName.ToLowerInvariant()))
+                        {
+                            _logger.LogError($"业务系统设备服务调时{deviceInfo.DeviceAddress}平台端物模型属性标识符{item.Key}是升级指令操作，被禁止。");
+                            messageBody.Commands.RemoveAll(d => d.Key == item.Key);
+                            continue;
+                        }
+                        commands.Add(tempPlatformThingModelInfo.IoTPlatformRawFieldName, item.Key);
+                    }
+
+                    if (commands == null || commands.Count <= 0)
+                    {
+                        _logger.LogError($"业务系统设备服务调时{deviceInfo.DeviceAddress}未匹配到具体数据模型，被禁止。");
+                        return HttpDataResultExtensions.Failed<Dictionary<string, object>>("业务系统设备服务调失败，未匹配到具体数据模型", -108);
+                    }
+
+
+                    var packetTaskInfo = GetDeviceTelemetryPacketTaskInfo(ioTDBOptions, input, deviceInfo.Adapt<DeviceCacheInfos>(), commands.Serialize());
+
+                    //数据写入遥测任务数据存储通道
+                    await ioTDBDataChannelManageService.DeviceTelemetryTaskWriterAsync(DataChannelManage.DeviceTelemetryTaskDataChannel.Writer, (DistributedMessageCenterConst.OneNETCommandIssuedEventName, packetTaskInfo));
+
+                    var queryResult = await CallDeviceServiceToOneNET(deviceInfo, new CallDeviceServiceForApiInput()
+                    {
+                        ServiceParams = messageBody.Commands,
+                        ServiceName = messageBody.ServiceName,
+                    });
+
+                    if (queryResult == null || queryResult.Count <= 0)
+                    {
+                        return HttpDataResultExtensions.Failed<Dictionary<string, object>>("业务系统设备服务调失败，OneNET平台未返回数据", -109);
+                    }
+
+                    Dictionary<string, object> result = new Dictionary<string, object>();
+                    foreach (var item in commands)
+                    {
+                        result.Add(item.Value, queryResult[item.Key]);
+                    }
+
+                    return HttpDataResultExtensions.Success<Dictionary<string, object>>(result);
+
+                }
+                else if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.CTWing)
+                {
+                    //await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.CTWingAepCommandIssuedEventName, input);
+                }
+                else
+                {
+                    return HttpDataResultExtensions.Failed<Dictionary<string, object>>("指令处理失败，当前设备平台类型异常", -110);
+                }
+
+                return HttpDataResultExtensions.Success<Dictionary<string, object>>("指令下发成功");
+            }
+            catch (UserFriendlyException ex)
+            {
+                _logger.LogError(ex, "接收业务系统指令信息时发生异常");
+                return HttpDataResultExtensions.Failed<Dictionary<string, object>>("设备属性数据获取失败，发送异常", -111);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "接收业务系统指令信息时发生异常");
+                return HttpDataResultExtensions.Failed<Dictionary<string, object>>("设备属性数据获取失败，发送异常", -112);
+            }
+        }
     }
 }

src/JiShe.IoT.Application/DeviceAggregation/DeviceAggregationService.cs

@@ -638,6 +638,60 @@ namespace JiShe.IoT.DeviceAggregation
         }
 
 
+        /// <summary>
+        /// 调用OneNET平台设备服务
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        public async Task<Dictionary<string, object>> CallDeviceServiceToOneNETForApiAsync(CallDeviceServiceForApiInput input)
+        {
+            try
+            {
+                if (string.IsNullOrWhiteSpace(input.ServiceName) || input.ServiceParams == null || input.ServiceParams.Count <= 0)
+                {
+                    throw new UserFriendlyException($"服务名称参数异常");
+                }
+
+                var deviceInfo = await deviceAppService.FindByIdAsync(input);
+                if (deviceInfo == null)
+                {
+                    throw new UserFriendlyException($"设备不存在");
+                }
+
+
+                //数据写入遥测任务数据存储通道
+                var commandRequest = new OpenApiRequest()
+                {
+                    Message = input.Serialize(),
+                };
+                var packetTaskInfo = GetDeviceTelemetryPacketTaskInfo(ioTDBOptions, commandRequest, deviceInfo.Adapt<DeviceCacheInfos>(), input.Serialize());
+                packetTaskInfo.TelemetryType = (int)DeviceTelemetryCommandTypeEnum.GetAttributeData;
+
+
+                if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET)
+                {
+                    await ioTDBDataChannelManageService.DeviceTelemetryTaskWriterAsync(DataChannelManage.DeviceTelemetryTaskDataChannel.Writer, (DistributedMessageCenterConst.OneNETUpgradeCommandIssuedEventName, packetTaskInfo));
+
+                    return await CallDeviceServiceToOneNET(deviceInfo, input);
+                }
+                else if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.CTWing)
+                {
+                    //await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.CTWingAepCommandIssuedEventName,commandRequest);
+                    //return true;
+                    throw new UserFriendlyException($"发送设备指令信息失败，CTWing暂未实现。");
+                }
+                else
+                {
+                    throw new UserFriendlyException($"发送设备指令信息失败，未找到对应的产品配置信息。");
+                }
+            }
+            catch (Exception)
+            {
+                throw;
+            }
+        }
+
+
         #region CTWing 设备操作 
         /// <summary>
         /// CTWing 设备创建

src/JiShe.IoT.Application/IoTDeviceBasicAppService.cs

@@ -34,12 +34,12 @@ namespace JiShe.IoT
     {
         protected readonly ILogger<IoTDeviceBasicAppService> logger;
         protected readonly IDeviceAppService deviceAppService;
+        protected readonly IOneNETProductService oneNETProductService;
         protected readonly IOneNETDeviceService oneNETDeviceService;
         protected readonly IReliableRedisPubSubService redisPubSubService;
         protected readonly IIoTDBDataChannelManageService ioTDBDataChannelManageService;
         protected readonly IoTDBOptions ioTDBOptions;
         protected readonly ServerApplicationOptions serverApplicationOptions;
-        protected readonly IOneNETProductService oneNETProductService;
         protected readonly IDeviceThingModelManagementAppService deviceThingModelService;
         protected readonly IIoTPlatformThingModelInfoAppService platformThingModelInfoAppService;
         protected readonly IDeviceUpgradeRecordService deviceUpgradeRecordService;
@@ -802,6 +802,70 @@ namespace JiShe.IoT
                 throw;
             }
         }
+
+        /// <summary>
+        /// 调用OneNET平台设备服务
+        /// </summary>
+        /// <param name="deviceInfo"></param>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        /// <exception cref="UserFriendlyException"></exception>
+        protected async Task<Dictionary<string, object>> CallDeviceServiceToOneNET(DeviceManagementInfoDto deviceInfo, CallDeviceServiceForApiInput input)
+        {
+            try
+            {
+                //检查下设备是否在线
+                var deviceOnlineStatus = await oneNETDeviceService.DeviceInfoDetailAsync(new DeviceInfoDetailInput()
+                {
+                    DeviceName = deviceInfo.IoTPlatformDeviceOpenInfo,
+                    OneNETAccountId = deviceInfo.IoTPlatformAccountId,
+                    ProductId = deviceInfo.IoTPlatformProductId,
+                });
+
+                if (deviceOnlineStatus == null || deviceOnlineStatus.Code != ResponeResultEnum.Success)
+                {
+                    throw new UserFriendlyException("获取平台设备信息失败");
+                }
+
+                if (deviceOnlineStatus.Data.Status != 1)
+                {
+                    throw new UserFriendlyException("设备不在线");
+                }
+
+                var deviceDataResult = await oneNETDeviceService.CallDeviceService(new CallDeviceServiceRequestInput()
+                {
+                    DeviceName = deviceInfo.IoTPlatformDeviceOpenInfo,
+                    OneNETAccountId = deviceInfo.IoTPlatformAccountId,
+                    ProductId = deviceInfo.IoTPlatformProductId,
+                    ThingModelIdentifier = input.ServiceName,
+                    ServiceParams = input.ServiceParams
+                });
+                if (deviceDataResult == null || deviceDataResult.Success == false || deviceDataResult.Data == null)
+                {
+                    throw new UserFriendlyException($"设备{deviceInfo.DeviceName}获取数据失败");
+                }
+
+                //调用平台设备服务
+                var platformThingModelInfo = await platformThingModelInfoAppService.FindByPlatformProductIdAsync(new IdInput<string>() { Id = deviceInfo.IoTPlatformProductId });
+                if (platformThingModelInfo == null || platformThingModelInfo.Count <= 0)
+                {
+                    return deviceDataResult.Data;
+                }
+
+                List<string> updateKeys = new List<string>()
+                {
+                        ThingModelFixedTypeConst.FIRMWARE_VERSION.ToLowerInvariant(),
+                        ThingModelFixedTypeConst.ReadingMode.ToLowerInvariant()
+                };
+                   
+
+                return deviceDataResult.Data;
+            }
+            catch (Exception ex)
+            {
+                throw;
+            }
+        }
         #endregion
 
         #region CTWing 设备操作

src/JiShe.IoT.Application/IoTPlatformAggregation/IoTPlatformAggregationService.cs

@@ -7,8 +7,12 @@ using JiShe.ServicePro.CTWingManagement.CTWingProducts;
 using JiShe.ServicePro.DeviceManagement.ThingModels;
 using JiShe.ServicePro.Enums;
 using JiShe.ServicePro.FreeRedisProvider;
+using JiShe.ServicePro.IoTDBManagement.TableModels;
+using JiShe.ServicePro.IoTDBManagement.TreeModels;
+using JiShe.ServicePro.OneNET.Provider.ReceiveModels;
 using JiShe.ServicePro.OneNETManagement.OneNETAccounts;
 using JiShe.ServicePro.OneNETManagement.OneNETProducts;
+using JiShe.ServicePro.OneNETManagement.Subscribers;
 using Microsoft.AspNetCore.DataProtection.KeyManagement;
 using Microsoft.Extensions.Logging;
 using System;
@@ -31,6 +35,9 @@ namespace JiShe.IoT.IoTPlatformAggregation
         private readonly ICTWingAccountService _ctwingAccountService;
         private readonly IOneNETAccountService _oneNETAccountService;
         private readonly IIoTPlatformThingModelInfoAppService _ioTPlatformThingModelInfoAppService;
+        private readonly IOneNETPulsarServiceSubscriptionService _oneNETPulsarServiceSubscriptionService;
+        private readonly ITableModelService _tableModelService;
+
 
 
         public IoTPlatformAggregationService(ILogger<IoTPlatformAggregationService> logger,
@@ -38,6 +45,8 @@ namespace JiShe.IoT.IoTPlatformAggregation
             IOneNETProductService oneNetProductService,
             ICTWingAccountService ctwingAccountService,
             IOneNETAccountService oneNETAccountService,
+            IOneNETPulsarServiceSubscriptionService oneNETPulsarServiceSubscriptionService,
+            ITableModelService tableModelService,
             IIoTPlatformThingModelInfoAppService ioTPlatformThingModelInfoAppService)
         {
             _logger = logger;
@@ -46,6 +55,8 @@ namespace JiShe.IoT.IoTPlatformAggregation
             _ctwingAccountService = ctwingAccountService;
             _oneNETAccountService = oneNETAccountService;
             _ioTPlatformThingModelInfoAppService = ioTPlatformThingModelInfoAppService;
+            _oneNETPulsarServiceSubscriptionService = oneNETPulsarServiceSubscriptionService;
+            _tableModelService = tableModelService;
         }
 
         /// <summary>
@@ -349,5 +360,37 @@ namespace JiShe.IoT.IoTPlatformAggregation
                 throw;
             }
         }
+
+
+
+        /// <summary>
+        /// OneNET 数据日志处理重新处理
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        public async Task IoTPlatformDataLogReHandlerAsync(QueryOneNETReceiveMessageInput input)
+        {
+            try
+            {
+                var oneNETDataLoglist = await _tableModelService.OneNETLogInfoPageAsync(input);
+                if (oneNETDataLoglist == null || oneNETDataLoglist.Items == null || oneNETDataLoglist.Items.Count <= 0)
+                {
+                    _logger.LogError($"{nameof(IoTPlatformDataLogReHandlerAsync)}OneNET数据日志处理重新处理失败，未获取到数据");
+                    return;
+                }
+
+                foreach (var item in oneNETDataLoglist.Items)
+                {
+                    string topicName = string.Equals(item.MessageType, IoTDBDataTypeConst.Data, StringComparison.OrdinalIgnoreCase) ? DistributedMessageCenterConst.OneNETThingModelPropertyChangeReceivedEventName : DistributedMessageCenterConst.OneNETThingModelEventChangeReceivedEventName;
+                    var tempOneNETReceiveBasicModel = item.RawMessage.Deserialize<OneNETReceiveBasicModel>();
+                    await _oneNETPulsarServiceSubscriptionService.OneNETReceiveThingModelHandlerAsync(topicName, item.IoTDataType, tempOneNETReceiveBasicModel, true);
+                }
+            }
+            catch (Exception)
+            {
+
+                throw;
+            }
+        }
     }
 }

src/JiShe.IoT.HttpApi/Controllers/IoTPlatformAggregationController.cs

@@ -6,6 +6,7 @@ using JiShe.ServicePro;
 using JiShe.ServicePro.Commons;
 using JiShe.ServicePro.Core;
 using JiShe.ServicePro.DeviceManagement.DeviceInfos;
+using JiShe.ServicePro.IoTDBManagement.TableModels;
 
 namespace JiShe.IoT.Controllers
 {
@@ -69,5 +70,18 @@ namespace JiShe.IoT.Controllers
             return await _iotPlatformAggregationService.UpdateIoTPlatformProductThingModelInfoAsync(input);
         }
 
+
+        /// <summary>
+        /// OneNET数据日志处理重新处理
+        /// </summary>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        [HttpPost(nameof(IoTPlatformDataLogReHandlerAsync))]
+        [SwaggerOperation(summary: "OneNET数据日志处理重新处理", Tags = new[] { "AggregationIoTPlatform" })]
+        public async Task IoTPlatformDataLogReHandlerAsync(QueryOneNETReceiveMessageInput input)
+        {
+            await _iotPlatformAggregationService.IoTPlatformDataLogReHandlerAsync(input);
+        }
+
     }
 }