// backend\tasks_assets.go package main import ( "context" "encoding/json" "errors" "net/http" "os" "path/filepath" "sort" "strings" "sync" "time" ) // --------------------------- // Tasks: Missing Assets erzeugen // --------------------------- type AssetsTaskState struct { ID string `json:"id"` Queued bool `json:"queued"` Running bool `json:"running"` QueuedAt time.Time `json:"queuedAt"` StartedAt time.Time `json:"startedAt"` FinishedAt *time.Time `json:"finishedAt,omitempty"` Total int `json:"total"` Done int `json:"done"` GeneratedThumbs int `json:"generatedThumbs"` GeneratedPreviews int `json:"generatedPreviews"` Skipped int `json:"skipped"` Error string `json:"error,omitempty"` CurrentFile string `json:"currentFile,omitempty"` CurrentQueue string `json:"currentQueue,omitempty"` CurrentPhase string `json:"currentPhase,omitempty"` CurrentLabel string `json:"currentLabel,omitempty"` Text string `json:"text,omitempty"` } type assetsPhaseSelection struct { Meta bool Thumb bool Teaser bool Sprites bool Analyze bool } type generateAssetsRequest struct { Meta *bool `json:"meta,omitempty"` Thumb *bool `json:"thumb,omitempty"` Teaser *bool `json:"teaser,omitempty"` Sprites *bool `json:"sprites,omitempty"` Analyze *bool `json:"analyze,omitempty"` } type assetsTaskJob struct { State AssetsTaskState Cancel context.CancelFunc FileCancels map[string]context.CancelFunc SkipFiles map[string]struct{} Selection assetsPhaseSelection } var assetsJobsMu sync.Mutex var assetsJobs = map[string]*assetsTaskJob{} func selectedAssetsPhasesFromSettings() assetsPhaseSelection { s := getSettings() return assetsPhaseSelection{ Meta: s.GenerateAssetsMeta, Thumb: s.GenerateAssetsThumb, Teaser: s.GenerateAssetsTeaser, Sprites: s.GenerateAssetsSprites, Analyze: s.GenerateAssetsAnalyze, } } func selectedAssetsPhasesFromRequest(req *generateAssetsRequest) assetsPhaseSelection { sel := selectedAssetsPhasesFromSettings() if req == nil { return sel } if req.Meta != nil { sel.Meta = *req.Meta } if req.Thumb != nil { sel.Thumb = *req.Thumb } if req.Teaser != nil { sel.Teaser = *req.Teaser } if req.Sprites != nil { sel.Sprites = *req.Sprites } if req.Analyze != nil { sel.Analyze = *req.Analyze } return sel } func (p assetsPhaseSelection) Any() bool { return p.Meta || p.Thumb || p.Teaser || p.Sprites || p.Analyze } func assetsTaskFileKey(file string) string { return strings.ToLower(strings.TrimSpace(filepath.Base(file))) } func snapshotAssetsJobs() []AssetsTaskState { assetsJobsMu.Lock() defer assetsJobsMu.Unlock() now := time.Now() out := make([]AssetsTaskState, 0, len(assetsJobs)) for id, job := range assetsJobs { if job == nil { delete(assetsJobs, id) continue } st := job.State if !st.Queued && !st.Running && st.FinishedAt != nil && now.Sub(*st.FinishedAt) > 4*time.Second { delete(assetsJobs, id) continue } out = append(out, st) } sort.Slice(out, func(i, j int) bool { ai := out[i].QueuedAt aj := out[j].QueuedAt if ai.IsZero() { ai = out[i].StartedAt } if aj.IsZero() { aj = out[j].StartedAt } return ai.Before(aj) }) return out } func updateAssetsJobState(jobID string, fn func(st *AssetsTaskState)) (AssetsTaskState, bool) { assetsJobsMu.Lock() job := assetsJobs[jobID] if job == nil { assetsJobsMu.Unlock() return AssetsTaskState{}, false } fn(&job.State) st := job.State assetsJobsMu.Unlock() notifyAssetsChanged() publishTaskState() return st, true } func removeAssetsJob(jobID string) { assetsJobsMu.Lock() delete(assetsJobs, jobID) assetsJobsMu.Unlock() notifyAssetsChanged() publishTaskState() } func registerAssetsJobFileCancel(jobID, file string, cancel context.CancelFunc) { key := assetsTaskFileKey(file) if key == "" { return } assetsJobsMu.Lock() job := assetsJobs[jobID] if job != nil { if job.FileCancels == nil { job.FileCancels = map[string]context.CancelFunc{} } job.FileCancels[key] = cancel } assetsJobsMu.Unlock() } func clearAssetsJobFileControl(jobID, file string) { key := assetsTaskFileKey(file) if key == "" { return } assetsJobsMu.Lock() job := assetsJobs[jobID] if job != nil { delete(job.FileCancels, key) delete(job.SkipFiles, key) } assetsJobsMu.Unlock() } func requestAssetsJobSkip(jobID, file string) bool { key := assetsTaskFileKey(file) if key == "" { return false } var ( cancel context.CancelFunc marked bool ) assetsJobsMu.Lock() job := assetsJobs[jobID] if job != nil { if job.SkipFiles == nil { job.SkipFiles = map[string]struct{}{} } job.SkipFiles[key] = struct{}{} cancel = job.FileCancels[key] marked = true } assetsJobsMu.Unlock() if cancel != nil { cancel() } return marked } func isAssetsJobSkipRequested(jobID, file string) bool { key := assetsTaskFileKey(file) if key == "" { return false } assetsJobsMu.Lock() job := assetsJobs[jobID] if job == nil { assetsJobsMu.Unlock() return false } _, ok := job.SkipFiles[key] assetsJobsMu.Unlock() return ok } func tasksGenerateAssets(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodGet: writeJSON(w, http.StatusOK, snapshotAssetsJobs()) return case http.MethodPost: var req generateAssetsRequest if r.Body != nil && r.ContentLength != 0 { _ = json.NewDecoder(r.Body).Decode(&req) } sel := selectedAssetsPhasesFromRequest(&req) jobID := newTaskID("generate-assets") ctx, cancel := context.WithCancel(context.Background()) job := &assetsTaskJob{ State: AssetsTaskState{ ID: jobID, Queued: true, Running: false, QueuedAt: time.Now(), StartedAt: time.Time{}, FinishedAt: nil, Total: 0, Done: 0, GeneratedThumbs: 0, GeneratedPreviews: 0, Skipped: 0, Error: "", CurrentFile: "", CurrentQueue: "", CurrentPhase: "", CurrentLabel: "", Text: "Wartet…", }, Cancel: cancel, FileCancels: map[string]context.CancelFunc{}, SkipFiles: map[string]struct{}{}, Selection: sel, } assetsJobsMu.Lock() assetsJobs[jobID] = job assetsJobsMu.Unlock() notifyAssetsChanged() publishTaskState() go runGenerateMissingAssetsJob(jobID, ctx) writeJSON(w, http.StatusOK, job.State) return case http.MethodDelete: id := strings.TrimSpace(r.URL.Query().Get("id")) if id == "" { http.Error(w, "id fehlt", http.StatusBadRequest) return } assetsJobsMu.Lock() job := assetsJobs[id] assetsJobsMu.Unlock() if job == nil || job.Cancel == nil { w.WriteHeader(http.StatusNoContent) return } job.Cancel() w.WriteHeader(http.StatusNoContent) return default: http.Error(w, "Nur GET/POST/DELETE", http.StatusMethodNotAllowed) return } } func publishAssetsTaskPhase(fileName string, queue string, phase string, state string, label string) { fileName = strings.TrimSpace(fileName) if fileName == "" { return } base := strings.TrimSuffix(fileName, filepath.Ext(fileName)) assetID := stripHotPrefix(base) publishFinishedPostworkPhase( fileName, assetID, queue, phase, state, label, "", ) } func clearAssetsTaskFinishedPostworkStates(fileName string, sel assetsPhaseSelection) { if sel.Meta { publishAssetsTaskPhase(fileName, "postwork", "meta", "missing", "") } if sel.Thumb { publishAssetsTaskPhase(fileName, "postwork", "thumb", "missing", "") } if sel.Teaser { publishAssetsTaskPhase(fileName, "postwork", "teaser", "missing", "") } if sel.Sprites { publishAssetsTaskPhase(fileName, "postwork", "sprites", "missing", "") } if sel.Analyze { publishAssetsTaskPhase(fileName, "enrich", "analyze", "missing", "") } } func fileExistsNonEmpty(path string) bool { path = strings.TrimSpace(path) if path == "" { return false } fi, err := os.Stat(path) return err == nil && fi != nil && !fi.IsDir() && fi.Size() > 0 } type orderedAssetItem struct { name string path string } func listAssetsDoneThenKeepByFilenameAsc(doneAbs string) []orderedAssetItem { indexedItems, _ := buildDoneIndex(doneAbs) doneItems := make([]orderedAssetItem, 0, len(indexedItems)) keepItems := make([]orderedAssetItem, 0, len(indexedItems)) seen := make(map[string]struct{}, len(indexedItems)) for _, entry := range indexedItems { if entry.job == nil { continue } full := strings.TrimSpace(entry.job.Output) if full == "" { continue } cleanKey := strings.ToLower(filepath.Clean(full)) if _, ok := seen[cleanKey]; ok { continue } seen[cleanKey] = struct{}{} it := orderedAssetItem{ name: filepath.Base(full), path: full, } if entry.fromKeep { keepItems = append(keepItems, it) } else { doneItems = append(doneItems, it) } } sort.Slice(doneItems, func(i, j int) bool { ai := strings.ToLower(doneItems[i].name) aj := strings.ToLower(doneItems[j].name) if ai != aj { return ai < aj } return strings.ToLower(filepath.Clean(doneItems[i].path)) < strings.ToLower(filepath.Clean(doneItems[j].path)) }) sort.Slice(keepItems, func(i, j int) bool { ai := strings.ToLower(keepItems[i].name) aj := strings.ToLower(keepItems[j].name) if ai != aj { return ai < aj } return strings.ToLower(filepath.Clean(keepItems[i].path)) < strings.ToLower(filepath.Clean(keepItems[j].path)) }) return append(doneItems, keepItems...) } func assetsTaskTruthForVideo(id string, videoPath string) finishedPhaseTruth { truth := finishedPhaseTruthForID(id) _, thumbPath, previewPath, spritePath, metaPath, err := assetPathsForID(id) if err != nil { return truth } // meta muss nicht nur logisch "ready" sein, sondern die Datei muss auch existieren truth.MetaReady = truth.MetaReady && fileExistsNonEmpty(metaPath) // dateibasierte Assets IMMER an echter Datei festmachen truth.ThumbReady = fileExistsNonEmpty(thumbPath) truth.TeaserReady = fileExistsNonEmpty(previewPath) truth.SpritesReady = fileExistsNonEmpty(spritePath) // Analyse gilt als fertig, wenn alle Standard-Ziele gespeichert sind. // Auch "keine Treffer" zählt als fertig. if strings.TrimSpace(videoPath) != "" { truth.AnalyzeReady = hasAIAnalysisForAllOutputGoals(videoPath, requiredAnalyzeGoals()) } return truth } func runGenerateMissingAssetsJob(jobID string, ctx context.Context) { assetsJobsMu.Lock() job := assetsJobs[jobID] if job == nil { assetsJobsMu.Unlock() return } sel := job.Selection assetsJobsMu.Unlock() // Worker-Ende: CancelFunc zurücksetzen (pro Run) defer func() { assetsJobsMu.Lock() job := assetsJobs[jobID] if job != nil { job.Cancel = nil } assetsJobsMu.Unlock() }() finishWithErr := func(err error) { now := time.Now() updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.Queued = false st.Running = false st.FinishedAt = &now st.CurrentFile = "" st.CurrentQueue = "" st.CurrentPhase = "" st.CurrentLabel = "" if err == nil { st.Error = "" return } if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { st.Error = "Abgebrochen" } else { st.Error = err.Error() } }) assetsJobsMu.Lock() job := assetsJobs[jobID] if job != nil { job.FileCancels = map[string]context.CancelFunc{} job.SkipFiles = map[string]struct{}{} } assetsJobsMu.Unlock() time.AfterFunc(4*time.Second, func() { removeAssetsJob(jobID) }) } // Wichtig: // Die Phasen kommen jetzt bereits von außen (Request-Overrides oder gespeicherte Settings) if !sel.Any() { finishWithErr(nil) return } if err := acquireExclusiveTask(ctx); err != nil { finishWithErr(err) return } defer releaseExclusiveTask() updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.Queued = false st.Running = true st.StartedAt = time.Now() st.Text = "" }) settings := getSettings() doneAbs, err := resolvePathRelativeToApp(settings.DoneDir) if err != nil || strings.TrimSpace(doneAbs) == "" { if err == nil { err = appErrorf("doneDir auflösung fehlgeschlagen: doneDir ist leer") } else { err = appErrorf("doneDir auflösung fehlgeschlagen: %w", err) } finishWithErr(err) return } items := listAssetsDoneThenKeepByFilenameAsc(doneAbs) type assetCandidate struct { name string path string id string beforeTruth finishedPhaseTruth } candidates := make([]assetCandidate, 0, len(items)) skippedCount := 0 initialDone := 0 for _, it := range items { if err := ctx.Err(); err != nil { finishWithErr(err) return } base := strings.TrimSuffix(it.name, filepath.Ext(it.name)) id := stripHotPrefix(base) if strings.TrimSpace(id) == "" { clearAssetsTaskFinishedPostworkStates(it.name, sel) skippedCount++ initialDone++ continue } vfi, verr := os.Stat(it.path) if verr != nil || vfi == nil || vfi.IsDir() || vfi.Size() <= 0 { clearAssetsTaskFinishedPostworkStates(it.name, sel) skippedCount++ initialDone++ continue } beforeTruth := assetsTaskTruthForVideo(id, it.path) needMeta := sel.Meta && !beforeTruth.MetaReady needThumb := sel.Thumb && !beforeTruth.ThumbReady needTeaser := sel.Teaser && !beforeTruth.TeaserReady needSprites := sel.Sprites && !beforeTruth.SpritesReady needAnalyze := sel.Analyze && !beforeTruth.AnalyzeReady // Bereits vollständig bzgl. ausgewählter Phasen -> als erledigt/skipped mitzählen if !needMeta && !needThumb && !needTeaser && !needSprites && !needAnalyze { clearAssetsTaskFinishedPostworkStates(it.name, sel) skippedCount++ initialDone++ continue } candidates = append(candidates, assetCandidate{ name: it.name, path: it.path, id: id, beforeTruth: beforeTruth, }) } updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.Total = len(items) // alle Dateien, inkl. keep + skips st.Done = initialDone // vorhandene/ungültige direkt als erledigt zählen st.GeneratedThumbs = 0 st.GeneratedPreviews = 0 st.Skipped = skippedCount st.Error = "" st.CurrentFile = "" st.CurrentQueue = "" st.CurrentPhase = "" st.CurrentLabel = "" st.Text = "" }) if len(items) == 0 { finishWithErr(nil) return } if len(candidates) == 0 { finishWithErr(nil) return } for _, it := range candidates { if err := ctx.Err(); err != nil { finishWithErr(err) return } fileCtx, fileCancel := context.WithCancel(ctx) registerAssetsJobFileCancel(jobID, it.name, fileCancel) finishFileControl := func() { fileCancel() clearAssetsJobFileControl(jobID, it.name) } skipCurrentFile := func() { clearAssetsTaskFinishedPostworkStates(it.name, sel) updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.Done++ st.Skipped++ st.CurrentFile = "" st.CurrentQueue = "" st.CurrentPhase = "" st.CurrentLabel = "" }) finishFileControl() } handleFileAbort := func(stepErr error) (skipFile bool, stopAll bool) { if stepErr == nil { if isAssetsJobSkipRequested(jobID, it.name) || !fileExistsNonEmpty(it.path) { return true, false } return false, false } if errors.Is(stepErr, context.Canceled) || errors.Is(stepErr, context.DeadlineExceeded) { // globaler Task wurde beendet if ctx.Err() != nil && !isAssetsJobSkipRequested(jobID, it.name) { finishFileControl() finishWithErr(context.Canceled) return false, true } // nur diese eine Datei wurde freigegeben/übersprungen return true, false } // Datei wurde evtl. während des Verarbeitungsschritts gelöscht/verschoben if isAssetsJobSkipRequested(jobID, it.name) || !fileExistsNonEmpty(it.path) { return true, false } return false, false } updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.CurrentFile = it.name }) if isAssetsJobSkipRequested(jobID, it.name) || !fileExistsNonEmpty(it.path) { skipCurrentFile() continue } id := it.id vfi, verr := os.Stat(it.path) if verr != nil || vfi == nil || vfi.IsDir() || vfi.Size() <= 0 { clearAssetsTaskFinishedPostworkStates(it.name, sel) updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.Done++ st.Skipped++ st.CurrentFile = "" st.CurrentQueue = "" st.CurrentPhase = "" st.CurrentLabel = "" }) finishFileControl() continue } beforeTruth := it.beforeTruth needMeta := sel.Meta && !beforeTruth.MetaReady needThumb := sel.Thumb && !beforeTruth.ThumbReady needTeaser := sel.Teaser && !beforeTruth.TeaserReady needSprites := sel.Sprites && !beforeTruth.SpritesReady needAnalyze := sel.Analyze && !beforeTruth.AnalyzeReady // Pfade einmalig über zentralen Helper _, _, _, _, metaPath, perr := assetPathsForID(id) if perr != nil { if needMeta { publishAssetsTaskPhase(it.name, "postwork", "meta", "error", "Meta fehlgeschlagen") } if needThumb { publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen") } if needTeaser { publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen") } if needSprites { publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen") } if needAnalyze { publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "Analyse fehlgeschlagen") } updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.Error = "mindestens ein Eintrag konnte nicht verarbeitet werden (siehe Logs)" st.Done++ st.CurrentFile = "" st.CurrentQueue = "" st.CurrentPhase = "" st.CurrentLabel = "" }) finishFileControl() appLogln("⚠️ assetPathsForID:", perr) continue } // SourceURL best-effort: aus bestehender meta.json sourceURL := "" if u, ok := readVideoMetaSourceURL(metaPath, vfi); ok { sourceURL = u } thumbGenerated := false previewGenerated := false // ---------------- // META // ---------------- if needMeta { updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.CurrentFile = it.name st.CurrentQueue = "postwork" st.CurrentPhase = "meta" st.CurrentLabel = "Meta" }) publishAssetsTaskPhase(it.name, "postwork", "meta", "running", "Meta") okMeta, merr := ensureMetaForVideoCtx(fileCtx, it.path, sourceURL) if skipFile, stopAll := handleFileAbort(merr); stopAll { return } else if skipFile { skipCurrentFile() continue } if merr != nil || !okMeta { publishAssetsTaskPhase(it.name, "postwork", "meta", "error", "Meta fehlgeschlagen") } else { publishAssetsTaskPhase(it.name, "postwork", "meta", "done", "Meta") } } // ---------------- // THUMB // ---------------- if needThumb { updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.CurrentFile = it.name st.CurrentQueue = "postwork" st.CurrentPhase = "thumb" st.CurrentLabel = "Vorschaubild" }) publishAssetsTaskPhase(it.name, "postwork", "thumb", "running", "Vorschaubild") thumbGeneratedNow, terr := ensureThumbForVideoCtx(fileCtx, it.path, sourceURL) if skipFile, stopAll := handleFileAbort(terr); stopAll { return } else if skipFile { skipCurrentFile() continue } if terr != nil { publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen") } else if assetsTaskTruthForVideo(id, it.path).ThumbReady { if thumbGeneratedNow { thumbGenerated = true } publishAssetsTaskPhase(it.name, "postwork", "thumb", "done", "Vorschaubild") } else { publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen") } } // ---------------- // TEASER // ---------------- if needTeaser { updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.CurrentFile = it.name st.CurrentQueue = "postwork" st.CurrentPhase = "teaser" st.CurrentLabel = "Teaser" }) publishAssetsTaskPhase(it.name, "postwork", "teaser", "running", "Teaser") genTeaser, terr := ensureTeaserForVideoCtx(fileCtx, it.path, sourceURL) if skipFile, stopAll := handleFileAbort(terr); stopAll { return } else if skipFile { skipCurrentFile() continue } if terr != nil { appLogln("❌ teaser generation fehlgeschlagen für", it.path+":", terr) publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen") } else if assetsTaskTruthForVideo(id, it.path).TeaserReady { if genTeaser { previewGenerated = true } publishAssetsTaskPhase(it.name, "postwork", "teaser", "done", "Teaser") } else { appLogln("❌ teaser generation fehlgeschlagen für", it.path+":", "preview.mp4 fehlt oder ist leer") publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen") } } // ---------------- // SPRITES // ---------------- if needSprites { updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.CurrentFile = it.name st.CurrentQueue = "postwork" st.CurrentPhase = "sprites" st.CurrentLabel = "Sprites" }) publishAssetsTaskPhase(it.name, "postwork", "sprites", "running", "Sprites") _, serr := ensureSpritesForVideoCtx(fileCtx, it.path, sourceURL) if skipFile, stopAll := handleFileAbort(serr); stopAll { return } else if skipFile { skipCurrentFile() continue } if serr != nil { publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen") } else if assetsTaskTruthForVideo(id, it.path).SpritesReady { publishAssetsTaskPhase(it.name, "postwork", "sprites", "done", "Sprites") } else { publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen") } } // ---------------- // ANALYZE // ---------------- if needAnalyze { updateAssetsJobState(jobID, func(st *AssetsTaskState) { st.CurrentFile = it.name st.CurrentQueue = "enrich" st.CurrentPhase = "analyze" st.CurrentLabel = "Analyse" }) publishAssetsTaskPhase(it.name, "enrich", "analyze", "running", "Analyse") _, aerr := ensureAnalyzeForVideoCtx(fileCtx, it.path, sourceURL, "highlights") if analysisSkippedByBadSprite(aerr) { publishFinishedPostworkPhase( it.name, id, "enrich", "analyze", "error", "Analyse übersprungen", aerr.Error(), ) continue } if skipFile, stopAll := handleFileAbort(aerr); stopAll { return } else if skipFile { skipCurrentFile() continue } if aerr != nil { publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "Analyse fehlgeschlagen") } else if assetsTaskTruthForVideo(id, it.path).AnalyzeReady { publishAssetsTaskPhase(it.name, "enrich", "analyze", "done", "Analyse") } else { publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "Analyse fehlgeschlagen") } } updateAssetsJobState(jobID, func(st *AssetsTaskState) { if thumbGenerated { st.GeneratedThumbs++ } if previewGenerated { st.GeneratedPreviews++ } st.Done++ st.CurrentFile = "" st.CurrentQueue = "" st.CurrentPhase = "" st.CurrentLabel = "" }) finishFileControl() } finishWithErr(nil) }