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',
|
|
params: { nrg_cons: 'KWH2500-4999', currency: 'EUR', unit: 'KWH' }
|
|
},
|
|
{
|
|
code: 'nrg_pc_202',
|
|
theme: 'energy',
|
|
params: { nrg_cons: 'GJ20-199', currency: 'EUR', unit: 'KWH' }
|
|
},
|
|
{ code: 'nrg_pc_204_h', theme: 'energy', params: { consom: '4161902', currency: 'EUR' } },
|
|
{ code: 'nrg_pc_202_h', theme: 'energy', params: { consom: '4141902', currency: 'EUR' } },
|
|
{ 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;
|
|
});
|
|
}
|