// backend\tasks_regenerate_assets.go package main import ( "context" "encoding/json" "errors" "fmt" "net/http" "os" "path/filepath" "strings" "sync" "time" ) const regenerateAssetsMaxParallel = 3 type regenerateAssetsRequest struct { File string `json:"file"` } type regenerateAssetsPhase struct { Queue string `json:"queue"` Phase string `json:"phase"` State string `json:"state"` Label string `json:"label,omitempty"` TS int64 `json:"ts"` } type regenerateAssetsJob struct { File string VideoPath string AssetID string Ctx context.Context Cancel context.CancelFunc State string Error string EnqueuedAt time.Time StartedAt time.Time EndedAt time.Time Phases map[string]regenerateAssetsPhase } var ( regenerateAssetsMu sync.Mutex regenerateAssetsJobs = map[string]*regenerateAssetsJob{} // key = file regenerateAssetsQueue []string // file names regenerateAssetsWorkersStarted bool ) func publishFinishedPostworkPhase( file string, assetID string, queue string, phase string, state string, label string, errMsg string, ) { setRegenerateAssetsJobPhase(file, queue, phase, state, label) ev := finishedPostworkEvent{ Type: "finished_postwork", File: file, AssetID: assetID, Queue: queue, State: state, Phase: phase, Label: label, Error: strings.TrimSpace(errMsg), TS: time.Now().UnixMilli(), } if b, err := json.Marshal(ev); err == nil { publishSSE("finishedPostwork", b) } } type regeneratePhaseTruth struct { MetaReady bool ThumbReady bool TeaserReady bool SpritesReady bool AnalyzeReady bool } func regeneratePhaseTruthForVideo(videoPath string, assetID string) regeneratePhaseTruth { videoPath = strings.TrimSpace(videoPath) assetID = strings.TrimSpace(assetID) if assetID == "" && videoPath != "" { assetID = assetIDFromVideoPath(videoPath) } var out regeneratePhaseTruth if assetID == "" { return out } _, thumbPath, previewPath, spritePath, metaPath, err := assetPathsForID(assetID) if err != nil { return out } fileExists := func(p string) bool { fi, err := os.Stat(strings.TrimSpace(p)) return err == nil && fi != nil && !fi.IsDir() && fi.Size() > 0 } out.ThumbReady = fileExists(thumbPath) out.TeaserReady = fileExists(previewPath) out.SpritesReady = fileExists(spritePath) // Meta ist nur "ready", wenn die meta.json gültig zur aktuellen Videodatei passt if videoPath != "" { if vfi, err := os.Stat(videoPath); err == nil && vfi != nil && !vfi.IsDir() && vfi.Size() > 0 { if meta, ok := readVideoMetaIfValid(metaPath, vfi); ok && meta != nil { out.MetaReady = true } } // KI nur dann fertig, wenn Sprite existiert UND AI in meta.json vorhanden ist out.AnalyzeReady = out.SpritesReady && hasAIResultsForOutput(videoPath) } return out } func firstMissingPrimaryRegeneratePhaseError(videoPath string, assetID string) error { truth := regeneratePhaseTruthForVideo(videoPath, assetID) if !truth.MetaReady { return errors.New("meta.json fehlt oder ist noch unvollständig") } if !truth.ThumbReady { return errors.New("preview.jpg fehlt oder ist leer") } if !truth.TeaserReady { return errors.New("preview.mp4 fehlt oder ist leer") } return nil } func firstMissingDeferredRegeneratePhaseError(videoPath string, assetID string) error { truth := regeneratePhaseTruthForVideo(videoPath, assetID) if !truth.SpritesReady { return errors.New("preview-sprite.jpg fehlt oder ist leer") } if !truth.AnalyzeReady { return errors.New("KI-Analyse fehlt oder wurde nicht in meta.json gespeichert") } return nil } func ensureRegenerateAssetsWorkersStarted() { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() if regenerateAssetsWorkersStarted { return } regenerateAssetsWorkersStarted = true for i := 0; i < regenerateAssetsMaxParallel; i++ { go regenerateAssetsWorker() } } func enqueueRegenerateAssetsJob(job *regenerateAssetsJob) { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() regenerateAssetsJobs[job.File] = job regenerateAssetsQueue = append(regenerateAssetsQueue, job.File) } func nextQueuedRegenerateAssetsJob() *regenerateAssetsJob { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() for len(regenerateAssetsQueue) > 0 { file := regenerateAssetsQueue[0] regenerateAssetsQueue = regenerateAssetsQueue[1:] job := regenerateAssetsJobs[file] if job == nil { continue } if job.State != "queued" { continue } job.State = "running" job.StartedAt = time.Now() return job } return nil } func finishRegenerateAssetsJob(file string, state string, errMsg string) { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() job := regenerateAssetsJobs[file] if job == nil { return } job.State = state job.Error = strings.TrimSpace(errMsg) job.EndedAt = time.Now() } func setRegenerateAssetsJobPhase( file string, queue string, phase string, state string, label string, ) { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() job := regenerateAssetsJobs[file] if job == nil { return } if job.Phases == nil { job.Phases = make(map[string]regenerateAssetsPhase) } key := strings.ToLower(strings.TrimSpace(queue)) + ":" + strings.ToLower(strings.TrimSpace(phase)) job.Phases[key] = regenerateAssetsPhase{ Queue: strings.TrimSpace(queue), Phase: strings.TrimSpace(phase), State: strings.TrimSpace(state), Label: strings.TrimSpace(label), TS: time.Now().UnixMilli(), } } func removeRegenerateAssetsJobLater(file string, after time.Duration) { time.AfterFunc(after, func() { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() delete(regenerateAssetsJobs, file) }) } func getRegenerateAssetsJobsSnapshot() []map[string]any { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() out := make([]map[string]any, 0, len(regenerateAssetsJobs)) for _, job := range regenerateAssetsJobs { phases := make([]map[string]any, 0, len(job.Phases)) for _, p := range job.Phases { phases = append(phases, map[string]any{ "queue": p.Queue, "phase": p.Phase, "state": p.State, "label": p.Label, "ts": p.TS, }) } out = append(out, map[string]any{ "file": job.File, "assetId": job.AssetID, "state": job.State, "error": job.Error, "phases": phases, "enqueuedAt": job.EnqueuedAt.UnixMilli(), "startedAt": func() int64 { if job.StartedAt.IsZero() { return 0 } return job.StartedAt.UnixMilli() }(), "endedAt": func() int64 { if job.EndedAt.IsZero() { return 0 } return job.EndedAt.UnixMilli() }(), }) } return out } func regenerateAssetsWorker() { for { job := nextQueuedRegenerateAssetsJob() if job == nil { time.Sleep(250 * time.Millisecond) continue } runRegenerateAssetsJob(job) } } func runRegenerateAssetsJob(job *regenerateAssetsJob) { file := job.File videoPath := job.VideoPath id := job.AssetID ctx := job.Ctx setRegenerateAssetsTaskRunning(file, file) cancelDeferredEnrichForFile(file) cancelJob := func() { finishRegenerateAssetsJob(file, "cancelled", "Abgebrochen") setRegenerateAssetsTaskError(file, "Abgebrochen") publishFinishedPostworkPhase(file, id, "postwork", "meta", "missing", "", "") publishFinishedPostworkPhase(file, id, "postwork", "thumb", "missing", "", "") publishFinishedPostworkPhase(file, id, "postwork", "teaser", "missing", "", "") publishFinishedPostworkPhase(file, id, "postwork", "sprites", "missing", "", "") publishFinishedPostworkPhase(file, id, "enrich", "analyze", "missing", "", "") removeRegenerateAssetsJobLater(file, 2*time.Second) } failJob := func(queue, phase, msg string) { msg = strings.TrimSpace(msg) if msg != "" { fmt.Printf("❌ regenerate-assets error file=%s queue=%s phase=%s error=%s\n", file, queue, phase, msg) } if queue != "" && phase != "" { publishFinishedPostworkPhase(file, id, queue, phase, "error", "Fehler", msg) } finishRegenerateAssetsJob(file, "error", msg) setRegenerateAssetsTaskError(file, msg) removeRegenerateAssetsJobLater(file, 5*time.Second) } if ctx.Err() == context.Canceled { cancelJob() return } // Initialstatus publishFinishedPostworkPhase(file, id, "postwork", "meta", "running", "Meta", "") publishFinishedPostworkPhase(file, id, "postwork", "thumb", "queued", "Vorschaubild", "") publishFinishedPostworkPhase(file, id, "postwork", "teaser", "queued", "Teaser", "") publishFinishedPostworkPhase(file, id, "postwork", "sprites", "queued", "Sprites", "") publishFinishedPostworkPhase(file, id, "enrich", "analyze", "queued", "KI-Analyse", "") // Alte Assets löschen if id != "" { if assetDir, _, _, _, metaPath, err := assetPathsForID(id); err == nil { _ = os.Remove(filepath.Join(assetDir, "preview.jpg")) _ = os.Remove(filepath.Join(assetDir, "preview.mp4")) _ = os.Remove(filepath.Join(assetDir, "preview-sprite.jpg")) _ = os.Remove(metaPath) } } // 1) Meta if _, err := ensureMetaForVideoCtx(ctx, videoPath, ""); err != nil { if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled { cancelJob() return } failJob("postwork", "meta", err.Error()) return } if !regeneratePhaseTruthForVideo(videoPath, id).MetaReady { failJob("postwork", "meta", "meta.json fehlt oder ist noch unvollständig") return } publishFinishedPostworkPhase(file, id, "postwork", "meta", "done", "Meta", "") // 2) Thumb publishFinishedPostworkPhase(file, id, "postwork", "thumb", "running", "Vorschaubild", "") if _, err := ensureThumbForVideoCtx(ctx, videoPath, ""); err != nil { if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled { cancelJob() return } failJob("postwork", "thumb", err.Error()) return } if !regeneratePhaseTruthForVideo(videoPath, id).ThumbReady { failJob("postwork", "thumb", "preview.jpg fehlt oder ist leer") return } publishFinishedPostworkPhase(file, id, "postwork", "thumb", "done", "Vorschaubild", "") // 3) Teaser publishFinishedPostworkPhase(file, id, "postwork", "teaser", "running", "Teaser", "") if _, err := ensureTeaserForVideoCtx(ctx, videoPath, ""); err != nil { if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled { cancelJob() return } failJob("postwork", "teaser", err.Error()) return } if !regeneratePhaseTruthForVideo(videoPath, id).TeaserReady { failJob("postwork", "teaser", "preview.mp4 fehlt oder ist leer") return } publishFinishedPostworkPhase(file, id, "postwork", "teaser", "done", "Teaser", "") // 4) Sprites publishFinishedPostworkPhase(file, id, "postwork", "sprites", "running", "Sprites", "") if _, err := ensureSpritesForVideoCtx(ctx, videoPath, ""); err != nil { if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled { cancelJob() return } failJob("postwork", "sprites", err.Error()) return } if !regeneratePhaseTruthForVideo(videoPath, id).SpritesReady { failJob("postwork", "sprites", "preview-sprite.jpg fehlt oder ist leer") return } publishFinishedPostworkPhase(file, id, "postwork", "sprites", "done", "Sprites", "") // 5) KI-Analyse publishFinishedPostworkPhase(file, id, "enrich", "analyze", "running", "KI-Analyse", "") if _, err := ensureAnalyzeForVideoCtx(ctx, videoPath, "", "nsfw"); err != nil { if errors.Is(err, context.Canceled) || ctx.Err() == context.Canceled { cancelJob() return } failJob("enrich", "analyze", err.Error()) return } if !regeneratePhaseTruthForVideo(videoPath, id).AnalyzeReady { failJob("enrich", "analyze", "KI-Analyse fehlt oder wurde nicht in meta.json gespeichert") return } publishFinishedPostworkPhase(file, id, "enrich", "analyze", "done", "KI-Analyse", "") finishRegenerateAssetsJob(file, "done", "") setRegenerateAssetsTaskDone(file, "Fertig") removeRegenerateAssetsJobLater(file, 3*time.Second) } func handleRegenerateAssets(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodPost && r.Method != http.MethodDelete { http.Error(w, "method not allowed", http.StatusMethodNotAllowed) return } if r.Method == http.MethodDelete { var req regenerateAssetsRequest _ = json.NewDecoder(r.Body).Decode(&req) file := strings.TrimSpace(req.File) if file == "" { regenerateAssetsMu.Lock() for _, j := range regenerateAssetsJobs { if j != nil && (j.State == "running" || j.State == "queued") { file = j.File break } } regenerateAssetsMu.Unlock() } if file == "" { http.Error(w, "missing file", http.StatusBadRequest) return } regenerateAssetsMu.Lock() job := regenerateAssetsJobs[file] regenerateAssetsMu.Unlock() if job != nil && job.Cancel != nil { job.Cancel() } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "ok": true, "cancelled": true, "file": file, }) return } var req regenerateAssetsRequest if err := json.NewDecoder(r.Body).Decode(&req); err != nil { http.Error(w, "invalid json body", http.StatusBadRequest) return } file := strings.TrimSpace(req.File) if file == "" { http.Error(w, "missing file", http.StatusBadRequest) return } if !isAllowedVideoExt(file) { http.Error(w, "nicht erlaubt", http.StatusForbidden) return } s := getSettings() doneAbs, err := resolvePathRelativeToApp(s.DoneDir) if err != nil { http.Error(w, "doneDir auflösung fehlgeschlagen: "+err.Error(), http.StatusInternalServerError) return } if strings.TrimSpace(doneAbs) == "" { http.Error(w, "doneDir ist leer", http.StatusBadRequest) return } videoPath, _, fi, err := resolveDoneFileByName(doneAbs, file) if err != nil { http.Error(w, "datei nicht gefunden", http.StatusNotFound) return } if fi == nil || fi.IsDir() || fi.Size() <= 0 { http.Error(w, "invalid output path", http.StatusBadRequest) return } ensureRegenerateAssetsWorkersStarted() regenerateAssetsMu.Lock() if existing := regenerateAssetsJobs[file]; existing != nil { regenerateAssetsMu.Unlock() w.Header().Set("Content-Type", "application/json") phases := make([]map[string]any, 0, len(existing.Phases)) for _, p := range existing.Phases { phases = append(phases, map[string]any{ "queue": p.Queue, "phase": p.Phase, "state": p.State, "label": p.Label, "ts": p.TS, }) } _ = json.NewEncoder(w).Encode(map[string]any{ "ok": true, "file": file, "state": existing.State, "error": existing.Error, "phases": phases, }) return } regenerateAssetsMu.Unlock() runCtx, runCancel := context.WithCancel(context.Background()) id := assetIDFromVideoPath(videoPath) if strings.TrimSpace(id) == "" { id = stripHotPrefix(strings.TrimSuffix(filepath.Base(file), filepath.Ext(file))) } job := ®enerateAssetsJob{ File: file, VideoPath: videoPath, AssetID: id, Ctx: runCtx, Cancel: runCancel, State: "queued", EnqueuedAt: time.Now(), Phases: make(map[string]regenerateAssetsPhase), } enqueueRegenerateAssetsJob(job) publishFinishedPostworkPhase(file, id, "postwork", "meta", "queued", "Meta", "") publishFinishedPostworkPhase(file, id, "postwork", "thumb", "queued", "Vorschaubild", "") publishFinishedPostworkPhase(file, id, "postwork", "teaser", "queued", "Teaser", "") publishFinishedPostworkPhase(file, id, "postwork", "sprites", "queued", "Sprites", "") publishFinishedPostworkPhase(file, id, "enrich", "analyze", "queued", "KI-Analyse", "") w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "ok": true, "file": file, "state": "queued", }) }