Add resumable bulk lyrics operations
This commit is contained in:
Jonathan Sykes
@@ -1,12 +1,31 @@
|
||||
import { audit } from './audit-log.js';
|
||||
import { mkdir, readFile, readdir, stat, writeFile } from 'node:fs/promises';
|
||||
import { join } from 'node:path';
|
||||
import { basename, join } from 'node:path';
|
||||
|
||||
const clientOf = db => typeof db?.execute === 'function' ? db : db?.db;
|
||||
const parse = value => { try { return JSON.parse(value || '{}'); } catch { return {}; } };
|
||||
const number = value => Number(value) || 0;
|
||||
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
|
||||
const zipCrcTable = Uint32Array.from({ length: 256 }, (_, value) => {
|
||||
let crc = value; for (let bit = 0; bit < 8; bit++) crc = crc & 1 ? 0xedb88320 ^ (crc >>> 1) : crc >>> 1;
|
||||
return crc >>> 0;
|
||||
});
|
||||
function zipStore(files) {
|
||||
const local = [], central = []; let offset = 0;
|
||||
for (const file of files) {
|
||||
const name = Buffer.from(file.name), data = Buffer.from(file.content);
|
||||
let crc = 0xffffffff; for (const byte of data) crc = zipCrcTable[(crc ^ byte) & 0xff] ^ (crc >>> 8); crc = (crc ^ 0xffffffff) >>> 0;
|
||||
const header = Buffer.alloc(30); header.writeUInt32LE(0x04034b50, 0); header.writeUInt16LE(20, 4); header.writeUInt16LE(0x0800, 6); header.writeUInt32LE(crc, 14); header.writeUInt32LE(data.length, 18); header.writeUInt32LE(data.length, 22); header.writeUInt16LE(name.length, 26);
|
||||
local.push(header, name, data);
|
||||
const directory = Buffer.alloc(46); directory.writeUInt32LE(0x02014b50, 0); directory.writeUInt16LE(20, 4); directory.writeUInt16LE(20, 6); directory.writeUInt16LE(0x0800, 8); directory.writeUInt32LE(crc, 16); directory.writeUInt32LE(data.length, 20); directory.writeUInt32LE(data.length, 24); directory.writeUInt16LE(name.length, 28); directory.writeUInt32LE(offset, 42);
|
||||
central.push(directory, name); offset += header.length + name.length + data.length;
|
||||
}
|
||||
const centralBytes = Buffer.concat(central), localBytes = Buffer.concat(local), end = Buffer.alloc(22);
|
||||
end.writeUInt32LE(0x06054b50, 0); end.writeUInt16LE(files.length, 8); end.writeUInt16LE(files.length, 10); end.writeUInt32LE(centralBytes.length, 12); end.writeUInt32LE(localBytes.length, 16);
|
||||
return Buffer.concat([localBytes, centralBytes, end]);
|
||||
}
|
||||
|
||||
export function registerAdminToolsRoutes(app, { db, adminAuth, media = null, backupDir = null, backfillStep = null }) {
|
||||
export function registerAdminToolsRoutes(app, { db, adminAuth, media = null, backupDir = null, backfillStep = null, pullLyrics = null, queueWhisper = null, autoStartJobs = true }) {
|
||||
const sql = clientOf(db);
|
||||
|
||||
const backupFile = date => join(backupDir || '.', `notes-${date}.json`);
|
||||
@@ -83,8 +102,9 @@ export function registerAdminToolsRoutes(app, { db, adminAuth, media = null, bac
|
||||
await sql.execute({ sql: 'DELETE FROM video_notes WHERE video_id=? AND kind=?', args: [note.videoId, note.kind] });
|
||||
}
|
||||
}
|
||||
await audit(db, 'admin', 'backup.restore', date, { videoId: body.videoId || null, restored: selected.length, safetySnapshot: safety.file });
|
||||
return c.json({ ok: true, restored: selected.length, safetySnapshot: safety.file });
|
||||
const safetySnapshot = basename(safety.file);
|
||||
await audit(db, 'admin', 'backup.restore', date, { videoId: body.videoId || null, restored: selected.length, safetySnapshot });
|
||||
return c.json({ ok: true, restored: selected.length, safetySnapshot });
|
||||
});
|
||||
|
||||
app.get('/api/admin/backups/export/sqlite-dump', adminAuth, async c => {
|
||||
@@ -202,6 +222,148 @@ export function registerAdminToolsRoutes(app, { db, adminAuth, media = null, bac
|
||||
return c.json({ ok: true, indexed });
|
||||
});
|
||||
|
||||
const librarySelect = `SELECT id,title,channel,hay,lyrics,rev,cacheStatus FROM (
|
||||
SELECT m.id,COALESCE(json_extract(m.card,'$.title'),m.id) AS title,COALESCE(json_extract(m.card,'$.channel'),'') AS channel,m.hay,
|
||||
n.data AS lyrics,n.rev AS rev,c.status AS cacheStatus FROM video_meta m
|
||||
LEFT JOIN video_notes n ON n.video_id=m.id AND n.kind='lyrics' LEFT JOIN media_cache c ON c.video_id=m.id
|
||||
UNION ALL
|
||||
SELECT u.id,u.title,COALESCE(u.artist,''),lower(u.title||' '||COALESCE(u.artist,'')),n.data,n.rev,NULL
|
||||
FROM uploads u LEFT JOIN video_notes n ON n.video_id=u.id AND n.kind='lyrics'
|
||||
)`;
|
||||
function lyricsFilter(q) {
|
||||
const words = String(q || '').normalize('NFD').replace(/[\u0300-\u036f]/g, '').toLowerCase().match(/[\p{L}\p{N}_-]+/gu) || [];
|
||||
return { where: words.length ? `WHERE ${words.map(() => 'lower(hay) LIKE ?').join(' AND ')}` : '', args: words.map(word => `%${word}%`) };
|
||||
}
|
||||
app.get('/api/admin/lyrics/bulk/library', adminAuth, async c => {
|
||||
const limit = Math.max(1, Math.min(100, Number(c.req.query('limit')) || 50));
|
||||
const offset = Math.max(0, Math.min(1_000_000, Number(c.req.query('offset')) || 0));
|
||||
const q = String(c.req.query('q') || '').slice(0, 120), filter = lyricsFilter(q);
|
||||
const [count, rows] = await Promise.all([
|
||||
sql.execute({ sql: `SELECT COUNT(*) AS total FROM (${librarySelect}) library ${filter.where}`, args: filter.args }),
|
||||
sql.execute({ sql: `${librarySelect} ${filter.where} ORDER BY title COLLATE NOCASE,id LIMIT ? OFFSET ?`, args: [...filter.args, limit, offset] }),
|
||||
]);
|
||||
return c.json({ ok: true, q, total: number(count.rows[0]?.total), limit, offset, songs: rows.rows.map(row => {
|
||||
const lyrics = parse(row.lyrics), lines = Array.isArray(lyrics.lines) ? lyrics.lines.length : 0;
|
||||
return { id: row.id, title: row.title || row.id, channel: row.channel || '', lines, rev: number(row.rev), cacheStatus: row.cacheStatus || null };
|
||||
}) }, 200, { 'Cache-Control': 'no-store' });
|
||||
});
|
||||
|
||||
app.post('/api/admin/lyrics/jobs', adminAuth, async c => {
|
||||
let body = {}; try { body = await c.req.json(); } catch {}
|
||||
const action = ['lrclib', 'whisper', 'export', 'clear'].includes(body.action) ? body.action : '';
|
||||
if (!action) return c.json({ ok: false, error: 'Choose LRCLIB, Whisper, export, or clear.' }, 400);
|
||||
let ids = Array.isArray(body.ids) ? body.ids : [];
|
||||
if (body.allFiltered === true) {
|
||||
const filter = lyricsFilter(String(body.query || '').slice(0, 120));
|
||||
const rows = await sql.execute({ sql: `${librarySelect} ${filter.where} ORDER BY title COLLATE NOCASE,id LIMIT 500`, args: filter.args });
|
||||
ids = rows.rows.map(row => row.id);
|
||||
}
|
||||
ids = [...new Set(ids.map(String).filter(id => /^[\w-]{11}$/.test(id) || /^upl_[a-f0-9]{12}$/.test(id)))].slice(0, 500);
|
||||
if (!ids.length) return c.json({ ok: false, error: 'Select at least one song.' }, 400);
|
||||
if (action === 'clear' && body.confirmClear !== true) return c.json({ ok: false, error: 'Clearing lyrics requires explicit confirmation.' }, 400);
|
||||
const id = crypto.randomUUID(), now = Math.floor(Date.now() / 1000);
|
||||
const state = { done: 0, total: ids.length, results: [], logs: [], fileReady: false };
|
||||
await sql.execute({ sql: 'INSERT INTO admin_lyrics_jobs(id,action,status,ids,state,created_at,updated_at) VALUES(?,?,\'queued\',?,?,?,?)', args: [id, action, JSON.stringify(ids), JSON.stringify(state), now, now] });
|
||||
await audit(db, 'admin', 'lyrics.bulk.start', id, { action, count: ids.length });
|
||||
return c.json({ ok: true, job: { id, action, status: 'queued', state } }, 202);
|
||||
});
|
||||
|
||||
app.get('/api/admin/lyrics/jobs', adminAuth, async c => {
|
||||
const rows = await sql.execute('SELECT * FROM admin_lyrics_jobs ORDER BY created_at DESC LIMIT 30');
|
||||
return c.json({ ok: true, jobs: rows.rows.map(row => ({ id: row.id, action: row.action, status: row.status, state: parse(row.state), error: row.error, createdAt: number(row.created_at), updatedAt: number(row.updated_at) })) }, 200, { 'Cache-Control': 'no-store' });
|
||||
});
|
||||
app.get('/api/admin/lyrics/jobs/:id', adminAuth, async c => {
|
||||
const row = (await sql.execute({ sql: 'SELECT * FROM admin_lyrics_jobs WHERE id=?', args: [c.req.param('id')] })).rows[0];
|
||||
return row ? c.json({ ok: true, job: { id: row.id, action: row.action, status: row.status, state: parse(row.state), error: row.error, createdAt: number(row.created_at), updatedAt: number(row.updated_at) } }) : c.json({ ok: false, error: 'Job not found.' }, 404);
|
||||
});
|
||||
app.post('/api/admin/lyrics/jobs/:id/:action', adminAuth, async c => {
|
||||
const id = c.req.param('id'), action = c.req.param('action');
|
||||
if (!['pause', 'resume', 'cancel'].includes(action)) return c.json({ ok: false, error: 'Unknown job action.' }, 400);
|
||||
const row = (await sql.execute({ sql: 'SELECT status FROM admin_lyrics_jobs WHERE id=?', args: [id] })).rows[0];
|
||||
if (!row) return c.json({ ok: false, error: 'Job not found.' }, 404);
|
||||
const next = action === 'pause' ? 'paused' : action === 'resume' ? 'queued' : 'cancelled';
|
||||
if (action === 'pause' && !['queued', 'running'].includes(row.status) || action === 'resume' && row.status !== 'paused' || action === 'cancel' && !['queued', 'running', 'paused'].includes(row.status)) return c.json({ ok: false, error: `Cannot ${action} a ${row.status} job.` }, 409);
|
||||
await sql.execute({ sql: 'UPDATE admin_lyrics_jobs SET status=?,updated_at=? WHERE id=?', args: [next, Math.floor(Date.now() / 1000), id] });
|
||||
await audit(db, 'admin', `lyrics.bulk.${action}`, id, { from: row.status, to: next });
|
||||
return c.json({ ok: true, status: next });
|
||||
});
|
||||
app.get('/api/admin/lyrics/jobs/:id/export', adminAuth, async c => {
|
||||
const row = (await sql.execute({ sql: 'SELECT action,status,state FROM admin_lyrics_jobs WHERE id=?', args: [c.req.param('id')] })).rows[0];
|
||||
const state = parse(row?.state);
|
||||
if (!row || row.action !== 'export' || row.status !== 'complete' || !state.fileReady || !backupDir) return c.json({ ok: false, error: 'Export is not ready.' }, 404);
|
||||
try { const data = await readFile(join(backupDir, 'bulk-lyrics', `${c.req.param('id')}.zip`)); return new Response(data, { headers: { 'Content-Type': 'application/zip', 'Content-Disposition': `attachment; filename="lyrics-${c.req.param('id')}.zip"` } }); }
|
||||
catch { return c.json({ ok: false, error: 'Export file is unavailable.' }, 404); }
|
||||
});
|
||||
|
||||
let workerBusy = false;
|
||||
let activeJobId = null;
|
||||
async function runBulkJob() {
|
||||
if (workerBusy) return;
|
||||
workerBusy = true;
|
||||
try {
|
||||
const row = (await sql.execute("SELECT * FROM admin_lyrics_jobs WHERE status IN ('queued','running') ORDER BY created_at,id LIMIT 1")).rows[0];
|
||||
if (!row) return;
|
||||
activeJobId = row.id;
|
||||
await sql.execute({ sql: "UPDATE admin_lyrics_jobs SET status='running',updated_at=? WHERE id=? AND status='queued'", args: [Math.floor(Date.now() / 1000), row.id] });
|
||||
const ids = parse(row.ids), state = parse(row.state); state.total = ids.length; state.results ||= []; state.logs ||= [];
|
||||
for (let index = number(state.done); index < ids.length; index++) {
|
||||
const current = (await sql.execute({ sql: 'SELECT status FROM admin_lyrics_jobs WHERE id=?', args: [row.id] })).rows[0]?.status;
|
||||
if (current !== 'running') return;
|
||||
const videoId = ids[index]; let result;
|
||||
try {
|
||||
if (row.action === 'lrclib') result = pullLyrics ? await pullLyrics(videoId) : { ok: false, error: 'LRCLIB worker is unavailable.' };
|
||||
else if (row.action === 'whisper') {
|
||||
const queued = queueWhisper ? await queueWhisper(videoId) : { status: 503, body: { ok: false, error: 'Whisper worker is unavailable.' } };
|
||||
result = { ...(queued.body || {}), ok: queued.status < 400 && queued.body?.ok, httpStatus: queued.status };
|
||||
} else if (row.action === 'clear') {
|
||||
const saved = await db.saveNote({ videoId, kind: 'lyrics', data: { lines: [], tags: [], offset: 0 }, source: 'admin-bulk-clear', updatedBy: 'admin bulk clear', force: true });
|
||||
const resolved = await db.autoResolveFlags(videoId, new Set());
|
||||
await audit(db, 'admin', 'lyrics.clear', videoId, { rev: saved.rev, resolvedReports: resolved, undoViaRevision: saved.rev - 1 });
|
||||
result = { ok: true, rev: saved.rev, resolvedReports: resolved };
|
||||
} else {
|
||||
const note = (await db.getNotes(videoId)).lyrics;
|
||||
result = note ? { ok: true, lines: note.data.lines?.length || 0 } : { ok: false, error: 'No saved lyrics to export.' };
|
||||
}
|
||||
} catch (error) { result = { ok: false, error: String(error.message || error).slice(0, 300) }; }
|
||||
const item = { id: videoId, ok: !!result?.ok, message: result?.ok ? (result.lines != null ? `${result.lines} lines` : 'Done') : result?.error || 'Action failed.' };
|
||||
state.done = index + 1; state.results.push(item); state.logs.push({ at: Date.now(), ...item }); if (state.logs.length > 80) state.logs.splice(0, state.logs.length - 80);
|
||||
await sql.execute({ sql: 'UPDATE admin_lyrics_jobs SET state=?,updated_at=? WHERE id=?', args: [JSON.stringify(state), Math.floor(Date.now() / 1000), row.id] });
|
||||
if (row.action === 'lrclib') await sleep(1200);
|
||||
}
|
||||
if (row.action === 'export') {
|
||||
const files = [];
|
||||
for (const videoId of ids) {
|
||||
const note = (await db.getNotes(videoId)).lyrics; if (!note?.data?.lines?.length) continue;
|
||||
let title = videoId;
|
||||
if (videoId.startsWith('upl_')) title = (await db.getUpload(videoId))?.title || videoId;
|
||||
else {
|
||||
const card = (await sql.execute({ sql: 'SELECT card FROM video_meta WHERE id=?', args: [videoId] })).rows[0]; title = parse(card?.card).title || videoId;
|
||||
}
|
||||
const safe = String(title).normalize('NFKC').replace(/[\\/:*?"<>|\u0000-\u001f]/g, '_').replace(/\s+/g, ' ').trim().slice(0, 100) || videoId;
|
||||
const lrc = note.data.lines.map(line => line.t == null ? line.text : `[${String(Math.floor(line.t / 60)).padStart(2, '0')}:${(line.t % 60).toFixed(2).padStart(5, '0')}]${line.text}`).join('\n') + '\n';
|
||||
files.push({ name: `${safe}__${videoId}.lrc`, content: lrc });
|
||||
}
|
||||
if (!backupDir) throw Error('Export storage is unavailable.');
|
||||
await mkdir(join(backupDir, 'bulk-lyrics'), { recursive: true });
|
||||
await writeFile(join(backupDir, 'bulk-lyrics', `${row.id}.zip`), zipStore(files));
|
||||
state.fileReady = true; state.exportedFiles = files.length;
|
||||
}
|
||||
await sql.execute({ sql: "UPDATE admin_lyrics_jobs SET status='complete',state=?,updated_at=? WHERE id=? AND status='running'", args: [JSON.stringify(state), Math.floor(Date.now() / 1000), row.id] });
|
||||
} catch (error) {
|
||||
console.error('[admin-tools] lyrics job failed:', error.message);
|
||||
if (activeJobId) {
|
||||
try {
|
||||
await sql.execute({ sql: "UPDATE admin_lyrics_jobs SET status='failed',error=?,updated_at=? WHERE id=? AND status='running'", args: [String(error.message || error).slice(0, 300), Math.floor(Date.now() / 1000), activeJobId] });
|
||||
await audit(db, 'admin', 'lyrics.bulk.failed', activeJobId, { error: String(error.message || error).slice(0, 300) });
|
||||
} catch (persistError) { console.warn('[admin-tools] could not persist lyrics job failure:', persistError.message); }
|
||||
}
|
||||
} finally { activeJobId = null; workerBusy = false; }
|
||||
}
|
||||
if (autoStartJobs) {
|
||||
sql.execute("UPDATE admin_lyrics_jobs SET status='queued',updated_at=unixepoch() WHERE status='running'").then(() => runBulkJob()).catch(error => console.warn('[admin-tools] could not resume lyrics jobs:', error.message));
|
||||
const timer = setInterval(runBulkJob, 1000); timer.unref?.();
|
||||
}
|
||||
|
||||
app.get('/api/admin/cache/health', adminAuth, async c => {
|
||||
const limit = Math.max(1, Math.min(100, Number(c.req.query('limit')) || 50));
|
||||
const offset = Math.max(0, Math.min(1_000_000, Number(c.req.query('offset')) || 0));
|
||||
@@ -297,4 +459,6 @@ export function registerAdminToolsRoutes(app, { db, adminAuth, media = null, bac
|
||||
return c.json({ ok: true, removed, days });
|
||||
});
|
||||
|
||||
return { runBulkJob };
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user