PersonService.java

package com.viontech.match.service;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.*;
import co.elastic.clients.elasticsearch._types.aggregations.*;
import co.elastic.clients.elasticsearch._types.query_dsl.*;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.bulk.IndexOperation;
import co.elastic.clients.elasticsearch.core.knn_search.KnnSearchQuery;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.HitsMetadata;
import co.elastic.clients.elasticsearch.core.search.SourceConfig;
import co.elastic.clients.elasticsearch.indices.DeleteIndexResponse;
import co.elastic.clients.elasticsearch.ingest.simulate.Document;
import co.elastic.clients.json.JsonData;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregation;
import co.elastic.clients.elasticsearch._types.aggregations.AggregationBuilders;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import com.alibaba.fastjson.JSON;
import com.viontech.keliu.model.BodyFeature;
import com.viontech.keliu.model.FaceFeature;
import com.viontech.keliu.model.Person;
import com.viontech.keliu.model.Pool;
import com.viontech.match.config.Constant;
import com.viontech.match.entity.PersonInfo;
import com.viontech.match.entity.SearchResultHit;
import com.viontech.match.entity.vo.RequestVo;
import com.viontech.match.entity.vo.ResponseVo;
import com.viontech.match.util.Utils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.client.RequestOptions;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.io.IOException;
import java.text.ParseException;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * .
 *
 * @author 谢明辉
 * @version 0.0.1
 */

@Service
@Slf4j
public class PersonService {
    private static final String[] FETCH_SOURCE = new String[]{"personId", "age", "gender", "fid", "counttime", "channelSerialNum", "body_type","unid"};
    @Resource
    private ElasticsearchClient client;
    @Resource
    private PoolService poolService;

    /**
     * 符合条件的数量
     * @param requestVo
     * @return
     * @throws Exception
     */
    public ResponseVo countMatchPerson(RequestVo requestVo) throws Exception {
        String rid = requestVo.getRid();
        String poolId = requestVo.getPersonPoolId();
        List<String> unionPersonPoolId = requestVo.getUnionPersonPoolId();
        Person person = requestVo.getPerson();
        List<String> poolIds = new ArrayList<>();

        if (unionPersonPoolId != null && unionPersonPoolId.size() > 0) {
            poolIds.addAll(unionPersonPoolId);
        }
        if (poolId != null) {
            poolIds.add(poolId);
        }
        log.info("人员匹配数量操作开始,PoolIds:[{}],rid:{}", poolIds, rid);
        long startTime = System.currentTimeMillis();
        try {
            String[] indices = poolIds.toArray(new String[poolIds.size()]);
//            SearchRequest request = new SearchRequest(indices);
//            IndicesOptions defaultIndicesOptions = request.indicesOptions();
//            EnumSet<IndicesOptions.Option> options = defaultIndicesOptions.getOptions();
//            options.add(IndicesOptions.Option.IGNORE_UNAVAILABLE);
//            EnumSet<IndicesOptions.WildcardStates> expandWildcards = defaultIndicesOptions.getExpandWildcards();
//            IndicesOptions newIndicesOptions = new IndicesOptions(options, expandWildcards);
//            request.indicesOptions(newIndicesOptions);

            long count = 0;
            List<BodyFeature> bodyFeatures = person.getBodyFeatures();
            Double[] feature = null;
            if (CollectionUtils.isEmpty(bodyFeatures) && null == person.getBodyMinScore()) {
                log.info("rid:{} CountMatchPerson人体特征为空并且最小分数为空，直接查数量,PoolIds:[{}],count:{}", rid, poolIds, count);
            } else if (CollectionUtils.isNotEmpty(bodyFeatures)){
                feature = bodyFeatures.get(0).getFeature();
                if (feature == null || feature.length < Constant.BODY_FEATURE_DIMS_2048) {
                    log.info("rid:{} 人体特征维数小于2048,跳过比对", rid);
                    ResponseVo error = ResponseVo.error(rid, "人体特征维数小于2048,跳过比对");
                    error.setCount(0);
                    return error;
                } else if (feature.length == Constant.BODY_FEATURE_DIMS_2110) {
                    feature = Utils.transferBodyFeature(feature);
                }

                if (feature.length > Constant.BODY_FEATURE_DIMS_2048) {
                    feature = Arrays.copyOfRange(feature, 3, Constant.BODY_FEATURE_DIMS_2048 + 3);
                }
            } else {
                log.info("rid:{} 参数错误", rid);
                ResponseVo error = ResponseVo.error(rid, "参数错误");
                error.setCount(0);
                return error;
            }
            Aggregation countAggregation = AggregationBuilders.valueCount(s -> s.field("unid"));


            Query query = null;
            if (feature == null) {
                BoolQuery boolQuery = getSearchSourceCountBuilder(feature, person, 1);
                query = boolQuery._toQuery();
            } else {
                Query scriptScoreQuery = getScriptScoreQuery(feature, person, 1);
                query = scriptScoreQuery;
            }

            SearchRequest request = new SearchRequest.Builder().index(Arrays.asList(indices)).ignoreUnavailable(true).aggregations("countId", countAggregation).size(0).query(query).build();
//            log.debug("rid:{} CountMatchPerson条件{}:{}", rid, indices, query.toString());

            SearchResponse<SearchResultHit> response = client.search(request, SearchResultHit.class);
            count = response.aggregations().size() == 0 ? 0 :(long) response.aggregations().get("countId").valueCount().value();

            ResponseVo result = ResponseVo.success(rid, "success");
            result.setCount(count);
            log.info("rid:{} CountMatchPerson完成,PoolIds:[{}],count:{}，耗时：{}", rid, poolIds, count, System.currentTimeMillis() - startTime);
            return result;
        } catch (ElasticsearchException e) {
            ResponseVo error = ResponseVo.error(rid, e.getMessage());
            error.setCount(0);
            return error;
        }
    }

    /**
     * 人员比对
     *
     * @param requestVo rid,personPoolId,unionPersonPoolId,person
     *
     * @return ResponseVo
     * @throws Exception --
     */
    public ResponseVo matchPerson(RequestVo requestVo) throws Exception {
        String rid = requestVo.getRid();
        String poolId = requestVo.getPersonPoolId();
        List<String> unionPersonPoolId = requestVo.getUnionPersonPoolId();
        Integer size = requestVo.getSize();
        Boolean agg = requestVo.getAgg();
        List<String> poolIds = new ArrayList<>();
        List<Person> matchFaces = new ArrayList<>();
        List<Person> matchBodies = new ArrayList<>();
        List<Pool> poolStatus = new ArrayList<>();

        if (unionPersonPoolId != null && unionPersonPoolId.size() > 0) {
            poolIds.addAll(unionPersonPoolId);
        }
        if (poolId != null) {
            poolIds.add(poolId);
        }
        log.info("人员匹配操作开始,PoolIds:[{}]，rid:{}", poolIds, rid);
        long startTime = System.currentTimeMillis();
        try {
            for (String id : poolIds) {
                Pool pool = new Pool();
                pool.setPersonPoolId(id);
                if (poolService.existPool(id)) {
                    //根据mallId判断特征池是否存在该mall的数据
                    boolean hasData = hasData(id, requestVo.getPerson().getMallId());
                    if (hasData) {
                        pool.setStatus(0);
                        List<Person> face = matchPerson(rid, id, requestVo.getPerson(), 0, size, agg);
                        matchFaces.addAll(face);
                        List<Person> body = matchPerson(rid, id, requestVo.getPerson(), 1, size, agg);
                        matchBodies.addAll(body);
                    } else {
                        pool.setStatus(2);
                    }

                } else {
                    pool.setStatus(1);
                }
                poolStatus.add(pool);
            }
            ResponseVo success = ResponseVo.success(rid, "success");
            if (matchFaces.size() > 0) {
                success.setMatchPersons(matchFaces);
            }
            if (matchBodies.size() > 0) {
                success.setMatchBodies(matchBodies);
            }
            success.setMatch(1);
            success.setPersonPoolStatus(poolStatus);
            log.info("人员匹配,Pool:{}, status:{}，rid:{},time：{}", poolIds, JSON.toJSONString(poolStatus), rid, System.currentTimeMillis() - startTime);
            return success;
        } catch (ElasticsearchException e) {
            if (e.status() == 400 && e.getMessage().contains(Constant.CLASS_CAST_EXCEPTION)) {
                for (String id : poolIds) {
                    RequestVo requestVo1 = new RequestVo();
                    requestVo1.setRid("1");
                    requestVo1.setPoolId(id);
                    requestVo1.setFlushPool(1);
                    poolService.deletePool(requestVo1);
                }
            }

            log.error("matchPerson", e);
            ResponseVo error = ResponseVo.error(rid, e.getMessage());
            error.setMatch(0);
            return error;
        }
    }

    /**
     * 修改人员特征(先删除再添加)
     *
     * @param requestVo rid,person, poolId,
     *
     * @return ResponseVo
     */
    public ResponseVo updatePerson(RequestVo requestVo) {
        String rid = requestVo.getRid();
        Person person = requestVo.getPerson();
        String personId = person.getPersonId();
        String poolId = requestVo.getPoolId();
        log.info("人员修改操作开始,poolId:[{}],personId:[{}]", poolId, personId);
        try {
            if (!poolService.existPool(poolId)) {
                poolService.createPool(requestVo, false);
            }
            DeleteByQueryResponse bulkByScrollResponse = deletePerson(poolId, personId);
            BulkResponse bulkItemResponses = addPerson(poolId, Collections.singletonList(person));
        } catch (Exception e) {
            log.error("人员修改操作异常", e);
            return ResponseVo.error(rid, "update failed");
        }
        log.info("人员修改操作完成,poolId:[{}],personId:[{}]", poolId, personId);
        return ResponseVo.success(rid);
    }

    public ResponseVo updateStaff(RequestVo requestVo) {
        String rid = requestVo.getRid();
        Person person = requestVo.getPerson();
        String personId = person.getPersonId();
        String poolId = requestVo.getPoolId();
//        log.info("店员修改操作开始,poolId:[{}],personId:[{}]", poolId, personId);
        try {
            if (!poolService.existPool(poolId)) {
                poolService.createPool(requestVo, false);
            }
            BulkResponse bulkItemResponses = addPerson(poolId, Collections.singletonList(person));
        } catch (Exception e) {
            log.error("店员修改操作异常", e);
            return ResponseVo.error(rid, "update failed");
        }
        log.info("店员修改操作完成,poolId:[{}],personId:[{}]", poolId, personId);
        return ResponseVo.success(rid);
    }

    /**
     * 添加人员
     *
     * @param poolId     特征池Id
     * @param personPool 要添加的人员列表
     *
     * @return 批量添加的结果
     * @throws IOException elasticsearch 所产生的异常
     */
    public BulkResponse addPerson(String poolId, List<Person> personPool) throws IOException {
//        BulkRequest bulkRequest = new BulkRequest.Builder().index(poolId).build();
        List<BulkOperation> operationList = new ArrayList<>();
        for (Person person : personPool) {
            if (person == null) {
                continue;
            }
            Integer age = person.getAge();
            String gender = person.getGender();
            String personId = person.getPersonId();
            Date personCountTime = person.getCounttime();
            String direction = person.getDirection();
            Long mallId = person.getMallId();
            //Long gateId = person.getGateId();
            Integer bodyType = person.getBodyType();
            String personChannelSerialNum = person.getChannelSerialNum();
            List<FaceFeature> faceFeatures = person.getFaceFeatures();
            if (CollectionUtils.isNotEmpty(faceFeatures)) {
                for (FaceFeature faceFeature : faceFeatures) {
                    Double[] feature = faceFeature.getFeature();
                    if (feature != null && feature.length == Constant.FACE_FEATURE_DIMS) {
                        String fid = faceFeature.getFid();
                        String unid = faceFeature.getUnid();
                        Long gateId = faceFeature.getGateId();
                        operationList.add(new BulkOperation.Builder().index(new IndexOperation.Builder<PersonInfo>().index(poolId).document(new PersonInfo(unid, personId, feature, null, age, gender, bodyType, personCountTime, fid, personChannelSerialNum, gateId, direction, mallId)).build()).build());
                    }
                }
            }
            List<BodyFeature> bodyFeatures = person.getBodyFeatures();
            if (CollectionUtils.isNotEmpty(bodyFeatures)) {
                for (BodyFeature bodyFeature : bodyFeatures) {
                    Double[] feature = bodyFeature.getFeature();
                    if (feature == null || feature.length < Constant.BODY_FEATURE_DIMS_2048) {
                        continue;
                    } else if (feature.length == Constant.BODY_FEATURE_DIMS_2110) {
                        feature = Utils.transferBodyFeature(feature);
                    }

                    if (feature.length > Constant.BODY_FEATURE_DIMS_2048) {
                        feature = Arrays.copyOfRange(feature, 3, Constant.BODY_FEATURE_DIMS_2048 + 3);
                    }

                    String fid = bodyFeature.getBid();
                    String unid = bodyFeature.getUnid();
                    Long gateId = bodyFeature.getGateId();
                    Date counttime = bodyFeature.getCounttime() == null ? personCountTime : bodyFeature.getCounttime();
                    String channelSerialNum = bodyFeature.getChannelSerialNum() == null ? personChannelSerialNum : person.getChannelSerialNum();
                    operationList.add(new BulkOperation.Builder().index(new IndexOperation.Builder<PersonInfo>().index(poolId).document(new PersonInfo(unid, personId, feature, null, age, gender, bodyType, counttime, fid, personChannelSerialNum, gateId, direction, mallId)).build()).build());
                }
            }
        }

        if (operationList.size() == 0) {
            return null;
        }
        BulkResponse bulk = client.bulk(new BulkRequest.Builder().index(poolId).operations(operationList).build());
//        poolService.refreshPool(poolId);
        return bulk;
    }

    /**
     * 删除人员
     *
     * @param poolId   特征池Id
     * @param personId 人员Id,对应数据库personUnid
     *
     * @return 删除结果
     * @throws IOException elasticsearch 所产生的异常
     */
    public DeleteByQueryResponse deletePerson(String poolId, String personId) throws IOException {
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest.Builder().index(poolId)
                .query(q -> q.match(m -> m.field("personId").query(personId)))
                .refresh(true).build();
        return client.deleteByQuery(deleteByQueryRequest);
    }

    public DeleteByQueryResponse deletePersonByMallId(String poolId, Long mallId) throws IOException {
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest.Builder().index(poolId)
                .query(q -> q.match(m -> m.field("mallId").query(mallId)))
                .refresh(true)
                .build();
        return client.deleteByQuery(deleteByQueryRequest);
    }

    public DeleteByQueryResponse deletePersonByFid(String poolId, List<String> fidList) throws IOException {
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest.Builder().index(poolId)
                .query(QueryBuilders.terms().field("fid").terms(new TermsQueryField.Builder().value(fidList.stream().map(FieldValue::of).collect(Collectors.toList())).build()).build()._toQuery())
                .refresh(true)
                .build();
        return client.deleteByQuery(deleteByQueryRequest);
    }

    public BulkByScrollResponse deletePersonByMallId(String poolId, Long mallId) throws IOException {
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(poolId)
                .setQuery(new TermQueryBuilder("mallId", mallId))
                .setRefresh(true);
        return client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
    }

    public BulkByScrollResponse deletePersonByFid(String poolId, List<String> fidList) throws IOException {
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(poolId)
                .setQuery(new TermsQueryBuilder("fid", fidList))
                .setRefresh(true);
        return client.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
    }


    /**
     * 人员匹配入口
     *
     * @param rid 请求唯一id
     * @param poolId 在哪个特征池中匹配
     * @param person 用来匹配的人员信息
     * @param type   匹配类型 0人脸 1人体
     *
     * @return 匹配结果
     * @throws Exception --
     */
    public List<Person> matchPerson(String rid, String poolId, Person person, int type, Integer size, Boolean agg) throws Exception {
        List<Person> matchResult = new ArrayList<>();
        int matchResultSize;
        if (type == 0) {
            matchFace(rid, poolId, matchResult, person, size, agg);
            matchResultSize = Constant.FACE_MATCH_RESULT_SIZE;
        } else {
            matchBody(rid, poolId, matchResult, person, size, agg);
            matchResultSize = Constant.BODY_MATCH_RESULT_SIZE;
        }
        if (size != null) {
            matchResultSize = size;
        }

        Map<String, Person> collect = matchResult.stream().collect(Collectors.toMap(Person::getPersonId, x -> x, (a, b) -> a.getScore() >= b.getScore() ? a : b));

        Stream<Person> stream = collect.values().stream().sorted(Comparator.comparingDouble(Person::getScore).reversed());
        if (matchResult.size() > matchResultSize) {
            stream = stream.limit(matchResultSize);
        }
        matchResult = stream.collect(Collectors.toList());
        return matchResult;
    }

    private void matchFace(String rid, String poolId, List<Person> matchResult, Person person, Integer size, Boolean agg) throws Exception {
        List<FaceFeature> faceFeatures = person.getFaceFeatures();
        if (CollectionUtils.isNotEmpty(faceFeatures)) {
            for (FaceFeature faceFeature : faceFeatures) {
                Double[] feature = faceFeature.getFeature();
                if (feature == null || feature.length != Constant.FACE_FEATURE_DIMS) {
                    log.info("人脸特征维数小于512,跳过比对");
                    continue;
                }

                SearchRequest searchRequest = getSearchRequest(rid, poolId, size == null ? Constant.FACE_MATCH_RESULT_SIZE : size, feature, person, 0, agg);
                matchResult.addAll(match0(searchRequest, agg));
            }
        } else {
            log.info("no face feature");
        }
    }

    private void matchBody(String rid, String poolId, List<Person> matchResult, Person person, Integer size, Boolean agg) throws Exception {
        List<BodyFeature> bodyFeatures = person.getBodyFeatures();
        if (CollectionUtils.isNotEmpty(bodyFeatures)) {
            for (BodyFeature faceFeature : bodyFeatures) {
                Double[] feature = faceFeature.getFeature();
                if (feature == null || feature.length < Constant.BODY_FEATURE_DIMS_2048) {
                    log.info("人体特征维数小于2048,跳过比对");
                    continue;
                } else if (feature.length == Constant.BODY_FEATURE_DIMS_2110) {
                    feature = Utils.transferBodyFeature(feature);
                }

                if (feature.length > Constant.BODY_FEATURE_DIMS_2048) {
                    feature = Arrays.copyOfRange(feature, 3, Constant.BODY_FEATURE_DIMS_2048 + 3);
                }

                SearchRequest searchRequest = getSearchRequest(rid, poolId, size == null ? Constant.BODY_MATCH_RESULT_SIZE : size, feature, person, 1, agg);

                matchResult.addAll(match0(searchRequest, agg));
            }
        }
    }

    private List<Float> getFeature(Double[] feature) {
        List<Double> list = Arrays.asList(feature);
        return list.stream().map(x -> x.floatValue()).collect(Collectors.toList());
    }

    private Query getScriptScoreQuery(Double[] feature, Person person, int type) {
        return new Query.Builder().knn(getSearchSourceBuilder(feature, person, type)).build();
    }

    private BoolQuery getSearchSourceCountBuilder(Double[] feature, Person person, int type) {
        List<Query> queries = new ArrayList<>();
        BoolQuery.Builder boolQuery = new BoolQuery.Builder();

        // 根据通道号过滤
        List<String> channelSerialNums = person.getChannelSerialNums();
        if (CollectionUtils.isNotEmpty(channelSerialNums)) {
            queries.add(QueryBuilders.terms().field("channelSerialNum").terms(new TermsQueryField.Builder().value(channelSerialNums.stream().map(FieldValue::of).collect(Collectors.toList())).build()).build()._toQuery());
        }

        // 根据监控点id进行过滤
        List<Long> gateIdIn = person.getGateIdIn();
        if (CollectionUtils.isNotEmpty(gateIdIn)) {
            queries.add(QueryBuilders.terms().field("gateId").terms(new TermsQueryField.Builder().value(gateIdIn.stream().map(FieldValue::of).collect(Collectors.toList())).build()).build()._toQuery());
        }

        // 根据人的ID进行过滤
        if (StringUtils.isNotEmpty(person.getPersonUnid())) {
            queries.add(QueryBuilders.term().field("personId").value(FieldValue.of(person.getPersonUnid())).build()._toQuery());
        }

        if (StringUtils.isNotBlank(person.getCompareDirection())) {
            queries.add(QueryBuilders.term().field("direction").value(FieldValue.of(person.getCompareDirection())).build()._toQuery());
        }
        //根据mallId过滤
        if (person.getMallId() != null) {
            queries.add(QueryBuilders.term().field("mallId").value(FieldValue.of(person.getMallId())).build()._toQuery());
        }

        Date counttimeGTE = person.getCounttimeGTE();
        Date counttimeLTE = person.getCounttimeLTE();
        if (counttimeGTE != null || counttimeLTE != null) {
            queries.add(QueryBuilders.range().term(s -> s.field("counttime").gte(Constant.DATE_FORMAT.get().format(counttimeGTE)).lte(Constant.DATE_FORMAT.get().format(counttimeLTE))).build()._toQuery());
        }

        return boolQuery.filter(queries).build();
    }

    /**
     *
     * @param feature
     * @param person
     * @param type 0,人脸，1人体
     * @return
     */
    private KnnQuery getSearchSourceBuilder(Double[] feature, Person person, int type) {
        List<Query> queries = new ArrayList<>();
        KnnQuery.Builder knnQuery = new KnnQuery.Builder();
        if (null != feature && type == 0) {
            knnQuery.field("data").queryVector(getFeature(feature)).numCandidates(Constant.FACE_MATCH_RESULT_SIZE);
        } else if (null != feature && type == 1){
            knnQuery.field("body").queryVector(getFeature(feature)).numCandidates(Constant.BODY_MATCH_RESULT_SIZE);
            queries.add(new Query.Builder().exists(s -> s.field("body").boost(1.0f)).build());
        }
        //根据mallId过滤
        if (person.getMallId() != null) {
            boolQuery.filter().add(QueryBuilders.termQuery("mallId", person.getMallId()));
        }
        // 根据通道号过滤
        List<String> channelSerialNums = person.getChannelSerialNums();
        if (CollectionUtils.isNotEmpty(channelSerialNums)) {
            queries.add(QueryBuilders.terms().field("channelSerialNum").terms(new TermsQueryField.Builder().value(channelSerialNums.stream().map(FieldValue::of).collect(Collectors.toList())).build()).build()._toQuery());
        }

        // 根据监控点id进行过滤
        List<Long> gateIdIn = person.getGateIdIn();
        if (CollectionUtils.isNotEmpty(gateIdIn)) {
            queries.add(QueryBuilders.terms().field("gateId").terms(new TermsQueryField.Builder().value(gateIdIn.stream().map(FieldValue::of).collect(Collectors.toList())).build()).build()._toQuery());
        }

        // 根据人的ID进行过滤
        if (StringUtils.isNotEmpty(person.getPersonUnid())) {
//            boolQuery.filter(QueryBuilders.termQuery("personId", person.getPersonUnid()));
            queries.add(QueryBuilders.term().field("personId").value(FieldValue.of(person.getPersonUnid())).build()._toQuery());
        }

        if (StringUtils.isNotBlank(person.getCompareDirection())) {
            queries.add(QueryBuilders.term().field("direction").value(FieldValue.of(person.getCompareDirection())).build()._toQuery());
        }

        if (person.getMallId() != null) {
            queries.add(QueryBuilders.term().field("mallId").value(FieldValue.of(person.getMallId())).build()._toQuery());
        }

        Date counttimeGTE = person.getCounttimeGTE();
        Date counttimeLTE = person.getCounttimeLTE();
        if (counttimeGTE != null || counttimeLTE != null) {
//            boolQuery.filter(rangeQueryBuilder);
            queries.add(QueryBuilders.range().term(s -> s.field("counttime").gte(Constant.DATE_FORMAT.get().format(counttimeGTE)).lte(Constant.DATE_FORMAT.get().format(counttimeLTE))).build()._toQuery());
        }

        return knnQuery.filter(queries).build();
    }

    private SearchRequest getSearchRequest(String rid, String poolId, Integer matchResultSize, Double[] feature, Person person, int type, Boolean agg) {
//        BoolQuery.Builder builder = getSearchSourceBuilder(feature, person, type);
        Query scriptScoreQuery = getScriptScoreQuery(feature, person, 1);
        if (agg) {
//            AggregationBuilders.max("max_score", s -> s.field("_score").script(new Script.Builder().source("_score").build()));
//            Aggregation aggregation = AggregationBuilders.max(s -> s.field("max_score").script(new Script.Builder().source("_score").build()));
//            Aggregation termsAgg = AggregationBuilders.terms("by_personId", TermsAggregationDefinition.builder("by_personId").field("personId").order("max_score", false));
//            Aggregation termsAgg2 = AggregationBuilders.terms(s -> s.field("personId").size(matchResultSize).size(matchResultSize));
//            MaxAggregation.Builder maxBucketAggregation = new MaxAggregation.Builder().field("max_score").script(new Script.Builder().source("_score").build());
//            TermsAggregation termsAgg = new TermsAggregation.Builder().field("personId").format("by_personId").size(matchResultSize).child(maxBucketAggregation);
//            MetricAg
//            termsAgg2.subAggregation(aggregation);
//            termsAgg2.order(BucketOrder.aggregation("max_score", false));
//            termsAgg.size(matchResultSize);

//            MaxAggregationBuilder maxScoreAgg = AggregationBuilders.max("max_score").script(new Script("_score"));
//            TermsAggregationBuilder personIdAgg = AggregationBuilders.terms("by_personId").field("personId");
//            personIdAgg.subAggregation(maxScoreAgg);
//            personIdAgg.order(BucketOrder.aggregation("max_score", false));
//            personIdAgg.size(matchResultSize);
//            builder.aggregation(personIdAgg);
        } else {
//            builder.size(matchResultSize);
        }
        SourceConfig sourceConfig = new SourceConfig.Builder().filter(s -> s.includes(Arrays.asList(FETCH_SOURCE))).build();
//        log.debug("rid:{} poolId:{} 匹配时参数:{}", rid, poolId, scriptScoreQuery.toString());
//        KnnSearch knnQuery = getKnnSearch(feature, person, type);
//        ScriptScoreQuery scriptScoreQuery = getScriptScoreQuery(feature, person, type);
        return new SearchRequest.Builder().index(poolId).query(scriptScoreQuery).source(sourceConfig).size(matchResultSize).minScore(person.getBodyMinScore().doubleValue()/100).build();
    }

    private List<Person> match0(SearchRequest searchRequest, Boolean agg) throws Exception {
        String poolId = searchRequest.index().get(0);
        List<Person> persons = new ArrayList<>();
//        log.info("request:{}", JSON.tojsonS);
        SearchResponse<SearchResultHit> search = client.search(searchRequest, SearchResultHit.class);
        if (agg) {
            Map aggregations = search.aggregations();
            TermsAggregation byPersonId = (TermsAggregation) aggregations.get("by_personId");
            log.info("aggregation:{}", byPersonId.toString());
//            List<Bucket> buckets = byPersonId.;
//            for (Terms.Bucket bucket : buckets) {
//                String personId = (String) bucket.getKey();
//                Max maxScore = bucket.getAggregations().get("max_score");
//                double value = maxScore.getValue();
//                Person person = new Person().setPersonId(personId).setScore((float) value).setPersonPoolId(poolId);
//                persons.add(person);
//            }
        } else {
            HitsMetadata hits = search.hits();
//            log.info("match0:{}", hits.toString());
            List<Hit<SearchResultHit>> hits1 = hits.hits();
            for (Hit item : hits1) {
                SearchResultHit hit = (SearchResultHit) item.source();
                Person p = new Person();
                p.setPersonId((String) hit.getPersonId());
                p.setAge(hit.getAge());
                p.setGender((String) hit.getGender());
                p.setChannelSerialNum(hit.getChannelSerialNum());
                p.setBodyType(hit.getBody_type());
                p.setCaptureUnid(hit.getUnid());

                p.setCounttime(Optional.ofNullable((String)hit.getCounttime())
                        .map(x -> {
                            try {
                            return Constant.DATE_FORMAT.get().parse(x);
                            } catch (ParseException e) {
                                return null;
                            }
                        })
                        .orElse(null));

                p.setScore(item.score().floatValue() * 100);
                p.setPersonPoolId(item.index());
                persons.add(p);
            }
        }
        return persons;
    }


    public boolean hasData(String poolId, Long mallId) throws IOException {
        log.info("判断特征池:{}是否存在mallId:{}的数据", poolId, mallId);
        if (mallId == null || mallId == 0L) {
            return true;
        }
        //店员库才需要判断是否有数据，并且重建库
        if (!poolId.contains("staff")) {
            return true;
        }
        BoolQueryBuilder builder = new BoolQueryBuilder();
        builder.filter(QueryBuilders.termQuery("mallId", mallId));
        CountRequest countRequest = new CountRequest(poolId);
        countRequest.query(builder);
        CountResponse response = client.count(countRequest, RequestOptions.DEFAULT);
        log.info("特征池:{}中mallId:{}的数据量为:{}", poolId, mallId, response.getCount());
        return response.getCount() > 0;
    }

}