using JiShe.IoT.BusinessSystemAggregation.Dto; using JiShe.IoT.DeviceAggregation; using JiShe.ServicePro; using JiShe.ServicePro.ApacheIoTDB.Provider.Model; using JiShe.ServicePro.ApacheIoTDB.Provider.Options; using JiShe.ServicePro.Core; using JiShe.ServicePro.DataChannelManages; using JiShe.ServicePro.DeviceManagement.DeviceInfos; using JiShe.ServicePro.DeviceManagement.ThingModels; using JiShe.ServicePro.Dto; using JiShe.ServicePro.Enums; using JiShe.ServicePro.FreeRedisProvider; using JiShe.ServicePro.IoTDBManagement.DataChannels; using JiShe.ServicePro.IoTDBManagement.TreeModels; using JiShe.ServicePro.OneNETManagement.OneNETDevices; using JiShe.ServicePro.OneNETManagement.OneNETProducts; using Mapster; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Volo.Abp; using Volo.Abp.Auditing; using static Volo.Abp.Ui.LayoutHooks.LayoutHooks; namespace JiShe.IoT.BusinessSystemAggregation { /// /// 业务系统聚合服务 /// /// 设备服务 /// OneNET设备服务 /// Redis发布订阅服务 /// 数据通道 /// IoTDBOptions /// OneNET产品服务 /// 设备端物模型服务 /// 平台端端物模型服务 /// 设备升级记录服务 /// 应用服务配置 /// IoTDB树模型服务 /// [DisableAuditing] public class BusinessSystemAggregationService( IDeviceAppService _deviceAppService, IOneNETDeviceService _oneNETDeviceService, IReliableRedisPubSubService _redisPubSubService, IIoTDBDataChannelManageService _ioTDBDataChannelManageService, IOptions _ioTDBOptions, IOptions _serverOptions, IOneNETProductService _oneNETProductService, IDeviceThingModelManagementAppService _deviceThingModelService, IIoTPlatformThingModelInfoAppService _platformThingModelInfoAppService, IDeviceUpgradeRecordService _deviceUpgradeRecordService, ITreeModelService _treeModelService, ILogger _logger) : IoTDeviceBasicAppService(_logger, _deviceAppService, _oneNETDeviceService, _redisPubSubService, _ioTDBDataChannelManageService, _ioTDBOptions, _serverOptions, _oneNETProductService, _deviceThingModelService, _platformThingModelInfoAppService, _deviceUpgradeRecordService, _treeModelService), IBusinessSystemAggregationService { private const string LUA_SCRIPT = @" local hashKey = KEYS[1] local fieldKeys = ARGV return redis.call('HMGET', hashKey, unpack(fieldKeys))"; /// /// 接收业务系统设置指令信息 /// [AllowAnonymous] public async Task ReceiveSetCommandInfoAsync(OpenApiRequest input) { try { var handleResult = HandleOpenApiRequest(input, serverApplicationOptions); if (handleResult.Success == false) { return handleResult; } var messageBody = handleResult.Data; if (messageBody == null || messageBody.Commands == null || messageBody.Commands.Count <= 0) { return HttpDataResultExtensions.Failed("设备指令不能为空", -103, ResponeResultEnum.Fail); } //限定来源类型必须为业务系统 if (messageBody.SourceType != DeviceTelemetrySourceTypeEnum.BusinessSystem) { return HttpDataResultExtensions.Failed("设备指令来源类型错误,业务系统传固定值2", -104, ResponeResultEnum.Fail); } var deviceInfo = await deviceAppService.FindByDeviceAddressAsync(messageBody.DeviceAddress); if (deviceInfo == null) { return HttpDataResultExtensions.Failed("设备不存在", -1041, ResponeResultEnum.Fail); } var packetTaskInfo = GetDeviceTelemetryPacketTaskInfo(ioTDBOptions, input, deviceInfo.Adapt(), messageBody.Commands.Serialize()); //将指令存储IoTDB数据库和Redis发布通道 if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET) { //获取设备对应的平台端物模型信息,校验前端传入的属性标识集合是否存在不合法的属性标识符 var platformThingModelInfo = await platformThingModelInfoAppService.FindByPlatformProductIdAsync(new IdInput() { Id = deviceInfo.IoTPlatformProductId }); if (platformThingModelInfo == null) { throw new UserFriendlyException($"业务系统推送指令时设备{deviceInfo.DeviceAddress}的平台端物模型信息不存在。"); } foreach (var item in messageBody.Commands) { var tempPlatformThingModelInfo = platformThingModelInfo.Where(d => d.IoTPlatformRawFieldName == item.Key).FirstOrDefault(); if (tempPlatformThingModelInfo == null) { throw new UserFriendlyException($"业务系统推送指令时设备设备{deviceInfo.DeviceAddress}平台端物模型信息不存在属性标识符{item.Key}。"); } //排除升级指令 if (tempPlatformThingModelInfo.StandardFieldName.ToLowerInvariant() == ThingModelFixedTypeConst.FIRMWARE_UPGRADE.ToLowerInvariant()) { throw new UserFriendlyException($"业务系统推送指令时设备{deviceInfo.DeviceAddress}平台端物模型属性标识符{item.Key}是升级指令操作,被禁止。"); } if (deviceInfo.IsNeedConfigDeviceModel && deviceInfo.DeviceThingModelDataId.HasValue && item.Key.ToLowerInvariant() == ThingModelFixedTypeConst.SpecialCommand.ToLowerInvariant()) { throw new UserFriendlyException($"业务系统推送指令时设备{deviceInfo.DeviceAddress}平台端物模型属性标识符{item.Key}是特殊指令操作,被禁止。"); } } //数据写入遥测任务数据存储通道 await ioTDBDataChannelManageService.DeviceTelemetryTaskWriterAsync(DataChannelManage.DeviceTelemetryTaskDataChannel.Writer, (DistributedMessageCenterConst.OneNETCommandIssuedEventName, packetTaskInfo)); await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.OneNETCommandIssuedEventName, input); } else if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.CTWing) { await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.CTWingAepCommandIssuedEventName, input); } else { return HttpDataResultExtensions.Failed("指令处理失败,当前设备平台类型异常", -105); } return HttpDataResultExtensions.Success("指令下发成功"); } catch (UserFriendlyException) { throw; // 重新抛出用户友好异常 } catch (Exception ex) { _logger.LogError(ex, "接收业务系统指令信息时发生异常"); return HttpDataResultExtensions.Failed("指令处理失败,发送异常", -106); } } /// /// 接收业务系统获取属性指令信息 /// [AllowAnonymous] public async Task>> ReceiveGetCommandInfoAsync(OpenApiRequest input) { try { var handleResult = HandleOpenApiRequest(input, serverApplicationOptions); if (handleResult.Success == false) { return HttpDataResultExtensions.Failed>("获取数据失败,签名校验异常", -101); } var messageBody = handleResult.Data; if (messageBody == null || messageBody.Commands == null || messageBody.Commands.Count <= 0) { return HttpDataResultExtensions.Failed>("获取数据失败,设备属性值不能为空", -103, ResponeResultEnum.Fail); } //限定来源类型必须为业务系统 if (messageBody.SourceType != DeviceTelemetrySourceTypeEnum.BusinessSystem) { return HttpDataResultExtensions.Failed>("获取数据失败,来源类型错误,业务系统传固定值2", -104, ResponeResultEnum.Fail); } var deviceInfo = await deviceAppService.FindByDeviceAddressAsync(messageBody.DeviceAddress); if (deviceInfo == null) { return HttpDataResultExtensions.Failed>("设备不存在", -105, ResponeResultEnum.Fail); } var packetTaskInfo = GetDeviceTelemetryPacketTaskInfo(ioTDBOptions, input, deviceInfo.Adapt(), messageBody.Commands.Serialize()); //将指令存储IoTDB数据库和Redis发布通道 if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.OneNET) { //获取设备对应的平台端物模型信息,校验前端传入的属性标识集合是否存在不合法的属性标识符 var platformThingModelInfo = await platformThingModelInfoAppService.FindByPlatformProductIdAsync(new IdInput() { Id = deviceInfo.IoTPlatformProductId }); if (platformThingModelInfo == null) { return HttpDataResultExtensions.Failed>($"业务系统推送指令时设备{deviceInfo.DeviceAddress}的平台端物模型信息不存在。", -106, ResponeResultEnum.Fail); } foreach (var item in messageBody.Commands) { var tempPlatformThingModelInfo = platformThingModelInfo.Where(d => d.IoTPlatformRawFieldName == item.Key).FirstOrDefault(); if (tempPlatformThingModelInfo == null) { _logger.LogError($"业务系统推送指令时设备设备{deviceInfo.DeviceAddress}平台端物模型信息不存在属性标识符{item.Key}。"); messageBody.Commands.RemoveAll(d => d.Key == item.Key); continue; } //排除升级指令 if (tempPlatformThingModelInfo.StandardFieldName.ToLowerInvariant() == ThingModelFixedTypeConst.FIRMWARE_UPGRADE.ToLowerInvariant()) { _logger.LogError($"业务系统推送指令时设备{deviceInfo.DeviceAddress}平台端物模型属性标识符{item.Key}是升级指令操作,被禁止。"); 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}是特殊指令操作,被禁止。"); messageBody.Commands.RemoveAll(d => d.Key == item.Key); continue; } } //数据写入遥测任务数据存储通道 await ioTDBDataChannelManageService.DeviceTelemetryTaskWriterAsync(DataChannelManage.DeviceTelemetryTaskDataChannel.Writer, (DistributedMessageCenterConst.OneNETCommandIssuedEventName, packetTaskInfo)); if (messageBody.Commands == null || messageBody.Commands.Count <= 0) { return HttpDataResultExtensions.Failed>("获取数据失败,OneNET平台未返回数据", -106); } var queryResult = await DevicePropertyValueToOneNET(deviceInfo, new DevicePropertyValueForApiInput() { PropertyList = messageBody.Commands.Select(d => d.Key).ToList() }); if (queryResult == null || queryResult.Count <= 0) { return HttpDataResultExtensions.Failed>("获取数据失败,OneNET平台未返回数据", -106); } return HttpDataResultExtensions.Success>(queryResult); } else if (deviceInfo.IoTPlatform == IoTPlatformTypeEnum.CTWing) { //await redisPubSubService.PublishReliableAsync(DistributedMessageCenterConst.CTWingAepCommandIssuedEventName, input); } else { return HttpDataResultExtensions.Failed>("指令处理失败,当前设备平台类型异常", -105); } return HttpDataResultExtensions.Success>("指令下发成功"); } catch (UserFriendlyException) { throw; // 重新抛出用户友好异常 } catch (Exception ex) { _logger.LogError(ex, "接收业务系统指令信息时发生异常"); return HttpDataResultExtensions.Failed>("指令处理失败,发送异常", -106); } } /// /// 业务系统批量查询设备数据,Msg 字段为 BatchQueryDeviceDataInfoInput 实体 /// /// /// [AllowAnonymous] public async Task>> BatchQueryDeviceDataInfoAsync(OpenApiRequest input) { try { var handleResult = HandleOpenApiRequest(input, serverApplicationOptions); if (handleResult.Success == false) { return HttpDataResultExtensions.Failed>(null, handleResult.Message, handleResult.LocationCode); } var messageBody = handleResult.Data; if (messageBody.DeviceAddresses == null || messageBody.DeviceAddresses.Count <= 0) { return HttpDataResultExtensions.Failed>(null, "设备地址不能为空", -103); } // 验证设备地址格式,防止注入攻击 foreach (var deviceAddress in messageBody.DeviceAddresses) { if (!IsValidDeviceAddress(deviceAddress)) { return HttpDataResultExtensions.Failed>(null, $"设备地址格式不正确: {deviceAddress}", -107); } } //执行脚本 var result = await FreeRedisProvider.Instance.EvalAsync ( LUA_SCRIPT, new[] { RedisConst.CacheAllDeviceInfoHashKey }, messageBody.DeviceAddresses.ToArray() ); List deviceCacheInfos = new List(); // 处理返回结果 if (result is object[] values) { foreach (var value in values) { if (value != null) { var tempFocusInfo = ServiceProJsonSerializer.Deserialize(value as string); deviceCacheInfos.Add(tempFocusInfo); } else { deviceCacheInfos.Add(null); // 添加空值以保持索引对应 } } } List queryResult = new List(); for (int i = 0; i < messageBody.DeviceAddresses.Count; i++) { var deviceAddress = messageBody.DeviceAddresses[i]; var deviceCacheInfo = deviceCacheInfos.Count > i ? deviceCacheInfos[i] : null; if (deviceCacheInfo == null) { _logger.LogError($"{nameof(BatchQueryDeviceDataInfoAsync)} 业务系统批量查询设备数据,设备地址:{deviceAddress}未找到设备地址缓存信息,消息体为:{input.Serialize()}"); continue; } var pageResult = await treeModelService.OpenRequestDeviceDataInfoPageAsync(new DeviceTreeModelDataInfoInput() { DeviceAddress = deviceAddress, DeviceType = messageBody.DeviceType, IoTDataType = messageBody.IoTDataType, IsNeedPaging = false, StartCreationTime = messageBody.BeginTime, EndCreationTime = messageBody.EndTime }); //todo 根据业务系统时间间隔要求进行过滤 if (pageResult.Items != null && pageResult.Items.Count > 0) { queryResult.AddRange(pageResult.Items); } } return HttpDataResultExtensions.Success(queryResult, "查询成功"); } catch (Exception ex) { _logger.LogError(ex, "批量查询设备数据时发生异常"); return HttpDataResultExtensions.Failed>(null, "查询设备数据失败", -106); } } /// /// 业务系统查询单个设备数据,Msg 字段为 QueryDeviceDataInfoInput 实体 /// /// /// [AllowAnonymous] public async Task>> QueryDeviceDataInfoAsync(OpenApiRequest input) { try { var handleResult = HandleOpenApiRequest(input, serverApplicationOptions); if (handleResult.Success == false) { return HttpDataResultExtensions.Failed>(null, handleResult.Message, handleResult.LocationCode); } var messageBody = handleResult.Data; if (string.IsNullOrWhiteSpace(messageBody.DeviceAddress)) { return HttpDataResultExtensions.Failed>(null, "设备地址不能为空", -103); } // 验证设备地址格式,防止注入攻击 if (!IsValidDeviceAddress(messageBody.DeviceAddress)) { return HttpDataResultExtensions.Failed>(null, $"设备地址格式不正确: {messageBody.DeviceAddress}", -107); } //执行脚本 var result = await FreeRedisProvider.Instance.EvalAsync ( LUA_SCRIPT, new[] { RedisConst.CacheAllDeviceInfoHashKey }, new List() { messageBody.DeviceAddress }.ToArray() ); List deviceCacheInfos = new List(); // 处理返回结果 if (result is object[] values) { foreach (var value in values) { if (value != null) { var tempFocusInfo = ServiceProJsonSerializer.Deserialize(value as string); deviceCacheInfos.Add(tempFocusInfo); } else { deviceCacheInfos.Add(null); } } } List queryResult = new List(); var deviceCacheInfo = deviceCacheInfos.FirstOrDefault(); if (deviceCacheInfo == null) { _logger.LogError($"{nameof(QueryDeviceDataInfoAsync)} 业务系统单个查询设备数据,设备地址:{messageBody.DeviceAddress}未找到设备地址缓存信息,消息体为:{input.Serialize()}"); return HttpDataResultExtensions.Failed>(null, "设备信息不存在", -108); } var pageResult = await treeModelService.OpenRequestDeviceDataInfoPageAsync(new DeviceTreeModelDataInfoInput() { DeviceAddress = messageBody.DeviceAddress, DeviceType = messageBody.DeviceType, IoTDataType = messageBody.IoTDataType, IsNeedPaging = false, StartCreationTime = messageBody.BeginTime, EndCreationTime = messageBody.EndTime, SubDeviceAddress = messageBody.SubDeviceAddress, }); //todo 根据业务系统时间间隔要求进行过滤 if (pageResult.Items != null && pageResult.Items.Count > 0) { queryResult.AddRange(pageResult.Items); } return HttpDataResultExtensions.Success(queryResult, "查询成功"); } catch (Exception ex) { _logger.LogError(ex, "查询单个设备数据时发生异常"); return HttpDataResultExtensions.Failed>(null, "查询设备数据失败", -106); } } /// /// 业务系统批量新增设备信息,Msg 字段为 BatchCreateDeviceBusinessSystemInput 实体 /// /// /// [AllowAnonymous] public async Task BatchCreateDeviceInfoAsync(OpenApiRequest input) { try { var handleResult = HandleOpenApiRequest(input, serverApplicationOptions); if (handleResult.Success == false) { return HttpDataResultExtensions.Failed(handleResult.Message, handleResult.LocationCode); } var messageBody = handleResult.Data; if (messageBody.DeviceInfos == null || messageBody.DeviceInfos.Count <= 0) { _logger.LogError($"{nameof(BatchCreateDeviceInfoAsync)} 业务系统批量新增设备数据,设备地址不能为空,消息体为:{input.Serialize()}"); return HttpDataResultExtensions.Failed("设备地址不能为空", -101); } if (messageBody.DeviceSourceType != DeviceSourceTypeEnum.Prepay && messageBody.DeviceSourceType != DeviceSourceTypeEnum.Energy) { _logger.LogError($"{nameof(BatchCreateDeviceInfoAsync)} 业务系统批量创建设备信息,设备来源异常,消息体为:{input.Serialize()}"); return HttpDataResultExtensions.Failed("业务系统批量创建设备信息,设备来源异常。", -102); } var createResult = false; if (messageBody.IoTPlatform == ServicePro.Enums.IoTPlatformTypeEnum.CTWing) { var batchCreateDeviceInput = input.Adapt(); batchCreateDeviceInput.AddressList = messageBody.DeviceInfos.Select(f => f.DeviceAddress.Trim()).ToList(); createResult = await CTWingDeviceBatchCreateAsync(batchCreateDeviceInput); } else if (messageBody.IoTPlatform == ServicePro.Enums.IoTPlatformTypeEnum.OneNET) { var batchCreateDeviceInput = input.Adapt(); batchCreateDeviceInput.AddressList = messageBody.DeviceInfos.Select(f => f.DeviceAddress.Trim()).ToList(); createResult = await OneNETDeviceBatchCreateAsync(batchCreateDeviceInput); } if (createResult == false) { _logger.LogError($"{nameof(BatchCreateDeviceInfoAsync)} 业务系统批量新增设备数据没有成功,消息体为:{input.Serialize()}"); return HttpDataResultExtensions.Failed("新增设备失败", -104); } return HttpDataResultExtensions.Success("新增设备成功"); } catch (Exception ex) { _logger.LogError(ex, "批量新增设备数据时发生异常"); return HttpDataResultExtensions.Failed("新增设备失败", -106); } } /// /// 验证设备地址格式是否合法 /// /// 设备地址 /// 是否合法 private bool IsValidDeviceAddress(string deviceAddress) { if (string.IsNullOrWhiteSpace(deviceAddress)) return false; // 可根据实际业务规则调整验证逻辑 // 这里简单检查是否包含非法字符 return !deviceAddress.Contains("..") && !deviceAddress.Contains("*") && !deviceAddress.Contains("~"); } } }