package examples.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.broadcast.Broadcast;
import scala.collection.immutable.Set;

import java.util.Map;

public class StatisticsUtil {

    public static void featureCoverRate(String record, Broadcast<Set<String>> featureNames, Map<String, Long> allMap, Map<String, Long> isShareMap, Map<String, Long> isReturnNoSelfMap) {
        String[] rSplit = record.split("\t");
        // 异常样本统计
        if (rSplit.length != 3) {
            StatisticsUtil.mapKeyAddOne(allMap, "errorSampleNum");
            StatisticsUtil.mapKeyAddOne(isShareMap, "errorSampleNum");
            StatisticsUtil.mapKeyAddOne(isReturnNoSelfMap, "errorSampleNum");
            return;
        }

        // 是否推荐场景
        JSONObject logKey = JSON.parseObject(rSplit[0]);
        String page = logKey.getString("page");
        String recommendPageType = logKey.getString("recommendpagetype");

        JSONObject labelKey = JSON.parseObject(rSplit[1]);
        String isShare = labelKey.getString("is_share");
        String isReturnNoSelf = logKey.getString("is_return_noself");

        if (StatisticsUtil.isRecommendScene(page, recommendPageType)) {
            StatisticsUtil.mapKeyAddOne(allMap, "recommendSampleNum");
            if (StringUtils.equals("1", isShare)) {
                StatisticsUtil.mapKeyAddOne(isShareMap, "recommendSampleNum");
            }
            if (StringUtils.equals("1", isReturnNoSelf)) {
                StatisticsUtil.mapKeyAddOne(isReturnNoSelfMap, "recommendSampleNum");
            }

        }

        // 正常数据统计
        StatisticsUtil.mapKeyAddOne(allMap, "normalSampleNum");
        if (StringUtils.equals("1", isShare)) {
            StatisticsUtil.mapKeyAddOne(isShareMap, "normalSampleNum");
        }
        if (StringUtils.equals("1", isReturnNoSelf)) {
            StatisticsUtil.mapKeyAddOne(isReturnNoSelfMap, "normalSampleNum");
        }

        // 特征大于0统计
        JSONObject feature = JSON.parseObject(rSplit[2]);
        for (String featureName : featureNames) {
            if (feature.containsKey(featureName) && feature.getDoubleValue(featureName) > 0) {
                StatisticsUtil.mapKeyAddOne(allMap, featureName);

                if (StringUtils.equals("1", isShare)) {
                    StatisticsUtil.mapKeyAddOne(isShareMap, featureName);
                }
                if (StringUtils.equals("1", isReturnNoSelf)) {
                    StatisticsUtil.mapKeyAddOne(isReturnNoSelfMap, featureName);
                }

            }
        }
    }

    private static boolean isRecommendScene(String page, String recommendPageType) {
        if (StringUtils.equals("详情后沉浸页", page)) {
            return true;
        } else if (StringUtils.equals("回流后沉浸页&内页feed", page) && StringUtils.isNotBlank(recommendPageType) && recommendPageType.endsWith("-pages/user-videos-share-recommend-detail")) {
            return true;
        }
        return false;
    }

    private static void mapKeyAddOne(Map<String, Long> resultMap, String key) {
        if (!resultMap.containsKey(key)) {
            resultMap.put(key, 0L);
        }

        resultMap.put(key, resultMap.get(key) + 1);
    }

}