diff options
Diffstat (limited to 'server/cmd')
| -rw-r--r-- | server/cmd/listend/main.go | 155 |
1 files changed, 151 insertions, 4 deletions
diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go index 495ed03..1b746f1 100644 --- a/server/cmd/listend/main.go +++ b/server/cmd/listend/main.go @@ -23,6 +23,7 @@ import ( "os" "os/signal" "path/filepath" + "sort" "sync" "syscall" "time" @@ -35,6 +36,7 @@ import ( "github.com/terst/listen/server/internal/signature" "github.com/terst/listen/server/internal/skips" "github.com/terst/listen/server/internal/store" + "github.com/terst/listen/server/internal/transcribe" "github.com/terst/listen/server/internal/transcript" ) @@ -46,6 +48,12 @@ func main() { pollEvery = flag.Duration("poll", 2*time.Hour, "how often to refresh every feed") tokenFile = flag.String("token-file", "", "file holding the API token (defaults to <data>/token)") once = flag.Bool("once", false, "run one poll cycle and exit, for checking a deployment") + + whisperBin = flag.String("whisper", "", "whisper-cli binary; empty disables recognition") + whisperModel = flag.String("whisper-model", "", "ggml model file for recognition") + whisperThreads = flag.Int("whisper-threads", 6, "cores one recognition run may use") + transcribeBudget = flag.Duration("transcribe-budget", 45*time.Minute, + "how long each cycle may spend recognising episodes") ) flag.Parse() @@ -83,8 +91,17 @@ func main() { decoder: decoder, transcripts: transcripts, harvester: transcript.Harvester{Store: transcripts}, - requests: make(chan request, 64), - wakeups: make(chan struct{}, 1), + recogniser: transcribe.Whisper{ + Binary: *whisperBin, + Model: *whisperModel, + Threads: *whisperThreads, + // Beside the audio rather than in /tmp: a decoded hour is 115MB of PCM, and the + // data directory is the volume with room for it. + WorkDir: filepath.Join(*dataDir, "work"), + }, + transcribeBudget: *transcribeBudget, + requests: make(chan request, 64), + wakeups: make(chan struct{}, 1), } server := api.New(state, worker, api.Options{ @@ -113,6 +130,9 @@ func main() { if *once { worker.pollAll(ctx) + // Recognition too, so a deployment check exercises the expensive path rather than + // discovering it is misconfigured hours later in a log nobody is reading. + worker.transcribeSome(ctx) if err := state.Save(statePath); err != nil { log.Fatalf("save: %v", err) } @@ -167,8 +187,14 @@ type worker struct { // they describe the copy that was mirrored. See transcript.Fits for when they do not. transcripts transcript.Store harvester transcript.Harvester - requests chan request - wakeups chan struct{} + // recogniser turns an episode's audio into its words where no usable published transcript + // exists. Zero when recognition is not configured, which is a working state. + recogniser transcribe.Whisper + // transcribeBudget is how long one cycle may spend recognising. Bounded so the backlog is + // worked through steadily rather than monopolising the machine -- see transcribeSome. + transcribeBudget time.Duration + requests chan request + wakeups chan struct{} } // wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll. @@ -192,6 +218,7 @@ func (w *worker) Request(feedURL, guid, enclosureURL, title string) { func (w *worker) run(ctx context.Context, every time.Duration) { w.pollAll(ctx) w.save() + w.transcribeSome(ctx) ticker := time.NewTicker(every) defer ticker.Stop() @@ -202,6 +229,10 @@ func (w *worker) run(ctx context.Context, every time.Duration) { case <-ticker.C: w.pollAll(ctx) w.save() + // Only on the timer. A wakeup is the phone pushing a subscription list and + // wanting that show polled now; following it with three quarters of an hour of + // recognition would make a new subscription feel broken. + w.transcribeSome(ctx) case <-w.wakeups: w.pollAll(ctx) w.save() @@ -387,6 +418,122 @@ func (w *worker) harvestTranscripts(ctx context.Context, show *store.Show, candi } } +// transcribeSome recognises the words in episodes that have none, for as long as the budget +// allows. +// +// Bounded rather than exhaustive, and run after polling rather than inside it. Recognition is +// the most expensive thing this server does -- measured at about 11x realtime with base.en on +// six threads, so an hour-long episode costs about five minutes -- and the backlog at the time +// this was built was 304 episodes. Spending a fixed slice of each cycle works through that in a +// few days without ever making a poll wait hours or taking the machine over. +// +// Newest first. An episode Peter is likely to play soon is worth having words for before one +// from the back of the catalogue. +func (w *worker) transcribeSome(ctx context.Context) { + if !w.recogniser.Available() || w.transcribeBudget <= 0 { + return + } + deadline := time.Now().Add(w.transcribeBudget) + done, failed := 0, 0 + // Attempts made during the current run of failures, so they can be given back if the run + // turns out to be the engine rather than the episodes. See transcribe.Breaker. + var breaker transcribe.Breaker + var failing []*store.Episode + + for _, e := range w.pendingRecognition() { + if ctx.Err() != nil || time.Now().After(deadline) { + break + } + started := time.Now() + result, err := w.recogniser.Transcribe(ctx, filepath.Join(w.mirror.AudioDir, e.Path)) + if ctx.Err() != nil { + // Interrupted rather than attempted. Recording it would quietly skip whatever was + // in flight across a redeploy, and this runs for days at a time. + return + } + e.TranscribeAttemptedAtUnix = time.Now().Unix() + if err != nil { + log.Printf("transcribe %q: %v", e.Title, err) + failed++ + failing = append(failing, e) + if breaker.Failure() { + giveBackAttempts(failing) + log.Printf("recognition abandoned after %d failures in a row -- "+ + "giving those attempts back, since this is the engine and not the episodes", + breaker.Run()) + w.save() + return + } + w.save() + continue + } + + stored := transcript.Transcript{ + EpisodeID: e.ID, + Source: transcript.SourceASR, + LanguageCode: result.LanguageCode, + Cues: result.Cues, + Ads: transcript.Spot(result.Cues), + } + if err := w.transcripts.Put(stored); err != nil { + log.Printf("transcribe %q: %v", e.Title, err) + failed++ + failing = append(failing, e) + if breaker.Failure() { + giveBackAttempts(failing) + log.Printf("recognition abandoned after %d failures in a row -- "+ + "giving those attempts back, since this is the engine and not the episodes", + breaker.Run()) + w.save() + return + } + w.save() + continue + } + e.TranscriptSource = string(transcript.SourceASR) + e.TranscriptCues = len(stored.Cues) + e.TranscriptAtUnix = time.Now().Unix() + // The note said why the publisher's transcript could not be used. There are words now, + // so it is answered rather than still true. + e.TranscriptNote = "" + done++ + breaker.Success() + failing = nil + log.Printf("transcribed %q: %d lines, %d ad candidate(s), took %s", + e.Title, len(stored.Cues), len(stored.Ads), time.Since(started).Round(time.Second)) + w.save() + } + + if done > 0 || failed > 0 { + log.Printf("recognition -- %d done, %d failed, %d still waiting", + done, failed, len(w.pendingRecognition())) + } +} + +// giveBackAttempts unmarks episodes that failed for a reason that was nothing to do with them, +// so they are tried again rather than hidden for ever. +func giveBackAttempts(episodes []*store.Episode) { + for _, e := range episodes { + e.TranscribeAttemptedAtUnix = 0 + } +} + +// pendingRecognition is every mirrored episode still without words, newest first. +func (w *worker) pendingRecognition() []*store.Episode { + var pending []*store.Episode + for _, show := range w.state.Shows() { + for _, e := range show.Episodes { + if e.NeedsRecognising() { + pending = append(pending, e) + } + } + } + sort.Slice(pending, func(i, j int) bool { + return pending[i].PublishedUnix > pending[j].PublishedUnix + }) + return pending +} + // detect runs the intro/outro search over a show's mirrored episodes and records the result. // // Every mirrored episode at once, which is the whole reason this is a server: the phone could |