// 视频转码API - 完整版本(支持自动恢复) import { NextRequest, NextResponse } from 'next/server'; import { getServerSession } from 'next-auth'; import { prisma } from '@/lib/prisma'; import { transcodeQueue, cleanupTranscodeResidues, VIDEO_MATERIAL_TYPES } from '@/lib/video-transcode-queue'; import { executeTranscodeTask } from '@/lib/video-transcode-task'; import { authOptions } from '@/lib/auth'; import { canUserViewMaterial, hasSystemAdminPermission } from '@/lib/department-permission'; export const dynamic = 'force-dynamic'; export const maxDuration = 3600; // 1小时 // 恢复标志:确保只执行一次 let recoveryChecked = false; const MATERIAL_ID_PATTERN = /^[A-Za-z0-9_-]{1,100}$/; const RATE_LIMIT_WINDOW_MS = 60 * 1000; const STATUS_RATE_LIMIT = 120; const START_RATE_LIMIT = 5; const requestCounts = new Map(); function consumeRateLimit(key: string, maxRequests: number): number | null { const now = Date.now(); const current = requestCounts.get(key); if (!current || current.resetAt <= now) { requestCounts.set(key, { count: 1, resetAt: now + RATE_LIMIT_WINDOW_MS }); return null; } if (current.count >= maxRequests) { return Math.max(1, Math.ceil((current.resetAt - now) / 1000)); } current.count += 1; return null; } function rateLimitResponse(retryAfter: number) { return NextResponse.json( { error: '请求过于频繁,请稍后再试' }, { status: 429, headers: { 'Retry-After': String(retryAfter) } }, ); } async function countInterruptedVideoTasks(): Promise { try { return await prisma.material.count({ where: { materialType: { in: VIDEO_MATERIAL_TYPES }, videoProcessingStatus: 'processing', videoSourceUrl: { not: null }, deletedAt: null, }, }); } catch (error) { console.error('[Transcode] Failed to count interrupted video tasks:', error); // 返回 -1 代表统计失败,后续逻辑会谨慎地触发兜底恢复 return -1; } } /** * R1修复:懒加载恢复中断的转码任务 * - 优先从新的 VideoTranscodeProgress 表恢复(持久化队列状态) * - 降级到旧的 Material.videoProcessingStatus 恢复(向后兼容) */ async function ensureRecoveryCompleted(): Promise { if (recoveryChecked) { return; } recoveryChecked = true; try { console.log('[Transcode] Starting recovery check...'); // R1修复:优先使用新的持久化队列恢复 let recovered = await transcodeQueue.recoverOnStartup(executeTranscodeTask); // 如果新表没有数据,并且存在真正被中断的任务,再降级到旧方法 if (recovered === 0) { const interruptedCount = await countInterruptedVideoTasks(); if (interruptedCount > 0 || interruptedCount === -1) { if (interruptedCount > 0) { console.log(`[Transcode] Found ${interruptedCount} interrupted processing tasks, running legacy recovery...`); } else { console.warn('[Transcode] Unable to verify interrupted tasks, running legacy recovery as a safeguard...'); } const { recovered: legacyRecovered, requeued } = await transcodeQueue.recoverInterruptedTasks( executeTranscodeTask, cleanupTranscodeResidues, { videoTypes: VIDEO_MATERIAL_TYPES } ); recovered += legacyRecovered + requeued; } else { console.log('[Transcode] No interrupted processing tasks detected, skipping legacy fallback.'); } } console.log(`[Transcode] Recovery completed: ${recovered} tasks recovered`); } catch (error) { console.error('[Transcode] Recovery failed:', error); // 恢复失败不影响正常流程,重置标志以便下次重试 recoveryChecked = false; } } // GET - 查询转码状态 export async function GET(request: NextRequest) { try { const session = await getServerSession(authOptions); if (!session?.user) { return NextResponse.json({ error: '请先登录' }, { status: 401 }); } const sessionUser = session.user as any; const isSystemAdministrator = hasSystemAdminPermission( sessionUser.role, Boolean(sessionUser.isSuperAdmin), ); const retryAfter = consumeRateLimit(`status:${String(sessionUser.id || 'unknown')}`, STATUS_RATE_LIMIT); if (retryAfter) { return rateLimitResponse(retryAfter); } if (isSystemAdministrator) { await ensureRecoveryCompleted(); } const { searchParams } = new URL(request.url); const materialId = String(searchParams.get('materialId') || '').trim(); if (!MATERIAL_ID_PATTERN.test(materialId)) { return NextResponse.json( { error: '素材ID格式无效' }, { status: 400 } ); } const material = await prisma.material.findUnique({ where: { id: materialId }, select: { id: true, userId: true, materialType: true, videoProcessingStatus: true, videoPreview540Url: true, videoPreviewClipUrl: true, videoCoverUrl: true, videoDuration: true, transcodeLastError: true, transcodeAttempts: true, } as any, }); if (!material) { return NextResponse.json( { error: '素材不存在' }, { status: 404 } ); } if (!isSystemAdministrator && (material as any).userId !== sessionUser.id) { const canView = await canUserViewMaterial( sessionUser.department, sessionUser.role, sessionUser.managedDepartment, (material as any).materialType, sessionUser.id, ); if (!canView) { return NextResponse.json({ error: '无权限查看该素材转码状态' }, { status: 403 }); } } // 获取转码队列状态 const queueStatus = transcodeQueue.getStatus(); const isProcessing = queueStatus.processingIds.includes(materialId); const waitingIndex = queueStatus.waiting.findIndex(id => id === materialId); const waitingTotal = queueStatus.waiting.length; return NextResponse.json({ materialId: material.id, status: material.videoProcessingStatus || 'pending', hasPreview540: !!material.videoPreview540Url, hasClip: !!material.videoPreviewClipUrl, hasCover: !!material.videoCoverUrl, duration: material.videoDuration, attempts: material.transcodeAttempts || 0, lastError: isSystemAdministrator ? material.transcodeLastError : null, progress: isProcessing ? queueStatus.progress.get(materialId) || null : null, stage: isProcessing ? queueStatus.stages.get(materialId) || null : null, queue: { waitingIndex: waitingIndex >= 0 ? waitingIndex + 1 : null, waitingTotal: isSystemAdministrator ? waitingTotal : waitingIndex >= 0 ? 1 : 0, processingIds: isProcessing ? [materialId] : [], maxConcurrent: isSystemAdministrator ? queueStatus.maxConcurrent : null, }, }); } catch (error) { console.error('[Transcode] GET error:', error); return NextResponse.json( { error: '查询转码状态失败' }, { status: 500 } ); } } // POST - 启动转码任务 export async function POST(request: NextRequest) { try { const session = await getServerSession(authOptions); if (!session?.user) { return NextResponse.json({ error: '请先登录' }, { status: 401 }); } const sessionUser = session.user as any; const isSystemAdministrator = hasSystemAdminPermission( sessionUser.role, Boolean(sessionUser.isSuperAdmin), ); const retryAfter = consumeRateLimit(`start:${String(sessionUser.id || 'unknown')}`, START_RATE_LIMIT); if (retryAfter) { return rateLimitResponse(retryAfter); } const contentLength = Number(request.headers.get('content-length') || '0'); if (Number.isFinite(contentLength) && contentLength > 2048) { return NextResponse.json({ error: '请求参数过大' }, { status: 413 }); } if (isSystemAdministrator) { await ensureRecoveryCompleted(); } const { materialId, forceRetry } = await request.json(); if (!MATERIAL_ID_PATTERN.test(String(materialId || '').trim())) { return NextResponse.json( { error: '素材ID格式无效' }, { status: 400 } ); } if (forceRetry !== undefined && typeof forceRetry !== 'boolean') { return NextResponse.json({ error: 'forceRetry 参数必须是布尔值' }, { status: 400 }); } const normalizedMaterialId = String(materialId).trim(); const material = await prisma.material.findFirst({ where: { id: normalizedMaterialId, deletedAt: null }, select: { id: true, userId: true, videoSourceUrl: true, videoCoverUrl: true, videoProcessingStatus: true, materialType: true, } as any, }); if (!material) { return NextResponse.json( { error: '素材不存在或已被删除' }, { status: 404 } ); } if (!isSystemAdministrator && (material as any).userId !== String(sessionUser.id || '')) { return NextResponse.json({ error: '只能启动自己素材的转码任务' }, { status: 403 }); } // 检查是否为视频类型 if (!VIDEO_MATERIAL_TYPES.includes((material as any).materialType)) { return NextResponse.json( { error: '此素材不是视频类型' }, { status: 400 } ); } // 检查是否有源视频 if (!(material as any).videoSourceUrl) { return NextResponse.json( { error: '未找到源视频文件' }, { status: 400 } ); } // 检查转码状态 if ((material as any).videoProcessingStatus === 'processing' && !forceRetry) { return NextResponse.json( { error: '转码任务正在进行中', materialId: normalizedMaterialId }, { status: 409 } ); } if ((material as any).videoProcessingStatus === 'done' && !forceRetry) { return NextResponse.json( { message: '转码已完成', materialId: normalizedMaterialId }, { status: 200 } ); } // 判断是否需要生成封面图(用户未上传时才生成) const needsCoverImage = !(material as any).videoCoverUrl; // 加入转码队列 await transcodeQueue.add( normalizedMaterialId, (material as any).videoSourceUrl, needsCoverImage, executeTranscodeTask ); console.log(`[Transcode] Task queued for material ${normalizedMaterialId}`); const queueStatus = transcodeQueue.getStatus(); return NextResponse.json({ message: '转码任务已启动', materialId: normalizedMaterialId, status: 'processing', queueStatus: isSystemAdministrator ? queueStatus : { waitingIndex: queueStatus.waiting.findIndex((id) => id === normalizedMaterialId) + 1 || null, processing: queueStatus.processingIds.includes(normalizedMaterialId), }, }); } catch (error) { console.error('[Transcode] POST error:', error); return NextResponse.json( { error: '转码任务启动失败' }, { status: 500 } ); } }