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.vo.EsConfigVO;
import cn.cslg.pas.common.vo.business.EsCountVO;
import cn.cslg.pas.domain.es.Patent;
import cn.cslg.pas.factorys.EsCountBuilderFactory.EsCountBuilderFactory;
import cn.cslg.pas.factorys.EsCountBuilderFactory.IEsCountBuilder;
import cn.cslg.pas.service.business.CommonService;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import com.alibaba.fastjson.JSON;
import lombok.RequiredArgsConstructor;
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.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy})
public class EsCountService {
    private final List<String> childList = Arrays.asList("childRaw");
    private final List<String> nestedList = Arrays.asList("PA", "IN", "PE");
    private final List<String> dateList = Arrays.asList("PD", "AD");

    private final ElasticsearchClient client;

    @Autowired
    private EsCountBuilderFactory esCountBuilderFactory;

    /**
     * 查询专利库中的专利分组聚合统计
     *
     * @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();
            String valueOne = vo.getValueOne();
            String valueTwo = vo.getValueTwo();
            Integer topN = vo.getTopN();

            SearchRequest.Builder builder = new SearchRequest.Builder();
            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);
                if (iEsCountBuilder.getField().contains(".")) {
                    String path = iEsCountBuilder.getField().substring(0, iEsCountBuilder.getField().indexOf("."));
                    iEsCountBuilder.setPath(path);
                }

            }
            Aggregation aggregation = iEsCountBuilder.createAggregation();
            builder.aggregations("Agg", aggregation);
            SearchResponse<Patent> response = client.search(builder.build(), Patent.class);
            Aggregate agg = response.aggregations().get("Agg");
            if (dateList.contains(field)) {
                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);
                }
            } else if (nestedList.contains(field)) {
                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);
                    }
                });
            } else if (childList.contains(field)) {
                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(bucket.key().stringValue());
                    dto.setNumber(bucket.docCount());
                    if (aggregate != null) {
                        dto.setNumber(aggregate.filter().docCount());
                    }
                    if (dto.getNumber() > 0) {
                        detailDTOS.add(dto);
                    }
                });
            } else {
                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);
                    }
                });
            }
        }
        esCountDTO.setDetailDTOS(detailDTOS);
        return esCountDTO;
    }
}