Source code
// Pulls every feed, reads new article pages, measures each report, groups
// reports into events, and scores outlets across the saved history.
const OUTLETS = require('./outlets');
const R = require('./rules');
const { fetchAll } = require('./rss');
const { fetchPage } = require('./fetchtext');
const { extractParagraphs } = require('./text');
const { analyze, articleId } = require('./analyze');
const { groupIntoEvents, timeline, relatedEvents } = require('./cluster');
const { scoreReport, outletScore, median, spearman } = require('./scale');
const { readJson, writeJson, readText, writeText, hasTextStore } = require('./store');
const P = require('./panels');
const REC = require('./records');
const { anchorFromData } = require('./scale');
const { isOpinion } = require('./opinion');
const archive = require('./archive');
const ST = require('./steering');
const TTL_MS = 10 * 60 * 1000;
const WINDOW_MS = 72 * 3600 * 1000;
const PAGE_BUDGET = 40; // article pages read per update
const PAGE_CONCURRENCY = 12;
const RESCORE_BUDGET = 80; // stored texts measured again per update after a rule change
const ANCHOR_PATH = 'method/anchor.json';
const CACHE_PATH = 'data/articles.json';
const OVERRIDES_PATH = 'grouping/overrides.json';
let cache = null;
let inflight = null;
let memorySaved = {};
let archivedAll = false; // first update on each server copies the whole live window into the archive
// Picks the more complete of two records for the same report: one read from
// the full page beats one read from the feed; then more attempts; then newer.
function better(a, b) {
if (!a) return b;
if (!b) return a;
if ((a.lexiconFp === R.LEXICON_FP) !== (b.lexiconFp === R.LEXICON_FP)) return a.lexiconFp === R.LEXICON_FP ? a : b;
if ((a.source === 'page') !== (b.source === 'page')) return a.source === 'page' ? a : b;
if ((a.attempts || 0) !== (b.attempts || 0)) return (a.attempts || 0) > (b.attempts || 0) ? a : b;
return String(a.analyzedAt) >= String(b.analyzedAt) ? a : b;
}
function mergeSaved(a, b) {
const out = { ...a };
for (const [id, f] of Object.entries(b || {})) out[id] = better(out[id], f);
return out;
}
const score = (f) => scoreReport(f);
async function pool(items, n, fn) {
const out = [];
let i = 0;
await Promise.all(Array.from({ length: Math.min(n, items.length) }, async () => {
while (i < items.length) { const k = i++; out[k] = await fn(items[k]); }
}));
return out;
}
async function readPage(article) {
const page = await fetchPage(article.link);
const paragraphs = page.ok ? extractParagraphs(page.html) : [];
const sponsor = ST.sponsorFrom({ html: page.ok ? page.html : '', link: article.link, categories: article.categories });
const meta = page.ok ? { links: P.articleLinks(page.html), table: P.hasDataTable(page.html), via: page.via || null } : {};
return { paragraphs, sponsor, meta, reason: page.ok ? (paragraphs.length ? null : 'no paragraphs found') : page.reason };
}
// Keeps the text of a page read with the outlet's permission, privately, so
// the report can be measured again after a rule change. Never published.
async function keepText(article, page) {
if (!page.paragraphs.length) return false;
try {
return await writeText(article.outletId, articleId(article.link), {
id: articleId(article.link), outletId: article.outletId, link: article.link, title: article.title, summary: article.summary || '',
published: article.published, categories: article.categories || [], paragraphs: page.paragraphs, meta: page.meta, sponsor: page.sponsor, readAt: new Date().toISOString(),
});
} catch (err) { console.error('text save failed', err.message); return false; }
}
// Measures a report again from its stored text (after a rule change).
async function remeasure(f) {
const t = await readText(f.outletId, f.id);
if (!t || !t.paragraphs || !t.paragraphs.length) return null;
const a = { title: t.title, link: t.link, summary: t.summary, published: t.published, outletId: t.outletId, categories: t.categories };
const g = analyze(a, t.paragraphs, score, t.meta || {});
g.cats = t.categories || [];
g.attempts = f.attempts || 1;
g.sponsor = t.sponsor;
g.textKept = true;
g.remeasured = new Date().toISOString();
g.capture = { ...g.capture, at: t.readAt || g.capture.at, remeasuredAt: g.remeasured };
return g;
}
async function build() {
const now = Date.now();
const results = await fetchAll(OUTLETS);
const recent = [];
const seen = new Set();
for (const r of results) for (const a of r.articles) {
if (!a.published) continue;
const t = Date.parse(a.published);
if (now - t > WINDOW_MS || t - now > 3600 * 1000 || seen.has(a.link)) continue;
seen.add(a.link);
recent.push(a);
}
// Saved measurements for every report seen in the last 120 days. Stored
// copies can lag a few seconds behind the latest write, so the copy held in
// memory is merged in and the more complete record of each report wins.
const saved = mergeSaved(await readJson(CACHE_PATH, {}), memorySaved);
let changed = false;
const toArchive = [];
// Reports measured with older word lists are measured again, or dropped
// when they have left the feeds.
// Since 3.1.0, a report whose page text was kept is measured again from
// that text, up to RESCORE_BUDGET per update; until then it is held out of
// the scores. A report with no kept text leaves the live scores.
const recentIds = new Set(recent.map((a) => articleId(a.link)));
const textOn = hasTextStore();
let rescored = 0;
for (const [id, f] of Object.entries(saved)) {
if (f.lexiconFp === R.LEXICON_FP) continue;
if (!f.held) toArchive.push(f); // the older measurement stays in the archive
if (recentIds.has(id)) { saved[id] = { ...f, attempts: 0, source: 'feed', stale: true, held: undefined }; changed = true; continue; }
if (textOn && f.source === 'page' && f.textKept !== false && rescored < RESCORE_BUDGET) {
rescored++;
const g = await remeasure(f);
if (g) { g.opinion = isOpinion(g.link, g.cats); saved[id] = g; toArchive.push(g); changed = true; continue; }
delete saved[id]; changed = true; continue;
}
if (textOn && f.source === 'page' && f.textKept !== false) { if (!f.held) { saved[id] = { ...f, held: true }; changed = true; } continue; }
delete saved[id];
changed = true;
}
const todo = recent.filter((a) => { const f = saved[articleId(a.link)]; return !f || f.stale || f.fv !== ST.FEATURES_V || (f.source === 'feed' && (f.attempts || 0) < 2); });
todo.sort((a, b) => b.published.localeCompare(a.published));
const batch = todo.slice(0, PAGE_BUDGET);
const pages = await pool(batch, PAGE_CONCURRENCY, readPage);
const kept = await pool(batch, PAGE_CONCURRENCY, (a) => keepText(a, pages[batch.indexOf(a)]));
batch.forEach((a, i) => {
const prev = saved[articleId(a.link)];
const f = analyze(a, pages[i].paragraphs, score, pages[i].meta);
f.textKept = Boolean(kept[i]);
if (pages[i].meta && pages[i].meta.via) f.via = pages[i].meta.via;
f.cats = a.categories || [];
f.attempts = (prev && prev.attempts ? prev.attempts : 0) + 1;
if (pages[i].reason) f.pageNote = pages[i].reason;
f.sponsor = pages[i].sponsor;
saved[f.id] = f;
toArchive.push(f);
changed = true;
});
// Ask the Internet Archive to keep a copy of up to 5 newly read pages per
// update, so a report can be checked against the page as it was read.
await requestArchiveCopies(batch.filter((a, i) => pages[i].paragraphs.length).slice(0, 5).map((a) => saved[articleId(a.link)]));
for (const a of recent) {
const id = articleId(a.link);
if (!saved[id]) { saved[id] = { ...analyze(a, [], score), cats: a.categories || [], attempts: 0, pending: true }; changed = true; }
}
// Opinion is decided from the outlet's own labels each time, so a change to
// the published label rules applies to every saved report.
for (const f of Object.values(saved)) f.opinion = isOpinion(f.link, f.cats);
const cutoff = now - R.CONSTANTS.historyDays * 86400000;
// Reports older than the live window leave the live scores but stay in the archive.
for (const [id, f] of Object.entries(saved)) if (Date.parse(f.published) < cutoff) { toArchive.push(f); delete saved[id]; changed = true; }
if (changed) {
try { await writeJson(CACHE_PATH, Object.fromEntries(Object.entries(saved).map(([k, v]) => [k, { ...v, pending: undefined }]))); } catch (err) { console.error('save failed', err.message); }
}
// Every report seen is archived, including ones read only from the feed.
for (const f of Object.values(saved)) if (f.pending) toArchive.push({ ...f, pending: undefined });
if (!archivedAll) toArchive.push(...Object.values(saved).map((f) => ({ ...f, pending: undefined })));
try { await archive.saveReports(toArchive); archivedAll = true; } catch (err) { console.error('archive failed', err.message); }
memorySaved = saved;
const scored = {};
for (const f of Object.values(saved)) if (!f.held) scored[f.id] = { f, s: score(f) };
const overrides = await readJson(OVERRIDES_PATH, { merges: [], splits: [] });
const grouped = applyOverrides(groupIntoEvents(recent), overrides);
const events = grouped
.filter((e) => e.outlets.length >= 2)
.map((e) => {
const reports = e.articles.map((a) => scored[articleId(a.link)]);
const temp = median(reports.map((r) => r.s.signedTemp));
const byOutlet = {};
for (const r of reports) (byOutlet[r.f.outletId] ||= []).push(r);
const outletScores = {};
for (const [o, list] of Object.entries(byOutlet)) {
const news = list.filter((r) => !r.f.opinion);
const pool = news.length ? news : list;
const use = pool.filter((r) => r.s.scored).length ? pool.filter((r) => r.s.scored) : pool;
outletScores[o] = { score: Math.round(use.reduce((n, r) => n + r.s.overall, 0) / use.length), full: use.some((r) => r.s.scored), opinion: !news.length };
}
const newsReports = reports.filter((r) => !r.f.opinion);
const plainest = [...(newsReports.length ? newsReports : reports)].sort((a, b) => b.s.overall - a.s.overall || a.f.published.localeCompare(b.f.published))[0];
return { ...e, reports, typicalTemp: temp, outletScores, timeline: timeline(e), title: plainest.f.title, titleFrom: plainest.f.outletId, facts: factCoverage(reports), related: relatedEvents(e, grouped.filter((x) => x.outlets.length >= 2)) };
})
.sort((a, b) => b.outlets.length - a.outlets.length || b.lastAt.localeCompare(a.lastAt));
const anchor = await dataAnchor(scored);
const outletScoresNow = outletTable(scored, anchor);
eventAdjusted(events, outletScoresNow);
const summaryScores = summaryTable(scored);
// How outlet rankings move between method 3.0.0's arithmetic and this
// method, on the same measurements.
const prevTable = outletTable(Object.fromEntries(Object.values(scored).map(({ f }) => [f.id, { f, s: scoreReport(f, { v300: true }) }])), null, { v300: true });
const pick = (t) => Object.fromEntries(Object.entries(t).filter(([, v]) => v.status === 'full' || v.status === 'provisional').map(([k, v]) => [k, v.raw]));
const rankChange = { from: '3.0.0', to: R.VERSION, rho: spearman(pick(prevTable), pick(outletScoresNow)), old: pick(prevTable), now: pick(outletScoresNow), note: 'The 3.0.0 arithmetic applied to the current measurements.' };
let records = {};
try { records = await REC.runChecks(Object.values(scored).map(({ f }) => f).filter((f) => Date.now() - Date.parse(f.published) < 14 * 86400000)); } catch (err) { console.error('record checks failed', err.message); }
const coverageMap = coverageTable(saved, results);
const ownParty = Object.values(scored).filter(({ f }) => /transparency party/i.test(`${f.title} ${(f.facts || []).join(' ')}`)).map(({ f, s: sc }) => ({ id: f.id, title: f.title, outletId: f.outletId, score: sc.overall }));
const placement = await frontPages(saved);
const steering = {
repetition: ST.repetition(saved, OUTLETS, { now }),
numbers: ST.numberFraming(saved, OUTLETS, { now }),
sponsors: ST.sponsors(saved, OUTLETS),
coverage: ST.coverage(saved, OUTLETS, groupIntoEvents, { now }),
placement,
};
try { await archive.saveDailyScores(outletScoresNow, OUTLETS); } catch (err) { console.error('score snapshot failed', err.message); }
return {
builtAt: new Date(now).toISOString(),
outlets: OUTLETS,
outletById: Object.fromEntries(OUTLETS.map((o) => [o.id, o])),
status: results.map((r) => ({ id: r.outlet.id, name: r.outlet.name, ok: r.ok, error: r.error || null, count: r.articles.length })),
recentCount: recent.length,
pagesRead: batch.length,
pagesPending: Math.max(0, todo.length - batch.length),
saved,
scored,
events,
eventById: Object.fromEntries(events.map((e) => [e.id, e])),
outletScores: outletScoresNow,
summaryScores,
anchor,
records,
textStore: textOn,
rescored,
held: Object.values(saved).filter((f) => f.held).length,
steering,
rankChange,
coverage: coverageMap,
ownParty,
overrides,
};
}
// Manual corrections to event grouping, kept in saved storage and published
// on the changes page. A split moves one report out into an event of its own;
// a merge joins one event into another.
function applyOverrides(events, ov) {
let list = events.map((e) => ({ ...e, articles: [...e.articles] }));
for (const sp of ov.splits || []) {
for (const e of list) {
const i = e.articles.findIndex((a) => articleId(a.link) === sp.report);
if (i < 0 || e.articles.length < 2) continue;
const [a] = e.articles.splice(i, 1);
list.push({ id: `s${sp.report.slice(0, 6)}`, articles: [a], outlets: [a.outletId], firstAt: a.published, lastAt: a.published, confidence: 1, edited: 'split' });
break;
}
}
for (const m of ov.merges || []) {
const into = list.find((e) => e.id === m.into);
const from = list.find((e) => e.id === m.from);
if (!into || !from || into === from) continue;
into.articles = [...into.articles, ...from.articles].sort((x, y) => x.published.localeCompare(y.published));
into.edited = 'merged';
list = list.filter((e) => e !== from);
}
for (const e of list) {
e.outlets = [...new Set(e.articles.map((a) => a.outletId))];
e.firstAt = e.articles[0].published;
e.lastAt = e.articles[e.articles.length - 1].published;
}
return list;
}
// Coverage map: how much of each outlet's saved coverage could be read in full.
function coverageTable(saved, results) {
const out = {};
for (const o of OUTLETS) out[o.id] = { reports: 0, page: 0, headlineOnly: 0, wire: 0, opinion: 0, reasons: {}, feedOk: true };
for (const r of results) if (out[r.outlet.id]) out[r.outlet.id].feedOk = r.ok;
for (const f of Object.values(saved)) {
const c = out[f.outletId];
if (!c) continue;
c.reports++;
if (f.source === 'page') c.page++; else { c.headlineOnly++; if (f.pageNote) c.reasons[f.pageNote] = (c.reasons[f.pageNote] || 0) + 1; }
if (f.wire) c.wire++;
if (f.opinion) c.opinion++;
}
for (const c of Object.values(out)) {
c.fullShare = c.reports ? c.page / c.reports : 0;
c.status = !c.reports ? 'none' : c.fullShare >= 0.6 ? 'full text' : c.fullShare > 0 ? 'partly' : 'headline only';
}
return out;
}
async function requestArchiveCopies(records) {
await Promise.allSettled(records.filter(Boolean).map(async (f) => {
try {
const r = await fetch(`https://web.archive.org/save/${f.link}`, { method: 'GET', headers: { 'user-agent': 'TransparencyNewsBot/2.0 (+https://transparencynews.org/method)' }, signal: AbortSignal.timeout(4000), redirect: 'manual' });
f.capture = { ...(f.capture || {}), saveRequested: r.status < 500 };
} catch { f.capture = { ...(f.capture || {}), saveRequested: false }; }
}));
}
// Front-page snapshot, taken at most once every three hours and saved, so
// every server shows the same latest snapshot. Each day's snapshots are archived.
const PLACEMENT_PATH = 'steering/placement.json';
const PLACEMENT_EVERY = 3 * 3600 * 1000;
let placementMem = null;
async function frontPages(saved) {
let last = placementMem || (await readJson(PLACEMENT_PATH, null));
if (last && Date.now() - Date.parse(last.at) < PLACEMENT_EVERY) { placementMem = last; return last; }
try {
const snap = await ST.placementSnapshot(OUTLETS, saved, fetchPage);
placementMem = snap;
await writeJson(PLACEMENT_PATH, snap);
await archive.savePlacement(snap);
last = snap;
} catch (err) { console.error('front pages failed', err.message); }
return last;
}
// Names and numbers that appear in two or more reports of an event, and
// which reports include each one. Descriptive only; it does not change scores.
function factCoverage(reports) {
const where = {};
for (const r of reports) for (const t of r.f.facts || []) (where[t.toLowerCase()] ||= { t, ids: new Set() }).ids.add(r.f.id);
return Object.values(where).filter((x) => x.ids.size >= 2).sort((a, b) => b.ids.size - a.ids.size).slice(0, 14).map((x) => ({ t: x.t, ids: [...x.ids] }));
}
// Outlet scores across the whole saved history. Only reports read from the
// full page with 150 or more own words are ranked (3.1.0); wire copy is the
// wire service's writing and opinion is labeled by the outlet, so both are
// left out of the outlet's own score.
function outletTable(scored, anchor, { v300 = false } = {}) {
const by = {};
for (const { f, s } of Object.values(scored)) if ((v300 ? s.scored : s.ranked) && !f.opinion && !f.wire) (by[f.outletId] ||= []).push({ s: s.overall, w: f.ownWords });
const table = {};
for (const o of OUTLETS) {
const list = by[o.id] || [];
const words = list.reduce((n, x) => n + x.w, 0);
const res = outletScore(list.map((x) => x.s), o.id, anchor || {});
let status = 'none';
if (res && list.length >= R.CONSTANTS.outletFloorReports && words >= R.CONSTANTS.outletFloorWords) status = list.length < R.CONSTANTS.provisionalBelow ? 'provisional' : 'full';
else if (list.length) status = 'too few';
table[o.id] = { ...(res || {}), words, status };
}
const shown = Object.entries(table).filter(([, v]) => v.status === 'full' || v.status === 'provisional');
for (const [, v] of shown) v.percentile = Math.round((shown.filter(([, w]) => w.score < v.score).length / Math.max(1, shown.length - 1)) * 100);
return table;
}
// Reports read only from the feed's headline and summary: averaged per
// outlet in a separate, labeled list, never in the ranking.
function summaryTable(scored) {
const by = {};
for (const { f, s } of Object.values(scored)) if (s.mode === 'summary' && !f.opinion && !f.wire && f.headWords) (by[f.outletId] ||= []).push(s);
const out = {};
for (const [o, list] of Object.entries(by)) {
const avg = (k) => { const v = list.map((x) => (k ? x.layers[k] : x.overall)).filter((n) => n != null); return v.length ? Math.round(v.reduce((a, b) => a + b, 0) / v.length) : null; };
out[o] = { n: list.length, overall: avg(null), headline: avg('headline'), words: Math.round(list.reduce((a, x) => a + x.steps.words.words, 0) / list.length) };
}
return out;
}
// Event-adjusted outlet score (3.1.0): the outlet's plain average plus how far
// its reports sit from the average report on the same events, using events
// with 3 or more outlets. It answers "compared with others covering the same
// news", so an outlet's topic mix matters less.
// adjusted = plain average + mean over its reports of (report − event average)
function eventAdjusted(events, table) {
const diffs = {};
for (const e of events) {
const ranked = e.reports.filter((r) => r.s.ranked && !r.f.opinion && !r.f.wire);
if (new Set(ranked.map((r) => r.f.outletId)).size < R.CONSTANTS.eventMinOutlets) continue;
const mean = ranked.reduce((a, r) => a + r.s.overall, 0) / ranked.length;
for (const r of ranked) (diffs[r.f.outletId] ||= []).push(r.s.overall - mean);
}
for (const [o, t] of Object.entries(table)) {
const d = diffs[o] || [];
t.eventN = d.length;
t.eventDiff = d.length ? Math.round((d.reduce((a, b) => a + b, 0) / d.length) * 10) / 10 : null;
t.eventAdjusted = d.length >= R.CONSTANTS.eventMinReports && t.raw != null ? Math.round(t.raw + t.eventDiff) : null;
}
}
// The data-based prior (μ0, k), recomputed once a quarter and saved with its
// date, so it changes on a published schedule rather than with every update.
let anchorMem = null;
async function dataAnchor(scored) {
let a = anchorMem || (await readJson(ANCHOR_PATH, null));
// A saved value set from too little data is replaced as soon as there is enough.
const thin = (x) => !x || x.outlets < R.CONSTANTS.anchorMinOutlets || x.reports < R.CONSTANTS.anchorMinReports;
if (a && !thin(a) && Date.now() - Date.parse(a.at) < R.CONSTANTS.anchorEveryDays * 86400000) { anchorMem = a; return a; }
const groups = {};
for (const { f, s } of Object.values(scored)) if (s.ranked && !f.opinion && !f.wire) (groups[f.outletId] ||= []).push(s.overall);
const fresh = anchorFromData(groups);
if (!fresh || thin(fresh)) return null; // the fixed starting values apply until there is enough data
a = { ...fresh, at: new Date().toISOString(), method: R.VERSION, rule: 'μ0 = average of the outlets\' plain averages; k = spread within outlets ÷ spread between outlets, kept between 2 and 30. Outlets with 5 or more ranked reports.' };
anchorMem = a;
try { await writeJson(ANCHOR_PATH, a); } catch (err) { console.error('anchor save failed', err.message); }
return a;
}
// One-time catch-up for a newly added outlet: reads every item in its feed
// published within the history window, so an outlet that posts once a day
// starts with the same history depth as the others.
async function backfill(outletId, limit = 60) {
const outlet = OUTLETS.find((o) => o.id === outletId);
if (!outlet) return { error: 'unknown outlet' };
const [res] = await fetchAll([outlet]);
const cutoff = Date.now() - R.CONSTANTS.historyDays * 86400000;
const saved = mergeSaved(await readJson(CACHE_PATH, {}), memorySaved);
const items = res.articles.filter((a) => a.published && Date.parse(a.published) >= cutoff && !(saved[articleId(a.link)] && saved[articleId(a.link)].lexiconFp === R.LEXICON_FP && saved[articleId(a.link)].source === 'page')).slice(0, limit);
if (!items.length) return { outlet: outlet.id, feedOk: res.ok, feedError: res.error || null, inFeed: res.articles.length, read: 0, fromPage: 0 };
const pages = await pool(items, PAGE_CONCURRENCY, readPage);
const kept = await pool(items, PAGE_CONCURRENCY, (a) => keepText(a, pages[items.indexOf(a)]));
items.forEach((a, i) => {
const f = analyze(a, pages[i].paragraphs, score, pages[i].meta);
f.textKept = Boolean(kept[i]);
f.cats = a.categories || [];
f.attempts = 1;
if (pages[i].reason) f.pageNote = pages[i].reason;
f.sponsor = pages[i].sponsor;
f.opinion = isOpinion(f.link, f.cats);
saved[f.id] = f;
});
await writeJson(CACHE_PATH, saved);
try { await archive.saveReports(items.map((a) => saved[articleId(a.link)])); } catch (err) { console.error('archive failed', err.message); }
memorySaved = saved;
cache = null;
return { outlet: outlet.id, feedOk: res.ok, inFeed: res.articles.length, read: items.length, fromPage: items.filter((_, i) => pages[i].paragraphs.length).length };
}
async function getData() {
if (cache && Date.now() - Date.parse(cache.builtAt) < TTL_MS) return cache;
if (!inflight) inflight = build().then((d) => { cache = d; inflight = null; return d; }, (e) => { inflight = null; throw e; });
return inflight;
}
module.exports = { getData, build, score, mergeSaved, backfill, applyOverrides, OVERRIDES_PATH, outletTable, summaryTable, eventAdjusted, remeasure, keepText };