package com.yeejoin.amos.message.kafka;

import com.alibaba.excel.metadata.Sheet;
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.apache.poi.ss.usermodel.*;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
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.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;
import org.typroject.tyboot.component.emq.EmqKeeper;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;

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

    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;

    /**
     * execl文件路径，读取excel
     */
//    @Value("${filter.excel.path:F:\\filterExcel11.xlsx}")
    @Value("${filter.excel.path:}")
    private String filePath;

    /**
     * 服务启动时，内存存储execl文档中需要的编码信息
     */
    private List<String> codeListInfo;

    private static final String topic1 = "romaSite/data/transmit";
    private static final String topic2 = "romaSite/data/eventAlarm";

    /**
     * 转发苏州，绍兴换流站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());
            log.info("kafka上报数据：{}", record.toString());
            if (messages.isPresent()) {
                JSONObject object = JSONObject.fromObject(record.value());
                com.alibaba.fastjson.JSONObject jsonObj = ClassToJsonUtil.class2json(object, commonMessage, record.topic());
                if ((StringUtils.isEmpty(filePath)) || (CollectionUtils.isEmpty(codeListInfo)) || (!ObjectUtils.isEmpty(jsonObj) && Boolean.TRUE.equals(isSendEmq(jsonObj)))) {
                    log.info("kafka上报mqtt数据:{}", JSON.toJSONString(jsonObj));
                    emqKeeper.getMqttClient().publish(String.valueOf(jsonObj.get("mqTopic")), JSON.toJSONString(jsonObj).getBytes("UTF-8"), 2, false);
                }
            }
        } catch (MqttException e) {
            log.error("换流站转发Kafka消息失败" + e.getMessage(), e);
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        } finally {
            ack.acknowledge();
        }
    }

    /**
     * 消费单条消息,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), 2, false);
                ack.acknowledge();
                log.info("消息转发成功" + messageObj);
            }
        } catch (Exception e) {
            log.error("消息转发失败" + e.getMessage(), e);
        }
    }

    @Override
    public void run(ApplicationArguments args) {
        codeListInfo = readExcelFile(filePath);
        log.info("excel文件数据:{}", JSON.toJSONString(codeListInfo));
    }

    /**
     * 判断是否发送emq消息
     * @return true 发送  false 不发送
     */
    private Boolean isSendEmq(com.alibaba.fastjson.JSONObject res) {
        String key = "";
        if (!StringUtils.isEmpty(res.get("mqTopic")) && topic1.equals(res.get("mqTopic").toString())) {
            key = res.getJSONObject("data").get("key").toString();
        } else if (!StringUtils.isEmpty(res.get("mqTopic")) && topic2.equals(res.get("mqTopic").toString())) {
            key = res.getJSONObject("data").getJSONArray("warns").getJSONObject(0).get("pointId").toString();
        }
        return !StringUtils.isEmpty(key) && codeListInfo.contains(key);
    }

    public static List<String> readExcelFile(String filePath) {
        try (FileInputStream fis = new FileInputStream(new File(filePath));
             Workbook workbook = new XSSFWorkbook(fis)) {
            com.alibaba.excel.metadata.Sheet sheet = workbook.getSheetAt(0); // 获取第一个工作表
            List<String> list = getColumnData(sheet);
            // 在这里处理list中的数据，例如打印、存储等操作
            return list;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return new ArrayList<>();
    }

    private static List<String> getColumnData(Sheet sheet) {
        List<String> list = new ArrayList<>();
        Iterator<Row> rowIterator = sheet.iterator();
        rowIterator.next(); // 跳过表头行
        while (rowIterator.hasNext()) {
            Row row = rowIterator.next();
            Cell cell = row.getCell(0);
            if (cell != null) {
                String cellValue = getCellValueAsString(cell);
                list.add(cellValue);
            }
        }
        return list;
    }

    private static String getCellValueAsString(Cell cell) {
        if (cell == null) {
            return "";
        }
        switch (cell.getCellType()) {
            case STRING:
                return cell.getStringCellValue();
            case NUMERIC:
                if (DateUtil.isCellDateFormatted(cell)) {
                    return cell.getDateCellValue().toString();
                } else {
                    return Double.toString(cell.getNumericCellValue());
                }
            case BOOLEAN:
                return Boolean.toString(cell.getBooleanCellValue());
            case FORMULA:
                return cell.getCellFormula();
            default:
                return "";
        }
    }
}


