package com.liquidnet.service.consumer.goblin.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.JsonUtils;
import com.liquidnet.service.consumer.goblin.dto.PhoneDto;
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.io.InputStream;
import java.net.URL;

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

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        log.debug("CONSUMER XLS [streamKey:{},messageId:{},stream:{},body:{}]",
                this.getRedisStreamKey(), message.getId(), message.getStream(), message.getValue());
        boolean result = this.consumerSqlDaoHandler(message.getValue().get("message"), Integer.parseInt(message.getValue().get("type")), message.getValue().get("skuId"));
        log.info("CONSUMER XLS_PATH RESULT:{} ==> MESSAGE_ID:{}", result, message.getId());

        try {
            stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
            stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
        } catch (Exception e) {
            log.error("#CONSUMER SQL RESULT:{} ==> DEL_REDIS_QUEUE_MSG_EXCEPTION[MESSAGE_ID:{},MSG:{}]", result, message.getId(), JsonUtils.toJson(message), e);
        } finally {
            try {
                stringRedisTemplate.opsForStream().acknowledge(getRedisStreamGroup(), message);
                stringRedisTemplate.opsForStream().delete(this.getRedisStreamKey(), message.getId());
            } catch (Exception ignored) {
            }
        }
    }

    private boolean consumerSqlDaoHandler(String xlsPath, Integer type, String skuId) {
        Boolean aBoolean = false;
        try {
            URL url = new URL(xlsPath);
            InputStream is = url.openStream();
            EasyExcel.read(is, PhoneDto.class, new PageReadListener<PhoneDto>(dataList -> {
                for (PhoneDto data : dataList) {
                    if(data.getMobile()==null){
                        continue;
                    }
                    String redisKey = GoblinRedisConst.REDIS_CAN_BUY.concat(skuId + ":").concat(data.getMobile());
                    if (type.equals(1)) {//添加
                        log.debug("添加 读取到一条数据{}", JSON.toJSONString(data));
                        redisUtil.set(redisKey, 0);
                    } else {
                        log.debug("删除 读取到一条数据{}", JSON.toJSONString(data));
                        redisUtil.del(redisKey);
                    }
                }
            })).sheet().doRead();
        } catch (Exception e) {
            aBoolean = false;
            log.error("CONSUMER XLS FAIL ==> {}", e.getMessage(), e);
            e.printStackTrace();
        } finally {
//            if (!aBoolean) {
//                HashMap<String, String> map = CollectionUtil.mapStringString();
//                map.put("message", xlsPath);
//                map.put("skuId", skuId);
//                map.put("type", type.toString());
//                stringRedisTemplate.opsForStream().add(StreamRecords.mapBacked(map).withStreamKey(this.getRedisStreamKey()));
//            }
        }
        return aBoolean;
    }

    protected abstract String getRedisStreamKey();

    protected abstract String getRedisStreamGroup();
}