Stan Runge
d19313035f
All checks were successful
Deploy production / deploy (push) Successful in 1m10s
250 lines
7.6 KiB
TypeScript
250 lines
7.6 KiB
TypeScript
// A2 Eurostat dissemination API (JSON-stat 2.0). Each dataset is pulled with a fixed filter
|
||
// (12 countries + EU/EA aggregates, one consumption band / unit) into stats.eurostat_obs, then
|
||
// every (dataset, key) becomes a normalised series `eurostat:<dataset>:<key>`.
|
||
import {
|
||
getJson,
|
||
log,
|
||
parsePeriod,
|
||
run,
|
||
sql,
|
||
writeSeries,
|
||
type ObsRow,
|
||
type SeriesRow
|
||
} from './common';
|
||
|
||
const BASE = 'https://ec.europa.eu/eurostat/api/dissemination/statistics/1.0/data';
|
||
export const EU_GEOS = [
|
||
'NL',
|
||
'DE',
|
||
'FR',
|
||
'BE',
|
||
'IT',
|
||
'ES',
|
||
'PL',
|
||
'SE',
|
||
'DK',
|
||
'AT',
|
||
'EU27_2020',
|
||
'EA20'
|
||
];
|
||
const GEO_REGION: Record<string, string> = {
|
||
NL: 'NLD',
|
||
DE: 'DEU',
|
||
FR: 'FRA',
|
||
BE: 'BEL',
|
||
IT: 'ITA',
|
||
ES: 'ESP',
|
||
PL: 'POL',
|
||
SE: 'SWE',
|
||
DK: 'DNK',
|
||
AT: 'AUT',
|
||
EU27_2020: 'EU27',
|
||
EA20: 'EA',
|
||
EA: 'EA',
|
||
UK: 'GBR',
|
||
JP: 'JPN',
|
||
US: 'USA'
|
||
};
|
||
|
||
type Dataset = {
|
||
code: string;
|
||
theme: string;
|
||
params: Record<string, string | string[]>;
|
||
geos?: string[];
|
||
};
|
||
|
||
export const EUROSTAT_DATASETS: Dataset[] = [
|
||
{
|
||
code: 'nrg_pc_204',
|
||
theme: 'energy',
|
||
// KWH3500 (band Dc) carries 1991–2007; the old nrg_pc_204_h table was folded into nrg_pc_204.
|
||
params: { nrg_cons: ['KWH2500-4999', 'KWH3500'], currency: 'EUR', unit: 'KWH' }
|
||
},
|
||
{
|
||
code: 'nrg_pc_202',
|
||
theme: 'energy',
|
||
// GJ83P70 (band D3, 83.7 GJ) carries the pre-2007 history (was nrg_pc_202_h); it is only published per GJ (gross calorific value).
|
||
params: { nrg_cons: ['GJ20-199', 'GJ83P70'], currency: 'EUR', unit: ['KWH', 'GJ_GCV'] }
|
||
},
|
||
{ code: 'nrg_ind_id', theme: 'energy', params: { siec: 'TOTAL' } },
|
||
{
|
||
code: 'irt_lt_mcby_m',
|
||
theme: 'economy',
|
||
params: {},
|
||
geos: ['NL', 'DE', 'FR', 'IT', 'BE', 'ES', 'AT', 'EA']
|
||
},
|
||
{ code: 'prc_hicp_minr', theme: 'economy', params: { coicop18: 'TOTAL', unit: 'RCH_A' } },
|
||
{ code: 'prc_hicp_manr', theme: 'economy', params: { coicop: 'CP00', unit: 'RCH_A' } },
|
||
{ code: 'prc_hpi_q', theme: 'economy', params: { purchase: 'TOTAL', unit: ['I15_Q', 'RCH_A'] } },
|
||
{ code: 'road_eqr_carpda', theme: 'energy', params: { unit: 'NR' } },
|
||
{
|
||
code: 'migr_imm1ctz',
|
||
theme: 'countries',
|
||
params: { citizen: 'TOTAL', age: 'TOTAL', sex: 'T', agedef: 'COMPLET', unit: 'NR' }
|
||
},
|
||
{
|
||
code: 'gov_10a_main',
|
||
theme: 'budget',
|
||
params: { unit: 'PC_GDP', sector: 'S13', na_item: ['TR', 'TE', 'B9'] }
|
||
},
|
||
{
|
||
code: 'gov_10a_exp',
|
||
theme: 'budget',
|
||
params: { unit: 'PC_GDP', sector: 'S13', na_item: 'TE' }
|
||
},
|
||
{
|
||
code: 'gov_10a_taxag',
|
||
theme: 'budget',
|
||
params: { unit: 'PC_GDP', sector: 'S13_S212', na_item: ['D2_REC', 'D5_REC', 'D61_REC', 'D995'] }
|
||
},
|
||
{
|
||
code: 'gov_10dd_edpt1',
|
||
theme: 'budget',
|
||
params: { unit: 'PC_GDP', sector: 'S13', na_item: ['GD', 'B9'] }
|
||
},
|
||
{
|
||
code: 'nama_10_gdp',
|
||
theme: 'economy',
|
||
params: { unit: ['CP_MEUR', 'CLV_PCH_PRE'], na_item: 'B1GQ' }
|
||
},
|
||
{ code: 'demo_pjan', theme: 'countries', params: { sex: 'T', age: 'TOTAL', unit: 'NR' } }
|
||
];
|
||
|
||
type JsonStat = {
|
||
label: string;
|
||
updated?: string;
|
||
id: string[];
|
||
size: number[];
|
||
dimension: Record<
|
||
string,
|
||
{ label: string; category: { index: Record<string, number>; label: Record<string, string> } }
|
||
>;
|
||
value: Record<string, number> | number[];
|
||
status?: Record<string, string> | string[];
|
||
};
|
||
|
||
function query(d: Dataset) {
|
||
const q = new URLSearchParams({ format: 'JSON', lang: 'EN' });
|
||
for (const g of d.geos ?? EU_GEOS) q.append('geo', g);
|
||
for (const [k, v] of Object.entries(d.params)) for (const x of [v].flat()) q.append(k, x);
|
||
return q.toString();
|
||
}
|
||
|
||
async function pull(d: Dataset): Promise<number> {
|
||
const qs = query(d);
|
||
const js = await getJson<JsonStat>(`${BASE}/${d.code}?${qs}`);
|
||
const order = js.id;
|
||
const codesByDim = order.map((id) => {
|
||
const idx = js.dimension[id].category.index;
|
||
const arr: string[] = [];
|
||
for (const [code, i] of Object.entries(idx)) arr[i] = code;
|
||
return arr;
|
||
});
|
||
const timePos = order.indexOf('time');
|
||
const geoPos = order.indexOf('geo');
|
||
const rows: {
|
||
dataset: string;
|
||
key: string;
|
||
geo: string;
|
||
period: string;
|
||
value: number | null;
|
||
status: string | null;
|
||
}[] = [];
|
||
const values = js.value;
|
||
const status = js.status ?? {};
|
||
const entries: [number, number][] = Array.isArray(values)
|
||
? values.map((v, i) => [i, v] as [number, number]).filter(([, v]) => v != null)
|
||
: Object.entries(values).map(([k, v]) => [Number(k), v]);
|
||
for (const [flat, value] of entries) {
|
||
let rest = flat;
|
||
const coord: number[] = new Array(order.length);
|
||
for (let p = order.length - 1; p >= 0; p--) {
|
||
coord[p] = rest % js.size[p];
|
||
rest = Math.floor(rest / js.size[p]);
|
||
}
|
||
const keyParts: string[] = [];
|
||
for (let p = 0; p < order.length; p++)
|
||
if (p !== timePos && order[p] !== 'freq') keyParts.push(codesByDim[p][coord[p]]);
|
||
rows.push({
|
||
dataset: d.code,
|
||
key: keyParts.join('.'),
|
||
geo: geoPos >= 0 ? codesByDim[geoPos][coord[geoPos]] : '',
|
||
period: codesByDim[timePos][coord[timePos]],
|
||
value: value ?? null,
|
||
status: (Array.isArray(status) ? status[flat] : status[String(flat)]) ?? null
|
||
});
|
||
}
|
||
const dims: Record<string, unknown> = {};
|
||
for (const id of order)
|
||
dims[id] = { label: js.dimension[id].label, labels: js.dimension[id].category.label };
|
||
await sql.begin(async (tx) => {
|
||
await tx`delete from stats.eurostat_obs where dataset = ${d.code}`;
|
||
for (let i = 0; i < rows.length; i += 5000)
|
||
await tx`insert into stats.eurostat_obs ${tx(rows.slice(i, i + 5000))} on conflict do nothing`;
|
||
await tx`insert into stats.eurostat_dataset (code, label, updated, query, dims, obs_count, fetched_at)
|
||
values (${d.code}, ${js.label}, ${js.updated ?? null}, ${qs}, ${tx.json(dims as never)}, ${rows.length}, now())
|
||
on conflict (code) do update set label = excluded.label, updated = excluded.updated, query = excluded.query,
|
||
dims = excluded.dims, obs_count = excluded.obs_count, fetched_at = now()`;
|
||
});
|
||
// Normalise: one series per key.
|
||
const order2 = order.filter((id) => id !== 'time' && id !== 'freq');
|
||
const byKey = new Map<string, typeof rows>();
|
||
for (const r of rows) byKey.set(r.key, [...(byKey.get(r.key) ?? []), r]);
|
||
const series: SeriesRow[] = [];
|
||
const obs: ObsRow[] = [];
|
||
for (const [key, list] of byKey) {
|
||
const parts = key.split('.');
|
||
const labels = order2.map((id, i) => ({
|
||
id,
|
||
code: parts[i],
|
||
label: js.dimension[id].category.label[parts[i]] ?? parts[i]
|
||
}));
|
||
const unit =
|
||
labels.find((l) => l.id === 'unit')?.label ??
|
||
labels.find((l) => l.id === 'currency')?.label ??
|
||
'';
|
||
const geo = labels.find((l) => l.id === 'geo');
|
||
const id = `eurostat:${d.code}:${key}`;
|
||
let freq = 'A';
|
||
for (const r of list) {
|
||
const p = parsePeriod(r.period);
|
||
if (!p) continue;
|
||
freq = p.freq;
|
||
obs.push({ series_id: id, period: p.date, value: r.value, status: r.status });
|
||
}
|
||
series.push({
|
||
series_id: id,
|
||
source_id: 'eurostat',
|
||
source_code: `${d.code}/${key}`,
|
||
title: `${js.label} · ${labels
|
||
.filter((l) => l.id !== 'geo' && l.id !== 'unit')
|
||
.map((l) => l.label)
|
||
.join(', ')}`.replace(/ · $/, ''),
|
||
unit,
|
||
freq,
|
||
region_code: geo ? (GEO_REGION[geo.code] ?? null) : null,
|
||
theme: d.theme,
|
||
meta: { dataset: d.code, dims: Object.fromEntries(labels.map((l) => [l.id, l.code])) }
|
||
});
|
||
}
|
||
await sql`delete from stats.series where source_id = 'eurostat' and meta->>'dataset' = ${d.code}`;
|
||
await writeSeries(series, obs);
|
||
log(`eurostat ${d.code}: ${rows.length} observations, ${series.length} series`);
|
||
return rows.length;
|
||
}
|
||
|
||
export async function eurostat(opts: { only?: string; job: string }) {
|
||
await run('eurostat', opts.job, async (note) => {
|
||
let total = 0;
|
||
for (const d of EUROSTAT_DATASETS) {
|
||
if (opts.only && !opts.only.split(',').includes(d.code)) continue;
|
||
try {
|
||
total += await pull(d);
|
||
} catch (error) {
|
||
note(`${d.code}: ${error instanceof Error ? error.message : String(error)}`);
|
||
}
|
||
}
|
||
return total;
|
||
});
|
||
}
|