|
|
@@ -0,0 +1,177 @@
|
|
|
+package com.webflux.launchadmin.mysql.service.sourceMaterial;
|
|
|
+
|
|
|
+
|
|
|
+import com.fasterxml.jackson.core.JsonParser;
|
|
|
+import com.fasterxml.jackson.core.JsonProcessingException;
|
|
|
+import com.fasterxml.jackson.databind.DeserializationContext;
|
|
|
+import com.fasterxml.jackson.databind.JsonDeserializer;
|
|
|
+import com.fasterxml.jackson.databind.ObjectMapper;
|
|
|
+import com.fasterxml.jackson.databind.SerializationFeature;
|
|
|
+import com.fasterxml.jackson.databind.module.SimpleModule;
|
|
|
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
|
|
|
+import com.webflux.launchadmin.config.BaseContextHandler;
|
|
|
+import com.webflux.launchadmin.global.BaseException;
|
|
|
+import com.webflux.launchadmin.mysql.controller.planNew.req.PosterTemplateRequset;
|
|
|
+import com.webflux.launchadmin.mysql.entity.listening.ListeningPlanNew;
|
|
|
+import com.webflux.launchadmin.mysql.entity.planNew.PosterTemplate;
|
|
|
+import com.webflux.launchadmin.mysql.service.sourceMaterial.data.sql.Where;
|
|
|
+import com.webflux.launchadmin.mysql.task.SyncPlanNewListenToRedis;
|
|
|
+import com.webflux.launchcommon.returnObj.Paged;
|
|
|
+import com.webflux.launchcommon.returnObj.RStatus;
|
|
|
+import jakarta.annotation.Resource;
|
|
|
+import org.springframework.data.domain.Sort;
|
|
|
+import org.springframework.data.r2dbc.core.R2dbcEntityTemplate;
|
|
|
+import org.springframework.data.relational.core.query.Criteria;
|
|
|
+import org.springframework.data.relational.core.query.Query;
|
|
|
+import org.springframework.data.relational.core.query.Update;
|
|
|
+import org.springframework.stereotype.Service;
|
|
|
+import org.springframework.web.bind.annotation.RequestBody;
|
|
|
+import reactor.core.publisher.Mono;
|
|
|
+
|
|
|
+import java.io.IOException;
|
|
|
+import java.time.LocalDateTime;
|
|
|
+import java.time.ZoneId;
|
|
|
+import java.time.ZonedDateTime;
|
|
|
+import java.time.format.DateTimeFormatter;
|
|
|
+import java.util.List;
|
|
|
+import java.util.function.Function;
|
|
|
+
|
|
|
+
|
|
|
+@Service
|
|
|
+public class SourceMaterialImpl implements SourceMaterialInterface{
|
|
|
+ @Resource
|
|
|
+ private R2dbcEntityTemplate template;
|
|
|
+ @Override
|
|
|
+ public <T> Mono<RStatus<Paged<T>>> page(List<Where> wheres, Class<T> domainType,Integer size ,Integer page){
|
|
|
+ return pageAndFunction(wheres, domainType, size , page, (x)-> x);
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T> Mono<RStatus<Paged<T>>> pageAndFunction(List<Where> wheres, Class<T> domainType, Integer size , Integer page, Function<List<T>,List<T>> function){
|
|
|
+ Criteria criteria = Criteria.empty();
|
|
|
+ if (!wheres.isEmpty()) {
|
|
|
+ for (Where entry: wheres){
|
|
|
+ String key = entry.getKey();
|
|
|
+ Object value = entry.getValue();
|
|
|
+ if(entry.getWhereType()==1){
|
|
|
+ criteria=criteria.and(key).is(value);
|
|
|
+ }
|
|
|
+ if(entry.getWhereType()==2){
|
|
|
+ criteria=criteria.and(key).like("%"+value+"%");
|
|
|
+ }
|
|
|
+ if(entry.getWhereType()==3){
|
|
|
+ criteria=criteria.and(key).in(value);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ Query query = Query.query(criteria);
|
|
|
+ Criteria finalCriteria = criteria;
|
|
|
+ Mono<Paged<T>> pagedMono = template.count(query, domainType).flatMap(fm ->
|
|
|
+ template.select(Query.query(finalCriteria)
|
|
|
+ .offset((long) (page - 1) * size)
|
|
|
+ .limit(size)
|
|
|
+ .sort(Sort.by(Sort.Order.asc("created_at"))), domainType)
|
|
|
+ .collectList()
|
|
|
+ .map(list -> {
|
|
|
+ List<T> apply = function.apply(list);
|
|
|
+ return new Paged<T>(fm, apply, page, size);
|
|
|
+ }
|
|
|
+ )
|
|
|
+ );
|
|
|
+ return RStatus.successList(pagedMono);
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T,E> Mono<RStatus<Paged<E>>> pageAndFunctionRtE(List<Where> wheres, Class<T> domainType, Integer size , Integer page, Function<List<T>,List<E>> function){
|
|
|
+ Criteria criteria = Criteria.empty();
|
|
|
+ if (!wheres.isEmpty()) {
|
|
|
+ for (Where entry: wheres){
|
|
|
+ String key = entry.getKey();
|
|
|
+ Object value = entry.getValue();
|
|
|
+ if(entry.getWhereType()==1){
|
|
|
+ criteria=criteria.and(key).is(value);
|
|
|
+ }
|
|
|
+ if(entry.getWhereType()==2){
|
|
|
+ criteria=criteria.and(key).like("%"+value+"%");
|
|
|
+ }
|
|
|
+ if(entry.getWhereType()==3){
|
|
|
+ criteria=criteria.and(key).in(value);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ Query query = Query.query(criteria);
|
|
|
+ Criteria finalCriteria = criteria;
|
|
|
+ Mono<Paged<E>> pagedMono = template.count(query, domainType).flatMap(fm ->
|
|
|
+ template.select(Query.query(finalCriteria)
|
|
|
+ .offset((long) (page - 1) * size)
|
|
|
+ .limit(size)
|
|
|
+ .sort(Sort.by(Sort.Order.asc("created_at"))), domainType)
|
|
|
+ .collectList()
|
|
|
+ .map(list -> {
|
|
|
+ List<E> apply = function.apply(list);
|
|
|
+ return new Paged<E>(fm, apply, page, size);
|
|
|
+ }
|
|
|
+ )
|
|
|
+ );
|
|
|
+ return RStatus.successList(pagedMono);
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T> Mono<RStatus<Long>> deleteById(String column, Long columnValue,Class<T> domainType) {
|
|
|
+ return RStatus.success(template.delete(Query.query(Criteria.where(column).is(columnValue)),domainType));
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T> Mono<RStatus<Long>> deleteByIdDeleteAt(String column, Long columnValue,Class<T> domainType,String delete) {
|
|
|
+ Update update =Update.update(delete, ZonedDateTime.now());
|
|
|
+ Query query = Query.query(Criteria.where("column").is(columnValue));
|
|
|
+ return RStatus.success(template.update(query, update, domainType));
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T> Mono<RStatus<T>> save( String ObjStr,Class<T> domainType) {
|
|
|
+ T t = strParseT(ObjStr, domainType);
|
|
|
+ Mono<T> insert = template.insert(t);
|
|
|
+ return RStatus.success(insert);
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T> Mono<RStatus<T>> update( String ObjStr,Class<T> domainType) {
|
|
|
+ T t = strParseT(ObjStr, domainType);
|
|
|
+ Mono<T> insert = template.update(t);
|
|
|
+ return RStatus.success(insert);
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T> String tToString(T t) {
|
|
|
+ ObjectMapper objectMapper = new ObjectMapper();
|
|
|
+ objectMapper.registerModule(new JavaTimeModule());
|
|
|
+ objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
|
|
|
+ try {
|
|
|
+ // 将 record 转换为 JSON 字符串
|
|
|
+ return objectMapper.writeValueAsString(t);
|
|
|
+ } catch (JsonProcessingException e) {
|
|
|
+ throw new BaseException(" 将对象转换为JSON字符串异常::"+e.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ @Override
|
|
|
+ public <T> T strParseT(String str,Class<T> domainType) {
|
|
|
+ // 创建 ObjectMapper 实例并配置
|
|
|
+ ObjectMapper objectMapper = new ObjectMapper();
|
|
|
+ objectMapper.registerModule(new JavaTimeModule());
|
|
|
+ SimpleModule module = new SimpleModule();
|
|
|
+ module.addDeserializer(ZonedDateTime.class, new CustomZonedDateTimeDeserializer());
|
|
|
+ objectMapper.registerModule(module);
|
|
|
+ try {
|
|
|
+ // 将 JSON 字符串转换为 record
|
|
|
+ return objectMapper.readValue(str,domainType);
|
|
|
+ } catch (JsonProcessingException e) {
|
|
|
+ throw new BaseException(" 将JSON字符串转换异常:::"+e.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+ public static class CustomZonedDateTimeDeserializer extends JsonDeserializer<ZonedDateTime> {
|
|
|
+ private static final DateTimeFormatter FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
|
|
+ @Override
|
|
|
+ public ZonedDateTime deserialize(JsonParser p, DeserializationContext ctxt)
|
|
|
+ throws IOException, JsonProcessingException {
|
|
|
+ String date = p.getText();
|
|
|
+ LocalDateTime localDateTime = LocalDateTime.parse(date, FORMATTER);
|
|
|
+ // 使用系统默认时区,可以根据需要修改
|
|
|
+ return ZonedDateTime.of(localDateTime, ZoneId.systemDefault());
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+}
|