package com.liquidnet.service.consumer.kylin.receiver;

import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.read.listener.PageReadListener;
import com.alibaba.fastjson.JSON;
import com.liquidnet.common.cache.redis.util.RedisUtil;
import com.liquidnet.commons.lang.util.CollectionUtil;
import com.liquidnet.service.consumer.kylin.dto.PhoneDto;
import com.liquidnet.service.consumer.kylin.dto.PhoneUnIdDto;
import com.liquidnet.service.consumer.kylin.service.IBaseDao;
import com.liquidnet.service.goblin.constant.GoblinRedisConst;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import java.net.URL;
import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.Map;

@Slf4j
public abstract class AbstractXlsRedisReceiver implements StreamListener<String, MapRecord<String, String, String>> {
    @Autowired
    private IBaseDao baseDao;
    @Autowired
    StringRedisTemplate stringRedisTemplate;
    @Autowired
    private RedisUtil redisUtil;

    private static final String SQL_INSERT_GOODS_BUY_ROSTER_LOG = "INSERT INTO goblin_goods_buy_roster_log (sku_id,buy_roster,buy_roster_type,parsing_result,created_at)VALUES(?,?,?,?,?)";

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        String redisStreamKey = this.getRedisStreamKey();
        log.debug("CONSUMER MSG[streamKey:{},messageId:{},stream:{},body:{}]", redisStreamKey, message.getId(), message.getStream(), message.getValue());
        boolean result = this.consumerMessageHandler(message.getValue());
        log.info("CONSUMER MSG RESULT:{} ==> [{}]MESSAGE_ID:{}", result, redisStreamKey, message.getId());

        try {
            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
        } catch (Exception e) {
            log.error("#CONSUMER MSG EX_ACK ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
        }
        try {
            stringRedisTemplate.opsForStream().delete(redisStreamKey, message.getId());
        } catch (Exception e) {
            log.error("#CONSUMER MSG EX_DEL ==> [{}]RESULT:{},MESSAGE:{}", redisStreamKey, result, message.getValue(), e);
        }
    }

    private boolean consumerMessageHandler(Map<String, String> message) {
        LinkedList<Object[]> objs = CollectionUtil.linkedListObjectArr();
        String xlsPath = null, skuId = null;
        String oXlsPath = null;
        Integer type = null;
        boolean aBoolean = false;
        try {
            String[] path = message.get("message").split(",");
            xlsPath = path[0];
            if (path.length > 1) {
                oXlsPath = path[1];
            }
            String finalSkuId = (skuId = message.get("skuId"));
            String listId;
//            try {
                listId = message.getOrDefault("listId", "");
//            }catch (Exception e){
//                listId = "";
//            }
            Integer finalType = (type = Integer.parseInt(message.get("type")));
            if (finalType.equals(1) || finalType.equals(2)) {
                EasyExcel.read(new URL(xlsPath).openStream(), PhoneDto.class, new PageReadListener<PhoneDto>(dataList -> {
                    for (PhoneDto data : dataList) {
                        if (data.getMobile() == null) {
                            continue;
                        }
                        String redisKey = GoblinRedisConst.REDIS_CAN_BUY.concat(finalSkuId + ":").concat(data.getMobile());
                        if (finalType.equals(1)) {//添加
                            if (log.isDebugEnabled()) {
                                log.debug("添加 读取到一条数据{}", JSON.toJSONString(data));
                            }
                            redisUtil.set(redisKey, 0);
                        } else {
                            if (log.isDebugEnabled()) {
                                log.debug("删除 读取到一条数据{}", JSON.toJSONString(data));
                            }
                            redisUtil.del(redisKey);
                        }
                    }
                })).sheet().doRead();
                objs.add(new Object[]{skuId, xlsPath, type, 1, LocalDateTime.now()});
                aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
            } else if (finalType.equals(3) || finalType.equals(4)) {
                if (xlsPath != null && !xlsPath.equals("")) {//添加
                    EasyExcel.read(new URL(xlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
                        for (HashMap<String, String> data : dataList) {
                            if (data.get(0) == null) {
                                continue;
                            }
                            if (finalType.equals(3)) {
                                redisUtil.set(GoblinRedisConst.REDIS_WHITE.concat(listId+":").concat(finalSkuId + ":").concat(data.get(0)), 1);
                            } else {
                                redisUtil.set(GoblinRedisConst.REDIS_BLACK.concat(listId+":").concat(finalSkuId + ":").concat(data.get(0)), 1);
                            }
                        }
                    })).sheet().doRead();
                }
                if (oXlsPath != null&& !oXlsPath.equals("")) {//删除旧的黑白名单
                    EasyExcel.read(new URL(oXlsPath).openStream(), new PageReadListener<HashMap<String, String>>(dataList -> {
                        for (HashMap<String, String> data : dataList) {
                            if (data.get(0) == null) {
                                continue;
                            }
                            if (finalType.equals(3)) {
                                redisUtil.del(GoblinRedisConst.REDIS_WHITE.concat(listId+":").concat(finalSkuId + ":").concat(data.get(0)));
                            } else {
                                redisUtil.del(GoblinRedisConst.REDIS_BLACK.concat(listId+":").concat(finalSkuId + ":").concat(data.get(0)));
                            }
                        }
                    })).sheet().doRead();
                }
                aBoolean = true;
            }
        } catch (Exception e) {
            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), message, e);
            try {
                if (type.equals(1) || type.equals(2)) {
                    objs.add(new Object[]{skuId, xlsPath, type, 2, LocalDateTime.now()});
                    aBoolean = baseDao.batchSql(SQL_INSERT_GOODS_BUY_ROSTER_LOG, objs);
                }
            } catch (Exception ignored) {
            }
        }
        return aBoolean;
    }

    protected abstract String getRedisStreamKey();

    protected abstract String getRedisStreamGroup();
}