summaryrefslogtreecommitdiff
path: root/server/cmd
diff options
context:
space:
mode:
Diffstat (limited to 'server/cmd')
-rw-r--r--server/cmd/listend/main.go696
1 files changed, 39 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