import { Injectable } from '@angular/core'; import { BehaviorSubject } from 'rxjs'; import { AuthCreditService } from './auth-credit.service'; import { BatchJob, BatchJobCreateInput, BatchJobItem, BatchJobRunResult, } from '../models/batch-job.model'; const JOBS_STORAGE_KEY = 'videoWorkflow.batchJobs.v1'; const ITEMS_STORAGE_KEY = 'videoWorkflow.batchJobItems.v1'; type BatchItemRunner = (item: BatchJobItem, job: BatchJob) => Promise; @Injectable({ providedIn: 'root' }) export class BatchJobService { private readonly jobsSubject = new BehaviorSubject(this.readJobs()); private readonly itemsSubject = new BehaviorSubject(this.readItems()); readonly jobs$ = this.jobsSubject.asObservable(); readonly items$ = this.itemsSubject.asObservable(); private runningJobId = ''; private stopRequested = false; constructor(private auth: AuthCreditService) {} get jobs(): BatchJob[] { return this.jobsSubject.value; } get items(): BatchJobItem[] { return this.itemsSubject.value; } listJobs(): BatchJob[] { const userId = this.currentUserId(); if (!userId) return []; return this.jobs.filter((job) => job.userId === userId); } listItems(batchJobId: string): BatchJobItem[] { return this.items.filter((item) => item.batchJobId === batchJobId); } create(input: BatchJobCreateInput): BatchJob { const userId = this.requireUserId(); const now = new Date().toISOString(); const jobId = this.createId('batch'); const items = input.items .map((item, index) => ({ id: this.createId('batch_item'), batchJobId: jobId, topicId: item.topicId, title: item.title?.trim() || `子任务 ${index + 1}`, input: this.cloneSerializable(item.input || {}), status: 'queued' as const, retryCount: 0, createdAt: now, updatedAt: now, })); const job: BatchJob = { id: jobId, userId, title: input.title?.trim() || '未命名批量任务', pipelineId: input.pipelineId, templateId: input.templateId, status: 'queued', total: items.length, completed: 0, failed: 0, estimatedCredits: Math.max(0, Math.ceil(Number(input.estimatedCredits || 0))), createdAt: now, updatedAt: now, }; this.setJobs([job, ...this.jobs]); this.setItems([...items, ...this.items]); return job; } pause(jobId: string): void { const job = this.findJob(jobId); if (!job || job.status !== 'running') return; this.stopRequested = true; this.patchJob(jobId, { status: 'paused' }); } cancel(jobId: string): void { this.stopRequested = this.runningJobId === jobId; this.patchJob(jobId, { status: 'cancelled' }); this.setItems(this.items.map((item) => item.batchJobId === jobId && (item.status === 'queued' || item.status === 'running') ? { ...item, status: 'cancelled', updatedAt: new Date().toISOString() } : item)); } retryItem(itemId: string): void { const item = this.items.find((row) => row.id === itemId); if (!item || item.status !== 'failed') return; this.patchItem(itemId, { status: 'queued', errorMessage: undefined, resultUrl: undefined, retryCount: Number(item.retryCount || 0) + 1, }); this.recalculateJob(item.batchJobId, 'queued'); } async run(jobId: string, runner: BatchItemRunner = this.mockRunner): Promise { const job = this.findJob(jobId); if (!job || job.status === 'cancelled') return; if (this.runningJobId && this.runningJobId !== jobId) { throw new Error('已有批量任务正在执行,请稍后再试'); } this.runningJobId = jobId; this.stopRequested = false; this.patchJob(jobId, { status: 'running' }); try { while (!this.stopRequested) { const currentJob = this.findJob(jobId); if (!currentJob || currentJob.status === 'cancelled') break; const nextItem = this.listItems(jobId).find((item) => item.status === 'queued'); if (!nextItem) break; this.patchItem(nextItem.id, { status: 'running', errorMessage: undefined }); try { const result = await runner(nextItem, currentJob); this.patchItem(nextItem.id, { status: 'completed', resultUrl: result?.resultUrl, generationTaskId: result?.generationTaskId, }); } catch (err: any) { this.patchItem(nextItem.id, { status: 'failed', errorMessage: err?.message || '子任务执行失败', }); } this.recalculateJob(jobId); } } finally { const finalJob = this.findJob(jobId); if (finalJob && finalJob.status === 'running') { this.recalculateJob(jobId); } this.runningJobId = ''; } } private mockRunner(item: BatchJobItem): Promise { return new Promise((resolve) => { window.setTimeout(() => resolve({ generationTaskId: `mock-${item.id}` }), 300); }); } private recalculateJob(jobId: string, fallbackStatus?: BatchJob['status']): void { const job = this.findJob(jobId); if (!job) return; const items = this.listItems(jobId); const completed = items.filter((item) => item.status === 'completed').length; const failed = items.filter((item) => item.status === 'failed').length; const cancelled = items.filter((item) => item.status === 'cancelled').length; const queued = items.filter((item) => item.status === 'queued').length; const running = items.filter((item) => item.status === 'running').length; const status = (() => { if (job.status === 'cancelled') return 'cancelled'; if (fallbackStatus) return fallbackStatus; if (running > 0) return 'running'; if (queued > 0) return job.status === 'paused' ? 'paused' : 'running'; if (completed === items.length && items.length > 0) return 'completed'; if (failed > 0 && completed + failed + cancelled === items.length) return 'failed'; return job.status; })(); this.patchJob(jobId, { completed, failed, status }); } private patchJob(id: string, patch: Partial): void { const now = new Date().toISOString(); this.setJobs(this.jobs.map((job) => job.id === id ? { ...job, ...patch, id, updatedAt: now } : job)); } private patchItem(id: string, patch: Partial): void { const now = new Date().toISOString(); this.setItems(this.items.map((item) => item.id === id ? { ...item, ...patch, id, updatedAt: now } : item)); } private findJob(id: string): BatchJob | undefined { const userId = this.currentUserId(); return this.jobs.find((job) => job.id === id && job.userId === userId); } private currentUserId(): string { return this.auth.currentUser?.objectId || 'local-guest'; } private requireUserId(): string { return this.currentUserId(); } private readJobs(): BatchJob[] { try { const parsed = JSON.parse(localStorage.getItem(JOBS_STORAGE_KEY) || '[]'); return Array.isArray(parsed) ? parsed : []; } catch { return []; } } private readItems(): BatchJobItem[] { try { const parsed = JSON.parse(localStorage.getItem(ITEMS_STORAGE_KEY) || '[]'); return Array.isArray(parsed) ? parsed : []; } catch { return []; } } private setJobs(jobs: BatchJob[]): void { const sorted = [...jobs].sort((a, b) => Date.parse(b.updatedAt) - Date.parse(a.updatedAt)).slice(0, 100); this.jobsSubject.next(sorted); localStorage.setItem(JOBS_STORAGE_KEY, JSON.stringify(sorted)); } private setItems(items: BatchJobItem[]): void { this.itemsSubject.next(items); localStorage.setItem(ITEMS_STORAGE_KEY, JSON.stringify(items)); } private createId(prefix: string): string { return `${prefix}_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`; } private cloneSerializable(value: T): T { return JSON.parse(JSON.stringify(value || {})); } }