Docs/Skills and code/src/
src/stages/25-qualify-channels-with-llm.js
import crypto from "node:crypto";
import fs from "node:fs";
import path from "node:path";
import { getDb } from "../db.js";
import { config } from "../../config.js";
import { callLLM, getLLMConfig, parseJsonObject } from "../llm.js";
import { buildQualificationInput } from "../analysis/channel-evidence.js";
import { shouldRetryIndividually } from "../analysis/qualification-retry-policy.js";
import { ensureChannelQualificationLlmSchema, normalizeLlmResult, qualificationPrompt } from "../analysis/channel-qualification-store.js";
const db = getDb();
ensureChannelQualificationLlmSchema(db);
const args = process.argv.slice(2);
const argValue = name => args.find(value => value.startsWith(`${name}=`))?.slice(name.length + 1);
const has = flag => args.includes(flag);
const force = has("--force");
const limitArg = argValue("--limit");
const batchArg = argValue("--batch-size");
const limit = limitArg == null ? Infinity : Number(limitArg);
const batchSize = Math.max(1, Math.min(20, Math.floor(Number(batchArg ?? config.qualification.llm.batchSize))));
if (!(limit > 0)) throw new Error("--limit must be a positive integer.");
const root = process.cwd();
const now = new Date();
const startedAt = now.toISOString();
const runId = `${startedAt.replaceAll(":", "-").replaceAll(".", "-")}-${crypto.randomBytes(3).toString("hex")}`;
const llm = getLLMConfig();
const promptVersion = config.qualification.promptVersion;
const channelsRows = db.prepare(`
SELECT c.*,
COALESCE(m.name,c.name) AS resolved_name,
COALESCE(m.description,c.description) AS resolved_description,
COALESCE(m.keywords_json,c.keywords_json) AS resolved_keywords_json,
COALESCE(m.subscriber_count,c.subscriber_count) AS resolved_subscriber_count,
COALESCE(m.subscriber_count_text,c.subscriber_count_text) AS resolved_subscriber_count_text,
COALESCE(m.video_count,c.video_count) AS resolved_video_count,
COALESCE(m.joined_date,c.joined_date) AS resolved_joined_date
FROM channels c LEFT JOIN channel_metadata m ON m.channel_url=c.channel_url
ORDER BY COALESCE(m.name,c.name) COLLATE NOCASE, c.channel_url
`).all();
const videosByChannel = new Map();
const videoRows = db.prepare(`
SELECT cv.channel_url, v.video_id, v.title, v.view_count, v.published_at, v.published_at_normalized, v.published_at_precision
FROM channel_videos cv JOIN videos v ON v.video_id=cv.video_id
UNION
SELECT v.channel_url, v.video_id, v.title, v.view_count, v.published_at, v.published_at_normalized, v.published_at_precision
FROM videos v WHERE v.channel_url IS NOT NULL
`).all();
for (const video of videoRows) {
if (!video.channel_url) continue;
if (!videosByChannel.has(video.channel_url)) videosByChannel.set(video.channel_url, new Map());
const map = videosByChannel.get(video.channel_url);
const prior = map.get(video.video_id);
if (!prior || (Number(video.view_count) || 0) > (Number(prior.view_count) || 0)) map.set(video.video_id, video);
}
const sourceChannels = channelsRows.map(row => {
let keywords = [];
try { const parsed = JSON.parse(row.resolved_keywords_json ?? "[]"); keywords = Array.isArray(parsed) ? parsed : []; } catch {}
return {
channel_url: row.channel_url, name: row.resolved_name, description: row.resolved_description,
keywords, keywords_json: row.resolved_keywords_json,
subscriber_count: row.resolved_subscriber_count, subscriber_count_text: row.resolved_subscriber_count_text,
video_count: row.resolved_video_count, joined_date: row.resolved_joined_date,
is_verified: row.is_verified, is_artist: row.is_artist,
videos: [...(videosByChannel.get(row.channel_url)?.values() ?? [])]
};
});
const existing = db.prepare(`
SELECT channel_url,input_hash,prompt_version,processing_status,created_at
FROM channel_qualification_llm_results
WHERE processing_status='succeeded' AND prompt_version=?
ORDER BY created_at DESC
`).all(promptVersion);
const latestSuccessful = new Map();
for (const row of existing) if (!latestSuccessful.has(row.channel_url)) latestSuccessful.set(row.channel_url, row);
const pendingAll = sourceChannels.map(channel => ({ channel, ...buildQualificationInput(channel) }))
.filter(item => force || latestSuccessful.get(item.channel.channel_url)?.input_hash !== item.inputHash);
const pending = Number.isFinite(limit) ? pendingAll.slice(0, Math.floor(limit)) : pendingAll;
const skippedUnchanged = sourceChannels.length - pendingAll.length;
const deferredByLimit = pendingAll.length - pending.length;
const reportDir = path.join(root, "reports");
fs.mkdirSync(reportDir, { recursive: true });
const reportPath = path.join(reportDir, `channel-qualification-llm-${startedAt.replaceAll(":", "-").replaceAll(".", "-")}.json`);
const configRecord = {
ecosystem_name: config.ecosystem.name,
prompt_version: promptVersion,
model: llm.model,
batch_size: batchSize,
max_video_titles: config.qualification.llm.maxVideoTitles,
skipped_unchanged: skippedUnchanged,
deferred_by_limit: deferredByLimit,
force
};
const insertRun = db.prepare(`INSERT INTO channel_qualification_llm_runs
(run_id,ecosystem_name,prompt_version,model,started_at,input_channel_count,requested_channel_count,success_count,failure_count,batch_size,config_json,status)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)`);
insertRun.run(runId, config.ecosystem.name, promptVersion, llm.model, startedAt, sourceChannels.length, pending.length, 0, 0, batchSize, JSON.stringify(configRecord), "running");
const insertResult = db.prepare(`INSERT OR REPLACE INTO channel_qualification_llm_results
(run_id,channel_url,input_hash,prompt_version,model,processing_status,classification,relevance_score,confidence,focus_specificity,channel_type,focus_summary,rationale,supporting_evidence_json,counter_evidence_json,topic_tags_json,formats_json,uncertainty_flags_json,raw_response_json,error_text,created_at)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`);
// A result is stored under the model that actually answered; with a Gemini-to-Ollama fallback
// that can differ from the configured primary model within one run.
const modelsUsed = {};
const saveResult = (item, normalized, status, rawResponse, errorText = null, model = llm.model) => {
if (status === "succeeded") modelsUsed[model] = (modelsUsed[model] ?? 0) + 1;
return insertResult.run(
runId, item.channel.channel_url, item.inputHash, promptVersion, model, status,
normalized?.classification ?? null, normalized?.relevance_score ?? null, normalized?.confidence ?? null,
normalized?.focus_specificity ?? null, normalized?.channel_type ?? null, normalized?.focus_summary ?? null,
normalized?.rationale ?? null, JSON.stringify(normalized?.supporting_evidence ?? []),
JSON.stringify(normalized?.counter_evidence ?? []), JSON.stringify(normalized?.topic_tags ?? []),
JSON.stringify(normalized?.formats ?? []), JSON.stringify(normalized?.uncertainty_flags ?? []),
rawResponse == null ? null : JSON.stringify({ content: String(rawResponse).slice(0, 18000) }), errorText, new Date().toISOString()
);
};
const system = qualificationPrompt();
const counts = { succeeded: 0, failed: 0, batches: 0, individual_retries: 0 };
const resultReport = [];
async function retryItemsIndividually(items, reason) {
if (!config.qualification.llm.retryMissingIndividually || !items.length) return false;
console.warn("[qualification:llm] Batch-level failure; retrying affected channels individually", {
affected: items.length, reason
});
for (const item of items) {
counts.individual_retries += 1;
await callBatch([item], { retry: true });
}
return true;
}
async function callBatch(items, { retry = false } = {}) {
if (!items.length) return;
counts.batches += retry ? 0 : 1;
const user = `Classify these channels as one batch. Return one result for every exact channel_url.\n\n${JSON.stringify({ channels: items.map(item => item.packet) })}`;
let response;
try {
response = await callLLM({ system, user, stream: false });
} catch (error) {
// A transport/model error is usually shared by the whole request (e.g. a
// missing model tag). Do not multiply a systemic failure into N calls.
for (const item of items) {
saveResult(item, null, "failed", null, `request_failed: ${error.message}`);
counts.failed += 1;
resultReport.push({ channel_url: item.channel.channel_url, status: "failed", error: error.message });
}
return;
}
let parsed;
try { parsed = parseJsonObject(response.content); }
catch (error) {
if (shouldRetryIndividually({ retry, enabled: config.qualification.llm.retryMissingIndividually, itemCount: items.length }) && await retryItemsIndividually(items, `response_json_invalid: ${error.message}`)) return;
for (const item of items) {
saveResult(item, null, "failed", response.content, `response_json_invalid: ${error.message}`, response.model);
counts.failed += 1;
resultReport.push({ channel_url: item.channel.channel_url, status: "failed", error: error.message });
}
return;
}
// Some models wrap the reply once more: [{"results":[...]}].
const resultRows = Array.isArray(parsed) ? parsed.flatMap(row => Array.isArray(row?.results) ? row.results : [row]) : parsed?.results;
const byUrl = new Map();
for (const row of Array.isArray(resultRows) ? resultRows : []) {
const url = String(row?.channel_url ?? row?.channelUrl ?? "").trim();
if (url && !byUrl.has(url)) byUrl.set(url, row);
}
const retryCandidates = [];
for (const item of items) {
const raw = byUrl.get(item.channel.channel_url);
if (!raw) { retryCandidates.push(item); continue; }
try {
const normalized = normalizeLlmResult(raw, item.channel.channel_url);
saveResult(item, normalized, "succeeded", response.content, null, response.model);
counts.succeeded += 1;
resultReport.push({ channel_url: item.channel.channel_url, channel_name: item.channel.name, status: "succeeded", classification: normalized.classification, confidence: normalized.confidence, topic_tags: normalized.topic_tags });
} catch (error) {
if (shouldRetryIndividually({ retry, enabled: config.qualification.llm.retryMissingIndividually, itemCount: retryCandidates.length })) {
retryCandidates.push(item);
console.warn("[qualification:llm] Invalid per-channel result; will retry that channel individually", { channel: item.channel.name, error: error.message });
} else {
saveResult(item, null, "failed", response.content, `result_invalid: ${error.message}`, response.model);
counts.failed += 1;
resultReport.push({ channel_url: item.channel.channel_url, status: "failed", error: error.message });
}
}
}
if (retryCandidates.length && !retry && config.qualification.llm.retryMissingIndividually) {
console.warn("[qualification:llm] Batch omitted or malformed channel rows; retrying only affected channels individually", { affected: retryCandidates.length });
for (const item of retryCandidates) {
counts.individual_retries += 1;
await callBatch([item], { retry: true });
}
} else if (retryCandidates.length) {
for (const item of retryCandidates) {
saveResult(item, null, "failed", response.content, "response_missing_or_invalid_channel_result", response.model);
counts.failed += 1;
resultReport.push({ channel_url: item.channel.channel_url, status: "failed", error: "response_missing_or_invalid_channel_result" });
}
}
}
console.log("[qualification:llm] Starting batched ecosystem qualification", {
runId, ecosystem: config.ecosystem.name, model: llm.model, fallback: llm.fallback?.model ?? null,
inputChannels: sourceChannels.length, pending: pending.length, skippedUnchanged, deferredByLimit,
batchSize, promptVersion, criteriaFromConfig: true
});
for (let offset = 0; offset < pending.length; offset += batchSize) {
const batch = pending.slice(offset, offset + batchSize);
console.log("[qualification:llm] Request batch", { batch: Math.floor(offset / batchSize) + 1, totalBatches: Math.ceil(pending.length / batchSize), channels: batch.length, first: batch[0]?.channel.name, last: batch.at(-1)?.channel.name });
await callBatch(batch);
}
const finishedAt = new Date().toISOString();
const finalStatus = counts.failed === 0 ? "completed" : (counts.succeeded > 0 ? "completed_with_failures" : "failed");
db.prepare(`UPDATE channel_qualification_llm_runs SET finished_at=?,success_count=?,failure_count=?,status=? WHERE run_id=?`)
.run(finishedAt, counts.succeeded, counts.failed, finalStatus, runId);
const report = { run_id: runId, started_at: startedAt, finished_at: finishedAt, status: finalStatus, model: llm.model, models_used: modelsUsed, prompt_version: promptVersion,
input_channels: sourceChannels.length, requested_channels: pending.length, skipped_unchanged: skippedUnchanged, deferred_by_limit: deferredByLimit, batch_size: batchSize, ...counts, results: resultReport };
fs.writeFileSync(reportPath, JSON.stringify(report, null, 2));
console.log("[qualification:llm] Complete", { ...report, report: reportPath });
if (counts.failed > 0) process.exitCode = 2;
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.