Docs/Skills and code/src/
src/stages/28-discovery-v2-enrich.js
import { config } from '../../config.js';
import { getDb } from '../db.js';
import { getChannel, getChannelVideos } from '../youtube-channel.js';
import { parseCount, uniqueStrings } from '../metadata-utils.js';
import { ensureAnalysisSchema } from '../analysis-schema.js';
import { ensureDiscoveryV2Schema } from '../discovery/v2-schema.js';
import { parseArgs } from '../discovery/v2-utils.js';
const args = parseArgs(process.argv.slice(2));
const db = ensureDiscoveryV2Schema();
ensureAnalysisSchema();
const now = () => new Date().toISOString();
const safeJson = value => { try { return JSON.stringify(value); } catch { return null; } };
const hasTable = name => Boolean(db.prepare("SELECT 1 FROM sqlite_master WHERE type='table' AND name=?").get(name));
const configValue = path => path.split('.').reduce((value, key) => value?.[key], config);
const integerArg = (value, fallback, min, max, label) => {
if (value == null || value === '') return fallback;
const number = Number(value);
if (!Number.isFinite(number) || number < min) throw new Error(`${label} must be between ${min} and ${max}; received ${value}.`);
return Math.min(max, Math.floor(number));
};
const phaseId = String(args.phase || db.prepare(
'SELECT phase_id FROM discovery_v2_phases ORDER BY started_at DESC LIMIT 1'
).get()?.phase_id || '');
if (!phaseId) throw new Error('No discovery phase exists. Run discovery:v2 first.');
const limit = integerArg(args.limit, 200, 1, 500, '--limit');
const offset = integerArg(args.offset, 0, 0, Number.MAX_SAFE_INTEGER, '--offset');
const configuredCatalogueLimit = configValue('discovery.catalogueLimit') ?? configValue('stage1_5.catalogueLimit') ?? 300;
const catalogueLimit = integerArg(args.catalogue, Number(configuredCatalogueLimit), 0, 300, '--catalogue');
const force = Boolean(args.force);
const delayMs = Math.max(0, Number(args.delay ?? 250) || 0);
function tableColumns(table) {
return new Set(db.prepare(`PRAGMA table_info("${table}")`).all().map(column => column.name));
}
const channelColumns = tableColumns('channels');
const metadataColumns = tableColumns('channel_metadata');
const snapshotColumns = hasTable('channel_metric_snapshots') ? tableColumns('channel_metric_snapshots') : new Set();
const videoColumns = tableColumns('videos');
const channelVideoColumns = tableColumns('channel_videos');
const orderBy = channelColumns.has('discovered_at') ? 'c.discovered_at' : 'c.channel_url';
const rows = db.prepare(`
SELECT c.*
FROM channels c
JOIN (
SELECT DISTINCT channel_url FROM discovery_v2_channel_events
WHERE phase_id=? AND was_new_channel=1
) e ON e.channel_url=c.channel_url
ORDER BY ${orderBy}, c.channel_url LIMIT ? OFFSET ?
`).all(phaseId, limit, offset);
// Build statements from the live SQLite schema. Ecosystem DBs may be at different
// migration levels, so unsupported optional columns are omitted instead of crashing.
function buildUpsert(table, columns, candidateFields, keyField) {
const fields = candidateFields.filter(field => columns.has(field));
if (!fields.includes(keyField)) throw new Error(`${table} is missing required key column ${keyField}`);
const names = fields.map(field => `"${field}"`).join(',');
const values = fields.map(field => `@${field}`).join(',');
const updates = fields.filter(field => field !== keyField).map(field => {
if (field === 'keywords_json') return `"${field}"=CASE WHEN excluded."${field}" IS NOT NULL AND excluded."${field}"<>'[]' THEN excluded."${field}" ELSE "${table}"."${field}" END`;
if (field === 'fetch_error') return `"${field}"=excluded."${field}"`;
if (field === 'fetched_at') return `"${field}"=excluded."${field}"`;
return `"${field}"=COALESCE(excluded."${field}","${table}"."${field}")`;
});
const conflict = updates.length ? `DO UPDATE SET ${updates.join(',')}` : 'DO NOTHING';
return db.prepare(`INSERT INTO "${table}" (${names}) VALUES (${values}) ON CONFLICT("${keyField}") ${conflict}`);
}
const metadataFields = [
'channel_url','fetched_at','channel_id','name','description',
'subscriber_count','subscriber_count_text','subscriber_count_precision',
'video_count','video_count_text','video_count_precision',
'total_view_count','total_view_count_text','total_view_count_precision',
'joined_date','joined_date_text','country','canonical_url','handle',
'keywords_json','thumbnail_url','banner_url','custom_links_json',
'is_verified','is_artist','metadata_source','metadata_raw_json','fetch_error'
];
const saveMetadata = buildUpsert('channel_metadata', metadataColumns, metadataFields, 'channel_url');
const channelUpdateFields = [
'channel_id','name','description','subscriber_count','subscriber_count_text','subscriber_count_precision',
'video_count','video_count_text','video_count_precision','total_view_count','total_view_count_text',
'total_view_count_precision','joined_date','joined_date_text','country','canonical_url','handle',
'keywords_json','thumbnail_url','banner_url','custom_links_json','is_verified','is_artist',
'metadata_source','metadata_raw_json','enriched_at','enrichment_error'
];
const channelSetFields = channelUpdateFields.filter(field => channelColumns.has(field));
const channelSetSql = channelSetFields.map(field => {
if (field === 'enriched_at') return '"enriched_at"=@fetched_at';
if (field === 'enrichment_error') return '"enrichment_error"=NULL';
if (field === 'keywords_json') return '"keywords_json"=CASE WHEN @keywords_json IS NOT NULL AND @keywords_json<>\'[]\' THEN @keywords_json ELSE "keywords_json" END';
return `"${field}"=COALESCE(@${field},"${field}")`;
}).join(',');
const updateChannel = db.prepare(`UPDATE channels SET ${channelSetSql} WHERE channel_url=@channel_url`);
const errorFields = ['channel_url','fetched_at','name','fetch_error'];
const saveError = buildUpsert('channel_metadata', metadataColumns, errorFields, 'channel_url');
let saveSnapshot = null;
if (snapshotColumns.size) {
const snapshotFields = [
'channel_url','observed_at','source','subscriber_count','subscriber_count_text','subscriber_count_precision',
'total_view_count','total_view_count_text','total_view_count_precision','video_count','video_count_text',
'video_count_precision','joined_date','joined_date_text','country','raw_metadata_json'
];
const fields = snapshotFields.filter(field => snapshotColumns.has(field));
if (fields.includes('channel_url') && fields.includes('observed_at') && fields.includes('source')) {
saveSnapshot = db.prepare(`INSERT INTO channel_metric_snapshots (${fields.map(f => `"${f}"`).join(',')}) VALUES (${fields.map(f => `@${f}`).join(',')})`);
}
}
const saveVideo = db.prepare(`
INSERT INTO videos(video_id,title,url,channel_url,channel_name,thumbnail_url,duration,view_count,published_at,discovered_at)
VALUES(@video_id,@title,@url,@channel_url,@channel_name,@thumbnail_url,@duration,@view_count,@published_at,@discovered_at)
ON CONFLICT(video_id) DO UPDATE SET
title=CASE WHEN videos.title IS NULL OR videos.title='' THEN excluded.title ELSE videos.title END,
url=COALESCE(videos.url,excluded.url),channel_url=COALESCE(videos.channel_url,excluded.channel_url),
channel_name=COALESCE(videos.channel_name,excluded.channel_name),thumbnail_url=COALESCE(videos.thumbnail_url,excluded.thumbnail_url),
duration=COALESCE(videos.duration,excluded.duration),view_count=COALESCE(videos.view_count,excluded.view_count),
published_at=COALESCE(videos.published_at,excluded.published_at)
`);
const saveChannelVideo = db.prepare(`
INSERT OR IGNORE INTO channel_videos(channel_url,video_id,source,position,discovered_at) VALUES(?,?,?,?,?)
`);
function deriveHandle(canonicalUrl, inputUrl, existingHandle) {
if (existingHandle) return existingHandle;
const url = String(canonicalUrl || inputUrl || '');
const match = url.match(/youtube\.com\/@([^/?#]+)/i);
return match?.[1] ? `@${match[1]}` : null;
}
function makeRecord(channel, row, fetchedAt) {
const canonicalUrl = channel.canonical_url ?? row.canonical_url ?? row.channel_url;
const keywords = uniqueStrings(channel.keywords ?? []);
const metadataRaw = channel.metadata_raw_json ?? safeJson({
channelMetadata: channel.channel?.metadata ?? null,
headerMetadataRows: channel.channel?.header?.content?.metadata?.metadata_rows ?? null,
aboutFieldsAvailable: Boolean(channel.joined_date_text || channel.country || channel.total_view_count_text)
});
return {
channel_url: row.channel_url,
fetched_at: fetchedAt,
channel_id: channel.channel_id ?? row.channel_id ?? null,
name: channel.name ?? row.name ?? null,
description: channel.description ?? row.description ?? null,
subscriber_count: parseCount(channel.subscriber_count) ?? parseCount(channel.subscriber_count_text) ?? row.subscriber_count ?? null,
subscriber_count_text: channel.subscriber_count_text ?? row.subscriber_count_text ?? null,
subscriber_count_precision: channel.subscriber_count_precision ?? row.subscriber_count_precision ?? null,
video_count: parseCount(channel.video_count) ?? parseCount(channel.video_count_text) ?? row.video_count ?? null,
video_count_text: channel.video_count_text ?? row.video_count_text ?? null,
video_count_precision: channel.video_count_precision ?? row.video_count_precision ?? null,
total_view_count: parseCount(channel.total_view_count) ?? parseCount(channel.total_view_count_text) ?? row.total_view_count ?? null,
total_view_count_text: channel.total_view_count_text ?? row.total_view_count_text ?? null,
total_view_count_precision: channel.total_view_count_precision ?? row.total_view_count_precision ?? null,
joined_date: channel.joined_date ?? row.joined_date ?? null,
joined_date_text: channel.joined_date_text ?? row.joined_date_text ?? null,
joined_date_precision: channel.joined_date_precision ?? row.joined_date_precision ?? null,
country: channel.country ?? row.country ?? null,
canonical_url: canonicalUrl,
handle: deriveHandle(canonicalUrl, row.channel_url, channel.handle ?? row.handle),
keywords_json: keywords.length ? JSON.stringify(keywords) : (row.keywords_json ?? '[]'),
thumbnail_url: channel.thumbnail_url ?? row.thumbnail_url ?? null,
banner_url: channel.banner_url ?? row.banner_url ?? null,
custom_links_json: channel.custom_links_json ?? (Array.isArray(channel.custom_links) ? JSON.stringify(channel.custom_links) : row.custom_links_json ?? null),
is_verified: channel.is_verified == null ? (row.is_verified ?? null) : Number(channel.is_verified),
is_artist: channel.is_artist == null ? (row.is_artist ?? null) : Number(channel.is_artist),
metadata_source: channel.metadata_source ?? row.metadata_source ?? 'youtubei_channel_header_and_about',
metadata_observed_at: channel.metadata_observed_at ?? fetchedAt,
metadata_raw_json: metadataRaw,
fetch_error: null
};
}
let fetched = 0, skipped = 0, failed = 0, catalogueVideos = 0, catalogueFailures = 0;
let metadataMissingAfterFetch = 0;
for (let i = 0; i < rows.length; i++) {
const row = rows[i];
const missing = !row.channel_id || !row.description || row.subscriber_count == null || row.video_count == null ||
row.total_view_count == null || !row.joined_date || !row.canonical_url || !row.handle ||
!row.metadata_source || !row.keywords_json || !row.thumbnail_url || !row.banner_url;
if (!force && !missing) { skipped++; continue; }
console.log(`[discovery-v2-enrich] ${i + 1}/${rows.length} channel=${row.name} url=${row.channel_url}`);
try {
const channel = await getChannel(row.channel_url);
const fetchedAt = now();
const record = makeRecord(channel, row, fetchedAt);
db.transaction(() => {
saveMetadata.run(record);
updateChannel.run(record);
if (saveSnapshot) saveSnapshot.run({
...record,
observed_at: record.metadata_observed_at ?? fetchedAt,
source: record.metadata_source ?? 'youtubei_channel_header_and_about',
raw_metadata_json: record.metadata_raw_json
});
})();
fetched++;
const missingFields = ['total_view_count','joined_date','canonical_url','handle','metadata_source'].filter(field => record[field] == null || record[field] === '');
if (missingFields.length) metadataMissingAfterFetch++;
console.log('[discovery-v2-enrich] channel metadata saved', {
channel: record.name,
subscriberCount: record.subscriber_count,
videoCount: record.video_count,
totalViewCount: record.total_view_count,
joinedDate: record.joined_date,
canonicalUrl: record.canonical_url,
handle: record.handle,
metadataSource: record.metadata_source,
stillMissing: missingFields
});
if (catalogueLimit > 0) {
for (const mode of ['Popular', 'Latest']) {
try {
const videos = await getChannelVideos(channel, mode, catalogueLimit);
const discoveredAt = now();
let savedThisFeed = 0;
db.transaction(() => {
videos.forEach((video, index) => {
if (!video.video_id) return;
const item = {
video_id: video.video_id,
title: video.title || '(untitled)',
url: video.url || `https://www.youtube.com/watch?v=${video.video_id}`,
channel_url: row.channel_url,
channel_name: record.name,
thumbnail_url: video.thumbnail_url || null,
duration: video.duration || null,
view_count: parseCount(video.view_count) ?? (typeof video.view_count === 'number' ? video.view_count : null),
published_at: video.published_at || null,
discovered_at: discoveredAt
};
saveVideo.run(item);
saveChannelVideo.run(row.channel_url, video.video_id, mode.toLowerCase(), index + 1, discoveredAt);
savedThisFeed++;
});
})();
catalogueVideos += savedThisFeed;
console.log('[discovery-v2-enrich] catalogue feed saved', {
channel: record.name, mode, returned: videos.length, saved: savedThisFeed, limit: catalogueLimit
});
} catch (error) {
catalogueFailures++;
console.error('[discovery-v2-enrich] catalogue feed failed', {
channel: record.name, mode, error: error instanceof Error ? error.message : String(error)
});
}
}
}
} catch (error) {
failed++;
const message = String(error?.stack || error).slice(0, 1800);
try { saveError.run({ channel_url: row.channel_url, fetched_at: now(), name: row.name, fetch_error: message }); } catch {}
console.error('[discovery-v2-enrich] channel failed', {
channel: row.name, url: row.channel_url, error: error instanceof Error ? error.message : String(error)
});
}
if (delayMs) await new Promise(resolve => setTimeout(resolve, delayMs));
}
console.log('[discovery-v2-enrich] complete', {
phaseId, candidates: rows.length, fetched, skipped, failed,
catalogueVideos, catalogueFailures, catalogueLimit, metadataMissingAfterFetch
});
console.log(`Next: npm run qualification:llm -- --limit=${Math.min(limit, rows.length)} ; then npm run queues:rate ; then npm run discovery:v2:score -- --phase=${phaseId}`);
Shown as it is in the repository at build time (2026-11-03); the home directory appears as <home>. If this page and the file differ, the file is right. Back to the catalogue.