241 lines
9.1 KiB
TypeScript
241 lines
9.1 KiB
TypeScript
// A1 CBS StatLine via the OData v4 API (long format: one row per measure × dimensions × period).
|
||
// Raw rows go to stats.cbs_obs (+ code lists in stats.cbs_code); the chart series in
|
||
// CBS_SERIES are then normalised into stats.series / stats.obs.
|
||
// Nightly: re-pull a table only when its `Modified` changed (or with --force).
|
||
import {
|
||
GOUDA_RING,
|
||
getJson,
|
||
log,
|
||
parsePeriod,
|
||
run,
|
||
sql,
|
||
writeSeries,
|
||
type ObsRow,
|
||
type SeriesRow
|
||
} from './common';
|
||
import { CBS_SERIES } from './cbs-series';
|
||
|
||
const BASE = 'https://datasets.cbs.nl/odata/v1/CBS';
|
||
const NL = ['NL01', 'NL00'];
|
||
const regionIn = (dim: string, extra: string[] = []) =>
|
||
'(' + [...NL, ...GOUDA_RING, 'CR28', ...extra].map((c) => `${dim} eq '${c}'`).join(' or ') + ')';
|
||
const kwbFilter =
|
||
`(startswith(WijkenEnBuurten,'WK0513') or startswith(WijkenEnBuurten,'BU0513') or ` +
|
||
[...NL, ...GOUDA_RING].map((c) => `WijkenEnBuurten eq '${c}'`).join(' or ') +
|
||
')';
|
||
|
||
export type CbsTable = { id: string; theme: string; filter?: string; note?: string };
|
||
|
||
/** Tables pulled in build step 1. Big tables are filtered to NL + Gouda + its ring. */
|
||
export const CBS_TABLES: CbsTable[] = [
|
||
// energy
|
||
{ id: '85592NED', theme: 'energy' },
|
||
{ id: '84672NED', theme: 'energy', note: 'discontinued 2018–2023 tariffs (history)' },
|
||
{ id: '81309NED', theme: 'energy', note: 'discontinued 2007–2022 prices (history)' },
|
||
{ id: '83140NED', theme: 'energy' },
|
||
{ id: '82610NED', theme: 'energy' },
|
||
{ id: '85523NED', theme: 'energy' },
|
||
{ id: '85005NED', theme: 'energy' },
|
||
{ id: '85237NED', theme: 'energy' },
|
||
// economy
|
||
{ id: '86141NED', theme: 'economy' },
|
||
{ id: '83136NED', theme: 'economy' },
|
||
{ id: '86144NED', theme: 'economy', filter: "Bestedingscategorieen eq 'T001112'" },
|
||
{ id: '85865NED', theme: 'economy' },
|
||
{ id: '85773NED', theme: 'economy' },
|
||
// budget
|
||
{ id: '84120NED', theme: 'budget' },
|
||
{ id: '84117NED', theme: 'budget' },
|
||
{ id: '85590NED', theme: 'budget' },
|
||
{ id: '80509ned', theme: 'budget' },
|
||
{ id: '83642NED', theme: 'budget' },
|
||
{ id: '83641NED', theme: 'budget' },
|
||
// gouda
|
||
{ id: '37230ned', theme: 'gouda', filter: regionIn('RegioS') },
|
||
{ id: '37259ned', theme: 'gouda', filter: regionIn('RegioS') },
|
||
{ id: '70072ned', theme: 'gouda', filter: regionIn('RegioS') },
|
||
{ id: '83625NED', theme: 'gouda' },
|
||
{ id: '85819NED', theme: 'gouda' },
|
||
{ id: '86079NED', theme: 'gouda', filter: regionIn('RegioS') },
|
||
{ id: '83162NED', theme: 'gouda' },
|
||
{ id: '70675ned', theme: 'gouda' },
|
||
// Kerncijfers wijken en buurten, one table per year (Gouda wijken/buurten + ring + NL)
|
||
{ id: '86358NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2026' },
|
||
{ id: '86165NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2025' },
|
||
{ id: '85984NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2024' },
|
||
{ id: '85618NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2023' },
|
||
{ id: '85318NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2022' },
|
||
{ id: '85039NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2021' },
|
||
{ id: '84799NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2020' },
|
||
{ id: '84583NED', theme: 'gouda', filter: kwbFilter, note: 'kwb 2019' }
|
||
];
|
||
|
||
type Props = { Title: string; Modified: string; Status: string; ObservationCount?: number };
|
||
type Page<T> = { value: T[]; '@odata.nextLink'?: string };
|
||
|
||
async function all<T>(url: string): Promise<T[]> {
|
||
const out: T[] = [];
|
||
let next: string | undefined = url;
|
||
while (next) {
|
||
const page: Page<T> = await getJson<Page<T>>(next);
|
||
out.push(...page.value);
|
||
next = page['@odata.nextLink'];
|
||
}
|
||
return out;
|
||
}
|
||
|
||
async function pullTable(t: CbsTable, force: boolean, note: (w: string) => void): Promise<number> {
|
||
const props = await getJson<Props>(`${BASE}/${t.id}/Properties`);
|
||
const [known] = await sql<{ modified: Date | null; filter: string | null }[]>`
|
||
select modified, filter from stats.cbs_table where table_id = ${t.id}`;
|
||
if (props.Status && props.Status !== 'Regulier' && !t.note?.includes('discontinued'))
|
||
note(
|
||
`${t.id} status is ${props.Status}: check the CBS catalog for its successor (base-year swap)`
|
||
);
|
||
if (
|
||
!force &&
|
||
known?.modified &&
|
||
new Date(props.Modified).getTime() === known.modified.getTime() &&
|
||
(known.filter ?? null) === (t.filter ?? null)
|
||
) {
|
||
log(`cbs ${t.id}: unchanged since ${props.Modified}`);
|
||
return 0;
|
||
}
|
||
log(`cbs ${t.id}: pulling "${props.Title.trim()}"${t.filter ? ' (filtered)' : ''}`);
|
||
const dims = await getJson<Page<{ Identifier: string; Kind: string; CodesUrl: string }>>(
|
||
`${BASE}/${t.id}/Dimensions`
|
||
);
|
||
const codes: Record<string, unknown>[] = [];
|
||
const measures = await all<{
|
||
Identifier: string;
|
||
Title: string;
|
||
Unit?: string;
|
||
Decimals?: number;
|
||
MeasureGroupId?: string;
|
||
Index?: number;
|
||
}>(`${BASE}/${t.id}/MeasureCodes`);
|
||
for (const m of measures)
|
||
codes.push({
|
||
table_id: t.id,
|
||
dimension: 'Measure',
|
||
code: m.Identifier.trim(),
|
||
title: (m.Title ?? '').trim() || m.Identifier,
|
||
unit: m.Unit?.trim() || null,
|
||
decimals: m.Decimals ?? null,
|
||
group_id: m.MeasureGroupId ?? null,
|
||
sort_index: m.Index ?? null
|
||
});
|
||
const dimIds: string[] = [];
|
||
for (const d of dims.value) {
|
||
dimIds.push(d.Identifier);
|
||
const list = await all<{
|
||
Identifier: string;
|
||
Title: string;
|
||
DimensionGroupId?: string;
|
||
Index?: number;
|
||
}>(d.CodesUrl);
|
||
for (const c of list)
|
||
codes.push({
|
||
table_id: t.id,
|
||
dimension: d.Identifier,
|
||
code: c.Identifier.trim(),
|
||
title: (c.Title ?? '').trim() || c.Identifier,
|
||
unit: null,
|
||
decimals: null,
|
||
group_id: c.DimensionGroupId ?? null,
|
||
sort_index: c.Index ?? null
|
||
});
|
||
}
|
||
const timeDim = dims.value.find((d) => d.Kind === 'TimeDimension')?.Identifier;
|
||
const url =
|
||
`${BASE}/${t.id}/Observations` + (t.filter ? `?$filter=${encodeURIComponent(t.filter)}` : '');
|
||
const raw = await all<Record<string, unknown>>(url);
|
||
const rows = raw.map((o) => {
|
||
const d: Record<string, string> = {};
|
||
for (const id of dimIds) if (id !== timeDim && o[id] != null) d[id] = String(o[id]).trim();
|
||
return {
|
||
table_id: t.id,
|
||
obs_id: Number(o.Id),
|
||
measure: String(o.Measure).trim(),
|
||
dims: sql.json(d),
|
||
period: timeDim ? String(o[timeDim]).trim() : '',
|
||
value: typeof o.Value === 'number' ? o.Value : null,
|
||
value_attribute:
|
||
o.ValueAttribute && o.ValueAttribute !== 'None' ? String(o.ValueAttribute) : null
|
||
};
|
||
});
|
||
await sql.begin(async (tx) => {
|
||
await tx`delete from stats.cbs_obs where table_id = ${t.id}`;
|
||
await tx`delete from stats.cbs_code where table_id = ${t.id}`;
|
||
for (let i = 0; i < codes.length; i += 5000)
|
||
await tx`insert into stats.cbs_code ${tx(codes.slice(i, i + 5000))} on conflict do nothing`;
|
||
for (let i = 0; i < rows.length; i += 5000)
|
||
await tx`insert into stats.cbs_obs ${tx(rows.slice(i, i + 5000))}`;
|
||
await tx`insert into stats.cbs_table (table_id, title, modified, status, filter, obs_count, fetched_at)
|
||
values (${t.id}, ${props.Title.trim()}, ${props.Modified}, ${props.Status ?? null}, ${t.filter ?? null}, ${rows.length}, now())
|
||
on conflict (table_id) do update set title = excluded.title, modified = excluded.modified,
|
||
status = excluded.status, filter = excluded.filter, obs_count = excluded.obs_count, fetched_at = now()`;
|
||
});
|
||
log(`cbs ${t.id}: ${rows.length} observations, ${codes.length} codes`);
|
||
return rows.length;
|
||
}
|
||
|
||
/** Normalise CBS_SERIES from the raw rows into stats.series / stats.obs. */
|
||
export async function normaliseCbs(): Promise<number> {
|
||
const series: SeriesRow[] = [];
|
||
const obs: ObsRow[] = [];
|
||
for (const def of CBS_SERIES) {
|
||
const regions = def.regions ?? [null];
|
||
for (const region of regions) {
|
||
const dims = { ...(def.dims ?? {}) };
|
||
if (region && def.regionDim) dims[def.regionDim] = region.code;
|
||
const rows = await sql<
|
||
{ period: string; value: number | null; value_attribute: string | null }[]
|
||
>`
|
||
select period, value, value_attribute from stats.cbs_obs
|
||
where table_id = ${def.table} and measure = ${def.measure} and dims @> ${sql.json(dims)}`;
|
||
const id = region ? `${def.id}:${region.code}` : def.id;
|
||
let n = 0;
|
||
const seen = new Set<string>();
|
||
for (const r of rows) {
|
||
const p = parsePeriod(r.period);
|
||
if (!p || p.freq !== def.freq || seen.has(p.date)) continue;
|
||
seen.add(p.date);
|
||
const value = r.value == null ? null : r.value * (def.scale ?? 1);
|
||
obs.push({ series_id: id, period: p.date, value, status: r.value_attribute });
|
||
n++;
|
||
}
|
||
if (!n) continue;
|
||
series.push({
|
||
series_id: id,
|
||
source_id: 'cbs',
|
||
source_code: `${def.table}/${def.measure}/${JSON.stringify(dims)}`,
|
||
title: def.title,
|
||
unit: def.unit,
|
||
freq: def.freq,
|
||
region_code: region?.region ?? def.region ?? 'NLD',
|
||
theme: def.theme,
|
||
meta: { table: def.table, measure: def.measure, dims }
|
||
});
|
||
}
|
||
}
|
||
return writeSeries(series, obs);
|
||
}
|
||
|
||
export async function cbs(opts: { force: boolean; only?: string; job: string }) {
|
||
let total = 0;
|
||
await run('cbs', opts.job, async (note) => {
|
||
for (const t of CBS_TABLES) {
|
||
if (opts.only && !opts.only.split(',').includes(t.id)) continue;
|
||
try {
|
||
total += await pullTable(t, opts.force, note);
|
||
} catch (error) {
|
||
note(`${t.id}: ${error instanceof Error ? error.message : String(error)}`);
|
||
}
|
||
}
|
||
const normalised = await normaliseCbs();
|
||
log(`cbs: ${normalised} normalised observations`);
|
||
return total;
|
||
});
|
||
}
|