218 lines
16 KiB
JavaScript
218 lines
16 KiB
JavaScript
import { randomUUID } from 'node:crypto';
|
|
import { db as defaultDb } from './db.js';
|
|
import * as catalog from './video-catalog.js';
|
|
import * as innertube from './innertube.js';
|
|
import * as searchCache from './search-cache.js';
|
|
import { audit } from './audit-log.js';
|
|
|
|
const MAX_SEEDS = 200, MAX_PLAN = 500, MAX_ERRORS = 50;
|
|
const clean = value => String(value || '').replace(/[\u0000-\u001f\u007f]/g, ' ').trim().slice(0, 200);
|
|
const fold = value => String(value || '').normalize('NFD').replace(/[\u0300-\u036f]/g, '').toLowerCase().replace(/\s+/g, ' ').trim();
|
|
const parse = value => { try { return JSON.parse(value || '{}'); } catch { return {}; } };
|
|
const sleepDefault = ms => new Promise(resolve => setTimeout(resolve, ms));
|
|
|
|
export const DEFAULT_PRESETS = [
|
|
'worship', 'praise', 'hymns', 'gospel', 'pagsamba', 'papuri', 'mga awit ng papuri', 'ebanghelyo',
|
|
'dayeg', 'pagdayeg', 'mga alawiton', 'maayong balita', 'adoración', 'alabanza', 'himnos', 'evangelio',
|
|
'Hillsong Worship', 'Elevation Worship', 'Bethel Music', 'Planetshakers', 'Don Moen', 'Sandi Patty',
|
|
];
|
|
export const DEFAULT_MODIFIERS = ['lyrics', 'live', 'acoustic', 'instrumental', 'karaoke'];
|
|
|
|
export function expandQueryPlan({ seeds = [], modifiers = [], maxQueries = 80 } = {}) {
|
|
const seedList = (Array.isArray(seeds) ? seeds : String(seeds).split(/\r?\n/)).map(clean).filter(Boolean).slice(0, MAX_SEEDS);
|
|
const mods = (Array.isArray(modifiers) ? modifiers : String(modifiers).split(/[,\r\n]+/)).map(clean).filter(Boolean).slice(0, 30);
|
|
const cap = Math.max(1, Math.min(MAX_PLAN, Number(maxQueries) || 1));
|
|
const out = [], seen = new Set();
|
|
for (const seed of seedList) for (const modifier of ['', ...mods]) {
|
|
const query = clean(modifier ? `${seed} ${modifier}` : seed), key = fold(query);
|
|
if (key && !seen.has(key)) { seen.add(key); out.push(query); if (out.length >= cap) return out; }
|
|
}
|
|
return out;
|
|
}
|
|
|
|
function publicJob(row, indexSize = 0) {
|
|
const state = parse(row.state);
|
|
const done = Number(state.queriesDone) || 0, total = JSON.parse(row.plan || '[]').length;
|
|
const elapsed = Math.max(0, Date.now() - Number(row.created_at));
|
|
const etaMs = done ? Math.max(0, Math.round(elapsed / done * (total - done))) : null;
|
|
return { id: row.id, status: row.status, plan: JSON.parse(row.plan || '[]'), options: parse(row.options),
|
|
queriesDone: done, queriesTotal: total, current: state.current || '', newVideos: Number(state.newVideos) || 0,
|
|
totalIndexSize: indexSize, errors: state.errors || [], log: state.log || [], etaMs,
|
|
createdAt: Number(row.created_at), updatedAt: Number(row.updated_at), error: row.error || null };
|
|
}
|
|
|
|
export function createDeepSearchService({ db = defaultDb, searchDeep = innertube.searchDeep, related = innertube.related,
|
|
browseUploads = innertube.browseUploads, ingest = catalog.ingest, putSearch = searchCache.put,
|
|
sleep = sleepDefault, random = Math.random, now = Date.now, autoStart = true } = {}) {
|
|
let worker = null;
|
|
let timer = null;
|
|
const log = (state, message) => { state.log ||= []; state.log.push({ at: now(), message: String(message).slice(0, 240) }); if (state.log.length > 100) state.log.splice(0, state.log.length - 100); };
|
|
async function indexSize() { return Number((await db.execute('SELECT COUNT(*) AS n FROM video_meta')).rows[0]?.n || 0); }
|
|
async function getJob(id) { return (await db.execute({ sql: 'SELECT * FROM deep_search_jobs WHERE id=?', args: [id] })).rows[0] || null; }
|
|
async function checkpoint(job, state, status = 'running', error = null) {
|
|
const r = await db.execute({ sql: "UPDATE deep_search_jobs SET state=?,status=?,error=?,updated_at=? WHERE id=? AND status='running'", args: [JSON.stringify(state), status, error, now(), job.id] });
|
|
if (!r.rowsAffected) { const stop = new Error('Deep search paused or cancelled.'); stop.name = 'DeepSearchStopped'; throw stop; }
|
|
}
|
|
async function reserveDaily(cap) {
|
|
const day = new Date(now()).toISOString().slice(0, 10);
|
|
const r = await db.execute({ sql: `INSERT INTO deep_search_daily(day,queries) VALUES(?,1)
|
|
ON CONFLICT(day) DO UPDATE SET queries=queries+1 WHERE queries<? RETURNING queries`, args: [day, cap] });
|
|
return r.rows.length ? Number(r.rows[0].queries) : null;
|
|
}
|
|
async function run(job) {
|
|
const options = parse(job.options), plan = JSON.parse(job.plan || '[]');
|
|
const state = { queriesDone: 0, newVideos: 0, errors: [], log: [], current: '', relatedSeen: [], ...parse(job.state) };
|
|
async function saveCards(query, cards, cacheCards = cards) {
|
|
const unique = [...new Map((cards || []).filter(c => c?.id).map(c => [c.id, c])).values()];
|
|
if (unique.length) {
|
|
const validIds = unique.filter(c => /^[\w-]{11}$/.test(c.id)).map(c => c.id);
|
|
const existing = validIds.length ? (await db.execute({ sql: `SELECT id FROM video_meta WHERE id IN (${validIds.map(() => '?').join(',')})`, args: validIds })).rows : [];
|
|
const present = new Set(existing.map(row => row.id));
|
|
await ingest(unique, 'deep');
|
|
state.newVideos += validIds.filter(id => !present.has(id)).length;
|
|
}
|
|
await putSearch(query, (cacheCards || unique).slice(0, 200));
|
|
}
|
|
async function requireDailyBudget() {
|
|
const used = await reserveDaily(Math.max(1, Number(options.dailyCap) || 1000));
|
|
if (used !== null) return;
|
|
state.dailyPausedDay = new Date(now()).toISOString().slice(0, 10);
|
|
log(state, 'Daily query cap reached; job will resume tomorrow.');
|
|
await checkpoint(job, state, 'paused');
|
|
const stop = new Error('Daily cap reached.'); stop.name = 'DeepSearchStopped'; throw stop;
|
|
}
|
|
try {
|
|
for (let qi = state.queriesDone; qi < plan.length; qi++) {
|
|
const query = plan[qi];
|
|
if (state.current !== query) state.currentIds = [];
|
|
state.current = query; await checkpoint(job, state);
|
|
await requireDailyBudget();
|
|
await checkpoint(job, state);
|
|
let cards, lastError;
|
|
for (let attempt = 0; attempt < 3; attempt++) {
|
|
try {
|
|
cards = await searchDeep(query, { limit: Math.max(1, Math.min(200, Number(options.resultsPerQuery) || 50)), maxPages: 14, timeoutMs: 30000,
|
|
pageDelayMs: Math.max(0, Number(options.delayMs) || 0), jitterMs: Math.max(0, Number(options.jitterMs) || 0),
|
|
onPage: async pageCards => {
|
|
await checkpoint(job, state);
|
|
const saved = new Set(state.currentIds || []);
|
|
const fresh = pageCards.filter(card => !saved.has(card.id));
|
|
await saveCards(query, fresh, pageCards);
|
|
state.currentIds = pageCards.map(card => card.id);
|
|
await checkpoint(job, state);
|
|
} });
|
|
await checkpoint(job, state);
|
|
lastError = null; break;
|
|
} catch (error) { if (error.name === 'DeepSearchStopped') throw error; lastError = error; if (attempt < 2) await sleep(Math.min(60000, (Number(options.delayMs) || 10000) * 2 ** attempt + Math.floor(random() * 3000))); }
|
|
}
|
|
if (lastError) { state.errors.push({ query, error: String(lastError.message).slice(0, 200) }); if (state.errors.length > MAX_ERRORS) state.errors.shift(); log(state, `Search failed: ${query}`); }
|
|
else {
|
|
const saved = new Set(state.currentIds || []);
|
|
const remaining = (cards || []).filter(card => !saved.has(card.id));
|
|
await saveCards(query, remaining, cards || []);
|
|
state.currentIds = (cards || []).map(card => card.id);
|
|
const depth = Math.max(0, Math.min(2, Number(options.relatedDepth) || 0));
|
|
if (depth) {
|
|
let frontier = (cards || []).slice(0, Math.max(1, Math.min(10, Number(options.relatedTop) || 3))).map(c => ({ id: c.id, depth: 1 }));
|
|
for (let level = 1; level <= depth && frontier.length; level++) {
|
|
const next = [];
|
|
for (const item of frontier) {
|
|
if (state.relatedSeen.includes(item.id)) continue;
|
|
state.relatedSeen.push(item.id);
|
|
try {
|
|
await sleep(Math.max(0, Number(options.delayMs) || 10000) + Math.floor(random() * (Number(options.jitterMs) || 3000)));
|
|
await checkpoint(job, state);
|
|
await requireDailyBudget();
|
|
const recs = await related(item.id); await checkpoint(job, state);
|
|
await saveCards(`${query} related ${item.id}`, recs);
|
|
if (level < depth) next.push(...recs.slice(0, 2).map(c => ({ id: c.id, depth: level + 1 })));
|
|
} catch (error) { if (error.name === 'DeepSearchStopped') throw error; state.errors.push({ query: `related:${item.id}`, error: String(error.message).slice(0, 200) }); }
|
|
}
|
|
frontier = next;
|
|
}
|
|
}
|
|
log(state, `Saved ${cards?.length || 0} results for “${query}”.`);
|
|
}
|
|
state.queriesDone = qi + 1; state.current = ''; state.currentIds = []; await checkpoint(job, state);
|
|
if (qi + 1 < plan.length) await sleep(Math.max(0, Number(options.delayMs) || 10000) + Math.floor(random() * (Number(options.jitterMs) || 3000)));
|
|
}
|
|
for (const source of [...(options.channels || []), ...(options.playlists || [])].slice(0, 50)) {
|
|
state.current = `browse ${source}`; await checkpoint(job, state);
|
|
try { await requireDailyBudget(); await checkpoint(job, state); const cards = await browseUploads(source, { limit: Math.max(1, Math.min(200, Number(options.resultsPerQuery) || 50)), pageDelayMs: options.delayMs, jitterMs: options.jitterMs }); await checkpoint(job, state); await saveCards(state.current, cards); log(state, `Browsed ${cards.length} videos from ${source}.`); }
|
|
catch (error) { if (error.name === 'DeepSearchStopped') throw error; state.errors.push({ query: state.current, error: String(error.message).slice(0, 200) }); }
|
|
await checkpoint(job, state);
|
|
}
|
|
state.current = ''; log(state, 'Deep search complete.'); await checkpoint(job, state, 'complete');
|
|
} catch (error) {
|
|
if (error.name !== 'DeepSearchStopped') {
|
|
state.errors.push({ query: state.current, error: String(error.message).slice(0, 200) });
|
|
await db.execute({ sql: "UPDATE deep_search_jobs SET state=?,status='failed',error=?,updated_at=? WHERE id=? AND status='running'", args: [JSON.stringify(state), String(error.message).slice(0, 300), now(), job.id] });
|
|
}
|
|
}
|
|
}
|
|
async function runNext() {
|
|
if (worker) return worker;
|
|
worker = (async () => {
|
|
// Recover jobs interrupted by process restart; the current query is safe to repeat.
|
|
await db.execute({ sql: "UPDATE deep_search_jobs SET status='queued',updated_at=? WHERE status='running'", args: [now()] });
|
|
const today = new Date(now()).toISOString().slice(0, 10);
|
|
const paused = (await db.execute("SELECT id,state FROM deep_search_jobs WHERE status='paused'")).rows;
|
|
for (const row of paused) if (parse(row.state).dailyPausedDay && parse(row.state).dailyPausedDay < today)
|
|
await db.execute({ sql: "UPDATE deep_search_jobs SET status='queued',updated_at=? WHERE id=? AND status='paused'", args: [now(), row.id] });
|
|
const job = (await db.execute({ sql: "SELECT * FROM deep_search_jobs WHERE status='queued' ORDER BY created_at LIMIT 1" })).rows[0];
|
|
if (job) {
|
|
await db.execute({ sql: "UPDATE deep_search_jobs SET status='running',updated_at=? WHERE id=? AND status='queued'", args: [now(), job.id] });
|
|
const current = await getJob(job.id); if (current?.status === 'running') await run(current);
|
|
}
|
|
return !!job;
|
|
})().finally(() => { worker = null; });
|
|
return worker;
|
|
}
|
|
const kick = () => runNext().catch(error => console.error('[deep search]', error.message));
|
|
function start() { if (!timer) { timer = setInterval(kick, 15000); timer.unref?.(); } kick(); }
|
|
if (autoStart) start();
|
|
async function estimate(body) {
|
|
const plan = expandQueryPlan(body);
|
|
const browseSources = [...(Array.isArray(body?.channels) ? body.channels : []), ...(Array.isArray(body?.playlists) ? body.playlists : [])].map(clean).filter(Boolean).slice(0, 50).length;
|
|
const operations = plan.length + browseSources + (Number(body?.relatedDepth) > 0 ? Math.min(plan.length * (Number(body?.relatedTop) || 3) * Math.min(2, Number(body.relatedDepth)), 100) : 0);
|
|
const today = new Date(now()).toISOString().slice(0, 10);
|
|
const used = Number((await db.execute({ sql: 'SELECT queries FROM deep_search_daily WHERE day=?', args: [today] })).rows[0]?.queries || 0);
|
|
return { plan, queries: plan.length, browseSources, estimatedOperations: operations,
|
|
estimatedMinutes: Math.ceil(operations * Math.max(1, Number(body?.delaySeconds) || 10) / 60),
|
|
indexSize: await indexSize(), dailyUsed: used, dailyRemaining: Math.max(0, (Number(body?.dailyCap) || 1000) - used) };
|
|
}
|
|
function register(app, adminAuth) {
|
|
app.post('/api/admin/deep-search/estimate', adminAuth, async c => { let body = {}; try { body = await c.req.json(); } catch {} return c.json({ ok: true, ...(await estimate(body)) }); });
|
|
app.get('/api/admin/deep-search/jobs', adminAuth, async c => c.json({ ok: true, indexSize: await indexSize(), jobs: (await db.execute('SELECT * FROM deep_search_jobs ORDER BY created_at DESC LIMIT 30')).rows.map(row => publicJob(row)) }));
|
|
app.post('/api/admin/deep-search/start', adminAuth, async c => {
|
|
let body; try { body = await c.req.json(); } catch { return c.json({ ok: false, error: 'Invalid JSON.' }, 400); }
|
|
const plan = expandQueryPlan(body); if (!plan.length) return c.json({ ok: false, error: 'Add at least one seed term.' }, 400);
|
|
const active = (await db.execute("SELECT id FROM deep_search_jobs WHERE status IN ('queued','running','paused') LIMIT 1")).rows[0];
|
|
if (active) return c.json({ ok: false, error: 'A deep-search job is already active.' }, 409);
|
|
const options = { resultsPerQuery: Math.max(1, Math.min(200, Number(body.resultsPerQuery) || 50)), relatedDepth: Math.max(0, Math.min(2, Number(body.relatedDepth) || 0)), relatedTop: Math.max(1, Math.min(10, Number(body.relatedTop) || 3)), delayMs: Math.max(1000, Math.min(300000, (Number(body.delaySeconds) || 10) * 1000)), jitterMs: Math.max(0, Math.min(300000, (Number(body.jitterSeconds) || 3) * 1000)), dailyCap: Math.max(1, Math.min(10000, Number(body.dailyCap) || 1000)), channels: (Array.isArray(body.channels) ? body.channels : []).map(clean).filter(Boolean).slice(0, 25), playlists: (Array.isArray(body.playlists) ? body.playlists : []).map(clean).filter(Boolean).slice(0, 25) };
|
|
const id = randomUUID(), time = now();
|
|
await db.execute({ sql: "INSERT INTO deep_search_jobs(id,status,plan,options,state,created_at,updated_at) VALUES(?,'queued',?,?,?,?,?)", args: [id, JSON.stringify(plan), JSON.stringify(options), JSON.stringify({ queriesDone: 0, newVideos: 0, errors: [], log: [] }), time, time] });
|
|
await audit(db, 'admin', 'deep-search.start', id, { queries: plan.length });
|
|
if (autoStart) kick(); return c.json({ ok: true, id }, 202);
|
|
});
|
|
for (const action of ['pause', 'resume', 'cancel']) app.post(`/api/admin/deep-search/:id/${action}`, adminAuth, async c => {
|
|
const id = c.req.param('id');
|
|
const allowed = action === 'pause' ? ['queued', 'running'] : action === 'resume' ? ['paused'] : ['queued', 'running', 'paused'];
|
|
const target = action === 'pause' ? 'paused' : action === 'resume' ? 'queued' : 'cancelled';
|
|
const marks = allowed.map(() => '?').join(',');
|
|
const r = await db.execute({ sql: `UPDATE deep_search_jobs SET status=?,updated_at=? WHERE id=? AND status IN (${marks})`, args: [target, now(), id, ...allowed] });
|
|
if (r.rowsAffected) await audit(db, 'admin', `deep-search.${action}`, id);
|
|
if (action === 'resume' && r.rowsAffected && autoStart) kick();
|
|
return c.json({ ok: !!r.rowsAffected, status: r.rowsAffected ? target : undefined }, r.rowsAffected ? 200 : 409);
|
|
});
|
|
return { estimate, runNext, kick, start };
|
|
}
|
|
return { estimate, runNext, kick, start, register };
|
|
}
|
|
|
|
export function registerDeepSearchRoutes(app, options = {}) {
|
|
const service = createDeepSearchService(options);
|
|
service.register(app, options.adminAuth);
|
|
return service;
|
|
}
|