package com.liquidnet.service.consumer.slowly.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.slowly.dto.PhoneDto;
import com.liquidnet.service.consumer.slowly.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.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;
        Integer type = null;
        boolean aBoolean = false;
        try {
            xlsPath = message.get("message");
            String finalSkuId = (skuId = message.get("skuId"));
            Integer finalType = (type = Integer.parseInt(message.get("type")));

            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);
        } catch (Exception e) {
            log.error("CONSUMER MSG EX_HANDLE ==> [{}]:{}", this.getRedisStreamKey(), message, e);
            try {
                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();
}