238 lines
10 KiB
TypeScript
238 lines
10 KiB
TypeScript
// Shared plumbing for the /stats collector (scripts/stats-fetch.ts): DB client, polite HTTP,
|
|
// fetch_run logging, and the normalised layer (stats.source / region / series / obs).
|
|
// Runs on the host (systemd user timers, see deploy/stats-collector) with DB_URL = app role.
|
|
import postgres from 'postgres';
|
|
|
|
if (!process.env.DB_URL) throw new Error('DB_URL is missing');
|
|
export const sql = postgres(process.env.DB_URL, {
|
|
max: 2,
|
|
connect_timeout: 15,
|
|
onnotice: () => {},
|
|
prepare: false
|
|
});
|
|
|
|
export const USER_AGENT =
|
|
'stanrunge.dev-stats/1.0 (+https://stanrunge.dev/stats; open-data collector; stan@stanrunge.dev)';
|
|
|
|
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
|
|
|
|
/** Minimum gap between requests per host (CBS ≤5 req/s → 250 ms; OECD ≤20/min → 3.1 s). */
|
|
const hostGapMs: Record<string, number> = {
|
|
'datasets.cbs.nl': 250,
|
|
'opendata.cbs.nl': 250,
|
|
'sdmx.oecd.org': 3100,
|
|
'fred.stlouisfed.org': 1000,
|
|
'ec.europa.eu': 500,
|
|
'stats.bis.org': 1000
|
|
};
|
|
const lastHit = new Map<string, number>();
|
|
|
|
export async function http(url: string, init: RequestInit = {}, tries = 4): Promise<Response> {
|
|
const host = new URL(url).host;
|
|
for (let attempt = 1; ; attempt++) {
|
|
const gap = hostGapMs[host] ?? 300;
|
|
const wait = (lastHit.get(host) ?? 0) + gap - Date.now();
|
|
if (wait > 0) await sleep(wait);
|
|
lastHit.set(host, Date.now());
|
|
let response: Response | undefined;
|
|
let failure: unknown;
|
|
try {
|
|
response = await fetch(url, {
|
|
...init,
|
|
headers: { 'user-agent': USER_AGENT, accept: '*/*', ...(init.headers ?? {}) },
|
|
signal: AbortSignal.timeout(180_000)
|
|
});
|
|
if (response.ok) return response;
|
|
if (response.status !== 429 && response.status < 500) {
|
|
const body = (await response.text()).slice(0, 300);
|
|
throw new Error(`HTTP ${response.status} for ${url}: ${body}`);
|
|
}
|
|
failure = new Error(`HTTP ${response.status} for ${url}`);
|
|
} catch (error) {
|
|
if (error instanceof Error && error.message.startsWith('HTTP 4')) throw error;
|
|
failure = error;
|
|
}
|
|
if (attempt >= tries) throw failure;
|
|
const retryAfter = Number(response?.headers.get('retry-after'));
|
|
await sleep(
|
|
Number.isFinite(retryAfter) && retryAfter > 0 ? retryAfter * 1000 : 2000 * 2 ** attempt
|
|
);
|
|
}
|
|
}
|
|
|
|
export const getJson = async <T = unknown>(url: string) => (await (await http(url)).json()) as T;
|
|
export const getText = async (url: string) => (await http(url)).text();
|
|
|
|
export function log(...parts: unknown[]) {
|
|
console.log(new Date().toISOString(), ...parts);
|
|
}
|
|
|
|
/** Wraps one collector job in a stats.fetch_run row (rows, ok, error). */
|
|
export async function run(
|
|
sourceId: string,
|
|
job: string,
|
|
fn: (note: (warning: string) => void) => Promise<number>
|
|
): Promise<number> {
|
|
const [{ run_id }] = await sql`
|
|
insert into stats.fetch_run (source_id, job) values (${sourceId}, ${job}) returning run_id`;
|
|
const warnings: string[] = [];
|
|
try {
|
|
const rows = await fn((w) => warnings.push(w));
|
|
await sql`update stats.fetch_run set finished_at = now(), rows = ${rows}, ok = true,
|
|
error = ${warnings.length ? warnings.join('; ').slice(0, 4000) : null} where run_id = ${run_id}`;
|
|
log(
|
|
`${sourceId}/${job}: ok, ${rows} rows${warnings.length ? `, ${warnings.length} warnings` : ''}`
|
|
);
|
|
return rows;
|
|
} catch (error) {
|
|
const message = error instanceof Error ? error.message : String(error);
|
|
await sql`update stats.fetch_run set finished_at = now(), ok = false,
|
|
error = ${[message, ...warnings].join('; ').slice(0, 4000)} where run_id = ${run_id}`;
|
|
log(`${sourceId}/${job}: FAILED ${message}`);
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
// ── Sources and regions ──────────────────────────────────────────────────────────────────
|
|
|
|
export const SOURCES = [
|
|
{
|
|
source_id: 'cbs',
|
|
name: 'CBS StatLine',
|
|
base_url: 'https://datasets.cbs.nl/odata/v1/CBS',
|
|
licence: 'CC BY 4.0',
|
|
key_required: false,
|
|
attribution: 'Source: CBS StatLine (Statistics Netherlands), CC BY 4.0'
|
|
},
|
|
{
|
|
source_id: 'eurostat',
|
|
name: 'Eurostat',
|
|
base_url: 'https://ec.europa.eu/eurostat/api/dissemination/statistics/1.0/data',
|
|
licence: 'CC BY 4.0',
|
|
key_required: false,
|
|
attribution: 'Source: Eurostat, CC BY 4.0'
|
|
},
|
|
{
|
|
source_id: 'fred',
|
|
name: 'FRED (Federal Reserve Bank of St. Louis)',
|
|
base_url: 'https://fred.stlouisfed.org/graph/fredgraph.csv',
|
|
licence: 'Per series (EIA/BLS/Fed public domain, OECD CC BY 4.0)',
|
|
key_required: false,
|
|
attribution: 'Source: FRED, Federal Reserve Bank of St. Louis (original series sources apply)'
|
|
},
|
|
{
|
|
source_id: 'bis',
|
|
name: 'BIS central bank policy rates',
|
|
base_url: 'https://stats.bis.org/api/v2/data/dataflow/BIS/WS_CBPOL/1.0',
|
|
licence: 'Free reuse with attribution',
|
|
key_required: false,
|
|
attribution: 'Source: BIS, central bank policy rates (WS_CBPOL)'
|
|
}
|
|
] as const;
|
|
|
|
export const REGIONS: { region_code: string; name: string; kind: string; parent: string | null }[] =
|
|
[
|
|
{ region_code: 'WLD', name: 'World', kind: 'aggregate', parent: null },
|
|
{ region_code: 'EU27', name: 'European Union (27)', kind: 'aggregate', parent: 'WLD' },
|
|
{ region_code: 'EA', name: 'Euro area', kind: 'aggregate', parent: 'EU27' },
|
|
{ region_code: 'NLD', name: 'Netherlands', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'USA', name: 'United States', kind: 'country', parent: 'WLD' },
|
|
{ region_code: 'KOR', name: 'South Korea', kind: 'country', parent: 'WLD' },
|
|
{ region_code: 'CHN', name: 'China', kind: 'country', parent: 'WLD' },
|
|
{ region_code: 'CUW', name: 'Curaçao', kind: 'country', parent: 'WLD' },
|
|
{ region_code: 'DEU', name: 'Germany', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'FRA', name: 'France', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'BEL', name: 'Belgium', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'ITA', name: 'Italy', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'ESP', name: 'Spain', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'POL', name: 'Poland', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'SWE', name: 'Sweden', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'DNK', name: 'Denmark', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'AUT', name: 'Austria', kind: 'country', parent: 'EU27' },
|
|
{ region_code: 'GBR', name: 'United Kingdom', kind: 'country', parent: 'WLD' },
|
|
{ region_code: 'JPN', name: 'Japan', kind: 'country', parent: 'WLD' },
|
|
{ region_code: 'IND', name: 'India', kind: 'country', parent: 'WLD' },
|
|
{ region_code: 'CR28', name: 'Oost-Zuid-Holland (COROP)', kind: 'corop', parent: 'NLD' },
|
|
{ region_code: 'GM0513', name: 'Gouda', kind: 'municipality', parent: 'CR28' },
|
|
{ region_code: 'GM0627', name: 'Waddinxveen', kind: 'municipality', parent: 'CR28' },
|
|
{ region_code: 'GM1901', name: 'Bodegraven-Reeuwijk', kind: 'municipality', parent: 'CR28' },
|
|
{ region_code: 'GM1892', name: 'Zuidplas', kind: 'municipality', parent: 'CR28' },
|
|
{ region_code: 'GM1931', name: 'Krimpenerwaard', kind: 'municipality', parent: 'CR28' },
|
|
{ region_code: 'GM0484', name: 'Alphen aan den Rijn', kind: 'municipality', parent: 'NLD' },
|
|
{ region_code: 'GM0632', name: 'Woerden', kind: 'municipality', parent: 'NLD' }
|
|
];
|
|
|
|
/** Gouda and its house-hunting ring (CBS gemeente codes). */
|
|
export const GOUDA_RING = ['GM0513', 'GM0627', 'GM1901', 'GM1892', 'GM1931', 'GM0484', 'GM0632'];
|
|
|
|
export async function ensureMeta() {
|
|
await sql`insert into stats.source ${sql(SOURCES as unknown as Record<string, unknown>[])}
|
|
on conflict (source_id) do update set name = excluded.name, base_url = excluded.base_url,
|
|
licence = excluded.licence, key_required = excluded.key_required, attribution = excluded.attribution`;
|
|
await sql`insert into stats.region ${sql(REGIONS)}
|
|
on conflict (region_code) do update set name = excluded.name, kind = excluded.kind, parent = excluded.parent`;
|
|
}
|
|
|
|
export type SeriesRow = {
|
|
series_id: string;
|
|
source_id: string;
|
|
source_code: string;
|
|
title: string;
|
|
unit: string;
|
|
freq: string;
|
|
region_code: string | null;
|
|
theme: string;
|
|
meta: Record<string, unknown>;
|
|
};
|
|
export type ObsRow = {
|
|
series_id: string;
|
|
period: string;
|
|
value: number | null;
|
|
status: string | null;
|
|
};
|
|
|
|
/** Replaces the normalised series + observations for the given series ids. */
|
|
export async function writeSeries(series: SeriesRow[], obs: ObsRow[]): Promise<number> {
|
|
if (!series.length) return 0;
|
|
const known = new Set(REGIONS.map((r) => r.region_code));
|
|
for (const s of series) if (s.region_code && !known.has(s.region_code)) s.region_code = null;
|
|
await sql.begin(async (tx) => {
|
|
for (let i = 0; i < series.length; i += 2000) {
|
|
const chunk = series
|
|
.slice(i, i + 2000)
|
|
.map((s) => ({ ...s, meta: tx.json(s.meta as never) }));
|
|
await tx`insert into stats.series ${tx(chunk)} on conflict (series_id) do update set
|
|
source_id = excluded.source_id, source_code = excluded.source_code, title = excluded.title,
|
|
unit = excluded.unit, freq = excluded.freq, region_code = excluded.region_code,
|
|
theme = excluded.theme, meta = excluded.meta`;
|
|
}
|
|
const ids = series.map((s) => s.series_id);
|
|
await tx`delete from stats.obs where series_id = any(${ids})`;
|
|
for (let i = 0; i < obs.length; i += 5000) {
|
|
await tx`insert into stats.obs ${tx(obs.slice(i, i + 5000), 'series_id', 'period', 'value', 'status')}
|
|
on conflict (series_id, period) do update set value = excluded.value, status = excluded.status, fetched_at = now()`;
|
|
}
|
|
});
|
|
return obs.length;
|
|
}
|
|
|
|
/** Period code → period start date (YYYY-MM-DD) and frequency. Handles CBS, Eurostat, ISO forms. */
|
|
export function parsePeriod(code: string): { date: string; freq: string } | null {
|
|
const c = code.trim();
|
|
let m: RegExpMatchArray | null;
|
|
if ((m = c.match(/^(\d{4})JJ00$/)) || (m = c.match(/^(\d{4})$/)))
|
|
return { date: `${m[1]}-01-01`, freq: 'A' };
|
|
if (
|
|
(m = c.match(/^(\d{4})MM(\d{2})$/)) ||
|
|
(m = c.match(/^(\d{4})-(\d{2})$/)) ||
|
|
(m = c.match(/^(\d{4})M(\d{2})$/))
|
|
)
|
|
return { date: `${m[1]}-${m[2]}-01`, freq: 'M' };
|
|
if ((m = c.match(/^(\d{4})KW0?(\d)$/)) || (m = c.match(/^(\d{4})-?Q(\d)$/)))
|
|
return { date: `${m[1]}-${String((Number(m[2]) - 1) * 3 + 1).padStart(2, '0')}-01`, freq: 'Q' };
|
|
if ((m = c.match(/^(\d{4})-?S(\d)$/)) || (m = c.match(/^(\d{4})HJ0?(\d)$/)))
|
|
return { date: `${m[1]}-${m[2] === '1' ? '01' : '07'}-01`, freq: 'S' };
|
|
if (/^\d{4}-\d{2}-\d{2}$/.test(c)) return { date: c, freq: 'D' };
|
|
return null;
|
|
}
|