using Confluent.Kafka; using Google.Protobuf.WellKnownTypes; using Microsoft.Extensions.Options; using Newtonsoft.Json; using System.ComponentModel.DataAnnotations; using TelpoDataService.Util.Clients; using TelpoDataService.Util.Entities.GpsLocationHistory; using TelpoDataService.Util.Models; using TelpoDataService.Util.QueryObjects; 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; private readonly GpsLocationHistoryAccessorClient _messageWanJiaAnClient; public WanJiaAnProcess( IHostEnvironment env, ILogger logger, HttpHelperAsync httpHelper, RedisUtil redis, MqProcessMessage serviceMqProcess, IOptions WanJiaAnConfig, GpsLocationHistoryAccessorClient messageWanJiaAnClient ) { _env = env; _logger = logger; _httpHelper = httpHelper; _redis = redis; _WanJiaAnConfig = WanJiaAnConfig.Value; _serviceMqProcess = serviceMqProcess; _messageWanJiaAnClient = messageWanJiaAnClient; } 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) { try { 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 tagRemark = ""; if (data.report_type == 1) tagRemark = "事件开始"; else if (data.report_type == 2) tagRemark = "事件结束"; else if (data.report_type == 3) tagRemark = "资源上报"; string tag = data?.event_type == 2 ? "哭声检测" : "遮脸提醒"; string eventId = $"{data?.event_id}_{data?.report_type}"; var wanjiaanAlarm = new WanjiaanAlarmEvent() { EventId = eventId, DeviceId = model.device_id, UserId = model.user_id, EventTime = TimeHelper.ConvertToLocalDateTime(model.event_time), EventType = data.event_type, ReportType = data.report_type, EventStart = TimeHelper.ConvertToLocalDateTime(data.event_start), ImageUrl = data.image, VideoUrl = data.url, Remark = $"{tagRemark}:{JsonConvert.SerializeObject(model)};", CreateTime = DateTime.Now, }; if (!string.IsNullOrEmpty(data.event_end)) wanjiaanAlarm.EventEnd = TimeHelper.ConvertToLocalDateTime(data.event_end); if (!string.IsNullOrEmpty(data.video_start)) wanjiaanAlarm.VideoStart = TimeHelper.ConvertToLocalDateTime(data.video_start); if (!string.IsNullOrEmpty(data.video_end)) wanjiaanAlarm.VideoEnd = TimeHelper.ConvertToLocalDateTime(data.video_end); await _messageWanJiaAnClient.AddAsync(wanjiaanAlarm);//数据保存 //推送 业务 //var messageObj = new //{ // messageId = eventId, // topic = "topic.push.third", // time = TimeHelper.ToDateTimeStr(wanjiaanAlarm.EventTime), // data = new // { // imei = wanjiaanAlarm.DeviceId, // wanjiaanAlarm.UserId, // wanjiaanAlarm.EventType, // wanjiaanAlarm.ReportType, // wanjiaanAlarm.EventStart, // wanjiaanAlarm.EventEnd, // wanjiaanAlarm.ImageUrl, // wanjiaanAlarm.VideoUrl, // wanjiaanAlarm.VideoStart, // wanjiaanAlarm.VideoEnd // } //}; //var header = new Dictionary { { "DataType", 32 } }; //await _serviceMqProcess.ProcessThirdhServer(wanjiaanAlarm.DeviceId, messageObj, header, $"{tag}({tagRemark})"); #region 注释 //var alarmEventDb = await _redis.GetWanjiaanAlarmEvent(eventId); //if (alarmEventDb != null) // wanjiaanAlarm.Remark = alarmEventDb.Remark + wanjiaanAlarm.Remark; //await _redis.SetWanjiaanAlarmEvent(wanjiaanAlarm.EventId, wanjiaanAlarm); //if (data.report_type == 3) //{ // await _messageWanJiaAnClient.AddAsync(wanjiaanAlarm); // await _redis.DelWanjiaanAlarmEvent(wanjiaanAlarm.EventId); //} //var param = new GeneralParam //{ // Filters = new List { // new QueryFilterCondition { // Key = nameof(WanjiaanAlarmEvent.DeviceId), // Value = device_id, // ValueType = QueryValueTypeEnum.String, // Operator = QueryOperatorEnum.Equal // }, new QueryFilterCondition { // Key = nameof(WanjiaanAlarmEvent.EventId), // Value = event_id, // ValueType = QueryValueTypeEnum.String, // Operator = QueryOperatorEnum.Equal // } //} //}; //var alarmEventDb =_messageWanJiaAnClient.GetFirst(param); //if (alarmEventDb != null) //{ // alarmEventDb.UserId = user_id; // alarmEventDb.EventTime = event_time; // alarmEventDb.EventType = event_type; // alarmEventDb.ReportType = report_type; // alarmEventDb.EventStart = event_start; // alarmEventDb.EventEnd = wanjiaanAlarm.EventEnd; // alarmEventDb.ImageUrl = image_url; // alarmEventDb.VideoUrl = video_url; // alarmEventDb.VideoStart = wanjiaanAlarm.VideoStart; // alarmEventDb.VideoEnd = wanjiaanAlarm.VideoEnd; // alarmEventDb.Remark = alarmEventDb.Remark + remark; // _messageWanJiaAnClient.Update(alarmEventDb); // await _redis.SetWanjiaanAlarmEvent(event_id, alarmEventDb); //} //else //{ // _messageWanJiaAnClient.Add(wanjiaanAlarm); // await _redis.SetWanjiaanAlarmEvent(event_id, wanjiaanAlarm); //} #endregion } } } } catch (Exception ex) { string s = ex.Message; } } } }