Transcripts: new and existing episodes are transcribed in the background (published transcript first, else whisper), with a status line in Health

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

transcripts.js

@@ -96,6 +96,12 @@ async function whisper(file, name) {
   return shape((j.segments || []).map(s => ({ t: Math.round(+s.start * 1000), text: clean(s.text) })).filter(x => x.text), 'Transcribed on your server');
 }
 /** Bring the audio over from this hub (as the person who asked, with their sign-in) into a temp file. */
+async function saveAudio(r, tmp) {
+  if (!r.ok || !r.body) throw new Error('Could not read the audio (' + r.status + ')');
+  if ((+r.headers.get('content-length') || 0) > MAX_AUDIO) throw new Error('That audio file is too large to transcribe');
+  let n = 0; const body = Readable.fromWeb(r.body); body.on('data', c => { n += c.length; if (n > MAX_AUDIO) body.destroy(new Error('That audio file is too large to transcribe')); });
+  await pipeline(body, fs.createWriteStream(tmp, { mode: 0o600 }));
+}
 async function fetchAudio(src, cookie, tmp) {
   let r; try { r = await fetch('http://127.0.0.1:' + cfg.port + src, { headers: { cookie: cookie || '' }, redirect: 'follow', signal: AbortSignal.timeout(2 * 3600e3) }); }
   catch (e) { throw new Error(`Could not read the audio from this server (${why(e)})`); }
@@ -104,16 +110,40 @@ async function fetchAudio(src, cookie, tmp) {
   let n = 0; const body = Readable.fromWeb(r.body); body.on('data', c => { n += c.length; if (n > MAX_AUDIO) body.destroy(new Error('That audio file is too large to transcribe')); });
   await pipeline(body, fs.createWriteStream(tmp, { mode: 0o600 }));
 }
-function startJob(key, src, cookie) {
+function startJob(key, src, cookie, getAudio) {
   jobs.set(key, { state: 'running', since: Date.now() });
   chain = chain.then(async () => {
     const tmp = path.join(os.tmpdir(), 'nehub-tr-' + key);
-    try { await fetchAudio(src, cookie, tmp); const r = await whisper(tmp, 'audio'); if (!r) throw new Error('No speech was found in that audio'); atomicWrite(fileOf(key), JSON.stringify(r)); jobs.delete(key); }
+    try { if (getAudio) await saveAudio(await getAudio(), tmp); else await fetchAudio(src, cookie, tmp); const r = await whisper(tmp, 'audio'); if (!r) throw new Error('No speech was found in that audio'); atomicWrite(fileOf(key), JSON.stringify(r)); jobs.delete(key); }
     catch (e) { jobs.set(key, { state: 'error', since: Date.now(), error: e.message }); setTimeout(() => jobs.delete(key), 10 * 60e3).unref(); }
     finally { fs.rm(tmp, { force: true }, () => {}); }
   });
 }
 
+// ---- in the background: new episodes as they are pulled, and the ones already in the library that have no transcript yet
+const auto = { on: false, keep: 5, tried: new Map(), last: 0, timer: null };
+const podSrc = id => '/api/podcast/podcastitems/' + id + '/file';
+async function autoTick() {
+  if (!cfg.api || !auto.on || [...jobs.values()].some(j => j.state === 'running')) return; // (one job at a time: the next round looks again when this one is done)
+  let queued = 0; const feeds = new Map();
+  for (const ep of await podcast.recentEpisodes(auto.keep)) {
+    if (queued >= 2) break;
+    const key = keyFor('pod', 'pod:' + ep.id, podSrc(ep.id)), t = auto.tried.get(key) || { n: 0, at: 0 };
+    if (fs.existsSync(fileOf(key)) || jobs.has(key) || t.n >= 3 || Date.now() - t.at < 6 * 3600e3 * t.n) continue; // (a failing one is retried a few times, further apart each time)
+    auto.tried.set(key, { n: t.n + 1, at: Date.now() });
+    let r = { why: '' }; try { r = await nativeTranscript(ep); } catch {}
+    if (r.transcript) { atomicWrite(fileOf(key), JSON.stringify(r.transcript)); continue; } // (the show publishes one: no need to make one)
+    startJob(key, '', '', () => podcast.episodeAudio(ep.id)); queued++;
+  }
+}
+function startAuto({ on, keep }) {
+  auto.on = !!on && !!cfg.api; auto.keep = Math.max(1, Math.min(50, +keep || 5));
+  if (auto.timer || !auto.on) return;
+  auto.timer = setInterval(() => autoTick().catch(e => console.warn('[transcripts]', e.message)), 10 * 60e3); auto.timer.unref();
+  setTimeout(() => autoTick().catch(e => console.warn('[transcripts]', e.message)), 90e3).unref();
+}
+function status() { let saved = 0; try { saved = fs.readdirSync(cfg.dir).filter(x => x.endsWith('.json')).length; } catch {} return { enabled: !!cfg.api, auto: auto.on, keep: auto.keep, saved, working: [...jobs.values()].filter(j => j.state === 'running').length, failed: [...jobs.values()].filter(j => j.state === 'error').length }; }
+
 const SRC_OK = [/^\/api\/podcast\/podcastitems\/[\w-]+\/file$/, /^\/api\/ab\/books\/[^?#\s]+$/]; // only audio this hub already serves to the person
 const nativeMiss = new Map(); // key -> { t, why }: a show with no transcript is not asked again for a day
 async function handle(req, res) {
@@ -142,4 +172,4 @@ async function handle(req, res) {
     return json(res, 200, { state: 'none', why, ...base });
   } catch (e) { json(res, e.status || 500, { error: e.message }); }
 }
-module.exports = { init, handle, parseCues, parseJson, transcriptLinks, shape, whisper, getText, nativeTranscript, enabled: () => true };
+module.exports = { init, startAuto, status, handle, parseCues, parseJson, transcriptLinks, shape, whisper, getText, nativeTranscript, enabled: () => true };