diff --git a/backend/chaturbate_autostart.go b/backend/chaturbate_autostart.go index 8845649..1af82b3 100644 --- a/backend/chaturbate_autostart.go +++ b/backend/chaturbate_autostart.go @@ -3,6 +3,7 @@ package main import ( + "encoding/json" "errors" "fmt" "net/url" @@ -17,7 +18,8 @@ type autoStartItem struct { userKey string url string blind bool - source string // "watched" | "manual" + source string // "watched" | "manual" | "resume" + force bool // true = ignoriert Autostart-Pause + Download-Limit } func normUser(s string) string { @@ -254,7 +256,37 @@ func clearAllPendingAutoStartOnStartup() error { continue } - if err := os.Remove(filepath.Join(dir, name)); err != nil && !errors.Is(err, os.ErrNotExist) { + path := filepath.Join(dir, name) + + raw, err := os.ReadFile(path) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + continue + } + return err + } + + var f pendingAutoStartFile + if err := json.Unmarshal(raw, &f); err != nil { + _ = os.Remove(path) + continue + } + + kept := make([]PendingAutoStartItem, 0, len(f.Items)) + for _, it := range f.Items { + if normalizePendingSourceServer(it.Source) == "resume" { + kept = append(kept, it) + } + } + + if len(kept) == 0 { + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return err + } + continue + } + + if err := savePendingAutoStartItems(strings.TrimSuffix(name, filepath.Ext(name)), kept); err != nil { return err } } @@ -301,9 +333,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { for { select { case <-scanTicker.C: - if isAutostartPaused() { - continue - } + autostartPaused := isAutostartPaused() s := getSettings() if !s.UseChaturbateAPI { @@ -340,6 +370,15 @@ func startChaturbateAutoStartWorker(store *ModelStore) { showByUser := map[string]string{} imageByUser := map[string]string{} + for _, r := range rooms { + u := normUser(r.Username) + if u == "" { + continue + } + showByUser[u] = normalizePendingShowServer(r.CurrentShow) + imageByUser[u] = selectBestRoomImageURL(r) + } + pendingAutoStartMu.Lock() manualItems, err := loadManualPendingAutoStartItemsForProvider("chaturbate") pendingAutoStartMu.Unlock() @@ -350,6 +389,16 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue } + pendingAutoStartMu.Lock() + resumeItems, err := loadResumePendingAutoStartItemsForProvider("chaturbate") + pendingAutoStartMu.Unlock() + if err != nil { + if verboseLogs() { + fmt.Println("⚠️ [autostart] load resume chaturbate pending failed:", err) + } + continue + } + manualByUser := map[string]PendingAutoStartItem{} manualOrder := make([]string, 0, len(manualItems)) @@ -376,13 +425,30 @@ func startChaturbateAutoStartWorker(store *ModelStore) { manualOrder = append(manualOrder, key) } - for _, r := range rooms { - u := normUser(r.Username) - if u == "" { + resumeByUser := map[string]PendingAutoStartItem{} + resumeOrder := make([]string, 0, len(resumeItems)) + + for _, it := range resumeItems { + key := normUser(it.ModelKey) + if key == "" { + key = chaturbateUserFromURL(it.URL) + } + if key == "" { continue } - showByUser[u] = normalizePendingShowServer(r.CurrentShow) - imageByUser[u] = selectBestRoomImageURL(r) + + it.ModelKey = key + it.URL = strings.TrimSpace(it.URL) + if it.URL == "" { + it.URL = fmt.Sprintf("https://chaturbate.com/%s/", key) + } + + if _, exists := resumeByUser[key]; exists { + continue + } + + resumeByUser[key] = it + resumeOrder = append(resumeOrder, key) } // laufende Jobs sammeln @@ -467,7 +533,8 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue } - if it.source == "manual" { + switch it.source { + case "manual": m, ok := manualByUser[it.userKey] if !ok { continue @@ -477,7 +544,21 @@ func startChaturbateAutoStartWorker(store *ModelStore) { if it.url == "" { continue } - } else { + + case "resume": + m, ok := resumeByUser[it.userKey] + if !ok { + continue + } + + it.url = strings.TrimSpace(m.URL) + if it.url == "" { + continue + } + + it.force = true + + default: m, ok := watchedByUser[it.userKey] if !ok { continue @@ -522,9 +603,98 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } offlineCandidates := make([]autoStartItem, 0, len(watchedOrder)) - nextPending := make([]PendingAutoStartItem, 0, len(watchedOrder)) + nextPending := make([]PendingAutoStartItem, 0, len(watchedOrder)+len(resumeOrder)+len(runningByUser)) now := time.Now() + resumePendingThisScan := map[string]bool{} + + // Sichtbare laufende Downloads bei private/hidden/away stoppen + // und als "resume" merken. Diese Resume-Einträge starten später + // unabhängig von Autostart-Pause und unabhängig vom Download-Limit. + for user, runningJob := range runningByUser { + if runningJob == nil { + continue + } + if runningJob.Hidden { + continue + } + + show := normalizePendingShowServer(showByUser[user]) + if show != "private" && show != "hidden" && show != "away" { + continue + } + + u := strings.TrimSpace(runningJob.SourceURL) + if u == "" { + u = fmt.Sprintf("https://chaturbate.com/%s/", user) + } + + img := strings.TrimSpace(imageByUser[user]) + + nextPending = append(nextPending, PendingAutoStartItem{ + ModelKey: user, + URL: u, + Mode: "wait_public", + CurrentShow: show, + ImageURL: img, + Source: "resume", + }) + + resumePendingThisScan[user] = true + + if verboseLogs() { + fmt.Println("⏸️ [autostart] stopped because model is no longer public:", user, show) + } + + stopJobsInternal([]*RecordJob{runningJob}) + } + + // Resume hat Vorrang und ignoriert Autostart-Pause. + for _, user := range resumeOrder { + itm := resumeByUser[user] + + u := strings.TrimSpace(itm.URL) + if u == "" { + u = fmt.Sprintf("https://chaturbate.com/%s/", user) + } + + show := normalizePendingShowServer(showByUser[user]) + img := strings.TrimSpace(imageByUser[user]) + + switch show { + case "public": + if runningByUser[user] != nil { + continue + } + if queued[user] { + continue + } + + queue = append(queue, autoStartItem{ + userKey: user, + url: u, + blind: false, + source: "resume", + force: true, + }) + queued[user] = true + + case "private", "hidden", "away", "offline", "unknown": + if resumePendingThisScan[user] { + continue + } + + nextPending = append(nextPending, PendingAutoStartItem{ + ModelKey: user, + URL: u, + Mode: "wait_public", + CurrentShow: show, + ImageURL: img, + Source: "resume", + }) + } + } + for _, user := range watchedOrder { m := watchedByUser[user] @@ -538,6 +708,9 @@ func startChaturbateAutoStartWorker(store *ModelStore) { switch show { case "public": + if autostartPaused { + continue + } if runningByUser[user] != nil { continue } @@ -566,11 +739,10 @@ func startChaturbateAutoStartWorker(store *ModelStore) { Source: "watched", }) - if runningJob := runningByUser[user]; runningJob != nil { - stopJobsInternal([]*RecordJob{runningJob}) - } - default: + if autostartPaused { + continue + } if runningByUser[user] != nil { continue } @@ -606,6 +778,9 @@ func startChaturbateAutoStartWorker(store *ModelStore) { switch show { case "public": + if autostartPaused { + continue + } if runningByUser[user] != nil { continue } @@ -629,6 +804,9 @@ func startChaturbateAutoStartWorker(store *ModelStore) { continue default: + if autostartPaused { + continue + } if runningByUser[user] != nil { continue } @@ -648,7 +826,8 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } } - // Nur EIN Offline-Kandidat gleichzeitig in die Queue + // Nur EIN Offline-Kandidat gleichzeitig in die Queue. + // Bei pausiertem Autostart werden oben keine normalen Offline-Kandidaten erzeugt. if !blindQueued && !hasHiddenProbeRunningForProvider("chaturbate") && len(offlineCandidates) > 0 { n := len(offlineCandidates) start := 0 @@ -675,7 +854,7 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } } - if selectedBlindUser != "" { + if selectedBlindUser != "" && !autostartPaused { if m, ok := watchedByUser[selectedBlindUser]; ok { u := resolveChaturbateURL(m) if u != "" { @@ -705,10 +884,6 @@ func startChaturbateAutoStartWorker(store *ModelStore) { pendingAutoStartMu.Unlock() case <-startTicker.C: - if isAutostartPaused() { - continue - } - s := getSettings() if !s.UseChaturbateAPI { continue @@ -736,18 +911,34 @@ func startChaturbateAutoStartWorker(store *ModelStore) { showByUser[u] = strings.ToLower(strings.TrimSpace(r.CurrentShow)) } - it := queue[0] - show := strings.TrimSpace(showByUser[it.userKey]) - isPublic := strings.Contains(show, "public") + paused := isAutostartPaused() - // Nicht-public nur einzeln nacheinander prüfen - if it.blind && hasHiddenProbeRunningForProvider("chaturbate") { + startIdx := -1 + for i, candidate := range queue { + if paused && !candidate.force { + continue + } + + startIdx = i + break + } + + if startIdx < 0 { continue } - queue = queue[1:] + it := queue[startIdx] + queue = append(queue[:startIdx], queue[startIdx+1:]...) delete(queued, it.userKey) + show := strings.TrimSpace(showByUser[it.userKey]) + isPublic := strings.Contains(show, "public") + + // Nicht-public nur einzeln nacheinander prüfen. + if it.blind && hasHiddenProbeRunningForProvider("chaturbate") { + continue + } + if isJobRunningForURL(it.url) { continue } @@ -756,8 +947,9 @@ func startChaturbateAutoStartWorker(store *ModelStore) { lastTry[it.userKey] = time.Now() job, err := startRecordingInternal(RecordRequest{ - URL: it.url, - Cookie: lastCookieHdr, + URL: it.url, + Cookie: lastCookieHdr, + IgnoreConcurrentLimit: it.force, }) if err != nil { if verboseLogs() { @@ -767,10 +959,34 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } if verboseLogs() { - fmt.Println("▶️ [autostart] started:", it.url) + if it.source == "resume" { + fmt.Println("▶️ [autostart] resumed:", it.url) + } else { + fmt.Println("▶️ [autostart] started:", it.url) + } } if job != nil { + if it.source == "resume" { + pendingAutoStartMu.Lock() + _ = removeResumePendingAutoStartItemForProvider("chaturbate", it.userKey) + pendingAutoStartMu.Unlock() + + go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, nil, func() { + pendingAutoStartMu.Lock() + _ = saveResumePendingAutoStartItemForProvider("chaturbate", PendingAutoStartItem{ + ModelKey: it.userKey, + URL: it.url, + Mode: "wait_public", + CurrentShow: "unknown", + Source: "resume", + }) + pendingAutoStartMu.Unlock() + }) + + continue + } + if it.source == "manual" { go chaturbateAbortIfNoOutput(job.ID, outputProbeMax, func() { pendingAutoStartMu.Lock() @@ -786,12 +1002,17 @@ func startChaturbateAutoStartWorker(store *ModelStore) { } } } else { + if paused && !it.force { + continue + } + lastBlindTry[it.userKey] = time.Now() job, err := startRecordingInternal(RecordRequest{ - URL: it.url, - Cookie: lastCookieHdr, - Hidden: true, + URL: it.url, + Cookie: lastCookieHdr, + Hidden: true, + IgnoreConcurrentLimit: it.force, }) if err != nil || job == nil { if verboseLogs() { diff --git a/backend/ml/train_detector_model.py b/backend/ml/train_detector_model.py index 0059807..2c90729 100644 --- a/backend/ml/train_detector_model.py +++ b/backend/ml/train_detector_model.py @@ -1,55 +1,245 @@ -# backend\ml\train_detector_model.py +# backend/ml/train_detector_model.py import argparse import json +import shutil from pathlib import Path + from ultralytics import YOLO +def emit_progress(stage, progress, message="", **extra): + out = { + "type": "progress", + "stage": stage, + "progress": max(0.0, min(1.0, float(progress))), + "message": message, + } + out.update(extra) + print(json.dumps(out, ensure_ascii=False), flush=True) + + +def count_yolo_samples(dataset_root: Path, split: str) -> int: + images_dir = dataset_root / "images" / split + labels_dir = dataset_root / "labels" / split + + if not images_dir.exists() or not labels_dir.exists(): + return 0 + + image_exts = {".jpg", ".jpeg", ".png", ".webp"} + count = 0 + + for image_path in images_dir.iterdir(): + if not image_path.is_file(): + continue + + if image_path.suffix.lower() not in image_exts: + continue + + label_path = labels_dir / f"{image_path.stem}.txt" + if label_path.exists() and label_path.stat().st_size > 0: + count += 1 + + return count + + +def safe_int(value, fallback): + try: + return int(value) + except Exception: + return fallback + + def main(): parser = argparse.ArgumentParser() parser.add_argument("--root", required=True) parser.add_argument("--base", default="yolo11n.pt") parser.add_argument("--epochs", default="80") parser.add_argument("--imgsz", default="640") + parser.add_argument("--device", default="cpu") + parser.add_argument("--workers", default="2") + parser.add_argument("--patience", default="20") args = parser.parse_args() root = Path(args.root).resolve() - yaml_path = root / "detector" / "dataset" / "dataset.yaml" + dataset_root = root / "detector" / "dataset" + yaml_path = dataset_root / "dataset.yaml" runs_dir = root / "detector" / "runs" + out_dir = root / "detector" / "model" + + epochs = max(1, safe_int(args.epochs, 80)) + imgsz = max(64, safe_int(args.imgsz, 640)) + workers = max(0, safe_int(args.workers, 2)) + patience = max(0, safe_int(args.patience, 20)) if not yaml_path.exists(): raise SystemExit(f"dataset.yaml not found: {yaml_path}") + train_count = count_yolo_samples(dataset_root, "train") + val_count = count_yolo_samples(dataset_root, "val") + + emit_progress( + "detector", + 0.01, + "YOLO-Dataset wird geprüft…", + trainSamples=train_count, + valSamples=val_count, + epochs=epochs, + imgsz=imgsz, + device=args.device, + ) + + if train_count <= 0: + raise SystemExit("no YOLO train samples found") + + if val_count <= 0: + raise SystemExit("no YOLO val samples found") + + emit_progress( + "detector", + 0.03, + "YOLO-Basismodell wird geladen…", + base=args.base, + ) + model = YOLO(args.base) + best_epoch = 0 + last_epoch = 0 + + def on_train_epoch_start(trainer): + epoch = int(getattr(trainer, "epoch", 0)) + 1 + total = int(getattr(trainer, "epochs", epochs) or epochs) + + emit_progress( + "detector", + 0.04 + 0.90 * ((epoch - 1) / max(1, total)), + f"Object Detector trainiert… Epoche {epoch}/{total}", + epoch=epoch, + epochs=total, + trainSamples=train_count, + valSamples=val_count, + ) + + def on_train_epoch_end(trainer): + nonlocal last_epoch + + epoch = int(getattr(trainer, "epoch", 0)) + 1 + total = int(getattr(trainer, "epochs", epochs) or epochs) + last_epoch = max(last_epoch, epoch) + + emit_progress( + "detector", + 0.04 + 0.90 * (epoch / max(1, total)), + f"Object Detector trainiert… Epoche {epoch}/{total}", + epoch=epoch, + epochs=total, + trainSamples=train_count, + valSamples=val_count, + ) + + def on_fit_epoch_end(trainer): + nonlocal best_epoch + + epoch = int(getattr(trainer, "epoch", 0)) + 1 + metrics = getattr(trainer, "metrics", None) or {} + + # Ultralytics nutzt je nach Version unterschiedliche Keys. + map50 = ( + metrics.get("metrics/mAP50(B)") + or metrics.get("metrics/mAP50") + or metrics.get("mAP50") + ) + map5095 = ( + metrics.get("metrics/mAP50-95(B)") + or metrics.get("metrics/mAP50-95") + or metrics.get("mAP50-95") + ) + + if map50 is not None or map5095 is not None: + best_epoch = epoch + + emit_progress( + "detector", + 0.04 + 0.90 * (epoch / max(1, epochs)), + f"Object Detector validiert… Epoche {epoch}/{epochs}", + epoch=epoch, + epochs=epochs, + mAP50=map50, + mAP5095=map5095, + ) + + model.add_callback("on_train_epoch_start", on_train_epoch_start) + model.add_callback("on_train_epoch_end", on_train_epoch_end) + model.add_callback("on_fit_epoch_end", on_fit_epoch_end) + + emit_progress( + "detector", + 0.05, + "Object Detector Training startet…", + trainSamples=train_count, + valSamples=val_count, + epochs=epochs, + ) + result = model.train( data=str(yaml_path), - epochs=int(args.epochs), - imgsz=int(args.imgsz), + epochs=epochs, + imgsz=imgsz, project=str(runs_dir), name="detect", exist_ok=True, - device="cpu", - workers=2, - patience=20, + device=args.device, + workers=workers, + patience=patience, + ) + + emit_progress( + "detector", + 0.96, + "Bestes YOLO-Modell wird übernommen…", + lastEpoch=last_epoch, + bestEpoch=best_epoch, ) best = runs_dir / "detect" / "weights" / "best.pt" - if not best.exists(): - raise SystemExit(f"best.pt not found after training: {best}") + last = runs_dir / "detect" / "weights" / "last.pt" + + if not best.exists(): + if last.exists(): + best = last + else: + raise SystemExit(f"best.pt not found after training: {best}") - out_dir = root / "detector" / "model" out_dir.mkdir(parents=True, exist_ok=True) final_model = out_dir / "best.pt" - final_model.write_bytes(best.read_bytes()) + shutil.copy2(best, final_model) - print(json.dumps({ + result_path = runs_dir / "detect" + status = { "ok": True, "model": str(final_model), - "runs": str(runs_dir / "detect"), - })) + "sourceModel": str(best), + "runs": str(result_path), + "trainSamples": train_count, + "valSamples": val_count, + "epochs": epochs, + "imgsz": imgsz, + "device": str(args.device), + } + + with (out_dir / "status.json").open("w", encoding="utf-8") as f: + json.dump(status, f, ensure_ascii=False, indent=2) + + emit_progress( + "detector", + 1.0, + "Object Detector fertig.", + **status, + ) + + print(json.dumps(status, ensure_ascii=False), flush=True) if __name__ == "__main__": diff --git a/backend/ml/train_scene_model.py b/backend/ml/train_scene_model.py index 8df3578..d9895a0 100644 --- a/backend/ml/train_scene_model.py +++ b/backend/ml/train_scene_model.py @@ -51,6 +51,15 @@ def target_from_annotation(annotation): return correction_target(annotation) +def emit_progress(stage, progress, message="", **extra): + out = { + "type": "progress", + "stage": stage, + "progress": max(0.0, min(1.0, float(progress))), + "message": message, + } + out.update(extra) + print(json.dumps(out, ensure_ascii=False), flush=True) def load_clip(): device = "cuda" if torch.cuda.is_available() else "cpu" @@ -157,31 +166,48 @@ def main(): model_dir.mkdir(parents=True, exist_ok=True) rows = read_jsonl(feedback_path) + total = max(1, len(rows)) + + emit_progress( + "scene", + 0.02, + "CLIP-Modell wird geladen…", + totalSamples=len(rows), + ) clip_model, processor, device = load_clip() + emit_progress( + "scene", + 0.08, + "CLIP-Embeddings werden vorbereitet…", + totalSamples=len(rows), + device=device, + ) + embeddings = [] labels = [] targets = [] used = 0 skipped = 0 - for row in rows: + for idx, row in enumerate(rows, start=1): sample_id = str(row.get("sampleId") or "").strip() - if not sample_id: - skipped += 1 - continue - - image_path = frames_dir / f"{sample_id}.jpg" - if not image_path.exists(): - skipped += 1 - continue - - label = target_from_annotation(row) - if not label: - label = "unknown" try: + if not sample_id: + skipped += 1 + continue + + image_path = frames_dir / f"{sample_id}.jpg" + if not image_path.exists(): + skipped += 1 + continue + + label = target_from_annotation(row) + if not label: + label = "unknown" + emb = embed_image(clip_model, processor, device, image_path) embeddings.append(emb) @@ -191,13 +217,33 @@ def main(): "sexPosition": label, }) used += 1 + except Exception as e: - print(f"skip {sample_id}: {repr(e)}") + print(f"skip {sample_id or ''}: {repr(e)}", flush=True) skipped += 1 + finally: + emit_progress( + "scene", + 0.08 + 0.78 * (idx / total), + f"Scene-Samples werden verarbeitet… {idx}/{len(rows)}", + currentSample=idx, + totalSamples=len(rows), + usedSamples=used, + skippedSamples=skipped, + ) + if used < 5: raise SystemExit(f"too few usable samples: {used}") + emit_progress( + "scene", + 0.88, + "Scene-Embeddings werden gespeichert…", + usedSamples=used, + skippedSamples=skipped, + ) + x = np.stack(embeddings).astype("float32") y = np.array(labels) @@ -210,9 +256,25 @@ def main(): with (model_dir / "scene_clip_targets.json").open("w", encoding="utf-8") as f: json.dump(targets, f, ensure_ascii=False, indent=2) + emit_progress( + "scene", + 0.93, + "Scene-KNN wird trainiert…", + usedSamples=used, + skippedSamples=skipped, + ) + knn = train_knn(x, y) joblib.dump(knn, model_dir / "scene_clip_knn.joblib") + emit_progress( + "scene", + 0.96, + "Scene-Logistic-Regression wird trainiert…", + usedSamples=used, + skippedSamples=skipped, + ) + lr_status = "skipped_single_class" lr = train_lr_if_possible(x, y) if lr is not None: @@ -246,6 +308,19 @@ def main(): with (model_dir / "status.json").open("w", encoding="utf-8") as f: json.dump(status, f, ensure_ascii=False, indent=2) + emit_progress( + "scene", + 1.0, + "CLIP-Scene-Positionsmodell fertig.", + usedSamples=used, + skippedSamples=skipped, + classes=sorted(counts.keys()), + logisticRegression=lr_status, + knn="trained", + ) + + print(json.dumps(status, ensure_ascii=False), flush=True) + if __name__ == "__main__": main() \ No newline at end of file diff --git a/backend/pending_autostart.go b/backend/pending_autostart.go index b81d070..2470b31 100644 --- a/backend/pending_autostart.go +++ b/backend/pending_autostart.go @@ -41,11 +41,106 @@ func normalizePendingSourceServer(v string) string { switch strings.TrimSpace(strings.ToLower(v)) { case "watched": return "watched" + case "resume": + return "resume" default: return "manual" } } +func loadResumePendingAutoStartItemsForProvider(provider string) ([]PendingAutoStartItem, error) { + provider = strings.ToLower(strings.TrimSpace(provider)) + if provider == "" { + return nil, errors.New("missing provider") + } + + items, err := loadPendingAutoStartItems(pendingAutoStartGlobalUserKey) + if err != nil { + return nil, err + } + + out := make([]PendingAutoStartItem, 0, len(items)) + for _, it := range items { + if normalizePendingSourceServer(it.Source) != "resume" { + continue + } + if pendingProviderFromURL(it.URL) != provider { + continue + } + + out = append(out, it) + } + + return out, nil +} + +func removeResumePendingAutoStartItemForProvider(provider, modelKey string) error { + provider = strings.ToLower(strings.TrimSpace(provider)) + modelKey = strings.ToLower(strings.TrimSpace(modelKey)) + + if provider == "" || modelKey == "" { + return nil + } + + items, err := loadPendingAutoStartItems(pendingAutoStartGlobalUserKey) + if err != nil { + return err + } + + next := make([]PendingAutoStartItem, 0, len(items)) + for _, it := range items { + if normalizePendingSourceServer(it.Source) == "resume" && + pendingProviderFromURL(it.URL) == provider && + strings.ToLower(strings.TrimSpace(it.ModelKey)) == modelKey { + continue + } + + next = append(next, it) + } + + return savePendingAutoStartItems(pendingAutoStartGlobalUserKey, next) +} + +func saveResumePendingAutoStartItemForProvider(provider string, item PendingAutoStartItem) error { + provider = strings.ToLower(strings.TrimSpace(provider)) + if provider == "" { + return errors.New("missing provider") + } + + items, err := loadPendingAutoStartItems(pendingAutoStartGlobalUserKey) + if err != nil { + return err + } + + item.ModelKey = strings.ToLower(strings.TrimSpace(item.ModelKey)) + item.URL = strings.TrimSpace(item.URL) + item.Mode = normalizePendingModeServer(item.Mode) + item.CurrentShow = normalizePendingShowServer(item.CurrentShow) + item.ImageURL = strings.TrimSpace(item.ImageURL) + item.Source = "resume" + + if item.ModelKey == "" || item.URL == "" { + return nil + } + + replaced := false + for i := range items { + if normalizePendingSourceServer(items[i].Source) == "resume" && + pendingProviderFromURL(items[i].URL) == provider && + strings.ToLower(strings.TrimSpace(items[i].ModelKey)) == item.ModelKey { + items[i] = item + replaced = true + break + } + } + + if !replaced { + items = append(items, item) + } + + return savePendingAutoStartItems(pendingAutoStartGlobalUserKey, items) +} + func normalizePendingModeServer(v string) string { if strings.TrimSpace(strings.ToLower(v)) == "probe_retry" { return "probe_retry" diff --git a/backend/postwork.go b/backend/postwork.go index 44d3734..24c1058 100644 --- a/backend/postwork.go +++ b/backend/postwork.go @@ -3,6 +3,7 @@ package main import ( "context" + "os" "strings" "sync" "time" @@ -11,10 +12,11 @@ import ( // Eine Nacharbeit (kann ffmpeg, ffprobe, thumbnails, rename, etc. enthalten) type PostWorkTask struct { Key string // z.B. Dateiname oder Job-ID, zum Deduplizieren + Path string // Datei, die während queued/running gegen Explorer-Löschen gelockt wird Run func(ctx context.Context) error Added time.Time - SortBucket int // 0 = /done, 1 = /done/keep - SortName string // Dateiname lower-case + SortBucket int + SortName string } type PostWorkQueue struct { @@ -33,6 +35,8 @@ type PostWorkQueue struct { waitingKeys []string // sortierte wartende Keys runningKeys map[string]struct{} // Keys, die gerade wirklich laufen (Semaphor gehalten) cancelByKey map[string]context.CancelFunc + + fileLocks map[string]*os.File } func NewPostWorkQueue(queueSize int, maxParallelFFmpeg int) *PostWorkQueue { @@ -53,6 +57,8 @@ func NewPostWorkQueue(queueSize int, maxParallelFFmpeg int) *PostWorkQueue { waitingKeys: make([]string, 0, queueSize), runningKeys: make(map[string]struct{}), cancelByKey: make(map[string]context.CancelFunc), + + fileLocks: make(map[string]*os.File), } pq.cond = sync.NewCond(&pq.mu) @@ -110,6 +116,20 @@ func removeQueuedTaskLocked(tasks []PostWorkTask, key string) []PostWorkTask { return tasks } +func postWorkTaskLockPath(task PostWorkTask) string { + return strings.TrimSpace(task.Path) +} + +func (pq *PostWorkQueue) unlockFileLocked(key string) { + f := pq.fileLocks[key] + if f == nil { + return + } + + _ = f.Close() + delete(pq.fileLocks, key) +} + // Enqueue dedupliziert nach Key (damit du nicht durch Events doppelt queue-st) func (pq *PostWorkQueue) Enqueue(task PostWorkTask) bool { if task.Key == "" || task.Run == nil { @@ -123,15 +143,30 @@ func (pq *PostWorkQueue) Enqueue(task PostWorkTask) bool { defer pq.mu.Unlock() if _, ok := pq.inflight[task.Key]; ok { - return false // schon queued oder läuft + return false } if pq.maxQueued > 0 && len(pq.queue) >= pq.maxQueued { return false } + lockPath := postWorkTaskLockPath(task) + var lockFile *os.File + + if lockPath != "" { + f, err := lockPostWorkFile(lockPath) + if err != nil { + return false + } + lockFile = f + } + pq.inflight[task.Key] = struct{}{} pq.queued++ + if lockFile != nil { + pq.fileLocks[task.Key] = lockFile + } + insertAt := len(pq.queue) for i, existing := range pq.queue { if lessPostWorkTask(task, existing) { @@ -204,6 +239,8 @@ func (pq *PostWorkQueue) RemoveQueued(key string) bool { } delete(pq.inflight, key) + pq.unlockFileLocked(key) + if pq.queued > 0 { pq.queued-- } @@ -247,6 +284,7 @@ func (pq *PostWorkQueue) workerLoop(id int) { delete(pq.runningKeys, task.Key) delete(pq.inflight, task.Key) delete(pq.cancelByKey, task.Key) + pq.unlockFileLocked(task.Key) if pq.queued > 0 { pq.queued-- } @@ -273,6 +311,10 @@ func (pq *PostWorkQueue) workerLoop(id int) { pq.mu.Lock() pq.removeWaitingKeyLocked(task.Key) pq.runningKeys[task.Key] = struct{}{} + + // Wichtig: Lock vor der eigentlichen Nacharbeit freigeben, + // damit moveToDoneDir/removeWithRetry unter Windows nicht blockiert werden. + pq.unlockFileLocked(task.Key) pq.mu.Unlock() if task.Run != nil { diff --git a/backend/postwork_lock_other.go b/backend/postwork_lock_other.go new file mode 100644 index 0000000..581095d --- /dev/null +++ b/backend/postwork_lock_other.go @@ -0,0 +1,13 @@ +// backend\postwork_lock_other.go + +//go:build !windows + +package main + +import "os" + +func lockPostWorkFile(path string) (*os.File, error) { + // Auf Unix verhindert ein offenes Handle kein Löschen. + // Der Fallback hält nur ein Handle, damit der Code plattformübergreifend baut. + return os.Open(path) +} diff --git a/backend/postwork_lock_windows.go b/backend/postwork_lock_windows.go new file mode 100644 index 0000000..91dba1d --- /dev/null +++ b/backend/postwork_lock_windows.go @@ -0,0 +1,34 @@ +// backend\postwork_lock_windows.go + +//go:build windows + +package main + +import ( + "os" + "syscall" +) + +const fileShareDelete = 0x00000004 + +func lockPostWorkFile(path string) (*os.File, error) { + ptr, err := syscall.UTF16PtrFromString(path) + if err != nil { + return nil, err + } + + handle, err := syscall.CreateFile( + ptr, + 0, // desired access: nur Handle halten, kein Lesen/Schreiben nötig + syscall.FILE_SHARE_READ|syscall.FILE_SHARE_WRITE, // bewusst OHNE FILE_SHARE_DELETE + nil, + syscall.OPEN_EXISTING, + syscall.FILE_ATTRIBUTE_NORMAL, + 0, + ) + if err != nil { + return nil, err + } + + return os.NewFile(uintptr(handle), path), nil +} diff --git a/backend/record.go b/backend/record.go index f793a12..cbf9499 100644 --- a/backend/record.go +++ b/backend/record.go @@ -39,6 +39,9 @@ type RecordRequest struct { Cookie string `json:"cookie,omitempty"` UserAgent string `json:"userAgent,omitempty"` Hidden bool `json:"hidden,omitempty"` + + // Intern: Resume nach private/hidden/away darf das normale Download-Limit ignorieren. + IgnoreConcurrentLimit bool `json:"-"` } type doneListResponse struct { diff --git a/backend/recorder.go b/backend/recorder.go index c97e7c4..d01c7bc 100644 --- a/backend/recorder.go +++ b/backend/recorder.go @@ -456,7 +456,7 @@ func startRecordingInternal(req RecordRequest) (*RecordJob, error) { // Limit-Check ATOMAR unter jobsMu s := getSettings() - if s.EnableConcurrentDownloadsLimit { + if s.EnableConcurrentDownloadsLimit && !req.IgnoreConcurrentLimit { max := s.MaxConcurrentDownloads if max < 1 { max = 1 @@ -1105,12 +1105,12 @@ func enqueuePostworkOrFail(job *RecordJob, out string, postTarget JobStatus) { okQueued := postWorkQ.Enqueue(PostWorkTask{ Key: postKey, + Path: out, Added: time.Now(), Run: func(ctx context.Context) error { return runQueuedPostwork(ctx, job, out, postTarget, postKey) }, }) - if okQueued { publishQueuedPostworkState(job, postKey, postFile, postAssetID) return diff --git a/backend/training.go b/backend/training.go index 669432a..b3236e3 100644 --- a/backend/training.go +++ b/backend/training.go @@ -3,6 +3,7 @@ package main import ( + "bufio" "crypto/sha1" "encoding/hex" "encoding/json" @@ -160,6 +161,139 @@ type TrainingStatsResponse struct { Labels TrainingStatsLabels `json:"labels"` } +type trainingProgressEvent struct { + Type string `json:"type"` + Stage string `json:"stage"` + Progress float64 `json:"progress"` // 0..1 + Message string `json:"message,omitempty"` + Epoch int `json:"epoch,omitempty"` + Epochs int `json:"epochs,omitempty"` +} + +func trainingScaleProgress(local float64, start int, end int) int { + if math.IsNaN(local) || math.IsInf(local, 0) { + local = 0 + } + + local = clamp01(local) + + if end < start { + end = start + } + + return start + int(math.Round(local*float64(end-start))) +} + +func trainingHandleProgressLine(line string, start int, end int, defaultStep string) bool { + line = strings.TrimSpace(line) + if line == "" { + return false + } + + var ev trainingProgressEvent + if err := json.Unmarshal([]byte(line), &ev); err != nil { + return false + } + + if ev.Type != "progress" { + return false + } + + progress := trainingScaleProgress(ev.Progress, start, end) + step := strings.TrimSpace(ev.Message) + if step == "" { + step = defaultStep + } + + trainingSetJobStatus(func(s *TrainingJobStatus) { + if progress > s.Progress { + s.Progress = progress + } + s.Step = step + }) + + return true +} + +func trainingRunCommandStreaming( + python string, + script string, + onLine func(line string) bool, + args ...string, +) (string, error) { + cmdArgs := append([]string{script}, args...) + cmd := exec.Command(python, cmdArgs...) + + cmd.SysProcAttr = &syscall.SysProcAttr{ + HideWindow: true, + CreationFlags: 0x08000000, + } + + stdout, err := cmd.StdoutPipe() + if err != nil { + return "", err + } + + stderr, err := cmd.StderrPipe() + if err != nil { + return "", err + } + + if err := cmd.Start(); err != nil { + return "", err + } + + var outMu sync.Mutex + var lines []string + + readPipe := func(scanner *bufio.Scanner) { + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + + for scanner.Scan() { + line := strings.TrimSpace(scanner.Text()) + if line == "" { + continue + } + + handled := false + if onLine != nil { + handled = onLine(line) + } + + // Progress-Events nicht in den finalen Output übernehmen. + if handled { + continue + } + + outMu.Lock() + lines = append(lines, line) + outMu.Unlock() + } + } + + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + readPipe(bufio.NewScanner(stdout)) + }() + + go func() { + defer wg.Done() + readPipe(bufio.NewScanner(stderr)) + }() + + err = cmd.Wait() + wg.Wait() + + outMu.Lock() + out := strings.Join(lines, "\n") + outMu.Unlock() + + return strings.TrimSpace(out), err +} + const minTrainingFeedbackCount = 5 const minDetectorTrainCount = 20 @@ -589,9 +723,17 @@ func trainingRunJob(root string, count int) { sceneOutput := "" sceneScript := trainingScriptPath("train_scene_model.py") - sceneOut, sceneErr := trainingRunCommand( + sceneOut, sceneErr := trainingRunCommandStreaming( python, sceneScript, + func(line string) bool { + return trainingHandleProgressLine( + line, + 10, + 45, + "CLIP-Scene-Positionsmodell wird trainiert…", + ) + }, "--root", root, ) @@ -650,9 +792,17 @@ func trainingRunJob(root string, count int) { }) detectorScript := trainingScriptPath("train_detector_model.py") - detectorOut, detectorErr := trainingRunCommand( + detectorOut, detectorErr := trainingRunCommandStreaming( python, detectorScript, + func(line string) bool { + return trainingHandleProgressLine( + line, + 60, + 98, + "Object Detector wird trainiert…", + ) + }, "--root", root, "--base", "yolo11n.pt", "--epochs", strconv.Itoa(trainingDetectorEpochs()), @@ -772,23 +922,22 @@ func trainingStatusHandler(w http.ResponseWriter, r *http.Request) { return } - if err := trainingEnsureDetectorDirs(root); err != nil { - trainingWriteError(w, http.StatusInternalServerError, err.Error()) - return - } + job := trainingGetJobStatus() - // Praktisch für kleine Datensätze: - // Wenn genug Train-Daten existieren, aber noch zu wenig Val-Daten, - // werden ein paar Train-Samples nach Val kopiert. - if err := trainingEnsureDetectorValidationSample(root); err != nil { - fmt.Println("⚠️ detector val sample ensure failed:", err) + if !job.Running { + if err := trainingEnsureDetectorDirs(root); err != nil { + trainingWriteError(w, http.StatusInternalServerError, err.Error()) + return + } + + if err := trainingEnsureDetectorValidationSample(root); err != nil { + fmt.Println("⚠️ detector val sample ensure failed:", err) + } } feedbackPath := filepath.Join(root, "feedback.jsonl") feedbackCount, _ := trainingCountAnnotations(feedbackPath) - job := trainingGetJobStatus() - detectorDatasetYAML := filepath.Join(root, "detector", "dataset", "dataset.yaml") detectorTrainImages := filepath.Join(root, "detector", "dataset", "images", "train") detectorTrainLabels := filepath.Join(root, "detector", "dataset", "labels", "train") diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 61ed1a5..7756ae8 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -4075,7 +4075,7 @@ export default function App() { ) : null} {selectedTab === 'training' ? ( - + ) : null} {selectedTab === 'categories' ? : null} diff --git a/frontend/src/components/ui/TrainingTab.tsx b/frontend/src/components/ui/TrainingTab.tsx index 0ff6739..283a6b1 100644 --- a/frontend/src/components/ui/TrainingTab.tsx +++ b/frontend/src/components/ui/TrainingTab.tsx @@ -141,6 +141,131 @@ type TrainingStats = { } } +type TrainingNoticeKind = 'success' | 'error' | 'info' | 'warning' + +type TrainingNotice = { + kind: TrainingNoticeKind + title: string + message: string + detail?: string +} + +function trainingNoticeClass(kind: TrainingNoticeKind) { + switch (kind) { + case 'success': + return { + wrap: 'border-emerald-200 bg-emerald-50 text-emerald-900 dark:border-emerald-400/30 dark:bg-emerald-500/10 dark:text-emerald-100', + icon: 'bg-emerald-500 text-white', + detail: 'text-emerald-800/80 dark:text-emerald-100/70', + } + case 'error': + return { + wrap: 'border-red-200 bg-red-50 text-red-900 dark:border-red-400/30 dark:bg-red-500/10 dark:text-red-100', + icon: 'bg-red-500 text-white', + detail: 'text-red-800/80 dark:text-red-100/70', + } + case 'warning': + return { + wrap: 'border-amber-200 bg-amber-50 text-amber-900 dark:border-amber-400/30 dark:bg-amber-500/10 dark:text-amber-100', + icon: 'bg-amber-500 text-white', + detail: 'text-amber-800/80 dark:text-amber-100/70', + } + default: + return { + wrap: 'border-indigo-200 bg-indigo-50 text-indigo-900 dark:border-indigo-400/30 dark:bg-indigo-500/10 dark:text-indigo-100', + icon: 'bg-indigo-500 text-white', + detail: 'text-indigo-800/80 dark:text-indigo-100/70', + } + } +} + +function TrainingNoticeCard(props: { + notice: TrainingNotice + onClose?: () => void +}) { + const cls = trainingNoticeClass(props.notice.kind) + + const icon = + props.notice.kind === 'success' + ? '✓' + : props.notice.kind === 'error' + ? '!' + : props.notice.kind === 'warning' + ? '⚠' + : 'i' + + return ( +
+
+ + +
+
+ {props.notice.title} +
+ +
+ {props.notice.message} +
+ + {props.notice.detail ? ( +
+ + Details anzeigen + + + 
+                {props.notice.detail}
+
+
+ ) : null} +
+ + {props.onClose ? ( + + ) : null} +
+
+ ) +} + +function backendText(data: any, fallback: string) { + return String( + data?.message || + data?.error || + data?.detail || + fallback + ).trim() +} + function countPercent(count: number, total: number) { if (!Number.isFinite(count) || !Number.isFinite(total) || total <= 0) return '0%' return `${Math.round((count / total) * 100)}%` @@ -1647,7 +1772,7 @@ function TrainingStatsModal(props: {
- Gesamt-Confidence + Daten-Confidence
@@ -1673,7 +1798,7 @@ function TrainingStatsModal(props: {
- Grober Wert aus Feedback-Menge, Boxen, Label-Abdeckung und Korrekturanteil. + Daten-Confidence aus Feedback-Menge, Boxen, Label-Abdeckung und Korrekturanteil. Kein direkter Modell-Qualitätswert.
@@ -1736,7 +1861,9 @@ function TrainingStatsModal(props: { ) } -export default function TrainingTab() { +export default function TrainingTab(props: { + onTrainingRunningChange?: (running: boolean) => void +}) { const [labels, setLabels] = useState(emptyLabels) const [sample, setSample] = useState(null) const [correction, setCorrection] = useState(() => predictionToCorrection(null)) @@ -1882,10 +2009,14 @@ export default function TrainingTab() { const loadNext = useCallback(async (opts?: { forceNew?: boolean refreshPrediction?: boolean + preserveNotice?: boolean }) => { setLoading(true) - setError(null) - setMessage(null) + + if (!opts?.preserveNotice) { + setError(null) + setMessage(null) + } try { const params = new URLSearchParams() @@ -2023,6 +2154,12 @@ export default function TrainingTab() { void loadTrainingStats() }, [statsModalOpen, loadTrainingStats]) + const onTrainingRunningChange = props.onTrainingRunningChange + + useEffect(() => { + onTrainingRunningChange?.(trainingRunning) + }, [trainingRunning, onTrainingRunningChange]) + useEffect(() => { if (!boxLabel) return @@ -2052,7 +2189,7 @@ export default function TrainingTab() { const wasRunning = wasTrainingRunningRef.current if (wasRunning && !trainingRunning && trainingStatus?.training?.finishedAt) { - void loadNext({ refreshPrediction: true }) + void loadNext({ refreshPrediction: true, preserveNotice: true }) } wasTrainingRunningRef.current = trainingRunning @@ -2125,37 +2262,16 @@ export default function TrainingTab() { return } - setTrainingProgress((prev) => (prev > 0 ? prev : 8)) - setTrainingStep((prev) => prev || 'Training wird vorbereitet…') + const serverProgress = Number(trainingStatus?.training?.progress ?? 0) + const serverStep = String(trainingStatus?.training?.step ?? '') - const startedAt = Date.now() - - const timer = window.setInterval(() => { - const elapsed = Date.now() - startedAt - - setTrainingProgress((prev) => { - const serverProgress = trainingStatus?.training?.progress - if (typeof serverProgress === 'number' && serverProgress > prev) { - return serverProgress - } - - if (elapsed > 90_000) { - setTrainingStep('Detector wird trainiert…') - return Math.min(prev + 0.4, 92) - } - - if (elapsed > 25_000) { - setTrainingStep('Detector wird trainiert…') - return Math.min(prev + 0.8, 80) - } - - setTrainingStep('Trainingsdaten werden verarbeitet…') - return Math.min(prev + 1.2, 55) - }) - }, 700) - - return () => window.clearInterval(timer) - }, [trainingRunning, trainingStatus?.training?.progress]) + setTrainingProgress(Number.isFinite(serverProgress) ? clampPercent(serverProgress) : 0) + setTrainingStep(serverStep || 'Training läuft…') + }, [ + trainingRunning, + trainingStatus?.training?.progress, + trainingStatus?.training?.step, + ]) const saveFeedback = useCallback( async (accepted: boolean) => { @@ -2194,11 +2310,17 @@ export default function TrainingTab() { const data = await res.json().catch(() => null) if (!res.ok) { - throw new Error(data?.error || `HTTP ${res.status}`) + throw new Error(backendText(data, `HTTP ${res.status}`)) } + setMessage( + accepted + ? 'Feedback gespeichert: Prediction wurde als korrekt übernommen.' + : 'Korrektur gespeichert.' + ) + await loadTrainingStatus() - await loadNext() + await loadNext({ preserveNotice: true }) } catch (e) { setError(e instanceof Error ? e.message : String(e)) } finally { @@ -2223,9 +2345,11 @@ export default function TrainingTab() { const data = await res.json().catch(() => null) if (!res.ok) { - throw new Error(data?.error || `HTTP ${res.status}`) + throw new Error(backendText(data, `HTTP ${res.status}`)) } + setMessage(backendText(data, 'Training wurde gestartet.')) + await loadTrainingStatus() // WICHTIG: @@ -2269,10 +2393,10 @@ export default function TrainingTab() { canTrain: false, }) - setMessage(data?.message || 'Alle Trainingsdaten wurden gelöscht.') + setMessage(backendText(data, 'Alle Trainingsdaten wurden gelöscht.')) await loadTrainingStatus() - await loadNext({ forceNew: true }) + await loadNext({ forceNew: true, preserveNotice: true }) } catch (e) { setError(e instanceof Error ? e.message : String(e)) } finally { @@ -2527,6 +2651,38 @@ export default function TrainingTab() { const showImageBoxes = !loading && !trainingRunning + const activeNotice = useMemo(() => { + if (error) { + return { + kind: 'error', + title: 'Aktion fehlgeschlagen', + message: error, + } + } + + if (message) { + const looksPartial = + message.toLowerCase().includes('übersprungen') || + message.toLowerCase().includes('fehlgeschlagen') + + return { + kind: looksPartial ? 'warning' : 'success', + title: looksPartial ? 'Training teilweise abgeschlossen' : 'Erfolg', + message, + } + } + + if (trainingRunning) { + return { + kind: 'info', + title: 'Training läuft', + message: shownTrainingStep || 'Training läuft im Hintergrund.', + } + } + + return null + }, [error, message, trainingRunning, shownTrainingStep]) + const detectorBoxesPanel = (
@@ -2668,6 +2824,22 @@ export default function TrainingTab() { return ( <> + {activeNotice ? ( +
+ { + setError(null) + setMessage(null) + } + } + /> +
+ ) : null} +
{/* Sidebar links */}
- - {message ? ( -
- {message} -
- ) : null} - - {error ? ( -
- {error} -
- ) : null} {/* Mitte */}