// 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 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(/&/g, '&').replace(/</g, '<').replace(/>/g, '>').replace(/"/g, '"').replace(/'/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(/&/g, '&') : ''; }; /** The transcript links an episode has in its feed. */ function transcriptLinks(feedXml, guid, fileUrl) { const items = String(feedXml).split(/]/i).slice(1); const unwrap = x => String(x || '').replace(//g, '').trim(); const item = items.find(b => { const g = /]*>([\s\S]*?)<\/guid>/i.exec(b); return g && guid && unwrap(g[1]) === guid; }) || items.find(b => { const e = /]*>/i.exec(b); return e && fileUrl && attr(e[0], 'url') === fileUrl; }); if (!item) return null; return [...item.matchAll(/]*?)\/?>/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 };