package com.muyu.parse.process;

import cn.hutool.json.JSONObject;
import com.muyu.common.kafka.config.KafkaProducerConfig;
import com.muyu.domain.MessageValue;
import com.muyu.domain.SysCar;
import com.muyu.domain.SysCarType;
import com.muyu.enterprise.cache.MessageValueCacheService;
import com.muyu.enterprise.cache.SysCarCacheService;
import com.muyu.enterprise.cache.SysCarTypeCacheService;
import com.muyu.parse.uitl.DataParseUtil;
import lombok.extern.log4j.Log4j2;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.List;


/**
 * @version 1.0
 * @Author xie ya ru
 * @Date 2024/9/28 21:14
 * @注释
 */
@Log4j2
@Configuration
public class ProcessData {



    //报文模版信息
    @Resource
    private  MessageValueCacheService messageValueCacheService;

    //车辆信息
    @Resource
    private  SysCarCacheService sysCarCacheService;

    //车辆类型信息
    @Resource
    private  SysCarTypeCacheService sysCarTypeCacheService;

    @Resource
    private  KafkaProducerConfig kafkaProducerConfig;

    private final static String topic = "sysCar_vin_topic";

    public  void DataConversion(String jsonVin ) {
        //设置数组存储车辆数据
        JSONObject jsonObject = new JSONObject();
        String vin = DataParseUtil.dataParsing(jsonVin);
        System.out.println("车辆转换的vin是："+vin);

        //判断vin是否存在缓存中
        if(sysCarCacheService.hashKey(vin)){
            //从Redis中获取车辆信息
            SysCar sysCar = sysCarCacheService.get(vin);
            //根据缓存车辆类型获取缓存报文
            SysCarType sysCarType = sysCarTypeCacheService.get(String.valueOf(sysCar.getCarType()));

            //获取报文模版信息
            List<MessageValue> messageValues = messageValueCacheService.get(String.valueOf(sysCarType.getMessageTemplateId()));
            for (MessageValue messageValue : messageValues) {
                //起始位下标
                Integer startIndex = messageValue.getMessageStartIndex() - 1;
                //结束位下标
                Integer endIndex = messageValue.getMessageEndIndex();
                //根据报文模版截取数据
                String value = vin.substring(startIndex, endIndex);
                //存入数据
                System.out.println("标签"+messageValue.getMessageLabel()+"值"+value);
                jsonObject.put(messageValue.getMessageLabel(), value);
            }
            sendKafkaMessage(jsonObject);
        }

    }



    private  void sendKafkaMessage(JSONObject jsonObject){
        ProducerRecord<Object, JSONObject> producerRecord = new ProducerRecord<>(topic, jsonObject);
        try {
            kafkaProducerConfig.kafkaProducer().send(new ProducerRecord<>(topic,jsonObject.toString()));
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
