package task import ( "context" "ffmpeg-gui/internal/ffmpeg" "ffmpeg-gui/internal/media" "fmt" "os" "sync" "time" "github.com/google/uuid" ) // Event types for frontend updates. const ( EventTaskUpdated = "task:updated" EventTaskProgress = "task:progress" EventTaskLog = "task:log" ) // EventCallback is called when a task state changes. The frontend will receive these. type EventCallback func(eventType string, data any) // Manager manages the task queue. type Manager struct { mu sync.Mutex tasks []*Task exec *ffmpeg.Executor hwAccel string // hardware accel method: "cuda", "d3d11va", "qsv", or "" onEvent EventCallback running bool ctx context.Context cancel context.CancelFunc } // NewManager creates a new task manager. func NewManager(exec *ffmpeg.Executor) *Manager { ctx, cancel := context.WithCancel(context.Background()) return &Manager{ exec: exec, ctx: ctx, cancel: cancel, } } // SetEventCallback sets the function called on task events. func (m *Manager) SetEventCallback(cb EventCallback) { m.onEvent = cb } // SetHWAccel sets the preferred hardware acceleration method. func (m *Manager) SetHWAccel(accel string) { m.hwAccel = accel } // Add adds a task to the queue and returns its ID. func (m *Manager) Add(t *Task) string { m.mu.Lock() defer m.mu.Unlock() t.ID = uuid.New().String()[:8] t.Status = StatusPending t.CreatedAt = time.Now() m.tasks = append(m.tasks, t) m.emit(EventTaskUpdated, t) return t.ID } // Start begins processing the queue. func (m *Manager) Start(taskID string) error { m.mu.Lock() t := m.find(taskID) if t == nil { m.mu.Unlock() return fmt.Errorf("task %s not found", taskID) } if t.Status != StatusPending { m.mu.Unlock() return fmt.Errorf("task %s is not pending", taskID) } t.Status = StatusRunning m.emit(EventTaskUpdated, t) m.mu.Unlock() go m.runTask(t) return nil } // StartAll starts all pending tasks sequentially. func (m *Manager) StartAll() { m.mu.Lock() defer m.mu.Unlock() m.running = true go m.processLoop() } // Stop stops processing tasks. func (m *Manager) Stop() { m.mu.Lock() defer m.mu.Unlock() m.running = false if m.cancel != nil { m.cancel() } } // Cancel cancels a specific task. func (m *Manager) Cancel(taskID string) error { m.mu.Lock() t := m.find(taskID) if t == nil { m.mu.Unlock() return fmt.Errorf("task %s not found", taskID) } if t.Status != StatusRunning && t.Status != StatusPending { m.mu.Unlock() return fmt.Errorf("cannot cancel task in status %s", t.Status) } if t.Cancel != nil { t.Cancel() } now := time.Now() t.Status = StatusCanceled t.CompletedAt = &now m.emit(EventTaskUpdated, t) m.mu.Unlock() return nil } // Remove removes a completed/failed/canceled task. func (m *Manager) Remove(taskID string) error { m.mu.Lock() defer m.mu.Unlock() for i, t := range m.tasks { if t.ID == taskID { if t.Status == StatusRunning { return fmt.Errorf("cannot remove running task") } if t.Cancel != nil { t.Cancel() } m.tasks = append(m.tasks[:i], m.tasks[i+1:]...) return nil } } return fmt.Errorf("task %s not found", taskID) } // List returns all tasks. func (m *Manager) List() []*Task { m.mu.Lock() defer m.mu.Unlock() result := make([]*Task, len(m.tasks)) copy(result, m.tasks) return result } // processLoop loops through pending tasks and runs them one at a time. func (m *Manager) processLoop() { for m.running { m.mu.Lock() var next *Task for _, t := range m.tasks { if t.Status == StatusPending { next = t break } } if next == nil { m.running = false m.mu.Unlock() return } next.Status = StatusRunning m.emit(EventTaskUpdated, next) m.mu.Unlock() m.runTask(next) } } // runTask executes a single task. func (m *Manager) runTask(t *Task) { info, err := media.GetInfo(m.exec, t.InputFile) if err != nil { m.completeTask(t, false, fmt.Sprintf("读取文件信息失败: %v", err)) return } // For burn tasks: extract internal subtitles to temp files first. // The subtitles filter (libass) hangs on Windows when re-opening the // input via si= — extracting avoids the file re-open entirely. var tempFiles []string if t.Type == TypeBurn { for i, sub := range t.Subtitle.Subtitles { if sub.Source == "internal" { tmpPath := t.OutputFile + fmt.Sprintf(".sub_%d_tmp.srt", i) if _, err := m.exec.RunSync("-y", "-i", t.InputFile, "-map", fmt.Sprintf("0:s:%d", sub.Index), "-c:s", "srt", tmpPath); err != nil { m.completeTask(t, false, fmt.Sprintf("提取字幕轨道 #%d 失败: %v", sub.Index, err)) return } t.Subtitle.Subtitles[i].Source = "external" t.Subtitle.Subtitles[i].FilePath = tmpPath tempFiles = append(tempFiles, tmpPath) } } } args := BuildArgs(t, m.hwAccel) t.Args = args cancel, progressCh, logCh, errCh := m.exec.Run(args, info.GetDurationSeconds()) m.mu.Lock() t.Cancel = cancel m.mu.Unlock() done := make(chan struct{}) go func() { for line := range logCh { m.mu.Lock() t.Logs = append(t.Logs, line) m.mu.Unlock() m.emit(EventTaskLog, map[string]any{ "taskId": t.ID, "line": line, }) } close(done) }() var lastEmit time.Time for p := range progressCh { m.mu.Lock() t.Progress = p m.mu.Unlock() now := time.Now() if now.Sub(lastEmit) > 250*time.Millisecond { lastEmit = now m.emit(EventTaskProgress, map[string]any{ "taskId": t.ID, "progress": p, }) } } <-done runErr := <-errCh // Clean up temp subtitle files for _, f := range tempFiles { os.Remove(f) } if runErr != nil { m.completeTask(t, false, fmt.Sprintf("%s失败: %v", taskTypeLabel(t.Type), runErr)) } else { m.completeTask(t, true, "") } } func (m *Manager) completeTask(t *Task, success bool, errMsg string) { m.mu.Lock() defer m.mu.Unlock() // Don't overwrite canceled status if t.Status == StatusCanceled { return } now := time.Now() if success { t.Status = StatusDone t.Progress.Percent = 100 } else { t.Status = StatusFailed t.Error = errMsg } t.CompletedAt = &now m.emit(EventTaskUpdated, t) // Emit final progress so frontend bar reaches 100% m.emit(EventTaskProgress, map[string]any{ "taskId": t.ID, "progress": t.Progress, }) } func (m *Manager) find(taskID string) *Task { for _, t := range m.tasks { if t.ID == taskID { return t } } return nil } func (m *Manager) emit(eventType string, data any) { if m.onEvent != nil { m.onEvent(eventType, data) } } func taskTypeLabel(typ Type) string { switch typ { case TypeRemux: return "封装" case TypeEncode: return "转码" case TypeBurn: return "字幕烧录" default: return "任务" } }