1 files changed, 149 insertions, 3 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