fix: archive generated videos to S3 immediately
Some checks failed
Lint / build (20) (push) Has been cancelled
Playwright Tests / test (push) Has been cancelled

This commit is contained in:
Your Name 2026-02-20 21:26:38 +08:00
parent 8386755e6d
commit ea2885769c

View File

@ -55,6 +55,7 @@ import type { DBMessage } from "@/lib/db/schema";
import { generatedImage, message as messageTable } from "@/lib/db/schema"; import { generatedImage, message as messageTable } from "@/lib/db/schema";
import { ChatSDKError } from "@/lib/errors"; import { ChatSDKError } from "@/lib/errors";
import { extractImageUrlsFromText } from "@/lib/image-url"; import { extractImageUrlsFromText } from "@/lib/image-url";
import { videoArchiver } from "@/lib/storage/video-archiver";
import { uploadFile } from "@/lib/storage/service"; import { uploadFile } from "@/lib/storage/service";
import type { ChatMessage } from "@/lib/types"; import type { ChatMessage } from "@/lib/types";
import type { AppUsage } from "@/lib/usage"; import type { AppUsage } from "@/lib/usage";
@ -357,6 +358,96 @@ function selectBestVideoSubModelForMode(
})[0]; })[0];
} }
function extractVideoUrlFromText(text: string): string | null {
const normalizedText = String(text || "");
if (!normalizedText) return null;
const htmlMatch = normalizedText.match(/<video[^>]+src="([^"]+)"/i);
if (htmlMatch?.[1]) return htmlMatch[1];
const markdownMatch = normalizedText.match(
/\[[^\]]*]\((https?:\/\/[^)\s]+\.mp4(?:\?[^)\s]*)?)\)/i,
);
if (markdownMatch?.[1]) return markdownMatch[1];
const plainMatch = normalizedText.match(/(https?:\/\/[^\s"'<>]+\.mp4(?:\?[^\s"'<>]+)?)/i);
return plainMatch?.[1] || null;
}
function replaceVideoUrlInText(text: string, oldUrl: string, newUrl: string): string {
if (!text || !oldUrl || !newUrl || oldUrl === newUrl) return text;
return text.split(oldUrl).join(newUrl);
}
function isAlreadyArchivedVideoUrl(url: string): boolean {
return /\/uploads\//i.test(url);
}
function scheduleVideoArchive(params: {
chatId: string;
messageId: string;
text: string;
temporaryVideoUrl: string;
taskId?: string;
metadata?: Record<string, any> | null;
}) {
const { chatId, messageId, text, temporaryVideoUrl, taskId, metadata } = params;
if (!temporaryVideoUrl || isAlreadyArchivedVideoUrl(temporaryVideoUrl)) {
return;
}
after(async () => {
try {
const permanentVideoUrl = await videoArchiver.archiveVideo({
messageId,
temporaryUrl: temporaryVideoUrl,
taskId: taskId || messageId,
metadata: {
duration:
typeof metadata?.duration === "number" && Number.isFinite(metadata.duration)
? metadata.duration
: undefined,
resolution:
typeof metadata?.resolution === "string" ? metadata.resolution : undefined,
ratio: typeof metadata?.ratio === "string" ? metadata.ratio : undefined,
},
});
const archivedText = replaceVideoUrlInText(text, temporaryVideoUrl, permanentVideoUrl);
const nextMetadata = {
...(metadata || {}),
temporaryVideoUrl,
permanentVideoUrl,
storageStatus: "archived",
archivedAt: Date.now(),
updatedAt: Date.now(),
};
await updateMessageById({
messageId,
chatId,
parts: [{ type: "text", text: archivedText }],
videoStorageStatus: "archived",
temporaryVideoUrl,
permanentVideoUrl,
archivedAt: new Date(),
videoMetadata: nextMetadata as any,
});
console.log("[Chat API] ✅ 视频已归档到永久存储:", {
messageId,
permanentVideoUrl,
});
} catch (error) {
console.error("[Chat API] ❌ 视频归档失败(保留临时地址):", {
messageId,
temporaryVideoUrl,
error,
});
}
});
}
async function replayAssistantMessagesForDuplicateRequest({ async function replayAssistantMessagesForDuplicateRequest({
chatId, chatId,
userMessageId, userMessageId,
@ -3638,10 +3729,6 @@ export async function POST(request: Request) {
) { ) {
const currentVideoProvider = (languageModel as any).provider as "qianwen-video" | "fal-video"; const currentVideoProvider = (languageModel as any).provider as "qianwen-video" | "fal-video";
console.log(`[Chat API] Using ${currentVideoProvider} video generation`); console.log(`[Chat API] Using ${currentVideoProvider} video generation`);
const needsAsyncReplacement = currentVideoProvider === "qianwen-video";
const scheduleVideoUrlReplacement = needsAsyncReplacement
? (await import("@/lib/ai/providers/qianwen-video")).scheduleVideoUrlReplacement
: null;
const qianwenVideoStream = createUIMessageStream({ const qianwenVideoStream = createUIMessageStream({
execute: async ({ writer }) => { execute: async ({ writer }) => {
@ -3699,15 +3786,23 @@ export async function POST(request: Request) {
const videoMarkdown = result.text || "视频生成失败"; const videoMarkdown = result.text || "视频生成失败";
console.log(`[Chat API] ${currentVideoProvider} video generated, length:`, videoMarkdown.length); console.log(`[Chat API] ${currentVideoProvider} video generated, length:`, videoMarkdown.length);
// 提取临时视频 URL 和提示词(用于后续替换) // 提取视频 URL(后续会自动转存到永久存储)
const videoUrlMatch = videoMarkdown.match(/<video\s+src="([^"]+)"/); const tempVideoUrl = extractVideoUrlFromText(videoMarkdown) || "";
const tempVideoUrl = videoUrlMatch ? videoUrlMatch[1] : "";
const promptMatch = videoMarkdown.match(/\*\*提示词\*\*:\s*(.+)/);
const prompt = promptMatch ? promptMatch[1].trim() : "生成视频";
// 生成一个统一的消息 ID // 生成一个统一的消息 ID
const messageId = generateUUID(); const messageId = generateUUID();
const videoMetadata = {
provider: currentVideoProvider,
model: finalModelId,
mode: effectiveVideoParams?.mode,
resolution: effectiveVideoParams?.resolution,
ratio: effectiveVideoParams?.ratio,
duration: effectiveVideoParams?.duration,
status: "succeeded",
updatedAt: Date.now(),
};
// 写入消息到流 // 写入消息到流
writer.write({ writer.write({
type: "data-appendMessage", type: "data-appendMessage",
@ -3730,29 +3825,25 @@ export async function POST(request: Request) {
parts: [{ type: "text", text: videoMarkdown }], parts: [{ type: "text", text: videoMarkdown }],
attachments: [], attachments: [],
createdAt: new Date(), createdAt: new Date(),
videoStorageStatus: null, videoStorageStatus: tempVideoUrl ? "temporary" : null,
temporaryVideoUrl: null, temporaryVideoUrl: tempVideoUrl || null,
permanentVideoUrl: null, permanentVideoUrl: null,
archivedAt: null, archivedAt: null,
videoMetadata: effectiveVideoParams videoMetadata: videoMetadata as any,
? {
provider: currentVideoProvider,
mode: effectiveVideoParams.mode,
resolution: effectiveVideoParams.resolution,
ratio: effectiveVideoParams.ratio,
duration: effectiveVideoParams.duration,
createdAt: Date.now(),
}
: null,
}, },
], ],
}); });
console.log(`[Chat API] ✅ ${currentVideoProvider} video message saved successfully!`); console.log(`[Chat API] ✅ ${currentVideoProvider} video message saved successfully!`);
// 启动异步视频替换任务(5分钟后执行) if (tempVideoUrl) {
if (needsAsyncReplacement && tempVideoUrl && scheduleVideoUrlReplacement) { scheduleVideoArchive({
console.log("[Chat API] 🕐 Scheduling video URL replacement in 5 minutes..."); chatId: id,
scheduleVideoUrlReplacement(id, messageId, tempVideoUrl, prompt); messageId,
text: videoMarkdown,
temporaryVideoUrl: tempVideoUrl,
taskId: messageId,
metadata: videoMetadata,
});
} }
} catch (saveError) { } catch (saveError) {
console.error(`[Chat API] ❌ Failed to save ${currentVideoProvider} video message:`, saveError); console.error(`[Chat API] ❌ Failed to save ${currentVideoProvider} video message:`, saveError);
@ -4343,7 +4434,18 @@ export async function POST(request: Request) {
console.log("[Chat API] ✅ Video message updated successfully"); console.log("[Chat API] ✅ Video message updated successfully");
if (temporaryVideoUrl) { if (temporaryVideoUrl) {
console.log("[Chat API] 📹 临时视频URL已保存,将在8小时后自动归档到S3"); console.log("[Chat API] 📹 临时视频URL已保存,开始异步归档到S3");
scheduleVideoArchive({
chatId: id,
messageId: videoMessageId,
text: cleanedText,
temporaryVideoUrl,
taskId:
typeof successMetadata.taskId === "string" && successMetadata.taskId
? successMetadata.taskId
: videoMessageId,
metadata: successMetadata,
});
} }
console.log("[Chat API] Video generation finished successfully"); console.log("[Chat API] Video generation finished successfully");
@ -4503,7 +4605,7 @@ export async function POST(request: Request) {
})), })),
}); });
// 检测助手消息中是否包含视频,如果有则启动异步替换任务 // 检测助手消息中是否包含视频,如果有则立即标记并归档到永久存储
for (const msg of messages) { for (const msg of messages) {
if (msg.role === "assistant" && msg.parts) { if (msg.role === "assistant" && msg.parts) {
const textContent = msg.parts const textContent = msg.parts
@ -4511,21 +4613,32 @@ export async function POST(request: Request) {
.map((part: any) => part.text) .map((part: any) => part.text)
.join(""); .join("");
// 检测是否包含视频标签 const tempVideoUrl = extractVideoUrlFromText(textContent);
const videoUrlMatch = textContent?.match(/<video[^>]+src="([^"]+)"/); if (tempVideoUrl) {
if (videoUrlMatch) { const genericVideoMetadata = {
const tempVideoUrl = videoUrlMatch[1]; provider: "stream",
const promptMatch = textContent?.match(/\*\*提示词\*\*:\s*(.+)/); model: actualModelId,
const prompt = promptMatch ? promptMatch[1].trim() : "生成视频"; status: "succeeded",
updatedAt: Date.now(),
};
console.log("[Chat API] 检测到视频消息,启动URL替换任务"); await updateMessageById({
console.log("[Chat API] 临时视频URL:", tempVideoUrl); messageId: msg.id,
chatId: id,
videoStorageStatus: "temporary",
temporaryVideoUrl: tempVideoUrl,
permanentVideoUrl: null,
archivedAt: null,
videoMetadata: genericVideoMetadata as any,
});
// 动态导入并启动替换任务(5分钟后执行) scheduleVideoArchive({
import("@/lib/ai/providers/qianwen-video").then(({ scheduleVideoUrlReplacement }) => { chatId: id,
scheduleVideoUrlReplacement(id, msg.id, tempVideoUrl, prompt); messageId: msg.id,
}).catch((err) => { text: textContent,
console.error("[Chat API] 启动视频替换任务失败:", err); temporaryVideoUrl: tempVideoUrl,
taskId: msg.id,
metadata: genericVideoMetadata,
}); });
} }
} }