`src/stages/12-embed-video-titles.js`

Code179 lines7.4 KBsrc/
import { filterQualifiedRows } from "../analysis/qualification-scope.js";
import crypto from "node:crypto";
import { getDb } from "../db.js";
import { ensureResearchSchema } from "../analysis/research-schema.js";
import { embedTexts, embeddingConfig } from "../analysis/embeddings.js";

const db = getDb();
ensureResearchSchema();
const embedding = embeddingConfig();
const inputVersion = "title-prefix-v1";
const allPending = process.argv.includes("--all");
const limitArg = process.argv.find(value => value.startsWith("--limit="));
const parsedLimit = limitArg ? Number(limitArg.split("=")[1]) : 250;
if (!allPending && (!Number.isFinite(parsedLimit) || parsedLimit < 1)) {
  throw new Error("--limit must be a positive number, or use --all to process every pending title.");
}
const limit = allPending ? Infinity : Math.floor(parsedLimit);
const batchArg = process.argv.find(value => value.startsWith("--batch-size="));
const parsedBatchSize = batchArg ? Number(batchArg.split("=")[1]) : 24;
if (!Number.isFinite(parsedBatchSize) || parsedBatchSize < 1) {
  throw new Error("--batch-size must be a positive number.");
}
const batchSize = Math.min(64, Math.floor(parsedBatchSize));
const force = process.argv.includes("--force");
const now = () => new Date().toISOString();
const titleHash = title => crypto.createHash("sha256").update(String(title ?? "").trim()).digest("hex");

const videos = db.prepare(`
  SELECT v.video_id, v.title, v.channel_url, v.channel_name, v.discovered_at,
    (SELECT COUNT(DISTINCT cv.channel_url) FROM channel_videos cv WHERE cv.video_id=v.video_id) AS linked_channel_count
  FROM videos v
  WHERE trim(COALESCE(v.title,'')) <> ''
  ORDER BY v.discovered_at, v.video_id
`).all();
videos.splice(0, videos.length, ...filterQualifiedRows(db, videos, "video", { label: "12-embed-video-titles.js" }));
const existing = new Map(db.prepare(`
  SELECT video_id, title_hash, input_version FROM video_embeddings WHERE model=?
`).all(embedding.model).map(row => [row.video_id, row]));

// Balance incremental work across the corpus instead of embedding one channel's titles first.
const coverageRows = db.prepare(`
  SELECT cv.channel_url, COUNT(DISTINCT ve.video_id) AS embedded_count
  FROM channel_videos cv
  JOIN video_embeddings ve ON ve.video_id=cv.video_id AND ve.model=?
  GROUP BY cv.channel_url
`).all(embedding.model);
const coverage = new Map(coverageRows.map(row => [row.channel_url, row.embedded_count]));
const channelByVideo = new Map();
for (const row of db.prepare(`SELECT video_id, MIN(channel_url) AS channel_url FROM channel_videos GROUP BY video_id`).all()) {
  channelByVideo.set(row.video_id, row.channel_url);
}

const pending = videos.filter(video => {
  const old = existing.get(video.video_id);
  return force || !old || old.title_hash !== titleHash(video.title) || old.input_version !== inputVersion;
}).map(video => ({
  ...video,
  balancingChannel: channelByVideo.get(video.video_id) ?? video.channel_url ?? "",
  title_hash: titleHash(video.title)
}));
function selectBalancedPending(items, channelCoverage, maxItems) {
  const groups = new Map();
  for (const item of items) {
    if (!groups.has(item.balancingChannel)) groups.set(item.balancingChannel, []);
    groups.get(item.balancingChannel).push(item);
  }
  for (const group of groups.values()) {
    group.sort((a, b) => String(a.discovered_at).localeCompare(String(b.discovered_at)) || a.video_id.localeCompare(b.video_id));
  }
  const offsets = new Map();
  const selectedPerChannel = new Map();
  const selected = [];
  while (selected.length < maxItems) {
    let bestChannel = null;
    let bestCoverage = Infinity;
    for (const [channel, group] of groups) {
      if ((offsets.get(channel) ?? 0) >= group.length) continue;
      const effectiveCoverage = (channelCoverage.get(channel) ?? 0) + (selectedPerChannel.get(channel) ?? 0);
      if (effectiveCoverage < bestCoverage || (effectiveCoverage === bestCoverage && String(channel).localeCompare(String(bestChannel)) < 0)) {
        bestCoverage = effectiveCoverage;
        bestChannel = channel;
      }
    }
    if (bestChannel == null) break;
    const offset = offsets.get(bestChannel) ?? 0;
    selected.push(groups.get(bestChannel)[offset]);
    offsets.set(bestChannel, offset + 1);
    selectedPerChannel.set(bestChannel, (selectedPerChannel.get(bestChannel) ?? 0) + 1);
  }
  return selected;
}

const selected = selectBalancedPending(pending, coverage, limit);

console.log("[video-embeddings] Starting title embedding", {
  totalVideos: videos.length,
  alreadyCurrent: videos.length - pending.length,
  pending: pending.length,
  selectedThisRun: selected.length,
  limit,
  batchSize,
  model: embedding.model,
  endpoint: `${embedding.baseUrl}/api/embed`,
  force,
  allPending,
  inputVersion
});

if (!selected.length) {
  console.log("[video-embeddings] No new or changed titles need embeddings.");
  process.exit(0);
}

const save = db.prepare(`
  INSERT INTO video_embeddings (video_id, model, dimensions, vector_json, title_hash, input_version, created_at)
  VALUES (?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT(video_id, model) DO UPDATE SET
    dimensions=excluded.dimensions,
    vector_json=excluded.vector_json,
    title_hash=excluded.title_hash,
    input_version=excluded.input_version,
    created_at=excluded.created_at
`);
let embedded = 0;
for (let start = 0; start < selected.length; start += batchSize) {
  const batch = selected.slice(start, start + batchSize);
  const batchNo = Math.floor(start / batchSize) + 1;
  const batchCount = Math.ceil(selected.length / batchSize);
  console.log(`[video-embeddings] Batch ${batchNo}/${batchCount}: requesting ${batch.length} titles.`, {
    firstVideoId: batch[0]?.video_id,
    lastVideoId: batch.at(-1)?.video_id
  });
  let vectors;
  try {
    vectors = await embedTexts(batch.map(video => `YouTube video title: ${video.title}`), embedding);
  } catch (error) {
    console.error("[video-embeddings] Ollama request failed. Earlier batches remain saved; rerun to resume.", {
      batch: batchNo,
      model: embedding.model,
      endpoint: `${embedding.baseUrl}/api/embed`,
      videoIds: batch.map(video => video.video_id),
      error: error instanceof Error ? error.message : String(error)
    });
    process.exitCode = 1;
    break;
  }
  if (vectors.some(vector => vector.length !== vectors[0].length)) {
    throw new Error(`[video-embeddings] Inconsistent vector dimensions in batch ${batchNo}. Check the configured model.`);
  }
  const timestamp = now();
  db.transaction(() => {
    batch.forEach((video, index) => {
      save.run(video.video_id, embedding.model, vectors[index].length,
        JSON.stringify(vectors[index]), video.title_hash, inputVersion, timestamp);
    });
  })();
  embedded += batch.length;
  console.log(`[video-embeddings] Saved batch ${batchNo}/${batchCount}.`, {
    embeddedThisRun: embedded,
    remainingInSelectedBatch: selected.length - embedded,
    dimensions: vectors[0]?.length
  });
}

const summary = db.prepare(`
  SELECT COUNT(*) AS rows, COUNT(DISTINCT video_id) AS videos,
    MIN(dimensions) AS min_dimensions, MAX(dimensions) AS max_dimensions
  FROM video_embeddings WHERE model=?
`).get(embedding.model);
console.log("[video-embeddings] Run complete", {
  embeddedThisRun: embedded,
  model: embedding.model,
  summary,
  next: pending.length > selected.length
    ? `Rerun to process the next ${Math.min(limit, pending.length - selected.length)} pending titles.`
    : "Run npm run niches:build to cluster titles and score channels."
});
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.