stats step 3: elections, party positions and opinion feeds (schema, collectors, timers); world-opinions data moved into stats.wo_*

scripts/stats/eurobarometer.ts

@@ -0,0 +1,135 @@
+// A16 Standard Eurobarometer, volume A (EU + candidate countries, weighted shares), via the
+// data.europa.eu catalogue (© European Union, reuse under the Commission's reuse policy).
+// Reads QA3 (most important issues facing our country), QA4 (… facing you personally), QA5
+// (… facing the EU) and the trust batteries (QA6_x national/international institutions, QA8_x
+// EU institutions; share "tend to trust"). Only waves published as xlsx; waves already loaded are
+// skipped unless --force.
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import { getJson, http, insertChunks, log, run, sql } from './common';
+import { readXlsx, type Cell } from './xlsx';
+
+const SEARCH =
+	'https://data.europa.eu/api/hub/search/search?q=standard%20eurobarometer&limit=100&filter=dataset';
+
+type Dist = {
+	title?: string | Record<string, string>;
+	download_url?: string[];
+	access_url?: string[];
+};
+type Dataset = { title?: string | Record<string, string>; distributions?: Dist[] };
+const en = (t: Dataset['title']) =>
+	typeof t === 'string' ? t : (t?.en ?? Object.values(t ?? {})[0] ?? '');
+
+type Row = {
+	wave: string;
+	fieldwork_end: string | null;
+	topic: string;
+	qcode: string;
+	item: string;
+	country: string;
+	share: number;
+};
+
+const TOPICS: [RegExp, string][] = [
+	[/^QA3$/, 'mii_country'],
+	[/^QA4$/, 'mii_personal'],
+	[/^QA5$/, 'mii_eu'],
+	[/^QA6_\d+$/, 'trust'],
+	[/^QA8_\d+$/, 'trust_eu']
+];
+
+const str = (c: Cell | undefined) => (c == null ? '' : String(c)).trim();
+const share = (c: Cell | undefined) => {
+	const n = typeof c === 'number' ? c : Number(str(c));
+	return str(c) !== '' && Number.isFinite(n) ? n : null;
+};
+
+export function parseVolumeA(sheets: Awaited<ReturnType<typeof readXlsx>>): Row[] {
+	const content = sheets.find((s) => s.name === 'Content');
+	if (!content) throw new Error('no Content sheet');
+	const wave = str(content.rows.find((r) => /^wave/i.test(str(r[0])))?.[1]);
+	const fw = str(content.rows.find((r) => /^fieldwork/i.test(str(r[0])))?.[1]);
+	const d = fw.match(/(\d{1,2})\/(\d{1,2})\/(\d{4})\s*$/);
+	const fieldwork_end = d ? `${d[3]}-${d[2].padStart(2, '0')}-${d[1].padStart(2, '0')}` : null;
+	if (!wave) throw new Error('no wave in Content sheet');
+	const questions = new Map<string, string>();
+	for (const r of content.rows) {
+		const sheet = str(r[0])
+			.match(/^"?([A-Z][A-Z0-9_]*)"?\)?$|"([A-Z][A-Z0-9_]*)"\)\s*$/)
+			?.slice(1)
+			.find(Boolean);
+		if (sheet) questions.set(sheet, str(r[2]));
+	}
+	const out: Row[] = [];
+	for (const s of sheets) {
+		const topic = TOPICS.find(([re]) => re.test(s.name))?.[1];
+		if (!topic) continue;
+		const h = s.rows.findIndex(
+			(r) => r.some((c) => str(c) === 'BE') && r.some((c) => str(c) === 'NL')
+		);
+		if (h < 0) continue;
+		const countries = s.rows[h].map((c) => str(c).split('\n').at(-1) ?? '');
+		// Answers come in pairs: a French row with counts, then an English row with shares.
+		for (let i = h + 2; i < s.rows.length; i++) {
+			const label = str(s.rows[i][1]);
+			const next = s.rows[i + 1];
+			if (!label || !next || !str(next[1])) continue;
+			const values = countries.map((_, c) => share(next[c]));
+			if (!values.some((v) => v != null && v <= 1)) continue;
+			const item = str(next[1]);
+			i++;
+			let name = item;
+			if (topic.startsWith('trust')) {
+				if (!/^tend to trust/i.test(item)) continue;
+				name = (questions.get(s.name) ?? s.name).split(':-').at(-1)?.trim() ?? s.name;
+			}
+			countries.forEach((country, c) => {
+				const v = values[c];
+				if (c >= 2 && country && v != null && v <= 1)
+					out.push({ wave, fieldwork_end, topic, qcode: s.name, item: name, country, share: v });
+			});
+		}
+	}
+	return out;
+}
+
+export async function eurobarometer(opts: { force: boolean; job: string }) {
+	await run('eurobarometer', opts.job, async (note) => {
+		const res = await getJson<{ result: { results: Dataset[] } }>(SEARCH);
+		const waves = res.result.results
+			.filter((d) => /^standard eurobarometer/i.test(en(d.title)))
+			.flatMap((d) =>
+				(d.distributions ?? [])
+					.filter((x) => /volume[ _]A\.xlsx$/i.test(en(x.title)))
+					.map((x) => ({ dataset: en(d.title), url: x.download_url?.[0] ?? x.access_url?.[0] }))
+			)
+			.filter((w): w is { dataset: string; url: string } => !!w.url);
+		const loaded = new Set(
+			(await sql`select distinct wave from stats.eb_item`).map((r) => r.wave as string)
+		);
+		let rows = 0;
+		for (const w of waves) {
+			const num = w.dataset.match(/(?:STD|Eurobarometer )(\d{2,3})/i)?.[1];
+			if (num && !opts.force && [...loaded].some((l) => l.split('.')[0] === num)) continue;
+			const file = join(tmpdir(), `eb-${process.pid}.xlsx`);
+			await Bun.write(file, await (await http(w.url)).arrayBuffer());
+			let items: Row[] = [];
+			try {
+				items = parseVolumeA(await readXlsx(file));
+			} catch (error) {
+				note(`${w.dataset}: ${error instanceof Error ? error.message : error}`);
+			}
+			await Bun.file(file).delete();
+			if (!items.length) continue;
+			const dedup = new Map(items.map((r) => [`${r.wave}|${r.topic}|${r.item}|${r.country}`, r]));
+			await sql.begin(async (tx) => {
+				await tx`delete from stats.eb_item where wave = ${items[0].wave}`;
+				await insertChunks(tx, 'eb_item', [...dedup.values()]);
+			});
+			log(`eurobarometer ${items[0].wave}: ${dedup.size} rows`);
+			rows += dedup.size;
+		}
+		return rows;
+	});
+}