// 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 { 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"` } 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() 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 { st := assetsTaskState assetsTaskMu.Unlock() writeJSON(w, http.StatusOK, st) return } // cancelbarer Context (pro Run) ctx, cancel := context.WithCancel(context.Background()) assetsTaskCancel = cancel now := time.Now() assetsTaskState = AssetsTaskState{ Running: true, Total: 0, Done: 0, GeneratedThumbs: 0, GeneratedPreviews: 0, Skipped: 0, StartedAt: now, FinishedAt: nil, Error: "", CurrentFile: "", CurrentQueue: "", CurrentPhase: "", CurrentLabel: "", } st := assetsTaskState assetsTaskMu.Unlock() // SSE: Start pushen notifyAssetsChanged() go runGenerateMissingAssets(ctx) writeJSON(w, http.StatusOK, st) return case http.MethodDelete: assetsTaskMu.Lock() cancel := assetsTaskCancel running := assetsTaskState.Running assetsTaskMu.Unlock() if !running || 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.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.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) { st.Error = "abgebrochen" } else { st.Error = err.Error() } }) } 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 updateAssetsState(func(st *AssetsTaskState) { st.Total = len(items) st.Done = 0 st.GeneratedThumbs = 0 st.GeneratedPreviews = 0 st.Skipped = 0 st.Error = "" }) for i, it := range items { if err := ctx.Err(); err != nil { finishWithErr(err) return } // aktuellen Dateinamen für UI setzen updateAssetsState(func(st *AssetsTaskState) { st.CurrentFile = it.name }) // ID aus Dateiname base := strings.TrimSuffix(it.name, filepath.Ext(it.name)) id := stripHotPrefix(base) if strings.TrimSpace(id) == "" { clearAssetsTaskFinishedPostworkStates(it.name) updateAssetsState(func(st *AssetsTaskState) { st.Done = i + 1 }) continue } // Datei-Info (validieren) vfi, verr := os.Stat(it.path) if verr != nil || vfi.IsDir() || vfi.Size() <= 0 { clearAssetsTaskFinishedPostworkStates(it.name) updateAssetsState(func(st *AssetsTaskState) { st.Done = i + 1 }) continue } // Vorher-Wahrheit ermitteln: was fehlt wirklich? beforeTruth := finishedPhaseTruthForID(id) needMeta := !beforeTruth.MetaReady needThumb := !beforeTruth.ThumbReady needTeaser := !beforeTruth.TeaserReady needSprites := !beforeTruth.SpritesReady needAnalyze := !beforeTruth.AnalyzeReady // Wenn gar nichts fehlt -> alte evtl. Live-States wegräumen und skippen if !needMeta && !needThumb && !needTeaser && !needSprites && !needAnalyze { clearAssetsTaskFinishedPostworkStates(it.name) updateAssetsState(func(st *AssetsTaskState) { st.Skipped++ st.Done = i + 1 }) continue } // 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 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 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 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 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 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) }