87 lines
6.7 KiB
JavaScript
87 lines
6.7 KiB
JavaScript
// Explicit Whisper requests produce reviewable drafts, never published notes.
|
|
import { randomUUID } from 'node:crypto';
|
|
import { db as sql } from './db.js';
|
|
import { audit } from './audit-log.js';
|
|
const ID = /^([\w-]{11}|upl_[a-f0-9]{12})$/;
|
|
const LEASE_MS = 120000;
|
|
function publicJob(row) {
|
|
if (!row) return null;
|
|
return { id: row.id, videoId: row.video_id, status: row.status, stage: row.stage,
|
|
baseRev: Number(row.base_rev), result: row.result ? JSON.parse(row.result) : null,
|
|
error: row.error, createdAt: Number(row.created_at), updatedAt: Number(row.updated_at) };
|
|
}
|
|
|
|
export function registerTranscriptionRoutes(app, { db, adminAuth, workerAuth, workerEnabled, sanitizeLyrics, now = Date.now }) {
|
|
const seen = () => sql.execute({ sql: 'INSERT INTO lyrics_worker_state (id,last_seen) VALUES (1,?) ON CONFLICT(id) DO UPDATE SET last_seen=excluded.last_seen', args: [now()] });
|
|
async function source(id) {
|
|
if (id.startsWith('upl_')) return db.getUpload(id);
|
|
const row = await db.getMedia(id); return row?.status === 'ready' ? row : null;
|
|
}
|
|
async function enqueue(videoId) {
|
|
const id = String(videoId || '');
|
|
if (!ID.test(id)) return { status: 400, body: { ok: false, error: 'invalid video id' } };
|
|
if (!workerEnabled) return { status: 503, body: { ok: false, error: 'Whisper is unavailable: the lyrics worker is not configured.' } };
|
|
const audio = await source(id);
|
|
if (!audio) return { status: 409, body: { ok: false, error: 'Save this video on the server before transcribing its audio.' } };
|
|
if (Number(audio.duration) > 3600) return { status: 422, body: { ok: false, error: 'Whisper requests are limited to one hour of audio.' } };
|
|
const notes = await db.getNotes(id);
|
|
// One active job per song and a bounded queue, including across restarts.
|
|
const existing = (await sql.execute({ sql: "SELECT * FROM lyric_transcriptions WHERE video_id=? AND status IN ('queued','running')", args: [id] })).rows[0];
|
|
if (existing) return { status: 200, body: { ok: true, job: publicJob(existing) } };
|
|
const active = Number((await sql.execute("SELECT COUNT(*) AS n FROM lyric_transcriptions WHERE status IN ('queued','running')")).rows[0].n);
|
|
if (active >= 25) return { status: 429, body: { ok: false, error: 'The transcription queue is full. Try again after a job finishes.' } };
|
|
const time = now();
|
|
await sql.execute({ sql: `INSERT OR IGNORE INTO lyric_transcriptions (id,video_id,base_rev,created_at,updated_at)
|
|
SELECT ?,?,?,?,? WHERE (SELECT COUNT(*) FROM lyric_transcriptions WHERE status IN ('queued','running')) < 25`,
|
|
args: [randomUUID(), id, notes.lyrics?.rev || 0, time, time] });
|
|
const row = (await sql.execute({ sql: "SELECT * FROM lyric_transcriptions WHERE video_id=? AND status IN ('queued','running')", args: [id] })).rows[0];
|
|
if (row) await audit(db, 'admin', 'lyrics.whisper.enqueue', id, { jobId: row.id });
|
|
return row ? { status: 202, body: { ok: true, job: publicJob(row) } } : { status: 429, body: { ok: false, error: 'The transcription queue is full.' } };
|
|
}
|
|
app.post('/api/admin/transcriptions/:video', adminAuth, async c => {
|
|
const result = await enqueue(c.req.param('video'));
|
|
return c.json(result.body, result.status);
|
|
});
|
|
app.get('/api/admin/transcriptions/:video', adminAuth, async c => {
|
|
const id = c.req.param('video');
|
|
if (!ID.test(id)) return c.json({ ok: false, error: 'invalid video id' }, 400);
|
|
const row = (await sql.execute({ sql: 'SELECT * FROM lyric_transcriptions WHERE video_id=? ORDER BY created_at DESC,rowid DESC LIMIT 1', args: [id] })).rows[0];
|
|
const worker = (await sql.execute('SELECT last_seen FROM lyrics_worker_state WHERE id=1')).rows[0];
|
|
return c.json({ ok: true, job: publicJob(row), workerOnline: !!worker && now() - Number(worker.last_seen) < 90000, enabled: !!workerEnabled }, 200, { 'Cache-Control': 'no-store' });
|
|
});
|
|
app.post('/api/lyrics-worker/claim', workerAuth, async c => {
|
|
await seen(); const time = now();
|
|
await sql.execute({ sql: "UPDATE lyric_transcriptions SET status='failed',stage='failed',error='The worker stopped repeatedly. Please retry.',updated_at=? WHERE status='running' AND lease_until<? AND attempts>=3", args: [time, time] });
|
|
// UPDATE RETURNING makes claiming atomic for multiple worker processes.
|
|
const row = (await sql.execute({ sql: `UPDATE lyric_transcriptions SET status='running',stage='loading-model',attempts=attempts+1,
|
|
lease_token=?,lease_until=?,updated_at=? WHERE id=(SELECT id FROM lyric_transcriptions
|
|
WHERE (status='queued' OR (status='running' AND lease_until<?)) AND attempts<3 ORDER BY created_at LIMIT 1) RETURNING *`,
|
|
args: [randomUUID(), time + LEASE_MS, time, time] })).rows[0];
|
|
return c.json({ ok: true, job: row ? { ...publicJob(row), lease: row.lease_token,
|
|
audioPath: row.video_id.startsWith('upl_') ? `/api/uploads/${row.video_id}` : `/api/media/${row.video_id}?a=1` } : null });
|
|
});
|
|
app.post('/api/lyrics-worker/jobs/:job', workerAuth, async c => {
|
|
let body;
|
|
try { const raw = await c.req.text(); if (raw.length > 400000) return c.json({ ok: false, error: 'transcript too large' }, 413); body = JSON.parse(raw); }
|
|
catch { return c.json({ ok: false, error: 'invalid JSON' }, 400); }
|
|
if (!body || typeof body.lease !== 'string') return c.json({ ok: false, error: 'missing lease' }, 400);
|
|
const time = now();
|
|
let result = null, status = 'running', error = null;
|
|
let stage = ['loading-model', 'downloading-audio', 'transcribing'].includes(body.stage) ? body.stage : 'transcribing';
|
|
if (body.status === 'complete') {
|
|
try { result = sanitizeLyrics(body.result); if (!result.lines.length) throw Error('No sung lyrics were found.'); }
|
|
catch (e) { return c.json({ ok: false, error: e.message }, 400); }
|
|
result.tags = ['auto-transcribed (whisper)']; status = 'complete'; stage = 'complete';
|
|
} else if (body.status === 'failed') { status = 'failed'; stage = 'failed'; error = String(body.error || 'Transcription failed.').slice(0, 300); }
|
|
const changed = await sql.execute({ sql: `UPDATE lyric_transcriptions SET status=?,stage=?,result=?,error=?,lease_until=?,updated_at=?
|
|
WHERE id=? AND status='running' AND lease_token=? AND lease_until>=?`,
|
|
args: [status, stage, result ? JSON.stringify(result) : null, error, time + LEASE_MS, time, c.req.param('job'), body.lease, time] });
|
|
if (!changed.rowsAffected) return c.json({ ok: false, error: 'This job lease has expired or finished.' }, 409);
|
|
await seen();
|
|
// Bound completed draft retention; active jobs are never deleted.
|
|
await sql.execute({ sql: "DELETE FROM lyric_transcriptions WHERE status IN ('complete','failed') AND updated_at<?", args: [time - 30 * 86400000] });
|
|
return c.json({ ok: true });
|
|
});
|
|
return { enqueue };
|
|
}
|