package com.yeejoin.amos.boot.module.jxiop.biz.emqx;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.FanWarningRecordServiceImpl;
import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.HealthStatusIndicatorServiceImpl;
import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.PvWarningRecordServiceImpl;
import com.yeejoin.amos.boot.module.jxiop.biz.tdMapper2.FanWaringRecordMapper;
import com.yeejoin.amos.boot.module.jxiop.biz.tdMapper2.PvWaringRecordMapper;
import com.yeejoin.amos.boot.module.jxiop.biz.tdengine.FanWarningRecord;
import com.yeejoin.amos.boot.module.jxiop.biz.tdengine.PvWarningRecord;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.typroject.tyboot.component.emq.EmqKeeper;
import org.typroject.tyboot.component.emq.EmqxListener;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@Component
@Slf4j
public class WarningRecordStatusMessage extends EmqxListener {

    @Autowired
    protected EmqKeeper emqKeeper;
    /**
     * 预警状态修改消息 - 标准化
     */
    public static final String WARNING_CHANGE_MESSAGE = "+/warning/change";


    @Autowired
    private FanWaringRecordMapper fanWaringRecordMapper;

    @Autowired
    private PvWaringRecordMapper pvWaringRecordMapper;

    @Autowired
    FanWarningRecordServiceImpl fanWarningRecordService;

    @Autowired
    PvWarningRecordServiceImpl pvWarningRecordService;

    @PostConstruct
    void init() throws Exception {
        emqKeeper.subscript(WARNING_CHANGE_MESSAGE, 2, this);
    }

    @Override
    public void processMessage(String topic, MqttMessage message) throws Exception {
        if (topic.contains(HealthStatusIndicatorServiceImpl.SMART_ANALYSE_PV)) {
            log.info("预警状态改变消息-光伏{}", new String(message.getPayload()));
            JSONArray ja = JSON.parseArray(new String(message.getPayload()));
            jxIopUpdatePv(ja);
        } else if (topic.contains(HealthStatusIndicatorServiceImpl.SMART_ANALYSE_FAN)) {
            log.info("预警状态改变消息-风电{}", new String(message.getPayload()));
            JSONArray ja = JSON.parseArray(new String(message.getPayload()));
            jxIopUpdateFan(ja);
        }
    }


    @Async("async")
    public void jxIopUpdateFan(JSONArray analysisResult) {
        log.info("修改预警状态信息：{}", analysisResult);
        List<JSONObject> taskList = JSONObject.parseArray(analysisResult.toJSONString(), JSONObject.class);
        List<String> traceIds = taskList.stream().map(t -> t.get("extAttr1").toString()).collect(Collectors.toList());
//        LambdaUpdateWrapper<IdxBizFanWarningRecord> lambda = new LambdaUpdateWrapper<>();
//        lambda.set(IdxBizFanWarningRecord::getDisposotionState, "已确认");
//        lambda.set(IdxBizFanWarningRecord::getStatus, "1");
//        lambda.set(IdxBizFanWarningRecord::getDisposotionDate, new Date());
//        lambda.in(IdxBizFanWarningRecord::getSequenceNbr, traceIds);
//        idxBizFanWarningRecordService.update(lambda);

        List<FanWarningRecord> list = new ArrayList<>();
        for (String traceId : traceIds) {
            FanWarningRecord fanWarningRecord = new FanWarningRecord();
            fanWarningRecord.setTs(Long.valueOf(traceId));
            fanWarningRecord.setDisposotionDate(DateUtil.now());
            fanWarningRecord.setStatus("1");
            fanWarningRecord.setDisposotionState("已确认");
            list.add(fanWarningRecord);
        }
        fanWaringRecordMapper.updateStatusByTs(list);

    }

    @Async("async")
    public void jxIopUpdatePv(JSONArray analysisResult) {
        log.info("修改预警状态信息：{}", analysisResult);
        List<JSONObject> taskList = JSONObject.parseArray(analysisResult.toJSONString(), JSONObject.class);
        List<String> traceIds = taskList.stream().map(t -> t.get("extAttr1").toString()).collect(Collectors.toList());
//        LambdaUpdateWrapper<IdxBizPvWarningRecord> lambda = new LambdaUpdateWrapper<>();
//        lambda.set(IdxBizPvWarningRecord::getDisposotionState, "已处置");
//        lambda.set(IdxBizPvWarningRecord::getStatus, "1");
//        lambda.set(IdxBizPvWarningRecord::getDisposotionDate, new Date());
//        lambda.in(IdxBizPvWarningRecord::getSequenceNbr, traceIds);
//        idxBizPvWarningRecordService.update(lambda);
        // td
        List<PvWarningRecord> list = new ArrayList<>();
        for (String traceId : traceIds) {
            PvWarningRecord pvWarningRecord = new PvWarningRecord();
            pvWarningRecord.setTs(Long.valueOf(traceId));
            pvWarningRecord.setDisposotionDate(DateUtil.now());
            pvWarningRecord.setStatus("1");
            pvWarningRecord.setDisposotionState("已确认");
            list.add(pvWarningRecord);
        }
        pvWaringRecordMapper.updateStatusByTs(list);
    }
}
