diff options
| author | Peter Stone <thepeterstone@gmail.com> | 2026-10-31 09:16:45 +0000 |
|---|---|---|
| committer | Peter Stone <thepeterstone@gmail.com> | 2026-10-31 09:16:45 +0000 |
| commit | ff40abd84236288cc0a82226460f8815ad31b76d (patch) | |
| tree | 429f31cdf2fe4b97cef3fdb4bfe53e84930add50 /server | |
| parent | cb3377b4856e3419df1cbc52044eda8bf68a006d (diff) | |
Deep clean, part one: get the pipeline out of the command
cmd/listend/main.go had reached 1,249 lines carrying nine passes, and cmd has no
test file -- so every judgement in the transcript pipeline lived where nothing
could reach it. That is not hypothetical: a refetch master blocked the recurring
detector for a fortnight because the condition was wrong in a place no test
could see.
internal/words now holds the five passes with judgement in them -- Harvest,
Recognise, Rederive, Recurring, SecondCopy -- behind a Pipeline that names its
dependencies. main.go is 631 lines and does what a command should: read flags,
wire things together, serve.
**Moving it found a latent bug.** Pruning an episode's audio cleared its
transcript fields but not its master ones, so a re-mirrored episode would claim
a master whose file had been deleted -- a second description it could not
produce, and subtraction finding nothing for it for ever. Exactly the shape of
the bug that had just cost a fortnight. Pipeline.Forget now clears all of it,
including the attempt timestamps, so a re-mirrored episode is done again rather
than skipped for ever. Pinned by a test that asserts it needs recognising
afterwards.
Also in this part: Gradle's file-system watching is off. Chaquopy extracts a
3,200-file Python environment into app/build mid-build and the watcher tripped
over re-registering it, printing "Caught exception: Already watching path" in
front of every build error this project produced all week. It costs nothing to
turn off -- measured twice each on a no-op compileDebugKotlin, 1.50s and 1.59s
with, 1.46s and 1.46s without -- because that extraction invalidates the VFS
anyway. A watcher holding stale state about a tree another tool rewrites causes
the "clean and try again" class of problem rather than curing it.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018AwnALZL6bD4VBtATDVv3B
Diffstat (limited to 'server')
| -rw-r--r-- | server/cmd/listend/main.go | 696 | ||||
| -rw-r--r-- | server/internal/words/pipeline.go | 706 | ||||
| -rw-r--r-- | server/internal/words/pipeline_test.go | 137 |
3 files changed, 882 insertions, 657 deletions
diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go index 43921bd..e1329a9 100644 --- a/server/cmd/listend/main.go +++ b/server/cmd/listend/main.go @@ -23,7 +23,6 @@ import ( "os" "os/signal" "path/filepath" - "sort" "sync" "syscall" "time" @@ -38,20 +37,9 @@ import ( "github.com/terst/listen/server/internal/store" "github.com/terst/listen/server/internal/transcribe" "github.com/terst/listen/server/internal/transcript" + "github.com/terst/listen/server/internal/words" ) -// tierRankOf orders the tiers for work: topical first, archive last. -func tierRankOf(tier string) int { - switch tier { - case store.TierTopical: - return 0 - case store.TierArchive: - return 2 - default: - return 1 - } -} - // minRecurringCorpus is how many of a show's episodes must have words before what they share // means anything. Three: the target and the two others a stretch has to turn up in. const minRecurringCorpus = 3 @@ -99,6 +87,7 @@ func main() { decoder := decode.Decoder{} transcriptDir := filepath.Join(*dataDir, "transcripts") transcripts := transcript.Store{Dir: transcriptDir} + secondCopyDir := filepath.Join(*dataDir, "work", "second") worker := &worker{ state: state, statePath: statePath, @@ -107,22 +96,26 @@ func main() { keep: *keep, mirror: &mirror.Mirror{AudioDir: audioDir, Probe: decoder}, decoder: decoder, - transcripts: transcripts, - harvester: transcript.Harvester{Store: transcripts}, - 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"), + words: &words.Pipeline{ + State: state, + AudioDir: audioDir, + Transcripts: transcripts, + Harvester: transcript.Harvester{Store: transcripts}, + 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"), + }, + SecondCopies: &mirror.Mirror{AudioDir: secondCopyDir, Probe: decoder}, + SecondCopyDir: secondCopyDir, + RecogniseBudget: *transcribeBudget, + SecondCopyBudget: *refetchBudget, }, - transcribeBudget: *transcribeBudget, - refetchMirror: &mirror.Mirror{AudioDir: filepath.Join(*dataDir, "work", "refetch"), Probe: decoder}, - refetchDir: filepath.Join(*dataDir, "work", "refetch"), - refetchBudget: *refetchBudget, - requests: make(chan request, 64), - wakeups: make(chan struct{}, 1), + requests: make(chan request, 64), + wakeups: make(chan struct{}, 1), } server := api.New(state, worker, api.Options{ @@ -150,12 +143,12 @@ func main() { defer stop() if *once { - worker.rederiveAds() + worker.words.Rederive() 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) - worker.refetchSome(ctx) + worker.words.Recognise(ctx) + worker.words.SecondCopy(ctx) if err := state.Save(statePath); err != nil { log.Fatalf("save: %v", err) } @@ -208,21 +201,12 @@ type worker struct { decoder decode.Decoder // transcripts holds each episode's words, and harvester takes the publishers' own where // they describe the copy that was mirrored. See transcript.Fits for when they do not. - transcripts transcript.Store - harvester transcript.Harvester - // 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 - // refetchMirror downloads a second copy of an episode into its own directory, so the one - // the phone plays is never touched. Its files are deleted as soon as they are recognised. - refetchMirror *mirror.Mirror - refetchDir string - refetchBudget time.Duration - requests chan request - wakeups chan struct{} + // words is the transcript side: taking publishers' transcripts, recognising the rest, and + // working out what to skip from what was said. It lives in its own package because none + // of it could be tested here. + words *words.Pipeline + requests chan request + wakeups chan struct{} } // wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll. @@ -244,12 +228,12 @@ func (w *worker) Request(feedURL, guid, enclosureURL, title string) { } func (w *worker) run(ctx context.Context, every time.Duration) { - w.rederiveAds() + w.words.Rederive() w.save() w.pollAll(ctx) w.save() - w.transcribeSome(ctx) - w.refetchSome(ctx) + w.words.Recognise(ctx) + w.words.SecondCopy(ctx) ticker := time.NewTicker(every) defer ticker.Stop() @@ -263,8 +247,8 @@ func (w *worker) run(ctx context.Context, every time.Duration) { // 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) - w.refetchSome(ctx) + w.words.Recognise(ctx) + w.words.SecondCopy(ctx) case <-w.wakeups: w.pollAll(ctx) w.save() @@ -367,14 +351,8 @@ func (w *worker) pollOne(ctx context.Context, feedURL string) { if e.Mirrored() { _ = w.mirror.Prune(e) } - // The words go with the audio. A transcript of a file nobody holds any more is of no - // use to the phone, which plays the server's bytes, and the directory would otherwise - // grow for ever. - if e.TranscriptExamined() { - _ = w.transcripts.Prune(e.ID) - e.TranscriptSource, e.TranscriptCues, e.TranscriptNote = "", 0, "" - e.TranscriptAtUnix = 0 - } + // The words go with the audio -- see words.Pipeline.Forget. + w.words.Forget(e) } fetched := 0 @@ -395,607 +373,11 @@ func (w *worker) pollOne(ctx context.Context, feedURL string) { if fetched > 0 { log.Printf("%s: mirrored %d new episode(s)", channel.Title, fetched) } - w.harvestTranscripts(ctx, show, kept) - w.findRecurring(show) + w.words.Harvest(ctx, show, kept) + w.words.Recurring(show) w.detect(ctx, show, kept) } -// harvestTranscripts takes the publishers' own transcripts for a show's mirrored episodes. -// -// Only where the publisher's master is what was mirrored, which transcript.Fits decides from the -// declared-versus-served length gap rather than from the words. A show using dynamic ad -// insertion publishes a transcript of the ad-free programme, and reading along with it would be -// minutes out of step with the audio the phone plays. -// -// Each episode is examined once. A refusal is a judgement about this copy, so re-fetching it on -// every poll would be work done for ever to reach the same answer -- and the refusal is itself -// the record of which episodes speech recognition would have to do. -func (w *worker) harvestTranscripts(ctx context.Context, show *store.Show, candidates []*store.Episode) { - taken, declined, spotted, masters := 0, 0, 0, 0 - for _, e := range candidates { - if ctx.Err() != nil { - return - } - // A declined episode is examined and has no words, but its publisher's transcript is - // still the ad-free master and is worth fetching -- it is what Inserted subtracts. - // Episodes declined before masters were kept arrive here once. - if !e.Mirrored() || e.TranscriptURL == "" { - continue - } - if e.TranscriptExamined() && !e.NeedsMaster() { - continue - } - result, err := w.harvester.Harvest(ctx, transcript.Offer{ - EpisodeID: e.ID, - URL: e.TranscriptURL, - Type: e.TranscriptType, - ServedMs: e.DurationMs, - DeclaredMs: e.DeclaredDurationMs, - }) - if err != nil { - // The download failed, so nothing has been decided and the next poll tries again. - log.Printf("%s: transcript for %q: %v", show.Title, e.Title, err) - continue - } - if result.Master > 0 { - e.MasterCues = result.Master - e.MasterNote = "" - masters++ - } else if e.NeedsMaster() { - // Offered, fetched, and not usable as a master either. Recorded so it is not - // fetched again on every poll for ever. - e.MasterNote = result.Note - } - switch { - case result.Stored: - e.TranscriptSource = string(transcript.SourcePublished) - e.TranscriptCues = result.Cues - e.TranscriptAtUnix = time.Now().Unix() - e.TranscriptNote = "" - taken++ - spotted += result.Ads - case result.Note != "": - e.TranscriptNote = result.Note - e.TranscriptAtUnix = time.Now().Unix() - declined++ - } - } - if taken > 0 || declined > 0 || masters > 0 { - // The ad count is reported, not acted on. These are candidates read out of the words, - // and nothing publishes them as ranges to skip -- see transcript.Spot. - log.Printf("%s: transcripts -- took %d, declined %d (%d kept as ad-free masters), "+ - "%d ad candidate(s) in the words", - show.Title, taken, declined, masters, spotted) - } -} - -// 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: w.adsIn(e, 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, %s, took %s", - e.Title, len(stored.Cues), adAccounting(e, 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())) - } -} - -/* -rederiveAds works out every stored episode's advertising again, from the words it already has. - -Recognition is the expensive part and its output does not change; which stretches of it are -advertising is a judgement that has. When subtraction learned to pull its edges back off the -programme, the episodes already transcribed kept the old boundaries, and re-recognising them to -pick up a better answer would cost hours of CPU for text that is already on disk. - -So it runs over the stored transcripts at startup instead -- eighteen alignments, a few -milliseconds each -- and the next improvement costs nothing to apply either. -*/ -func (w *worker) rederiveAds() { - changed := 0 - for _, show := range w.state.Shows() { - for _, e := range show.Episodes { - if e.MasterCues == 0 || !e.HasTranscript() { - continue - } - held, err := w.transcripts.Get(e.ID) - if err != nil { - continue - } - found := w.adsIn(e, held.Cues) - // Publishing is decided every time, even when the candidates have not moved: what - // is fit to act on is a separate judgement from what was found, and it changes on - // its own. Only the file is left alone when nothing moved. - // - // Only for episodes this pass owns. An episode with no master has its breaks from - // findRecurring, and re-deriving subtraction for it would clear them -- it has - // nothing to subtract. - if e.MasterCues > 0 { - w.publishWordBreaks(e, found) - } - if sameCandidates(found, held.Ads) { - continue - } - held.Ads = found - if err := w.transcripts.Put(held); err != nil { - log.Printf("re-deriving %q: %v", e.Title, err) - continue - } - changed++ - } - } - published, breaks := 0, 0 - for _, show := range w.state.Shows() { - for _, e := range show.Episodes { - if len(e.WordBreaks) > 0 { - published++ - breaks += len(e.WordBreaks) - } - } - } - log.Printf("re-derived the advertising in %d stored transcript(s); "+ - "%d episode(s) publish %d break(s) found in their words", changed, published, breaks) -} - -func sameCandidates(left, right []transcript.Candidate) bool { - if len(left) != len(right) { - return false - } - for i := range left { - if left[i].StartMs != right[i].StartMs || left[i].EndMs != right[i].EndMs { - return false - } - } - return true -} - -/* -findRecurring works out what each of a show's episodes says every week. - -Per show rather than per episode, because the comparison is against the show's other episodes. -Eight of them are mirrored within a minute of each other, so they carry one advertising campaign -and eight different programmes -- the text analogue of what makes the audio scan work, using -transcripts already on disk and costing no download and no recognition. - -Published as the show's own audio rather than as advertising. What recurs is the sponsor reads -and the furniture together, and nothing measurable separates them: ad copy turns up in 5% of the -advertising and 4% of the furniture, and length and position overlap throughout. The listener's -own setting governs a show's own audio, so claiming the weaker of the two is the safe mistake. - -Skipped for an episode that already has measured advertising. Subtraction knows which stretches -are inserted; this only knows that they repeat, and the better answer should not be overwritten -by the vaguer one. -*/ -func (w *worker) findRecurring(show *store.Show) { - type held struct { - episode *store.Episode - cues []transcript.Cue - } - var corpus []held - for _, e := range show.Episodes { - if !e.Mirrored() || !e.HasTranscript() { - continue - } - stored, err := w.transcripts.Get(e.ID) - if err != nil { - continue - } - corpus = append(corpus, held{e, stored.Cues}) - } - if len(corpus) < minRecurringCorpus { - return - } - - published, refused := 0, 0 - for i, target := range corpus { - // Skipped only where subtraction actually produced something, not merely where a - // master exists. A refetch master that carried the same campaign yields no candidates - // at all, and treating "has a master" as "has a better answer" left those episodes - // with nothing -- Tangle's "The Israeli elections" has a 229-second gap, a refetch - // master, no subtraction candidates, and was excluded from this. - if len(target.episode.WordBreaks) > 0 { - continue - } - others := make([][]transcript.Cue, 0, len(corpus)-1) - for j, other := range corpus { - if i != j { - others = append(others, other.cues) - } - } - found := transcript.Recurring(target.cues, others) - fit, why := transcript.PublishableRecurring(found, target.episode.DurationMs) - target.episode.WordAdsRefused = why - target.episode.WordBreaks = nil - for _, c := range fit { - target.episode.WordBreaks = append(target.episode.WordBreaks, store.Break{ - StartMs: c.StartMs, - EndMs: c.EndMs, - // What recurs is the show's own recurring audio, which is what this kind - // means. The listener decides whether to skip it. - Kind: string(skips.KindCue), - }) - } - if len(fit) > 0 { - published++ - } - if why != "" { - refused++ - } - } - if published > 0 || refused > 0 { - log.Printf("%s: recurring text -- %d episode(s) publish it, %d refused, from a corpus of %d", - show.Title, published, refused, len(corpus)) - } -} - -/* -publishWordBreaks turns what the words found into stretches the phone will actually seek past. - -The only place anything read out of a transcript reaches playback. Everything else in this -pipeline produces candidates to look at; transcript.Publishable decides which of them are fit to -act on, and refuses the rest with a reason rather than silently dropping them. - -Kept in their own field so the ad scan, which rewrites its own list from scratch on every -detection pass, cannot wipe them -- and so that which detector found a break stays answerable. -*/ -func (w *worker) publishWordBreaks(e *store.Episode, found []transcript.Candidate) { - fit, why := transcript.Publishable(found, e.DurationMs, e.DeclaredDurationMs) - e.WordAdsRefused = why - e.WordBreaks = nil - for _, c := range fit { - e.WordBreaks = append(e.WordBreaks, store.Break{ - StartMs: c.StartMs, - EndMs: c.EndMs, - Kind: string(skips.KindAd), - }) - } -} - -/* -adsIn works out which stretches of a recognised episode are advertising. - -By subtraction where the publisher's ad-free master is held, and by reading the ad copy where it -is not. They are not close in quality and the choice is not a preference. Measured on Darknet -Diaries' "Conti": the feed declares 59:17 and the file served runs 63:33, so 4:15 of advertising -was inserted -- subtraction found 4:20 of it in three pods, each starting on the first word of -an ad read and ending on the last, while reading the copy found 0:16. 102% against 6%. - -Subtraction wins because it is a measurement. The master says what the programme is, the -recognised copy says what was played, and the difference is the advertising by definition -- -there is no phrase list to be wrong about, and no show whose hosts happen to talk like an -advertisement. -*/ -func (w *worker) adsIn(e *store.Episode, cues []transcript.Cue) []transcript.Candidate { - if e.MasterCues > 0 { - if master, err := w.transcripts.Master(e.ID); err == nil { - if found := transcript.Inserted(master.Cues, cues); len(found) > 0 { - return found - } - } - } - return transcript.Spot(cues) -} - -// 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 - } -} - -/* -adAccounting describes what was found against what the lengths say is there. - -The declared-versus-served gap comes from the feed and the file rather than from any detector, -which makes it the one number able to contradict one. - -It bounds only the advertising a CDN *inserted* into this copy, which is not all the advertising -in it. A show that bakes its sponsor reads into the file it publishes has no gap at all and can -still be a third advertisements -- The 404 Media Podcast declares within 11 seconds of what it -serves and carries Mack Weldon, Incogni and Warby Parker reads in the middle of it. So finding -more than the gap implies is not over-reporting, and saying so in the readout matters: the first -reading of a "728%" line was that the detector had gone wrong, and it had not. -*/ -func adAccounting(e *store.Episode, found []transcript.Candidate) string { - var total int64 - for _, c := range found { - total += c.EndMs - c.StartMs - } - how := "read from the ad copy" - switch { - case e.MasterSource == string(transcript.SourceRefetch): - how = "by subtracting a second fetch" - case e.MasterCues > 0: - how = "by subtracting the publisher's master" - } - gap := e.DurationMs - e.DeclaredDurationMs - switch { - case gap <= 0: - return fmt.Sprintf("%d ad candidate(s) %s, %s; nothing was inserted into this copy, "+ - "so anything found is baked into what the publisher serves", - len(found), how, clock(total)) - case total > gap: - // Deliberately does not say why it is over. Two causes look identical here and the - // first guess was the wrong one: advertising baked into what the publisher serves, - // which the gap cannot see, and candidate edges reaching past the advertisement. - // Subtraction answers "not in the master", which is a superset of "advertising" -- - // a title card, a music sting and whisper's own [MUSIC] markers are all absent from a - // publisher's written transcript too. Checked on Heavyweight "#70 Cayden", where a - // 134-second candidate opens on `Today's episode, "Caiton."` and `[MUSIC]`. - return fmt.Sprintf("%d ad candidate(s) %s, %s against %s inserted -- over by %s, "+ - "which is either baked-in advertising or candidate edges reaching too far", - len(found), how, clock(total), clock(gap), clock(total-gap)) - default: - return fmt.Sprintf("%d ad candidate(s) %s, %s of the %s inserted (%d%%)", - len(found), how, clock(total), clock(gap), total*100/gap) - } -} - -/* -refetchSome fetches a second copy of episodes that have no master, and recognises it. - -The extension of subtraction to the rest of the library. Only 38 of 336 mirrored episodes are -offered a publisher's transcript, and only 18 of those are the ad-carrying kind subtraction -works on -- so the method that measures itself at 92-104% against the length gap covers about -five per cent of what Peter listens to. - -A second fetch stands in for the master. Advertising is inserted per request, which this project -already measured the hard way: the same Radiolab episode came back 7:29 of advertising on one -fetch and 5:34 on another. So two copies agree on the programme and differ on the advertising, -and subtracting one from the other says where the advertising in the other one is. What it -cannot see is a campaign both copies happened to carry -- the exact opposite blind spot to the -audio scan, whose whole method is finding what a batch shares. - -The second copy's audio is deleted the moment it has been recognised. The phone plays the -mirrored bytes and nothing else may claim that slot; this is evidence, not a recording. -*/ -func (w *worker) refetchSome(ctx context.Context) { - if !w.recogniser.Available() || w.refetchBudget <= 0 { - return - } - deadline := time.Now().Add(w.refetchBudget) - done, failed := 0, 0 - var breaker transcribe.Breaker - var failing []*store.Episode - - for _, e := range w.pendingRefetch() { - if ctx.Err() != nil || time.Now().After(deadline) { - break - } - found, secondMs, err := w.refetchOne(ctx, e) - if ctx.Err() != nil { - return - } - e.RefetchAttemptedAtUnix = time.Now().Unix() - if err != nil { - log.Printf("refetching %q: %v", e.Title, err) - failed++ - failing = append(failing, e) - if breaker.Failure() { - for _, given := range failing { - given.RefetchAttemptedAtUnix = 0 - } - log.Printf("refetching abandoned after %d failures in a row -- giving those "+ - "attempts back, since this is the machine and not the episodes", breaker.Run()) - w.save() - return - } - w.save() - continue - } - breaker.Success() - failing = nil - done++ - // The two lengths are the first thing to look at when this finds nothing: two fetches - // that came back the same length probably carried the same campaign, and there is - // then nothing for subtraction to see. - log.Printf("second copy of %q: %s against the mirrored %s, %d lines, %s", - e.Title, clock(secondMs), clock(e.DurationMs), e.MasterCues, adAccounting(e, found)) - w.save() - } - if done > 0 || failed > 0 { - log.Printf("second copies -- %d done, %d failed, %d still waiting", - done, failed, len(w.pendingRefetch())) - } -} - -// refetchOne downloads one episode again, recognises it, keeps the words and throws the audio -// away. -func (w *worker) refetchOne( - ctx context.Context, e *store.Episode, -) ([]transcript.Candidate, int64, error) { - // Its own id and its own directory, so nothing can mistake this for the copy the phone - // plays even for the moment it exists. - second := &store.Episode{ID: e.ID + "-second", EnclosureURL: e.EnclosureURL} - if err := w.refetchMirror.Fetch(ctx, second); err != nil { - return nil, 0, err - } - defer func() { _ = w.refetchMirror.Prune(second) }() - - result, err := w.recogniser.Transcribe(ctx, filepath.Join(w.refetchDir, second.Path)) - if err != nil { - return nil, second.DurationMs, err - } - if err := w.transcripts.PutMaster(transcript.Transcript{ - EpisodeID: e.ID, - Source: transcript.SourceRefetch, - LanguageCode: result.LanguageCode, - Cues: result.Cues, - }); err != nil { - return nil, second.DurationMs, err - } - e.MasterCues = len(result.Cues) - e.MasterSource = string(transcript.SourceRefetch) - e.MasterNote = "" - - // Re-derive this episode's advertising now that there is something to subtract it from. - held, err := w.transcripts.Get(e.ID) - if err != nil { - return nil, second.DurationMs, err - } - held.Ads = w.adsIn(e, held.Cues) - if err := w.transcripts.Put(held); err != nil { - return nil, second.DurationMs, err - } - w.publishWordBreaks(e, held.Ads) - return held.Ads, second.DurationMs, nil -} - -/* -pendingRefetch is every episode with its own words, no master, and advertising to find. - -Newest first: an episode about to be played is worth the work before one from the back of a -catalogue. -*/ -func (w *worker) pendingRefetch() []*store.Episode { - now := time.Now().Unix() - var pending []*store.Episode - for _, show := range w.state.Shows() { - for _, e := range show.Episodes { - if !e.NeedsRefetch(now) { - continue - } - // Without a length gap nothing was inserted, so a second fetch would be the same - // recording and there would be nothing to subtract. - if !transcript.HasInsertedAdvertising(e.DurationMs, e.DeclaredDurationMs) { - continue - } - pending = append(pending, e) - } - } - sort.Slice(pending, func(i, j int) bool { - return pending[i].PublishedUnix > pending[j].PublishedUnix - }) - return pending -} - -/* -pendingRecognition is every mirrored episode still without words, best first. - -An episode holding the publisher's ad-free master comes first, whatever its date. Recognising -one of those produces an answer that checks itself: subtraction found 4:20 of the 4:15 the -lengths imply on Darknet Diaries and 3:32 of 3:29 on Heavyweight. Recognising anything else -leaves the ad copy to be read, which across the first nineteen real episodes found 0%, 0%, 15% -and 29% of what the lengths say is there. Spending the budget where the result can be trusted -first is worth more than spending it in date order. - -Newest first within each group, because an episode about to be played is worth having words for -before one from the back of a catalogue. -*/ -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) - } - } - } - // Three things decide the order, and they are in the order they matter. A topical show is - // worthless by Thursday. An episode whose answer checks itself against the length gap is - // worth more than one that cannot. And within that, newest first. - rank := map[string]int{} - for _, show := range w.state.Shows() { - for _, e := range show.Episodes { - rank[e.ID] = tierRankOf(w.state.Tier(show.FeedURL)) - } - } - sort.SliceStable(pending, func(i, j int) bool { - left, right := pending[i], pending[j] - if rank[left.ID] != rank[right.ID] { - return rank[left.ID] < rank[right.ID] - } - if (left.MasterCues > 0) != (right.MasterCues > 0) { - return left.MasterCues > 0 - } - return left.PublishedUnix > right.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 diff --git a/server/internal/words/pipeline.go b/server/internal/words/pipeline.go new file mode 100644 index 0000000..b37efa3 --- /dev/null +++ b/server/internal/words/pipeline.go @@ -0,0 +1,706 @@ +/* +Package words turns an episode's audio into what was said, and what was said into what to skip. + +Everything here used to live in cmd/listend, where nothing could test it: the command has no +tests and nine passes of real logic had accumulated in one 1,249-line file. These five are the +ones with judgement in them. + +They run in a fixed order and each depends on the last. Harvest takes a publisher's transcript +where it describes the copy that was mirrored, and keeps it as an ad-free master where it does +not. Recognise does the episodes with no words yet. Rederive works out the advertising again +from words already on disk, which costs nothing and is how an improvement to the rules reaches +episodes transcribed days ago. Recurring finds what a show says every week. SecondCopy fetches +an episode again, for the shows whose episodes are still worth hearing weeks later. + +The expensive ones are bounded by a budget per cycle and abandon themselves after three failures +in a row, because the failure that matters is the systemic one: a misconfigured engine marking +every episode in the library as attempted in a few seconds. +*/ +package words + +import ( + "context" + "fmt" + "log" + "path/filepath" + "sort" + "time" + + "github.com/terst/listen/server/internal/mirror" + "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" +) + +// minRecurringCorpus is how many of a show's episodes must have words before what they share +// means anything. Three: the target and the two others a stretch has to turn up in. +const minRecurringCorpus = 3 + +/* +Pipeline is the transcript side of the server, and everything it needs. + +Fields rather than constructor arguments because there are eight of them and they are all +plainly named; a constructor would be eight positional arguments or eight setters. + +A zero Recogniser is a working state, not a broken one: the server still takes publishers' +transcripts and still finds what a show repeats. Only recognition and second copies go quiet. +*/ +type Pipeline struct { + State *store.State + Transcripts transcript.Store + // AudioDir is where the mirrored episodes live -- the bytes the phone plays and the ones + // recognition reads. + AudioDir string + Harvester transcript.Harvester + Recogniser transcribe.Whisper + + // SecondCopies downloads an episode again into its own directory, so the copy the phone + // plays is never touched. Its files are deleted as soon as they have been recognised. + SecondCopies *mirror.Mirror + SecondCopyDir string + + // RecogniseBudget and SecondCopyBudget bound what one cycle may spend, so a backlog is + // worked through steadily rather than taking the machine over. + RecogniseBudget time.Duration + SecondCopyBudget time.Duration + + // Save persists the state. Called as each episode finishes rather than at the end, + // because these passes run for tens of minutes and a restart in the middle should not + // lose the ones already done. + Save func() +} + +/* +Forget drops everything known about an episode's words. + +Called when its audio is reclaimed. A transcript of a file nobody holds is of no use to the +phone, which plays the server's bytes, and the directory would otherwise grow for ever. + +Every field goes, not just the obvious ones. An episode that kept MasterCues after its master +file was deleted would claim a second description it could not produce, and subtraction would +quietly find nothing for it for ever -- which is how the refetch master came to block the +recurring detector a fortnight ago. +*/ +func (p *Pipeline) Forget(e *store.Episode) { + _ = p.Transcripts.Prune(e.ID) + e.TranscriptSource, e.TranscriptCues, e.TranscriptNote = "", 0, "" + e.TranscriptAtUnix = 0 + e.MasterCues, e.MasterNote, e.MasterSource = 0, "", "" + e.TranscribeAttemptedAtUnix, e.RefetchAttemptedAtUnix = 0, 0 + e.WordBreaks, e.WordAdsRefused = nil, "" +} + +// tierRankOf orders the tiers for work: topical first, archive last. +func tierRankOf(tier string) int { + switch tier { + case store.TierTopical: + return 0 + case store.TierArchive: + return 2 + default: + return 1 + } +} + +// clock renders a duration the way every readout in this project does. +func clock(ms int64) string { + s := ms / 1000 + return fmt.Sprintf("%d:%02d", s/60, s%60) +} + +// harvestTranscripts takes the publishers' own transcripts for a show's mirrored episodes. +// +// Only where the publisher's master is what was mirrored, which transcript.Fits decides from the +// declared-versus-served length gap rather than from the words. A show using dynamic ad +// insertion publishes a transcript of the ad-free programme, and reading along with it would be +// minutes out of step with the audio the phone plays. +// +// Each episode is examined once. A refusal is a judgement about this copy, so re-fetching it on +// every poll would be work done for ever to reach the same answer -- and the refusal is itself +// the record of which episodes speech recognition would have to do. +func (p *Pipeline) Harvest(ctx context.Context, show *store.Show, candidates []*store.Episode) { + taken, declined, spotted, masters := 0, 0, 0, 0 + for _, e := range candidates { + if ctx.Err() != nil { + return + } + // A declined episode is examined and has no words, but its publisher's transcript is + // still the ad-free master and is worth fetching -- it is what Inserted subtracts. + // Episodes declined before masters were kept arrive here once. + if !e.Mirrored() || e.TranscriptURL == "" { + continue + } + if e.TranscriptExamined() && !e.NeedsMaster() { + continue + } + result, err := p.Harvester.Harvest(ctx, transcript.Offer{ + EpisodeID: e.ID, + URL: e.TranscriptURL, + Type: e.TranscriptType, + ServedMs: e.DurationMs, + DeclaredMs: e.DeclaredDurationMs, + }) + if err != nil { + // The download failed, so nothing has been decided and the next poll tries again. + log.Printf("%s: transcript for %q: %v", show.Title, e.Title, err) + continue + } + if result.Master > 0 { + e.MasterCues = result.Master + e.MasterNote = "" + masters++ + } else if e.NeedsMaster() { + // Offered, fetched, and not usable as a master either. Recorded so it is not + // fetched again on every poll for ever. + e.MasterNote = result.Note + } + switch { + case result.Stored: + e.TranscriptSource = string(transcript.SourcePublished) + e.TranscriptCues = result.Cues + e.TranscriptAtUnix = time.Now().Unix() + e.TranscriptNote = "" + taken++ + spotted += result.Ads + case result.Note != "": + e.TranscriptNote = result.Note + e.TranscriptAtUnix = time.Now().Unix() + declined++ + } + } + if taken > 0 || declined > 0 || masters > 0 { + // The ad count is reported, not acted on. These are candidates read out of the words, + // and nothing publishes them as ranges to skip -- see transcript.Spot. + log.Printf("%s: transcripts -- took %d, declined %d (%d kept as ad-free masters), "+ + "%d ad candidate(s) in the words", + show.Title, taken, declined, masters, spotted) + } +} + +// 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 (p *Pipeline) Recognise(ctx context.Context) { + if !p.Recogniser.Available() || p.RecogniseBudget <= 0 { + return + } + deadline := time.Now().Add(p.RecogniseBudget) + 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 p.pendingRecognition() { + if ctx.Err() != nil || time.Now().After(deadline) { + break + } + started := time.Now() + result, err := p.Recogniser.Transcribe(ctx, filepath.Join(p.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()) + p.Save() + return + } + p.Save() + continue + } + + stored := transcript.Transcript{ + EpisodeID: e.ID, + Source: transcript.SourceASR, + LanguageCode: result.LanguageCode, + Cues: result.Cues, + Ads: p.adsIn(e, result.Cues), + } + if err := p.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()) + p.Save() + return + } + p.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, %s, took %s", + e.Title, len(stored.Cues), adAccounting(e, stored.Ads), + time.Since(started).Round(time.Second)) + p.Save() + } + + if done > 0 || failed > 0 { + log.Printf("recognition -- %d done, %d failed, %d still waiting", + done, failed, len(p.pendingRecognition())) + } +} + +/* +rederiveAds works out every stored episode's advertising again, from the words it already has. + +Recognition is the expensive part and its output does not change; which stretches of it are +advertising is a judgement that has. When subtraction learned to pull its edges back off the +programme, the episodes already transcribed kept the old boundaries, and re-recognising them to +pick up a better answer would cost hours of CPU for text that is already on disk. + +So it runs over the stored transcripts at startup instead -- eighteen alignments, a few +milliseconds each -- and the next improvement costs nothing to apply either. +*/ +func (p *Pipeline) Rederive() { + changed := 0 + for _, show := range p.State.Shows() { + for _, e := range show.Episodes { + if e.MasterCues == 0 || !e.HasTranscript() { + continue + } + held, err := p.Transcripts.Get(e.ID) + if err != nil { + continue + } + found := p.adsIn(e, held.Cues) + // Publishing is decided every time, even when the candidates have not moved: what + // is fit to act on is a separate judgement from what was found, and it changes on + // its own. Only the file is left alone when nothing moved. + // + // Only for episodes this pass owns. An episode with no master has its breaks from + // findRecurring, and re-deriving subtraction for it would clear them -- it has + // nothing to subtract. + if e.MasterCues > 0 { + p.publishWordBreaks(e, found) + } + if sameCandidates(found, held.Ads) { + continue + } + held.Ads = found + if err := p.Transcripts.Put(held); err != nil { + log.Printf("re-deriving %q: %v", e.Title, err) + continue + } + changed++ + } + } + published, breaks := 0, 0 + for _, show := range p.State.Shows() { + for _, e := range show.Episodes { + if len(e.WordBreaks) > 0 { + published++ + breaks += len(e.WordBreaks) + } + } + } + log.Printf("re-derived the advertising in %d stored transcript(s); "+ + "%d episode(s) publish %d break(s) found in their words", changed, published, breaks) +} + +func sameCandidates(left, right []transcript.Candidate) bool { + if len(left) != len(right) { + return false + } + for i := range left { + if left[i].StartMs != right[i].StartMs || left[i].EndMs != right[i].EndMs { + return false + } + } + return true +} + +/* +findRecurring works out what each of a show's episodes says every week. + +Per show rather than per episode, because the comparison is against the show's other episodes. +Eight of them are mirrored within a minute of each other, so they carry one advertising campaign +and eight different programmes -- the text analogue of what makes the audio scan work, using +transcripts already on disk and costing no download and no recognition. + +Published as the show's own audio rather than as advertising. What recurs is the sponsor reads +and the furniture together, and nothing measurable separates them: ad copy turns up in 5% of the +advertising and 4% of the furniture, and length and position overlap throughout. The listener's +own setting governs a show's own audio, so claiming the weaker of the two is the safe mistake. + +Skipped for an episode that already has measured advertising. Subtraction knows which stretches +are inserted; this only knows that they repeat, and the better answer should not be overwritten +by the vaguer one. +*/ +func (p *Pipeline) Recurring(show *store.Show) { + type held struct { + episode *store.Episode + cues []transcript.Cue + } + var corpus []held + for _, e := range show.Episodes { + if !e.Mirrored() || !e.HasTranscript() { + continue + } + stored, err := p.Transcripts.Get(e.ID) + if err != nil { + continue + } + corpus = append(corpus, held{e, stored.Cues}) + } + if len(corpus) < minRecurringCorpus { + return + } + + published, refused := 0, 0 + for i, target := range corpus { + // Skipped only where subtraction actually produced something, not merely where a + // master exists. A refetch master that carried the same campaign yields no candidates + // at all, and treating "has a master" as "has a better answer" left those episodes + // with nothing -- Tangle's "The Israeli elections" has a 229-second gap, a refetch + // master, no subtraction candidates, and was excluded from this. + if len(target.episode.WordBreaks) > 0 { + continue + } + others := make([][]transcript.Cue, 0, len(corpus)-1) + for j, other := range corpus { + if i != j { + others = append(others, other.cues) + } + } + found := transcript.Recurring(target.cues, others) + fit, why := transcript.PublishableRecurring(found, target.episode.DurationMs) + target.episode.WordAdsRefused = why + target.episode.WordBreaks = nil + for _, c := range fit { + target.episode.WordBreaks = append(target.episode.WordBreaks, store.Break{ + StartMs: c.StartMs, + EndMs: c.EndMs, + // What recurs is the show's own recurring audio, which is what this kind + // means. The listener decides whether to skip it. + Kind: string(skips.KindCue), + }) + } + if len(fit) > 0 { + published++ + } + if why != "" { + refused++ + } + } + if published > 0 || refused > 0 { + log.Printf("%s: recurring text -- %d episode(s) publish it, %d refused, from a corpus of %d", + show.Title, published, refused, len(corpus)) + } +} + +/* +publishWordBreaks turns what the words found into stretches the phone will actually seek past. + +The only place anything read out of a transcript reaches playback. Everything else in this +pipeline produces candidates to look at; transcript.Publishable decides which of them are fit to +act on, and refuses the rest with a reason rather than silently dropping them. + +Kept in their own field so the ad scan, which rewrites its own list from scratch on every +detection pass, cannot wipe them -- and so that which detector found a break stays answerable. +*/ +func (p *Pipeline) publishWordBreaks(e *store.Episode, found []transcript.Candidate) { + fit, why := transcript.Publishable(found, e.DurationMs, e.DeclaredDurationMs) + e.WordAdsRefused = why + e.WordBreaks = nil + for _, c := range fit { + e.WordBreaks = append(e.WordBreaks, store.Break{ + StartMs: c.StartMs, + EndMs: c.EndMs, + Kind: string(skips.KindAd), + }) + } +} + +/* +adsIn works out which stretches of a recognised episode are advertising. + +By subtraction where the publisher's ad-free master is held, and by reading the ad copy where it +is not. They are not close in quality and the choice is not a preference. Measured on Darknet +Diaries' "Conti": the feed declares 59:17 and the file served runs 63:33, so 4:15 of advertising +was inserted -- subtraction found 4:20 of it in three pods, each starting on the first word of +an ad read and ending on the last, while reading the copy found 0:16. 102% against 6%. + +Subtraction wins because it is a measurement. The master says what the programme is, the +recognised copy says what was played, and the difference is the advertising by definition -- +there is no phrase list to be wrong about, and no show whose hosts happen to talk like an +advertisement. +*/ +func (p *Pipeline) adsIn(e *store.Episode, cues []transcript.Cue) []transcript.Candidate { + if e.MasterCues > 0 { + if master, err := p.Transcripts.Master(e.ID); err == nil { + if found := transcript.Inserted(master.Cues, cues); len(found) > 0 { + return found + } + } + } + return transcript.Spot(cues) +} + +// 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 + } +} + +/* +adAccounting describes what was found against what the lengths say is there. + +The declared-versus-served gap comes from the feed and the file rather than from any detector, +which makes it the one number able to contradict one. + +It bounds only the advertising a CDN *inserted* into this copy, which is not all the advertising +in it. A show that bakes its sponsor reads into the file it publishes has no gap at all and can +still be a third advertisements -- The 404 Media Podcast declares within 11 seconds of what it +serves and carries Mack Weldon, Incogni and Warby Parker reads in the middle of it. So finding +more than the gap implies is not over-reporting, and saying so in the readout matters: the first +reading of a "728%" line was that the detector had gone wrong, and it had not. +*/ +func adAccounting(e *store.Episode, found []transcript.Candidate) string { + var total int64 + for _, c := range found { + total += c.EndMs - c.StartMs + } + how := "read from the ad copy" + switch { + case e.MasterSource == string(transcript.SourceRefetch): + how = "by subtracting a second fetch" + case e.MasterCues > 0: + how = "by subtracting the publisher's master" + } + gap := e.DurationMs - e.DeclaredDurationMs + switch { + case gap <= 0: + return fmt.Sprintf("%d ad candidate(s) %s, %s; nothing was inserted into this copy, "+ + "so anything found is baked into what the publisher serves", + len(found), how, clock(total)) + case total > gap: + // Deliberately does not say why it is over. Two causes look identical here and the + // first guess was the wrong one: advertising baked into what the publisher serves, + // which the gap cannot see, and candidate edges reaching past the advertisement. + // Subtraction answers "not in the master", which is a superset of "advertising" -- + // a title card, a music sting and whisper's own [MUSIC] markers are all absent from a + // publisher's written transcript too. Checked on Heavyweight "#70 Cayden", where a + // 134-second candidate opens on `Today's episode, "Caiton."` and `[MUSIC]`. + return fmt.Sprintf("%d ad candidate(s) %s, %s against %s inserted -- over by %s, "+ + "which is either baked-in advertising or candidate edges reaching too far", + len(found), how, clock(total), clock(gap), clock(total-gap)) + default: + return fmt.Sprintf("%d ad candidate(s) %s, %s of the %s inserted (%d%%)", + len(found), how, clock(total), clock(gap), total*100/gap) + } +} + +/* +refetchSome fetches a second copy of episodes that have no master, and recognises it. + +The extension of subtraction to the rest of the library. Only 38 of 336 mirrored episodes are +offered a publisher's transcript, and only 18 of those are the ad-carrying kind subtraction +works on -- so the method that measures itself at 92-104% against the length gap covers about +five per cent of what Peter listens to. + +A second fetch stands in for the master. Advertising is inserted per request, which this project +already measured the hard way: the same Radiolab episode came back 7:29 of advertising on one +fetch and 5:34 on another. So two copies agree on the programme and differ on the advertising, +and subtracting one from the other says where the advertising in the other one is. What it +cannot see is a campaign both copies happened to carry -- the exact opposite blind spot to the +audio scan, whose whole method is finding what a batch shares. + +The second copy's audio is deleted the moment it has been recognised. The phone plays the +mirrored bytes and nothing else may claim that slot; this is evidence, not a recording. +*/ +func (p *Pipeline) SecondCopy(ctx context.Context) { + if !p.Recogniser.Available() || p.SecondCopyBudget <= 0 { + return + } + deadline := time.Now().Add(p.SecondCopyBudget) + done, failed := 0, 0 + var breaker transcribe.Breaker + var failing []*store.Episode + + for _, e := range p.pendingRefetch() { + if ctx.Err() != nil || time.Now().After(deadline) { + break + } + found, secondMs, err := p.refetchOne(ctx, e) + if ctx.Err() != nil { + return + } + e.RefetchAttemptedAtUnix = time.Now().Unix() + if err != nil { + log.Printf("refetching %q: %v", e.Title, err) + failed++ + failing = append(failing, e) + if breaker.Failure() { + for _, given := range failing { + given.RefetchAttemptedAtUnix = 0 + } + log.Printf("refetching abandoned after %d failures in a row -- giving those "+ + "attempts back, since this is the machine and not the episodes", breaker.Run()) + p.Save() + return + } + p.Save() + continue + } + breaker.Success() + failing = nil + done++ + // The two lengths are the first thing to look at when this finds nothing: two fetches + // that came back the same length probably carried the same campaign, and there is + // then nothing for subtraction to see. + log.Printf("second copy of %q: %s against the mirrored %s, %d lines, %s", + e.Title, clock(secondMs), clock(e.DurationMs), e.MasterCues, adAccounting(e, found)) + p.Save() + } + if done > 0 || failed > 0 { + log.Printf("second copies -- %d done, %d failed, %d still waiting", + done, failed, len(p.pendingRefetch())) + } +} + +// refetchOne downloads one episode again, recognises it, keeps the words and throws the audio +// away. +func (p *Pipeline) refetchOne( + ctx context.Context, e *store.Episode, +) ([]transcript.Candidate, int64, error) { + // Its own id and its own directory, so nothing can mistake this for the copy the phone + // plays even for the moment it exists. + second := &store.Episode{ID: e.ID + "-second", EnclosureURL: e.EnclosureURL} + if err := p.SecondCopies.Fetch(ctx, second); err != nil { + return nil, 0, err + } + defer func() { _ = p.SecondCopies.Prune(second) }() + + result, err := p.Recogniser.Transcribe(ctx, filepath.Join(p.SecondCopyDir, second.Path)) + if err != nil { + return nil, second.DurationMs, err + } + if err := p.Transcripts.PutMaster(transcript.Transcript{ + EpisodeID: e.ID, + Source: transcript.SourceRefetch, + LanguageCode: result.LanguageCode, + Cues: result.Cues, + }); err != nil { + return nil, second.DurationMs, err + } + e.MasterCues = len(result.Cues) + e.MasterSource = string(transcript.SourceRefetch) + e.MasterNote = "" + + // Re-derive this episode's advertising now that there is something to subtract it from. + held, err := p.Transcripts.Get(e.ID) + if err != nil { + return nil, second.DurationMs, err + } + held.Ads = p.adsIn(e, held.Cues) + if err := p.Transcripts.Put(held); err != nil { + return nil, second.DurationMs, err + } + p.publishWordBreaks(e, held.Ads) + return held.Ads, second.DurationMs, nil +} + +/* +pendingRefetch is every episode with its own words, no master, and advertising to find. + +Newest first: an episode about to be played is worth the work before one from the back of a +catalogue. +*/ +func (p *Pipeline) pendingRefetch() []*store.Episode { + now := time.Now().Unix() + var pending []*store.Episode + for _, show := range p.State.Shows() { + for _, e := range show.Episodes { + if !e.NeedsRefetch(now) { + continue + } + // Without a length gap nothing was inserted, so a second fetch would be the same + // recording and there would be nothing to subtract. + if !transcript.HasInsertedAdvertising(e.DurationMs, e.DeclaredDurationMs) { + continue + } + pending = append(pending, e) + } + } + sort.Slice(pending, func(i, j int) bool { + return pending[i].PublishedUnix > pending[j].PublishedUnix + }) + return pending +} + +/* +pendingRecognition is every mirrored episode still without words, best first. + +An episode holding the publisher's ad-free master comes first, whatever its date. Recognising +one of those produces an answer that checks itself: subtraction found 4:20 of the 4:15 the +lengths imply on Darknet Diaries and 3:32 of 3:29 on Heavyweight. Recognising anything else +leaves the ad copy to be read, which across the first nineteen real episodes found 0%, 0%, 15% +and 29% of what the lengths say is there. Spending the budget where the result can be trusted +first is worth more than spending it in date order. + +Newest first within each group, because an episode about to be played is worth having words for +before one from the back of a catalogue. +*/ +func (p *Pipeline) pendingRecognition() []*store.Episode { + var pending []*store.Episode + for _, show := range p.State.Shows() { + for _, e := range show.Episodes { + if e.NeedsRecognising() { + pending = append(pending, e) + } + } + } + // Three things decide the order, and they are in the order they matter. A topical show is + // worthless by Thursday. An episode whose answer checks itself against the length gap is + // worth more than one that cannot. And within that, newest first. + rank := map[string]int{} + for _, show := range p.State.Shows() { + for _, e := range show.Episodes { + rank[e.ID] = tierRankOf(p.State.Tier(show.FeedURL)) + } + } + sort.SliceStable(pending, func(i, j int) bool { + left, right := pending[i], pending[j] + if rank[left.ID] != rank[right.ID] { + return rank[left.ID] < rank[right.ID] + } + if (left.MasterCues > 0) != (right.MasterCues > 0) { + return left.MasterCues > 0 + } + return left.PublishedUnix > right.PublishedUnix + }) + return pending +} diff --git a/server/internal/words/pipeline_test.go b/server/internal/words/pipeline_test.go new file mode 100644 index 0000000..180b807 --- /dev/null +++ b/server/internal/words/pipeline_test.go @@ -0,0 +1,137 @@ +package words + +import ( + "testing" + + "github.com/terst/listen/server/internal/store" + "github.com/terst/listen/server/internal/transcript" +) + +/* +The passes that used to live in cmd/listend, where nothing could reach them. + +Nine of them had accumulated in one 1,249-line command with no test file, which is how a +refetch master came to block the recurring detector for a fortnight: the condition was wrong in +a place no test could see. +*/ + +func TestForgettingAnEpisodeLeavesNothingBehind(t *testing.T) { + dir := t.TempDir() + store0 := transcript.Store{Dir: dir} + if err := store0.Put(transcript.Transcript{ + EpisodeID: "e1", Cues: []transcript.Cue{{EndMs: 1000, Text: "words"}}, + }); err != nil { + t.Fatalf("Put: %v", err) + } + if err := store0.PutMaster(transcript.Transcript{ + EpisodeID: "e1", Cues: []transcript.Cue{{EndMs: 1000, Text: "master"}}, + }); err != nil { + t.Fatalf("PutMaster: %v", err) + } + + episode := &store.Episode{ + ID: "e1", TranscriptCues: 400, TranscriptSource: "asr", MasterCues: 300, + MasterSource: "refetch", TranscribeAttemptedAtUnix: 1, RefetchAttemptedAtUnix: 1, + WordBreaks: []store.Break{{StartMs: 0, EndMs: 1000}}, + } + (&Pipeline{Transcripts: store0}).Forget(episode) + + if store0.Has("e1") || store0.HasMaster("e1") { + t.Error("files outlived the episode") + } + if episode.TranscriptCues != 0 || episode.MasterCues != 0 { + t.Errorf("counts survived: %+v", episode) + } + if episode.MasterSource != "" || episode.TranscriptSource != "" { + t.Errorf("sources survived: %+v", episode) + } + if len(episode.WordBreaks) != 0 { + t.Error("breaks from a transcript that no longer exists survived") + } +} + +/* +An episode must not keep a master count after the master file is gone. + +It would claim a second description it cannot produce, and subtraction would find nothing for +it for ever -- which is exactly how a refetch master blocked the recurring detector: the +episode looked like it had a better answer and did not. +*/ +func TestForgettingClearsTheAttemptsSoTheWorkCanBeRedone(t *testing.T) { + episode := &store.Episode{ + ID: "e1", TranscribeAttemptedAtUnix: 1_700_000_000, + RefetchAttemptedAtUnix: 1_700_000_000, Path: "e1.mp3", + } + (&Pipeline{Transcripts: transcript.Store{Dir: t.TempDir()}}).Forget(episode) + + if !episode.NeedsRecognising() { + t.Error("a re-mirrored episode would never be recognised again") + } +} + +/* +The work order: topical shows first, archive last. + +A five-hour back-catalogue episode arriving the same morning as the news would otherwise take +most of a cycle's recognition budget before the news was looked at. +*/ +func TestTheWorkOrderPutsTopicalShowsFirst(t *testing.T) { + if tierRankOf(store.TierTopical) >= tierRankOf(store.TierOrdinary) { + t.Error("a topical show does not come first") + } + if tierRankOf(store.TierArchive) <= tierRankOf(store.TierOrdinary) { + t.Error("an archive show does not come last") + } + if tierRankOf("") != tierRankOf(store.TierOrdinary) { + t.Error("a show nobody placed is not treated as ordinary, so it could be starved") + } +} + +func TestRecognitionIsOrderedByTierThenByWhatChecksItself(t *testing.T) { + state := store.New() + state.SetFeeds([]string{"https://news", "https://usual"}) + state.SetTiers(map[string]string{"https://news": store.TierTopical}) + state.Merge("https://usual", "Usual", nil) + state.Merge("https://news", "News", nil) + + usual := state.Show("https://usual") + usual.Episodes = []*store.Episode{ + {ID: "old-with-master", Path: "a.mp3", MasterCues: 10, PublishedUnix: 100}, + {ID: "new-plain", Path: "b.mp3", PublishedUnix: 900}, + } + news := state.Show("https://news") + news.Episodes = []*store.Episode{{ID: "news-old", Path: "c.mp3", PublishedUnix: 1}} + + got := (&Pipeline{State: state}).pendingRecognition() + + var order []string + for _, e := range got { + order = append(order, e.ID) + } + want := []string{"news-old", "old-with-master", "new-plain"} + for i := range want { + if i >= len(order) || order[i] != want[i] { + t.Fatalf("got %v, want %v", order, want) + } + } +} + +// A show with too few recognised episodes says nothing about what it repeats. +func TestRecurringNeedsEnoughEpisodesToMeanAnything(t *testing.T) { + state := store.New() + state.Merge("https://f", "A Show", nil) + show := state.Show("https://f") + show.Episodes = []*store.Episode{ + {ID: "e1", Path: "a.mp3", TranscriptCues: 10}, + {ID: "e2", Path: "b.mp3", TranscriptCues: 10}, + } + + // Two is below the floor, so nothing is read and nothing is published. + (&Pipeline{State: state, Transcripts: transcript.Store{Dir: t.TempDir()}}).Recurring(show) + + for _, e := range show.Episodes { + if len(e.WordBreaks) != 0 { + t.Errorf("%s published breaks from a corpus of two", e.ID) + } + } +} |
