package com.yeejoin.amos.message.kafka;

import com.alibaba.fastjson.JSON;
import com.yeejoin.amos.message.utils.ClassToJsonUtil;
import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.core.io.Resource;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import org.typroject.tyboot.component.emq.EmqKeeper;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import static com.yeejoin.amos.message.kafka.Constant.*;

/**
 * kafka 消费服务
 *
 * @author litw
 * @create 2022/11/1 10:06
 **/
@Slf4j
@Service
public class KafkaConsumerService {

    private static final String MQTT_TOPIC = "romaSite/data/transmit";
    private static final String PROVINCE_MQTT_TOPIC = "province/data/transport";
    private static final String MQTT_TOPIC_EVENT_ALARM = "romaSite/data/eventAlarm";
    @Autowired
    protected EmqKeeper emqKeeper;
    @Value("${system.zxj}")
    private boolean isZxj;
    @Value("classpath:/json/commonMessage.json")
    private Resource commonMessage;

//    /**
//     * 批量消费kafka消息
//     * Kafka消息转emq
//     *
//     * @param consumerRecords messages
//     * @param ack             ack
//     */
//    @KafkaListener(id = "consumerSingle", groupId = "zhTestGroup", topics = "#{'${kafka.topics}'.split(',')}")
//    public void listen1(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
//        try {
//            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//                Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
//                if (kafkaMessage.isPresent()) {
//                    JSONObject messageObj = JSONObject.fromObject(kafkaMessage.get());
//                    if (messageObj.has(TOPIC)) {
//                        emqKeeper.getMqttClient().publish(messageObj.optString(TOPIC), messageObj.getJSONObject(DATA).toString()
//                                .getBytes(StandardCharsets.UTF_8), 0, false);
//                    }
//                    log.info("kafka消费zhTestGroup消息{}", messageObj);
//                }
//            }
//        } catch (Exception e) {
//            log.error("kafka失败，当前失败的批次: data:{}, {}", consumerRecords, e);
//        } finally {
//            ack.acknowledge();
//        }
//    }

    /**
     * 批量消费kafka消息
     * 监听数据表变化kafka数据转发emq
     *
     * @param consumerRecords messages
     * @param ack             ack
     */
    @KafkaListener(id = "provinceMessage", groupId = "province", topics = "#{'${kafka.topics}'.split(',')}")
    public void listen(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        try {
            if (isZxj) {
                for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                    Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
                    if (kafkaMessage.isPresent()) {
                        JSONObject messageObj = JSONObject.fromObject(kafkaMessage.get());
                        String type = messageObj.optString(TYPE);
                        String table = messageObj.optString(TABLE);
                        if (Arrays.asList(INSERT, UPDATE).contains(type) && DBTableTypeEnum.have(table) != null) {
                            JSONObject data = (JSONObject) messageObj.getJSONArray(DATA).get(0);
                            data.put(DB_TYPE, type);
                            data.put(TABLE, table);
                            emqKeeper.getMqttClient().publish(PROVINCE_MQTT_TOPIC, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
                            log.info("kafka消费province消息{}", messageObj);
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("kafka失败，当前失败的批次: data:{}, {}", consumerRecords, e);
        } finally {
            ack.acknowledge();
        }
    }

    /**
     * 转发苏州，绍兴换流站Kafka数据对emq
     *
     * @param record record
     * @param ack    ack
     */
    @KafkaListener(id = "kafkaRoma", groupId = "kafkaRoma", topics = "#{'${queue.kafka.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
    public void kafkaListener(ConsumerRecord<?, String> record, Acknowledgment ack) {
        try {
            Optional<?> messages = Optional.ofNullable(record.value());
            if (messages.isPresent()) {
//                JSONObject messageObj = JSONObject.fromObject(record.value());
//                if (messageObj.getJSONObject(BODY).isEmpty()) {
//                    messageObj.put(DATA_TYPE, STATE);
//                }
//                JSONObject object = JSONObject.fromObject(record.value());
//                String json = ClassToJsonUtil.class2json(object, commonMessage, MQTT_TOPIC);
//                emqKeeper.getMqttClient().publish(MQTT_TOPIC, json.getBytes(StandardCharsets.UTF_8), 0, false);

                JSONObject object = JSONObject.fromObject(record.value());
                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
            }
        } catch (MqttException e) {
            log.error("换流站转发Kafka消息失败" + e.getMessage(), e);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        } finally {
            ack.acknowledge();
        }
    }

    @KafkaListener(id = "kafkaConsumerEventAlarm", groupId = "kafkaConsumerGroupEventAlarm", topics = "#{'${queue.kafka.eventAlarm.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
    public void kafkaConsumerEventAlarm(ConsumerRecord<?, String> record, Acknowledgment ack) {
        Optional<?> message = Optional.ofNullable(record.value());
        if (message.isPresent()) {
            try {
//                JSONObject messageObj = JSONObject.fromObject(record.value());
//                JSONArray dataArray = messageObj.getJSONArray("data");
//                JSONArray jsonArray = new JSONArray();
//                String timestamp = "";
//                for (Object obj : dataArray) {
//                    JSONObject finallyObj = new JSONObject();
//                    com.alibaba.fastjson.JSONObject detail = com.alibaba.fastjson.JSONObject.parseObject(com.alibaba.fastjson.JSONObject.toJSONString(obj));
//                    finallyObj.put("eventtextL1", detail.get("description"));
//                    finallyObj.put("pointId", detail.get("astId"));
//                    finallyObj.put("time", detail.get("dateTime"));
//                    jsonArray.add(finallyObj);
//                    timestamp = detail.get("dateTime").toString();
//                }
//                JSONObject jsonObjectMessage = new JSONObject();
//                jsonObjectMessage.put("warns", jsonArray);
//                jsonObjectMessage.put("timestamp", timestamp);

//                JSONObject object = JSONObject.fromObject(record.value());
//                String json = ClassToJsonUtil.class2json(object, commonMessage, MQTT_TOPIC_EVENT_ALARM);
//                emqKeeper.getMqttClient().publish(MQTT_TOPIC_EVENT_ALARM, json.getBytes(StandardCharsets.UTF_8), 0, false);

                JSONObject object = JSONObject.fromObject(record.value());
                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
                ack.acknowledge();
            } catch (MqttException e) {
                log.error("解析数据失败,{}", e.getMessage());
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
        }
    }
//
//    /**
//     * 韶山换流对接Kafka
//     * @param record record
//     * @param ack ack
//     */
//    @KafkaListener(id = "kafkaConsumer", groupId = "kafkaConsumerGroup", topics = "#{'${queue.kafka.shaoshan.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
//    public void kafkaConsumer(ConsumerRecord<?, String> record, Acknowledgment ack) {
//        Optional<?> message = Optional.ofNullable(record.value());
//        if (message.isPresent()) {
//            try {
////                JSONObject messageObj = JSONObject.fromObject(record.value());
////                JSONObject data = messageObj.getJSONObject("body");
////                JSONObject object = JSONObject.fromObject(record.value());
////                String json = ClassToJsonUtil.class2json(object, commonMessage, MQTT_TOPIC_EVENT_ALARM);
////                emqKeeper.getMqttClient().publish(MQTT_TOPIC_EVENT_ALARM, json.getBytes(StandardCharsets.UTF_8), 0, false);
//
//                JSONObject object = JSONObject.fromObject(record.value());
//                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
//                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
//                ack.acknowledge();
//            } catch (MqttException e) {
//                log.error("解析数据失败,{}", e.getMessage());
//            } catch (UnsupportedEncodingException e) {
//                e.printStackTrace();
//            }
//        }
//    }
//
//    /**
//     * 事件告警对接Kafka
//     * @param record record
//     * @param ack ack
//     *            groupId = kafkaConsumerGroup
//     *            该消息的消息格式为
//     *            {"data_class":"realdata","data_type":"alarm","op_type":"subscribe_emergency","condition":{"station_psr_id":"50edcb6c1b8a811030493c80a2014950ed9d4f59e8","station_name":"中州换流站","alarm_type":"yx_bw"},"data":[{"psrId":"D017020000000000000000999","astId":"D017020000000000000000999","equipType":"ASTType_0000111","eventType":"OtherSignal","alarmSource":"OWS","alarmLevel":"3","description":"2024-03-11 09:06:17::585 S2WCL12A E3.C01软水器再生结束信号 出现","dateTime":"2024-03-11 09:06:17.585"}]}
//     */
//
//    @KafkaListener(id = "kafkaConsumerEventAlarm", groupId = "kafkaConsumerGroupEventAlarm", topics = "#{'${queue.kafka.eventAlarm.topics}'.split(',')}", containerFactory = "kafkaRomaContainerFactory")
//    public void kafkaConsumerEventAlarm(ConsumerRecord<?, String> record, Acknowledgment ack) {
//        Optional<?> message = Optional.ofNullable(record.value());
//        if (message.isPresent()) {
//            try {
////                JSONObject messageObj = JSONObject.fromObject(record.value());
////                JSONArray dataArray = messageObj.getJSONArray("data");
////                JSONArray jsonArray = new JSONArray();
////                String timestamp = "";
////                for (Object obj : dataArray) {
////                    JSONObject finallyObj = new JSONObject();
////                    com.alibaba.fastjson.JSONObject detail = com.alibaba.fastjson.JSONObject.parseObject(com.alibaba.fastjson.JSONObject.toJSONString(obj));
////                    finallyObj.put("eventtextL1", detail.get("description"));
////                    finallyObj.put("pointId", detail.get("astId"));
////                    finallyObj.put("time", detail.get("dateTime"));
////                    jsonArray.add(finallyObj);
////                    timestamp = detail.get("dateTime").toString();
////                }
////                JSONObject jsonObjectMessage = new JSONObject();
////                jsonObjectMessage.put("warns", jsonArray);
////                jsonObjectMessage.put("timestamp", timestamp);
//
//                JSONObject object = JSONObject.fromObject(record.value());
//                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
//                emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 0, false);
//                ack.acknowledge();
//            } catch (MqttException e) {
//                log.error("解析数据失败,{}", e.getMessage());
//            } catch (UnsupportedEncodingException e) {
//                e.printStackTrace();
//            }
//        }
//    }


    /**
     * 消费单条消息,topics 可以监听多个topic，如：topics = {"topic1", "topic2"}
     *
     * @param message 消息
     */
    @KafkaListener(id = "consumerSingle", groupId = "${kafka.station.groupId}", topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
    public void consumerSingle(String message, Acknowledgment ack) {
        JSONObject messageObj = JSONObject.fromObject(message);
        try {
            String topic = null;
            JSONObject data = null;
            if (messageObj.has("topic")) {
                topic = messageObj.getString("topic");
                data = messageObj.getJSONObject("data");
                emqKeeper.getMqttClient().publish(topic, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
                ack.acknowledge();
                log.info("消息转发成功" + messageObj);
            }
        } catch (Exception e) {
            log.error("消息转发失败" + e.getMessage(), e);
        }
    }


    ///**
    // * 省级消息转发
    // *
    // * @param message 省级消息
    // * @param ack     ack
    // */
    //@KafkaListener(id = "provinceMessage", groupId = "province", topics = "#{'${kafka.topics}'.split(',')}", concurrency = "2")
    //public void consumerSingle1(String message, Acknowledgment ack) {
    //    log.info("省级消息转发打印" + message);
    //    if(isZxj) {
    //        Optional<?> messages = Optional.ofNullable(message);
    //        if (messages.isPresent()) {
    //            try {
    //                JSONObject jsonObject = JSONObject.fromObject(message);
    //                String type = jsonObject.optString("type");
    //                String table = jsonObject.optString("table");
    //                if (Arrays.asList(Constant.INSERT, Constant.UPDATE).contains(type) && DBTableTypeEnum.have(table) != null) {
    //                    if (Arrays.asList("INSERT", "UPDATE").contains(type)) {
    //                        JSONArray array = jsonObject.getJSONArray("data");
    //                        JSONObject data = (JSONObject)array.get(0);
    //                        data.put("dbType", type);
    //                        data.put("table", table);
    //                        emqKeeper.getMqttClient().publish(PROVINCE_MQTT_TOPIC, data.toString().getBytes(StandardCharsets.UTF_8), 0, false);
    //                        log.info("省级消息: {}", data);
    //                    }
    //                }
    //            } catch (MqttException e) {
    //                log.error("消息转发失败" + e.getMessage(), e);
    //            }
    //            ack.acknowledge();
    //        }
    //    }
    //}


/*    @KafkaListener(id = "consumerBatch", topicPartitions = {
            @TopicPartition(topic = "hello-batch1", partitions = "0"),
            @TopicPartition(topic = "hello-batch2", partitionOffsets = @PartitionOffset(partition = "2", initialOffset = "4"))
    })*/
//    /**
//     * 批量消费消息
//     * @param messages
//     */
//    @KafkaListener(id = "consumerBatch", topics = "test-batch")
//    public void consumerBatch(List<ConsumerRecord<String, String>> messages) {
//        log.info("consumerBatch =====> messageSize: {}", messages.size());
//        log.info(messages.toString());
//    }

//    /**
//     * 指定消费异常处理器
//     * @param message
//     */
//    @KafkaListener(id = "consumerException", topics = "kafka-test-topic", errorHandler = "consumerAwareListenerErrorHandler")
//    public void consumerException(String message) {
//        throw new RuntimeException("consumer exception");
//    }
//
//    /**
//     * 验证ConsumerInterceptor
//     * @param message
//     */
//    @KafkaListener(id = "interceptor", topics = "consumer-interceptor")
//    public void consumerInterceptor(String message) {
//        log.info("consumerInterceptor ====> message: {}", message);
//    }
//
//
//
    //kafka的监听器，topic为"zhTest"，消费者组为"zhTestGroup"
    //@KafkaListener(topics = "test", groupId = "zhTestGroup")
    //public void listenZhugeGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
    //    String value = record.value();
    //    System.out.println(value);
    //    System.out.println(record);
    //    //手动提交offset
    //    ack.acknowledge();
    //}
}


