473 lines
18 KiB
Go
473 lines
18 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"os"
|
|
"path/filepath"
|
|
"slices"
|
|
"strings"
|
|
"testing"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/config"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
|
|
"gitea.maximumdirect.net/eric/narratio/internal/stage"
|
|
)
|
|
|
|
func TestExecuteStagesDelegatesEmptyAudioAcceptanceToWhisperX(t *testing.T) {
|
|
for _, rejected := range []bool{false, true} {
|
|
name := "accepted"
|
|
if rejected {
|
|
name = "rejected"
|
|
}
|
|
t.Run(name, func(t *testing.T) {
|
|
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice"})
|
|
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
|
if err := os.WriteFile(filepath.Join(paths.AudioDir, "alice.flac"), nil, 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
fake := &whisperx.FakeClient{}
|
|
if rejected {
|
|
fake.Err = errors.New("service rejected audio")
|
|
}
|
|
transcribe := mustSelectStage(t, "transcribe")
|
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: fake},
|
|
})
|
|
if !errors.Is(err, fake.Err) {
|
|
t.Fatalf("transcribe error = %v, want %v", err, fake.Err)
|
|
}
|
|
if len(fake.RequestsSnapshot()) != 1 {
|
|
t.Fatal("empty audio was not sent to service")
|
|
}
|
|
m := loadTranscribeRecoveryManifest(t, store, cfg)
|
|
if rejected {
|
|
if len(m.Stages["transcribe"].TranscribeSpeakers) != 0 {
|
|
t.Fatal("rejected audio was checkpointed")
|
|
}
|
|
return
|
|
}
|
|
record, ok := m.Stages["transcribe"].TranscribeSpeakers["alice"]
|
|
if !ok || record.AudioSize != 0 {
|
|
t.Fatalf("missing zero-size checkpoint: %#v", record)
|
|
}
|
|
retry := &whisperx.FakeClient{}
|
|
summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: retry},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(summary.Skipped) != 1 || len(retry.RequestsSnapshot()) != 0 {
|
|
t.Fatal("successful empty-audio result was not reused")
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestExecuteStagesRetriesOnlyUncommittedTranscribeSpeakers(t *testing.T) {
|
|
cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"a", "b", "c", "d"})
|
|
concurrency := 1
|
|
cfg.Pipeline.WhisperX.Concurrency = &concurrency
|
|
cause := errors.New("speaker c failed")
|
|
failing := &whisperx.FakeClient{TranscribeFn: func(
|
|
_ context.Context,
|
|
req whisperx.TranscribeRequest,
|
|
) (whisperx.TranscribeResult, error) {
|
|
if req.SpeakerID == "c" {
|
|
return whisperx.TranscribeResult{}, cause
|
|
}
|
|
if req.SpeakerID == "d" {
|
|
t.Fatalf("speaker d was dispatched after c failed")
|
|
}
|
|
return writeTranscribeRecoveryResult(req)
|
|
}}
|
|
transcribe := mustSelectStage(t, "transcribe")
|
|
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: baseStore, WhisperX: failing},
|
|
}); !errors.Is(err, cause) {
|
|
t.Fatalf("failed executeStages() error = %v, want %v", err, cause)
|
|
}
|
|
failed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
|
|
if failed.Stages["transcribe"].Status != manifest.StatusFailed {
|
|
t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status)
|
|
}
|
|
if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"a", "b"}) {
|
|
t.Fatalf("checkpointed speakers = %#v, want a and b", got)
|
|
}
|
|
if failed.Stages["prepare"].Status != manifest.StatusSucceeded {
|
|
t.Fatalf("prepare status = %q, want unchanged success", failed.Stages["prepare"].Status)
|
|
}
|
|
if failed.Stages["merge"].Status != manifest.StatusStale {
|
|
t.Fatalf("merge status = %q, want stale after transcribe failure", failed.Stages["merge"].Status)
|
|
}
|
|
|
|
retry := &whisperx.FakeClient{}
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: baseStore, WhisperX: retry},
|
|
}); err != nil {
|
|
t.Fatalf("retry executeStages() error = %v", err)
|
|
}
|
|
if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"c", "d"}) {
|
|
t.Fatalf("retry speakers = %#v, want c and d", got)
|
|
}
|
|
completed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
|
|
if completed.Stages["transcribe"].Status != manifest.StatusSucceeded {
|
|
t.Fatalf("transcribe status = %q, want succeeded", completed.Stages["transcribe"].Status)
|
|
}
|
|
if got := transcribeSpeakerKeys(completed); !slices.Equal(got, []string{"a", "b", "c", "d"}) {
|
|
t.Fatalf("completed speakers = %#v", got)
|
|
}
|
|
}
|
|
|
|
func TestExecuteStagesRebuildsTranscribeAggregateAfterFinalSaveFailure(t *testing.T) {
|
|
cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
|
|
store := &transcribeAggregateSaveFailingStore{
|
|
Store: baseStore,
|
|
Err: errors.New("aggregate save failed"),
|
|
}
|
|
fake := &whisperx.FakeClient{}
|
|
transcribe := mustSelectStage(t, "transcribe")
|
|
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: fake},
|
|
}); !errors.Is(err, store.Err) {
|
|
t.Fatalf("executeStages() error = %v, want aggregate save failure", err)
|
|
}
|
|
failed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
|
|
if failed.Stages["transcribe"].Status != manifest.StatusFailed {
|
|
t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status)
|
|
}
|
|
if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice", "bob"}) {
|
|
t.Fatalf("checkpointed speakers = %#v, want both speakers", got)
|
|
}
|
|
|
|
retry := &whisperx.FakeClient{}
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: baseStore, WhisperX: retry},
|
|
}); err != nil {
|
|
t.Fatalf("retry executeStages() error = %v", err)
|
|
}
|
|
if got := len(retry.RequestsSnapshot()); got != 0 {
|
|
t.Fatalf("retry adapter requests = %d, want aggregate-only rebuild", got)
|
|
}
|
|
completed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
|
|
if completed.Stages["transcribe"].Status != manifest.StatusSucceeded || len(completed.Stages["transcribe"].Outputs) != 2 {
|
|
t.Fatalf("rebuilt aggregate = %#v", completed.Stages["transcribe"])
|
|
}
|
|
}
|
|
|
|
func TestExecuteStagesRepairsSucceededTranscribeAggregateWithoutRequests(t *testing.T) {
|
|
for _, test := range []struct {
|
|
name string
|
|
mutate func(*manifest.Manifest)
|
|
removed bool
|
|
}{
|
|
{name: "removed speaker", removed: true, mutate: func(m *manifest.Manifest) {
|
|
m.Inputs = slices.DeleteFunc(m.Inputs, func(in manifest.InputRecord) bool { return filepath.Base(in.Path) == "bob.flac" })
|
|
}},
|
|
{name: "missing output", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs = m.Stages["transcribe"].Outputs[:1] }},
|
|
{name: "duplicate output", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[1] = m.Stages["transcribe"].Outputs[0] }},
|
|
{name: "wrong path", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].LocalPath = "transcripts/raw/other.json" }},
|
|
{name: "wrong checksum", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].Checksum = "obsolete" }},
|
|
{name: "wrong producer", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].ProducerRunID = "other-run" }},
|
|
} {
|
|
t.Run(test.name, func(t *testing.T) {
|
|
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
|
|
transcribe := mustSelectStage(t, "transcribe")
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: &whisperx.FakeClient{}},
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
m := loadTranscribeRecoveryManifest(t, store, cfg)
|
|
test.mutate(m)
|
|
m.MarkStageSucceeded("merge", time.Now().UTC(), nil)
|
|
if err := store.Save(context.Background(), manifestPathFor(cfg), m); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
retry := &whisperx.FakeClient{}
|
|
summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: retry},
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(summary.Skipped) != 0 || len(retry.RequestsSnapshot()) != 0 {
|
|
t.Fatalf("want aggregate repair without HTTP; summary=%#v requests=%#v", summary, retry.RequestsSnapshot())
|
|
}
|
|
completed := loadTranscribeRecoveryManifest(t, store, cfg)
|
|
want := []string{"alice", "bob"}
|
|
if test.removed {
|
|
want = []string{"alice"}
|
|
}
|
|
record := completed.Stages["transcribe"]
|
|
if !slices.Equal(transcribeSpeakerKeys(completed), want) || len(record.Outputs) != len(want) {
|
|
t.Fatalf("unexpected repaired state: %#v", record)
|
|
}
|
|
for _, output := range record.Outputs {
|
|
speaker := record.TranscribeSpeakers[strings.TrimSuffix(filepath.Base(output.LocalPath), ".json")]
|
|
if output.Checksum != speaker.Output.Checksum || output.ProducerRunID != speaker.ProducerRunID {
|
|
t.Fatalf("aggregate evidence does not match checkpoint: %#v", output)
|
|
}
|
|
}
|
|
if completed.Stages["merge"].Status != manifest.StatusStale {
|
|
t.Fatal("aggregate repair did not stale merge")
|
|
}
|
|
if test.removed {
|
|
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
|
for _, name := range []string{"speakers.yml", "autocorrect.yml"} {
|
|
if err := os.WriteFile(filepath.Join(paths.InputsDir, name), []byte("{}"), 0o644); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
merge := &seriatim.FakeRunner{}
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{mustSelectStage(t, "merge")}, RunOptions{
|
|
Env: &Env{ManifestStore: store, Seriatim: merge},
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(merge.NormalizeRequests) != 1 || filepath.Base(merge.NormalizeRequests[0].InputTranscriptPath) != "alice.json" {
|
|
t.Fatalf("removed speaker reached merge: %#v", merge.NormalizeRequests)
|
|
}
|
|
if _, err := os.Stat(filepath.Join(paths.TranscriptsRawDir, "bob.json")); err != nil {
|
|
t.Fatalf("incidental transcript was removed: %v", err)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
func TestExecuteStagesResumesInterruptedRunningTranscribeState(t *testing.T) {
|
|
cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
|
|
concurrency := 1
|
|
cfg.Pipeline.WhisperX.Concurrency = &concurrency
|
|
stageErr := errors.New("speaker b failed")
|
|
terminalErr := errors.New("terminal save interrupted")
|
|
store := &interruptAfterTranscribeCheckpointStore{Store: baseStore, Err: terminalErr}
|
|
fake := &whisperx.FakeClient{TranscribeFn: func(
|
|
_ context.Context,
|
|
req whisperx.TranscribeRequest,
|
|
) (whisperx.TranscribeResult, error) {
|
|
if req.SpeakerID == "bob" {
|
|
return whisperx.TranscribeResult{}, stageErr
|
|
}
|
|
return writeTranscribeRecoveryResult(req)
|
|
}}
|
|
transcribe := mustSelectStage(t, "transcribe")
|
|
|
|
_, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: fake},
|
|
})
|
|
if !errors.Is(err, stageErr) || !errors.Is(err, terminalErr) {
|
|
t.Fatalf("executeStages() error = %v, want stage and terminal failures", err)
|
|
}
|
|
interrupted := loadTranscribeRecoveryManifest(t, baseStore, cfg)
|
|
if interrupted.Stages["transcribe"].Status != manifest.StatusRunning {
|
|
t.Fatalf("transcribe status = %q, want durable running state", interrupted.Stages["transcribe"].Status)
|
|
}
|
|
if got := transcribeSpeakerKeys(interrupted); !slices.Equal(got, []string{"alice"}) {
|
|
t.Fatalf("checkpointed speakers = %#v, want alice", got)
|
|
}
|
|
|
|
retry := &whisperx.FakeClient{}
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: baseStore, WhisperX: retry},
|
|
}); err != nil {
|
|
t.Fatalf("retry executeStages() error = %v", err)
|
|
}
|
|
if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) {
|
|
t.Fatalf("retry speakers = %#v, want bob", got)
|
|
}
|
|
}
|
|
|
|
func TestExecuteStagesPersistsForcedTranscribeResetAcrossRetry(t *testing.T) {
|
|
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
|
|
concurrency := 1
|
|
cfg.Pipeline.WhisperX.Concurrency = &concurrency
|
|
transcribe := mustSelectStage(t, "transcribe")
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: &whisperx.FakeClient{}},
|
|
}); err != nil {
|
|
t.Fatalf("baseline executeStages() error = %v", err)
|
|
}
|
|
|
|
cause := errors.New("forced bob failed")
|
|
forced := &whisperx.FakeClient{TranscribeFn: func(
|
|
_ context.Context,
|
|
req whisperx.TranscribeRequest,
|
|
) (whisperx.TranscribeResult, error) {
|
|
if req.SpeakerID == "bob" {
|
|
return whisperx.TranscribeResult{}, cause
|
|
}
|
|
return writeTranscribeRecoveryResult(req)
|
|
}}
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Force: true,
|
|
Env: &Env{ManifestStore: store, WhisperX: forced},
|
|
}); !errors.Is(err, cause) {
|
|
t.Fatalf("forced executeStages() error = %v, want %v", err, cause)
|
|
}
|
|
failed := loadTranscribeRecoveryManifest(t, store, cfg)
|
|
if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice"}) {
|
|
t.Fatalf("speakers after forced failure = %#v, want only replacement alice", got)
|
|
}
|
|
|
|
retry := &whisperx.FakeClient{}
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: retry},
|
|
}); err != nil {
|
|
t.Fatalf("retry executeStages() error = %v", err)
|
|
}
|
|
if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) {
|
|
t.Fatalf("retry speakers = %#v, want bob", got)
|
|
}
|
|
}
|
|
|
|
func TestExecuteStagesUpgradesLegacyTranscribeSuccessOnce(t *testing.T) {
|
|
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice"})
|
|
m := loadTranscribeRecoveryManifest(t, store, cfg)
|
|
m.MarkStageSucceeded("transcribe", time.Now().UTC(), []manifest.ArtifactRecord{{
|
|
Kind: "transcript_raw", LocalPath: "transcripts/raw/alice.json",
|
|
}})
|
|
if err := store.Save(context.Background(), manifestPathFor(cfg), m); err != nil {
|
|
t.Fatalf("save legacy transcribe success: %v", err)
|
|
}
|
|
transcribe := mustSelectStage(t, "transcribe")
|
|
upgrade := &whisperx.FakeClient{}
|
|
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: upgrade},
|
|
}); err != nil {
|
|
t.Fatalf("upgrade executeStages() error = %v", err)
|
|
}
|
|
if got := requestedTranscribeSpeakers(upgrade); !slices.Equal(got, []string{"alice"}) {
|
|
t.Fatalf("upgrade speakers = %#v, want alice", got)
|
|
}
|
|
upgraded := loadTranscribeRecoveryManifest(t, store, cfg)
|
|
if upgraded.Stages["transcribe"].TranscribeStateVersion != manifest.TranscribeStateContractVersion {
|
|
t.Fatalf("transcribe state version = %d, want current", upgraded.Stages["transcribe"].TranscribeStateVersion)
|
|
}
|
|
|
|
reused := &whisperx.FakeClient{}
|
|
summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
|
|
Env: &Env{ManifestStore: store, WhisperX: reused},
|
|
})
|
|
if err != nil {
|
|
t.Fatalf("reuse executeStages() error = %v", err)
|
|
}
|
|
if len(summary.Skipped) != 1 || len(reused.RequestsSnapshot()) != 0 {
|
|
t.Fatalf("reuse summary = %#v, requests = %#v, want skipped without HTTP", summary, reused.RequestsSnapshot())
|
|
}
|
|
}
|
|
|
|
type transcribeAggregateSaveFailingStore struct {
|
|
manifest.Store
|
|
Err error
|
|
Failed bool
|
|
}
|
|
|
|
func (store *transcribeAggregateSaveFailingStore) Save(
|
|
ctx context.Context,
|
|
path string,
|
|
m *manifest.Manifest,
|
|
) error {
|
|
record := m.Stages["transcribe"]
|
|
if !store.Failed && record != nil && record.Status == manifest.StatusSucceeded &&
|
|
record.TranscribeStateVersion == manifest.TranscribeStateContractVersion {
|
|
store.Failed = true
|
|
return store.Err
|
|
}
|
|
return store.Store.Save(ctx, path, m)
|
|
}
|
|
|
|
type interruptAfterTranscribeCheckpointStore struct {
|
|
manifest.Store
|
|
Err error
|
|
Interrupted bool
|
|
}
|
|
|
|
func (store *interruptAfterTranscribeCheckpointStore) Save(
|
|
ctx context.Context,
|
|
path string,
|
|
m *manifest.Manifest,
|
|
) error {
|
|
if store.Interrupted {
|
|
return store.Err
|
|
}
|
|
record := m.Stages["transcribe"]
|
|
err := store.Store.Save(ctx, path, m)
|
|
if err == nil && record != nil && record.Status == manifest.StatusRunning && len(record.TranscribeSpeakers) == 1 {
|
|
store.Interrupted = true
|
|
}
|
|
return err
|
|
}
|
|
|
|
func setupTranscribeRecoveryFixture(t *testing.T, speakers []string) (*config.Config, *manifest.LocalStore) {
|
|
t.Helper()
|
|
cfg := testConfig(t)
|
|
store := &manifest.LocalStore{}
|
|
artifactStore := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
|
|
paths, err := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID)
|
|
if err != nil {
|
|
t.Fatalf("prepare session layout: %v", err)
|
|
}
|
|
m := manifest.New(cfg.Session.SessionID, time.Now().UTC())
|
|
m.Campaign = cfg.Session.Campaign
|
|
m.Inputs = make([]manifest.InputRecord, 0, len(speakers))
|
|
for _, speaker := range speakers {
|
|
relative := filepath.ToSlash(filepath.Join("audio", speaker+".flac"))
|
|
if err := os.WriteFile(filepath.Join(paths.AudioDir, speaker+".flac"), []byte("audio-"+speaker), 0o644); err != nil {
|
|
t.Fatalf("write audio fixture: %v", err)
|
|
}
|
|
m.Inputs = append(m.Inputs, manifest.InputRecord{Kind: "audio", Path: relative})
|
|
}
|
|
m.MarkStageSucceeded("prepare", time.Now().UTC(), nil)
|
|
m.MarkStageSucceeded("merge", time.Now().UTC(), nil)
|
|
if err := store.Save(context.Background(), paths.ManifestPath, m); err != nil {
|
|
t.Fatalf("save recovery fixture: %v", err)
|
|
}
|
|
return cfg, store
|
|
}
|
|
|
|
func writeTranscribeRecoveryResult(req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
|
|
if err := os.MkdirAll(filepath.Dir(req.OutputRawTranscriptPath), 0o755); err != nil {
|
|
return whisperx.TranscribeResult{}, err
|
|
}
|
|
if err := os.WriteFile(req.OutputRawTranscriptPath, []byte(`{"segments":[]}`), 0o644); err != nil {
|
|
return whisperx.TranscribeResult{}, err
|
|
}
|
|
return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil
|
|
}
|
|
|
|
func loadTranscribeRecoveryManifest(t *testing.T, store manifest.Store, cfg *config.Config) *manifest.Manifest {
|
|
t.Helper()
|
|
m, err := store.Load(context.Background(), manifestPathFor(cfg))
|
|
if err != nil {
|
|
t.Fatalf("load transcribe recovery manifest: %v", err)
|
|
}
|
|
return m
|
|
}
|
|
|
|
func transcribeSpeakerKeys(m *manifest.Manifest) []string {
|
|
keys := make([]string, 0, len(m.Stages["transcribe"].TranscribeSpeakers))
|
|
for key := range m.Stages["transcribe"].TranscribeSpeakers {
|
|
keys = append(keys, key)
|
|
}
|
|
slices.Sort(keys)
|
|
return keys
|
|
}
|
|
|
|
func requestedTranscribeSpeakers(fake *whisperx.FakeClient) []string {
|
|
requests := fake.RequestsSnapshot()
|
|
speakers := make([]string, 0, len(requests))
|
|
for _, request := range requests {
|
|
speakers = append(speakers, request.SpeakerID)
|
|
}
|
|
return speakers
|
|
}
|