using Confluent.Kafka; using Microsoft.Extensions.Options; using Newtonsoft.Json; using TelpoPush.WanJiaAn.Worker.Common; using TelpoPush.WanJiaAn.Worker.Models.Config; using TelpoPush.WanJiaAn.Worker.Models.Enum; using TelpoPush.WanJiaAn.Worker.Models.MqTemplates; using TelpoPush.WanJiaAn.Worker.Models.PushTemplates; using TelpoPush.WanJiaAn.Worker.Service.Cache; using TelpoPush.WanJiaAn.Worker.Service.Mq; namespace TelpoPush.WanJiaAn.Worker.Handlers { public class WanJiaAnProcess { private readonly static object _syncLocker = new object(); private readonly IHostEnvironment _env; private readonly ILogger _logger; private readonly HttpHelperAsync _httpHelper; private readonly RedisUtil _redis; private readonly MqProcessMessage _serviceMqProcess; private readonly WanJiaAnConfig _WanJiaAnConfig; public WanJiaAnProcess( IHostEnvironment env, ILogger logger, HttpHelperAsync httpHelper, RedisUtil redis, MqProcessMessage serviceMqProcess, IOptions WanJiaAnConfig ) { _env = env; _logger = logger; _httpHelper = httpHelper; _redis = redis; _WanJiaAnConfig = WanJiaAnConfig.Value; _serviceMqProcess = serviceMqProcess; } public async Task SendWanJiaAn(string? message, string topic, Headers headers) { #region 数据初始验证 bool isHandle = true; BaseModel model = null; string imei = ""; if (!string.IsNullOrEmpty(message)) { model = JsonConvert.DeserializeObject(message); if (model != null) { var Jo = JsonConvert.DeserializeObject>(message); if (Jo.ContainsKey("device_id")) imei = Jo["device_id"].ToString(); if (string.IsNullOrEmpty(imei)) { _logger.LogInformation($"[数据信息不完整] device_id信息不存在：{message}"); isHandle = false; } //else // await _redis.GetGpsDevice(imei); } else { _logger.LogInformation($"[数据信息不完整] 数据解析异常：{message}"); isHandle = false; } } else { _logger.LogInformation($"[数据信息不完整] message数据异常：{message}"); isHandle = false; } #endregion if (isHandle) { lock (_syncLocker) { #region 注释 Headers 解析 //Headers 解析 //HeadersDto headersDto = new HeadersDto(); //try //{ // foreach (var item in headers) // { // if (item.Key == KafkaHeader.DataType) // headersDto.DataType = BitConverter.ToInt32(item.GetValueBytes(), 0); // else if (item.Key == KafkaHeader.AlarmType) // headersDto.AlarmType = BitConverter.ToInt32(item.GetValueBytes(), 0); // else if (item.Key == KafkaHeader.OperType) // headersDto.OperType = BitConverter.ToInt32(item.GetValueBytes(), 0); // } //} //catch (Exception ex) //{ // _logger.LogError($"当前工作线程Headers异常,{ex.Message}|{ex.Source}|{ex.StackTrace}"); //} #endregion try { #region 注释 //string dataType = headersDto.DataType != null ? "_" + headersDto.DataType : ""; //string alarmType = headersDto.AlarmType != null ? "_" + headersDto.AlarmType : ""; //string operType = headersDto.OperType != null ? "_" + headersDto.OperType : ""; //string key = dataType + alarmType + operType; //var dataStatus = _redis.IsDateStatus(model, imei, key).Result; //过滤 //if (headersDto.DataType == (int)MqDataType.TemperatureInfo // || headersDto.DataType == (int)MqDataType.Temperature1Info // || headersDto.DataType == (int)MqDataType.BindDevice // || headersDto.DataType == (int)MqDataType.WanJiaAnInfo // || headersDto.DataType == (int)MqDataType.HeartRateInfo // || headersDto.DataType == (int)MqDataType.HeartRate1Info // || headersDto.DataType == (int)MqDataType.Spo2Info // || headersDto.DataType == (int)MqDataType.Spo21Info // || headersDto.DataType == (int)MqDataType.BloodPressInfo // || headersDto.DataType == (int)MqDataType.BloodPress1Info // || headersDto.DataType == (int)MqDataType.SportResult // || headersDto.DataType == (int)MqDataType.BloodSugar // ) // dataStatus.isPush = true; //dataStatus.isPush = true; //if (dataStatus.isPush) //{ //switch (topic) //{ // case "topic.push.third": // switch (headersDto.DataType) // { // case (int)MqDataType.AlarmInfo: //报警消息 // break; // case (int)MqDataType.TemperatureInfo: //体温消息 // break; // case (int)MqDataType.WanJiaAnInfo: //定位消息 // // DataServicePusWanJiaAn(model, imei); // break; // case (int)MqDataType.StepInfo: //步数消息 // break; // case (int)MqDataType.BatteryLevelInfo: //电量消息 // break; // case (int)MqDataType.DeviceCallLog: //设备通话记录 // break; // case (int)MqDataType.DeviceSmsLog: //设备短信记录 // break; // case (int)MqDataType.DeviceConfigInfo: //设备配置信息 // break; // case (int)MqDataType.Status: //设备状态（offline,online） // break; // case (int)MqDataType.Active: //设备激活状态 // break; // case (int)MqDataType.reply: //指令回调 // break; // case (int)MqDataType.Weather: //天气查询 // break; // case (int)MqDataType.ReadMsg: //短消息阅读 // break; // case (int)MqDataType.StudyAINotifyStatusUpload: //学习能力状态 // break; // case (int)MqDataType.HeartRateInfo: //心率 // break; // case (int)MqDataType.HeartRate1Info: //周期性心率 // break; // case (int)MqDataType.Spo2Info: //血氧 // break; // case (int)MqDataType.Spo21Info: //周期性血氧 // break; // case (int)MqDataType.Temperature1Info: //周期性报体温数据 // break; // case (int)MqDataType.DrownReportInfo: //防溺水告警 // break; // case (int)MqDataType.WearStatusInfo: //手表未佩戴告警 // break; // case (int)MqDataType.BloodPressInfo: //血压 // break; // case (int)MqDataType.BloodPress1Info: //周期性血压 // break; // case (int)MqDataType.PsychInfo: //心理监测 // break; // case (int)MqDataType.AiCallResult: //AI呼叫结果回调 // case (int)MqDataType.CrossBorder: //越界上报(围栏进出告警) // break; // case (int)MqDataType.SportResult: //运动数据上报 // break; // case (int)MqDataType.BindDevice: //绑定业务 // break; // case (int)MqDataType.BloodSugar: //血糖业务 // break; // default: // break; // } // break; // default: // break; //} //} //else // _logger.LogInformation($"数据未处理(历史数据):{JsonConvert.SerializeObject(dataStatus)}"); #endregion switch (topic) { case "topic.wanjiaan.push.telpo": DataServicePusWanJiaAn(model); break; default: break; } } catch (Exception ex) { _logger.LogError($"当前工作线程异常: {ex.Message}|{ex.Source}|{ex.StackTrace}"); } } } } //位置 public async Task DataServicePusWanJiaAn(BaseModel model) { if (model.data != null) { var alarmEvent = JsonConvert.DeserializeObject(model.data.ToString()); if (alarmEvent?.alarm_event != null) { var data = alarmEvent?.alarm_event; if (data?.event_type == 2 || data?.event_type == 52)//2:哭声检测;52:遮脸提醒 { string event_id = data.event_id; string device_id = model.device_id; string user_id = model.user_id; int event_type = data.event_type; int report_type = data.report_type; DateTime event_time = TimeHelper.ConvertToLocalDateTime(model.event_time); DateTime event_start = TimeHelper.ConvertToLocalDateTime(data.event_start); if (!string.IsNullOrEmpty(data.event_end)) { DateTime event_end = TimeHelper.ConvertToLocalDateTime(data.event_end); } if (!string.IsNullOrEmpty(data.video_start)) { DateTime video_start = TimeHelper.ConvertToLocalDateTime(data.video_start); } if (!string.IsNullOrEmpty(data.video_end)) { DateTime video_end = TimeHelper.ConvertToLocalDateTime(data.video_end); } string image_url = data.image; string video_url = data.url; string remark = JsonConvert.SerializeObject(model); DateTime create_time = DateTime.Now; var obj = new { }; } } } } } }