package com.szwl.service.es; import cn.com.crbank.ommo.esclient.EsBaseService; import cn.com.crbank.ommo.exception.MyException; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.metadata.IPage; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.szwl.model.bean.ChartBean; import com.szwl.model.bean.ChartColumn; import com.szwl.model.bo.ChartType; import com.szwl.model.entity.TEquipment; import com.szwl.model.entity.TOrder; import com.szwl.model.query.StatisticsParam; import com.szwl.model.query.TCoinOrderParam; import com.szwl.model.query.TEquipmentParam; import com.szwl.model.query.TOrderParam; import com.szwl.model.utils.DateUtils; import com.szwl.service.TOrderService; import lombok.extern.slf4j.Slf4j; import org.apache.commons.collections4.CollectionUtils; import org.apache.commons.lang3.StringUtils; import org.elasticsearch.action.search.SearchRequest; import org.elasticsearch.action.search.SearchResponse; import org.elasticsearch.client.RequestOptions; import org.elasticsearch.client.indices.GetIndexRequest; import org.elasticsearch.index.query.BoolQueryBuilder; import org.elasticsearch.search.aggregations.Aggregation; import org.elasticsearch.search.aggregations.AggregationBuilder; import org.elasticsearch.search.aggregations.AggregationBuilders; import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval; import org.elasticsearch.search.aggregations.bucket.histogram.Histogram; import org.elasticsearch.search.aggregations.metrics.ParsedSum; import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder; import org.elasticsearch.search.builder.SearchSourceBuilder; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import java.io.IOException; import java.lang.reflect.InvocationTargetException; import java.math.BigDecimal; import java.math.RoundingMode; import java.text.ParseException; import java.text.SimpleDateFormat; import java.util.ArrayList; import java.util.Calendar; import java.util.Date; import java.util.List; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; @Slf4j @Service public class EsTOrderService extends EsBaseService { private static boolean InitEsTableStatus = false; // 是否正常完成 es 初始化 public static final int MAX_ROW = 1000; @Autowired TOrderService tOrderService; @Autowired EsTEquipmentService esTEquipmentService; @Autowired EsTCoinOrderService esTCoinOrderService; @Override public String getTableName() { return "es_t_order"; } // public String getTableName() { // return "es_t_order"; // } @Override public TOrder getInstanceOfEntity() { return new TOrder(); } @Override public void setInitTableStatus(boolean flag) { InitEsTableStatus = flag; } @Override public boolean getInitTableStatus() { return InitEsTableStatus; } @Override public void initTableFun() { try{ String tableName = getTableName(); GetIndexRequest request = new GetIndexRequest(tableName); // DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(this.getTableName()); // restHighLevelClient.indices().delete(deleteIndexRequest,RequestOptions.DEFAULT); boolean isExists = restHighLevelClient.indices().exists(request, RequestOptions.DEFAULT); if (!isExists) { log.info("es 索引 开始创建"+tableName); createTable(); // 初始化旧流水 int num = 0; while (true) { int limit = MAX_ROW; // int offset= num * MAX_ROW; int offset= num; // TOrderExample example = new TOrderExample(); // example.setLimit(limit); // example.setOffset(offset); // List list = tOrderService.selectByOption(example); LambdaQueryWrapper query = Wrappers.lambdaQuery(); SimpleDateFormat ft = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); String time = "2022-07-01 00:00:00"; Date date = ft.parse(time); query.gt(TOrder::getCreateDate,date); Page page = new Page<>(offset, limit, true); IPage iPage = tOrderService.page(page, query); List list = iPage.getRecords(); insertBatch(list); num++; if(list.size()< MAX_ROW){ // 数据小于 最大值 ,证明后面已无数据,则跳出 break; } } } else { log.info("es 索引 "+tableName+" 已存在不再创建"); } InitEsTableStatus = true; }catch (Exception e){ log.error("ElasticsearchRunner InitEsTOrderThread 发生错误:{}" , e); throw new MyException("ElasticsearchRunner InitEsTOrderThread 发生错误:" + e.getMessage()); } } @Override public String getEntityPrimaryKey(TOrder tOrder) { return String.valueOf(tOrder.getId()); } @Override public TOrder setEntityPrimaryKey(TOrder tOrder, String value) { return null; } // @Override // public TOrder setEntityPrimaryKey(TOrder tOrder, String value) { // return tOrder.setId(Long.parseLong(value)); // } /** * 根据时间 重新同步es * @param * @return */ public void updateEsByDate(String startTime, String endTime) { try { startTime = startTime.replace("/", "-"); endTime = endTime.replace("/", "-"); Date start = DateUtils.parseDate(startTime, DateUtils.PATTERN_yyyy_MM_dd_HH_mm_ss, new Date()); Date end = DateUtils.parseDate(endTime, DateUtils.PATTERN_yyyy_MM_dd_HH_mm_ss, new Date()); // 查询es 现有的数据,删除 TOrderParam param = new TOrderParam(); param.setCreateDate_start(start); param.setCreateDate_end(end); List list_es = this.selectEntityByEqualToOption(param); for (TOrder entity : list_es) { this.deleteTableById(String.valueOf(entity.getId())); } // 插入 新的 // List list = tOrderService.selectForEs(param); LambdaQueryWrapper query = Wrappers.lambdaQuery(); query.gt(TOrder::getCreateDate,start); query.lt(TOrder::getCreateDate,end); List list = tOrderService.list(query); if (CollectionUtils.isNotEmpty(list)) { insertBatch(list); } } catch (Exception e) { log.error(this.getTableName()+" updateEsByDate 发生错误:{}", e); throw new MyException(this.getTableName()+" updateEsByDate 发生错误:" + e.getMessage()); } } /** * es 查询条件 * @param param * @return */ public BoolQueryBuilder getStatisticsParam2QueryBuilder(StatisticsParam param){ BoolQueryBuilder boolQueryBuilder; String startDate = param.getStartDate().replace("/","-"); String endDate = param.getEndDate().replace("/","-"); Date start = DateUtils.parseDate(startDate+" 00:00:00", DateUtils.PATTERN_yyyy_MM_dd_HH_mm_ss,new Date()); Date end = DateUtils.parseDate(endDate+" 23:59:59", DateUtils.PATTERN_yyyy_MM_dd_HH_mm_ss,new Date()); if(StringUtils.equals("0",param.getIfForeign())){ // 国内用户 TOrderParam tOrderParam = new TOrderParam(); tOrderParam.setStatus(1); tOrderParam.setCreateDate_start(start); tOrderParam.setCreateDate_end(end); if(StringUtils.isNotEmpty(param.getAdminId())){ // 所属商家id tOrderParam.setAdminId(Long.parseLong(param.getAdminId())); } if(StringUtils.isNotEmpty(param.getEquipmentId())){ // 设备id tOrderParam.setEquipmentId(Long.parseLong(param.getEquipmentId())); } if(param.getEquipmentIds()!=null&¶m.getEquipmentIds().size()>0){ // 设备id tOrderParam.setEquipmentId_inList(param.getEquipmentIds()); } if(StringUtils.isNotEmpty(param.getPayType())){ // 支付方式 tOrderParam.setFrpCode(param.getPayType()); } if (null != param.getAgencyId()){ tOrderParam.setAgencyId(param.getAgencyId()); tOrderParam.setType_gt(1); } if(null != param.getMerchantId()){ tOrderParam.setMerchantId(param.getMerchantId()); tOrderParam.setType_gt(2); } if(param.getAdminIds()!=null&¶m.getAdminIds().size()>0){ tOrderParam.setAdminId_inList(param.getAdminIds()); } boolQueryBuilder = this.getParam2QueryBuilder(tOrderParam); }else{ // 国外用户 TEquipmentParam tEquipmentParam = new TEquipmentParam(); if(StringUtils.isNotEmpty(param.getEquipmentId())){ // 设备id tEquipmentParam.setId(Long.parseLong(param.getEquipmentId())); } List equipmentList = esTEquipmentService.selectEntityByEqualToOption(tEquipmentParam); List clientIdList = equipmentList.stream().map(TEquipment::getClientId).collect(Collectors.toList()); TCoinOrderParam tCoinOrderParam = new TCoinOrderParam(); tCoinOrderParam.setCreateDate_start(start); tCoinOrderParam.setCreateDate_end(end); tCoinOrderParam.setClientId_inList(clientIdList); if(param.getAdminIds().size()>0){ tCoinOrderParam.setAdminId_inList(param.getAdminIds()); } if(StringUtils.isNotEmpty(param.getAdminId())){ // 所属商家id tCoinOrderParam.setAdminId(Long.parseLong(param.getAdminId())); } if(param.getClientIds().size()>0){ // 设备 tCoinOrderParam.setClientId_inList(param.getClientIds()); } boolQueryBuilder = esTCoinOrderService.getParam2QueryBuilder(tCoinOrderParam); } return boolQueryBuilder; } public ChartColumn getStatistics(StatisticsParam param) { String msg = ""; String format = "yyyy-MM"; DateHistogramInterval interval = DateHistogramInterval.DAY; if (ChartType.day.toString().equals(param.getChartType())) { interval = DateHistogramInterval.HOUR; format = "HH点"; msg = "日统计"; } if (ChartType.week.toString().equals(param.getChartType())) { interval = DateHistogramInterval.DAY; format = "yyyy-MM-dd"; msg = "周统计"; } if (ChartType.month.toString().equals(param.getChartType())) { interval = DateHistogramInterval.DAY; format = "MM月dd"; msg = "月统计"; } if (ChartType.year.toString().equals(param.getChartType())) { interval = DateHistogramInterval.MONTH; format = "MM月"; msg = "年统计"; } ChartColumn chartColumn = getStatistics(param,format,interval); if (ChartType.week.toString().equals(param.getChartType())) { ArrayList categories = new ArrayList<>(); for (String day :chartColumn.getCategories()) { int week = DateUtils.getWeek(DateUtils.parseDate(day, DateUtils.PATTERN_yyyy_MM_dd,new Date())); String finalCategories = "周" + (week==0?"日":week); categories.add(finalCategories); } chartColumn.setCategories(categories); } return chartColumn; } public ChartColumn getStatistics(StatisticsParam param, String format, DateHistogramInterval interval) { try { String aggregationResultName = "aggregationResult";// 仅为名称,用以获取返回结果 String aggSumName = "sales"; // 仅为名称,用以获取返回结果 String aggName_productNumber = "productNumber"; // 仅为名称,用以获取返回结果 //Bool查询 BoolQueryBuilder queryBuilder = getStatisticsParam2QueryBuilder(param); // 时间聚合 AggregationBuilder dateHistogram = AggregationBuilders.dateHistogram(aggregationResultName) .field("createDate") .calendarInterval(interval) .format(format); // 根据字段price 求和 sum, sales 仅为名称,用以获取返回结果 SumAggregationBuilder sumAggregationBuilder = AggregationBuilders.sum(aggSumName).field("price"); dateHistogram.subAggregation(sumAggregationBuilder); SumAggregationBuilder sumAggregationBuilder2 = AggregationBuilders.sum(aggName_productNumber).field("productNumber"); dateHistogram.subAggregation(sumAggregationBuilder2); SearchSourceBuilder sourceBuilder = new SearchSourceBuilder(); // 指定size为0,不返回文档 因为只需要数量 sourceBuilder.query(queryBuilder).aggregation(dateHistogram).size(0); SearchRequest searchRequest = new SearchRequest(getTableName()); searchRequest.source(sourceBuilder); log.debug("sourceBuilder:{}", sourceBuilder); SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT); Aggregation agg = searchResponse.getAggregations().get(aggregationResultName); List buckets = ((Histogram) agg).getBuckets(); List chartBeanList = new ArrayList<>(); for (Histogram.Bucket bucket : buckets) { ChartBean chartBean = new ChartBean(); chartBean.setCategorie(bucket.getKeyAsString()); // chartBean.setSaleNum(Float.valueOf(bucket.getDocCount())); //统计商品数 ParsedSum saleNum = bucket.getAggregations().get(aggName_productNumber); String saleNumStr = format1((float) saleNum.getValue()); if(StringUtils.isEmpty(saleNumStr)){ saleNumStr = "1"; } chartBean.setSaleNum(Float.valueOf(saleNumStr)); //统计订单数 chartBean.setOrderNum(Float.valueOf(bucket.getDocCount())); // chartBean.setSaleNum(Long.valueOf(bucket.getDocCount()).intValue()); ParsedSum sum = bucket.getAggregations().get(aggSumName); String s = format1((float) sum.getValue()); chartBean.setSalePrice(Float.valueOf(s)); // chartBean.setSalePrice(Double.valueOf(sum.getValue()).intValue()); chartBeanList.add(chartBean); } ChartColumn chartColumn = new ChartColumn(chartBeanList); return chartColumn; } catch (Exception e) { log.error(getTableName() + " es根据条件 聚合查询 报错:{}", e); throw new MyException(getTableName() + " es根据条件 聚合查询 报错:" + e.getMessage()); } } public static String format1(Float value){ BigDecimal bd = new BigDecimal(value);//创建一个bd对象,将要转换的值value传入构造函数 bd = bd.setScale(2, RoundingMode.HALF_UP);//调用setScale方法进行数据格式化,保留两位小数,采用四舍五入规则 return bd.toString(); //返回bd对象的值(转化为string形式) } public void tongbuByHour() throws ParseException { String startDate = null; String endDate = null; SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); Calendar calendar = Calendar.getInstance(); calendar.setTime(new Date()); endDate = sdf.format(calendar.getTime()); calendar.add(calendar.HOUR_OF_DAY, -1); // calendar.add(calendar.DATE, -4); startDate = sdf.format(calendar.getTime()); Date start = sdf.parse(startDate); Date end = sdf.parse(endDate); // 查询es 现有的数据 //1,同步已支付的订单 int num = 0; while (true) { int limit = MAX_ROW; int offset= num * MAX_ROW; LambdaQueryWrapper query = Wrappers.lambdaQuery(); query.gt(TOrder::getCreateDate,start); query.lt(TOrder::getCreateDate,end); query.eq(TOrder::getStatus,1); Page page = new Page<>(offset, limit, true); IPage iPage = tOrderService.page(page, query); List list_es = iPage.getRecords(); if(list_es.size()>0){ insertBatch(list_es); for(TOrder order:list_es){ try { updateDataById(order); } catch (Exception e) { e.printStackTrace(); } } } num++; if(list_es.size()< MAX_ROW){ // 数据小于 最大值 ,证明后面已无数据,则跳出 break; } } //2,同步已退款的订单 int num2 = 0; while (true) { int limit = MAX_ROW; int offset= num2 * MAX_ROW; LambdaQueryWrapper query = Wrappers.lambdaQuery(); query.gt(TOrder::getCreateDate,start); query.lt(TOrder::getCreateDate,end); query.eq(TOrder::getStatus,3); Page page = new Page<>(offset, limit, true); IPage iPage = tOrderService.page(page, query); List list_es = iPage.getRecords(); if(list_es.size()>0){ insertBatch(list_es); for(TOrder order:list_es){ try { updateDataById(order); } catch (Exception e) { e.printStackTrace(); } } } num2++; if(list_es.size()< MAX_ROW){ // 数据小于 最大值 ,证明后面已无数据,则跳出 break; } } // 查询es 现有的数据 // LambdaQueryWrapper query = Wrappers.lambdaQuery(); // query.gt(TOrder::getCreateDate,start); // query.lt(TOrder::getCreateDate,end); // List list_es = tOrderService.list(query); // if (CollectionUtils.isNotEmpty(list_es)) { // insertBatch(list_es); // if(list_es.size()>0){ // for(TOrder order : list_es){ // try { // updateDataById(order); // } catch (Exception e) { // e.printStackTrace(); // } // } // } // } } public void tongbuByDay() throws ParseException { String startDate = null; String endDate = null; SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); Calendar calendar = Calendar.getInstance(); calendar.setTime(new Date()); endDate = sdf.format(calendar.getTime()); calendar.add(calendar.DATE, -1); // calendar.add(calendar.DATE, -4); startDate = sdf.format(calendar.getTime()); Date start = sdf.parse(startDate); Date end = sdf.parse(endDate); // 查询es 现有的数据 //1,同步已支付的订单 int num = 0; while (true) { int limit = MAX_ROW; int offset= num * MAX_ROW; LambdaQueryWrapper query = Wrappers.lambdaQuery(); query.gt(TOrder::getCreateDate,start); query.lt(TOrder::getCreateDate,end); query.eq(TOrder::getStatus,1); Page page = new Page<>(offset, limit, true); IPage iPage = tOrderService.page(page, query); List list_es = iPage.getRecords(); if(list_es.size()>0){ insertBatch(list_es); for(TOrder order:list_es){ try { updateDataById(order); } catch (Exception e) { e.printStackTrace(); } } } num++; if(list_es.size()< MAX_ROW){ // 数据小于 最大值 ,证明后面已无数据,则跳出 break; } } //2,同步已退款的订单 int num2 = 0; while (true) { int limit = MAX_ROW; int offset= num2 * MAX_ROW; LambdaQueryWrapper query = Wrappers.lambdaQuery(); query.gt(TOrder::getCreateDate,start); query.lt(TOrder::getCreateDate,end); query.eq(TOrder::getStatus,3); Page page = new Page<>(offset, limit, true); IPage iPage = tOrderService.page(page, query); List list_es = iPage.getRecords(); if(list_es.size()>0){ insertBatch(list_es); for(TOrder order:list_es){ try { updateDataById(order); } catch (Exception e) { e.printStackTrace(); } } } num2++; if(list_es.size()< MAX_ROW){ // 数据小于 最大值 ,证明后面已无数据,则跳出 break; } } // 查询es 现有的数据 // LambdaQueryWrapper query = Wrappers.lambdaQuery(); // query.gt(TOrder::getCreateDate,start); // query.lt(TOrder::getCreateDate,end); // List list_es = tOrderService.list(query); // if (CollectionUtils.isNotEmpty(list_es)) { // insertBatch(list_es); // if(list_es.size()>0){ // for(TOrder order : list_es){ // try { // updateDataById(order); // } catch (Exception e) { // e.printStackTrace(); // } // } // } // } } }