package com.sunwin.metro.bean;


import cn.hutool.json.JSONObject;
import lombok.*;
import org.apache.flink.types.Row;

import java.math.BigDecimal;

/**
 * @author kane
 */
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
@ToString(exclude="rule")
public class WindowRule {

    private int ruleRelationId;

    private String type;

    private String aggField;

    private String aggFunction;

    private String operator;

    private BigDecimal threshold;

    private Long windowSeconds;

    private String controlCmd;

    private Long windowSecondsSplit;

    private int windowCounts;

    private int windowCountsSplit;

    private String timeState;

    private String countState;



    public static String getWindowRule(Row rule) {
        com.alibaba.fastjson.JSONObject jsonObject = new com.alibaba.fastjson.JSONObject();
        if(rule.getField("rule_id")!=null){
            jsonObject.put("rule_id",rule.getField("rule_id"));
        }
        if(rule.getField("type")!=null){
            jsonObject.put("type",rule.getField("type"));
        }
        if(rule.getField("time_state")!=null){
            jsonObject.put("time_state",rule.getField("time_state"));
        }
        if(rule.getField("count_state")!=null){
            jsonObject.put("count_state",rule.getField("count_state"));
        }
        if(rule.getField("control_cmd")!=null){
            jsonObject.put("control_cmd",rule.getField("control_cmd"));
        }
        if(rule.getField("agg_function")!=null){
            jsonObject.put("agg_function",rule.getField("agg_function"));
        }
        if(rule.getField("operator")!=null){
            jsonObject.put("operator",rule.getField("operator"));
        }
        if(rule.getField("threshold")!=null){
            jsonObject.put("threshold",rule.getField("threshold"));
        }
        if(rule.getField("agg_field")!=null){
            jsonObject.put("agg_field",rule.getField("agg_field"));
        }
        if(rule.getField("win_second")!=null){
            jsonObject.put("win_second",rule.getField("win_second"));
        }
        if(rule.getField("win_count")!=null){
            jsonObject.put("win_count",rule.getField("win_count"));
        }
        if(rule.getField("win_count_split")!=null){
            jsonObject.put("win_count_split",rule.getField("win_count_split"));
        }
        if(rule.getField("win_second_split")!=null){
            jsonObject.put("win_second_split",rule.getField("win_second_split"));
        }
       return jsonObject.toString();
    }

    public WindowRule(JSONObject rule) {
        this.ruleRelationId = rule.getInt("rule_id");
        this.type=rule.getStr("type");
        this.timeState=rule.getStr("time_state","0");
        this.countState=rule.getStr("count_state","0");
        this.controlCmd=rule.getStr("control_cmd");
        this.aggFunction=rule.getStr("agg_function");
        this.operator=rule.getStr("operator","↑/↓");
        this.threshold=new BigDecimal(rule.getStr("threshold","0"));
        this.aggField=rule.getStr("agg_field");
        this.windowSeconds=rule.getLong("win_second",0L);
        this.windowSecondsSplit=rule.getLong("win_second_split",5L);
        this.windowCounts=rule.getInt("win_count",0);
        this.windowCountsSplit=rule.getInt("win_count_split",1);

    }



    /**
    判断Event是否匹配该规则
    */
    public boolean filter(Event event, WindowRule windowRule){
        String eType = windowRule.getType();
        return event.getType().equals(eType);
    }

    public boolean isWindowTime(){
        return "1".equals(this.timeState.trim());
    }



    public boolean apply(BigDecimal comparisonValue) {
        switch (operator) {
          case "=":
            return comparisonValue.compareTo(threshold) == 0;
          case "!=":
            return comparisonValue.compareTo(threshold) != 0;
          case ">":
            return comparisonValue.compareTo(threshold) > 0;
          case "<":
            return comparisonValue.compareTo(threshold) < 0;
          case "<=":
            return comparisonValue.compareTo(threshold) <= 0;
          case ">=":
            return comparisonValue.compareTo(threshold) >= 0;
          default:
            throw new RuntimeException("Unknown limit operator type: " + operator);
        }
    }



}
