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 }); }
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 [];
  return [...item.matchAll(/<podcast:transcript\s+([^>]*?)\/?>/gi)].map(m => ({ url: attr(m[1], 'url'), type: attr(m[1], 'type').toLowerCase() })).filter(x => x.url);
}
async function nativeTranscript(info) {
  if (!info || !info.feedUrl) return null;
  const links = transcriptLinks(await getText(info.feedUrl), info.guid, info.fileUrl);
  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 r; } catch {}
  }
  return null;
}

// ---- 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');
async function whisper(file, name) {
  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');
  const r = await fetch(cfg.api + '/v1/audio/transcriptions', { method: 'POST', body: form, signal: AbortSignal.timeout(6 * 3600e3) });
  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 fetchAudio(src, cookie, tmp) {
  const r = await fetch('http://127.0.0.1:' + cfg.port + src, { headers: { cookie: cookie || '' }, redirect: 'follow', signal: AbortSignal.timeout(2 * 3600e3) });
  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) {
  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); }
    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 }, () => {}); }
  });
}

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 -> time: 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 });
    }
    if (kind === 'pod' && !(nativeMiss.get(key) > Date.now() - 864e5)) {
      let r = null; try { r = await nativeTranscript(await podcast.feedInfo(me, key0.slice(4))); } catch {}
      if (r) { atomicWrite(fileOf(key), JSON.stringify(r)); return json(res, 200, { state: 'ready', ...base, transcript: r }); }
      nativeMiss.set(key, Date.now());
    }
    return json(res, 200, { state: 'none', ...base });
  } catch (e) { json(res, e.status || 500, { error: e.message }); }
}
module.exports = { init, handle, parseCues, parseJson, transcriptLinks, shape, whisper, getText, nativeTranscript, enabled: () => true };