定时同步luma任务记录

luma/src/main/java/com/yhlxj/service/midjourney/LumaService.java

@@ -34,4 +34,11 @@ public interface LumaService {
     LumaUserConversation getVideoConversationByTid(LumaUser user, Long rid);
 
     String updateCover(Long id, Long userId) throws IOException;
+
+    void syncConversation();
+
+    /**
+     * 任务失败回滚次数
+     */
+    void recoverUserLimit(Long userId);
 }

luma/src/main/java/com/yhlxj/service/midjourney/impl/LumaServiceImpl.java

@@ -1,6 +1,8 @@
 package com.yhlxj.service.midjourney.impl;
 
 import cn.hutool.core.bean.BeanUtil;
+import cn.hutool.core.date.DateTime;
+import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.map.MapUtil;
 import cn.hutool.core.util.StrUtil;
 import cn.hutool.http.HttpRequest;
@@ -8,7 +10,6 @@ import cn.hutool.http.HttpResponse;
 import cn.hutool.http.HttpUtil;
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
-import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
 import com.cyksj.common.exception.BusinessRuntimeException;
 import com.cyksj.common.task.GlobalThreadPoolTaskExecutor;
@@ -179,34 +180,7 @@ public class LumaServiceImpl implements LumaService {
         if (lumaUserConversation == null) {
             throw BusinessRuntimeException.getInstance("视频不存在");
         }
-        Long providerId = lumaUserConversation.getProviderId();
-        LumaUserApiProvider apiProvider = lumaUserApiProviderMapper.selectById(providerId);
-        String url = apiProvider.getDomain() + "/luma/generations/" + lumaUserConversation.getTaskId();
-        HttpRequest request = HttpRequest.get(url);
-        request.header("Authorization", apiProvider.getAuth());
-
-        String body = request.setConnectionTimeout(50000).execute().body();
-        log.info("action body:{}", body);
-        SubmitResult submitResult = Jsons.parseObject(body, SubmitResult.class);
-        if (submitResult.getId() == null) {
-            throw BusinessRuntimeException.getInstance("视频不存在");
-        }
-        if (FINISH_STATUS.contains(submitResult.getState())) {
-            String id = submitResult.getId();
-            if (!FINISH_STATUS.contains(lumaUserConversation.getStatus())) {
-                lumaUserConversation.setStatus(submitResult.getState());
-                SubmitResult.Video video = submitResult.getVideo();
-                lumaUserConversation.setFinishTime(System.currentTimeMillis());
-                //出水印
-                String removeVideoWatermarkUrl = removeVideoWatermark(lumaUserConversation.getProviderId(),lumaUserConversation.getTaskId());
-                lumaUserConversation.setVideoUrl(removeVideoWatermarkUrl);
-                lumaUserConversation.setThumbnail(video.getThumbnail());
-                TASK_EXECUTOR.execute(() -> {
-                    //同步
-                    sync(lumaUserConversation);
-                });
-            }
-        }
+        SubmitResult submitResult = syncConversation(lumaUserConversation);
         return submitResult;
     }
 
@@ -405,6 +379,39 @@ public class LumaServiceImpl implements LumaService {
         return null;
     }
 
+    @Override
+    public void syncConversation() {
+        List<LumaUserConversation> pending = conversationMapper.selectList(Wrappers.lambdaQuery(LumaUserConversation.class)
+                .eq(LumaUserConversation::getStatus, status.get(0))
+                .orderByAsc(LumaUserConversation::getId)
+                .last("limit 50"));
+        log.info("正在定时同步更新任务状态--->size:{}", pending.size());
+        DateTime limitTime = DateUtil.offsetMinute(DateTime.now(), -30);
+        pending.forEach(conversation -> {
+            if (conversation.getCreatedTime().compareTo(limitTime) <= 0) {
+                log.info("任务超过过30m，任务taskId:{}直接失败，返还执行次数", conversation.getTaskId());
+                conversation.setStatus(FINISH_STATUS.get(1));
+                conversationMapper.updateById(conversation);
+                recoverUserLimit(conversation.getUserId());
+                return;
+            }
+            try {
+                syncConversation(conversation);
+            } catch (Exception ex) {
+                log.error("定时同步任务执行状态失败：{}", StringUtil.getErrorText(ex));
+            }
+        });
+    }
+
+    /**
+     * 恢复次数
+     */
+    @Override
+    public void recoverUserLimit(Long id){
+        redisService.incr(RedisService.key.LUMA_NUM_LIMIT.getName() + id, 1L);
+        lumaUserMapper.incrNum(id, 1);
+    }
+
     public void sync(LumaUserConversation conversation) {
         if (FINISH_STATUS.contains(conversation.getStatus())) {
             //重复更新数据
@@ -416,7 +423,7 @@ public class LumaServiceImpl implements LumaService {
             LumaUserConversation dbConversation = conversationMapper.selectOne(Wrappers.lambdaQuery(LumaUserConversation.class).eq(LumaUserConversation::getTaskId, conversation.getTaskId())
                     .eq(LumaUserConversation::getUserId, userId).orderByDesc(LumaUserConversation::getId).last("limit 1"));
             if (dbConversation != null) {
-                if ("completed".equals(dbConversation.getStatus())) {
+                if (FINISH_STATUS.contains(dbConversation.getStatus())) {
                     return;
                 }
                 conversation.setProgress("100%");
@@ -437,12 +444,8 @@ public class LumaServiceImpl implements LumaService {
                     }
                 });
                 //失败返还次数
-                if ("failed".equals(conversation.getStatus())){
-                    LambdaUpdateWrapper<LumaUser> wrapper = Wrappers.lambdaUpdate(LumaUser.class)
-                            .eq(LumaUser::getId, userId);
-                    Object num = redisService.incr(RedisService.key.LUMA_NUM_LIMIT.getName() + userId, 1L);
-                    wrapper.set(LumaUser::getNum, num);
-                    lumaUserMapper.update(null, wrapper);
+                if (FINISH_STATUS.get(1).equals(conversation.getStatus())){
+                    recoverUserLimit(conversation.getUserId());
                 }
             }
         }
@@ -520,4 +523,36 @@ public class LumaServiceImpl implements LumaService {
         }
         return null;
     }
+
+    public SubmitResult syncConversation(LumaUserConversation lumaUserConversation) throws Exception {
+        Long providerId = lumaUserConversation.getProviderId();
+        LumaUserApiProvider apiProvider = lumaUserApiProviderMapper.selectById(providerId);
+        String url = apiProvider.getDomain() + "/luma/generations/" + lumaUserConversation.getTaskId();
+        HttpRequest request = HttpRequest.get(url);
+        request.header("Authorization", apiProvider.getAuth());
+
+        String body = request.setConnectionTimeout(50000).execute().body();
+        log.info("action body:{}", body);
+        SubmitResult submitResult = Jsons.parseObject(body, SubmitResult.class);
+        if (submitResult.getId() == null) {
+            throw BusinessRuntimeException.getInstance("视频不存在");
+        }
+        if (FINISH_STATUS.contains(submitResult.getState())) {
+            String id = submitResult.getId();
+            if (!FINISH_STATUS.contains(lumaUserConversation.getStatus())) {
+                lumaUserConversation.setStatus(submitResult.getState());
+                SubmitResult.Video video = submitResult.getVideo();
+                lumaUserConversation.setFinishTime(System.currentTimeMillis());
+                //出水印
+                String removeVideoWatermarkUrl = removeVideoWatermark(lumaUserConversation.getProviderId(),lumaUserConversation.getTaskId());
+                lumaUserConversation.setVideoUrl(removeVideoWatermarkUrl);
+                lumaUserConversation.setThumbnail(video.getThumbnail());
+                TASK_EXECUTOR.execute(() -> {
+                    //同步
+                    sync(lumaUserConversation);
+                });
+            }
+        }
+        return submitResult;
+    }
 }

luma/src/main/java/com/yhlxj/service/task/Scheduler.java

@@ -0,0 +1,23 @@
+package com.yhlxj.service.task;
+
+import com.yhlxj.service.midjourney.LumaService;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+
+@Component
+@Slf4j
+@RequiredArgsConstructor
+public class Scheduler {
+
+    private final LumaService lumaService;
+
+    @Scheduled(cron = "0 0/2 * * * ?")
+    public void syscConversation() {
+        lumaService.syncConversation();
+    }
+
+}

luma/src/main/java/com/yhlxj/web/mirror/LumaController.java

@@ -112,7 +112,7 @@ public class LumaController {
         try {
             conversation = lumaService.submitVideo(user, videoDTO, num);
         } catch (Exception e) {
-            recoverUserLimit(user.getId());
+            lumaService.recoverUserLimit(user.getId());
             log.error("提交video任务异常，{}", StringUtil.getErrorText(e));
             throw BusinessRuntimeException.getInstance(e.getMessage());
         }
@@ -191,7 +191,7 @@ public class LumaController {
             log.info("userId:{}扩展taskId:{}视频", user.getId(), taskId);
             submitResult = lumaService.extendedVideo(user, videoDTO);
         } catch (Exception e) {
-            recoverUserLimit(user.getId());
+            lumaService.recoverUserLimit(user.getId());
             log.error("扩展taskId:{}视频任务异常，{}", taskId, StringUtil.getErrorText(e));
             throw BusinessRuntimeException.getInstance(e.getMessage());
         }
@@ -245,12 +245,4 @@ public class LumaController {
         }
         return num;
     }
-
-    /**
-     * 恢复次数
-     */
-    public void recoverUserLimit(Long id){
-        redisService.incr(RedisService.key.LUMA_NUM_LIMIT.getName() + id, 1L);
-        lumaUserMapper.incrNum(id, 1);
-    }
 }