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.PatentColumnDTO;
import cn.cslg.pas.common.dto.PatentDTO;
import cn.cslg.pas.common.dto.business.EsCountDTO;
import cn.cslg.pas.common.dto.business.EsCountDetailDTO;
import cn.cslg.pas.common.model.request.QueryRequest;
import cn.cslg.pas.common.vo.EsConfigVO;
import cn.cslg.pas.common.vo.business.EsCountVO;
import cn.cslg.pas.common.vo.business.PatentNoVO;
import cn.cslg.pas.domain.es.Patent;
import cn.cslg.pas.factorys.EsBuilderFactory.IQueryBuilder;
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._types.query_dsl.HasChildQuery;
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 co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.elasticsearch.client.elc.ElasticsearchAggregations;
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;
import java.util.stream.Stream;

@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy})
public class EsCountService {
    private final List<String> termList = Arrays.asList();
    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_v1");
            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");
            List<EsCountDetailDTO> countDetailDTOS = new ArrayList<>();
            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 {
                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);
                    }
                });
            }
//            detailDTOS.addAll(countDetailDTOS);
        }
        esCountDTO.setDetailDTOS(detailDTOS);
        return esCountDTO;
    }

    /**
     * 根据专利号查询专利详情
     * @param vo
     * @return
     * @throws IOException
     */
    public PatentColumnDTO selectPatentDetail(PatentNoVO vo) throws IOException {
        PatentColumnDTO dto = new PatentColumnDTO();
        SearchRequest.Builder builder = new SearchRequest.Builder();
        //设置查询索引
        builder.index("patent");
        Query q = QueryBuilders.term(i -> i.field("patent_no.keyword").value(vo.getPatentNo()));
        Query query = QueryBuilders.bool(i -> i.must(q));
        builder.query(query);
        SearchResponse<Patent> response = client.search(builder.build(), Patent.class);
        List<PatentColumnDTO> list = new ArrayList<>();
        List<Hit<Patent>> hits = response.hits().hits();
        for (Hit<Patent> hit : hits) {
            Patent esMess = hit.source();
            BeanUtils.copyProperties(esMess, dto);
        }
        return dto;
    }
}