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.config.KafkaConsumerConfig;
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.commons.lang3.ObjectUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author LiuLin
 * @date 2023年08月01日 17:27
 */
@Slf4j
@Component
public class KafkaConsumerWithThread implements CommandLineRunner {
    final private static AtomicLong sendThreadPoolCounter = new AtomicLong(0);
    final public static ExecutorService pooledExecutor =
            Executors.newFixedThreadPool(2 + Runtime.getRuntime().availableProcessors(),
                    createThreadFactory());
    //iot转发实时消息存入influxdb前缀
    private static final String MEASUREMENT = "iot_data_";
    private static final String INDICATORS = "indicators_";
    private static final String TOTAL_CALL_DATA = "total_call_data_";
    private static final String ES_INDEX_NAME_JX = "jxiop_equipments";
    //装备更新最新消息存入influxdb前缀
    private static final String TRUE = "true";
    private static final String FALSE = "false";
    //装备更新最新消息存入influxdb固定时间
    private static final Long TIME = 1688558007051L;
    @Autowired
    protected KafkaProducerService kafkaProducerService;
    @Autowired
    private KafkaConsumerConfig consumerConfig;
    @Autowired
    private InfluxDbConnection influxDbConnection;
    @Autowired
    private RedisUtils redisUtils;

    @Value("${kafka.alarm.topic}")
    private String alarmTopic;

    @Value("${kafka.topic}")
    private String topic;

    @Autowired
    private ElasticSearchUtil elasticSearchUtil;

    private static ThreadFactory createThreadFactory() {
        return runnable -> {
            Thread thread = new Thread(runnable);
            thread.setName(String.format("clojure-agent-send-pool-%d", KafkaConsumerWithThread.sendThreadPoolCounter.getAndIncrement()));
            return thread;
        };
    }

    @Override
    public void run(String... args) {
        Thread thread = new Thread(new KafkaConsumerThread(consumerConfig.consumerConfigs(), topic));
        //4.启动线程
        thread.start();
    }

    private void processRecord(ConsumerRecord<String, String> record, Map<Object, Object> equipmentIndexVOMap) {
        // 处理消息记录
        //log.info("监听Kafka集群message:{}",record.value());
        JSONObject jsonObject = JSONObject.parseObject(record.value());
        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;
        String signalType = jsonObject.getString("signalType");

        try {
            if (equipmentIndexVOMap.get(key) != null) {
                SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                EquipmentIndexVO equipmentSpeIndex = (EquipmentIndexVO) equipmentIndexVOMap.get(key);

                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("isAlarm", String.valueOf(equipmentSpeIndex.getIsAlarm()));
                fieldsMap.put("equipmentSpecificName", equipmentSpeIndex.getEquipmentSpecificName());
                String valueLabel = valueTranslate(value, equipmentSpeIndex.getValueEnum());
                fieldsMap.put("traceId", traceId);
                fieldsMap.put("value", value);
                fieldsMap.put("valueLabel", valueLabel.isEmpty() ? value : valueLabel);
                fieldsMap.put("equipmentIndexName", equipmentSpeIndex.getEquipmentIndexName());
                fieldsMap.put("unit", equipmentSpeIndex.getUnitName());
                fieldsMap.put("createdTime", simpleDateFormat.format(new Date()));
                fieldsMap.put("equipmentIndex", JSON.toJSONString(equipmentSpeIndex));
                influxDbConnection.insert(MEASUREMENT + gatewayId, tagsMap, fieldsMap);

                //总召入库
                if(!"transformation".equals(signalType)){
                    influxDbConnection.insert(TOTAL_CALL_DATA + gatewayId, tagsMap, fieldsMap);
                }
                influxDbConnection.insert(INDICATORS + gatewayId, tagsMap, fieldsMap, TIME, TimeUnit.MILLISECONDS);

                ////更新数据入ES库
                Map<String, Object> paramJson = new HashMap<>();
                if (Arrays.asList(TRUE, FALSE).contains(value)) {
                    paramJson.put("value", value);
                }else{
                    paramJson.put("valueDouble", Float.parseFloat(value));
                }
                paramJson.put("valueLabel", valueLabel.isEmpty() ? value : valueLabel);
                paramJson.put("createdTime", 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) {
            log.error("告警枚举转换异常" + e.getMessage(), e);
        }
        return "";
    }

    public class KafkaConsumerThread implements Runnable {
        private final KafkaConsumer<String, String> kafkaConsumer;
        Map<Object, Object> equipmentIndexVOMap = redisUtils.hmget(RedisKey.EQUIP_INDEX_ADDRESS);

        public KafkaConsumerThread(Properties props, String topic) {
            this.kafkaConsumer = new KafkaConsumer<>(props);
            this.kafkaConsumer.subscribe(Collections.singletonList(topic));
        }

        @Override
        public void run() {
            while (true) {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    pooledExecutor.submit(() -> {
                        processRecord(record, equipmentIndexVOMap);
                    });
                }
            }
        }
    }
}
