diff options
| -rw-r--r-- | server/cmd/listend/main.go | 152 | ||||
| -rw-r--r-- | server/internal/store/store.go | 18 | ||||
| -rw-r--r-- | server/internal/store/store_test.go | 50 | ||||
| -rw-r--r-- | server/internal/transcript/fits.go | 10 | ||||
| -rw-r--r-- | server/internal/transcript/store.go | 15 |
5 files changed, 241 insertions, 4 deletions
diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go index 9914284..d8c9d0c 100644 --- a/server/cmd/listend/main.go +++ b/server/cmd/listend/main.go @@ -54,6 +54,8 @@ func main() { 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") + refetchBudget = flag.Duration("refetch-budget", 20*time.Minute, + "how long each cycle may spend fetching and recognising second copies") ) flag.Parse() @@ -100,6 +102,9 @@ func main() { WorkDir: filepath.Join(*dataDir, "work"), }, 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), } @@ -134,6 +139,7 @@ func main() { // 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) if err := state.Save(statePath); err != nil { log.Fatalf("save: %v", err) } @@ -194,8 +200,13 @@ type worker struct { // 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{} + // 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{} } // wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll. @@ -222,6 +233,7 @@ func (w *worker) run(ctx context.Context, every time.Duration) { w.pollAll(ctx) w.save() w.transcribeSome(ctx) + w.refetchSome(ctx) ticker := time.NewTicker(every) defer ticker.Stop() @@ -236,6 +248,7 @@ func (w *worker) run(ctx context.Context, every time.Duration) { // 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) case <-w.wakeups: w.pollAll(ctx) w.save() @@ -633,7 +646,10 @@ func adAccounting(e *store.Episode, found []transcript.Candidate) string { total += c.EndMs - c.StartMs } how := "read from the ad copy" - if e.MasterCues > 0 { + 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 @@ -660,6 +676,136 @@ func adAccounting(e *store.Episode, found []transcript.Candidate) string { } /* +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, 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++ + log.Printf("second copy of %q: %d lines, %s", e.Title, 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, 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, err + } + defer func() { _ = w.refetchMirror.Prune(second) }() + + result, err := w.recogniser.Transcribe(ctx, filepath.Join(w.refetchDir, second.Path)) + if err != nil { + return nil, err + } + if err := w.transcripts.PutMaster(transcript.Transcript{ + EpisodeID: e.ID, + Source: transcript.SourceRefetch, + LanguageCode: result.LanguageCode, + Cues: result.Cues, + }); err != nil { + return nil, 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, err + } + held.Ads = w.adsIn(e, held.Cues) + if err := w.transcripts.Put(held); err != nil { + return nil, err + } + return held.Ads, 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 { + var pending []*store.Episode + for _, show := range w.state.Shows() { + for _, e := range show.Episodes { + if !e.NeedsRefetch() { + 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 diff --git a/server/internal/store/store.go b/server/internal/store/store.go index 7f81bd5..2a6992b 100644 --- a/server/internal/store/store.go +++ b/server/internal/store/store.go @@ -73,6 +73,14 @@ type Episode struct { // states as everywhere else: words, a reason, or nobody has looked. MasterCues int `json:"masterCues,omitempty"` MasterNote string `json:"masterNote,omitempty"` + // MasterSource says what kind of master it is: "published" for the publisher's own ad-free + // transcript, "refetch" for a recognised second fetch of the same episode. They are not + // equally good and the readout should not pretend otherwise -- a refetch carries its own + // advertising, so anything both copies happen to share is invisible to it. + MasterSource string `json:"masterSource,omitempty"` + // RefetchAttemptedAtUnix is when a second copy was last fetched for this episode, so a + // failure is not retried every cycle. Recorded only when an attempt finished. + RefetchAttemptedAtUnix int64 `json:"refetchAttemptedAtUnix,omitempty"` // TranscribeAttemptedAtUnix is when speech recognition last ran over this episode, whether // it worked or not, so a failure is not retried on every poll for ever. Recorded only when @@ -110,6 +118,16 @@ func (e *Episode) NeedsMaster() bool { e.MasterCues == 0 && e.MasterNote == "" } +// NeedsRefetch reports whether a second copy of this episode is worth fetching. +// +// Only for an episode that has its own words but no master to subtract them from, and only +// where the lengths say advertising was inserted: without a gap the two fetches would be the +// same recording and there would be nothing to find. +func (e *Episode) NeedsRefetch() bool { + return e.Mirrored() && e.HasTranscript() && e.EnclosureURL != "" && + e.MasterCues == 0 && e.MasterNote == "" && e.RefetchAttemptedAtUnix == 0 +} + // NeedsRecognising reports whether speech recognition should be run over this episode. // // Deliberately not the negation of [Episode.TranscriptExamined]. An episode whose publisher's diff --git a/server/internal/store/store_test.go b/server/internal/store/store_test.go index c04dc29..9020b17 100644 --- a/server/internal/store/store_test.go +++ b/server/internal/store/store_test.go @@ -435,3 +435,53 @@ func TestAFinishedAttemptIsNotRepeated(t *testing.T) { t.Error("an episode already attempted would be attempted again") } } + +/* +Which episodes are worth fetching a second time. + +Advertising is inserted per request, so two fetches of the same episode carry different +campaigns and agree on the programme. Subtracting one from the other therefore says where the +advertising is -- the same arithmetic as a publisher's master, with a second recording standing +in for one. + +Only worth doing where there is advertising to find. An episode whose served length matches what +the feed declares has none inserted, and the second fetch would be the same recording. +*/ + +func TestAnEpisodeWithItsOwnWordsAndNoMasterIsWorthRefetching(t *testing.T) { + e := Episode{ + Path: "abc.mp3", EnclosureURL: "https://cdn/1.mp3", TranscriptCues: 400, + } + if !e.NeedsRefetch() { + t.Error("an episode with words and nothing to subtract them from was passed over") + } +} + +func TestAnEpisodeThatAlreadyHasAMasterIsNotRefetched(t *testing.T) { + e := Episode{ + Path: "abc.mp3", EnclosureURL: "https://cdn/1.mp3", TranscriptCues: 400, MasterCues: 300, + } + if e.NeedsRefetch() { + t.Error("would fetch a second copy for an episode that already has a master") + } +} + +// Nothing to subtract from until its own words exist. +func TestAnEpisodeWithNoWordsYetIsNotRefetched(t *testing.T) { + e := Episode{Path: "abc.mp3", EnclosureURL: "https://cdn/1.mp3"} + if e.NeedsRefetch() { + t.Error("would fetch a second copy before there was a first to compare it against") + } +} + +// A finished attempt is not repeated: a publisher that will not serve it again will not serve +// it again next cycle either. +func TestAFinishedRefetchIsNotRepeated(t *testing.T) { + e := Episode{ + Path: "abc.mp3", EnclosureURL: "https://cdn/1.mp3", TranscriptCues: 400, + RefetchAttemptedAtUnix: 1_700_000_000, + } + if e.NeedsRefetch() { + t.Error("would fetch a second copy again") + } +} diff --git a/server/internal/transcript/fits.go b/server/internal/transcript/fits.go index b0495d0..c22958b 100644 --- a/server/internal/transcript/fits.go +++ b/server/internal/transcript/fits.go @@ -76,6 +76,16 @@ func Fits(lastCueEndMs, servedMs, declaredMs int64) Verdict { return Verdict{OK: true} } +// HasInsertedAdvertising reports whether this copy is longer than the feed says the programme +// is, by more than the feed could be approximate about. +// +// The one fact about an episode that comes from neither a detector nor a transcript. An episode +// without it has nothing inserted to find, so there is no point fetching a second copy to +// compare against -- the two would be the same recording. +func HasInsertedAdvertising(servedMs, declaredMs int64) bool { + return declaredMs > 0 && servedMs-declaredMs > insertionMs +} + // LastCueEndMs is where a transcript stops describing the audio. func LastCueEndMs(cues []Cue) int64 { if len(cues) == 0 { diff --git a/server/internal/transcript/store.go b/server/internal/transcript/store.go index 9450b3a..50feae4 100644 --- a/server/internal/transcript/store.go +++ b/server/internal/transcript/store.go @@ -19,6 +19,15 @@ const ( // SourceASR is speech recognition over the mirrored bytes: approximate words, of exactly // the audio the phone will play. SourceASR Source = "asr" + // SourceRefetch is speech recognition over a *second* fetch of the same episode, kept as a + // master for subtraction. It is not ad-free -- it carries whatever campaign was running + // when it was fetched -- but advertising is inserted per request, so the two copies differ + // in exactly the places the advertising sits. What they share is the programme. + // + // Subtracting it therefore reports the mirrored copy's advertising minus anything both + // copies happened to carry, which is the one thing it cannot see. That is the opposite + // blind spot to the audio scan, which only finds what a batch of episodes shares. + SourceRefetch Source = "refetch" ) // Transcript is one episode's words and when they were said. @@ -110,9 +119,13 @@ func (s Store) Put(t Transcript) error { // inserted into, and subtracting it from a recognised reading of the served file is what says // where. See Inserted. func (s Store) PutMaster(t Transcript) error { + source := t.Source + if source == "" { + source = SourcePublished + } return s.Put(Transcript{ EpisodeID: t.EpisodeID + masterSuffix, - Source: SourcePublished, + Source: source, LanguageCode: t.LanguageCode, Cues: t.Cues, }) |