Microservice/pqs
5
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

代码调整

Browse Source
This commit is contained in:
hongawen
2022-11-07 09:11:09 +08:00
parent c53cd9b2dc
commit 0e5f9cc8df
51 changed files with 1469 additions and 851 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/CoustomReportFeignClientFallbackFactory.java
View File

@@ -35,8 +35,8 @@ public class CoustomReportFeignClientFallbackFactory implements FallbackFactory<
Enum<?> finalExceptionEnum = exceptionEnum;
return new CoustmReportFeignClient() {
@Override
public HttpResult<Boolean> batchReport(LineParam reportParam){
log.error("{}异常，降级处理，异常为：{}", "自定义报表预处理任务: ", throwable.toString());
public HttpResult<Boolean> batchReport(@RequestBody LineParam reportParam){
log.error("{}异常，降级处理，异常为：{}", "Date数据转Day数据: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
};

2
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/DayDataFeignClientFallbackFactory.java
View File

@@ -33,7 +33,7 @@ public class DayDataFeignClientFallbackFactory implements FallbackFactory<DayDat
Enum<?> finalExceptionEnum = exceptionEnum;
return new DayDataFeignClient() {
@Override
public HttpResult<Boolean> dayDataHanlder(LineParam jobParam){
public HttpResult<Boolean> dayDataHanlder(@RequestBody LineParam jobParam){
log.error("{}异常，降级处理，异常为：{}", "Date数据转Day数据: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}

4
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/IntegrityFeignClientFallbackFactory.java
View File

@@ -30,8 +30,8 @@ public class IntegrityFeignClientFallbackFactory implements FallbackFactory<Inte
Enum<?> finalExceptionEnum = exceptionEnum;
return new IntegrityFeignClient() {
@Override
public HttpResult<String> computeDataIntegrity(LineParam lineParam) {
log.error("{}异常，降级处理，异常为：{}", "数据完整性预处理: ", throwable.toString());
public HttpResult<String> computeDataIntegrity(@RequestBody @Validated LineParam lineParam) {
log.error("{}异常，降级处理，异常为：{}", "数据完整性处理: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
};

4
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/LimitTargetFeignClientFallbackFactory.java
View File

@@ -34,8 +34,8 @@ public class LimitTargetFeignClientFallbackFactory implements FallbackFactory<Li
Enum<?> finalExceptionEnum = exceptionEnum;
return new LimitTargetFeignClient() {
@Override
public HttpResult<String> getLimitTargetData(LineParam lineParam) {
log.error("{}异常，降级处理，异常为：{}", "越限次数数据预处理: ", throwable.toString());
public HttpResult<String> getLimitTargetData(@RequestBody @Validated LineParam lineParam) {
log.error("{}异常，降级处理，异常为：{}", "越限数据: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
};

4
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/LimitrateFeignClientFallbackFactory.java
View File

@@ -33,8 +33,8 @@ public class LimitrateFeignClientFallbackFactory implements FallbackFactory<Limi
Enum<?> finalExceptionEnum = exceptionEnum;
return new LimitrateFeignClient() {
@Override
public HttpResult<Boolean> limitRateHanlder(LineParam limitRateHanlderParam ){
log.error("{}异常，降级处理，异常为：{}", "越限是否数据预处理: ", throwable.toString());
public HttpResult<Boolean> limitRateHanlder(@RequestBody LineParam limitRateHanlderParam ){
log.error("{}异常，降级处理，异常为：{}", "越限数据处理: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
};

2
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/NormalFeignClientFallbackFactory.java
View File

@@ -28,7 +28,7 @@ public class NormalFeignClientFallbackFactory implements FallbackFactory<NormalL
return new NormalLimitFeignClient() {
@Override
public HttpResult<String> getNormLimitData() {
log.error("{}异常，降级处理，异常为：{}", "告警数据预处理: ", throwable.toString());
log.error("{}异常，降级处理，异常为：{}", "告警数据: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
};

4
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/OnlineRateFeignClientFallbackFactory.java
View File

@@ -30,8 +30,8 @@ public class OnlineRateFeignClientFallbackFactory implements FallbackFactory<Onl
Enum<?> finalExceptionEnum = exceptionEnum;
return new OnlineRateFeignClient() {
@Override
public HttpResult<String> getOnlineRateData(LineParam lineParam){
log.error("{}异常，降级处理，异常为：{}", "在线率数据预处理: ", throwable.toString());
public HttpResult<String> getOnlineRateData(@RequestBody @Validated LineParam lineParam){
log.error("{}异常，降级处理，异常为：{}", "在线率: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
};

4
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/api/line/fallback/PollutionFeignClientFallbackFactory.java
View File

@@ -31,8 +31,8 @@ public class PollutionFeignClientFallbackFactory implements FallbackFactory<Poll
Enum<?> finalExceptionEnum = exceptionEnum;
return new PollutionFeignClient() {
@Override
public HttpResult<String> processPollutionData(LineParam lineParam){
log.error("{}异常，降级处理，异常为：{}", "污区数据预处理: ", throwable.toString());
public HttpResult<String> processPollutionData(@RequestBody @Validated LineParam lineParam){
log.error("{}异常，降级处理，异常为：{}", "污区数据: ", throwable.toString());
throw new BusinessException(finalExceptionEnum);
}
};

11
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/controller/line/DayDataController.java
View File

@@ -6,7 +6,7 @@ import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.device.pq.api.LineFeignClient;
import com.njcn.prepare.harmonic.pojo.param.LimitRateHanlderParam;
import com.njcn.prepare.harmonic.pojo.param.LineParam;
import com.njcn.prepare.harmonic.service.line.DayDataService;
import com.njcn.web.controller.BaseController;
import io.swagger.annotations.Api;
@@ -20,6 +20,7 @@ import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
@@ -47,7 +48,8 @@ public class DayDataController extends BaseController {
@ApiImplicitParam(value = "jobParam",name = "jobParam",required = true)
@PostMapping("dayDataHanlder")
@OperateInfo(info = LogEnum.BUSINESS_MEDIUM)
public HttpResult<Boolean> dayDataHanlder(@RequestBody LimitRateHanlderParam jobParam ){
public HttpResult<Boolean> dayDataHanlder(@RequestBody LineParam jobParam ){
log.info(LocalDateTime.now()+"dayDataHanlder开始执行");
String methodDescribe = getMethodDescribe("dayDataHanlder");
Boolean result = true;
List<String> indexLists = new ArrayList<> ();
@@ -56,9 +58,10 @@ public class DayDataController extends BaseController {
}else{
indexLists = jobParam.getLineIds ();
}
String startTime = jobParam.getDataDate ()+" "+"00:00:00";
String endTime = jobParam.getDataDate ()+" "+"23:59:59";
dayDataService.dayDataJobHandler (indexLists,jobParam.getStartTime (),jobParam.getEndTime ());
dayDataService.dayDataJobHandler (indexLists,startTime,endTime);
if (result){
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, true, methodDescribe);
} else {

9
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/controller/line/LimitrateController.java
View File

@@ -6,7 +6,7 @@ import com.njcn.common.pojo.enums.response.CommonResponseEnum;
import com.njcn.common.pojo.response.HttpResult;
import com.njcn.common.utils.HttpResultUtil;
import com.njcn.device.pq.api.LineFeignClient;
import com.njcn.prepare.harmonic.pojo.param.LimitRateHanlderParam;
import com.njcn.prepare.harmonic.pojo.param.LineParam;
import com.njcn.prepare.harmonic.service.Impl.line.LimitRateService;
import com.njcn.web.controller.BaseController;
import io.swagger.annotations.Api;
@@ -48,7 +48,7 @@ public class LimitrateController extends BaseController {
@ApiImplicitParam(value = "limitRateHanlderParam",name = "limitRateHanlderParam",required = true)
@PostMapping("LimitRateHanlder")
@OperateInfo(info = LogEnum.BUSINESS_MEDIUM)
public HttpResult<Boolean> limitRateHanlder(@RequestBody LimitRateHanlderParam limitRateHanlderParam ){
public HttpResult<Boolean> limitRateHanlder(@RequestBody LineParam limitRateHanlderParam ){
String methodDescribe = getMethodDescribe("limitRateHanlder");
Boolean result = true;
List<String> indexLists = new ArrayList<> ();
@@ -57,9 +57,10 @@ public class LimitrateController extends BaseController {
}else{
indexLists = limitRateHanlderParam.getLineIds ();
}
String startTime = limitRateHanlderParam.getDataDate ()+" "+"00:00:00";
String endTime = limitRateHanlderParam.getDataDate ()+" "+"23:59:59";
limitRateService.limitRateJobHandler (indexLists,limitRateHanlderParam.getStartTime (),limitRateHanlderParam.getEndTime ());
limitRateService.limitRateJobHandler (indexLists,startTime,endTime);
if (result){
return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, true, methodDescribe);
} else {

2
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/mapper/line/mapping/ExcelRptTempMapper.xml
View File

@@ -2,7 +2,7 @@
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
<mapper namespace="com.njcn.prepare.harmonic.mapper.line.ExcelRptTempMapper">
<select id="getActiveTempList" resultType="com.njcn.prepare.harmonic.pojo.po.ExcelRptTemp">
<select id="getActiveTempList" resultType="ExcelRptTemp">
SELECT
DISTINCT t1.*
FROM

2
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/pojo/po/DataPolluctionPO.java
View File

@@ -12,7 +12,7 @@ import java.time.Instant;
* @createTime 2022/10/21 13:45
*/
@Data
@Measurement(name = "data_polluction")
@Measurement(name = "harmonic_pollution")
public class DataPolluctionPO {
@Column(name = "line_id")

3
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/pojo/po/PqsCommunicatePO.java
View File

@@ -22,6 +22,9 @@ public class PqsCommunicatePO {
@Column(name = "line_id")
private String lineId;
@Column(name = "dev_id")
private String devId;
@Column(name = "type")
private Integer type;

4
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/service/Impl/line/OnlineRateServiceImpl.java
View File

@@ -77,7 +77,7 @@ public class OnlineRateServiceImpl implements OnlineRateService {
* 获取pqs_communicate数据
*/
private List<PqsCommunicatePO> getCommunicateData(String lineId){
QueryResult sqlResult = influxDbUtils.query("SELECT * FROM pqs_communicate where line_id = '" + lineId +"' order by time desc limit 1 tz('Asia/Shanghai')");
QueryResult sqlResult = influxDbUtils.query("SELECT * FROM pqs_communicate where dev_id = '" + lineId +"' order by time desc limit 1 tz('Asia/Shanghai')");
InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
return resultMapper.toPOJO(sqlResult, PqsCommunicatePO.class);
}
@@ -86,7 +86,7 @@ public class OnlineRateServiceImpl implements OnlineRateService {
* 获取范围时间内的pqs_communicate数据
*/
private List<PqsCommunicatePO> getCommunicateData(String lineId, String date){
QueryResult sqlResult = influxDbUtils.query("SELECT * FROM pqs_communicate where time >= '" + date + " 00:00:00' and time <= '" + date + " 23:59:59' and line_id = '" + lineId +"' order by time asc tz('Asia/Shanghai')");
QueryResult sqlResult = influxDbUtils.query("SELECT * FROM pqs_communicate where time >= '" + date + " 00:00:00' and time <= '" + date + " 23:59:59' and dev_id = '" + lineId +"' order by time asc tz('Asia/Shanghai')");
InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
return resultMapper.toPOJO(sqlResult, PqsCommunicatePO.class);
}

12
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/service/Impl/line/PollutionServiceImpl.java
View File

@@ -106,14 +106,14 @@ public class PollutionServiceImpl implements PollutionService {
pollutionList = processPollutionList(lineIdList,harmonicVoltageList,harmonicCurrentList,frequencyDeviationList,voltageDeviationList,threePhaseVoltageList,negativeSequenceList,interharmonicVoltageList,voltageFlickerList);
Date dateOut = new Date();
//入表data_polluction
//入表harmonic_pollution
if (StrUtil.isNotBlank(lineParam.getDataDate())){
dateOut = DateUtil.parse(lineParam.getDataDate());
}
insertPolluction(pollutionList,dateOut.getTime());
LogUtil.njcnDebug(log, "监测点污染指标数据data_polluction插入耗时：{}", timer.intervalRestart());
LogUtil.njcnDebug(log, "监测点污染指标数据harmonic_pollution插入耗时：{}", timer.intervalRestart());
}else {
//获取data_polluction数据
//获取harmonic_pollution数据
pollutionList = getDataPolluction(lineParam);
}
//MySql入表污区图表等
@@ -153,7 +153,7 @@ public class PollutionServiceImpl implements PollutionService {
List<PollutionDTO> pollutionDTOList = new ArrayList<>();
InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
for (String lineId : lineList){
String sql="SELECT * FROM data_polluction where line_id = '" + lineId +"' "+processDate(lineParam.getDataDate(),lineParam.getType());
String sql="SELECT * FROM harmonic_pollution where line_id = '" + lineId +"' "+processDate(lineParam.getDataDate(),lineParam.getType());
QueryResult dataPolluctionResult = influxDbUtils.query(sql);
List<DataPolluctionPO> threePhaseList = resultMapper.toPOJO(dataPolluctionResult, DataPolluctionPO.class);
for (DataPolluctionPO dataPolluction : threePhaseList){
@@ -624,7 +624,7 @@ public class PollutionServiceImpl implements PollutionService {
}
/**
* 监测点污染指标数据入表 data_polluction
* 监测点污染指标数据入表 harmonic_pollution
*/
private void insertPolluction(List<PollutionDTO> list, long time){
List<String> records = new ArrayList<String>();
@@ -640,7 +640,7 @@ public class PollutionServiceImpl implements PollutionService {
fields.put("i_all",item.getIAll());
fields.put("v_inharm",item.getVInharm());
fields.put("plt",item.getPlt());
Point point = influxDbUtils.pointBuilder("data_polluction", time, TimeUnit.MILLISECONDS,tags, fields);
Point point = influxDbUtils.pointBuilder("harmonic_pollution", time, TimeUnit.MILLISECONDS,tags, fields);
BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).tag("line_id", item.getLineId()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
batchPoints.point(point);
records.add(batchPoints.lineProtocol());

130
pqs-prepare/harmonic-prepare/src/main/java/com/njcn/prepare/harmonic/service/Impl/line/ReportServiceImpl.java
View File

@@ -7,6 +7,7 @@ import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.cloud.commons.lang.StringUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.njcn.common.config.GeneralInfo;
import com.njcn.common.pojo.constant.BizParamConstant;
import com.njcn.common.pojo.enums.common.DataStateEnum;
import com.njcn.common.pojo.exception.BusinessException;
@@ -16,6 +17,9 @@ import com.njcn.harmonic.pojo.dto.ReportTemplateDTO;
import com.njcn.influxdb.param.InfluxDBSqlConstant;
import com.njcn.influxdb.param.InfluxDBTableConstant;
import com.njcn.influxdb.utils.InfluxDbUtils;
import com.njcn.minio.bo.MinIoUploadResDTO;
import com.njcn.minio.config.MinIoProperties;
import com.njcn.minio.utils.MinIoUtils;
import com.njcn.prepare.harmonic.constant.Param;
import com.njcn.prepare.harmonic.mapper.line.ExcelRptMapper;
import com.njcn.prepare.harmonic.mapper.line.ExcelRptTempMapper;
@@ -25,9 +29,18 @@ import com.njcn.prepare.harmonic.pojo.po.ExcelRptTemp;
import com.njcn.prepare.harmonic.service.line.ReportService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.fileupload.FileItem;
import org.apache.commons.fileupload.disk.DiskFileItemFactory;
import org.apache.commons.io.IOUtils;
import org.influxdb.dto.QueryResult;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.multipart.commons.CommonsMultipartFile;
import javax.annotation.Resource;
import java.io.*;
import java.net.URL;
import java.util.*;
import java.util.stream.Collectors;
@@ -49,6 +62,14 @@ public class ReportServiceImpl implements ReportService {
private final ExcelRptMapper excelRptMapper;
private final GeneralInfo generalInfo;
@Resource
private MinIoUtils minIoUtils;
@Resource
private MinIoProperties minIoProperties;
@Override
public boolean batchReport(LineParam reportParam) {
@@ -77,7 +98,8 @@ public class ReportServiceImpl implements ReportService {
for (ExcelRptTemp excelRptTemp : reportTemplateList) {
try {
//获取content解析数据
jsonArray = JSONUtil.parseArray(excelRptTemp.getContent());
String objectUrl = minIoUtils.getObjectUrl(minIoProperties.getBucket(), excelRptTemp.getContent(), 7 * 24 * 60 * 60);
jsonArray = JSONUtil.parseArray(urlToString(objectUrl));
dataList = getDataList(jsonArray);
} catch (Exception e) {
throw new BusinessException(HarmonicResponseEnum.CUSTOM_REPORT_JSON);
@@ -120,7 +142,9 @@ public class ReportServiceImpl implements ReportService {
//月：例如2022十月份，传入2022-10-01进行匹配，有则更新无则插入
//周：例如2022年第五周，传入2022-01-23（周一）进行匹配，有则更新无则插入
//日：直接插入，无需配对
String afterContent = jsonArray.toString();
//文件上传到Minio服务器，存入文件名
MinIoUploadResDTO minIoUploadResDTO = contentToMinio(jsonArray.toString());
String afterContent = minIoUploadResDTO.getMinFileName();
if (BizParamConstant.STAT_BIZ_DAY.equals(reportParam.getType().toString())){
rptInsert(reportParam, lineId, excelRptTemp, afterContent);
}else{
@@ -299,4 +323,106 @@ public class ReportServiceImpl implements ReportService {
}
/**
* 上传文件到Minio
*
* @param content 文件
* @return 成功标记
*/
private MinIoUploadResDTO contentToMinio(String content) {
//上传到minio
String businessTempPath = generalInfo.getBusinessTempPath();
File file = stringToFile(content, businessTempPath + File.separator + "a.json");
MultipartFile multiFile = getMultipartFile(file);
try {
//把名称存入数据
MinIoUploadResDTO upload = minIoUtils.upload(multiFile, minIoProperties.getBucket(), "report/");
return upload;
} catch (Exception e) {
throw new BusinessException(HarmonicResponseEnum.CUSTOM_REPORT_FILE);
}
}
/**
* 将字符串写入指定文件
*
* @param res 原字符串
* @param filePath 文件路径
* @return 成功标记
*/
public File stringToFile(String res, String filePath) {
boolean flag = true;
BufferedReader bufferedReader = null;
BufferedWriter bufferedWriter = null;
File distFile = new File(filePath);
try {
if (!distFile.getParentFile().exists()){
distFile.getParentFile().mkdirs();
}
bufferedReader = new BufferedReader(new StringReader(res));
bufferedWriter = new BufferedWriter(new FileWriter(distFile));
//先清空
bufferedWriter.write("");
char buf[] = new char[1024]; //字符缓冲区
int len;
while ((len = bufferedReader.read(buf)) != -1) {
bufferedWriter.write(buf, 0, len);
}
bufferedWriter.flush();
bufferedReader.close();
bufferedWriter.close();
} catch (IOException e) {
e.printStackTrace();
} finally {
if (bufferedReader != null) {
try {
bufferedReader.close();
} catch (IOException e) {
e.printStackTrace();
}
}
}
return distFile;
}
/**
* 将文件转成Multipart
*
* @param file 文件
* @return 成功标记
*/
private MultipartFile getMultipartFile(File file) {
FileItem item = new DiskFileItemFactory().createItem("file"
, MediaType.MULTIPART_FORM_DATA_VALUE
, true
, file.getName());
try (InputStream input = new FileInputStream(file);
OutputStream os = item.getOutputStream()) {
// 流转移
IOUtils.copy(input, os);
} catch (Exception e) {
throw new IllegalArgumentException("Invalid file: " + e, e);
}
return new CommonsMultipartFile(item);
}
/**
* 将文件Url读取，转为String
*
* @param objectUrl 文件url
* @return 成功标记
*/
private String urlToString(String objectUrl) throws IOException {
URL url = new URL(objectUrl);
BufferedReader in = new BufferedReader(new InputStreamReader(url.openStream()));
StringBuffer buffer = new StringBuffer();
String line = " ";
while ((line = in.readLine()) != null){
buffer.append(line);
}
return buffer.toString();
}
}