diff --git a/internal/service/flowmusic_test.go b/internal/service/flowmusic_test.go index 7357f30..9178df8 100644 --- a/internal/service/flowmusic_test.go +++ b/internal/service/flowmusic_test.go @@ -2912,6 +2912,116 @@ func TestGenerateReturnsMultipleClips(t *testing.T) { } } +func TestGenerateSoftFailsOptionalWavCache404(t *testing.T) { + ctx := context.Background() + clipID := "11111111-1111-1111-1111-111111111111" + mediaServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/song.mp3": + w.Header().Set("Content-Type", "audio/mpeg") + _, _ = w.Write([]byte("fake-mp3")) + case "/song.wav": + http.NotFound(w, r) + default: + t.Fatalf("unexpected media path: %s", r.URL.Path) + } + })) + t.Cleanup(mediaServer.Close) + + flowServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/__api/conversation": + writeTestJSON(t, w, map[string]string{"job_id": "job-1"}) + case "/__api/messages/job-1/stream": + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte(`data: {"clip_ids":["` + clipID + `"]}` + "\n\n")) + _, _ = w.Write([]byte("data: [DONE]\n\n")) + case "/__api/clips": + writeTestJSON(t, w, map[string]any{ + "clips": map[string]any{ + clipID: map[string]string{ + "id": clipID, + "title": "Soft Wav", + "audio_url": mediaServer.URL + "/song.mp3", + "wav_url": mediaServer.URL + "/song.wav", + }, + }, + }) + default: + t.Fatalf("unexpected FlowMusic path: %s", r.URL.Path) + } + })) + t.Cleanup(flowServer.Close) + + dir := t.TempDir() + cfg := config.Config{ + DataDir: dir, + CacheDir: filepath.Join(dir, "tmp"), + DatabaseDriver: "sqlite", + DatabaseURL: filepath.Join(dir, "flowmusic2api.db"), + FlowMusicBaseURL: flowServer.URL, + UpstreamTimeout: time.Second, + GenerationTimeout: time.Second, + TokenRefreshLead: time.Minute, + TokenRefreshInterval: time.Hour, + DefaultAdminUser: "admin", + DefaultAdminPassword: "admin", + DefaultAPIKey: "test-api-key", + } + db, err := store.New(ctx, cfg) + if err != nil { + t.Fatalf("store.New() error = %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + if err := db.Migrate(ctx); err != nil { + t.Fatalf("Migrate() error = %v", err) + } + if err := db.EnsureDefaults(ctx); err != nil { + t.Fatalf("EnsureDefaults() error = %v", err) + } + if err := db.UpdateCacheConfig(ctx, domain.CacheConfig{ + Enabled: true, + StorageMode: "local", + BaseURL: "https://cdn.example.test", + }); err != nil { + t.Fatalf("UpdateCacheConfig() error = %v", err) + } + if _, err := db.CreateAccount(ctx, domain.Account{ + Email: "soft-wav@example.test", + ProtocolMode: "bearer", + FlowBearer: "flow-bearer", + }); err != nil { + t.Fatalf("CreateAccount() error = %v", err) + } + + flow := NewFlowMusicClient(cfg) + accounts := NewAccountService(cfg, db, flow) + cache := storage.NewCache(cfg, db, NewHTTPClient(cfg, "")) + generation := NewGenerationService(cfg, db, accounts, flow, cache) + + var sawWavFallback bool + out, err := generation.GenerateWithProgress(ctx, "prompt", "lyria", func(progress GenerationProgress) { + if strings.Contains(progress.Message, "WAV 暂不可用") { + sawWavFallback = true + } + }) + if err != nil { + t.Fatalf("GenerateWithProgress() error = %v, want soft-fail success", err) + } + if len(out.Clips) != 1 || out.Clips[0].Wav == nil { + t.Fatalf("unexpected clips: %+v", out.Clips) + } + if out.Clips[0].Wav.URL != mediaServer.URL+"/song.wav" || out.Clips[0].Wav.OriginalURL != mediaServer.URL+"/song.wav" { + t.Fatalf("wav should fall back to original URL: %+v", out.Clips[0].Wav) + } + if !strings.HasPrefix(out.Clips[0].Audio.URL, "https://cdn.example.test/") { + t.Fatalf("audio should still be cached: %+v", out.Clips[0].Audio) + } + if !sawWavFallback { + t.Fatalf("expected WAV fallback progress message") + } +} + func TestGenerateFailsWhenFlowMusicReturnsNoAudioClips(t *testing.T) { ctx := context.Background() clipID := "11111111-1111-1111-1111-111111111111" diff --git a/internal/service/generation.go b/internal/service/generation.go index 417e63a..5c097d9 100644 --- a/internal/service/generation.go +++ b/internal/service/generation.go @@ -234,63 +234,17 @@ func (s *GenerationService) GenerateWithProgress(ctx context.Context, prompt, mo } hasAudio := false for _, clip := range clips { - if clip.AudioURL == "" && clip.WavURL == "" { - continue - } - item := newClipOutput(clip) - if clip.AudioURL != "" { - emit("caching", fmt.Sprintf("缓存音频文件: %s", firstNonEmpty(clip.Title, clip.ID)), 82) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.AudioURL) - }) - if err != nil { - cacheErr := fmt.Errorf("cache audio: %w", err) - s.failLog(ctx, logID, account.ID, reqPayload, start, cacheErr) - s.recordFailure(ctx, account.ID) - return output, cacheErr - } - item.Audio = ref + item, ok, err := s.buildClipOutput(ctx, clip, runtime.MaxAttempts, emit) + if err != nil { + cacheErr := err + s.failLog(ctx, logID, account.ID, reqPayload, start, cacheErr) + s.recordFailure(ctx, account.ID) + return output, cacheErr } - if clip.WavURL != "" { - emit("caching", fmt.Sprintf("缓存 WAV 文件: %s", firstNonEmpty(clip.Title, clip.ID)), 86) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.WavURL) - }) - if err != nil { - cacheErr := fmt.Errorf("cache wav: %w", err) - s.failLog(ctx, logID, account.ID, reqPayload, start, cacheErr) - s.recordFailure(ctx, account.ID) - return output, cacheErr - } - item.Wav = &ref + if !ok { + continue } hasAudio = true - if clip.ImageURL != "" { - emit("caching", fmt.Sprintf("缓存封面: %s", firstNonEmpty(clip.Title, clip.ID)), 90) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.ImageURL) - }) - if err != nil { - cacheErr := fmt.Errorf("cache image: %w", err) - s.failLog(ctx, logID, account.ID, reqPayload, start, cacheErr) - s.recordFailure(ctx, account.ID) - return output, cacheErr - } - item.Image = &ref - } - if clip.VideoURL != "" { - emit("caching", fmt.Sprintf("缓存视频: %s", firstNonEmpty(clip.Title, clip.ID)), 92) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.VideoURL) - }) - if err != nil { - cacheErr := fmt.Errorf("cache video: %w", err) - s.failLog(ctx, logID, account.ID, reqPayload, start, cacheErr) - s.recordFailure(ctx, account.ID) - return output, cacheErr - } - item.Video = &ref - } output.Clips = append(output.Clips, item) } if !hasAudio { @@ -395,51 +349,14 @@ func (s *GenerationService) LookupResult(ctx context.Context, lookup GenerationR } hasAudio := false for _, clip := range clips { - if clip.AudioURL == "" && clip.WavURL == "" { - continue - } - item := newClipOutput(clip) - if clip.AudioURL != "" { - emit("caching", fmt.Sprintf("缓存音频文件: %s", firstNonEmpty(clip.Title, clip.ID)), 82) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.AudioURL) - }) - if err != nil { - return output, fmt.Errorf("cache audio: %w", err) - } - item.Audio = ref + item, ok, err := s.buildClipOutput(ctx, clip, runtime.MaxAttempts, emit) + if err != nil { + return output, err } - if clip.WavURL != "" { - emit("caching", fmt.Sprintf("缓存 WAV 文件: %s", firstNonEmpty(clip.Title, clip.ID)), 86) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.WavURL) - }) - if err != nil { - return output, fmt.Errorf("cache wav: %w", err) - } - item.Wav = &ref + if !ok { + continue } hasAudio = true - if clip.ImageURL != "" { - emit("caching", fmt.Sprintf("缓存封面: %s", firstNonEmpty(clip.Title, clip.ID)), 90) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.ImageURL) - }) - if err != nil { - return output, fmt.Errorf("cache image: %w", err) - } - item.Image = &ref - } - if clip.VideoURL != "" { - emit("caching", fmt.Sprintf("缓存视频: %s", firstNonEmpty(clip.Title, clip.ID)), 92) - ref, err := retryValue(ctx, runtime.MaxAttempts, func() (domain.MediaRef, error) { - return s.cache.CacheURL(ctx, clip.VideoURL) - }) - if err != nil { - return output, fmt.Errorf("cache video: %w", err) - } - item.Video = &ref - } output.Clips = append(output.Clips, item) } if !hasAudio { @@ -463,6 +380,96 @@ func newClipOutput(clip ClipResult) ClipOutput { } } +type progressEmitter func(stage, message string, progress int) + +// buildClipOutput caches media for one clip. +// Primary audio cache failures are hard errors; wav/image/video soft-fail to original URLs +// because FlowMusic often returns wav_url before the object is actually available (HTTP 404). +func (s *GenerationService) buildClipOutput(ctx context.Context, clip ClipResult, attempts int, emit progressEmitter) (ClipOutput, bool, error) { + if clip.AudioURL == "" && clip.WavURL == "" { + return ClipOutput{}, false, nil + } + item := newClipOutput(clip) + label := firstNonEmpty(clip.Title, clip.ID) + if clip.AudioURL != "" { + if emit != nil { + emit("caching", fmt.Sprintf("缓存音频文件: %s", label), 82) + } + ref, err := s.cacheRequiredURL(ctx, attempts, clip.AudioURL) + if err != nil { + return ClipOutput{}, false, fmt.Errorf("cache audio: %w", err) + } + item.Audio = ref + } + if clip.WavURL != "" { + if emit != nil { + emit("caching", fmt.Sprintf("缓存 WAV 文件: %s", label), 86) + } + ref, fallback, err := s.cacheOptionalURL(ctx, attempts, clip.WavURL) + if err != nil { + return ClipOutput{}, false, fmt.Errorf("cache wav: %w", err) + } + item.Wav = &ref + if fallback && emit != nil { + emit("caching", fmt.Sprintf("WAV 暂不可用，已回退原链接: %s", label), 86) + } + } + if clip.ImageURL != "" { + if emit != nil { + emit("caching", fmt.Sprintf("缓存封面: %s", label), 90) + } + ref, fallback, err := s.cacheOptionalURL(ctx, attempts, clip.ImageURL) + if err != nil { + return ClipOutput{}, false, fmt.Errorf("cache image: %w", err) + } + item.Image = &ref + if fallback && emit != nil { + emit("caching", fmt.Sprintf("封面缓存失败，已回退原链接: %s", label), 90) + } + } + if clip.VideoURL != "" { + if emit != nil { + emit("caching", fmt.Sprintf("缓存视频: %s", label), 92) + } + ref, fallback, err := s.cacheOptionalURL(ctx, attempts, clip.VideoURL) + if err != nil { + return ClipOutput{}, false, fmt.Errorf("cache video: %w", err) + } + item.Video = &ref + if fallback && emit != nil { + emit("caching", fmt.Sprintf("视频缓存失败，已回退原链接: %s", label), 92) + } + } + return item, true, nil +} + +func originalMediaRef(sourceURL string) domain.MediaRef { + sourceURL = strings.TrimSpace(sourceURL) + return domain.MediaRef{OriginalURL: sourceURL, URL: sourceURL} +} + +func (s *GenerationService) cacheRequiredURL(ctx context.Context, attempts int, sourceURL string) (domain.MediaRef, error) { + return retryValue(ctx, attempts, func() (domain.MediaRef, error) { + return s.cache.CacheURL(ctx, sourceURL) + }) +} + +// cacheOptionalURL caches when possible; on download/cache errors returns the original URL +// so optional assets (wav/image/video) never fail an otherwise successful generation. +func (s *GenerationService) cacheOptionalURL(ctx context.Context, attempts int, sourceURL string) (domain.MediaRef, bool, error) { + sourceURL = strings.TrimSpace(sourceURL) + if sourceURL == "" { + return domain.MediaRef{}, false, nil + } + ref, err := retryValue(ctx, attempts, func() (domain.MediaRef, error) { + return s.cache.CacheURL(ctx, sourceURL) + }) + if err != nil { + return originalMediaRef(sourceURL), true, nil + } + return ref, false, nil +} + func (s *GenerationService) lookupAccount(ctx context.Context, accountID int64, leaseTTL time.Duration) (*domain.Account, func(), error) { if accountID > 0 { account, err := s.db.GetAccount(ctx, accountID)