Files
ReelShortWeb/src/views/contentManagement/upload/useUploadQueue.ts
T

353 lines
10 KiB
TypeScript

// Author: fruit.com.cn
/** 剧集批量上传队列:手动开始、文件并发、进度加权、断点会话、失败隔离 */
import { computed, ref } from 'vue'
import { getUploadProgressApi } from '@/api/admin/media'
import { probeVideoDuration, resumeUpload, uploadFile } from '@/utils/uploader'
import {
listPendingSessions,
matchFileForSession,
removeUploadSession,
type UploadSessionRecord,
upsertUploadSession
} from './uploadSessions'
export type UploadItemStatus = 'waiting' | 'uploading' | 'merging' | 'done' | 'failed' | 'paused'
export interface UploadQueueItem {
id: string
file: File
fileName: string
fileSize: number
lastModified: number
percent: number
status: UploadItemStatus
error: string
uploadId: string
assetId?: number
url?: string
durationSec?: number
}
export interface UseUploadQueueOptions {
/** 当前短剧 id(调用时读取,避免切换剧后闭包过期) */
getDramaId: () => string
bizType?: string
}
const FILE_CONCURRENCY = 2
const CHUNK_CONCURRENCY = 3
const DEFAULT_BIZ_TYPE = 'EPISODE'
const MERGING_PERCENT = 95
const MAX_PERCENT = 100
let queueSeq = 0
function nextItemId(): string {
queueSeq += 1
return `uq-${Date.now()}-${queueSeq}`
}
function extractErrorMessage(error: unknown): string {
if (error instanceof Error && error.message) return error.message
return String(error || 'upload failed')
}
function isHttpNotFound(error: unknown): boolean {
const status = (error as { response?: { status?: number } })?.response?.status
return status === 404
}
/** 文件名自然排序(数字感知),保证队列序号稳定。 */
function compareFileName(a: string, b: string): number {
return a.localeCompare(b, undefined, { numeric: true, sensitivity: 'base' })
}
/**
* 创建上传队列控制器。
* @param options dramaId 与可选 bizType
*/
export function useUploadQueue(options: UseUploadQueueOptions) {
const items = ref<UploadQueueItem[]>([])
const paused = ref(false)
/** 用户是否已点过「开始上传」;选文件阶段不派发、不占用并发。 */
const started = ref(false)
const bizType = options.bizType || DEFAULT_BIZ_TYPE
let running = 0
const overallPercent = computed(() => {
const list = items.value
if (!list.length) return 0
const totalBytes = list.reduce((sum, item) => sum + item.fileSize, 0)
if (totalBytes <= 0) return 0
const weighted = list.reduce((sum, item) => sum + item.fileSize * item.percent, 0)
return Math.min(MAX_PERCENT, Math.round(weighted / totalBytes))
})
const isUploading = computed(() =>
items.value.some((item) => item.status === 'uploading' || item.status === 'merging')
)
/** 关闭弹窗需二次确认:仅在已开始且仍有未完成任务时。 */
const hasActiveUpload = computed(() => {
if (!started.value) return false
return items.value.some(
(item) =>
item.status === 'waiting' ||
item.status === 'uploading' ||
item.status === 'merging' ||
item.status === 'paused'
)
})
const canStart = computed(() => {
if (isUploading.value) return false
return items.value.some((item) => item.status === 'waiting' || item.status === 'paused')
})
const doneItems = computed(() => items.value.filter((item) => item.status === 'done'))
const failedCount = computed(() => items.value.filter((item) => item.status === 'failed').length)
/** 已开始且队列已无进行中任务时,可展示结果汇总。 */
const showResultSummary = computed(
() => started.value && items.value.length > 0 && !hasActiveUpload.value
)
function persistSession(item: UploadQueueItem, uploadId: string): void {
item.uploadId = uploadId
upsertUploadSession({
uploadId,
fileName: item.fileName,
fileSize: item.fileSize,
lastModified: item.lastModified,
dramaId: String(options.getDramaId()),
bizType,
createdAt: Date.now()
})
}
function clearSession(uploadId: string): void {
if (uploadId) removeUploadSession(uploadId)
}
function sortItemsByName(): void {
items.value.sort((a, b) => compareFileName(a.fileName, b.fileName))
}
async function runOne(item: UploadQueueItem, resumeId?: string): Promise<void> {
item.status = 'uploading'
item.error = ''
try {
const durationSec = await probeVideoDuration(item.file)
item.durationSec = durationSec
const uploadOptions = {
bizType,
concurrency: CHUNK_CONCURRENCY,
durationSec,
onSession: (uploadId: string) => persistSession(item, uploadId),
onProgress: (percent: number) => {
item.percent = percent
if (percent >= MERGING_PERCENT && item.status === 'uploading') {
item.status = 'merging'
}
}
}
const asset = resumeId
? await resumeUpload(item.file, resumeId, uploadOptions)
: await uploadFile(item.file, uploadOptions)
item.percent = MAX_PERCENT
item.status = 'done'
item.assetId = asset.id
item.url = asset.url
if (asset.durationSec != null) item.durationSec = asset.durationSec
clearSession(item.uploadId || resumeId || '')
} catch (error) {
item.status = 'failed'
item.error = extractErrorMessage(error)
}
}
function pump(): void {
if (!started.value || paused.value) return
while (running < FILE_CONCURRENCY && !paused.value) {
const next = items.value.find((item) => item.status === 'waiting')
if (!next) return
// 立刻占位,避免同步 while 循环把同一项派发两次
next.status = 'uploading'
running += 1
const resumeId = next.uploadId || undefined
void runOne(next, resumeId).finally(() => {
running -= 1
if (!paused.value) pump()
})
}
}
/** 入队本地文件(调用方已完成类型/大小校验);只展示待上传,不发起请求。 */
function enqueue(files: File[]): void {
const sorted = [...files].sort((a, b) => compareFileName(a.name, b.name))
for (const file of sorted) {
items.value.push({
id: nextItemId(),
file,
fileName: file.name,
fileSize: file.size,
lastModified: file.lastModified,
percent: 0,
status: 'waiting',
error: '',
uploadId: ''
})
}
sortItemsByName()
}
/** 开始或继续派发(含断点续传项上的 uploadId)。 */
function start(): void {
if (!items.value.length) return
started.value = true
paused.value = false
for (const item of items.value) {
if (item.status === 'paused') item.status = 'waiting'
}
void pump()
}
/** 暂停:不派新任务;进行中分片跑完。 */
function pause(): void {
if (!started.value) return
paused.value = true
for (const item of items.value) {
if (item.status === 'waiting') item.status = 'paused'
}
}
/** 重试单个失败项。 */
function retry(id: string): void {
const item = items.value.find((row) => row.id === id)
if (!item || item.status !== 'failed') return
item.error = ''
item.percent = item.uploadId ? item.percent : 0
item.status = paused.value ? 'paused' : 'waiting'
if (started.value && !paused.value) void pump()
}
/** 重试全部失败项。 */
function retryAll(): void {
for (const item of items.value) {
if (item.status === 'failed') {
item.error = ''
item.status = paused.value ? 'paused' : 'waiting'
}
}
if (started.value && !paused.value) void pump()
}
/** 仅选择阶段可移除;已开始上传后禁止直接移除,避免脏会话。 */
function remove(id: string): void {
if (started.value) return
const index = items.value.findIndex((row) => row.id === id)
if (index < 0) return
const [removed] = items.value.splice(index, 1)
clearSession(removed.uploadId)
}
/** 清空队列并清理本队列会话。 */
function clear(): void {
for (const item of items.value) {
clearSession(item.uploadId)
}
items.value = []
paused.value = false
started.value = false
running = 0
}
/** 读取当前剧未完成会话(24h 内)。 */
function getPendingSessions(): UploadSessionRecord[] {
return listPendingSessions(String(options.getDramaId()))
}
/**
* 用本地重选文件匹配会话并入队(待上传);不自动开始,由用户点「开始上传」续传。
* @returns matched / unmatched 会话列表
*/
async function resumeFromSessions(
sessions: UploadSessionRecord[],
files: File[]
): Promise<{ matched: UploadSessionRecord[]; unmatched: UploadSessionRecord[] }> {
const matched: UploadSessionRecord[] = []
const unmatched: UploadSessionRecord[] = []
for (const session of sessions) {
const file = matchFileForSession(session, files)
if (!file) {
unmatched.push(session)
continue
}
try {
const progressRes = await getUploadProgressApi(session.uploadId)
const progress = progressRes?.data
if (!progress) {
unmatched.push(session)
continue
}
if (String(progress.status || '').toLowerCase() === 'completed') {
clearSession(session.uploadId)
continue
}
items.value.push({
id: nextItemId(),
file,
fileName: file.name,
fileSize: file.size,
lastModified: file.lastModified,
percent: Math.min(MERGING_PERCENT, Number(progress.percent) || 0),
status: 'waiting',
error: '',
uploadId: session.uploadId
})
matched.push(session)
} catch (error) {
if (isHttpNotFound(error)) {
clearSession(session.uploadId)
} else {
unmatched.push(session)
}
}
}
sortItemsByName()
return { matched, unmatched }
}
/** 放弃续传:清理给定会话。 */
function abandonSessions(sessions: UploadSessionRecord[]): void {
for (const session of sessions) {
clearSession(session.uploadId)
}
}
return {
items,
overallPercent,
hasActiveUpload,
isUploading,
canStart,
doneItems,
failedCount,
showResultSummary,
started,
paused,
enqueue,
start,
pause,
retry,
retryAll,
remove,
clear,
getPendingSessions,
resumeFromSessions,
abandonSessions
}
}