|
|
@@ -30,13 +30,25 @@ import com.ufo.base.common.utils.StringUtils;
|
|
|
import com.ufo.module.system.domain.SysDictData;
|
|
|
import com.ufo.project.config.domain.ConfigStation;
|
|
|
import com.ufo.project.data.domain.DataFileSyncFiles;
|
|
|
+import com.ufo.project.forecast.mediumterm.power.domain.DataMediumTermPowerForecastIssued;
|
|
|
import com.ufo.project.forecast.mediumterm.power.domain.DataMediumTermPowerForecastRaw;
|
|
|
+import com.ufo.project.forecast.mediumterm.power.domain.DataMediumTermPowerForecastReport;
|
|
|
+import com.ufo.project.forecast.mediumterm.power.mapper.DataMediumTermPowerForecastIssuedMapper;
|
|
|
import com.ufo.project.forecast.mediumterm.power.mapper.DataMediumTermPowerForecastRawMapper;
|
|
|
+import com.ufo.project.forecast.mediumterm.power.mapper.DataMediumTermPowerForecastReportMapper;
|
|
|
+import com.ufo.project.forecast.shortterm.power.domain.DataShortTermPowerForecastIssued;
|
|
|
import com.ufo.project.forecast.shortterm.power.domain.DataShortTermPowerForecastRaw;
|
|
|
+import com.ufo.project.forecast.shortterm.power.domain.DataShortTermPowerForecastReport;
|
|
|
+import com.ufo.project.forecast.shortterm.power.mapper.DataShortTermPowerForecastIssuedMapper;
|
|
|
import com.ufo.project.forecast.shortterm.power.mapper.DataShortTermPowerForecastRawMapper;
|
|
|
+import com.ufo.project.forecast.shortterm.power.mapper.DataShortTermPowerForecastReportMapper;
|
|
|
+import com.ufo.project.forecast.supershort.domain.DataSuperShortTermForecastIssued;
|
|
|
import com.ufo.project.forecast.supershort.domain.DataSuperShortTermForecastRaw;
|
|
|
+import com.ufo.project.forecast.supershort.domain.DataSuperShortTermForecastReport;
|
|
|
import com.ufo.project.forecast.supershort.domain.DataUltraShortTermPowerForecasting;
|
|
|
+import com.ufo.project.forecast.supershort.mapper.DataSuperShortTermForecastIssuedMapper;
|
|
|
import com.ufo.project.forecast.supershort.mapper.DataSuperShortTermForecastRawMapper;
|
|
|
+import com.ufo.project.forecast.supershort.mapper.DataSuperShortTermForecastReportMapper;
|
|
|
import com.ufo.project.forecast.supershort.mapper.DataUltraShortTermPowerForecastingMapper;
|
|
|
|
|
|
/**
|
|
|
@@ -44,14 +56,14 @@ import com.ufo.project.forecast.supershort.mapper.DataUltraShortTermPowerForecas
|
|
|
*
|
|
|
* 处理形如 {@code /home/power/upload/power/JMJD01/MediumTerm/2026-09-2026090312.csv} 的功率csv文件:
|
|
|
* 1. 逐级目录匹配字典 station_file_code_mapping(场站编码-文件映射)的key, 取value为场站编码;
|
|
|
- * 2. 目录含 MediumTerm 为中期、ShortTerm 为短期、UltraShortTerm 为超短期(优先判定), 分别写入
|
|
|
+ * 2. 目录含 MediumTerm 为中期、ShortTerm 为短期、UltraShortTerm 为超短期(优先判定), 分别 upsert
|
|
|
* data_medium_term_power_forecast_raw / data_short_term_power_forecast_raw /
|
|
|
- * data_super_short_term_forecast_raw;
|
|
|
+ * data_super_short_term_forecast_raw 及对应 _issued / _report 表(report表仅batch_no为空的行);
|
|
|
* 3. 文件名中 yyyyMMddHH(超短期为yyyyMMddHHmm, 已是UTC)转为起报时间start_time(epoch秒);
|
|
|
* 4. EasyExcel读csv, 按fieldMappingConfig映射(time -> forecast_time, ISO格式如 2026-09-19T12:00Z,
|
|
|
* 本身即UTC; power -> power), 经纬度/场站等其余字段按任务约定取固定元数据与场站信息;
|
|
|
- * 5. 按(station_id, start_time)查询已有记录, 以(start_time, forecast_time)匹配,
|
|
|
- * 已存在的批量更新power, 新数据批量插入, 每500条一批.
|
|
|
+ * 5. 每张目标表按场站字段(raw按station_id, issued/report按entity_id)与start_time查询已有记录,
|
|
|
+ * 以forecast_time匹配, 已存在的批量更新power, 新数据批量插入, 每500条一批.
|
|
|
*
|
|
|
* @author ufo
|
|
|
* @date 2026-09-22
|
|
|
@@ -126,6 +138,24 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
@Autowired
|
|
|
private DataUltraShortTermPowerForecastingMapper dataUltraShortTermPowerForecastingMapper;
|
|
|
|
|
|
+ @Autowired
|
|
|
+ private DataShortTermPowerForecastIssuedMapper dataShortTermPowerForecastIssuedMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DataShortTermPowerForecastReportMapper dataShortTermPowerForecastReportMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DataMediumTermPowerForecastIssuedMapper dataMediumTermPowerForecastIssuedMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DataMediumTermPowerForecastReportMapper dataMediumTermPowerForecastReportMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DataSuperShortTermForecastIssuedMapper dataSuperShortTermForecastIssuedMapper;
|
|
|
+
|
|
|
+ @Autowired
|
|
|
+ private DataSuperShortTermForecastReportMapper dataSuperShortTermForecastReportMapper;
|
|
|
+
|
|
|
@Override
|
|
|
protected String parseAndSave(DataFileSyncFiles file)
|
|
|
{
|
|
|
@@ -165,7 +195,7 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
return "csv无有效数据行";
|
|
|
}
|
|
|
|
|
|
- int[] result = new int[]{0, 0};
|
|
|
+ int[] result = new int[6];
|
|
|
switch (term.kind)
|
|
|
{
|
|
|
case SHORT:
|
|
|
@@ -181,8 +211,8 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
default:
|
|
|
break;
|
|
|
}
|
|
|
- log.info("[DataFileProcesserPower] 处理完成, file={}, 场站={}, {}, start_time={}, 更新{}条, 新增{}条",
|
|
|
- csvFile.getName(), stationCode, term.subjectName, startTime, result[0], result[1]);
|
|
|
+ log.info("[DataFileProcesserPower] 处理完成, file={}, 场站={}, {}, start_time={}, raw更新{}新增{}, 下发更新{}新增{}, 上报更新{}新增{}",
|
|
|
+ csvFile.getName(), stationCode, term.subjectName, startTime, result[0], result[1], result[2], result[3], result[4], result[5]);
|
|
|
return "";
|
|
|
}
|
|
|
|
|
|
@@ -310,45 +340,96 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 短期: 查重分类 + 500一批更新/插入 data_short_term_power_forecast_raw
|
|
|
+ * 短期: raw/issued/report 三表各做一次查重分类 + 500一批更新/插入
|
|
|
*
|
|
|
- * @return int[]{更新条数, 新增条数}
|
|
|
+ * @return int[]{raw更新, raw新增, issued更新, issued新增, report更新, report新增}
|
|
|
*/
|
|
|
private int[] saveShortTerm(Term term, ConfigStation station, long startTime, Map<Long, String> powerByForecastTime)
|
|
|
{
|
|
|
- return saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ int[] raw = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "stationId", station.getId(),
|
|
|
DataShortTermPowerForecastRaw::new,
|
|
|
dataShortTermPowerForecastRawMapper::selectDataShortTermPowerForecastRawList,
|
|
|
+ (row, forecastTime, power) -> fillPowerCommon(row, term, station, startTime, forecastTime, power),
|
|
|
dataShortTermPowerForecastRawMapper::batchUpdateDataShortTermPowerForecastRaw,
|
|
|
dataShortTermPowerForecastRawMapper::batchInsertDataShortTermPowerForecastRaw);
|
|
|
+ int[] issued = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "entityId", String.valueOf(station.getId()),
|
|
|
+ DataShortTermPowerForecastIssued::new,
|
|
|
+ dataShortTermPowerForecastIssuedMapper::selectByEntityAndStartTime,
|
|
|
+ (row, forecastTime, power) -> fillIssuedReport(row, term, station, startTime, forecastTime, power, false),
|
|
|
+ dataShortTermPowerForecastIssuedMapper::batchUpdateDataShortTermPowerForecastIssued,
|
|
|
+ dataShortTermPowerForecastIssuedMapper::batchInsertDataShortTermPowerForecastIssued);
|
|
|
+ int[] report = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "entityId", String.valueOf(station.getId()),
|
|
|
+ DataShortTermPowerForecastReport::new,
|
|
|
+ dataShortTermPowerForecastReportMapper::selectByEntityAndStartTime,
|
|
|
+ (row, forecastTime, power) -> fillIssuedReport(row, term, station, startTime, forecastTime, power, true),
|
|
|
+ dataShortTermPowerForecastReportMapper::batchUpdateDataShortTermPowerForecastReport,
|
|
|
+ dataShortTermPowerForecastReportMapper::insertBatch);
|
|
|
+ return new int[]{raw[0], raw[1], issued[0], issued[1], report[0], report[1]};
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 中期: 查重分类 + 500一批更新/插入 data_medium_term_power_forecast_raw
|
|
|
+ * 中期: raw/issued/report 三表各做一次查重分类 + 500一批更新/插入
|
|
|
*
|
|
|
- * @return int[]{更新条数, 新增条数}
|
|
|
+ * @return int[]{raw更新, raw新增, issued更新, issued新增, report更新, report新增}
|
|
|
*/
|
|
|
private int[] saveMediumTerm(Term term, ConfigStation station, long startTime, Map<Long, String> powerByForecastTime)
|
|
|
{
|
|
|
- return saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ int[] raw = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "stationId", station.getId(),
|
|
|
DataMediumTermPowerForecastRaw::new,
|
|
|
dataMediumTermPowerForecastRawMapper::selectDataMediumTermPowerForecastRawList,
|
|
|
+ (row, forecastTime, power) -> fillPowerCommon(row, term, station, startTime, forecastTime, power),
|
|
|
dataMediumTermPowerForecastRawMapper::batchUpdateDataMediumTermPowerForecastRaw,
|
|
|
dataMediumTermPowerForecastRawMapper::batchInsertDataMediumTermPowerForecastRaw);
|
|
|
+ int[] issued = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "entityId", String.valueOf(station.getId()),
|
|
|
+ DataMediumTermPowerForecastIssued::new,
|
|
|
+ dataMediumTermPowerForecastIssuedMapper::selectByEntityAndStartTime,
|
|
|
+ (row, forecastTime, power) -> fillIssuedReport(row, term, station, startTime, forecastTime, power, false),
|
|
|
+ dataMediumTermPowerForecastIssuedMapper::batchUpdateDataMediumTermPowerForecastIssued,
|
|
|
+ dataMediumTermPowerForecastIssuedMapper::batchInsertDataMediumTermPowerForecastIssued);
|
|
|
+ int[] report = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "entityId", String.valueOf(station.getId()),
|
|
|
+ DataMediumTermPowerForecastReport::new,
|
|
|
+ dataMediumTermPowerForecastReportMapper::selectByEntityAndStartTime,
|
|
|
+ (row, forecastTime, power) -> fillIssuedReport(row, term, station, startTime, forecastTime, power, true),
|
|
|
+ dataMediumTermPowerForecastReportMapper::batchUpdateDataMediumTermPowerForecastReport,
|
|
|
+ dataMediumTermPowerForecastReportMapper::insertBatch);
|
|
|
+ return new int[]{raw[0], raw[1], issued[0], issued[1], report[0], report[1]};
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 超短期: 查重分类 + 500一批更新/插入 data_super_short_term_forecast_raw
|
|
|
+ * 超短期: raw/issued/report 三表各做一次查重分类 + 500一批更新/插入
|
|
|
*
|
|
|
- * @return int[]{更新条数, 新增条数}
|
|
|
+ * @return int[]{raw更新, raw新增, issued更新, issued新增, report更新, report新增}
|
|
|
*/
|
|
|
private int[] saveUltraShortTerm(Term term, ConfigStation station, long startTime, Map<Long, String> powerByForecastTime)
|
|
|
{
|
|
|
- return saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ int[] raw = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "stationId", station.getId(),
|
|
|
DataSuperShortTermForecastRaw::new,
|
|
|
dataSuperShortTermForecastRawMapper::selectDataSuperShortTermForecastRawList,
|
|
|
+ (row, forecastTime, power) -> fillPowerCommon(row, term, station, startTime, forecastTime, power),
|
|
|
dataSuperShortTermForecastRawMapper::batchUpdateDataSuperShortTermForecastRaw,
|
|
|
dataSuperShortTermForecastRawMapper::batchInsertDataSuperShortTermForecastRaw);
|
|
|
+ int[] issued = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "entityId", String.valueOf(station.getId()),
|
|
|
+ DataSuperShortTermForecastIssued::new,
|
|
|
+ dataSuperShortTermForecastIssuedMapper::selectByEntityAndStartTime,
|
|
|
+ (row, forecastTime, power) -> fillIssuedReport(row, term, station, startTime, forecastTime, power, false),
|
|
|
+ dataSuperShortTermForecastIssuedMapper::batchUpdateDataSuperShortTermForecastIssued,
|
|
|
+ dataSuperShortTermForecastIssuedMapper::batchInsertDataSuperShortTermForecastIssued);
|
|
|
+ int[] report = saveTerm(term, station, startTime, powerByForecastTime,
|
|
|
+ "entityId", String.valueOf(station.getId()),
|
|
|
+ DataSuperShortTermForecastReport::new,
|
|
|
+ dataSuperShortTermForecastReportMapper::selectByEntityAndStartTime,
|
|
|
+ (row, forecastTime, power) -> fillIssuedReport(row, term, station, startTime, forecastTime, power, true),
|
|
|
+ dataSuperShortTermForecastReportMapper::batchUpdateDataSuperShortTermForecastReport,
|
|
|
+ dataSuperShortTermForecastReportMapper::insertBatch);
|
|
|
+ return new int[]{raw[0], raw[1], issued[0], issued[1], report[0], report[1]};
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
@@ -411,18 +492,18 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
}
|
|
|
|
|
|
/**
|
|
|
- * 三条链路共用: 按(station_id, start_time)查已有记录, 以forecast_time匹配,
|
|
|
- * 已有置Id进更新组、其余进插入组, 各按BATCH_SIZE一批写库;
|
|
|
- * 三类实体无公共父类故stationId/startTime/Id/forecastTime经反射读写(与fillPowerCommon同法)
|
|
|
+ * 共用upsert: 按stationField/stationValue与start_time查已有记录, 以forecast_time匹配,
|
|
|
+ * 已有置Id进更新组、其余经rowFiller填充后进插入组, 各按BATCH_SIZE一批写库;
|
|
|
+ * 各表实体无公共父类故字段经反射读写(与fillPowerCommon同法)
|
|
|
*
|
|
|
* @return int[]{更新条数, 新增条数}
|
|
|
*/
|
|
|
private <T> int[] saveTerm(Term term, ConfigStation station, long startTime, Map<Long, String> powerByForecastTime,
|
|
|
- Supplier<T> factory, Function<T, List<T>> selectList,
|
|
|
- Consumer<List<T>> batchUpdate, Consumer<List<T>> batchInsert)
|
|
|
+ String stationField, Object stationValue, Supplier<T> factory, Function<T, List<T>> selectList,
|
|
|
+ RowFiller rowFiller, Consumer<List<T>> batchUpdate, Consumer<List<T>> batchInsert)
|
|
|
{
|
|
|
T query = factory.get();
|
|
|
- setEntityField(query, "stationId", station.getId());
|
|
|
+ setEntityField(query, stationField, stationValue);
|
|
|
setEntityField(query, "startTime", startTime);
|
|
|
List<T> dbList = selectList.apply(query);
|
|
|
Map<Long, Long> existIds = new HashMap<>();
|
|
|
@@ -439,17 +520,17 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
List<T> newList = new ArrayList<>();
|
|
|
for (Map.Entry<Long, String> row : powerByForecastTime.entrySet())
|
|
|
{
|
|
|
- T raw = factory.get();
|
|
|
- fillPowerCommon(raw, term, station, startTime, row.getKey(), row.getValue());
|
|
|
+ T entity = factory.get();
|
|
|
+ rowFiller.fill(entity, row.getKey(), row.getValue());
|
|
|
Long existId = existIds.get(row.getKey());
|
|
|
if (existId != null)
|
|
|
{
|
|
|
- setEntityField(raw, "Id", existId);
|
|
|
- dbExistList.add(raw);
|
|
|
+ setEntityField(entity, "Id", existId);
|
|
|
+ dbExistList.add(entity);
|
|
|
}
|
|
|
else
|
|
|
{
|
|
|
- newList.add(raw);
|
|
|
+ newList.add(entity);
|
|
|
}
|
|
|
}
|
|
|
for (List<T> batch : partition(dbExistList, BATCH_SIZE))
|
|
|
@@ -507,6 +588,40 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * 写入issued/report两表链路共有的元数据/场站/时间/功率字段(字段同名, 无公共父类故用反射);
|
|
|
+ * predays短中期写入, preQrts仅超短期; batch_no恒不写(置空),
|
|
|
+ * createTimeEpoch仅短/超短期report实体存在该字段且insertBatch按列取值, 中期report由库端取当前epoch秒
|
|
|
+ */
|
|
|
+ void fillIssuedReport(Object row, Term term, ConfigStation station, long startTime, long forecastTime, String power, boolean isReport)
|
|
|
+ {
|
|
|
+ Map<String, Object> values = new LinkedHashMap<>();
|
|
|
+ values.put("dataSourcegrabId", term.id);
|
|
|
+ values.put("dataSourcegrabPatternName", term.patternName);
|
|
|
+ values.put("entityId", String.valueOf(station.getId()));
|
|
|
+ values.put("entityType", 1L);
|
|
|
+ values.put("entityName", station.getStationName());
|
|
|
+ values.put("startTime", startTime);
|
|
|
+ values.put("forecastTime", forecastTime);
|
|
|
+ if (term.kind == Kind.ULTRA)
|
|
|
+ {
|
|
|
+ values.put("preQrts", term.preQrts);
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ values.put("predays", term.preDays);
|
|
|
+ }
|
|
|
+ if (isReport && term.kind != Kind.MEDIUM)
|
|
|
+ {
|
|
|
+ values.put("createTimeEpoch", System.currentTimeMillis() / 1000);
|
|
|
+ }
|
|
|
+ values.put("power", power);
|
|
|
+ for (Map.Entry<String, Object> value : values.entrySet())
|
|
|
+ {
|
|
|
+ setEntityField(row, value.getKey(), value.getValue());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
/**
|
|
|
* 解析文件名时间戳为起报epoch秒(UTC): 短期/中期10位yyyyMMddHH, 超短期12位yyyyMMddHHmm
|
|
|
*/
|
|
|
@@ -578,6 +693,13 @@ public class DataFileProcesserPowerImpl extends AbstraceDataFileProcesser
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ /** upsert行填充器: 按预测时间/功率填充一条新实体 */
|
|
|
+ @FunctionalInterface
|
|
|
+ interface RowFiller
|
|
|
+ {
|
|
|
+ void fill(Object entity, long forecastTime, String power);
|
|
|
+ }
|
|
|
+
|
|
|
/** 时效类型(路由到各自保存链路与实体) */
|
|
|
enum Kind
|
|
|
{
|