hub/transcripts.js

// Words for spoken audio (podcasts, audiobooks), shown in the same screen as song lyrics.
// 1. A transcript the show itself publishes (the Podcast 2.0 <podcast:transcript> tag in its feed: SRT, VTT or JSON), fetched on request.
// 2. Otherwise, speech-to-text on this server: an OpenAI-compatible Whisper service (TRANSCRIBE_API_URL, e.g. the bundled faster-whisper/speaches
//    container). Only on request, one job at a time; the result is kept on disk, so an episode is transcribed once.
const fs = require('fs'), path = require('path'), os = require('os'), crypto = require('crypto');
const { Readable } = require('stream'), { pipeline } = require('stream/promises');
const atomicWrite = require('./atomic');
const users = require('./users');
const podcast = require('./podcast');

const cfg = { api: '', model: '', dir: '', port: 8080 };
const MAX_AUDIO = 1.5 * 1024 ** 3, MAX_TEXT = 6 * 1024 * 1024;
function init({ dataDir, apiUrl, model, port }) { cfg.api = (apiUrl || '').replace(/\/$/, ''); cfg.model = model || 'Systran/faster-whisper-small'; cfg.dir = path.join(dataDir, 'transcripts'); cfg.port = port || 8080; fs.mkdirSync(cfg.dir, { recursive: true, mode: 0o700 }); }
/** "fetch failed" says nothing: name the step and the reason (connection refused, timed out...). */
const why = e => { const c = e && e.cause; return (c && (c.code || c.message)) || (e && e.message) || 'unknown error'; };
const fail = (status, message) => Object.assign(new Error(message), { status });
const json = (res, code, o) => { res.writeHead(code, { 'content-type': 'application/json', 'cache-control': 'no-store' }); res.end(JSON.stringify(o)); };

// ---- reading transcript files
const tms = (h, m, s, f) => (+(h || 0) * 3600 + +m * 60 + +s) * 1000 + Math.round(+('0.' + (f || '0')) * 1000);
const clean = t => String(t || '').replace(/<[^>]+>/g, '').replace(/&amp;/g, '&').replace(/&lt;/g, '<').replace(/&gt;/g, '>').replace(/&quot;/g, '"').replace(/&#39;/g, "'").replace(/\s+/g, ' ').trim();
/** SubRip and WebVTT: "00:01:02,500 --> 00:01:05,000" then the words. */
function parseCues(text) {
  const out = [], re = /(?:(\d+):)?(\d{1,2}):(\d{2})[.,](\d{1,3})\s*-->\s*(?:(\d+):)?(\d{1,2}):(\d{2})[.,](\d{1,3})[^\n]*\n([\s\S]*?)(?=\n\s*\n|$)/g; let m;
  while ((m = re.exec(String(text).replace(/\r/g, '')))) { const body = clean(m[9]); if (body) out.push({ t: tms(m[1], m[2], m[3], m[4]), text: body }); }
  return out;
}
/** Podcast 2.0 JSON: { segments: [{ startTime, endTime, speaker, body }] } */
function parseJson(text) {
  let j; try { j = JSON.parse(text); } catch { return []; }
  return (j.segments || []).map(s => ({ t: Math.round(+s.startTime * 1000), text: (s.speaker ? s.speaker + ': ' : '') + clean(s.body) })).filter(x => x.text && Number.isFinite(x.t));
}
const withEnds = lines => { lines.sort((a, b) => a.t - b.t); lines.forEach((l, i) => { l.e = i + 1 < lines.length ? lines[i + 1].t : l.t + 6000; }); return lines; };
const shape = (lines, source) => lines.length ? { found: true, source, sync: 'line', lines: withEnds(lines) } : null;

// ---- the show's own transcript
const PRIVATE = /^(localhost|.*\.local|.*\.internal|0\.0\.0\.0|127\.|10\.|192\.168\.|169\.254\.|172\.(1[6-9]|2\d|3[01])\.|\[?::1\]?$|\[?f[cd][0-9a-f]{2}:)/i;
async function getText(url, hops = 0) { // only public http(s) addresses; redirects are checked again
  const u = new URL(url); if (!/^https?:$/.test(u.protocol) || PRIVATE.test(u.hostname)) throw fail(400, 'That address is not allowed');
  const r = await fetch(u, { redirect: 'manual', signal: AbortSignal.timeout(15000), headers: { 'user-agent': 'NEHub/1 (+transcripts)', accept: 'text/*, application/json, application/xml, */*' } });
  if (r.status >= 300 && r.status < 400 && r.headers.get('location') && hops < 3) return getText(new URL(r.headers.get('location'), u).href, hops + 1);
  if (!r.ok) throw fail(502, 'Status ' + r.status);
  const len = +r.headers.get('content-length') || 0; if (len > MAX_TEXT) throw fail(413, 'Too large');
  const text = await r.text(); if (text.length > MAX_TEXT) throw fail(413, 'Too large'); return text;
}
const attr = (s, k) => { const m = new RegExp(k + '\\s*=\\s*"([^"]*)"', 'i').exec(s); return m ? m[1].replace(/&amp;/g, '&') : ''; };
/** The transcript links an episode has in its feed. */
function transcriptLinks(feedXml, guid, fileUrl) {
  const items = String(feedXml).split(/<item[\s>]/i).slice(1);
  const unwrap = x => String(x || '').replace(/<!\[CDATA\[|\]\]>/g, '').trim();
  const item = items.find(b => { const g = /<guid[^>]*>([\s\S]*?)<\/guid>/i.exec(b); return g && guid && unwrap(g[1]) === guid; })
    || items.find(b => { const e = /<enclosure[^>]*>/i.exec(b); return e && fileUrl && attr(e[0], 'url') === fileUrl; });
  if (!item) return null;
  return [...item.matchAll(/<podcast:transcript\s+([^>]*?)\/?>/gi)].map(m => ({ url: attr(m[1], 'url'), type: attr(m[1], 'type').toLowerCase() })).filter(x => x.url);
}
/** { transcript } when the show publishes one we can read, otherwise { why }: the reason, so a missing transcript can be explained. */
async function nativeTranscript(info) {
  if (!info) return { why: 'Could not read this episode from the podcast service.' };
  if (!info.feedUrl) return { why: 'The podcast service did not give the show\'s feed address.' };
  let xml; try { xml = await getText(info.feedUrl); } catch (e) { return { why: 'Could not download the show\'s feed (' + e.message + ').' }; }
  const links = transcriptLinks(xml, info.guid, info.fileUrl);
  if (!links) return { why: 'Could not find this episode in the show\'s feed (looked for ' + (info.guid ? 'id ' + String(info.guid).slice(0, 60) : 'its audio address') + ').' };
  if (!links.length) return { why: 'The show\'s feed lists no transcript for this episode.' };
  const rank = l => /json/.test(l.type) ? 0 : /vtt/.test(l.type) ? 1 : /subrip|srt/.test(l.type) ? 2 : 9;
  for (const l of links.sort((a, b) => rank(a) - rank(b))) {
    if (rank(l) > 2) continue; // (plain HTML or text has no times: not worth showing as a transcript)
    try { const t = await getText(l.url), lines = /json/.test(l.type) ? parseJson(t) : parseCues(t); const r = shape(lines, 'Published by the show'); if (r) return { transcript: r }; } catch {}
  }
  return { why: 'The show lists a transcript (' + links.map(l => l.type || 'unknown type').join(', ') + ') but not in a timed format this can follow, or it could not be downloaded.' };
}

// ---- speech to text on this server
const jobs = new Map(); // key -> { state: 'running' | 'error', since, error }
let chain = Promise.resolve(); // one at a time: it is heavy
const fileOf = key => path.join(cfg.dir, key + '.json');
const keyFor = (kind, id, src) => crypto.createHash('sha1').update(kind + '|' + id + '|' + src).digest('hex');
let modelReady = null; // the service must have the model on disk before it can use it: check, and ask for it once if it is missing
async function ensureModel() {
  const id = cfg.model, enc = id.split('/').map(encodeURIComponent).join('/'), seen = [];
  const call = async (method, url, body) => { const r = await fetch(cfg.api + url, { method, headers: body ? { 'content-type': 'application/json' } : {}, body, signal: AbortSignal.timeout(30 * 60e3) }); if (r.ok || r.status === 409) return true; seen.push(`${method} ${url}: ${r.status} ${(await r.text()).slice(0, 120)}`); return false; };
  if ((await fetch(cfg.api + '/v1/models/' + enc, { signal: AbortSignal.timeout(20000) })).ok) return; // already installed
  // (the download call has differed between versions of the service, so the known forms are tried in turn)
  if (await call('POST', '/v1/models/' + enc) || await call('POST', '/v1/models', JSON.stringify({ id })) || await call('POST', '/v1/models?model_name=' + encodeURIComponent(id))) return;
  throw new Error('The transcription service would not download the model ' + id + ' (' + seen.join('; ') + ')');
}
const modelOnce = () => modelReady || (modelReady = ensureModel().catch(e => { modelReady = null; throw e; }));
async function whisper(file, name) {
  try { await modelOnce(); } catch (e) { throw /would not download/.test(e.message) ? e : new Error(`Could not reach the transcription service at ${cfg.api} (${why(e)}). Is the whisper container running? Check its log.`); }
  const form = new FormData();
  form.append('file', await fs.openAsBlob(file), name || 'audio');
  form.append('model', cfg.model); form.append('response_format', 'verbose_json'); form.append('timestamp_granularities[]', 'segment');
  let r; try { r = await fetch(cfg.api + '/v1/audio/transcriptions', { method: 'POST', body: form, signal: AbortSignal.timeout(6 * 3600e3) }); }
  catch (e) { throw new Error(`Could not reach the transcription service at ${cfg.api} (${why(e)}). Is the whisper container running? Check its log.`); }
  if (!r.ok) throw new Error('The transcription service said ' + r.status + ': ' + (await r.text()).slice(0, 200));
  const j = await r.json();
  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)})`); }
  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 }));
}
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 { 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) {
  const u = new URL(req.url, 'http://x'), q = u.searchParams, key0 = String(q.get('key') || ''), src = String(q.get('src') || '');
  const kind = key0.startsWith('pod:') ? 'pod' : key0.startsWith('ab:') ? 'ab' : '';
  if (!kind || key0.length > 200 || !SRC_OK.some(re => re.test(src))) return json(res, 400, { error: 'That cannot be transcribed' });
  const key = keyFor(kind, key0, src), me = users.current();
  const base = { canTranscribe: !!cfg.api };
  try {
    if (fs.existsSync(fileOf(key))) return json(res, 200, { state: 'ready', ...base, transcript: JSON.parse(fs.readFileSync(fileOf(key), 'utf8')) });
    const job = jobs.get(key); if (job) return json(res, 200, { state: job.state, since: job.since, error: job.error, ...base });
    if (req.method === 'POST') {
      if (!cfg.api) return json(res, 400, { error: 'Transcribing is not set up on this server (see the whisper service in docker-compose.yml)' });
      startJob(key, src, req.headers.cookie); return json(res, 200, { state: 'running', since: Date.now(), ...base });
    }
    let why = '';
    if (kind === 'pod') {
      const miss = nativeMiss.get(key);
      if (miss && miss.t > Date.now() - 6 * 3600e3) why = miss.why;
      else {
        let r; try { r = await nativeTranscript(await podcast.feedInfo(me, key0.slice(4))); } catch (e) { r = { why: 'Checking the show\'s feed failed (' + e.message + ').' }; }
        if (r.transcript) { atomicWrite(fileOf(key), JSON.stringify(r.transcript)); return json(res, 200, { state: 'ready', ...base, transcript: r.transcript }); }
        why = r.why; nativeMiss.set(key, { t: Date.now(), why });
      }
    }
    return json(res, 200, { state: 'none', why, ...base });
  } catch (e) { json(res, e.status || 500, { error: e.message }); }
}
module.exports = { init, startAuto, status, handle, parseCues, parseJson, transcriptLinks, shape, whisper, getText, nativeTranscript, enabled: () => true };