PAS/src/main/java/cn/cslg/pas/service/business/es/EsCountService.java

package cn.cslg.pas.service.business.es;

import cn.cslg.pas.common.dto.business.EsCountDTO;
import cn.cslg.pas.common.dto.business.EsCountDetailDTO;
import cn.cslg.pas.common.utils.parseQueryToTree.expressManager;
import cn.cslg.pas.common.utils.parseQueryToTree.operateNode;
import cn.cslg.pas.common.utils.parseQueryToTree.treeNode;
import cn.cslg.pas.common.vo.EsConfigVO;
import cn.cslg.pas.common.vo.business.EsCountVO;
import cn.cslg.pas.domain.es.Patent;
import cn.cslg.pas.factorys.EsAnalysisBuilderFactory.EsAnalysisBuilderFactory;
import cn.cslg.pas.factorys.EsAnalysisBuilderFactory.IEsAnalysisBuilder;
import cn.cslg.pas.factorys.EsCountBuilderFactory.EsCountBuilderFactory;
import cn.cslg.pas.factorys.EsCountBuilderFactory.IEsCountBuilder;
import cn.cslg.pas.service.business.CommonService;
import cn.cslg.pas.service.query.FormatQueryService;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch._types.query_dsl.QueryBuilders;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import com.alibaba.fastjson.JSON;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy})
public class EsCountService {
    private final List<String> childList = Arrays.asList("field");
    private final List<String> nestedList = Arrays.asList("PA", "IN", "PE", "SAT", "MAT", "SRH", "MRH");
    private final List<String> dateList = Arrays.asList("PD", "AD", "GD");
    private final List<String> numberList = Arrays.asList("QPN", "QDPN", "SFN", "IFN", "PFN");


    private final ElasticsearchClient client;

    @Autowired
    private EsCountBuilderFactory esCountBuilderFactory;
    @Autowired
    private EsAnalysisBuilderFactory esAnalysisBuilderFactory;
    @Autowired
    private FormatQueryService formatQueryService;

    /**
     * 查询专利库中的专利分组聚合统计
     *
     * @param countVOS
     * @return
     * @throws Exception
     */
    public EsCountDTO esCountSearch(List<EsCountVO> countVOS) throws Exception {
        EsCountDTO esCountDTO = new EsCountDTO();
        List<EsCountDetailDTO> detailDTOS = new ArrayList<>();
        for (EsCountVO vo : countVOS) {
            String field = vo.getField();
            Integer topN = vo.getTopN();
            String condition = vo.getCondition();
            //查询es返回数据
            SearchRequest.Builder builder = new SearchRequest.Builder();
            Aggregation aggregation = this.selectAggregation(builder, vo);
            if (StringUtils.isNotEmpty(condition)) {
                //解析检索条件
                treeNode tree = expressManager.getInstance().Parse(condition, false);
                //从es中检索数据
                Query query = formatQueryService.EsQueryToQuery((operateNode) tree, "patent");
                Aggregation filtersAgg = null;
                if (aggregation != null) {
                    filtersAgg = new Aggregation.Builder().filters(new FiltersAggregation.Builder()
                            .filters(i -> i.array(Arrays.asList(query))).build())
                            .aggregations(new HashMap() {{
                                put("filters_agg", aggregation);
                            }}).build();
                } else {
                    filtersAgg = AggregationBuilders.filter(i -> i.bool(j -> j.must(query)));
                }
                builder.aggregations("Agg", filtersAgg);
            } else {
                builder.aggregations("Agg", aggregation);
            }
            SearchResponse<Patent> response = client.search(builder.build(), Patent.class);

            Aggregate agg = response.aggregations().get("Agg");
            if (StringUtils.isNotEmpty(condition)) {
                if (StringUtils.isNotEmpty(field)) {
                    List<FiltersBucket> filtersBuckets = agg.filters().buckets().array();
                    if (dateList.contains(field)) {
                        filtersBuckets.forEach(filtersBucket -> {
                            this.getFiltersCountDTO(filtersBucket, condition, detailDTOS);
                            Aggregate filtersAgg = filtersBucket.aggregations().get("filters_agg");
                            this.getDateCountDTOS(filtersAgg, field, topN, detailDTOS);
                        });
                    } else if (nestedList.contains(field)) {
                        filtersBuckets.forEach(filtersBucket -> {
                            this.getFiltersCountDTO(filtersBucket, condition, detailDTOS);
                            Aggregate filtersAgg = filtersBucket.aggregations().get("filters_agg");
                            this.getNestedCountDTOS(filtersAgg, field, detailDTOS);
                        });
                    } else if (childList.contains(field)) {
                        filtersBuckets.forEach(filtersBucket -> {
                            this.getFiltersCountDTO(filtersBucket, condition, detailDTOS);
                            Aggregate filtersAgg = filtersBucket.aggregations().get("filters_agg");
                            this.getChildCountDTOS(filtersAgg, field, detailDTOS);
                        });
                    } else {
                        filtersBuckets.forEach(filtersBucket -> {
                            this.getFiltersCountDTO(filtersBucket, condition, detailDTOS);
                            Aggregate filtersAgg = filtersBucket.aggregations().get("filters_agg");
                            this.getTermCountDTOS(filtersAgg, field, detailDTOS);
                        });
                    }
                } else {
                    this.getFilterCountDTO(agg, condition, detailDTOS);
                }
            } else {
                if (dateList.contains(field)) {
                    this.getDateCountDTOS(agg, field, topN, detailDTOS);
                } else if (nestedList.contains(field)) {
                    this.getNestedCountDTOS(agg, field, detailDTOS);
                } else if (childList.contains(field)) {
                    this.getChildCountDTOS(agg, field, detailDTOS);
                } else {
                    this.getTermCountDTOS(agg, field, detailDTOS);
                }
            }
        }
        esCountDTO.setDetailDTOS(detailDTOS);
        return esCountDTO;
    }

    /**
     * 专利的聚合分析
     *
     * @param countVOS
     * @return
     * @throws Exception
     */
    public EsCountDTO esAnalysisSearch(List<EsCountVO> countVOS) throws Exception {
        EsCountDTO esCountDTO = new EsCountDTO();
        List<EsCountDetailDTO> detailDTOS = new ArrayList<>();
        for (EsCountVO vo : countVOS) {
            String field = vo.getField();
            Integer topN = vo.getTopN();
            Boolean ifHaveChild = vo.getIfHaveChild();
            List<String> values = vo.getValues();
            //查询es返回数据
            SearchRequest.Builder builder = new SearchRequest.Builder();
            builder.index("patent");
            IEsAnalysisBuilder iEsAnalysisBuilder = null;
            String json = CommonService.readJsonFile("esAnalysis.json");
            List<EsConfigVO> esConfigVOS = JSON.parseArray(json, EsConfigVO.class);
            EsConfigVO esConfigVO = esConfigVOS.stream().filter(item -> item.getField().equals(field))
                    .findFirst().orElse(null);
            if (esConfigVO != null) {
                iEsAnalysisBuilder = esAnalysisBuilderFactory.getClass(esConfigVO.getEsClass());
                iEsAnalysisBuilder.setField(esConfigVO.getEsField());
                iEsAnalysisBuilder.setTopN(topN);
                iEsAnalysisBuilder.setIfHaveChild(ifHaveChild);
                for (String value : values) {
                    if (iEsAnalysisBuilder.getField().contains(".")) {
                        String path = iEsAnalysisBuilder.getField().substring(0, iEsAnalysisBuilder.getField().indexOf("."));
                        iEsAnalysisBuilder.setPath(path);
                    }
                    iEsAnalysisBuilder.setFieldValue(value);
                    Aggregation aggregation = iEsAnalysisBuilder.createAnalyseAgg();
                    builder.aggregations("Agg", aggregation);
                    SearchResponse<Patent> response = client.search(builder.build(), Patent.class);

                    Aggregate agg = response.aggregations().get("Agg");
                    if (dateList.contains(field)) {
                        this.getDateAnalysisDTOS(agg, field, value, topN, detailDTOS);
                    } else if (nestedList.contains(field)) {
                        this.getNestedCountDTOS(agg, field, detailDTOS);
                    } else if (childList.contains(field)) {
                        this.getChildAnalysisDTOS(agg, field, value, detailDTOS);
                    } else if (numberList.contains(field)) {
                        this.getNumberAnalysisDTOS(agg, field, value, detailDTOS);
                    } else {
                        this.getTermCountDTOS(agg, field, detailDTOS);
                    }
                }
            }
        }
        esCountDTO.setDetailDTOS(detailDTOS);
        return esCountDTO;
    }

    public EsCountDTO esAnalysis(EsCountVO vo) throws Exception {
        EsCountDTO esCountDTO = new EsCountDTO();
        List<EsCountDetailDTO> detailDTOS = new ArrayList<>();
        String field = vo.getField();
        Integer topN = vo.getTopN();
        Boolean ifHaveChild = vo.getIfHaveChild();
        List<String> values = vo.getValues();

        //查询es返回数据
        SearchRequest.Builder builder = new SearchRequest.Builder();
        builder.index("patent");
        IEsAnalysisBuilder iEsAnalysisBuilder = null;
        String json = CommonService.readJsonFile("esAnalysis.json");
        List<EsConfigVO> esConfigVOS = JSON.parseArray(json, EsConfigVO.class);
        EsConfigVO esConfigVO = esConfigVOS.stream().filter(item -> item.getField().equals(field))
                .findFirst().orElse(null);
        if (esConfigVO != null) {
            iEsAnalysisBuilder = esAnalysisBuilderFactory.getClass(esConfigVO.getEsClass());
            iEsAnalysisBuilder.setField(esConfigVO.getEsField());
            iEsAnalysisBuilder.setTopN(topN);
            iEsAnalysisBuilder.setIfHaveChild(ifHaveChild);
            for (String value : values) {
                if (iEsAnalysisBuilder.getField().contains(".")) {
                    String path = iEsAnalysisBuilder.getField().substring(0, iEsAnalysisBuilder.getField().indexOf("."));
                    iEsAnalysisBuilder.setPath(path);
                }
                iEsAnalysisBuilder.setFieldValue(value);
                Aggregation aggregation = iEsAnalysisBuilder.createAnalyseAgg();
                builder.aggregations("Agg", aggregation);
                SearchResponse<Patent> response = client.search(builder.build(), Patent.class);

                Aggregate agg = response.aggregations().get("Agg");
                if (dateList.contains(field)) {
                    this.getDateAnalysisDTOS(agg, field, value, topN, detailDTOS);
                } else if (nestedList.contains(field)) {
                    this.getNestedCountDTOS(agg, field, detailDTOS);
                } else if (childList.contains(field)) {
                    this.getChildAnalysisDTOS(agg, field, value, detailDTOS);
                } else if (numberList.contains(field)) {
                    this.getNumberAnalysisDTOS(agg, field, value, detailDTOS);
                } else {
                    this.getTermCountDTOS(agg, field, detailDTOS);
                }
            }
        }
        esCountDTO.setDetailDTOS(detailDTOS);
        return esCountDTO;
    }

    /**
     * 获取聚合后的aggregation
     *
     * @param builder
     * @param vo
     * @return
     * @throws Exception
     */
    public Aggregation selectAggregation(SearchRequest.Builder builder, EsCountVO vo) throws Exception {
        String valueOne = vo.getValueOne();
        String valueTwo = vo.getValueTwo();
        Integer topN = vo.getTopN();
        Boolean ifHaveChild = vo.getIfHaveChild();
        String field = vo.getField();
        Aggregation aggregation = null;

        builder.index("patent");
        IEsCountBuilder iEsCountBuilder = null;
        String json = CommonService.readJsonFile("esCount.json");
        List<EsConfigVO> esConfigVOS = JSON.parseArray(json, EsConfigVO.class);
        EsConfigVO esConfigVO = esConfigVOS.stream().filter(item -> item.getField().equals(field)).findFirst().orElse(null);
        if (esConfigVO != null) {
            iEsCountBuilder = esCountBuilderFactory.getClass(esConfigVO.getEsClass());
            iEsCountBuilder.setField(esConfigVO.getEsField());
            iEsCountBuilder.setValueOne(valueOne);
            iEsCountBuilder.setValueTwo(valueTwo);
            iEsCountBuilder.setTopN(topN);
            iEsCountBuilder.setIfHaveChild(ifHaveChild);
            if (iEsCountBuilder.getField().contains(".")) {
                String path = iEsCountBuilder.getField().substring(0, iEsCountBuilder.getField().indexOf("."));
                iEsCountBuilder.setPath(path);
            }
            aggregation = iEsCountBuilder.createAggregation();
        }
        return aggregation;
    }

    /**
     * 获取Filter聚合返回数据
     *
     * @param agg
     * @param condition
     * @return
     */
    public void getFilterCountDTO(Aggregate agg, String condition, List<EsCountDetailDTO> detailDTOS) {
        EsCountDetailDTO filterDTO = new EsCountDetailDTO();
        filterDTO.setField("condition");
        filterDTO.setName(condition);
        filterDTO.setNumber(agg.filter().docCount());
        if (filterDTO.getNumber() > 0) {
            detailDTOS.add(filterDTO);
        }
    }

    /**
     * 获取Filters聚合返回数据
     *
     * @param filtersBucket
     * @param condition
     * @return
     */
    public void getFiltersCountDTO(FiltersBucket filtersBucket, String condition, List<EsCountDetailDTO> detailDTOS) {
        EsCountDetailDTO filtersDTO = new EsCountDetailDTO();
        filtersDTO.setField("condition");
        filtersDTO.setName(condition);
        filtersDTO.setNumber(filtersBucket.docCount());
        if (filtersDTO.getNumber() > 0) {
            detailDTOS.add(filtersDTO);
        }
    }

    /**
     * 获取Terms聚合后数据
     *
     * @param agg
     * @param field
     * @param detailDTOS
     */
    public void getTermCountDTOS(Aggregate agg, String field, List<EsCountDetailDTO> detailDTOS) {
        List<StringTermsBucket> list = agg.sterms().buckets().array();
        list.forEach(bucket -> {
            EsCountDetailDTO dto = new EsCountDetailDTO();
            dto.setField(field);
            Aggregate aggregate = bucket.aggregations().get("filter_agg");
            dto.setName(bucket.key().stringValue());
            dto.setNumber(bucket.docCount());
            if (aggregate != null) {
                dto.setNumber(aggregate.filter().docCount());
            }
            if (dto.getNumber() > 0) {
                detailDTOS.add(dto);
            }
        });
    }

    /**
     * 获取children聚合后数据
     *
     * @param agg
     * @param field
     * @param detailDTOS
     */
    public void getChildCountDTOS(Aggregate agg, String field, List<EsCountDetailDTO> detailDTOS) {
        Aggregate childAgg = agg.children().aggregations().get("child_agg");
        List<StringTermsBucket> list = childAgg.sterms().buckets().array();
        list.forEach(bucket -> {
            EsCountDetailDTO dto = new EsCountDetailDTO();
            dto.setField(field);
            Aggregate aggregate = bucket.aggregations().get("filter_agg");
            List<StringTermsBucket> buckets = aggregate.sterms().buckets().array();
            for (StringTermsBucket termsBucket : buckets) {
                dto.setName(termsBucket.key().stringValue());
                dto.setNumber(termsBucket.docCount());
                if (dto.getNumber() > 0) {
                    detailDTOS.add(dto);
                }
            }
        });
    }

    /**
     * 获取children分析后数据
     *
     * @param agg
     * @param field
     * @param value
     * @param detailDTOS
     */
    public void getChildAnalysisDTOS(Aggregate agg, String field, String value, List<EsCountDetailDTO> detailDTOS) {
        Aggregate childAgg = agg.children().aggregations().get("child_agg");
        List<StringTermsBucket> list = childAgg.sterms().buckets().array();
        list.forEach(bucket -> {
            EsCountDetailDTO dto = new EsCountDetailDTO();
            dto.setField(field);
            Aggregate aggregate = bucket.aggregations().get("filter_agg");
            dto.setName(value);
            dto.setNumber(aggregate.filter().docCount());
            if (dto.getNumber() > 0) {
                detailDTOS.add(dto);
            }
        });
    }

    /**
     * 获取range分析后数据
     *
     * @param agg
     * @param field
     * @param value
     * @param detailDTOS
     */
    public void getNumberAnalysisDTOS(Aggregate agg, String field,String value, List<EsCountDetailDTO> detailDTOS) {
        List<RangeBucket> list = agg.range().buckets().array();
        list.forEach(bucket -> {
            EsCountDetailDTO dto = new EsCountDetailDTO();
            dto.setField(field);
            dto.setName(value);
            dto.setNumber(bucket.docCount());
            if (dto.getNumber() > 0) {
                detailDTOS.add(dto);
            }
        });
    }

    /**
     * 获取dateHistogram聚合后数据
     *
     * @param agg
     * @param field
     * @param detailDTOS
     */
    public void getDateCountDTOS(Aggregate agg, String field, Integer topN, List<EsCountDetailDTO> detailDTOS) {
        List<DateHistogramBucket> list = agg.dateHistogram().buckets().array();
        List<EsCountDetailDTO> esCountDetailDTOS = new ArrayList<>();
        list.forEach(bucket -> {
            EsCountDetailDTO dto = new EsCountDetailDTO();
            dto.setField(field);
            Aggregate aggregate = bucket.aggregations().get("filter_agg");
            dto.setName(bucket.keyAsString());
            dto.setNumber(bucket.docCount());
            if (aggregate != null) {
                dto.setNumber(aggregate.filter().docCount());
            }
            if (dto.getNumber() > 0) {
                esCountDetailDTOS.add(dto);
            }
        });
        if (!CollectionUtils.isEmpty(esCountDetailDTOS)) {
            List<EsCountDetailDTO> collect = esCountDetailDTOS.stream()
                    .sorted(Comparator.comparing(EsCountDetailDTO::getName).reversed()).limit(topN).collect(Collectors.toList());
            detailDTOS.addAll(collect);
        }
    }

    /**
     * 获取dateHistogram分析后数据
     *
     * @param agg
     * @param field
     * @param value
     * @param topN
     * @param detailDTOS
     */
    public void getDateAnalysisDTOS(Aggregate agg, String field, String value, Integer topN, List<EsCountDetailDTO> detailDTOS) {
        List<DateHistogramBucket> list = agg.dateHistogram().buckets().array();
        List<EsCountDetailDTO> esCountDetailDTOS = new ArrayList<>();
        if (value.contains("H")) {
            long sum = 0L;
            for (DateHistogramBucket bucket : list) {
                Aggregate aggregate = bucket.aggregations().get("filter_agg");
                long docCount = aggregate.filter().docCount();
                sum += docCount;
            }
            EsCountDetailDTO dto = new EsCountDetailDTO();
            dto.setField(field);
            dto.setName(value);
            dto.setNumber(sum);
            if (dto.getNumber() > 0) {
                esCountDetailDTOS.add(dto);
            }
        } else {
            list.forEach(bucket -> {
                EsCountDetailDTO dto = new EsCountDetailDTO();
                dto.setField(field);
                Aggregate aggregate = bucket.aggregations().get("filter_agg");
                dto.setName(value);
                dto.setNumber(bucket.docCount());
                if (aggregate != null) {
                    dto.setNumber(aggregate.filter().docCount());
                }
                if (dto.getNumber() > 0) {
                    esCountDetailDTOS.add(dto);
                }
            });
        }

        if (!CollectionUtils.isEmpty(esCountDetailDTOS)) {
            List<EsCountDetailDTO> collect = esCountDetailDTOS.stream()
                    .sorted(Comparator.comparing(EsCountDetailDTO::getName).reversed()).limit(topN).collect(Collectors.toList());
            detailDTOS.addAll(collect);
        }
    }

    /**
     * 获取nested聚合后数据
     *
     * @param agg
     * @param field
     * @param detailDTOS
     */
    public void getNestedCountDTOS(Aggregate agg, String field, List<EsCountDetailDTO> detailDTOS) {
        Aggregate termsAgg = agg.nested().aggregations().get("terms_agg");
        List<StringTermsBucket> list = termsAgg.sterms().buckets().array();
        list.forEach(bucket -> {
            EsCountDetailDTO dto = new EsCountDetailDTO();
            dto.setField(field);
            Aggregate aggregate = bucket.aggregations().get("filter_agg");
            dto.setName(bucket.key().stringValue());
            dto.setNumber(bucket.docCount());
            if (aggregate != null) {
                dto.setNumber(aggregate.filter().docCount());
            }
            if (dto.getNumber() > 0) {
                detailDTOS.add(dto);
            }
        });
    }
}