Skip to content

fix: WAV 缓存 404 不再导致整次生成失败 #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
genz27 merged 1 commit into from
Jul 30, 2026
Merged

fix: WAV 缓存 404 不再导致整次生成失败 #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
110 changes: 110 additions & 0 deletions internal/service/flowmusic_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
199 changes: 103 additions & 96 deletions internal/service/generation.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand All @@ -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)
Expand Down
Loading