package com.sunwin.metro.drools;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.sunwin.metro.constant.Descriptors;
import com.sunwin.metro.bean.RuleBase;
import com.sunwin.metro.governance.bean.Expression;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;

import static com.sunwin.metro.StartRule.getPercent;

/**
 * @author kane
 * @date 2020/01/27
 */
public class NormalRuleProcess extends BroadcastProcessFunction<List<Tuple4<String, String, Long,String>>, RuleBase, Tuple5<String, String, Integer, String,String>> {
    private static final Logger logger = LoggerFactory.getLogger(NormalRuleProcess.class);
    private RuleBase latestRuleBase;

    private transient MetroProcessor metroProcessor = null;

    public NormalRuleProcess(RuleBase ruleBase) {
        this.latestRuleBase = ruleBase;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        metroProcessor = new MetroProcessorWithRules(MetroProcessor.LOG_PKG);
        metroProcessor.loadRules(this.latestRuleBase);
    }

    @Override
    public void processElement(List<Tuple4<String, String, Long,String>> logEntry, ReadOnlyContext readOnlyContext, Collector<Tuple5<String, String, Integer, String,String>> collector) throws Exception {
        JSONArray array = new JSONArray();
        int fieldAll = 0;
        int countError = 0;
        int countAll = logEntry.size();
        for (Tuple4<String, String, Long,String> tuple2 : logEntry) {
            JSONObject jsonObject = JSONObject.parseObject(tuple2.f1);
            fieldAll = fieldAll + jsonObject.keySet().size();
                try {
                    List<JSONObject> result = metroProcessor.execute(jsonObject);
                    for (JSONObject item : result) {
                        String flag = item.getString("flag");
                        if (!"规则：".equals(flag)) {
                            array.add(item);
                            String[] split = flag.split("/");
                            countError = countError + split.length;
                        }
                    }
                } catch (Exception ex) {
                    logger.error("设备数据规则解析失败: " + ex.toString());
                }
        }

        JSONObject jsonObject = new JSONObject();
        //字段准确率
        jsonObject.put("字段准确性", getPercent(fieldAll - countError, fieldAll));
        //记录准确率
        jsonObject.put("记录准确性", getPercent(countAll - array.size(), countAll));
        if (array.size() > 0) {
            collector.collect(new Tuple5<>(logEntry.get(0).f0, jsonObject.toString(), countAll, array.toString(),logEntry.get(0).f3));
        } else {
            collector.collect(new Tuple5<>(logEntry.get(0).f0, jsonObject.toString(), countAll, null,logEntry.get(0).f3));
        }
    }

    @Override
    public void processBroadcastElement(RuleBase ruleBase, Context context, Collector<Tuple5<String, String, Integer, String,String>> collector) throws Exception {
        BroadcastState<String, RuleBase> ruleState = context.getBroadcastState(Descriptors.RULE_STATE_DESCRIPTOR);
        ruleState.put("logRule", ruleBase);
        if (latestRuleBase != null && StringUtils.equals(latestRuleBase.getHash(), ruleBase.getHash())) {
            logger.info("接收到相同的规则: {}", ruleBase.getName());
            return;
        }
        logger.info("更新 " + ruleBase.getDroolsRules().size() + " 规则, 规则名: " + ruleBase.getName());
        //加载规则到模型中
        if (metroProcessor != null) {
            if (!metroProcessor.loadRules(ruleBase)) {
                logger.error("加载规则失败");
            } else {
                logger.info("规则正在更新, hash值:{}", ruleBase.getHash());
                latestRuleBase = ruleBase;
            }
        } else {
            metroProcessor = new MetroProcessorWithRules(MetroProcessor.LOG_PKG);
            metroProcessor.loadRules(latestRuleBase);
            latestRuleBase = ruleBase;
        }
    }
}
