//package com.yeejoin.equip.kafka;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONArray;
//import com.alibaba.fastjson.JSONObject;
//import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
//import com.yeejoin.equip.entity.EquipmentIndexVO;
//import com.yeejoin.equip.utils.ElasticSearchUtil;
//import com.yeejoin.equip.utils.RedisKey;
//import com.yeejoin.equip.utils.RedisUtils;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerRecord;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.beans.factory.annotation.Value;
//import org.springframework.kafka.annotation.KafkaListener;
//import org.springframework.kafka.support.Acknowledgment;
//import org.springframework.stereotype.Service;
//import org.springframework.util.CollectionUtils;
//import org.springframework.util.ObjectUtils;
//
//import java.text.SimpleDateFormat;
//import java.util.*;
//
///**
// * @author LiuLin
// * @date 2023/6/25
// * @apiNote kafka 消费服务类
// */
//@Slf4j
//@Service
//public class KafkaConsumerService {
//
//    //iot转发实时消息存入influxdb前缀
//    private static final String MEASUREMENT = "iot_data_";
//    //装备更新最新消息存入influxdb前缀
//    private static final String TRUE = "true";
//    private static final String FALSE = "false";
//    //装备更新最新消息存入influxdb固定时间
//    private static final String ES_INDEX_NAME_JX = "jxiop_equipments";
//    @Autowired
//    protected KafkaProducerService kafkaProducerService;
//    @Autowired
//    private InfluxDbConnection influxDbConnection;
//    @Autowired
//    private RedisUtils redisUtils;
//    @Value("${kafka.alarm.topic}")
//    private String alarmTopic;
//    @Autowired
//    private ElasticSearchUtil elasticSearchUtil;
//
//    @KafkaListener(topics = "#{'${kafka.topic}'.split(',')}", groupId = "messageConsumerGroup")
//    public void listen(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
//        try {
//            if (CollectionUtils.isEmpty(consumerRecords)) {
//                return;
//            }
//            Map<Object, Object> equipmentIndexVOMap = redisUtils.hmget(RedisKey.EQUIP_INDEX_ADDRESS);
//
//            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//                Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
//                if (kafkaMessage.isPresent()) {
//                    String message = (String) kafkaMessage.get();
//                    this.handleMessage(message, equipmentIndexVOMap);
//                }
//            }
//        } catch (Exception e) {
//            log.error("kafka失败，当前失败的批次。data:{}", consumerRecords);
//            e.printStackTrace();
//        } finally {
//            ack.acknowledge();
//        }
//    }
//
//    private void handleMessage(String message, Map<Object, Object> equipmentIndexVOMap) {
//
//        JSONObject jsonObject = JSONObject.parseObject(message);
//        String dataType = jsonObject.getString("dataType");
//        String indexAddress = jsonObject.getString("address");
//        String traceId = jsonObject.getString("traceId");
//        String gatewayId = jsonObject.getString("gatewayId");
//        String value = jsonObject.getString("value");
//        String key = indexAddress + "_" + gatewayId;
//
//        try {
//            if (equipmentIndexVOMap.get(key) != null) {
//                SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
//                EquipmentIndexVO equipmentSpeIndex = (EquipmentIndexVO) equipmentIndexVOMap.get(key);
//                log.info("接收到iot消息: 指标名称:{},地址:{},值:{},网关{}",
//                        equipmentSpeIndex.getEquipmentIndexName(), indexAddress, value, gatewayId);
//
//                Map<String, String> tagsMap = new HashMap<>();
//                Map<String, Object> fieldsMap = new HashMap<>();
//                tagsMap.put("equipmentsIdx", key);
//                fieldsMap.put("address", indexAddress);
//                fieldsMap.put("gatewayId", gatewayId);
//                fieldsMap.put("dataType", dataType);
//                fieldsMap.put("equipmentIndexName", equipmentSpeIndex.getEquipmentIndexName());
//                fieldsMap.put("equipmentIndex", JSON.toJSONString(equipmentSpeIndex));
//                fieldsMap.put("equipmentSpecificName", equipmentSpeIndex.getEquipmentSpecificName());
//                fieldsMap.put("isAlarm", String.valueOf(equipmentSpeIndex.getIsAlarm()));
//
//                String valueLabel = valueTranslate(value, equipmentSpeIndex.getValueEnum());
//                fieldsMap.put("traceId", traceId);
//                fieldsMap.put("value", value);
//                fieldsMap.put("valueLabel", valueLabel.isEmpty() ? value : valueLabel);
//                fieldsMap.put("unit", equipmentSpeIndex.getUnitName());
//                fieldsMap.put("createdTime", simpleDateFormat.format(new Date()));
//                influxDbConnection.insert(MEASUREMENT + gatewayId, tagsMap, fieldsMap);
//
//                //更新数据入ES库
//                Map<String, Object> paramJson = new HashMap<>();
//                if (Arrays.asList(TRUE, FALSE).contains(value)) {
//                    paramJson.put("value", value);
//                } else {
//                    paramJson.put("value", Float.parseFloat(value));
//                }
//                paramJson.put("valueLabel", valueLabel.isEmpty() ? value : valueLabel);
//                paramJson.put("createdTime", simpleDateFormat.format(new Date()));
//                paramJson.put("unit", equipmentSpeIndex.getUnitName());
//                elasticSearchUtil.updateData(ES_INDEX_NAME_JX, key, JSON.toJSONString(paramJson));
//
//                if (equipmentSpeIndex.getIsAlarm() != null && 1 == equipmentSpeIndex.getIsAlarm()) {
//                    fieldsMap.putAll(tagsMap);
//                    //kafkaProducerService.sendMessageAsync(alarmTopic, JSON.toJSONString(fieldsMap));
//                }
//            }
//        } catch (Exception e) {
//            log.error("Iot透传消息解析入库失败" + e.getMessage(), e);
//        }
//
//    }
//
//    private String valueTranslate(String value, String enumStr) {
//        if (ObjectUtils.isEmpty(enumStr)) {
//            return "";
//        }
//        try {
//            JSONArray jsonArray = JSONArray.parseArray(enumStr);
//            for (int i = 0; i < jsonArray.size(); i++) {
//                JSONObject jsonObject = jsonArray.getJSONObject(i);
//                if (jsonObject.get("key").equals(value)) {
//                    return jsonObject.getString("label");
//                }
//            }
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//        return "";
//    }
//}
//
//
