package com.sunwin.metro;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.google.common.collect.Lists;
import com.sunwin.metro.bean.*;
import com.sunwin.metro.constant.Descriptors;
import com.sunwin.metro.drools.MetroProcessor;
import com.sunwin.metro.drools.NormalRuleProcess;
import com.sunwin.metro.governance.bean.Expression;
import com.sunwin.metro.governance.bean.Integrity;
import com.sunwin.metro.governance.bean.Timeliness;
import com.sunwin.metro.governance.bean.Uniformity;
import com.sunwin.metro.process.BroadcastMeterMes;
import com.sunwin.metro.process.ErrorMetersProcess;
import com.sunwin.metro.process.PrometheusProcess;
import com.sunwin.metro.source.CreateMetroMes;
import com.sunwin.metro.source.MetroSubSource;

import com.sunwin.metro.utils.*;
import lombok.SneakyThrows;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.stringtemplate.v4.ST;
import org.stringtemplate.v4.STGroup;
import org.stringtemplate.v4.STGroupFile;
import pro.husk.mysql.MySQL;


import java.math.BigDecimal;
import java.sql.*;
import java.text.DecimalFormat;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

import static org.apache.flink.table.api.Expressions.$;


/**
 * @author xuYJ
 * @description: 程序入口 区间隧道风机 BAS_TVF（环境与设备监控系统）  排热风机 BAS_TEF（环境与设备监控系统）  不间断电源 ups（不间断电源系统） ups巡检仪 UPS_XJY（不间断电源系统）  雨水泵 BAS_AT（环境与设备监控系统）
 * @create: 2020-08-26 14:10
 */
public class StartRule {
    private static final Logger logger = LoggerFactory.getLogger(StartRule.class);

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置重启机制（固定重启机制，重启5次，每隔3秒）(测试过程可以关闭重启机制)
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(10, org.apache.flink.api.common.time.Time.of(5, TimeUnit.SECONDS)));
        // env.setStateBackend(new RocksDBStateBackend("hdfs://192.168.20.63:8020/flink/flink-checkpoint/rule/", true).getCheckpointBackend());
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        env.getCheckpointConfig().setCheckpointStorage("hdfs://192.168.20.63:8020/flink/flink-checkpoint/rules/");
        // 每隔1分钟进行启动一个检查点
        env.enableCheckpointing(10000 * 6);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000 * 3);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000 * 10);
        env.getCheckpointConfig().setTolerableCheckpointFailureNumber(100);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, build);
        //生成广播流
        DataStreamSource<Map<String, List<String>>> addSource = env
                .addSource(new MetroSubSource());
        //用来删除Prometheus失效的节点
        addSource.keyBy(new MeterKey()).process(new DeletePush()).print();
        BroadcastStream<Map<String, List<String>>> broadcast = addSource
                .setParallelism(1).broadcast(Descriptors.DESCRIPTOR);
        DataStreamSource<String> source = env.addSource(new CreateMetroMes());
        SingleOutputStreamOperator<Tuple2<String, String>> connect = source
                .connect(broadcast)
                .process(new BroadcastMeterMes());
        SingleOutputStreamOperator<Tuple2<String, String>> process = connect.process(new ErrorMetersProcess(Descriptors.ERROR_METER));
        connect.filter(new FilterFunction<Tuple2<String, String>>() {
            @Override
            public boolean filter(Tuple2<String, String> value) throws Exception {
                return !(value.f1.contains("true"));
            }
        })
                .keyBy((KeySelector<Tuple2<String, String>, String>) tuple2 -> tuple2.f0)
                //整理后的数据发送到Prometheus
                .process(new PrometheusProcess()).print();

        //DataStream<Tuple2<String, String>> sideOutput = process.getSideOutput(Descriptors.ERROR_METER);
        //SingleOutputStreamOperator<Tuple2<String, String>> filter = process.filter(new FilterFunction<Tuple2<String, String>>() {
        //    @Override
        //    public boolean filter(Tuple2<String, String> value) throws Exception {
        //        return "ups巡检仪".equals(value.f0);
        //    }
        //});
        //filter.addSink(org.apache.flink.connector.jdbc.JdbcSink.sink(
        //        "insert ignore  ups_xjy_rule (name,create_time , zdl , zdy , wd ) values (?,?,?,?,?)",
        //        (ps, t) -> {
        //            UpsXjyMes upsXjyMes = JSONObject.parseObject(t.f1, UpsXjyMes.class);
        //            ps.setString(1, upsXjyMes.getNum() + "_" + upsXjyMes.getStation() + "_" + upsXjyMes.getName());
        //            ps.setString(2, upsXjyMes.getTime());
        //            ps.setFloat(3, BigDecimal.valueOf(upsXjyMes.getZdl()).floatValue());
        //            ps.setFloat(4, BigDecimal.valueOf(upsXjyMes.getZdy()).floatValue());
        //            ps.setFloat(5, BigDecimal.valueOf(upsXjyMes.getWd()).floatValue());
        //
        //        },
        //        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        //                .withUrl("jdbc:mysql://192.168.20.61:3306/rule-engine?characterEncoding=utf-8&serverTimezone=UTC")
        //                .withDriverName("com.mysql.jdbc.Driver")
        //                .withPassword("root")
        //                .withUsername("root")
        //                .build()));

        //SingleOutputStreamOperator<Tuple2<String, String>> filter2 = process.filter(new FilterFunction<Tuple2<String, String>>() {
        //    @Override
        //    public boolean filter(Tuple2<String, String> value) throws Exception {
        //        return "排热风机".equals(value.f0);
        //    }
        //});
        //filter2.addSink(org.apache.flink.connector.jdbc.JdbcSink.sink(
        //        "insert ignore  bas_tvf_rule (name,create_time , x_vibr , y_vibr , faxlet ) values (?,?,?,?,?)",
        //        (ps, t) -> {
        //            BasTvfMes basTvfMes = JSONObject.parseObject(t.f1, BasTvfMes.class);
        //            ps.setString(1, basTvfMes.getNum() + "_" + basTvfMes.getStation() + "_" + basTvfMes.getName());
        //            ps.setString(2, basTvfMes.getTime());
        //            ps.setFloat(3, BigDecimal.valueOf(basTvfMes.getAi_x_vibr()).floatValue());
        //            ps.setFloat(4, BigDecimal.valueOf(basTvfMes.getAi_y_vibr()).floatValue());
        //            ps.setFloat(5, BigDecimal.valueOf(basTvfMes.getAi_faxlet()).floatValue());
        //
        //        },
        //        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        //                .withUrl("jdbc:mysql://192.168.20.61:3306/rule-engine?characterEncoding=utf-8&serverTimezone=UTC")
        //                .withDriverName("com.mysql.jdbc.Driver")
        //                .withPassword("root")
        //                .withUsername("root")
        //                .build()));

        String source1 =
                "CREATE TABLE mes_governance (" +
                        "  meter_type STRING ," +
                        "  meter_mes  STRING " +
                        ") WITH (" +
                        "'connector' = 'mysql-cdc'," +
                        "'hostname' = '192.168.20.61'," +
                        "'port' = '3306'," +
                        "'username' = 'root'," +
                        "'password' = 'root'," +
                        "'database-name' = 'rule-engine'," +
                        "'table-name' = 'mes_governance'" +
                        ")";

        tableEnv.executeSql(source1);
        Table testSink = tableEnv.from("mes_governance");
        DataStream<Tuple2<Boolean, Row>> stream = tableEnv.toRetractStream(testSink, Row.class);
        SingleOutputStreamOperator<Tuple2<String, String>> map = stream.filter((FilterFunction<Tuple2<Boolean, Row>>) value -> value.f0)
                .map(new MapFunction<Tuple2<Boolean, Row>, Tuple2<String, String>>() {
                    @Override
                    public Tuple2<String, String> map(Tuple2<Boolean, Row> value) throws Exception {
                        return new Tuple2<>(Objects.requireNonNull(value.f1.getField("meter_type")).toString(), Objects.requireNonNull(value.f1.getField("meter_mes")).toString());
                    }
                });
        DataStream<Tuple2<String, String>> union = process.union(map);
        long window = 60 * 15;
        //long window = 60;
        //专门适应采集及时性
        SingleOutputStreamOperator<List<Tuple4<String, String, Long, String>>> process3 = union.map(new MapFunction<Tuple2<String, String>, Tuple4<String, String, Long, String>>() {
            @Override
            public Tuple4<String, String, Long, String> map(Tuple2<String, String> value) throws Exception {
                String appType = JSONObject.parseObject(value.f1).getString("appType");
                return new Tuple4<>(value.f0 + "_" + appType, value.f1, System.currentTimeMillis(), appType);
            }
        }).keyBy(new KeySelector<Tuple4<String, String, Long, String>, String>() {
            @Override
            public String getKey(Tuple4<String, String, Long, String> value) throws Exception {
                return value.f0;
            }
        }).window(TumblingProcessingTimeWindows.of(Time.seconds(window))).process(new ProcessWindowFunction<Tuple4<String, String, Long, String>, List<Tuple4<String, String, Long, String>>, String, TimeWindow>() {
            @Override
            public void process(String s, Context context, Iterable<Tuple4<String, String, Long, String>> elements, Collector<List<Tuple4<String, String, Long, String>>> out) throws Exception {
                ArrayList<Tuple4<String, String, Long, String>> arrayList = Lists.newArrayList(elements);
                //窗口内部去重
                List<Tuple4<String, String, Long, String>> newList = new ArrayList<>();
                arrayList.stream().filter(distinctByKey(p -> p.f1))
                        .forEach(newList::add);
                out.collect(newList);
            }
        });

        DataStreamSource<Tuple2<String, JSONObject>> source2 = env.addSource(new RichSourceFunction<Tuple2<String, JSONObject>>() {
            Connection connection;
            PreparedStatement pStatement;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                connection = MysqlUtil.getCon();
            }

            @Override
            public void close() throws Exception {
                super.close();
                if (pStatement != null) {
                    pStatement.close();
                }
                if (connection != null) {
                    connection.close();
                }
            }

            @Override
            public void run(SourceContext<Tuple2<String, JSONObject>> ctx) throws Exception {
                while (true) {
                    try {
                        pStatement = connection.prepareStatement("SELECT *  FROM  t_data_rule;");
                        ResultSet results = pStatement.executeQuery();
                        while (results.next()) {
                            JSONObject jsonObject = new JSONObject();
                            String ruleId = String.valueOf(results.getInt("id"));
                            jsonObject.put("rule_id", ruleId);
                            jsonObject.put("model", results.getString("target_name"));
                            jsonObject.put("type", getMeterName(results.getString("equipment_type_name")));
                            jsonObject.put("appType", String.valueOf(results.getInt("application_id")));
                            jsonObject.put("mes", results.getString("expr"));
                            jsonObject.put("flag", results.getInt("flag"));
                            ctx.collect(new Tuple2<>("1组", jsonObject));
                        }
                    } catch (Exception e) {
                        System.out.println("连接postgres规则库异常");
                        e.printStackTrace();
                        connection = MysqlUtil.getCon2();
                    }
                    Thread.sleep(5000L);
                }
            }

            @SneakyThrows
            @Override
            public void cancel() {
                if (pStatement != null) {
                    pStatement.close();
                }
                if (connection != null) {
                    connection.close();
                }
            }
        });
        //侧边流输出3种规则
        SingleOutputStreamOperator<String> outputStreamOperator1 = source2.keyBy(new KeySelector<Tuple2<String, JSONObject>, String>() {
            @Override
            public String getKey(Tuple2<String, JSONObject> value) throws Exception {
                return value.f0;
            }
        }).process(new KeyedProcessFunction<String, Tuple2<String, JSONObject>, String>() {
            private transient MapState<String, String> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                MapStateDescriptor<String, String> descriptor = new MapStateDescriptor<>("mapState", TypeInformation.of(String.class), TypeInformation.of(String.class));
                mapState = getRuntimeContext().getMapState(descriptor);
            }

            @Override
            public void close() throws Exception {
                super.close();
            }

            @Override
            public void processElement(Tuple2<String, JSONObject> value, Context ctx, Collector<String> out) throws Exception {
                JSONObject object = value.f1;
                String ruleId = object.getString("rule_id");
                String model = object.getString("model");
                boolean flag = "1".equals(object.getString("flag"));
                String type = object.getString("type");
                String appType = object.getString("appType");

                if (mapState.isEmpty()) {
                    putMapState(value, ctx, object, ruleId, model, flag, type, appType);
                    mapState.put(ruleId, Md5Util.getMd5(value.f1.toString()));
                } else {
                    ArrayList<String> arrayList = new ArrayList<>();
                    mapState.keys().forEach(arrayList::add);
                    if (!arrayList.contains(ruleId)) {
                        putMapState(value, ctx, object, ruleId, model, flag, type, appType);
                    } else {
                        String s = mapState.get(ruleId);
                        if (!s.equals(Md5Util.getMd5(object.toString()))) {
                            putMapState(value, ctx, object, ruleId, model, flag, type, appType);
                        } else {
                            System.out.println("扫描到一样的规则");
                        }
                    }
                }
            }

            private void putMapState(Tuple2<String, JSONObject> value, KeyedProcessFunction<String, Tuple2<String, JSONObject>, String>.Context ctx, JSONObject object, String ruleId, String model, boolean flag, String type, String appType) throws Exception {
                if ("及时性（数据采集）".equals(model)) {
                    JSONObject jsonObject = JSONObject.parseObject(object.getString("mes"));
                    int inTime = jsonObject.getInteger("in");
                    ctx.output(Descriptors.IN_TIME_RULE_1, new Timeliness(ruleId, type, inTime, flag, appType));
                } else if ("及时性（数据入库）".equals(model)) {
                    JSONObject jsonObject = JSONObject.parseObject(object.getString("mes"));
                    int outTime = jsonObject.getInteger("out");
                    ctx.output(Descriptors.OUT_TIME_RULE_1, new Timeliness(ruleId, type, outTime, flag, appType));
                } else if ("完整性".equals(model)) {
                    List<String> mes = JSONArray.parseArray(object.getString("mes"), String.class);
                    ctx.output(Descriptors.COMPLETE_RULE_1, new Integrity(ruleId, type, mes, flag, appType));
                } else if ("一致性".equals(model)) {
                    List<JSONObject> mes = JSONArray.parseArray(object.getString("mes"), JSONObject.class);
                    ctx.output(Descriptors.UNIFORMITY_RULE_1, new Uniformity(ruleId, type, mes, flag, appType));
                } else if ("准确性".equals(model)) {
                    ctx.output(Descriptors.CURRENT_RULE_1, value.f1);
                }
                mapState.put(ruleId, Md5Util.getMd5(value.f1.toString()));
                System.out.println("成功加入缓存");
            }
        }).setParallelism(1);
        BroadcastStream<Timeliness> broadcast2 = outputStreamOperator1.getSideOutput(Descriptors.IN_TIME_RULE_1).broadcast(Descriptors.IN_TIME_RULE);
        BroadcastStream<Timeliness> broadcast3 = outputStreamOperator1.getSideOutput(Descriptors.OUT_TIME_RULE_1).broadcast(Descriptors.OUT_TIME_RULE);
        BroadcastStream<Integrity> broadcastStream = outputStreamOperator1.getSideOutput(Descriptors.COMPLETE_RULE_1).broadcast(Descriptors.COMPLETE_RULE);
        BroadcastStream<Uniformity> uniformityBroadcastStream = outputStreamOperator1.getSideOutput(Descriptors.UNIFORMITY_RULE_1).broadcast(Descriptors.UNIFORMITY_RULE);
        DataStream<JSONObject> current = outputStreamOperator1.getSideOutput(Descriptors.CURRENT_RULE_1);
        //进行数据采集时间及时性判断
        SingleOutputStreamOperator<Tuple5<String, String, Integer, String, String>> result2 = process3.connect(broadcast2).process(new BroadcastProcessFunction<List<Tuple4<String, String, Long, String>>, Timeliness, Tuple5<String, String, Integer, String, String>>() {
            @Override
            public void processElement(List<Tuple4<String, String, Long, String>> value, ReadOnlyContext ctx, Collector<Tuple5<String, String, Integer, String, String>> out) throws Exception {
                ReadOnlyBroadcastState<String, List<Timeliness>> broadcastState = ctx.getBroadcastState(Descriptors.IN_TIME_RULE);
                JSONArray inArray = new JSONArray();
                int countAll = value.size();
                for (Tuple4<String, String, Long, String> tuple2 : value) {
                    String[] s = tuple2.f0.split("_");
                    if (broadcastState.contains(getMeterName(s[0]))) {
                        List<Timeliness> timelinessList = broadcastState.get(getMeterName(s[0]));
                        JSONObject inJson = JSONObject.parseObject(tuple2.f1);
                        for (Timeliness item : timelinessList) {
                            if (item.getAppType().equals(tuple2.f3)) {
                                if (tuple2.f2 - inJson.getLong("eventTime") > item.getCollectTime()) {
                                    if (inJson.containsKey("ruleId")) {
                                        String ruleId = inJson.getString("ruleId");
                                        inJson.put("ruleId", ruleId + "_" + item.getRuleId());
                                    } else {
                                        inJson.put("ruleId", item.getRuleId());
                                    }
                                }
                            }
                        }
                        if (inJson.containsKey("ruleId")) {
                            inArray.add(inJson);
                        }
                    }
                }
                //输出准确率 + 异常数据集合
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("记录及时性（数据采集）", getPercent(countAll - inArray.size(), countAll));
                if (inArray.size() > 0) {
                    out.collect(new Tuple5<>(value.get(0).f0, jsonObject.toString(), countAll, inArray.toString(), value.get(0).f3));
                } else {
                    out.collect(new Tuple5<>(value.get(0).f0, jsonObject.toString(), countAll, null, value.get(0).f3));
                }
            }

            @Override
            public void processBroadcastElement(Timeliness value, Context ctx, Collector<Tuple5<String, String, Integer, String, String>> out) throws Exception {
                BroadcastState<String, List<Timeliness>> ctxBroadcastState = ctx.getBroadcastState(Descriptors.IN_TIME_RULE);
                if (value.isFlag()) {
                    if (ctxBroadcastState.contains(value.getType())) {
                        List<Timeliness> timelines = ctxBroadcastState.get(value.getType());
                        timelines.removeIf(item -> item.getRuleId().equals(value.getRuleId()));
                        timelines.add(value);
                        ctxBroadcastState.put(value.getType(), timelines);
                    } else {
                        ArrayList<Timeliness> arrayList = new ArrayList<>();
                        arrayList.add(value);
                        ctxBroadcastState.put(value.getType(), arrayList);
                    }
                } else {
                    if (ctxBroadcastState.contains(value.getType())) {
                        List<Timeliness> timelines = ctxBroadcastState.get(value.getType());
                        timelines.removeIf(timeliness1 -> timeliness1.toString().equals(value.toString()));
                        ctxBroadcastState.put(value.getType(), timelines);
                    }
                }
            }
        });

        //进行数据输出及时性判断
        SingleOutputStreamOperator<Tuple5<String, String, Integer, String, String>> result = process3.connect(broadcast3).process(new BroadcastProcessFunction<List<Tuple4<String, String, Long, String>>, Timeliness, Tuple5<String, String, Integer, String, String>>() {
            @Override
            public void processElement(List<Tuple4<String, String, Long, String>> value, ReadOnlyContext ctx, Collector<Tuple5<String, String, Integer, String, String>> out) throws Exception {
                ReadOnlyBroadcastState<String, List<Timeliness>> broadcastState = ctx.getBroadcastState(Descriptors.OUT_TIME_RULE);
                Long timestamp = ctx.timestamp();
                JSONArray outArray = new JSONArray();
                int countAll = value.size();
                for (Tuple4<String, String, Long, String> tuple2 : value) {
                    String[] s = tuple2.f0.split("_");
                    if (broadcastState.contains(getMeterName(s[0]))) {
                        List<Timeliness> timelinessList = broadcastState.get(getMeterName(s[0]));
                        JSONObject outJson = JSONObject.parseObject(tuple2.f1);
                        for (Timeliness item : timelinessList) {
                            if (item.getAppType().equals(tuple2.f3)) {
                                long l = ctx.currentProcessingTime();
                                if (l - timestamp > item.getCollectTime()) {
                                    if (outJson.containsKey("ruleId")) {
                                        String ruleId = outJson.getString("ruleId");
                                        outJson.put("ruleId", ruleId + "_" + item.getRuleId());
                                    } else {
                                        outJson.put("ruleId", item.getRuleId());
                                    }
                                }
                            }
                        }
                        if (outJson.containsKey("ruleId")) {
                            outArray.add(outJson);
                        }
                    }
                }
                //输出准确率 + 异常数据集合
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("记录及时性（数据入库）", getPercent(countAll - outArray.size(), countAll));
                if (outArray.size() > 0) {
                    out.collect(new Tuple5<>(value.get(0).f0, jsonObject.toString(), countAll, outArray.toString(), value.get(0).f3));
                } else {
                    out.collect(new Tuple5<>(value.get(0).f0, jsonObject.toString(), countAll, null, value.get(0).f3));
                }
            }

            @Override
            public void processBroadcastElement(Timeliness value, Context ctx, Collector<Tuple5<String, String, Integer, String, String>> out) throws Exception {
                BroadcastState<String, List<Timeliness>> ctxBroadcastState = ctx.getBroadcastState(Descriptors.OUT_TIME_RULE);
                if (value.isFlag()) {
                    if (ctxBroadcastState.contains(value.getType())) {
                        List<Timeliness> timelines = ctxBroadcastState.get(value.getType());
                        timelines.removeIf(item -> item.getRuleId().equals(value.getRuleId()));
                        timelines.add(value);
                        ctxBroadcastState.put(value.getType(), timelines);
                    } else {
                        ArrayList<Timeliness> arrayList = new ArrayList<>();
                        arrayList.add(value);
                        ctxBroadcastState.put(value.getType(), arrayList);
                    }
                } else {
                    if (ctxBroadcastState.contains(value.getType())) {
                        List<Timeliness> timelines = ctxBroadcastState.get(value.getType());
                        timelines.removeIf(timeliness1 -> timeliness1.toString().equals(value.toString()));
                        ctxBroadcastState.put(value.getType(), timelines);
                    }
                }
            }
        });

        //完整性
        SingleOutputStreamOperator<Tuple5<String, String, Integer, String, String>> result3 = process3.connect(broadcastStream).process(new BroadcastProcessFunction<List<Tuple4<String, String, Long, String>>, Integrity, Tuple5<String, String, Integer, String, String>>() {
            @Override
            public void processElement(List<Tuple4<String, String, Long, String>> value, ReadOnlyContext ctx, Collector<Tuple5<String, String, Integer, String, String>> out) throws Exception {
                ReadOnlyBroadcastState<String, List<Integrity>> broadcastState = ctx.getBroadcastState(Descriptors.COMPLETE_RULE);
                int filedCount = 0;
                int countError = 0;
                JSONArray array = new JSONArray();
                int countAll = value.size();
                for (Tuple4<String, String, Long, String> tuple2 : value) {
                    String[] s = tuple2.f0.split("_");
                    JSONObject jsonObject = JSONObject.parseObject(tuple2.f1);
                    Set<String> keySet = jsonObject.keySet();
                    filedCount = filedCount + keySet.size();
                    if (broadcastState.contains(getMeterName(s[0]))) {
                        List<Integrity> timelinessList = broadcastState.get(getMeterName(s[0]));
                        for (Integrity item : timelinessList) {
                            if (item.getAppType().equals(tuple2.f3)) {
                                List<String> fields = item.getFields();
                                for (String field : fields) {
                                    if (keySet.contains(field)) {
                                        if (null == jsonObject.get(field) || jsonObject.getString(field).isEmpty() || jsonObject.getString(field).trim().isEmpty() || "null".equals(jsonObject.getString(field)) || "NULL".equals(jsonObject.getString(field))) {
                                            jsonObject.putIfAbsent(field, null);
                                            countError++;
                                            if (jsonObject.containsKey("ruleId")) {
                                                String ruleId = jsonObject.getString("ruleId");
                                                jsonObject.put("ruleId", ruleId + "_" + item.getRuleId());
                                            } else {
                                                jsonObject.put("ruleId", item.getRuleId());
                                            }
                                            break;
                                        }
                                    } else {
                                        countError++;
                                        if (jsonObject.containsKey("ruleId")) {
                                            String ruleId = jsonObject.getString("ruleId");
                                            jsonObject.put("ruleId", ruleId + "_" + item.getRuleId());
                                        } else {
                                            jsonObject.put("ruleId", item.getRuleId());
                                        }
                                        break;
                                    }
                                }
                            }
                        }
                        if (jsonObject.containsKey("ruleId")) {
                            array.add(jsonObject);
                        }
                    }
                }
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("字段完整性", getPercent(filedCount - countError, filedCount));
                jsonObject.put("记录完整性", getPercent(countAll - array.size(), countAll));
                if (array.size() > 0) {
                    out.collect(new Tuple5<>(value.get(0).f0, jsonObject.toString(), countAll, array.toString(SerializerFeature.WriteMapNullValue), value.get(0).f3));
                } else {
                    out.collect(new Tuple5<>(value.get(0).f0, jsonObject.toString(), countAll, null, value.get(0).f3));
                }
            }

            @Override
            public void processBroadcastElement(Integrity value, Context ctx, Collector<Tuple5<String, String, Integer, String, String>> out) throws Exception {
                BroadcastState<String, List<Integrity>> broadcastState = ctx.getBroadcastState(Descriptors.COMPLETE_RULE);
                if (value.isFlag()) {
                    if (broadcastState.contains(value.getType())) {
                        List<Integrity> timelines = broadcastState.get(value.getType());
                        timelines.removeIf(item -> item.getRuleId().equals(value.getRuleId()));
                        timelines.add(value);
                        broadcastState.put(value.getType(), timelines);
                    } else {
                        ArrayList<Integrity> arrayList = new ArrayList<>();
                        arrayList.add(value);
                        broadcastState.put(value.getType(), arrayList);
                    }
                } else {
                    if (broadcastState.contains(value.getType())) {
                        List<Integrity> timelines = broadcastState.get(value.getType());
                        timelines.removeIf(timeliness1 -> timeliness1.toString().equals(value.toString()));
                        broadcastState.put(value.getType(), timelines);
                    }
                }
            }
        });

        //一致性
        SingleOutputStreamOperator<Tuple3<String, List<Tuple3<String, String, String>>, String>> process2 = process3.connect(uniformityBroadcastStream).process(new BroadcastProcessFunction<List<Tuple4<String, String, Long, String>>, Uniformity, Tuple3<String, List<Tuple3<String, String, String>>, String>>() {
            @Override
            public void processElement(List<Tuple4<String, String, Long, String>> value, ReadOnlyContext ctx, Collector<Tuple3<String, List<Tuple3<String, String, String>>, String>> out) throws Exception {
                ArrayList<Tuple3<String, String, String>> arrayList = new ArrayList<>();
                ReadOnlyBroadcastState<String, List<Uniformity>> broadcastState = ctx.getBroadcastState(Descriptors.UNIFORMITY_RULE);
                for (Tuple4<String, String, Long, String> tuple2 : value) {
                    String[] s = tuple2.f0.split("_");
                    JSONObject jsonObject = JSONObject.parseObject(tuple2.f1);
                    if (broadcastState.contains(getMeterName(s[0]))) {
                        JSONObject keys = new JSONObject();
                        List<Uniformity> timelinessList = broadcastState.get(getMeterName(s[0]));
                        for (Uniformity item : timelinessList) {
                            if (item.getAppType().equals(tuple2.f3)) {
                                List<JSONObject> fields = item.getFields();
                                for (JSONObject field : fields) {
                                    if (jsonObject.containsKey(field.getString("dataField")) && (null != jsonObject.get(field.getString("dataField")))) {
                                        if (field.containsKey("dataType")) {
                                            keys.put(item.getRuleId() + "-" + field.getString("dataField") + "-" + field.getString("dataType"), jsonObject.get(field.getString("dataField")));
                                        } else {
                                            keys.put(item.getRuleId() + "-" + field.getString("dataField") + "-" + "all", jsonObject.get(field.getString("dataField")));
                                        }
                                    }
                                }
                            }
                        }
                        arrayList.add(new Tuple3<>(tuple2.f0, tuple2.f1, keys.toString()));
                    }
                }
                if (arrayList.isEmpty()) {
                    out.collect(new Tuple3<>(value.get(0).f0 + "_" + value.size(), arrayList, value.get(0).f3));
                } else {
                    out.collect(new Tuple3<>(value.get(0).f0, arrayList, value.get(0).f3));
                }
            }

            @Override
            public void processBroadcastElement(Uniformity value, Context ctx, Collector<Tuple3<String, List<Tuple3<String, String, String>>, String>> out) throws Exception {
                BroadcastState<String, List<Uniformity>> broadcastState = ctx.getBroadcastState(Descriptors.UNIFORMITY_RULE);
                if (value.isFlag()) {
                    if (broadcastState.contains(value.getType())) {
                        List<Uniformity> timelines = broadcastState.get(value.getType());
                        timelines.removeIf(item -> item.getRuleId().equals(value.getRuleId()));
                        timelines.add(value);
                        broadcastState.put(value.getType(), timelines);
                    } else {
                        ArrayList<Uniformity> arrayList = new ArrayList<>();
                        arrayList.add(value);
                        broadcastState.put(value.getType(), arrayList);
                    }
                } else {
                    if (broadcastState.contains(value.getType())) {
                        List<Uniformity> timelines = broadcastState.get(value.getType());
                        timelines.removeIf(timeliness1 -> timeliness1.toString().equals(value.toString()));
                        broadcastState.put(value.getType(), timelines);
                    }
                }
            }
        });

        //输出一致性结果
        SingleOutputStreamOperator<Tuple5<String, String, Integer, String, String>> result4 = process2.map(new MapFunction<Tuple3<String, List<Tuple3<String, String, String>>, String>, Tuple5<String, String, Integer, String, String>>() {
            @Override
            public Tuple5<String, String, Integer, String, String> map(Tuple3<String, List<Tuple3<String, String, String>>, String> value) throws Exception {
                int filedCount = 0;
                int countError = 0;
                JSONArray array = new JSONArray();
                int countAll = value.f1.size();
                for (Tuple3<String, String, String> tuple3 : value.f1) {
                    JSONObject input = JSONObject.parseObject(tuple3.f2);
                    JSONObject output = JSONObject.parseObject(tuple3.f1);
                    filedCount = filedCount + output.keySet().size();
                    for (String s : input.keySet()) {
                        String[] split = s.split("-");
                        String ruleId = split[0];
                        String key = split[1];
                        String type = split[2];
                        if (output.containsKey(key)) {
                            if (!(output.get(key).equals(input.get(s)) && DataType.getType(output.get(key)).get("类型").equals(type)) && (!"all".equals(type))) {
                                countError++;
                                if (output.containsKey("ruleId")) {
                                    output.put("ruleId", output.getString("ruleId") + "_" + ruleId);
                                } else {
                                    output.put("ruleId", ruleId);
                                }
                            }
                        }
                    }
                    if (output.containsKey("ruleId")) {
                        array.add(output);
                    }
                }
                JSONObject jsonObject = new JSONObject();
                jsonObject.put("字段一致性", getPercent(filedCount - countError, filedCount));
                jsonObject.put("记录一致性", getPercent(countAll - array.size(), countAll));
                if (array.size() > 0) {
                    return new Tuple5<>(value.f0, jsonObject.toString(), countAll, array.toString(), value.f2);
                } else if (value.f1.isEmpty()) {
                    String[] split = value.f0.split("_");
                    return new Tuple5<>(split[0], "{\"记录一致性\":\"100.00%\",\"字段一致性\":\"100.00%\"}", Integer.parseInt(split[2]), null, value.f2);
                } else {
                    return new Tuple5<>(value.f0, jsonObject.toString(), countAll, null, value.f2);
                }
            }
        });


        //准确性
        BroadcastStream<RuleBase> broadcast1 = current.keyBy((KeySelector<JSONObject, String>) value -> value.getString("model")).process(new KeyedProcessFunction<String, JSONObject, RuleBase>() {
            private transient MapState<String, DroolsRule> mapState;
            private boolean flag;

            @Override
            public void open(Configuration parameters) throws Exception {
                MapStateDescriptor<String, DroolsRule> descriptor = new MapStateDescriptor<>("mapState", TypeInformation.of(String.class), TypeInformation.of(DroolsRule.class));
                mapState = getRuntimeContext().getMapState(descriptor);
                flag = true;
                super.open(parameters);
            }

            @Override
            public void processElement(JSONObject tuple2, Context ctx, Collector<RuleBase> out) throws Exception {
                STGroup stg = new STGroupFile(Descriptors.RULE_ADDRESS);
                if (flag) {
                    ST sqlTemplate = stg.getInstanceOf("sqlTemplate");
                    String ruleId = "0000000001";
                    //计算表达式
                    String expression = "(ai_faxlet<-1000000)";
                    String type = "BasTefMes";
                    String state = "1";
                    sqlTemplate.add("ruleId", ruleId);
                    sqlTemplate.add("expression", expression);
                    sqlTemplate.add("class", type);
                    String script = sqlTemplate.render();
                    mapState.put(ruleId, new DroolsRule(ruleId, "logrules", script, state));
                    flag = false;
                }
                JSONObject jsonObject = JSONObject.parseObject(tuple2.getString("mes"));
                ST sqlTemplate = stg.getInstanceOf("sqlTemplate");
                String ruleId = tuple2.getString("rule_id");
                String expression = jsonObject.getString("expression");
                String type = tuple2.getString("type");
                String state = tuple2.getString("flag");
                sqlTemplate.add("ruleId", ruleId);
                sqlTemplate.add("expression", expression);
                sqlTemplate.add("class", type);
                String script = sqlTemplate.render();
                if ("1".equals(state)) {
                    mapState.put(ruleId, new DroolsRule(ruleId, "logrules", script, state));
                } else {
                    if (!mapState.isEmpty()) {
                        if (mapState.contains(ruleId)) {
                            mapState.remove(ruleId);
                        }
                    }
                }
                int count = 0;
                if (!mapState.isEmpty()) {
                    Iterable<DroolsRule> values = mapState.values();
                    JSONArray array = new JSONArray();
                    for (DroolsRule value : values) {
                        count++;
                        array.add(value);
                    }
                    RuleBase ruleBase = RuleBase.createRuleBase(array);
                    out.collect(ruleBase);
                }
            }
        }).setParallelism(1).broadcast(Descriptors.RULE_STATE_DESCRIPTOR);
        //获取初始化规则
        RuleBase ruleBase = getInitRuleBase();
        logger.debug("第一次初始化规则 : " + ruleBase.toString());

        SingleOutputStreamOperator<Tuple5<String, String, Integer, String, String>> result5 = process3.connect(broadcast1).process(new NormalRuleProcess(ruleBase));

        DataStream<Tuple5<String, String, Integer, String, String>> unionRes = result2.union(result3).union(result4).union(result5).union(result);
        SingleOutputStreamOperator<Tuple6<String, String, Integer, String, String, String>> operator = unionRes.map(new MapFunction<Tuple5<String, String, Integer, String, String>, Tuple6<String, String, Integer, String, String, String>>() {
            @Override
            public Tuple6<String, String, Integer, String, String, String> map(Tuple5<String, String, Integer, String, String> value) throws Exception {
                return new Tuple6<>(value.f0, value.f1, value.f2, value.f3, UUID.randomUUID().toString(), value.f4);
            }
        });

        operator.addSink(new RichSinkFunction<Tuple6<String, String, Integer, String, String, String>>() {
            private Connection con;
            private PreparedStatement statement;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                con = MysqlUtil.getCon();

            }

            @Override
            public void close() throws Exception {
                super.close();
                if (statement != null) {
                    statement.close();
                }
                if (con != null) {
                    con.close();
                }
            }

            @Override
            public void invoke(Tuple6<String, String, Integer, String, String, String> value, Context context) throws Exception {
                Date date = new Date();
                SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String dateStr = format.format(date);
                String[] s = value.f0.split("_");
                String device = s[0];
                String appType = value.f5;
                JSONObject jsonObject = JSONObject.parseObject(value.f1);
                int count = value.f2;
                String resMes = value.f3;
                String uuid = value.f4;
                char e = '%';
                try {
                    String sql = "insert into governance_result (app_type,model,field_res,record_res,rule_res,mes_count,create_time,uuid,device_type) values (?,?,?,?,?,?,?,?,?) ON CONFLICT(app_type,model,create_time,device_type) DO NOTHING";
                    statement = con.prepareStatement(sql);
                    statement.setString(1, appType);
                    if (jsonObject.containsKey("记录一致性")) {
                        Float recordRes = deleteChar(jsonObject.getString("记录一致性"), e);
                        Float fieldRes = deleteChar(jsonObject.getString("字段一致性"), e);
                        statement.setString(2, "一致性");
                        statement.setFloat(3, fieldRes);
                        statement.setFloat(4, recordRes);
                        statement.setString(5, resMes);
                    } else if (jsonObject.containsKey("记录完整性")) {
                        Float recordRes = deleteChar(jsonObject.getString("记录完整性"), e);
                        Float fieldRes = deleteChar(jsonObject.getString("字段完整性"), e);
                        statement.setString(2, "完整性");
                        statement.setFloat(3, fieldRes);
                        statement.setFloat(4, recordRes);
                        statement.setString(5, resMes);
                    } else if (jsonObject.containsKey("记录准确性")) {
                        Float recordRes = deleteChar(jsonObject.getString("记录准确性"), e);
                        Float fieldRes = deleteChar(jsonObject.getString("字段准确性"), e);
                        statement.setString(2, "准确性");
                        statement.setFloat(3, fieldRes);
                        statement.setFloat(4, recordRes);
                        statement.setString(5, resMes);
                    } else if (jsonObject.containsKey("记录及时性（数据入库）")) {
                        statement.setString(2, "及时性（数据入库）");
                        Float out = deleteChar(jsonObject.getString("记录及时性（数据入库）"), e);
                        statement.setFloat(3, -1);
                        statement.setFloat(4, out);
                        statement.setString(5, resMes);
                    } else if (jsonObject.containsKey("记录及时性（数据采集）")) {
                        statement.setString(2, "及时性（数据采集）");
                        Float in = deleteChar(jsonObject.getString("记录及时性（数据采集）"), e);
                        statement.setFloat(3, -1);
                        statement.setFloat(4, in);
                        statement.setString(5, resMes);
                    }
                    statement.setInt(6, count);
                    statement.setString(7, dateStr);
                    statement.setString(8, uuid);
                    statement.setString(9, device);
                    statement.addBatch();
                    statement.executeBatch();
                    con.commit();
                } catch (Exception e1) {
                    System.out.println("存储postgres规则结果异常");
                    e1.printStackTrace();
                    con = MysqlUtil.getCon2();
                }
            }
        });
        operator.filter(new FilterFunction<Tuple6<String, String, Integer, String, String, String>>() {
            @Override
            public boolean filter(Tuple6<String, String, Integer, String, String, String> value) throws Exception {
                return null != value.f3;
            }
        }).flatMap(new FlatMapFunction<Tuple6<String, String, Integer, String, String, String>, Tuple3<String, String, String>>() {
            @Override
            public void flatMap(Tuple6<String, String, Integer, String, String, String> value, Collector<Tuple3<String, String, String>> out) throws Exception {
                JSONArray jsonArray = JSONArray.parseArray(value.f3);
                for (Object o : jsonArray) {
                    JSONObject jsonObject = JSONObject.parseObject(JSONObject.toJSONString(o, SerializerFeature.WriteMapNullValue));

                    if (jsonObject.containsKey("ruleId")) {
                        String ruleIds = jsonObject.getString("ruleId");
                        String[] s = ruleIds.split("_");
                        for (String s1 : s) {
                            out.collect(new Tuple3<>(s1.trim(), jsonObject.toString(SerializerFeature.WriteMapNullValue), value.f4));
                        }
                    } else {
                        String ruleIds = jsonObject.getString("flag");
                        String[] s = ruleIds.split("-/");
                        for (String s1 : s) {
                            if (s1.contains("规则：")) {
                                s1 = s1.replace("规则：", "");
                            }
                            out.collect(new Tuple3<>(s1.trim(), jsonObject.toString(), value.f4));
                        }
                    }
                }
            }
        }).addSink(org.apache.flink.connector.jdbc.JdbcSink.sink(
                "insert into rule_mes (rule_id,rule_mes,create_time,uuid) values (?,?,?,?) ON CONFLICT(rule_id,rule_mes,uuid) DO NOTHING",
                (ps, t) -> {
                    Date date = new Date();
                    SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    String dateStr = format.format(date);
                    ps.setString(1, t.f0);
                    ps.setString(2, t.f1);
                    ps.setString(3, dateStr);
                    ps.setString(4, t.f2);
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:postgresql://192.168.20.72:5432/urtmpdb?characterEncoding=utf-8&serverTimezone=UTC")
                        .withDriverName("org.postgresql.Driver")
                        .withPassword("sw12345")
                        .withUsername("postgres")
                        .build()));
        env.execute();
    }

    /**
     * 删除某个字符
     */
    public static Float deleteChar(String str, char delChar) {
        StringBuilder delStr = new StringBuilder();
        char[] bytes = str.toCharArray();
        int iSize = bytes.length;
        for (int i = bytes.length - 1; i >= 0; i--) {
            if (bytes[i] == delChar) {
                if (iSize - 1 - i >= 0) {
                    System.arraycopy(bytes, i + 1, bytes, i, iSize - 1 - i);
                }
                iSize--;
            }
        }
        for (int i = 0; i < iSize; i++) {
            delStr.append(bytes[i]);
        }
        return Float.parseFloat(delStr.toString());
    }

    /**
     * 用来初始化规则，防止广播流规则为空
     */
    private static RuleBase getInitRuleBase() throws SQLException {
        JSONArray array = new JSONArray();
        STGroup stg = new STGroupFile(Descriptors.RULE_ADDRESS);
        ST sqlTemplate = stg.getInstanceOf("sqlTemplate");
        String ruleId = "0000000002";
        String expression = "(ai_faxlet<-1000000)";
        String type = "BasTefMes";
        String state = "1";
        sqlTemplate.add("ruleId", ruleId);
        sqlTemplate.add("expression", expression);
        sqlTemplate.add("class", type);
        String script = sqlTemplate.render();
        array.add(new com.sunwin.metro.bean.DroolsRule(ruleId, type, script, state));
        System.out.println(array);
        return RuleBase.createRuleBase(array);
    }

    /**
     * 求百分比
     */
    public static String getPercent(int x, int y) {
        double d1 = x * 1.0;
        double d2 = y * 1.0;
        // 设置保留几位小数， “.”后面几个零就保留几位小数，这里设置保留四位小数
        DecimalFormat decimalFormat = new DecimalFormat("##.00%");
        if (d1 == 0) {
            return "0%";
        }
        return decimalFormat.format(d1 / d2);
    }

    private static String getMeterName(String name) {
        switch (name) {
            case "隧道风机":
                return "BasTefMes";
            case "雨水泵":
                return "BasAtMes";
            case "排热风机":
                return "BasTvfMes";
            case "ups巡检仪":
                return "UpsXjyMes";
            case "不间断电源":
                return "UpsMes";
            default:
                return name;
        }

    }

    /**
     * 集合去重
     */
    private static <T> Predicate<T> distinctByKey(Function<? super T, ?> keyExtractor) {
        Map<Object, Boolean> seen = new ConcurrentHashMap<>(16);
        //putIfAbsent方法添加键值对，如果map集合中没有该key对应的值，则直接添加，并返回null，如果已经存在对应的值，则依旧为原来的值。
        //如果返回null表示添加数据成功(不重复)，不重复(null==null :TRUE)
        return t -> seen.putIfAbsent(keyExtractor.apply(t), Boolean.TRUE) == null;
    }

}
