// backend\tasks_assets.go package main import ( "context" "errors" "fmt" "net/http" "os" "path/filepath" "strings" "sync" "time" ) // --------------------------- // Tasks: Missing Assets erzeugen // --------------------------- type AssetsTaskState struct { Queued bool `json:"queued"` Running bool `json:"running"` Total int `json:"total"` Done int `json:"done"` GeneratedThumbs int `json:"generatedThumbs"` GeneratedPreviews int `json:"generatedPreviews"` Skipped int `json:"skipped"` StartedAt time.Time `json:"startedAt"` FinishedAt *time.Time `json:"finishedAt,omitempty"` 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"` } var assetsTaskMu sync.Mutex var assetsTaskState AssetsTaskState var assetsTaskCancel context.CancelFunc // updateAssetsState mutiert den State atomar und triggert danach SSE notify. // notifyAssetsChanged() muss außerhalb des Locks passieren. func updateAssetsState(fn func(st *AssetsTaskState)) AssetsTaskState { assetsTaskMu.Lock() fn(&assetsTaskState) st := assetsTaskState assetsTaskMu.Unlock() notifyAssetsChanged() publishTaskState() return st } func snapshotAssetsState() AssetsTaskState { assetsTaskMu.Lock() st := assetsTaskState assetsTaskMu.Unlock() return st } func getGenerateAssetsTaskStatus() AssetsTaskState { return snapshotAssetsState() } func tasksGenerateAssets(w http.ResponseWriter, r *http.Request) { switch r.Method { case http.MethodGet: // GET bleibt als Fallback/Debug möglich (UI nutzt SSE) st := snapshotAssetsState() writeJSON(w, http.StatusOK, st) return case http.MethodPost: assetsTaskMu.Lock() if assetsTaskState.Running || assetsTaskState.Queued { st := assetsTaskState assetsTaskMu.Unlock() writeJSON(w, http.StatusOK, st) return } ctx, cancel := context.WithCancel(context.Background()) assetsTaskCancel = cancel assetsTaskState = AssetsTaskState{ Queued: true, Running: false, Total: 0, Done: 0, GeneratedThumbs: 0, GeneratedPreviews: 0, Skipped: 0, StartedAt: time.Time{}, FinishedAt: nil, Error: "", CurrentFile: "", CurrentQueue: "", CurrentPhase: "", CurrentLabel: "", Text: "Wartet…", } st := assetsTaskState assetsTaskMu.Unlock() notifyAssetsChanged() publishTaskState() go func() { if err := acquireExclusiveTask(ctx); err != nil { now := time.Now() updateAssetsState(func(st *AssetsTaskState) { st.Queued = false st.Running = false st.FinishedAt = &now if errors.Is(err, context.Canceled) { st.Error = "Abgebrochen" } else { st.Error = err.Error() } }) return } defer releaseExclusiveTask() updateAssetsState(func(st *AssetsTaskState) { st.Queued = false st.Running = true st.StartedAt = time.Now() st.Text = "" }) runGenerateMissingAssets(ctx) }() writeJSON(w, http.StatusOK, st) return case http.MethodDelete: assetsTaskMu.Lock() cancel := assetsTaskCancel running := assetsTaskState.Running queued := assetsTaskState.Queued assetsTaskMu.Unlock() if (!running && !queued) || cancel == nil { w.WriteHeader(http.StatusNoContent) return } // canceln: Worker merkt das beim nächsten ctx.Err() und beendet sauber cancel() // UI sofort informieren (ohne Running künstlich auf false zu setzen — // das macht der Worker zuverlässig im finishWithErr(context.Canceled)) st := updateAssetsState(func(st *AssetsTaskState) { if st.Running || st.Queued { st.Queued = false st.Running = false st.Error = "Abgebrochen" } }) writeJSON(w, http.StatusOK, st) 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) { defs := []struct { queue string phase string }{ {queue: "postwork", phase: "meta"}, {queue: "postwork", phase: "thumb"}, {queue: "postwork", phase: "teaser"}, {queue: "postwork", phase: "sprites"}, {queue: "enrich", phase: "analyze"}, } for _, def := range defs { publishAssetsTaskPhase(fileName, def.queue, def.phase, "missing", "") } } func runGenerateMissingAssets(ctx context.Context) { // Worker-Ende: CancelFunc zurücksetzen (pro Run) defer func() { assetsTaskMu.Lock() assetsTaskCancel = nil assetsTaskMu.Unlock() }() finishWithErr := func(err error) { now := time.Now() updateAssetsState(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() } }) } abortIfCanceled := func(stepErr error, fileName string) bool { if errors.Is(stepErr, context.Canceled) || errors.Is(stepErr, context.DeadlineExceeded) || ctx.Err() != nil { if strings.TrimSpace(fileName) != "" { clearAssetsTaskFinishedPostworkStates(fileName) } finishWithErr(context.Canceled) return true } return false } s := getSettings() doneAbs, err := resolvePathRelativeToApp(s.DoneDir) if err != nil || strings.TrimSpace(doneAbs) == "" { finishWithErr(fmt.Errorf("doneDir auflösung fehlgeschlagen: %v", err)) return } type item struct { name string path string } indexedItems, sortedIdx := buildDoneIndex(doneAbs) // gleiche Grundreihenfolge wie FinishedDownloads-Default: // includeKeep = true, sort = completed_desc order := sortedIdx["1|completed_desc"] seen := map[string]struct{}{} items := make([]item, 0, len(order)) for _, idx := range order { if idx < 0 || idx >= len(indexedItems) { continue } j := indexedItems[idx].job if j == nil { continue } full := strings.TrimSpace(j.Output) if full == "" { continue } if _, ok := seen[full]; ok { continue } seen[full] = struct{}{} items = append(items, item{ name: filepath.Base(full), path: full, }) } // Initialisierung: Total etc. + SSE Push type assetCandidate struct { name string path string id string beforeTruth finishedPhaseTruth } candidates := make([]assetCandidate, 0, len(items)) skippedCount := 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) skippedCount++ continue } vfi, verr := os.Stat(it.path) if verr != nil || vfi == nil || vfi.IsDir() || vfi.Size() <= 0 { clearAssetsTaskFinishedPostworkStates(it.name) skippedCount++ continue } beforeTruth := finishedPhaseTruthForID(id) needMeta := !beforeTruth.MetaReady needThumb := !beforeTruth.ThumbReady needTeaser := !beforeTruth.TeaserReady needSprites := !beforeTruth.SpritesReady needAnalyze := !beforeTruth.AnalyzeReady // Bereits vollständig -> gar nicht erst in die Arbeitsliste aufnehmen if !needMeta && !needThumb && !needTeaser && !needSprites && !needAnalyze { clearAssetsTaskFinishedPostworkStates(it.name) skippedCount++ continue } candidates = append(candidates, assetCandidate{ name: it.name, path: it.path, id: id, beforeTruth: beforeTruth, }) } updateAssetsState(func(st *AssetsTaskState) { st.Total = len(candidates) st.Done = 0 st.GeneratedThumbs = 0 st.GeneratedPreviews = 0 st.Skipped = skippedCount st.Error = "" st.CurrentFile = "" st.CurrentQueue = "" st.CurrentPhase = "" st.CurrentLabel = "" }) if len(candidates) == 0 { finishWithErr(nil) return } for i, it := range candidates { if err := ctx.Err(); err != nil { finishWithErr(err) return } updateAssetsState(func(st *AssetsTaskState) { st.CurrentFile = it.name }) id := it.id vfi, verr := os.Stat(it.path) if verr != nil || vfi == nil || vfi.IsDir() || vfi.Size() <= 0 { clearAssetsTaskFinishedPostworkStates(it.name) updateAssetsState(func(st *AssetsTaskState) { st.Done = i + 1 }) continue } beforeTruth := it.beforeTruth needMeta := !beforeTruth.MetaReady needThumb := !beforeTruth.ThumbReady needTeaser := !beforeTruth.TeaserReady needSprites := !beforeTruth.SpritesReady needAnalyze := !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", "KI-Analyse fehlgeschlagen") } updateAssetsState(func(st *AssetsTaskState) { st.Error = "mindestens ein Eintrag konnte nicht verarbeitet werden (siehe Logs)" st.Done = i + 1 }) fmt.Println("⚠️ 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 { updateAssetsState(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(ctx, it.path, sourceURL) if abortIfCanceled(merr, it.name) { return } if merr != nil || !okMeta { publishAssetsTaskPhase(it.name, "postwork", "meta", "error", "Meta fehlgeschlagen") } else { publishAssetsTaskPhase(it.name, "postwork", "meta", "done", "Meta") } } // ---------------- // THUMB // ---------------- if needThumb { updateAssetsState(func(st *AssetsTaskState) { st.CurrentFile = it.name st.CurrentQueue = "postwork" st.CurrentPhase = "thumb" st.CurrentLabel = "Vorschaubild" }) publishAssetsTaskPhase(it.name, "postwork", "thumb", "running", "Vorschaubild") resThumb, terr := ensurePrimaryAssetsForVideoWithProgressCtx(ctx, it.path, sourceURL, nil) if abortIfCanceled(terr, it.name) { return } if terr != nil { publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen") } else if finishedPhaseTruthForID(id).ThumbReady { if resThumb.ThumbGenerated { thumbGenerated = true } publishAssetsTaskPhase(it.name, "postwork", "thumb", "done", "Vorschaubild") } else { publishAssetsTaskPhase(it.name, "postwork", "thumb", "error", "Vorschaubild fehlgeschlagen") } } // ---------------- // TEASER // ---------------- if needTeaser { updateAssetsState(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(ctx, it.path, sourceURL) if abortIfCanceled(terr, it.name) { return } if terr != nil { publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen") } else if finishedPhaseTruthForID(id).TeaserReady { if genTeaser { previewGenerated = true } publishAssetsTaskPhase(it.name, "postwork", "teaser", "done", "Teaser") } else { publishAssetsTaskPhase(it.name, "postwork", "teaser", "error", "Teaser fehlgeschlagen") } } // ---------------- // SPRITES // ---------------- if needSprites { updateAssetsState(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(ctx, it.path, sourceURL) if abortIfCanceled(serr, it.name) { return } if serr != nil { publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen") } else if finishedPhaseTruthForID(id).SpritesReady { publishAssetsTaskPhase(it.name, "postwork", "sprites", "done", "Sprites") } else { publishAssetsTaskPhase(it.name, "postwork", "sprites", "error", "Sprites fehlgeschlagen") } } // ---------------- // ANALYZE // ---------------- if needAnalyze { updateAssetsState(func(st *AssetsTaskState) { st.CurrentFile = it.name st.CurrentQueue = "enrich" st.CurrentPhase = "analyze" st.CurrentLabel = "KI-Analyse" }) publishAssetsTaskPhase(it.name, "enrich", "analyze", "running", "KI-Analyse") _, aerr := ensureAnalyzeForVideoCtx(ctx, it.path, sourceURL, "nsfw") if abortIfCanceled(aerr, it.name) { return } if aerr != nil { publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "KI-Analyse fehlgeschlagen") } else if finishedPhaseTruthForID(id).AnalyzeReady { publishAssetsTaskPhase(it.name, "enrich", "analyze", "done", "KI-Analyse") } else { publishAssetsTaskPhase(it.name, "enrich", "analyze", "error", "KI-Analyse fehlgeschlagen") } } updateAssetsState(func(st *AssetsTaskState) { if thumbGenerated { st.GeneratedThumbs++ } if previewGenerated { st.GeneratedPreviews++ } st.Done = i + 1 }) } finishWithErr(nil) }