Docs/Skills and code/src/

src/stages/07-analyze-corpus.js

Code402 lines20.5 KBsrc/

import { filterQualifiedRows } from "../analysis/qualification-scope.js";
import fs from "node:fs";
import path from "node:path";
import { fileURLToPath } from "node:url";
import { getDb } from "../db.js";
import { config } from "../../config.js";
import { ensureResearchSchema } from "../analysis/research-schema.js";
import { ageInDays, isRecencyEligible } from "../analysis/temporal.js";

const db = getDb();
ensureResearchSchema();
const here = path.dirname(fileURLToPath(import.meta.url));
const outputDir = path.resolve(here, "../../reports");
fs.mkdirSync(outputDir, { recursive: true });
const now = new Date();
const stamp = now.toISOString().replaceAll(":", "-").replaceAll(".", "-");

function tableCount(table) {
  return db.prepare(`SELECT COUNT(*) AS n FROM ${table}`).get().n;
}
function safeJson(text, fallback = {}) {
  try { return JSON.parse(text); } catch { return fallback; }
}
function round(value, digits = 3) {
  return Number.isFinite(value) ? Number(value.toFixed(digits)) : null;
}

const overview = {
  generatedAt: now.toISOString(),
  ecosystem: config.ecosystem.name,
  counts: Object.fromEntries([
    "channels", "channel_metadata", "channel_profiles", "channel_qualifications",
    "videos", "channel_videos", "concepts", "channel_concepts", "patterns",
    "channel_patterns", "video_profiles", "video_concepts", "video_patterns",
    "video_observations", "editorial_annotations", "vocabulary_terms", "vocabulary_evidence",
    "vocabulary_embeddings", "vocabulary_relations", "video_feature_values",
    "video_embeddings", "channel_priority_runs", "video_niche_clusters", "video_niche_memberships", "channel_priority_scores"
  ].map(table => {
    try { return [table, tableCount(table)]; } catch { return [table, null]; }
  }))
};

const qualifications = db.prepare(`
  SELECT classification, COUNT(*) AS channels
  FROM channel_qualifications GROUP BY classification ORDER BY channels DESC
`).all();
const editorial = db.prepare(`
  SELECT editorial_status, important, COUNT(*) AS n
  FROM editorial_annotations WHERE object_type='channel'
  GROUP BY editorial_status, important ORDER BY editorial_status, important
`).all();

const conceptFrequency = db.prepare(`
  SELECT c.concept_type AS type, c.canonical_name AS name,
         COUNT(DISTINCT cc.channel_url) AS channel_count,
         COUNT(*) AS links,
         GROUP_CONCAT(DISTINCT ch.name) AS example_channels
  FROM concepts c
  JOIN channel_concepts cc ON cc.concept_id=c.id
  JOIN channels ch ON ch.channel_url=cc.channel_url
  WHERE c.ecosystem=?
  GROUP BY c.id
  ORDER BY channel_count DESC, links DESC, c.canonical_name COLLATE NOCASE
`).all(config.ecosystem.name);
const patternFrequency = db.prepare(`
  SELECT p.pattern_type AS type, p.canonical_name AS name,
         COUNT(DISTINCT cp.channel_url) AS channel_count,
         COUNT(*) AS links,
         GROUP_CONCAT(DISTINCT ch.name) AS example_channels
  FROM patterns p
  JOIN channel_patterns cp ON cp.pattern_id=p.id
  JOIN channels ch ON ch.channel_url=cp.channel_url
  WHERE p.ecosystem=?
  GROUP BY p.id
  ORDER BY channel_count DESC, links DESC, p.canonical_name COLLATE NOCASE
`).all(config.ecosystem.name);

const channels = db.prepare(`
  SELECT c.channel_url, c.name, c.discovered_at,
    COALESCE(m.subscriber_count,c.subscriber_count) AS subscribers,
    COALESCE(m.video_count,c.video_count) AS declared_videos,
    cp.profile_json,
    q.classification AS machine_classification,
    q.score AS machine_score,
    ea.important, ea.editorial_status, ea.note AS editorial_note,
    (SELECT COUNT(DISTINCT cc.concept_id) FROM channel_concepts cc WHERE cc.channel_url=c.channel_url) AS concept_count,
    (SELECT COUNT(DISTINCT cp2.pattern_id) FROM channel_patterns cp2 WHERE cp2.channel_url=c.channel_url) AS pattern_count,
    (SELECT COUNT(*) FROM channel_videos cv WHERE cv.channel_url=c.channel_url) AS sampled_video_count
  FROM channels c
  LEFT JOIN channel_metadata m ON m.channel_url=c.channel_url
  LEFT JOIN channel_profiles cp ON cp.channel_url=c.channel_url
  LEFT JOIN channel_qualifications q ON q.channel_url=c.channel_url
  LEFT JOIN editorial_annotations ea ON ea.object_type='channel' AND ea.object_id=c.channel_url
  ORDER BY c.name COLLATE NOCASE
`).all();
channels.splice(0, channels.length, ...filterQualifiedRows(db, channels, "channel", { label: "07-analyze-corpus.js" }));

const conceptRows = db.prepare(`
  SELECT cc.channel_url, c.concept_type AS type, c.normalized_name AS term
  FROM channel_concepts cc JOIN concepts c ON c.id=cc.concept_id
  WHERE c.ecosystem=?
`).all(config.ecosystem.name);
conceptRows.splice(0, conceptRows.length, ...filterQualifiedRows(db, conceptRows, "channel", { label: "07-analyze-corpus.js" }));
const patternRows = db.prepare(`
  SELECT cp.channel_url, p.pattern_type AS type, p.normalized_name AS term
  FROM channel_patterns cp JOIN patterns p ON p.id=cp.pattern_id
  WHERE p.ecosystem=?
`).all(config.ecosystem.name);
patternRows.splice(0, patternRows.length, ...filterQualifiedRows(db, patternRows, "channel", { label: "07-analyze-corpus.js" }));
const vocabByChannel = new Map(channels.map(c => [c.channel_url, new Set()]));
for (const row of [...conceptRows, ...patternRows]) {
  if (!vocabByChannel.has(row.channel_url)) vocabByChannel.set(row.channel_url, new Set());
  vocabByChannel.get(row.channel_url).add(`${row.type}:${row.term}`);
}

// Transparent greedy clusters: nearest-neighbour graph, not a claim that nature has exactly K clusters.
const similarities = [];
for (let i = 0; i < channels.length; i++) {
  const a = channels[i];
  const aSet = vocabByChannel.get(a.channel_url) ?? new Set();
  let neighbors = [];
  for (let j = 0; j < channels.length; j++) {
    if (i === j) continue;
    const b = channels[j];
    const bSet = vocabByChannel.get(b.channel_url) ?? new Set();
    let intersection = 0;
    for (const item of aSet) if (bSet.has(item)) intersection++;
    const union = aSet.size + bSet.size - intersection;
    const jaccard = union ? intersection / union : 0;
    if (jaccard > 0) neighbors.push({ channelUrl: b.channel_url, channel: b.name, jaccard: round(jaccard), sharedTerms: intersection });
  }
  neighbors.sort((x, y) => y.jaccard - x.jaccard || y.sharedTerms - x.sharedTerms);
  similarities.push({
    channelUrl: a.channel_url,
    channel: a.name,
    uniqueVocabularyTerms: aSet.size,
    nearestNeighbors: neighbors.slice(0, 5)
  });
}

const channelBreadth = channels.map(channel => {
  const set = vocabByChannel.get(channel.channel_url) ?? new Set();
  const terms = [...set];
  const counts = new Map();
  for (const term of terms) {
    const type = term.split(":", 1)[0];
    counts.set(type, (counts.get(type) ?? 0) + 1);
  }
  const total = terms.length;
  const shares = [...counts.values()].map(n => n / (total || 1));
  const entropy = shares.reduce((sum, p) => sum - (p > 0 ? p * Math.log(p) : 0), 0);
  const maxTypeShare = shares.length ? Math.max(...shares) : null;
  return {
    channel: channel.name,
    channelUrl: channel.channel_url,
    vocabularyBreadth: total,
    conceptCount: channel.concept_count,
    patternCount: channel.pattern_count,
    sampledVideoCount: channel.sampled_video_count,
    machineClassification: channel.machine_classification ?? "unknown",
    machineScore: channel.machine_score,
    editorialStatus: channel.editorial_status ?? "unreviewed",
    important: Boolean(channel.important),
    typeCounts: Object.fromEntries(counts),
    typeEntropy: round(entropy),
    largestTypeShare: round(maxTypeShare)
  };
}).sort((a,b) => b.vocabularyBreadth - a.vocabularyBreadth);

const videoCoverage = db.prepare(`
  SELECT COUNT(*) AS total,
    SUM(CASE WHEN view_count IS NOT NULL AND view_count > 0 THEN 1 ELSE 0 END) AS with_views,
    SUM(CASE WHEN published_at_raw IS NOT NULL AND trim(published_at_raw) <> '' THEN 1 ELSE 0 END) AS with_published_at_raw,
    SUM(CASE WHEN published_at_normalized IS NOT NULL THEN 1 ELSE 0 END) AS with_published_at_normalized,
    SUM(CASE WHEN published_at_precision LIKE 'exact_%' THEN 1 ELSE 0 END) AS exact_dates,
    SUM(CASE WHEN published_at_precision LIKE 'estimated_%' THEN 1 ELSE 0 END) AS estimated_dates,
    SUM(CASE WHEN published_at_precision = 'unknown' THEN 1 ELSE 0 END) AS unparseable_dates
  FROM videos
`).get();
const subscriberCoverage = db.prepare(`
  SELECT COUNT(*) AS total,
    SUM(CASE WHEN COALESCE(m.subscriber_count,c.subscriber_count) > 0 THEN 1 ELSE 0 END) AS with_subscribers
  FROM channels c LEFT JOIN channel_metadata m ON m.channel_url=c.channel_url
`).get();

const performanceCandidates = db.prepare(`
  SELECT v.video_id, v.title, v.url, v.channel_url, v.channel_name,
    v.view_count, v.published_at, v.published_at_raw,
    v.published_at_normalized, v.published_at_precision,
    COALESCE(m.subscriber_count,c.subscriber_count) AS subscribers
  FROM videos v
  LEFT JOIN channels c ON c.channel_url=v.channel_url
  LEFT JOIN channel_metadata m ON m.channel_url=v.channel_url
  WHERE v.view_count > 0
  ORDER BY v.view_count DESC
`).all().map(v => {
  const eligible = isRecencyEligible(v.published_at_precision);
  const ageDays = eligible ? ageInDays(v.published_at_normalized, now) : null;
  const viewsPerDay = ageDays ? v.view_count / ageDays : null;
  const viewsPerSubscriber = v.subscribers > 0 ? v.view_count / v.subscribers : null;
  return {
    videoId: v.video_id, title: v.title, url: v.url, channel: v.channel_name,
    views: v.view_count, publishedAtRaw: v.published_at_raw ?? v.published_at,
    publishedAtNormalized: v.published_at_normalized,
    publishedAtPrecision: v.published_at_precision ?? "not_normalized",
    subscribers: v.subscribers,
    ageDays: ageDays == null ? null : round(ageDays, 1),
    viewsPerDay: viewsPerDay == null ? null : round(viewsPerDay, 1),
    viewsPerSubscriber: viewsPerSubscriber == null ? null : round(viewsPerSubscriber, 2),
    recencyEligible: eligible,
    caveat: "snapshot-based candidate only; relative publication dates are estimates and coarse month/year estimates are excluded from recency ranking"
  };
});
const relativeOutliers = performanceCandidates
  .filter(v => v.viewsPerSubscriber != null && v.ageDays != null && v.ageDays <= 365 * 3)
  .sort((a,b) => (b.viewsPerSubscriber * Math.log10(1 + (b.viewsPerDay ?? 0))) - (a.viewsPerSubscriber * Math.log10(1 + (a.viewsPerDay ?? 0))))
  .slice(0, 100);

const allVideoSignals = db.prepare(`
  SELECT vo.video_id, vo.channel_url, vo.source, vo.observation_json,
         v.title, v.channel_name, v.view_count, v.published_at_raw,
         v.published_at_normalized, v.published_at_precision
  FROM video_observations vo JOIN videos v ON v.video_id=vo.video_id
  WHERE vo.ecosystem=?
`).all(config.ecosystem.name).map(row => ({
  ...row,
  observation: safeJson(row.observation_json),
  publishedMs: isRecencyEligible(row.published_at_precision) && row.published_at_normalized
    ? Date.parse(row.published_at_normalized) : NaN,
  recencyEligible: isRecencyEligible(row.published_at_precision)
}));
function valuesForObservation(observation, field) {
  if (field === "entities" || field === "themes") return Array.isArray(observation[field]) ? observation[field] : [];
  const mapping = {
    angle: [observation.angle],
    format_family: [observation.format_family ?? observation.format],
    hook_mechanism: [observation.hook_mechanism],
    hook_text: [observation.hook_text ?? (observation.schema_version?.includes("v1") ? observation.hook : "")],
    promise_type: [observation.promise_type],
    promise_text: [observation.promise_text ?? observation.promise],
    packaging_pattern: [observation.packaging_pattern ?? observation.packaging]
  };
  return mapping[field] ?? [];
}
function aggregateVideoSignals(field) {
  const terms = new Map();
  for (const row of allVideoSignals) {
    for (const raw of valuesForObservation(row.observation, field)) {
      const name = String(raw && typeof raw === "object" ? raw.name ?? "" : raw ?? "").trim();
      if (!name) continue;
      const key = name.toLocaleLowerCase();
      if (!terms.has(key)) terms.set(key, { type: field, name, videoCount: 0, channelSet: new Set(), recentVideoCount: 0, totalViews: 0, videosWithViews: 0 });
      const entry = terms.get(key);
      entry.videoCount++;
      if (row.channel_url) entry.channelSet.add(row.channel_url);
      if (row.recencyEligible && Number.isFinite(row.publishedMs) && row.publishedMs >= now.getTime() - 180 * 86400000) entry.recentVideoCount++;
      if (row.view_count != null && row.view_count > 0) { entry.totalViews += row.view_count; entry.videosWithViews++; }
    }
  }
  return [...terms.values()].map(entry => ({
    type: entry.type, name: entry.name, videoCount: entry.videoCount,
    channelCount: entry.channelSet.size, recentVideoCount: entry.recentVideoCount,
    meanViews: entry.videosWithViews ? Math.round(entry.totalViews / entry.videosWithViews) : null,
    videosWithViews: entry.videosWithViews
  })).sort((a,b) => b.videoCount-a.videoCount || b.channelCount-a.channelCount || a.name.localeCompare(b.name));
}
const signalFields = ["entities", "themes", "angle", "format_family", "hook_mechanism", "hook_text", "promise_type", "promise_text", "packaging_pattern"];
const videoSignalFrequency = Object.fromEntries(signalFields.map(field => [field, aggregateVideoSignals(field)]));
const recentCutoff = now.getTime() - 180 * 86400000;
const recentObservationCoverage = {
  titleSignals: allVideoSignals.length,
  signalsWithEligiblePublicationDates: allVideoSignals.filter(v => v.recencyEligible).length,
  signalsPublishedInLast180Days: allVideoSignals.filter(v => v.recencyEligible && Number.isFinite(v.publishedMs) && v.publishedMs >= recentCutoff).length,
  interpretation: "Only video_observations are counted; relative dates with month/year precision are excluded from strict recency counts. Run npm run normalize:temporal after ingestion."
};

const referenceCoverage = db.prepare(`
  SELECT COUNT(*) AS important_channels,
    SUM(CASE WHEN EXISTS (SELECT 1 FROM channel_videos cv WHERE cv.channel_url=ea.object_id) THEN 1 ELSE 0 END) AS with_sampled_videos,
    SUM(CASE WHEN EXISTS (SELECT 1 FROM channel_profiles cp WHERE cp.channel_url=ea.object_id) THEN 1 ELSE 0 END) AS with_channel_profiles,
    SUM(CASE WHEN EXISTS (
      SELECT 1 FROM video_observations vo
      WHERE vo.channel_url=ea.object_id AND vo.source='channel_sample_title_signal_v2'
    ) THEN 1 ELSE 0 END) AS with_v2_video_signals
  FROM editorial_annotations ea WHERE ea.object_type='channel' AND ea.important=1
`).get();

const vocabularyStatus = db.prepare(`
  SELECT term_type, status, COUNT(*) AS terms
  FROM vocabulary_terms WHERE ecosystem=?
  GROUP BY term_type, status ORDER BY term_type, status
`).all(config.ecosystem.name);
const vocabularyRelationStatus = db.prepare(`
  SELECT r.relation_type, r.status, COUNT(*) AS relations,
    ROUND(AVG(r.similarity), 4) AS mean_similarity, ROUND(MAX(r.similarity), 4) AS max_similarity
  FROM vocabulary_relations r
  JOIN vocabulary_terms a ON a.term_id=r.term_a_id
  WHERE a.ecosystem=?
  GROUP BY r.relation_type, r.status ORDER BY r.relation_type, r.status
`).all(config.ecosystem.name);
const centralVocabulary = conceptFrequency.slice(0, 100);
const centralPatterns = patternFrequency.slice(0, 100);
const report = {
  ...overview,
  qualifications,
  editorialAnnotations: editorial,
  dataQuality: {
    videoViews: videoCoverage,
    channelSubscribers: subscriberCoverage,
    recentObservationCoverage,
    referenceCoverage,
    vocabularyStatus,
    vocabularyRelationStatus,
    warnings: [
      "Machine qualification is shown as a recorded label, not accepted as ground truth.",
      "Legacy concepts and patterns remain uncanonicalized proposals; the new vocabulary layer does not merge them automatically.",
      "Embedding similarity proposes same-type relationships only; it does not prove aliases or create canonical concepts.",
      "Channel similarity is typed-token Jaccard similarity, not an objective ontology.",
      "Recency analysis excludes relative publication strings with month/year precision; normalized estimates remain estimates.",
      "A single view-count/subscriber snapshot cannot establish growth rate or prove a breakout."
    ]
  },
  centralVocabulary,
  centralPatterns,
  videoSignalFrequency,
  channelBreadth,
  channelSimilarities: similarities,
  performanceCandidates: relativeOutliers
};

function mdTable(rows, columns) {
  const cell = value => String(value ?? "").replaceAll("|", "\\|").replaceAll("\n", " ");
  return [
    `| ${columns.map(c => c.label).join(" | ")} |`,
    `| ${columns.map(() => "---").join(" | ")} |`,
    ...rows.map(row => `| ${columns.map(c => cell(row[c.key])).join(" | ")} |`)
  ].join("\n");
}
const md = [
  `# Corpus analysis — ${config.ecosystem.name}`,
  `Generated: ${report.generatedAt}`,
  "",
  "## Coverage",
  mdTable(Object.entries(report.counts).map(([table,count]) => ({table,count})), [{key:"table",label:"Table"},{key:"count",label:"Rows"}]),
  "",
  "## Recorded qualification labels (not ground truth)",
  mdTable(qualifications, [{key:"classification",label:"Recorded label"},{key:"channels",label:"Channels"}]),
  "",
  "## Most repeated channel vocabulary",
  mdTable(centralVocabulary.slice(0,40), [{key:"type",label:"Type"},{key:"name",label:"Raw term"},{key:"channel_count",label:"Channels"},{key:"links",label:"Links"}]),
  "",
  "## Most repeated channel patterns",
  mdTable(centralPatterns.slice(0,40), [{key:"type",label:"Type"},{key:"name",label:"Raw pattern"},{key:"channel_count",label:"Channels"},{key:"links",label:"Links"}]),
  "",
  "## Broadest observed vocabularies",
  mdTable(channelBreadth.slice(0,40).map(c => ({channel:c.channel, vocabularyBreadth:c.vocabularyBreadth, conceptCount:c.conceptCount, patternCount:c.patternCount, editorialStatus:c.editorialStatus})), [{key:"channel",label:"Channel"},{key:"vocabularyBreadth",label:"Distinct typed terms"},{key:"conceptCount",label:"Concept links"},{key:"patternCount",label:"Pattern links"},{key:"editorialStatus",label:"Human status"}]),
  "",
  "## Video-level title signals (when present)",
  ...signalFields.flatMap(field => [
    `### ${field}`,
    mdTable(videoSignalFrequency[field].slice(0,20), [{key:"name",label:"Observed term"},{key:"videoCount",label:"Videos"},{key:"channelCount",label:"Channels"},{key:"recentVideoCount",label:"Last 180d"},{key:"meanViews",label:"Mean views where known"}]),
    ""
  ]),
  "## Candidate relative-performance outliers",
  mdTable(relativeOutliers.slice(0,40).map(v => ({title:v.title, channel:v.channel, views:v.views, subscribers:v.subscribers, viewsPerSubscriber:v.viewsPerSubscriber, viewsPerDay:v.viewsPerDay, publishedAt:v.publishedAtRaw, precision:v.publishedAtPrecision})), [{key:"title",label:"Video"},{key:"channel",label:"Channel"},{key:"views",label:"Views"},{key:"subscribers",label:"Subs"},{key:"viewsPerSubscriber",label:"Views/sub"},{key:"viewsPerDay",label:"Views/day"},{key:"publishedAt",label:"Raw published"},{key:"precision",label:"Date precision"}]),
  "",
  "## Vocabulary review status",
  mdTable(vocabularyStatus, [{key:"term_type",label:"Type"},{key:"status",label:"Status"},{key:"terms",label:"Terms"}]),
  "",
  "### Similarity proposals (not merges)",
  mdTable(vocabularyRelationStatus, [{key:"relation_type",label:"Relation"},{key:"status",label:"Review status"},{key:"relations",label:"Count"},{key:"mean_similarity",label:"Mean cosine"},{key:"max_similarity",label:"Max cosine"}]),
  "",
  "## Human-reference coverage",
  mdTable([referenceCoverage], [{key:"important_channels",label:"Important channels"},{key:"with_sampled_videos",label:"With sampled videos"},{key:"with_channel_profiles",label:"With channel profiles"},{key:"with_v2_video_signals",label:"With v2 title signals"}]),
  "",
  "## Interpretation warnings",
  ...report.dataQuality.warnings.map(w => `- ${w}`),
  ""
].join("\n");

const jsonPath = path.join(outputDir, `corpus-analysis-${stamp}.json`);
const mdPath = path.join(outputDir, `corpus-analysis-${stamp}.md`);
fs.writeFileSync(jsonPath, JSON.stringify(report, null, 2));
fs.writeFileSync(mdPath, md);
console.log("[analysis] Corpus report complete.");
console.log(JSON.stringify({
  jsonPath, mdPath,
  channels: report.counts.channels,
  channelProfiles: report.counts.channel_profiles,
  videos: report.counts.videos,
  videoTitleObservations: report.counts.video_observations,
  videosWithViews: videoCoverage.with_views ?? 0,
  videosWithRawPublicationDate: videoCoverage.with_published_at_raw ?? 0,
  videosWithNormalizedPublicationDate: videoCoverage.with_published_at_normalized ?? 0,
  exactPublicationDates: videoCoverage.exact_dates ?? 0,
  estimatedPublicationDates: videoCoverage.estimated_dates ?? 0,
  importantChannelCoverage: referenceCoverage,
  relativePerformanceCandidates: relativeOutliers.length,
  warning: "Candidate ranking is exploratory, not proof of momentum."
}, null, 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.