Skip to content

Y
yaoai-video
Commit a8a18380 authored by yaoke.yk's avatar yaoke.yk
Browse files
Options

分镜页面生成视频转tos自己桶

parent 27725d8e
Show whitespace changes
Inline Side-by-side
Showing with 551 additions and 90 deletions
design-docs/视频任务持久化与轮询重构.md 0 → 100644
View file @ a8a18380
# 视频任务持久化与轮询重构
# 视频任务持久化与轮询重构
> 日期：2026-04-27
> 涉及模块：`yaoai-pipeline`、`yaoai-domain`、`yaoai-bootstrap`
> 关联文档：[08-异步任务与AI工作流](./08-异步任务与AI工作流.md)
## 1. 背景
线上反馈两类视频问题：
1. **历史视频链接过期 403**
```
Code: AccessDenied
Message: Request has expired
ExpiresSeconds: 86400 // 24h
```
2. **复杂分镜偶发「视频生成超时（超过 10 分钟）」**，但 Ark 控制台查到任务实际仍在跑/已成功
两个问题共用一段代码：`VideoTaskAsyncProcessor.pollUntilDone`。
## 2. 根因分析
### 2.1 链接过期
`Seedance` 接口返回的 `videoUrl` 指向**火山 Ark 自家** TOS 桶（私有），它给我们的是预签名 URL，TTL 24h。
旧代码把这个临时 URL 直接持久化到 `ai_tasks.result_video_url`：
```java
// 旧实现
done.setResultVideoUrl(result.getVideoUrl()); // ← Ark 临时链接
```
24h 后用户访问就必然 `AccessDenied`。
### 2.2 误判超时
旧轮询实现：
```java
@Async
processTextToVideo(...) {
submit to Ark
pollUntilDone(taskId, externalId); // ← 阻塞 10 分钟
}
private void pollUntilDone(...) {
for (i = 0; i < 120; i++) { // 120 × 5s = 10min
Thread.sleep(5_000);
query Ark;
if (terminal) return;
}
mark as "视频生成超时（超过 10 分钟）"; // ← 硬判失败
}
```
问题：
- **占用 @Async 线程 10+ 分钟**，并发任务一多线程池就堵
- **服务重启会丢任务**：内存里的轮询循环消失，但 DB 里的状态仍是 `submitted`，永远没人继续处理
- **10 分钟阈值太紧**：复杂分镜（多参考图 + 配音 + 多镜头描述）经常 12–15 分钟才出，被误标失败
## 3. 新架构
### 3.1 流程对比
**改造前（in-thread polling）**
```
HTTP 请求 → @Async 线程
├─ 提交 Ark
├─ 阻塞循环查询 Ark （持续 10 分钟）
├─ 拿到 URL，原样写库 ❌（Ark 临时 URL）
└─ 超时 → 硬标 failed ❌
```
**改造后（scheduled scanner + 转存）**
```
HTTP 请求 → @Async 线程（短命）
├─ 提交 Ark
├─ DB: status=submitted, external_task_id=xxx
└─ 立即结束 ✅
@Scheduled (每 10s)：
VideoTaskPoller
├─ 扫 ai_tasks WHERE status IN ('submitted','running')
├─ for each：
│ ├─ 若 created_at 超 30min 仍非终态 → 标 failed（硬兜底）
│ └─ 否则 → VideoTaskCompletionService.pollOnce(task)
│
└─ pollOnce()
├─ 查 Ark 状态
├─ succeeded → persistVideoToTos() → 存 publicUrl
├─ failed → 写 error_message
└─ running → 同步状态
persistVideoToTos()：
├─ HTTP 下载 Ark 临时 URL（趁 24h 内）
├─ 上传到我们 public-read bucket
│ key = {tenantId}/{projectId}/video/{taskId}.mp4
└─ 返回 tosService.publicUrl(key) ✅ 永不过期
```
### 3.2 组件职责
| 组件 | 职责 |
|---|---|
| `VideoTaskAsyncProcessor` | 提交 Ark + 写初始状态，**无轮询** |
| `VideoTaskPoller` | `@Scheduled` 调度器：扫描、超时兜底、批量分发 |
| `VideoTaskCompletionService` | 单条任务的查询/持久化逻辑（可被未来管理后台补救接口复用） |
| `AiTaskMapper.findPollableVideoTasks` | 拉所有 `submitted/running` 且有 externalTaskId 的任务 |
## 4. 关键改动文件
| 文件 | 类型 | 说明 |
|---|---|---|
| `yaoai-bootstrap/.../YaoAiApplication.java` | 改 | 加 `@EnableScheduling` |
| `yaoai-domain/.../AiTaskMapper.java` | 改 | 加 `findPollableVideoTasks(int limit)` |
| `yaoai-pipeline/.../async/VideoTaskAsyncProcessor.java` | 改（精简） | 删除 `pollUntilDone` + `persistVideoToTos`，提交后立即返回 |
| `yaoai-pipeline/.../async/VideoTaskCompletionService.java` | 新增 | `pollOnce(task)` + `persistVideoToTos(...)` |
| `yaoai-pipeline/.../async/VideoTaskPoller.java` | 新增 | `@Scheduled` 调度器 |
## 5. 配置项
| Key | 默认值 | 说明 |
|---|---|---|
| `yaoai.video.poll-interval-ms` | `10000` | `VideoTaskPoller` 轮询间隔 |
| `BATCH_LIMIT`（常量） | `50` | 单轮最多处理任务数，避免短时打爆 Ark |
| `HARD_TIMEOUT`（常量） | `30 min` | 任务硬超时阈值 |
> 常量后续可按需要外化到配置。
## 6. TOS Key 命名约定
```
{tenantId}/{projectId}/video/{taskId}.mp4
```
与 `TosService.buildKey` 的多租户隔离规范保持一致。
> 配套前提：bucket 已设为 `public-read`，且 `TosServiceImpl.upload` 调用时设了 `ACLType.ACL_PUBLIC_READ`。
## 7. 边界与已知限制
### 7.1 已存量过期 URL 救不回来
2026-04-27 之前完成的视频任务仍存的是 Ark 临时链接。Ark 那边的源文件可能已被清理（24h+），无法补救，需要让用户**重新生成**。
### 7.2 转存失败的兜底策略
`persistVideoToTos` 失败时（网络异常、TOS 写入失败），不抛错阻塞任务完成 —— 退而求其次写回 Ark 原始 URL（24h 内仍可用），打 ERROR 日志。
### 7.3 单实例轮询
当前 `@Scheduled` 在所有应用实例上都会执行。多实例部署时会有重复查询 Ark 的浪费（DB 写入是幂等的所以不会出错）。如未来扩到 K8s 多 Pod，应改用：
- ShedLock 之类的分布式锁
- 或者把扫描器单独放在 `yaoai-task` 模块只跑一份
### 7.4 Ark API 限流
单轮处理 50 条任务 × 一次 API 调用，10s 一轮 → 5 QPS。Ark 文档限额内（目前 60 RPM）。如果同时跑的视频任务超过 50，会有积压（下一轮再处理）。
## 8. 后续可选优化
| 项 | 优先级 | 说明 |
|---|---|---|
| 管理后台「重新拉取」补救接口 | 中 | 按 task id 重查 Ark；放在 `yaoai-admin` + `yaoai-admin-web` |
| `result_tos_key` 落库 | 低 | 当前只存 publicUrl 字符串；存 key 可解耦 bucket 迁移 |
| 多实例部署的分布式锁 | 低 | 见 7.3 |
| 视频生成耗时指标上报 | 低 | 用于评估 30 分钟阈值是否合理 |
## 9. 测试要点
- [ ] 提交一个新视频任务 → 5 分钟内完成 → 链接是 `aivideo-2026041417.tos-cn-beijing.volces.com/...` 而非 `tos-1az-front-azc...`
- [ ] 提交后立刻杀进程重启 → 任务能被 Poller picked up 并最终完成
- [ ] 模拟 Ark 长时间不返回 → 30 分钟后任务被标 failed
- [ ] Ark 返回 succeeded 但视频下载失败 → 任务仍标 succeeded，URL 是 Ark 原始 URL（兜底）
- [ ] 24h 后访问新生成视频的 URL → 仍可正常访问（验证 publicUrl 永久性）
doc/html/src/app/components/Layout.tsx
View file @ a8a18380
......@@ -128,7 +128,7 @@ export function Layout() {
</div>
{/* Product Name and Version */}
<div className="flex items-baseline gap-1">
<span className="text-sm font-semibold text-foreground">DramaStudio</span>
<span className="text-sm font-semibold text-foreground">Dramix</span>
<span className="text-xs text-muted-foreground">V1</span>
</div>
</Link>
......@@ -269,7 +269,7 @@ export function Layout() {
</div>
{/* Product Name and Version */}
<div className="flex items-baseline gap-1">
<span className="text-sm font-semibold text-foreground">DramaStudio</span>
<span className="text-sm font-semibold text-foreground">Dramix</span>
<span className="text-xs text-muted-foreground">V1</span>
</div>
</Link>
......
doc/html/src/app/pages/LoginPage.tsx
View file @ a8a18380
......@@ -19,7 +19,7 @@ export function LoginPage() {
<div className="min-h-screen flex items-center justify-center bg-background px-4">
<Card className="w-full max-w-sm">
<CardHeader className="text-center">
<CardTitle className="text-2xl">YaoAI Comic Studio</CardTitle>
<CardTitle className="text-2xl">Dramix</CardTitle>
<CardDescription>登录您的账号</CardDescription>
</CardHeader>
<CardContent>
......
doc/html/src/app/pages/RegisterPage.tsx
View file @ a8a18380
......@@ -30,7 +30,7 @@ export function RegisterPage() {
<Card className="w-full max-w-sm">
<CardHeader className="text-center">
<CardTitle className="text-2xl">创建账号</CardTitle>
<CardDescription>免费开始使用 YaoAI Comic Studio</CardDescription>
<CardDescription>免费开始使用 Dramix</CardDescription>
</CardHeader>
<CardContent>
<form onSubmit={handleSubmit} className="space-y-4">
......
doc/html/src/app/pages/StoryboardWorkspace.tsx
View file @ a8a18380
import { useState, useEffect, useMemo } from "react";
import { useState, useEffect, useMemo, useRef } from "react";
import { useParams, useNavigate } from "react-router";
import {
Plus, Wand2, Play, Trash2,
......@@ -78,7 +78,7 @@ export function StoryboardWorkspace() {
const eid = activeEpisodeId;
const { data: storyboards = [], isLoading } = useStoryboards(pid, eid);
const { data: videoTasks = [] } = useVideoTasks(pid);
const { data: videoTasks = [], isFetched: videoTasksFetched } = useVideoTasks(pid);
const { data: characters = [] } = useCharacters(pid);
const { data: scenes = [] } = useScenes(pid);
const { data: assemblyTask } = useAssemblyTask(pid, eid);
......@@ -144,6 +144,10 @@ export function StoryboardWorkspace() {
return "bg-orange-400 animate-pulse";
};
// 编辑器状态只在切换分镜卡片时从最近任务回填一次。
// 之后输入框完全归用户掌控，不会被 videoTasks 刷新（如点击生成后 mutation 失效）覆盖。
const lastSyncedExpandedIdRef = useRef<string | null>(null);
// Reset on episode change
useEffect(() => {
setExpandedId(null);
......@@ -153,6 +157,7 @@ export function StoryboardWorkspace() {
setPropKeys([]);
setShortDescription("");
setActiveVideoSbId(null);
lastSyncedExpandedIdRef.current = null;
}, [activeEpisodeId]);
// Auto-expand first storyboard on load
......@@ -163,9 +168,12 @@ export function StoryboardWorkspace() {
}
}, [storyboards.length]);
// Sync editor state when expanded card changes
useEffect(() => {
if (!expanded) return;
if (!videoTasksFetched) return; // 等首次拉到任务列表再回填，避免空数据回填后被锁
if (lastSyncedExpandedIdRef.current === expandedId) return;
lastSyncedExpandedIdRef.current = expandedId;
setShortDescription(expanded.shortDescription ?? "");
const lastTask = getTaskForSb(expanded);
const initialChars = lastTask?.characterImageKeys
......@@ -177,7 +185,7 @@ export function StoryboardWorkspace() {
if (typeof lastTask?.videoDuration === "number") setSelectedDuration(lastTask.videoDuration);
if (lastTask?.videoRatio) setSelectedRatio(lastTask.videoRatio);
if (typeof lastTask?.generateAudio === "boolean") setAudioOn(lastTask.generateAudio);
}, [expandedId, videoTasks.length]);
}, [expandedId, videoTasksFetched]);
const handleEpisodeSelect = (ep: Episode) => {
setActiveEpisodeId(ep.id);
......@@ -318,7 +326,7 @@ export function StoryboardWorkspace() {
width: 50,
height: 56,
borderRadius: 16,
fontSize: 22,
fontSize: 20,
backgroundColor: active ? "#1c64ff" : "#ffffff",
color: active ? "#ffffff" : "#8b95a7",
boxShadow: active ? "0 12px 24px rgba(28, 100, 255, 0.28)" : "0 1px 2px rgba(15, 23, 42, 0.05)",
......
yaoai-admin-web/index.html
View file @ a8a18380
......@@ -3,7 +3,7 @@
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0" />
<title>YaoAI Admin</title>
<title>Dramix Admin</title>
</head>
<body>
<div id="app"></div>
......
yaoai-admin-web/src/App.vue
View file @ a8a18380
......@@ -26,9 +26,9 @@ async function logout() {
<div v-else class="admin-shell">
<aside class="side-rail" :class="{ collapsed }">
<div class="brand-mark">
<div class="brand-sigil">Y</div>
<div class="brand-sigil">D</div>
<div class="brand-copy">
<b>YaoAI</b>
<b>Dramix</b>
<span>Operations</span>
</div>
</div>
......
yaoai-admin-web/src/styles.css
View file @ a8a18380
......@@ -196,36 +196,134 @@ body {
.login-scene {
min-height: 100vh;
display: grid;
grid-template-columns: 1fr 420px;
grid-template-columns: 1fr 460px;
background:
linear-gradient(135deg, rgba(31, 122, 90, 0.16), transparent 45%),
radial-gradient(1100px 600px at -10% -10%, rgba(31, 122, 90, 0.18), transparent 60%),
radial-gradient(700px 480px at 110% 110%, rgba(180, 95, 36, 0.10), transparent 55%),
var(--paper);
}
.login-brief {
position: relative;
display: flex;
flex-direction: column;
justify-content: center;
padding: 8vw;
padding: 7vw 6vw;
gap: 28px;
}
.brief-brand {
display: flex;
align-items: center;
gap: 12px;
}
.brief-sigil {
width: 40px;
height: 40px;
display: grid;
place-items: center;
background: #d7a94b;
color: #18231d;
border-radius: 8px;
font-weight: 800;
font-size: 18px;
}
.brief-copy {
display: grid;
line-height: 1.15;
}
.brief-copy b {
font-size: 18px;
color: var(--ink);
letter-spacing: 0.5px;
}
.brief-copy span {
color: var(--muted);
font-size: 12px;
}
.login-brief h1 {
max-width: 760px;
max-width: 720px;
margin: 0;
font-family: Georgia, "Times New Roman", serif;
font-size: 58px;
letter-spacing: 0;
font-family: "Source Han Serif SC", "Songti SC", Georgia, serif;
font-size: 46px;
line-height: 1.25;
letter-spacing: 0.5px;
color: var(--ink);
}
.login-brief > p {
max-width: 540px;
margin: 0;
color: var(--muted);
font-size: 15px;
line-height: 1.7;
}
.brief-points {
margin: 8px 0 0;
padding: 0;
list-style: none;
display: grid;
gap: 10px;
}
.brief-points li {
display: flex;
align-items: center;
gap: 10px;
color: var(--ink);
font-size: 14px;
}
.brief-points i {
width: 6px;
height: 6px;
border-radius: 50%;
background: var(--accent);
display: inline-block;
}
.login-card {
margin: auto 42px;
padding: 28px;
align-self: center;
margin: 0 42px;
padding: 36px 32px 28px;
background: var(--panel);
border: 1px solid var(--line);
border-radius: 8px;
border-radius: 12px;
box-shadow: var(--shadow);
}
.login-card-head {
margin-bottom: 22px;
}
.login-card-head h2 {
margin: 0 0 6px;
font-size: 22px;
color: var(--ink);
letter-spacing: 0.5px;
}
.login-card-head p {
margin: 0;
color: var(--muted);
font-size: 13px;
}
.login-card-foot {
margin-top: 18px;
padding-top: 14px;
border-top: 1px dashed var(--line);
color: var(--muted);
font-size: 12px;
text-align: center;
}
.status-active {
color: var(--accent);
}
......@@ -246,6 +344,19 @@ body {
.login-scene {
grid-template-columns: 1fr;
}
.login-brief {
padding: 48px 32px 16px;
gap: 20px;
}
.login-brief h1 {
font-size: 32px;
}
.login-card {
margin: 0 32px 48px;
}
}
@media (max-width: 720px) {
......
yaoai-admin-web/src/views/LoginView.vue
View file @ a8a18380
......@@ -23,20 +23,38 @@ async function submit() {
<template>
<main class="login-scene">
<section class="login-brief">
<h1>Platform operations, measured without guesswork.</h1>
<p>集中管理租户、积分、模型成本和审计轨迹。</p>
<div class="brief-brand">
<div class="brief-sigil">D</div>
<div class="brief-copy">
<b>Dramix</b>
<span>Operations Console</span>
</div>
</div>
<h1>把每一次运营动作<br />都留在可追溯的轨道上。</h1>
<p>租户、积分、模型成本与审计日志，集中在一个克制的工作台。</p>
<ul class="brief-points">
<li><i></i>跨租户查询、套餐与状态调整</li>
<li><i></i>积分流水与模型成本对账</li>
<li><i></i>运营操作全量审计与回溯</li>
</ul>
</section>
<section class="login-card">
<h2>运营后台</h2>
<header class="login-card-head">
<h2>欢迎回来</h2>
<p>请使用运营账号登录 Dramix 后台</p>
</header>
<el-form label-position="top" @submit.prevent="submit">
<el-form-item label="账号">
<el-input v-model="form.username" :prefix-icon="User" size="large" />
<el-input v-model="form.username" :prefix-icon="User" size="large" placeholder="请输入运营账号" />
</el-form-item>
<el-form-item label="密码">
<el-input v-model="form.password" :prefix-icon="Lock" show-password size="large" @keyup.enter="submit" />
<el-input v-model="form.password" :prefix-icon="Lock" show-password size="large" placeholder="请输入登录密码" @keyup.enter="submit" />
</el-form-item>
<el-button type="primary" size="large" :loading="loading" style="width: 100%" @click="submit">登录</el-button>
</el-form>
<footer class="login-card-foot">
登录即代表你已阅读并同意《运营人员使用规范》。
</footer>
</section>
</main>
</template>
yaoai-comic-studio/yaoai-ai-providers/src/main/java/com/yaoai/ai/providers/service/impl/SeedanceServiceImpl.java
View file @ a8a18380
......@@ -9,9 +9,11 @@ import com.yaoai.common.exception.BizException;
import com.yaoai.common.exception.ErrorCode;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestClient;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
......@@ -26,10 +28,14 @@ public class SeedanceServiceImpl implements SeedanceService {
public SeedanceServiceImpl(ArkProperties properties) {
this.properties = properties;
SimpleClientHttpRequestFactory rf = new SimpleClientHttpRequestFactory();
rf.setConnectTimeout(Duration.ofSeconds(10));
rf.setReadTimeout(Duration.ofMinutes(2));
this.restClient = RestClient.builder()
.baseUrl(properties.getBaseUrl())
.defaultHeader("Authorization", "Bearer " + properties.getApiKey())
.defaultHeader("Content-Type", "application/json")
.requestFactory(rf)
.build();
}
......
yaoai-comic-studio/yaoai-ai-providers/src/main/java/com/yaoai/ai/providers/service/impl/SeedreamServiceImpl.java
View file @ a8a18380
......@@ -6,9 +6,11 @@ import com.yaoai.common.exception.BizException;
import com.yaoai.common.exception.ErrorCode;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestClient;
import java.time.Duration;
import java.util.List;
import java.util.Map;
......@@ -21,10 +23,14 @@ public class SeedreamServiceImpl implements SeedreamService {
public SeedreamServiceImpl(ArkProperties properties) {
this.properties = properties;
SimpleClientHttpRequestFactory rf = new SimpleClientHttpRequestFactory();
rf.setConnectTimeout(Duration.ofSeconds(10));
rf.setReadTimeout(Duration.ofMinutes(2));
this.restClient = RestClient.builder()
.baseUrl(properties.getBaseUrl())
.defaultHeader("Authorization", "Bearer " + properties.getApiKey())
.defaultHeader("Content-Type", "application/json")
.requestFactory(rf)
.build();
}
......
yaoai-comic-studio/yaoai-bootstrap/src/main/java/com/yaoai/YaoAiApplication.java
View file @ a8a18380
......@@ -7,6 +7,7 @@ import org.springframework.cache.annotation.EnableCaching;
@SpringBootApplication
@EnableCaching
@org.springframework.scheduling.annotation.EnableAsync
@org.springframework.scheduling.annotation.EnableScheduling
public class YaoAiApplication {
public static void main(String[] args) {
......
yaoai-comic-studio/yaoai-domain/src/main/java/com/yaoai/domain/mapper/AiTaskMapper.java
View file @ a8a18380
......@@ -26,4 +26,14 @@ public interface AiTaskMapper extends BaseMapper<AiTask> {
ORDER BY COALESCE(s.sequence_num, 9999), t.created_at
""")
List<AiTask> findSucceededByEpisodeOrdered(Long episodeId, Long tenantId);
/** 待轮询的视频任务：已提交 Ark 且未到终态。limit 控制单轮处理上限，避免单次扫描挤占线程。 */
@Select("""
SELECT * FROM ai_tasks
WHERE status IN ('submitted','running')
AND external_task_id IS NOT NULL
ORDER BY created_at ASC
LIMIT #{limit}
""")
List<AiTask> findPollableVideoTasks(int limit);
}
yaoai-comic-studio/yaoai-pipeline/src/main/java/com/yaoai/pipeline/async/VideoTaskAsyncProcessor.java
View file @ a8a18380
package com.yaoai.pipeline.async;
import com.yaoai.ai.providers.model.VideoTaskResult;
import com.yaoai.ai.providers.service.SeedanceService;
import com.yaoai.billing.dto.BillingChargeRequest;
import com.yaoai.billing.service.BillingService;
......@@ -34,6 +33,8 @@ public class VideoTaskAsyncProcessor {
* - preImageKeys 为空 → 调用 Seedream 文生图，再提交 Seedance
* - preImageKeys 非空 → 直接用这些 TOS key 作为 @图1、@图2... 参考图提交 Seedance
*
* 提交 Ark 后立即结束，不阻塞 @Async 线程；任务完成由 {@link VideoTaskPoller} 定时扫描处理。
*
* @param preImageKeys 按 @图1/@图2... 顺序排列的 TOS key 列表，空列表则走文生图
* @param durationSeconds 视频时长（秒），<=0 则使用默认值 5
* @param ratio 视频宽高比，如 "16:9" / "9:16" / "1:1"，空/非法值由下游兜底
......@@ -96,7 +97,7 @@ public class VideoTaskAsyncProcessor {
log.info("Seedance task submitted: taskId={}, externalId={}, images={}, duration={}s",
taskId, externalTaskId, presignedUrls.size(), duration);
// 更新任务为 submitted
// 更新任务为 submitted；后续状态查询交给 VideoTaskPoller
AiTask update = new AiTask();
update.setId(taskId);
update.setStatus("submitted");
......@@ -105,10 +106,7 @@ public class VideoTaskAsyncProcessor {
aiTaskMapper.updateById(update);
billingService.charge(videoCharge);
// 轮询 Ark 直到任务完成（最多等 10 分钟，每 5 秒查一次）
pollUntilDone(taskId, externalTaskId);
log.info("Async text-to-video done: taskId={}", taskId);
log.info("Async text-to-video submitted: taskId={}, externalId={}", taskId, externalTaskId);
} catch (Exception e) {
log.error("Async text-to-video failed: taskId={}", taskId, e);
......@@ -119,58 +117,4 @@ public class VideoTaskAsyncProcessor {
aiTaskMapper.updateById(failure);
}
}
/**
* 轮询 Ark 任务状态，直到 succeeded/failed 或超时（10 分钟）。
* 结果写回 DB，前端轮询 listByProject 即可自动感知。
*/
private void pollUntilDone(Long taskId, String externalTaskId) {
int maxAttempts = 120; // 120 * 5s = 10 min
for (int i = 0; i < maxAttempts; i++) {
try {
Thread.sleep(5_000);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
log.warn("Video poll interrupted: taskId={}", taskId);
return;
}
try {
VideoTaskResult result = seedanceService.getTaskStatus(externalTaskId);
String status = result.getStatus();
log.debug("Polling Ark: taskId={}, externalId={}, status={}", taskId, externalTaskId, status);
if ("succeeded".equals(status) || "failed".equals(status)) {
AiTask done = new AiTask();
done.setId(taskId);
done.setStatus(status);
if (result.getVideoUrl() != null) {
done.setResultVideoUrl(result.getVideoUrl());
}
if (result.getErrorMessage() != null) {
done.setErrorMessage(result.getErrorMessage());
}
aiTaskMapper.updateById(done);
log.info("Video task completed: taskId={}, status={}, url={}",
taskId, status, result.getVideoUrl());
return;
}
// 仍在执行中 — 更新为 running 便于前端区分 submitted/running
if ("running".equals(status)) {
AiTask running = new AiTask();
running.setId(taskId);
running.setStatus("running");
aiTaskMapper.updateById(running);
}
} catch (Exception e) {
log.warn("Ark poll error (will retry): taskId={}, error={}", taskId, e.getMessage());
}
}
// 超时
AiTask timeout = new AiTask();
timeout.setId(taskId);
timeout.setStatus("failed");
timeout.setErrorMessage("视频生成超时（超过 10 分钟）");
aiTaskMapper.updateById(timeout);
log.error("Video task timed out: taskId={}", taskId);
}
}
yaoai-comic-studio/yaoai-pipeline/src/main/java/com/yaoai/pipeline/async/VideoTaskCompletionService.java 0 → 100644
View file @ a8a18380
package com.yaoai.pipeline.async;
import com.yaoai.ai.providers.model.VideoTaskResult;
import com.yaoai.ai.providers.service.SeedanceService;
import com.yaoai.domain.entity.AiTask;
import com.yaoai.domain.mapper.AiTaskMapper;
import com.yaoai.storage.service.TosService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.io.ByteArrayInputStream;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
/**
* 单条视频任务的查询与完成处理：查 Ark → 更新 DB → 转存视频到自家 TOS。
* 给 {@link VideoTaskPoller} 调用，未来若加管理后台补救接口也可复用。
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class VideoTaskCompletionService {
private final AiTaskMapper aiTaskMapper;
private final SeedanceService seedanceService;
private final TosService tosService;
private final HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(10))
.build();
/**
* 单次轮询一个任务：查 Ark 状态并按状态更新 DB；succeeded 时把视频转存到我们桶。
* 返回此任务在本轮处理后是否进入终态（succeeded/failed）。
*/
public boolean pollOnce(AiTask task) {
Long taskId = task.getId();
String externalTaskId = task.getExternalTaskId();
try {
VideoTaskResult result = seedanceService.getTaskStatus(externalTaskId);
String status = result.getStatus();
log.debug("Polling Ark: taskId={}, externalId={}, status={}", taskId, externalTaskId, status);
if ("succeeded".equals(status) || "failed".equals(status)) {
AiTask done = new AiTask();
done.setId(taskId);
done.setStatus(status);
if (result.getVideoUrl() != null) {
String persistentUrl = persistVideoToTos(
task.getTenantId(), task.getProjectId(), taskId, result.getVideoUrl());
done.setResultVideoUrl(persistentUrl);
}
if (result.getErrorMessage() != null) {
done.setErrorMessage(result.getErrorMessage());
}
aiTaskMapper.updateById(done);
log.info("Video task completed: taskId={}, status={}, url={}",
taskId, status, done.getResultVideoUrl());
return true;
}
// running 时同步状态，便于前端区分 submitted/running
if ("running".equals(status) && !"running".equals(task.getStatus())) {
AiTask running = new AiTask();
running.setId(taskId);
running.setStatus("running");
aiTaskMapper.updateById(running);
}
return false;
} catch (Exception e) {
log.warn("Ark poll error (will retry): taskId={}, error={}", taskId, e.getMessage());
return false;
}
}
/**
* 把 Ark 临时视频 URL 下载下来上传到我们 public-read TOS bucket，返回永不过期的公开 URL。
* 失败时退回原始 URL（24h 内仍可用）+ 错误日志，不阻塞任务完成流程。
*/
public String persistVideoToTos(Long tenantId, Long projectId, Long taskId, String arkVideoUrl) {
try {
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(arkVideoUrl))
.timeout(Duration.ofMinutes(2))
.GET()
.build();
HttpResponse<byte[]> response = httpClient.send(request, HttpResponse.BodyHandlers.ofByteArray());
if (response.statusCode() != 200) {
log.warn("Download Ark video failed (status={}), keep raw URL: taskId={}", response.statusCode(), taskId);
return arkVideoUrl;
}
byte[] bytes = response.body();
String key = String.format("%d/%d/video/%d.mp4", tenantId, projectId, taskId);
try (var is = new ByteArrayInputStream(bytes)) {
tosService.upload(key, is, bytes.length, "video/mp4");
}
String publicUrl = tosService.publicUrl(key);
log.info("Video persisted to TOS: taskId={}, key={}, size={}", taskId, key, bytes.length);
return publicUrl;
} catch (Exception e) {
log.error("Persist Ark video to TOS failed, fallback to raw URL: taskId={}", taskId, e);
return arkVideoUrl;
}
}
}
yaoai-comic-studio/yaoai-pipeline/src/main/java/com/yaoai/pipeline/async/VideoTaskPoller.java 0 → 100644
View file @ a8a18380
package com.yaoai.pipeline.async;
import com.yaoai.domain.entity.AiTask;
import com.yaoai.domain.mapper.AiTaskMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
/**
* 定时扫描所有未到终态的视频任务并查询 Ark 状态。
* 取代之前 {@link VideoTaskAsyncProcessor} 中阻塞 10 分钟的 in-thread 轮询：
* - 不再占用 @Async 线程池
* - 服务重启后自动 picked up 未完成的任务
* - 硬性超时改为基于 created_at 的 30 分钟阈值
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class VideoTaskPoller {
/** 单轮处理上限，防止 Ark 接口短时被打爆 */
private static final int BATCH_LIMIT = 50;
/** 任务从创建到必须终态的最大允许时长 */
private static final Duration HARD_TIMEOUT = Duration.ofMinutes(30);
private final AiTaskMapper aiTaskMapper;
private final VideoTaskCompletionService completionService;
@Scheduled(fixedDelayString = "${yaoai.video.poll-interval-ms:10000}", initialDelay = 10_000)
public void poll() {
List<AiTask> pending;
try {
pending = aiTaskMapper.findPollableVideoTasks(BATCH_LIMIT);
} catch (Exception e) {
log.error("Load pollable video tasks failed", e);
return;
}
if (pending.isEmpty()) return;
log.debug("Polling {} pending video task(s)", pending.size());
LocalDateTime deadline = LocalDateTime.now().minus(HARD_TIMEOUT);
for (AiTask task : pending) {
// 硬超时兜底：避免 Ark 永不返回时任务永远停留在 submitted/running
if (task.getCreatedAt() != null && task.getCreatedAt().isBefore(deadline)) {
AiTask timeout = new AiTask();
timeout.setId(task.getId());
timeout.setStatus("failed");
timeout.setErrorMessage("视频生成超时（超过 " + HARD_TIMEOUT.toMinutes() + " 分钟）");
aiTaskMapper.updateById(timeout);
log.warn("Video task hard-timeout: taskId={}, createdAt={}", task.getId(), task.getCreatedAt());
continue;
}
completionService.pollOnce(task);
}
}
}
yaoai-comic-studio/yaoai-pipeline/src/main/java/com/yaoai/pipeline/service/impl/AssetGenPipelineServiceImpl.java
View file @ a8a18380
......@@ -33,6 +33,7 @@ import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
......@@ -86,7 +87,9 @@ public class AssetGenPipelineServiceImpl implements AssetGenPipelineService {
private final ProjectMapper projectMapper;
private final ObjectMapper objectMapper;
private final BillingService billingService;
private final HttpClient httpClient = HttpClient.newHttpClient();
private final HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(10))
.build();
@Override
public List<Character> extractCharacters(Long projectId, Long tenantId) {
......@@ -724,7 +727,11 @@ public class AssetGenPipelineServiceImpl implements AssetGenPipelineService {
}
private byte[] downloadBytes(String url) throws Exception {
HttpRequest request = HttpRequest.newBuilder().uri(URI.create(url)).GET().build();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(url))
.timeout(Duration.ofMinutes(2))
.GET()
.build();
HttpResponse<byte[]> response = httpClient.send(request, HttpResponse.BodyHandlers.ofByteArray());
if (response.statusCode() != 200) {
throw new BizException(ErrorCode.INTERNAL_ERROR, "下载图片失败 HTTP " + response.statusCode());
......
yaoai-comic-studio/yaoai-pipeline/src/main/java/com/yaoai/pipeline/service/impl/ImageGenPipelineServiceImpl.java
View file @ a8a18380
......@@ -17,6 +17,7 @@ import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.time.Duration;
import java.util.Map;
@Slf4j
......@@ -27,7 +28,9 @@ public class ImageGenPipelineServiceImpl implements ImageGenPipelineService {
private final SeedreamService seedreamService;
private final TosService tosService;
private final BillingService billingService;
private final HttpClient httpClient = HttpClient.newHttpClient();
private final HttpClient httpClient = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(10))
.build();
@Override
public String generateAndStore(Long tenantId, Long projectId, String prompt) {
......@@ -80,6 +83,7 @@ public class ImageGenPipelineServiceImpl implements ImageGenPipelineService {
try {
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create(url))
.timeout(Duration.ofMinutes(2))
.GET()
.build();
HttpResponse<byte[]> response = httpClient.send(request, HttpResponse.BodyHandlers.ofByteArray());
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment