490 lines
12 KiB
Go
490 lines
12 KiB
Go
// 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()
|
|
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 {
|
|
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()
|
|
|
|
notifyAssetsChanged()
|
|
publishTaskState()
|
|
|
|
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)
|
|
}
|