// backend\tasks_regenerate_assets.go package main import ( "context" "encoding/json" "errors" "net/http" "os" "path/filepath" "sort" "strings" "sync" "time" ) 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 ) 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("Analyse fehlt oder wurde nicht in meta.json gespeichert") } return nil } func markRegenerateAssetsJobRunning(file string) *regenerateAssetsJob { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() job := regenerateAssetsJobs[file] if job == nil { return nil } if job.State != "queued" { return nil } job.State = "running" job.StartedAt = time.Now() return job } 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 regeneratePhaseOrder(queue string, phase string) int { q := strings.ToLower(strings.TrimSpace(queue)) p := strings.ToLower(strings.TrimSpace(phase)) switch q + ":" + p { case "postwork:meta": return 0 case "postwork:thumb": return 1 case "postwork:teaser": return 2 case "postwork:sprites": return 3 case "enrich:analyze": return 4 default: return 999 } } func getRegenerateAssetsJobsSnapshot() []map[string]any { regenerateAssetsMu.Lock() defer regenerateAssetsMu.Unlock() jobs := make([]*regenerateAssetsJob, 0, len(regenerateAssetsJobs)) for _, job := range regenerateAssetsJobs { if job != nil { jobs = append(jobs, job) } } sort.SliceStable(jobs, func(i, j int) bool { ai := jobs[i].EnqueuedAt.UnixMilli() aj := jobs[j].EnqueuedAt.UnixMilli() if ai != aj { return ai < aj } return jobs[i].File < jobs[j].File }) out := make([]map[string]any, 0, len(jobs)) for _, job := range jobs { phaseList := make([]regenerateAssetsPhase, 0, len(job.Phases)) for _, p := range job.Phases { phaseList = append(phaseList, p) } sort.SliceStable(phaseList, func(i, j int) bool { oi := regeneratePhaseOrder(phaseList[i].Queue, phaseList[i].Phase) oj := regeneratePhaseOrder(phaseList[j].Queue, phaseList[j].Phase) if oi != oj { return oi < oj } return phaseList[i].TS < phaseList[j].TS }) phases := make([]map[string]any, 0, len(phaseList)) for _, p := range phaseList { 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": func() int64 { if job.EnqueuedAt.IsZero() { return 0 } return 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 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 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", "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) Analyse publishFinishedPostworkPhase(file, id, "enrich", "analyze", "running", "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", "Analyse fehlt oder wurde nicht in meta.json gespeichert") return } publishFinishedPostworkPhase(file, id, "enrich", "analyze", "done", "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 } regenerateAssetsMu.Lock() if existing := regenerateAssetsJobs[file]; existing != nil { state := existing.State errText := existing.Error phaseList := make([]regenerateAssetsPhase, 0, len(existing.Phases)) for _, p := range existing.Phases { phaseList = append(phaseList, p) } regenerateAssetsMu.Unlock() sort.SliceStable(phaseList, func(i, j int) bool { oi := regeneratePhaseOrder(phaseList[i].Queue, phaseList[i].Phase) oj := regeneratePhaseOrder(phaseList[j].Queue, phaseList[j].Phase) if oi != oj { return oi < oj } return phaseList[i].TS < phaseList[j].TS }) phases := make([]map[string]any, 0, len(phaseList)) for _, p := range phaseList { phases = append(phases, map[string]any{ "queue": p.Queue, "phase": p.Phase, "state": p.State, "label": p.Label, "ts": p.TS, }) } w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "ok": true, "file": file, "state": state, "error": errText, "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), } regenerateAssetsMu.Lock() regenerateAssetsJobs[file] = job regenerateAssetsMu.Unlock() 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", "Analyse", "") go func(file string, runCtx context.Context) { if err := acquireExclusiveTask(runCtx); err != nil { if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { finishRegenerateAssetsJob(file, "cancelled", "Abgebrochen") setRegenerateAssetsTaskError(file, "Abgebrochen") removeRegenerateAssetsJobLater(file, 2*time.Second) return } finishRegenerateAssetsJob(file, "error", err.Error()) setRegenerateAssetsTaskError(file, err.Error()) removeRegenerateAssetsJobLater(file, 5*time.Second) return } defer releaseExclusiveTask() runningJob := markRegenerateAssetsJobRunning(file) if runningJob == nil { return } runRegenerateAssetsJob(runningJob) }(file, runCtx) w.Header().Set("Content-Type", "application/json") _ = json.NewEncoder(w).Encode(map[string]any{ "ok": true, "file": file, "state": "queued", }) }