WIP
This commit is contained in:
felix
89
src/chat/index.js
Normal file
89
src/chat/index.js
Normal file
@@ -0,0 +1,89 @@
|
||||
// The turn endpoint. Mount it on the service that answers:
|
||||
//
|
||||
// mountChat(app, { path: '/assistant', answer });
|
||||
//
|
||||
// answer(turn) is the only thing the service writes. See the plan, §3, for
|
||||
// the turn object. This file knows nothing about what the answer is.
|
||||
import express from 'express';
|
||||
import {
|
||||
readConversation, readHistory, findAssistant,
|
||||
writeMessage, setDocumentUrl, dropDocument,
|
||||
groupsQuery, groupsUpdate,
|
||||
} from './store.js';
|
||||
|
||||
const TEXT = {
|
||||
failed: 'Er ging iets mis. Probeer het opnieuw.',
|
||||
};
|
||||
const TITLE_LENGTH = 80;
|
||||
|
||||
export function mountChat(app, { path = '/assistant', answer }) {
|
||||
if (typeof answer !== 'function') throw new Error('mountChat needs an answer(turn) function');
|
||||
|
||||
app.post(`${path}/conversations/:id/turns`, express.json(), async (req, res) => {
|
||||
const content = String(req.body?.content ?? '').trim();
|
||||
if (!content) return res.status(400).json({ error: 'content is required' });
|
||||
|
||||
// 1. the access check: can this session read the conversation?
|
||||
let conversation;
|
||||
try {
|
||||
conversation = await readConversation(req.params.id, req.get('mu-session-id'));
|
||||
} catch (e) {
|
||||
console.error('[chat] conversation read failed:', e);
|
||||
return res.status(500).json({ error: e.message });
|
||||
}
|
||||
if (!conversation) return res.status(404).json({ error: 'no such conversation' });
|
||||
|
||||
// 2. everything from here on carries the captured groups
|
||||
const query = sparql => groupsQuery(sparql, conversation.groups);
|
||||
const update = sparql => groupsUpdate(sparql, conversation.groups);
|
||||
|
||||
let assistant, history, userMessage;
|
||||
try {
|
||||
assistant = process.env.CHAT_ASSISTANT_URI || await findAssistant(query);
|
||||
history = await readHistory(query, conversation.uri, assistant);
|
||||
userMessage = await writeMessage(update, {
|
||||
conversationUri: conversation.uri,
|
||||
content,
|
||||
maker: conversation.creator,
|
||||
title: conversation.title ? undefined : content.slice(0, TITLE_LENGTH),
|
||||
});
|
||||
} catch (e) {
|
||||
console.error('[chat] could not record the question:', e);
|
||||
return res.status(500).json({ error: e.message });
|
||||
}
|
||||
|
||||
// 3. answer now; the assistant's message comes when it comes
|
||||
res.status(202).json({ id: userMessage.id });
|
||||
|
||||
const turn = {
|
||||
conversation: { uri: conversation.uri, id: conversation.id, title: conversation.title },
|
||||
history,
|
||||
content,
|
||||
query,
|
||||
update,
|
||||
groups: conversation.groups,
|
||||
finished: false,
|
||||
say: (text, attachments = []) => writeMessage(update, {
|
||||
conversationUri: conversation.uri, content: text, maker: assistant, attachments,
|
||||
}),
|
||||
setUrl: (documentUri, url) => setDocumentUrl(update, documentUri, url, conversation.uri),
|
||||
drop: (documentUri) => dropDocument(update, documentUri),
|
||||
};
|
||||
|
||||
try {
|
||||
const text = await answer(turn);
|
||||
if (typeof text === 'string' && text.trim()) await turn.say(text.trim());
|
||||
} catch (e) {
|
||||
console.error('[chat] turn failed:', e);
|
||||
try {
|
||||
await turn.say(TEXT.failed);
|
||||
} catch (e2) {
|
||||
console.error('[chat] could not write the failure message:', e2);
|
||||
}
|
||||
} finally {
|
||||
turn.finished = true;
|
||||
}
|
||||
});
|
||||
|
||||
console.log(`[chat] turns at POST ${path}/conversations/:id/turns`);
|
||||
}
|
||||
226
src/chat/store.js
Normal file
226
src/chat/store.js
Normal file
@@ -0,0 +1,226 @@
|
||||
// The chat's reads and writes. Copy the folder as is; nothing here knows the
|
||||
// service it lives in. Full URIs everywhere: the auth layer rejects a prefixed
|
||||
// name with no PREFIX line with an opaque 500.
|
||||
import { randomUUID } from 'crypto';
|
||||
import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime } from 'mu';
|
||||
|
||||
const ENDPOINT = process.env.MU_SPARQL_ENDPOINT;
|
||||
const HISTORY_LIMIT = Number(process.env.CHAT_HISTORY_LIMIT || 20);
|
||||
|
||||
// The mu template logs mu.query/mu.update the same way (helpers/mu/sparql.js);
|
||||
// these fetch paths bypass it, so log here under the same switches.
|
||||
const LOG_QUERIES = process.env.LOG_SPARQL_QUERIES != undefined
|
||||
? /^(true|1)$/i.test(process.env.LOG_SPARQL_QUERIES)
|
||||
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
|
||||
const LOG_UPDATES = process.env.LOG_SPARQL_UPDATES != undefined
|
||||
? /^(true|1)$/i.test(process.env.LOG_SPARQL_UPDATES)
|
||||
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
|
||||
function logSparql(kind, sparql) {
|
||||
console.log(`[sparql] ${kind}:\n${sparql}`);
|
||||
}
|
||||
|
||||
const T = {
|
||||
type: 'http://www.w3.org/1999/02/22-rdf-syntax-ns#type',
|
||||
uuid: 'http://mu.semte.ch/vocabularies/core/uuid',
|
||||
thread: 'http://rdfs.org/sioc/ns#Thread',
|
||||
post: 'http://rdfs.org/sioc/ns#Post',
|
||||
instantMessage: 'http://rdfs.org/sioc/types#InstantMessage',
|
||||
hasContainer: 'http://rdfs.org/sioc/ns#has_container',
|
||||
content: 'http://rdfs.org/sioc/ns#content',
|
||||
lastActivity: 'http://rdfs.org/sioc/ns#last_activity_date',
|
||||
title: 'http://purl.org/dc/terms/title',
|
||||
created: 'http://purl.org/dc/terms/created',
|
||||
maker: 'http://xmlns.com/foaf/0.1/maker',
|
||||
attachment: 'https://www.w3.org/ns/activitystreams#attachment',
|
||||
document: 'https://www.w3.org/ns/activitystreams#Document',
|
||||
name: 'https://www.w3.org/ns/activitystreams#name',
|
||||
mediaType: 'https://www.w3.org/ns/activitystreams#mediaType',
|
||||
url: 'https://www.w3.org/ns/activitystreams#url',
|
||||
softwareAgent: 'http://www.w3.org/ns/prov#SoftwareAgent',
|
||||
};
|
||||
const MESSAGE_BASE = 'http://data.lblod.info/id/chat-messages/';
|
||||
const DOCUMENT_BASE = 'http://data.lblod.info/id/chat-documents/';
|
||||
|
||||
const u = sparqlEscapeUri;
|
||||
const s = sparqlEscapeString;
|
||||
// The template helper writes ^^xsd:dateTime; the auth layer happens to bind
|
||||
// xsd, but the rule here is full URIs, so spell the type out.
|
||||
const dt = (value) => sparqlEscapeDateTime(value)
|
||||
.replace('xsd:dateTime', '<http://www.w3.org/2001/XMLSchema#dateTime>');
|
||||
|
||||
// --- the two ways to talk to the auth layer ---------------------------------
|
||||
|
||||
// With the caller's session, inside the request. Returns the JSON result and
|
||||
// the allowed-groups header, which is the identity we carry past the 202.
|
||||
// A bare fetch rather than mu.query: the header is on the response, and this
|
||||
// way we do not depend on what the template's helper chooses to return.
|
||||
export async function sessionQuery(sparql, sessionId) {
|
||||
if (LOG_QUERIES) logSparql('query', sparql);
|
||||
const res = await fetch(ENDPOINT, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
Accept: 'application/sparql-results+json',
|
||||
'mu-session-id': sessionId,
|
||||
},
|
||||
body: new URLSearchParams({ query: sparql }).toString(),
|
||||
});
|
||||
if (!res.ok) throw new Error(`query failed (${res.status}): ${firstLine(await res.text())}`);
|
||||
return { json: await res.json(), groups: res.headers.get('mu-auth-allowed-groups') };
|
||||
}
|
||||
|
||||
// With the captured groups, after the 202.
|
||||
export async function groupsQuery(sparql, groups) {
|
||||
if (LOG_QUERIES) logSparql('query', sparql);
|
||||
const res = await fetch(ENDPOINT, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
Accept: 'application/sparql-results+json',
|
||||
'mu-auth-allowed-groups': groups,
|
||||
},
|
||||
body: new URLSearchParams({ query: sparql }).toString(),
|
||||
});
|
||||
if (!res.ok) throw new Error(`query failed (${res.status}): ${firstLine(await res.text())}`);
|
||||
return res.json();
|
||||
}
|
||||
|
||||
export async function groupsUpdate(sparql, groups) {
|
||||
if (LOG_UPDATES) logSparql('update', sparql);
|
||||
const res = await fetch(ENDPOINT, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/sparql-update', 'mu-auth-allowed-groups': groups },
|
||||
body: sparql,
|
||||
});
|
||||
if (!res.ok) throw new Error(`update failed (${res.status}): ${firstLine(await res.text())}`);
|
||||
}
|
||||
|
||||
function firstLine(text) { return String(text).split('\n')[0].slice(0, 300); }
|
||||
const bindings = (json) => json.results.bindings;
|
||||
|
||||
// --- reads --------------------------------------------------------------------
|
||||
|
||||
// The conversation by uuid, as the caller sees it. null when not readable.
|
||||
// Also returns the allowed groups from the same round trip.
|
||||
export async function readConversation(id, sessionId) {
|
||||
const { json, groups } = await sessionQuery(`
|
||||
SELECT ?conversation ?title ?creator WHERE {
|
||||
?conversation ${u(T.type)} ${u(T.thread)} ;
|
||||
${u(T.uuid)} ${s(id)} .
|
||||
OPTIONAL { ?conversation ${u(T.title)} ?title . }
|
||||
OPTIONAL { ?conversation ${u(T.maker)} ?creator . }
|
||||
} LIMIT 1`, sessionId);
|
||||
const b = bindings(json)[0];
|
||||
if (!b) return null;
|
||||
if (!groups) throw new Error('no mu-auth-allowed-groups header on the conversation read; cannot carry the identity past the 202');
|
||||
return { uri: b.conversation.value, id, title: b.title?.value, creator: b.creator?.value, groups };
|
||||
}
|
||||
|
||||
// The last HISTORY_LIMIT messages, oldest first, as { role, content }.
|
||||
export async function readHistory(queryFn, conversationUri, assistantUri) {
|
||||
const json = await queryFn(`
|
||||
SELECT ?content ?maker ?created WHERE {
|
||||
?message ${u(T.hasContainer)} ${u(conversationUri)} ;
|
||||
${u(T.content)} ?content ;
|
||||
${u(T.maker)} ?maker ;
|
||||
${u(T.created)} ?created .
|
||||
} ORDER BY DESC(?created) LIMIT ${HISTORY_LIMIT}`);
|
||||
return bindings(json).reverse().map(b => ({
|
||||
role: b.maker.value === assistantUri ? 'assistant' : 'user',
|
||||
content: b.content.value,
|
||||
}));
|
||||
}
|
||||
|
||||
// The one assistant the caller can read. Throws unless there is exactly one.
|
||||
export async function findAssistant(queryFn) {
|
||||
const json = await queryFn(`
|
||||
SELECT ?agent WHERE { ?agent ${u(T.type)} ${u(T.softwareAgent)} . } LIMIT 2`);
|
||||
const rows = bindings(json);
|
||||
if (rows.length !== 1) {
|
||||
throw new Error(`expected one prov:SoftwareAgent, found ${rows.length}; seed one or set CHAT_ASSISTANT_URI`);
|
||||
}
|
||||
return rows[0].agent.value;
|
||||
}
|
||||
|
||||
// --- writes -------------------------------------------------------------------
|
||||
|
||||
// One message, both types, its attachments, and the conversation touched, in
|
||||
// one update. `title` is only written when given (the first question names
|
||||
// an unnamed conversation). Returns { uri, id, documents: [{ uri, id }] }.
|
||||
export async function writeMessage(updateFn, { conversationUri, content, maker, attachments = [], title }) {
|
||||
const id = randomUUID();
|
||||
const uri = `${MESSAGE_BASE}${id}`;
|
||||
const now = dt(new Date());
|
||||
const documents = attachments.map(a => {
|
||||
const id = randomUUID();
|
||||
return { ...a, id, uri: `${DOCUMENT_BASE}${id}` };
|
||||
});
|
||||
|
||||
// Predicate pairs are joined with a semicolon, so no list ever ends on one.
|
||||
const pairs = [
|
||||
`${u(T.type)} ${u(T.post)}`,
|
||||
`${u(T.type)} ${u(T.instantMessage)}`,
|
||||
`${u(T.uuid)} ${s(id)}`,
|
||||
`${u(T.hasContainer)} ${u(conversationUri)}`,
|
||||
`${u(T.content)} ${s(content)}`,
|
||||
`${u(T.created)} ${now}`,
|
||||
];
|
||||
if (maker) pairs.push(`${u(T.maker)} ${u(maker)}`);
|
||||
for (const d of documents) pairs.push(`${u(T.attachment)} ${u(d.uri)}`);
|
||||
|
||||
const conversationPairs = [`${u(T.lastActivity)} ${now}`];
|
||||
if (title) conversationPairs.push(`${u(T.title)} ${s(title)}`);
|
||||
|
||||
await updateFn(`
|
||||
DELETE {
|
||||
${u(conversationUri)} ${u(T.lastActivity)} ?old .
|
||||
}
|
||||
INSERT {
|
||||
${u(uri)} ${pairs.join(' ;\n ')} .
|
||||
${documents.map(d => `
|
||||
${u(d.uri)} ${[
|
||||
`${u(T.type)} ${u(T.document)}`,
|
||||
`${u(T.uuid)} ${s(d.id)}`,
|
||||
`${u(T.name)} ${s(d.name)}`,
|
||||
`${u(T.mediaType)} ${s(d.mediaType || 'application/octet-stream')}`,
|
||||
...(d.url ? [`${u(T.url)} ${s(d.url)}`] : []),
|
||||
].join(' ;\n ')} .`).join('')}
|
||||
${u(conversationUri)} ${conversationPairs.join(' ;\n ')} .
|
||||
}
|
||||
WHERE {
|
||||
OPTIONAL { ${u(conversationUri)} ${u(T.lastActivity)} ?old . }
|
||||
}`);
|
||||
|
||||
return { uri, id, documents: documents.map(({ uri, id }) => ({ uri, id })) };
|
||||
}
|
||||
|
||||
// The file is ready.
|
||||
export async function setDocumentUrl(updateFn, documentUri, url, conversationUri) {
|
||||
const now = dt(new Date());
|
||||
await updateFn(`
|
||||
DELETE {
|
||||
${u(documentUri)} ${u(T.url)} ?oldUrl .
|
||||
${u(conversationUri)} ${u(T.lastActivity)} ?old .
|
||||
}
|
||||
INSERT {
|
||||
${u(documentUri)} ${u(T.url)} ${s(url)} .
|
||||
${u(conversationUri)} ${u(T.lastActivity)} ${now} .
|
||||
}
|
||||
WHERE {
|
||||
OPTIONAL { ${u(documentUri)} ${u(T.url)} ?oldUrl . }
|
||||
OPTIONAL { ${u(conversationUri)} ${u(T.lastActivity)} ?old . }
|
||||
}`);
|
||||
}
|
||||
|
||||
// The file failed. The card disappears; the caller says why in a message.
|
||||
export async function dropDocument(updateFn, documentUri) {
|
||||
await updateFn(`
|
||||
DELETE {
|
||||
?message ${u(T.attachment)} ${u(documentUri)} .
|
||||
${u(documentUri)} ?p ?o .
|
||||
}
|
||||
WHERE {
|
||||
${u(documentUri)} ?p ?o .
|
||||
OPTIONAL { ?message ${u(T.attachment)} ${u(documentUri)} . }
|
||||
}`);
|
||||
}
|
||||
133
src/db.js
Normal file
133
src/db.js
Normal file
@@ -0,0 +1,133 @@
|
||||
import { query as muQuery, update as muUpdate,
|
||||
sparqlEscapeUri, sparqlEscapeString, sparqlEscapeInt,
|
||||
sparqlEscapeDateTime, sparqlEscapeDate, sparqlEscapeBool } from 'mu';
|
||||
|
||||
export { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeInt,
|
||||
sparqlEscapeDateTime, sparqlEscapeDate, sparqlEscapeBool };
|
||||
|
||||
// The caller's session. Everything that touches report data.
|
||||
// mu attaches mu-session-id / mu-auth-allowed-groups from the incoming
|
||||
// request itself; we never set headers here.
|
||||
export async function sessionQuery(sparql) { return muQuery(sparql); }
|
||||
export async function sessionUpdate(sparql) { return muUpdate(sparql); }
|
||||
|
||||
// The fetch paths below bypass the template's query helpers, so their
|
||||
// logging (LOG_SPARQL_ALL / LOG_SPARQL_QUERIES / LOG_SPARQL_UPDATES, the
|
||||
// same switches helpers/mu/sparql.js reads) is mirrored here.
|
||||
const LOG_QUERIES = process.env.LOG_SPARQL_QUERIES != undefined
|
||||
? /^(true|1)$/i.test(process.env.LOG_SPARQL_QUERIES)
|
||||
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
|
||||
const LOG_UPDATES = process.env.LOG_SPARQL_UPDATES != undefined
|
||||
? /^(true|1)$/i.test(process.env.LOG_SPARQL_UPDATES)
|
||||
: /^(true|1)$/i.test(process.env.LOG_SPARQL_ALL || '');
|
||||
function logSparql(kind, sparql) {
|
||||
console.log(`[sparql] ${kind}:\n${sparql}`);
|
||||
}
|
||||
|
||||
// No session. Only what the stack makes public. Everything shown to the LLM.
|
||||
// Bare HTTP POST: muQuery always attaches the request's session, which would
|
||||
// widen or narrow the answer depending on the caller.
|
||||
export async function publicQuery(sparql) {
|
||||
if (LOG_QUERIES) logSparql('query', sparql);
|
||||
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
'Accept': 'application/sparql-results+json',
|
||||
},
|
||||
body: new URLSearchParams({ query: sparql }).toString(),
|
||||
});
|
||||
if (!res.ok) {
|
||||
const text = await res.text();
|
||||
throw new Error(`public query failed (${res.status}): ${stripStoreError(text)}`);
|
||||
}
|
||||
return res.json();
|
||||
}
|
||||
|
||||
// A bare update, the same trick as publicQuery. muUpdate would attach the
|
||||
// request's session; job writes after the 202 must not depend on it.
|
||||
export async function plainUpdate(sparql, extraHeaders = {}) {
|
||||
if (LOG_UPDATES) logSparql('update', sparql);
|
||||
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/sparql-update',
|
||||
...extraHeaders,
|
||||
},
|
||||
body: sparql,
|
||||
});
|
||||
if (!res.ok) {
|
||||
const text = await res.text();
|
||||
throw new Error(`update failed (${res.status}): ${stripStoreError(text)}`);
|
||||
}
|
||||
}
|
||||
|
||||
// Background half of a run. Call captureGroups() once inside the request,
|
||||
// then send the captured header on every query after the 202: the template
|
||||
// stops attaching the session once the request is answered, and the auth
|
||||
// layer would silently answer from the public graph.
|
||||
export async function captureGroups(sessionQueryFn) {
|
||||
const probe = await sessionQueryFn('SELECT ?s WHERE { ?s ?p ?o } LIMIT 1');
|
||||
const header = probe.headers?.['mu-auth-allowed-groups'];
|
||||
if (!header) {
|
||||
throw new Error('no mu-auth-allowed-groups header on the first response; cannot carry the identity past the 202');
|
||||
}
|
||||
return header;
|
||||
}
|
||||
|
||||
export async function groupsQuery(sparql, allowedGroups) {
|
||||
if (LOG_QUERIES) logSparql('query', sparql);
|
||||
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
'Accept': 'application/sparql-results+json',
|
||||
'mu-auth-allowed-groups': allowedGroups,
|
||||
},
|
||||
body: new URLSearchParams({ query: sparql }).toString(),
|
||||
});
|
||||
if (!res.ok) {
|
||||
const text = await res.text();
|
||||
throw new Error(`query failed (${res.status}): ${stripStoreError(text)}`);
|
||||
}
|
||||
return res.json();
|
||||
}
|
||||
|
||||
export async function groupsUpdate(sparql, allowedGroups, extraHeaders = {}) {
|
||||
if (LOG_UPDATES) logSparql('update', sparql);
|
||||
const res = await fetch(`${process.env.MU_SPARQL_ENDPOINT}`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/sparql-update',
|
||||
'mu-auth-allowed-groups': allowedGroups,
|
||||
...extraHeaders,
|
||||
},
|
||||
body: sparql,
|
||||
});
|
||||
if (!res.ok) {
|
||||
const text = await res.text();
|
||||
throw new Error(`update failed (${res.status}): ${stripStoreError(text)}`);
|
||||
}
|
||||
return res.json();
|
||||
}
|
||||
|
||||
// The store's error text may name graphs or internals we do not pass on.
|
||||
export function stripStoreError(text) {
|
||||
return String(text).split('\n')[0].slice(0, 300);
|
||||
}
|
||||
|
||||
// The per-request session bundle the MCP tools use. Built here (next to the
|
||||
// other mu-touching code) so llm/ never imports from 'mu' or sessionQuery
|
||||
// (CI checks 1 and 2). Read tools use the caller's session through muQuery;
|
||||
// run_report captures the groups once and runs in the background with them.
|
||||
export async function buildSession() {
|
||||
const query = async (sparql) => muQuery(sparql);
|
||||
let groupsCache = null;
|
||||
return {
|
||||
query,
|
||||
get groups() {
|
||||
if (!groupsCache) groupsCache = captureGroups(query);
|
||||
return groupsCache;
|
||||
},
|
||||
};
|
||||
}
|
||||
114
src/llm/describe.js
Normal file
114
src/llm/describe.js
Normal file
@@ -0,0 +1,114 @@
|
||||
// Turn a shapes graph into the flat list an LLM reads well (plan §7.3). Not
|
||||
// Turtle: a plain text menu of entities, their fields and the code lists.
|
||||
// Code lists under INLINE_VALUES_MAX are inlined at boot from the public
|
||||
// graph (publicQuery) and refreshed every VALUES_TTL, so the LLM never has
|
||||
// to search the short ones.
|
||||
|
||||
import { fieldsOf, entityForClass } from '../runner/profile.js';
|
||||
import { publicQuery } from '../db.js';
|
||||
|
||||
const INLINE_VALUES_MAX = Number(process.env.INLINE_VALUES_MAX || 50);
|
||||
const VALUES_TTL = Number(process.env.VALUES_TTL || 3600) * 1000;
|
||||
|
||||
// In-memory cache: profile+field → { values, at }. Refreshed lazily when stale.
|
||||
const cache = new Map();
|
||||
|
||||
export async function codeListValues(profile) {
|
||||
// Returns a map: "shapeUri|predicate|inverse" → [{ uri, label }]
|
||||
const out = {};
|
||||
for (const shape of profile.shapes) {
|
||||
for (const f of shape.fields) {
|
||||
if (!f.class) continue;
|
||||
const key = `${shape.uri}|${f.path}|${f.inverse ? 1 : 0}`;
|
||||
out[key] = await loadCodeList(profile, f, key);
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
async function loadCodeList(profile, f, key) {
|
||||
if (!f.class) return [];
|
||||
const hit = cache.get(key);
|
||||
if (hit && Date.now() - hit.at < VALUES_TTL) return hit.values;
|
||||
let values = [];
|
||||
try {
|
||||
values = await queryCodeList(f.class);
|
||||
} catch (e) {
|
||||
// public graph read failed; report nothing, the LLM will use lookup_values
|
||||
values = [];
|
||||
}
|
||||
cache.set(key, { values, at: Date.now() });
|
||||
return values;
|
||||
}
|
||||
|
||||
async function queryCodeList(type) {
|
||||
const q = `SELECT ?uri ?label WHERE {
|
||||
?uri a <${type}> .
|
||||
OPTIONAL { ?uri <http://www.w3.org/2004/02/skos/core#prefLabel> ?label . }
|
||||
} ORDER BY ?label LIMIT ${INLINE_VALUES_MAX + 1}`;
|
||||
const r = await publicQuery(q);
|
||||
const bindings = r.results.bindings;
|
||||
return bindings.slice(0, INLINE_VALUES_MAX).map(b => ({
|
||||
uri: b.uri.value,
|
||||
label: b.label?.value || b.uri.value,
|
||||
}));
|
||||
}
|
||||
|
||||
export function describeProfile(profile, codeLists = {}) {
|
||||
const lines = [];
|
||||
lines.push('Een spec is Turtle, precies in deze vorm (kopieer deze prefixes):');
|
||||
lines.push('@prefix rep: <http://mu.semte.ch/vocabularies/reporting/> .');
|
||||
lines.push('@prefix sh: <http://www.w3.org/ns/shacl#> .');
|
||||
lines.push('@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .');
|
||||
lines.push('@prefix dct: <http://purl.org/dc/terms/> .');
|
||||
lines.push('');
|
||||
lines.push('<http://data.lblod.info/id/report-specs/<uw-id>> a rep:ReportSpec , sh:NodeShape ;');
|
||||
lines.push(' dct:title "..." ;');
|
||||
lines.push(` rep:profile <${profile.uri}> ;`);
|
||||
lines.push(' sh:targetClass <de rdf:type uit de lijst hieronder> ;');
|
||||
lines.push(' sh:property [ ... ] ; # filters, optioneel');
|
||||
lines.push(' rep:columns ( [ sh:path ( ... ) ; rdfs:label "..." ] ... ) .');
|
||||
lines.push('Elke kolom eindigt op een waardeveld. rep:self als sh:path geeft de eigen URI.');
|
||||
lines.push('');
|
||||
lines.push('sh:targetClass is de volledige URI tussen haakjes achter de entiteit hieronder.');
|
||||
lines.push('Gebruik in sh:path de predicate-URI die achter elke veldnaam staat, niet de veldnaam zelf.');
|
||||
lines.push('Een datumfilter is sh:minInclusive of sh:maxInclusive met "jjjj-mm-dd"^^xsd:dateTime');
|
||||
lines.push('(kopieer ook het xsd-prefix hierboven niet: voeg @prefix xsd: <http://www.w3.org/2001/XMLSchema#> toe).');
|
||||
lines.push('');
|
||||
for (const shape of profile.shapes) {
|
||||
lines.push(`${shape.label || shape.targetClass} — sh:targetClass <${shape.targetClass}>`);
|
||||
for (const f of shape.fields) {
|
||||
const key = `${shape.uri}|${f.path}|${f.inverse ? 1 : 0}`;
|
||||
const kind = f.datatype ? datatypeName(f.datatype)
|
||||
: f.class ? `→ ${conceptLabel(f.class, codeLists[key])}`
|
||||
: f.node ? `→ ${linkedLabel(profile, f.node)}`
|
||||
: '?';
|
||||
const dir = f.inverse ? ' (omgekeerd)' : '';
|
||||
const pathShown = typeof f.path === 'string' ? f.path : '?';
|
||||
lines.push(` ${f.name || pathShown} sh:path <${pathShown}>${dir} ${kind}`);
|
||||
}
|
||||
lines.push('');
|
||||
}
|
||||
return lines.join('\n');
|
||||
}
|
||||
|
||||
function conceptLabel(type, values) {
|
||||
const head = values.slice(0, INLINE_VALUES_MAX);
|
||||
if (!head.length) return short(type);
|
||||
const names = head.map(v => v.label).join(', ');
|
||||
const more = values.length >= INLINE_VALUES_MAX ? ', …' : '';
|
||||
return `${short(type)} — ${head.length} waarden: ${names}${more}`;
|
||||
}
|
||||
|
||||
function linkedLabel(profile, nodeUri) {
|
||||
const s = profile.shapes.find(x => x.uri === nodeUri);
|
||||
return s ? `${s.label || s.targetClass} (veel)` : short(nodeUri);
|
||||
}
|
||||
|
||||
function datatypeName(dt) {
|
||||
return dt.split('#').pop();
|
||||
}
|
||||
|
||||
function short(uri) {
|
||||
return uri.replace(/^https?:\/\/[^/]+\//, '…/').slice(-60);
|
||||
}
|
||||
111
src/llm/lookup.js
Normal file
111
src/llm/lookup.js
Normal file
@@ -0,0 +1,111 @@
|
||||
// lookup_values: search a code list too long to inline. One publicQuery,
|
||||
// regex on str(?label) so language-tagged labels still match. Returns the
|
||||
// matches, the total, and an `exact` flag when a label equals the term (case
|
||||
// aside) so the LLM can settle disambiguation on its own.
|
||||
|
||||
import { publicQuery, sparqlEscapeString } from '../db.js';
|
||||
import { fieldsOf, entityForClass } from '../runner/profile.js';
|
||||
|
||||
const LIMIT = 25;
|
||||
|
||||
// field is "shapeLabel.fieldLabel" as the LLM writes it, or a predicate URI.
|
||||
// We resolve it to the field's sh:class to know what type to search.
|
||||
export async function lookupValues(profile, term, fieldSpec) {
|
||||
const type = resolveClass(profile, fieldSpec);
|
||||
if (!type) return { total: 0, exact: null, matches: [], error: `field "${fieldSpec}" not found in profile "${profile.title}"` };
|
||||
|
||||
const q = `SELECT ?uri ?label ?type WHERE {
|
||||
?uri <http://www.w3.org/2004/02/skos/core#prefLabel> ?label ; a ?type .
|
||||
FILTER(REGEX(str(?label), ${sparqlEscapeString(escapeRegex(term))}, "i"))
|
||||
} ORDER BY ?label LIMIT ${LIMIT * 8}`;
|
||||
const r = await publicQuery(q);
|
||||
const bindings = r.results.bindings;
|
||||
// The profile's sh:class is the type the data should carry, but code lists
|
||||
// sometimes type their entries differently (MAR codes are
|
||||
// ext:supervision/Nomenclature, the profile says skos:Concept). Prefer rows
|
||||
// matching the asked class; fall back to all rows. The query reads more
|
||||
// rows than LIMIT so an exact hit deeper in the alphabet still surfaces.
|
||||
const wanted = bindings.filter(b => type && b.type?.value === type);
|
||||
const rows = wanted.length ? wanted : bindings;
|
||||
// Order: an exact label (case aside) first, then the rest as the store
|
||||
// ordered them. "Gent" must surface before "AGB Erfgoed Gent".
|
||||
const lower = term.toLowerCase();
|
||||
const exactRow = rows.find(b => b.label.value.toLowerCase() === lower);
|
||||
const rest = rows.filter(b => b !== exactRow);
|
||||
const ordered = exactRow ? [exactRow, ...rest] : rest;
|
||||
let matches = ordered.slice(0, LIMIT).map(b => ({
|
||||
label: b.label.value,
|
||||
uri: b.uri.value,
|
||||
}));
|
||||
if (!matches.length) {
|
||||
const byCode = await codeQuery(profile, type, term);
|
||||
matches = byCode;
|
||||
}
|
||||
const exact = matches.find(m => m.label.toLowerCase() === term.toLowerCase());
|
||||
return {
|
||||
total: matches.length >= LIMIT ? `${LIMIT}+` : matches.length,
|
||||
exact: exact ? exact.uri : null,
|
||||
matches,
|
||||
};
|
||||
}
|
||||
|
||||
// MAR codes: the label says "MAR7300 - …" but the searchable code sits on
|
||||
// ext:supervision/nomenclatureCode. One extra query, only when the label
|
||||
// search came up empty.
|
||||
async function codeQuery(profile, type, term) {
|
||||
const q = `SELECT ?uri ?label WHERE {
|
||||
?uri <http://mu.semte.ch/vocabularies/ext/supervision/nomenclatureCode> ?code ;
|
||||
<http://www.w3.org/2004/02/skos/core#prefLabel> ?label .
|
||||
FILTER(CONTAINS(LCASE(str(?code)), LCASE(${sparqlEscapeString(term)})))
|
||||
} ORDER BY ?label LIMIT ${LIMIT}`;
|
||||
try {
|
||||
const r = await publicQuery(q);
|
||||
return r.results.bindings.map(b => ({ label: b.label.value, uri: b.uri.value }));
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
function resolveClass(profile, fieldSpec) {
|
||||
if (fieldSpec.includes('/')) return null;
|
||||
// "shapeLabel.fieldLabel" → find the field carrying sh:class
|
||||
const dot = fieldSpec.lastIndexOf('.');
|
||||
if (dot < 0) {
|
||||
// bare predicate URI
|
||||
for (const shape of profile.shapes) {
|
||||
for (const f of shape.fields) {
|
||||
if (f.path === fieldSpec && f.class) return f.class;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
const shapeLabel = fieldSpec.slice(0, dot);
|
||||
const fieldLabel = fieldSpec.slice(dot + 1);
|
||||
const shape = profile.shapes.find(s => s.label === shapeLabel);
|
||||
if (!shape) return null;
|
||||
const f = shape.fields.find(x => x.name === fieldLabel);
|
||||
// sh:node marks a link, not a code list; the searchable type is the linked
|
||||
// shape's sh:targetClass ("melding.bestuurseenheid" → besluit:Bestuurseenheid).
|
||||
if (f?.class) return f.class;
|
||||
if (f?.node) {
|
||||
const linked = profile.shapes.find(s => s.uri === f.node);
|
||||
return linked?.targetClass || null;
|
||||
}
|
||||
// "bestuurseenheid.naam": the model chained link→value (shape.field on the
|
||||
// linked shape). Resolve it: find the shape whose *name or label* matches,
|
||||
// and whose rdf:type is in the data, then search that type.
|
||||
for (const s of profile.shapes) {
|
||||
if (s.label === shapeLabel || shortType(s.targetClass) === shapeLabel) {
|
||||
return s.targetClass;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function shortType(cls) {
|
||||
return cls.split('#').pop() || cls.split('/').pop();
|
||||
}
|
||||
|
||||
function escapeRegex(s) {
|
||||
return String(s).replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
|
||||
}
|
||||
163
src/llm/loop.js
Normal file
163
src/llm/loop.js
Normal file
@@ -0,0 +1,163 @@
|
||||
// The agent loop. One code path, one provider: POST /v1/chat/completions
|
||||
// with tool calls, OpenAI-compatible (OpenAI, Azure, Ollama, LiteLLM, vLLM,
|
||||
// Mistral). No provider SDKs. Three variables: LLM_BASE_URL, LLM_MODEL,
|
||||
// LLM_API_KEY.
|
||||
//
|
||||
// The loop is the only stateful party: it keeps the message history, calls the
|
||||
// MCP tools through the same handlers a real MCP client would, and stops when
|
||||
// the model writes no more tool calls. validate_spec is driven up to
|
||||
// MAX_REPAIR_ROUNDS, then run_report.
|
||||
//
|
||||
// This module is what the ten-question check runs against. It does not import
|
||||
// from 'mu' or sessionQuery; it receives the tool handlers and a session
|
||||
// bundle from the caller, so it works against any profile set and any session.
|
||||
|
||||
import { buildTools } from './tools.js';
|
||||
|
||||
const MAX_REPAIR_ROUNDS = Number(process.env.MAX_REPAIR_ROUNDS || 3);
|
||||
const LLM_BASE_URL = process.env.LLM_BASE_URL;
|
||||
const LLM_MODEL = process.env.LLM_MODEL;
|
||||
const LLM_API_KEY = process.env.LLM_API_KEY;
|
||||
// LOG_LLM=1 dumps every request to and response from the provider, verbatim.
|
||||
const LOG_LLM = /^(true|1)$/i.test(process.env.LOG_LLM || '');
|
||||
|
||||
const SYSTEM_PROMPT = `You write report specs. You never write SPARQL and you never see report data.
|
||||
|
||||
The profile lists what can be asked. Compose paths by chaining fields.
|
||||
|
||||
Rules:
|
||||
- A column must end on a value, not on a link. Chain one more hop.
|
||||
- Filters go in sh:property. Columns go in rep:columns, in the order you want.
|
||||
- To match a topic rather than a value, list the Dutch words and compounds you
|
||||
would expect and put them all in rep:anyOf.
|
||||
- If a lookup returns nothing, or more than one plausible match, ask the user
|
||||
in Dutch, offering labels. Never guess and never show a URI.
|
||||
- You cannot count, sort, take the first N, or compare two subjects. Say so.
|
||||
- The report's title is the dct:title you write in the spec. There is no other
|
||||
title input.
|
||||
- One conversation may ask for several reports, one run_report per report.
|
||||
|
||||
Always prefix the spec with @prefix lines. Use the prefixes the profile declares.
|
||||
Write the spec as one Turtle block. Then call validate_spec. Fix what it says,
|
||||
up to three rounds. Then call run_report.
|
||||
|
||||
Earlier turns are context only. Answer the last one.`;
|
||||
|
||||
// ask(questionOrMessages, profiles, session) → { text, trace }
|
||||
// A string is one question, as before. An array is the conversation so far,
|
||||
// [{ role: 'user' | 'assistant', content }], newest last. The system prompt
|
||||
// is always this file's own; a system message in the input is dropped, so a
|
||||
// caller cannot widen it. The loop appends every model reply and tool result
|
||||
// to the messages, so a caller can keep one conversation across many reports.
|
||||
// Returns the model's final text plus a trace of every tool call it made (for
|
||||
// the leak test and for debugging).
|
||||
export async function ask(questionOrMessages, profiles, session) {
|
||||
const { handlers } = buildTools(profiles, session);
|
||||
const trace = [];
|
||||
|
||||
const turns = typeof questionOrMessages === 'string'
|
||||
? [{ role: 'user', content: questionOrMessages }]
|
||||
: questionOrMessages.filter(m => m.role !== 'system');
|
||||
const messages = [{ role: 'system', content: SYSTEM_PROMPT }, ...turns];
|
||||
|
||||
for (let round = 0; round < 12; round++) {
|
||||
const reply = await chat(messages);
|
||||
messages.push(reply.message);
|
||||
const calls = reply.message.tool_calls || [];
|
||||
if (!calls.length) {
|
||||
return { text: reply.message.content || '', trace };
|
||||
}
|
||||
for (const call of calls) {
|
||||
const args = parseArgs(call.function.arguments);
|
||||
trace.push({ name: call.function.name, args });
|
||||
const fn = handlers.get(call.function.name);
|
||||
let resultText;
|
||||
if (!fn) {
|
||||
resultText = `unknown tool "${call.function.name}"`;
|
||||
} else {
|
||||
try {
|
||||
const out = await fn(args);
|
||||
resultText = out.content?.[0]?.text ?? JSON.stringify(out);
|
||||
} catch (e) {
|
||||
resultText = `tool failed: ${String(e.message || e).split('\n')[0]}`;
|
||||
}
|
||||
}
|
||||
messages.push({ role: 'tool', tool_call_id: call.id, content: resultText });
|
||||
trace[trace.length - 1].result = resultText;
|
||||
}
|
||||
}
|
||||
return { text: '(loop did not finish)', trace };
|
||||
}
|
||||
|
||||
async function chat(messages) {
|
||||
// Providers echo extra fields (litellm adds provider_specific_fields, Mistral
|
||||
// rejects them on the way back). Send only the portable wire shape.
|
||||
const wire = messages.map(m => {
|
||||
const out = { role: m.role, content: m.content ?? null };
|
||||
if (m.tool_calls) {
|
||||
out.tool_calls = m.tool_calls.map(c => ({
|
||||
id: c.id,
|
||||
type: 'function',
|
||||
function: { name: c.function.name, arguments: c.function.arguments },
|
||||
}));
|
||||
}
|
||||
if (m.role === 'tool' && m.tool_call_id) out.tool_call_id = m.tool_call_id;
|
||||
return out;
|
||||
});
|
||||
const body = {
|
||||
model: LLM_MODEL,
|
||||
messages: wire,
|
||||
tools: toolDefs(),
|
||||
tool_choice: 'auto',
|
||||
};
|
||||
if (LOG_LLM) {
|
||||
console.log(`[llm] POST ${LLM_BASE_URL}/chat/completions request:\n${
|
||||
JSON.stringify(body, null, 2)}`);
|
||||
}
|
||||
const res = await fetch(`${LLM_BASE_URL}/chat/completions`, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
...(LLM_API_KEY ? { Authorization: `Bearer ${LLM_API_KEY}` } : {}),
|
||||
},
|
||||
body: JSON.stringify(body),
|
||||
});
|
||||
if (!res.ok) {
|
||||
const t = await res.text();
|
||||
throw new Error(`LLM ${res.status}: ${t.slice(0, 300)}`);
|
||||
}
|
||||
const data = await res.json();
|
||||
if (LOG_LLM) {
|
||||
console.log(`[llm] response:\n${JSON.stringify(data, null, 2)}`);
|
||||
}
|
||||
return { message: data.choices[0].message };
|
||||
}
|
||||
|
||||
function parseArgs(s) {
|
||||
if (!s) return {};
|
||||
try { return JSON.parse(s); } catch { return {}; }
|
||||
}
|
||||
|
||||
// messagesFor(question) — a fresh conversation: system prompt plus the user
|
||||
// question. Turn this into your own history array to follow up in the same
|
||||
// conversation.
|
||||
export function messagesFor(question) {
|
||||
return [
|
||||
{ role: 'system', content: SYSTEM_PROMPT },
|
||||
{ role: 'user', content: question },
|
||||
];
|
||||
}
|
||||
|
||||
// The tool definitions sent to the model. Same names and args as the MCP
|
||||
// schemas, trimmed to what the model needs.
|
||||
import { toolSchemas } from './schemas.js';
|
||||
function toolDefs() {
|
||||
return toolSchemas().map(t => ({
|
||||
type: 'function',
|
||||
function: {
|
||||
name: t.name,
|
||||
description: t.description,
|
||||
parameters: t.inputSchema,
|
||||
},
|
||||
}));
|
||||
}
|
||||
80
src/llm/schemas.js
Normal file
80
src/llm/schemas.js
Normal file
@@ -0,0 +1,80 @@
|
||||
// The JSON schemas for the eight tools, in the shape the MCP ListTools call
|
||||
// returns. Keep these short and Dutch-aware: the descriptions are what the
|
||||
// LLM reads to decide which tool to call.
|
||||
|
||||
export function toolSchemas() {
|
||||
return [
|
||||
{
|
||||
name: 'list_profiles',
|
||||
description: 'List the report profiles available, with their id and title.',
|
||||
inputSchema: { type: 'object', properties: {}, required: [] },
|
||||
},
|
||||
{
|
||||
name: 'describe_profile',
|
||||
description: 'Describe one profile: its entities, fields, relations and labels. Inline the values of short code lists.',
|
||||
inputSchema: {
|
||||
type: 'object',
|
||||
properties: { profile_id: { type: 'string', description: 'The profile URI returned by list_profiles' } },
|
||||
required: ['profile_id'],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'validate_spec',
|
||||
description: 'Validate a report spec (Turtle). Returns "ok" or one clear sentence saying what is wrong and what to write instead. No database is touched.',
|
||||
inputSchema: {
|
||||
type: 'object',
|
||||
properties: { spec: { type: 'string', description: 'The spec as Turtle text' } },
|
||||
required: ['spec'],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'lookup_values',
|
||||
description: 'Search a code list (too long to inline) for a term. Returns matches, a total, and an exact flag when a label equals the term.',
|
||||
inputSchema: {
|
||||
type: 'object',
|
||||
properties: {
|
||||
profile_id: { type: 'string' },
|
||||
field: { type: 'string', description: 'The field as "entityLabel.fieldLabel", e.g. "melding.type dossier"' },
|
||||
term: { type: 'string', description: 'The Dutch word or name to search for' },
|
||||
},
|
||||
required: ['profile_id', 'field', 'term'],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'preview_queries',
|
||||
description: 'Show the SPARQL queries the runner would run for a valid spec, as text. Lets you check before running.',
|
||||
inputSchema: {
|
||||
type: 'object',
|
||||
properties: { spec: { type: 'string' } },
|
||||
required: ['spec'],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'run_report',
|
||||
description: 'Run a valid spec and return the report URI, the row count and the file URI. Runs as the caller, so it sees only what they may see. Waits up to RUN_TIMEOUT. The report is named by the dct:title in the spec.',
|
||||
inputSchema: {
|
||||
type: 'object',
|
||||
properties: { spec: { type: 'string' } },
|
||||
required: ['spec'],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'report_status',
|
||||
description: 'Read the status of a job by its URI, and the file once it succeeds.',
|
||||
inputSchema: {
|
||||
type: 'object',
|
||||
properties: { job_uri: { type: 'string' } },
|
||||
required: ['job_uri'],
|
||||
},
|
||||
},
|
||||
{
|
||||
name: 'export_spec',
|
||||
description: 'Return the Turtle spec that produced a report, by report URI.',
|
||||
inputSchema: {
|
||||
type: 'object',
|
||||
properties: { report_uri: { type: 'string' } },
|
||||
required: ['report_uri'],
|
||||
},
|
||||
},
|
||||
];
|
||||
}
|
||||
144
src/llm/tools.js
Normal file
144
src/llm/tools.js
Normal file
@@ -0,0 +1,144 @@
|
||||
// The eight MCP tools the LLM calls. Each maps to a runner/ function or a
|
||||
// pure profile/db read. Tools that touch session data (run_report,
|
||||
// report_status, export_spec) live here and call into runner/; they never
|
||||
// import sessionQuery themselves (CI check 2). What they return is always a
|
||||
// job URI, a status, a count, a link, column names or an error we wrote,
|
||||
// never a cell value (CI check 3, the §8 guarantee).
|
||||
//
|
||||
// The shape every tool returns is { content: [{ type: 'text', text }] } so the
|
||||
// loop can pass it straight back to the model.
|
||||
|
||||
import { checkSpec } from '../runner/check.js';
|
||||
import { parseSpec } from '../runner/spec.js';
|
||||
import { seedPageQuery } from '../runner/seed.js';
|
||||
import { describeProfile, codeListValues } from './describe.js';
|
||||
import { run, slug } from '../runner/run.js';
|
||||
import { lookupValues } from './lookup.js';
|
||||
|
||||
const MAX_PATH_DEPTH = Number(process.env.MAX_PATH_DEPTH || 8);
|
||||
const RUN_TIMEOUT = Number(process.env.RUN_TIMEOUT || 60) * 1000;
|
||||
|
||||
// buildTools(profiles, session) → { tools: Tool[], handlers: Map<name, fn> }
|
||||
// `session` is the request's { query, update, groups, capture } bundle the app
|
||||
// builds once per /mcp request. run_report captures the identity and runs in
|
||||
// the background exactly like POST /reports does, but waits up to
|
||||
// RUN_TIMEOUT so the LLM gets the result in one turn when it is quick.
|
||||
|
||||
export function buildTools(profiles, session) {
|
||||
const profileList = () => [...profiles.values()].map(p => ({ id: p.uri, title: p.title }));
|
||||
|
||||
const findProfile = (id) => profiles.get(id);
|
||||
|
||||
async function describe_profile({ profile_id }) {
|
||||
const p = findProfile(profile_id);
|
||||
if (!p) return text(`no profile <${profile_id}>. Available: ${profileList().map(x => x.title).join(', ')}.`);
|
||||
const inlined = await codeListValues(p, profiles);
|
||||
return text(describeProfile(p, inlined));
|
||||
}
|
||||
|
||||
async function validate_spec({ spec }) {
|
||||
let parsed;
|
||||
try { parsed = parseSpec(spec); }
|
||||
catch (e) { return text(`the spec did not parse: ${e.message}`); }
|
||||
const p = findProfile(parsed.profileUri);
|
||||
if (!p) return text(`no profile <${parsed.profileUri}>. Available: ${profileList().map(x => x.title).join(', ')}.`);
|
||||
const errors = checkSpec(parsed, p, MAX_PATH_DEPTH, profiles);
|
||||
return text(errors.length ? errors[0] : 'ok');
|
||||
}
|
||||
|
||||
async function lookup_values({ profile_id, field, term }) {
|
||||
const p = findProfile(profile_id);
|
||||
if (!p) return text(`no profile <${profile_id}>.`);
|
||||
const res = await lookupValues(p, term, field);
|
||||
return text(JSON.stringify(res));
|
||||
}
|
||||
|
||||
async function preview_queries({ spec }) {
|
||||
let parsed;
|
||||
try { parsed = parseSpec(spec); }
|
||||
catch (e) { return text(`the spec did not parse: ${e.message}`); }
|
||||
const p = findProfile(parsed.profileUri);
|
||||
if (!p) return text(`no profile <${parsed.profileUri}>.`);
|
||||
const errors = checkSpec(parsed, p, MAX_PATH_DEPTH, profiles);
|
||||
if (errors.length) return text(`the spec is not valid: ${errors[0]}`);
|
||||
return text(buildQueriesText(parsed, p));
|
||||
}
|
||||
|
||||
async function list_profiles() {
|
||||
return text(profileList().map(x => `${x.title} <${x.id}>`).join('\n'));
|
||||
}
|
||||
|
||||
async function run_report({ spec }) {
|
||||
let parsed;
|
||||
try { parsed = parseSpec(spec); }
|
||||
catch (e) { return text(`the spec did not parse: ${e.message}`); }
|
||||
const p = findProfile(parsed.profileUri);
|
||||
if (!p) return text(`no profile <${parsed.profileUri}>.`);
|
||||
const errors = checkSpec(parsed, p, MAX_PATH_DEPTH, profiles);
|
||||
if (errors.length) return text(`the spec is not valid: ${errors[0]}`);
|
||||
try {
|
||||
const g = await session.groups;
|
||||
// The spec's own dct:title names the report; there is no other input.
|
||||
const name = parsed.title || 'report';
|
||||
await session.onReportStart?.({ title: name, fileName: `${slug(name)}.csv` });
|
||||
const running = run(g, parsed, p, name);
|
||||
// The end callback fires even when withTimeout below has given up: the
|
||||
// run is not cancelled, and the file arrives when it arrives. The
|
||||
// rejection handler also keeps a late failure from crashing Node 20.
|
||||
running
|
||||
.then((r) => session.onReportEnd?.(null, r), (e) => session.onReportEnd?.(e))
|
||||
.catch((e) => console.error('[reports] end callback failed:', e));
|
||||
const result = await withTimeout(running, RUN_TIMEOUT);
|
||||
return text(`done. report <${result.reportUri}>, ${result.rowCount} rows, file <${result.fileUri}>.`);
|
||||
} catch (e) {
|
||||
return text(`the report failed: ${String(e.message || e).split('\n')[0]}`);
|
||||
}
|
||||
}
|
||||
|
||||
async function report_status({ job_uri }) {
|
||||
const r = await session.query(
|
||||
`SELECT ?status WHERE { <${job_uri}> <http://www.w3.org/ns/adms#status> ?status } LIMIT 1`);
|
||||
const b = r.results.bindings[0];
|
||||
if (!b) return text(`no job <${job_uri}>.`);
|
||||
const status = b.status.value.split('/').pop();
|
||||
let link = '';
|
||||
if (status === 'success') {
|
||||
const f = await session.query(
|
||||
`SELECT ?file WHERE { <${job_uri}> <http://redpencil.data.gift/vocabularies/tasks/resultsContainer> ?c . ?c <http://redpencil.data.gift/vocabularies/tasks/hasFile> ?file } LIMIT 1`);
|
||||
const fb = f.results.bindings[0];
|
||||
if (fb) link = ` file <${fb.file.value}>.`;
|
||||
}
|
||||
return text(`status: ${status}.${link}`);
|
||||
}
|
||||
|
||||
async function export_spec({ report_uri }) {
|
||||
const r = await session.query(
|
||||
`SELECT ?spec WHERE { <${report_uri}> <http://mu.semte.ch/vocabularies/ext/spec> ?spec } LIMIT 1`);
|
||||
const b = r.results.bindings[0];
|
||||
return text(b ? b.spec.value : `no spec stored on <${report_uri}>.`);
|
||||
}
|
||||
|
||||
const handlers = new Map([
|
||||
['list_profiles', list_profiles],
|
||||
['describe_profile', describe_profile],
|
||||
['validate_spec', validate_spec],
|
||||
['lookup_values', lookup_values],
|
||||
['preview_queries', preview_queries],
|
||||
['run_report', run_report],
|
||||
['report_status', report_status],
|
||||
['export_spec', export_spec],
|
||||
]);
|
||||
|
||||
return { handlers };
|
||||
}
|
||||
|
||||
function text(s) { return { content: [{ type: 'text', text: s }] }; }
|
||||
|
||||
function buildQueriesText(spec, profile) {
|
||||
const seed = seedPageQuery(spec, Number(process.env.SEED_PAGE_SIZE || 5000), 0);
|
||||
return ['-- seed query', seed, ''].join('\n');
|
||||
}
|
||||
|
||||
function withTimeout(p, ms) {
|
||||
return Promise.race([p, new Promise((_, rej) => setTimeout(() => rej(new Error(`report took longer than ${ms / 1000}s`)), ms))]);
|
||||
}
|
||||
50
src/report-assistant.js
Normal file
50
src/report-assistant.js
Normal file
@@ -0,0 +1,50 @@
|
||||
// The report assistant: what the chat says when this service answers a turn.
|
||||
// Runs the iteration-1 loop with the conversation so far, and turns a report
|
||||
// run into an interim message with a pending file.
|
||||
import { ask } from './llm/loop.js';
|
||||
|
||||
const TEXT = {
|
||||
making: 'Ik maak het rapport. Het verschijnt hier zodra het klaar is.',
|
||||
failed: 'Het rapport is mislukt. Stel de vraag opnieuw, of anders geformuleerd.',
|
||||
};
|
||||
|
||||
// whenProfiles() resolves once the SHACL profiles are loaded (app.js keeps
|
||||
// the load promise), so the chat mounts before the profiles are in.
|
||||
export function reportAssistant(whenProfiles) {
|
||||
return async function answer(turn) {
|
||||
// the pending file of the current run; the end callback needs the one
|
||||
// the start callback made. One report per turn: a second run_report in
|
||||
// the same turn overwrites this before the first end callback fires.
|
||||
let pendingDocument = null;
|
||||
|
||||
const session = {
|
||||
query: turn.query,
|
||||
groups: turn.groups,
|
||||
async onReportStart({ fileName }) {
|
||||
const said = await turn.say(TEXT.making, [{ name: fileName, mediaType: 'text/csv' }]);
|
||||
pendingDocument = said.documents[0].uri;
|
||||
},
|
||||
async onReportEnd(error, result) {
|
||||
const documentUri = pendingDocument;
|
||||
pendingDocument = null;
|
||||
if (!documentUri) return;
|
||||
if (error) {
|
||||
await turn.drop(documentUri);
|
||||
if (turn.finished) await turn.say(TEXT.failed);
|
||||
return;
|
||||
}
|
||||
await turn.setUrl(documentUri, downloadHref(result.fileUri));
|
||||
},
|
||||
};
|
||||
|
||||
const profiles = await whenProfiles();
|
||||
const { text } = await ask([...turn.history, { role: 'user', content: turn.content }], profiles, session);
|
||||
return text;
|
||||
};
|
||||
}
|
||||
|
||||
// The file service serves /files/<uuid>/download, and the logical file URI
|
||||
// that registerFile returns ends in that uuid.
|
||||
function downloadHref(fileUri) {
|
||||
return `/files/${fileUri.split('/').pop()}/download`;
|
||||
}
|
||||
83
src/runner/assemble.js
Normal file
83
src/runner/assemble.js
Normal file
@@ -0,0 +1,83 @@
|
||||
// Values to rows. Subjects in stable order, dedup on term, collect per cell.
|
||||
// rep:self columns take the subject URI; a column with no values is an
|
||||
// empty cell, never a dropped row. rep:row expands one subject into several
|
||||
// lines; at most one column per spec may do that (checked in check.js).
|
||||
|
||||
const SH = 'http://www.w3.org/ns/shacl#';
|
||||
const REP = 'http://mu.semte.ch/vocabularies/reporting/';
|
||||
const XSD = 'http://www.w3.org/2001/XMLSchema#';
|
||||
|
||||
export const NUMERIC_DATATYPES = new Set([
|
||||
'integer', 'decimal', 'double', 'float', 'long', 'int', 'short', 'byte',
|
||||
'nonNegativeInteger', 'positiveInteger', 'nonPositiveInteger', 'negativeInteger',
|
||||
'unsignedLong', 'unsignedInt', 'unsignedShort', 'unsignedByte',
|
||||
].map(d => XSD + d));
|
||||
export const TEMPORAL_DATATYPES = new Set([XSD + 'date', XSD + 'dateTime']);
|
||||
|
||||
export function assemble(subjects, values, spec) {
|
||||
const sorted = [...subjects].sort();
|
||||
const rows = [spec.columns.map(c => c.label)];
|
||||
for (const subject of sorted) {
|
||||
const perColumn = values.get(subject) || new Map();
|
||||
const cells = spec.columns.map((col, i) => {
|
||||
if (!col.path.length) return [subject]; // rep:self
|
||||
const list = dedup(perColumn.get(i) || []);
|
||||
return collect(list, col);
|
||||
});
|
||||
const lines = Math.max(1, ...cells.map(c => c.length));
|
||||
for (let r = 0; r < lines; r++) {
|
||||
// every cell except the expanding one keeps its value on each line
|
||||
rows.push(cells.map(c => c[Math.min(r, c.length - 1)]));
|
||||
}
|
||||
}
|
||||
return rows;
|
||||
}
|
||||
|
||||
function collect(terms, col) {
|
||||
switch (col.collect) {
|
||||
case SH + 'groupConcat':
|
||||
case null:
|
||||
case undefined:
|
||||
return [terms.map(t => t.value).join(col.separator ?? ',')];
|
||||
case SH + 'min':
|
||||
return [extreme(terms, col, -1)];
|
||||
case SH + 'max':
|
||||
return [extreme(terms, col, 1)];
|
||||
case REP + 'row':
|
||||
return terms.length ? terms.map(t => t.value) : [''];
|
||||
default:
|
||||
throw new Error(`column "${col.label}" has an unknown rep:collect <${col.collect}>. Use sh:groupConcat, rep:row, sh:min or sh:max.`);
|
||||
}
|
||||
}
|
||||
|
||||
// min/max over values that share one datatype. Numbers compare as numbers,
|
||||
// dates and times as instants. Mixed datatypes throw: check.js should have
|
||||
// caught it, and a silently wrong cell is worse than a failed job.
|
||||
function extreme(terms, col, direction) {
|
||||
if (!terms.length) return '';
|
||||
const datatypes = [...new Set(terms.map(t => t.datatype?.value || ''))];
|
||||
if (datatypes.length > 1) {
|
||||
throw new Error(`column "${col.label}" asks for sh:${direction < 0 ? 'min' : 'max'} but its values carry ${datatypes.length} datatypes (${datatypes.join(', ')}).`);
|
||||
}
|
||||
const dt = datatypes[0];
|
||||
const key = NUMERIC_DATATYPES.has(dt) ? t => Number(t.value)
|
||||
: TEMPORAL_DATATYPES.has(dt) ? t => new Date(t.value).getTime()
|
||||
: t => t.value;
|
||||
const cmp = (a, b) => a < b ? -1 : a > b ? 1 : 0;
|
||||
const best = terms.reduce((a, b) => (cmp(key(b), key(a)) * direction > 0 ? b : a));
|
||||
return best.value;
|
||||
}
|
||||
|
||||
// Two terms are the same when the URIs match, or lexical form, datatype
|
||||
// and language tag all match.
|
||||
export function dedup(terms) {
|
||||
const seen = new Set();
|
||||
const out = [];
|
||||
for (const t of terms) {
|
||||
const key = t.termType === 'NamedNode'
|
||||
? `u:${t.value}`
|
||||
: `l:${t.value}|${t.datatype?.value || ''}|${t.language || ''}`;
|
||||
if (!seen.has(key)) { seen.add(key); out.push(t); }
|
||||
}
|
||||
return out;
|
||||
}
|
||||
311
src/runner/check.js
Normal file
311
src/runner/check.js
Normal file
@@ -0,0 +1,311 @@
|
||||
import { entityForClass, fieldsOf } from './profile.js';
|
||||
import { NUMERIC_DATATYPES } from './assemble.js';
|
||||
|
||||
const SH = 'http://www.w3.org/ns/shacl#';
|
||||
const REP = 'http://mu.semte.ch/vocabularies/reporting/';
|
||||
const XSD = 'http://www.w3.org/2001/XMLSchema#';
|
||||
const XSD_DATE = XSD + 'date';
|
||||
const XSD_DATETIME = XSD + 'dateTime';
|
||||
|
||||
// The validator. checkSpec(spec, profile, maxPathDepth) runs the named checks
|
||||
// in order and returns the FIRST error as a one-element array (empty if ok).
|
||||
// Every message names the thing that is wrong in the profile's own words and
|
||||
// says what to write instead. No database call anywhere in this file — that is
|
||||
// what lets an LLM iterate for free.
|
||||
|
||||
const ANYOF_MAX_TERMS = 24;
|
||||
const ANYOF_MAX_TERM_LENGTH = 100;
|
||||
|
||||
// 1 + 2. the profile names itself and is one we have. knownProfiles is the
|
||||
// Map app.js loaded at boot; omit it and the check is skipped (fixtures).
|
||||
function profileNamed(spec, knownProfiles) {
|
||||
if (!spec.profileUri) {
|
||||
return `no rep:profile named. Available: ${knownProfiles ? [...knownProfiles.values()].map(p => p.title).join(', ') : '(none)'} .`;
|
||||
}
|
||||
if (knownProfiles && !knownProfiles.has(spec.profileUri)) {
|
||||
return `no profile <${spec.profileUri}>. Available: ${knownProfiles ? [...knownProfiles.values()].map(p => p.title).join(', ') : '(none)'}.`;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 3 + 4. the target class is in the profile
|
||||
function targetClass(spec, profile) {
|
||||
if (!spec.targetClass) {
|
||||
return `no sh:targetClass. In "${profile.title}" you can list: ${shapeList(profile)}.`;
|
||||
}
|
||||
if (!entityForClass(profile, spec.targetClass)) {
|
||||
return `"${profile.title}" cannot list <${spec.targetClass}>. It has: ${shapeList(profile)}.`;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function shapeList(profile) {
|
||||
return profile.shapes.map(s => s.label || s.targetClass).join(', ');
|
||||
}
|
||||
|
||||
// 5. columns exist
|
||||
function columnsPresent(spec) {
|
||||
if (!spec.columns.length) return 'no columns. Add at least one rep:columns entry.';
|
||||
return null;
|
||||
}
|
||||
|
||||
// 6. every column has a label
|
||||
function columnLabel(spec) {
|
||||
for (const col of spec.columns) {
|
||||
if (!col.label) return 'a column has no rdfs:label.';
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 7. labels unique
|
||||
function columnLabelsUnique(spec) {
|
||||
const seen = new Map();
|
||||
for (const col of spec.columns) {
|
||||
if (seen.has(col.label)) {
|
||||
return `two columns are called "${col.label}". Labels must differ.`;
|
||||
}
|
||||
seen.set(col.label, true);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 8 + 9 + 10. every path resolves, columns end on a value, depth capped
|
||||
function pathsResolve(spec, profile, maxPathDepth) {
|
||||
const startShapeUri = entityForClass(profile, spec.targetClass);
|
||||
for (const col of spec.columns) {
|
||||
const err = walk(profile, startShapeUri, col.path, maxPathDepth,
|
||||
`column "${col.label}"`, true);
|
||||
if (err) return err;
|
||||
}
|
||||
for (const [i, filter] of spec.filters.entries()) {
|
||||
const err = walk(profile, startShapeUri, filter.path, maxPathDepth,
|
||||
`filter ${i + 1}`, false);
|
||||
if (err) return err;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 11. sh:min/sh:max only on numbers and dates
|
||||
function collectMinMaxTyped(spec, profile) {
|
||||
const startShapeUri = entityForClass(profile, spec.targetClass);
|
||||
for (const col of spec.columns) {
|
||||
if (col.collect !== SH + 'min' && col.collect !== SH + 'max') continue;
|
||||
const name = col.collect === SH + 'min' ? 'sh:min' : 'sh:max';
|
||||
if (!col.path.length) return `${name} on a rep:self column does not apply. A subject URI is not a number or date.`;
|
||||
const last = col.path[col.path.length - 1];
|
||||
const shapeUri = shapeAfter(profile, startShapeUri, col.path);
|
||||
if (!shapeUri) continue; // pathsResolve already reported the real error
|
||||
const f = fieldsOf(profile, shapeUri)
|
||||
.find(x => x.path === last.predicate && !!x.inverse === !!last.inverse);
|
||||
if (!f) continue;
|
||||
const kind = f.datatype === XSD_DATETIME || f.datatype === XSD_DATE ? 'date'
|
||||
: NUMERIC_DATATYPES.has(f.datatype) ? 'number'
|
||||
: null;
|
||||
if (!kind) {
|
||||
return `"${f.name || last.predicate}" is text, so ${name} does not apply. Use rep:row, or leave it out to join the values.`;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 13. at most one rep:row column
|
||||
function oneRowColumn(spec) {
|
||||
const rowCols = spec.columns.filter(c => c.collect === REP + 'row');
|
||||
if (rowCols.length > 1) {
|
||||
const names = rowCols.map(c => `"${c.label}"`).join(' and ');
|
||||
return `only one column may use rep:row. ${names} both do.`;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 12 + 14. columns carry no constraints; known collect mode
|
||||
function columnPure(spec) {
|
||||
for (const col of spec.columns) {
|
||||
if (Object.keys(col.constraints || {}).length) {
|
||||
const which = Object.keys(col.constraints)[0];
|
||||
return `a column carries ${which}. Constraints belong in sh:property.`;
|
||||
}
|
||||
if (col.collect && ![SH + 'groupConcat', SH + 'min', SH + 'max', REP + 'row'].includes(col.collect)) {
|
||||
return `column "${col.label}" has an unknown rep:collect <${col.collect}>. Use sh:groupConcat (default), rep:row, sh:min or sh:max.`;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 15. no label inside a filter
|
||||
function filterLabel(spec) {
|
||||
for (const filter of spec.filters) {
|
||||
if ((filter.constraints || {}).label) {
|
||||
return 'a filter carries rdfs:label. Move it to rep:columns to show it.';
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 15b. a filter with a path but no constraint selects everything; the model
|
||||
// almost certainly meant to filter. Refuse and say what to add.
|
||||
function filterConstraint(spec) {
|
||||
const MEANINGFUL = ['minCount', 'maxCount', 'hasValue', 'in', 'minInclusive',
|
||||
'maxInclusive', 'minExclusive', 'maxExclusive', 'pattern', 'flags', 'anyOf'];
|
||||
for (const [i, filter] of spec.filters.entries()) {
|
||||
const c = filter.constraints || {};
|
||||
if (!MEANINGFUL.some(k => c[k] !== undefined && c[k] !== null)) {
|
||||
const p = filter.path.map(h => h.predicate).join(' → ');
|
||||
return `filter ${i + 1} (path ${p}) has no condition. Add one: sh:hasValue for an exact value, sh:in for a list, rep:anyOf for words, or remove the filter.`;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 16. sh:maxCount above 0 needs counting
|
||||
function maxCountZero(spec) {
|
||||
for (const [i, filter] of spec.filters.entries()) {
|
||||
const c = filter.constraints;
|
||||
if (c.maxCount !== null && c.maxCount !== undefined && c.maxCount > 0) {
|
||||
return `sh:maxCount ${c.maxCount} needs counting, which this service cannot do. Use sh:maxCount 0 for "has none", or drop it.`;
|
||||
}
|
||||
if (c.minCount !== null && c.minCount !== undefined && c.minCount > 2) {
|
||||
return `sh:minCount ${c.minCount} needs counting. Use sh:minCount 1 (at least one) or sh:minCount 2 (two distinct values); anything higher is beyond this service.`;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 17 + 18. rep:anyOf size limits
|
||||
function anyOfLimits(spec) {
|
||||
for (const [i, filter] of spec.filters.entries()) {
|
||||
const c = filter.constraints;
|
||||
if (!c.anyOf) continue;
|
||||
if (c.anyOf.length > ANYOF_MAX_TERMS) {
|
||||
return `rep:anyOf has ${c.anyOf.length} terms. The limit is ${ANYOF_MAX_TERMS}. Use sh:in with exact URIs instead (find them with lookup_values).`;
|
||||
}
|
||||
for (const term of c.anyOf) {
|
||||
if (term.type === 'uri') {
|
||||
return 'a rep:anyOf term is a URI. rep:anyOf takes Dutch words, not URIs. Use sh:in with the exact URI instead.';
|
||||
}
|
||||
if (term.value && term.value.length > ANYOF_MAX_TERM_LENGTH) {
|
||||
return `a rep:anyOf term is ${term.value.length} characters. The limit is ${ANYOF_MAX_TERM_LENGTH}.`;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 19. sh:in mixes URIs and literals
|
||||
function inHomogeneous(spec) {
|
||||
for (const filter of spec.filters) {
|
||||
const c = filter.constraints;
|
||||
if (!c.in) continue;
|
||||
const uris = c.in.filter(t => t.type === 'uri').length;
|
||||
if (uris > 0 && uris < c.in.length) {
|
||||
return 'sh:in mixes URIs and text. Pick one.';
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
// 20. typed literals that will not parse
|
||||
function literalsParse(spec) {
|
||||
const constraintTerms = [];
|
||||
for (const filter of spec.filters) {
|
||||
const c = filter.constraints;
|
||||
for (const key of ['hasValue', 'minInclusive', 'maxInclusive', 'minExclusive', 'maxExclusive']) {
|
||||
if (c[key]) constraintTerms.push(c[key]);
|
||||
}
|
||||
if (c.in) constraintTerms.push(...c.in);
|
||||
}
|
||||
for (const t of constraintTerms) {
|
||||
if (t.type !== 'literal' || !t.datatype) continue;
|
||||
if (t.datatype === XSD_DATETIME && !parsesAsDate(t.value, true)) {
|
||||
return `"${t.value}" is not a valid xsd:dateTime.`;
|
||||
}
|
||||
if (t.datatype === XSD_DATE && !parsesAsDate(t.value, false)) {
|
||||
return `"${t.value}" is not a valid xsd:date.`;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function parsesAsDate(value, withTime) {
|
||||
const d = new Date(value);
|
||||
if (Number.isNaN(d.getTime())) return false;
|
||||
return withTime ? value.includes('T') : /^\d{4}-\d{2}-\d{2}$/.test(value);
|
||||
}
|
||||
|
||||
const CHECKS = [
|
||||
['no sh:targetClass or unknown', targetClass],
|
||||
['no columns', columnsPresent],
|
||||
['column without rdfs:label', columnLabel],
|
||||
['two columns with one label', columnLabelsUnique],
|
||||
['column carries constraints', columnPure],
|
||||
['filter carries rdfs:label', filterLabel],
|
||||
['filter without a condition', filterConstraint],
|
||||
['sh:maxCount above 0', maxCountZero],
|
||||
['rep:anyOf too big', anyOfLimits],
|
||||
['sh:in mixes URIs and literals', inHomogeneous],
|
||||
['typed literal does not parse', literalsParse],
|
||||
['two rep:row columns', oneRowColumn],
|
||||
['path does not resolve', pathsResolve],
|
||||
['sh:min/sh:max on a non-number', collectMinMaxTyped],
|
||||
];
|
||||
|
||||
export function checkSpec(spec, profile, maxPathDepth, knownProfiles = null) {
|
||||
const profileErr = profileNamed(spec, knownProfiles);
|
||||
if (profileErr) return [profileErr];
|
||||
for (const [name, check] of CHECKS) {
|
||||
const err = check(spec, profile, maxPathDepth);
|
||||
if (err) return [err];
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
// The shape a column's path lands in: follow sh:node through every hop but
|
||||
// the last. walk() has already validated the path, so any gap means the walk
|
||||
// reported the real error and this can bail.
|
||||
function shapeAfter(profile, startShapeUri, hops) {
|
||||
let shapeUri = startShapeUri;
|
||||
for (let i = 0; i < hops.length - 1; i++) {
|
||||
const fields = fieldsOf(profile, shapeUri);
|
||||
const f = fields.find(x => x.path === hops[i].predicate && !!x.inverse === !!hops[i].inverse);
|
||||
if (!f || !f.node) return null;
|
||||
shapeUri = f.node;
|
||||
}
|
||||
return shapeUri;
|
||||
}
|
||||
|
||||
function walk(profile, startShapeUri, hops, maxPathDepth, what, isColumn) {
|
||||
if (hops.length > maxPathDepth) {
|
||||
return `${what} walks ${hops.length} hops. The limit is ${maxPathDepth}.`;
|
||||
}
|
||||
if (!hops.length) return null;
|
||||
let shapeUri = startShapeUri;
|
||||
for (let i = 0; i < hops.length; i++) {
|
||||
const hop = hops[i];
|
||||
const fields = fieldsOf(profile, shapeUri);
|
||||
const f = fields.find(x => x.path === hop.predicate && !!x.inverse === !!hop.inverse);
|
||||
if (!f) {
|
||||
const have = fields.map(x => (x.inverse ? `geen ${x.name} (${x.path}, omgekeerd)` : `${x.name} (${x.path})`)).join(', ');
|
||||
const shapeLabel = profile.shapes.find(s => s.uri === shapeUri)?.label || shapeUri;
|
||||
return `at "${shapeLabel}" there is no "${hop.predicate}". It has: ${have || '(nothing)'}.`;
|
||||
}
|
||||
if (i < hops.length - 1) {
|
||||
if (!f.node) {
|
||||
return `${what} hop ${i + 1} <${hop.predicate}> holds a value, not a link. You cannot walk past it.`;
|
||||
}
|
||||
shapeUri = f.node;
|
||||
} else if (isColumn) {
|
||||
if (!f.datatype && !f.class) {
|
||||
const valueFields = fieldsOf(profile, f.node)
|
||||
.filter(x => x.datatype || x.class)
|
||||
.map(x => `${x.name} (${x.path})`);
|
||||
return `column "${colLabel(what)}" ends on a link. Add a hop: ${valueFields.join(' or ')}.`;
|
||||
}
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function colLabel(what) {
|
||||
const m = what.match(/"([^"]+)"/);
|
||||
return m ? m[1] : what;
|
||||
}
|
||||
67
src/runner/columns.js
Normal file
67
src/runner/columns.js
Normal file
@@ -0,0 +1,67 @@
|
||||
import { sparqlEscapeUri } from '../db.js';
|
||||
|
||||
// Fetch column values for known subjects. Columns are grouped by their
|
||||
// shared non-leaf hops; two columns share a query only when every hop
|
||||
// before the leaf is identical, and the leaf is read the same way.
|
||||
// rep:self columns (empty path) are filled by assemble.js.
|
||||
|
||||
const SUBJECT_CHUNK_SIZE = Number(process.env.SUBJECT_CHUNK_SIZE || 250);
|
||||
|
||||
export async function fetchColumns(sessionQuery, subjects, spec, values = new Map(), onChunk = null) {
|
||||
const groups = groupColumns(spec.columns);
|
||||
|
||||
for (const group of groups) {
|
||||
for (let i = 0; i < subjects.length; i += SUBJECT_CHUNK_SIZE) {
|
||||
const chunk = subjects.slice(i, i + SUBJECT_CHUNK_SIZE);
|
||||
const result = await sessionQuery(groupQuery(group, chunk));
|
||||
collect(result, group, values);
|
||||
}
|
||||
if (onChunk) await onChunk(group);
|
||||
}
|
||||
return values;
|
||||
}
|
||||
|
||||
// Group columns by their shared prefix (every hop before the leaf) and the
|
||||
// direction of the leaf. A column with an empty path (rep:self) joins no group.
|
||||
export function groupColumns(columns) {
|
||||
const groups = new Map();
|
||||
columns.forEach((col, index) => {
|
||||
if (!col.path.length) return; // rep:self
|
||||
const leaf = col.path[col.path.length - 1];
|
||||
const key = col.path.map(h => `${h.inverse ? '^' : ''}${h.predicate}`).join(' > ');
|
||||
if (!groups.has(key)) {
|
||||
groups.set(key, { hops: col.path, columns: [] });
|
||||
}
|
||||
groups.get(key).columns.push({ index, leaf: leaf.predicate });
|
||||
});
|
||||
return [...groups.values()];
|
||||
}
|
||||
|
||||
export function groupQuery(group, chunk) {
|
||||
const subjects = chunk.map(sparqlEscapeUri).join(' ');
|
||||
const lines = [];
|
||||
let prev = '?s';
|
||||
// every hop except the last gets its own variable; the leaf is ?v
|
||||
group.hops.slice(0, -1).forEach((hop, i) => {
|
||||
const v = `?x${i}`;
|
||||
if (hop.inverse) lines.push(`${v} ${sparqlEscapeUri(hop.predicate)} ${prev} .`);
|
||||
else lines.push(`${prev} ${sparqlEscapeUri(hop.predicate)} ${v} .`);
|
||||
prev = v;
|
||||
});
|
||||
const leaf = group.hops[group.hops.length - 1];
|
||||
if (leaf.inverse) lines.push(`?v ${sparqlEscapeUri(leaf.predicate)} ${prev} .`);
|
||||
else lines.push(`${prev} ${sparqlEscapeUri(leaf.predicate)} ?v .`);
|
||||
return `SELECT ?s ?v WHERE {\n VALUES ?s { ${subjects} }\n ${lines.join('\n ')}\n}`;
|
||||
}
|
||||
|
||||
function collect(result, group, values) {
|
||||
for (const b of result.results.bindings) {
|
||||
let perColumn = values.get(b.s.value);
|
||||
if (!perColumn) { perColumn = new Map(); values.set(b.s.value, perColumn); }
|
||||
for (const col of group.columns) {
|
||||
let list = perColumn.get(col.index);
|
||||
if (!list) { list = []; perColumn.set(col.index, list); }
|
||||
list.push(b.v);
|
||||
}
|
||||
}
|
||||
}
|
||||
27
src/runner/csv.js
Normal file
27
src/runner/csv.js
Normal file
@@ -0,0 +1,27 @@
|
||||
import { mkdirSync, writeFileSync } from 'fs';
|
||||
|
||||
// Rows to a CSV file. CSV_SEPARATOR is ';' by default.
|
||||
// A cell is quoted only when it holds the separator, a quote or a newline —
|
||||
// this matches the shape of today's loket-report-generation-service output.
|
||||
// A quote inside a cell is doubled, per RFC 4180.
|
||||
|
||||
const CSV_SEPARATOR = process.env.CSV_SEPARATOR || ';';
|
||||
const SHARE_DIR = process.env.SHARE_DIR || '/share';
|
||||
|
||||
export function rowsToCsv(rows) {
|
||||
return rows.map(row => row.map(escapeCell).join(CSV_SEPARATOR)).join('\n');
|
||||
}
|
||||
|
||||
function escapeCell(cell) {
|
||||
const s = String(cell ?? '');
|
||||
return s.includes(CSV_SEPARATOR) || s.includes('"') || /[\n\r]/.test(s)
|
||||
? `"${s.replace(/"/g, '""')}"`
|
||||
: s;
|
||||
}
|
||||
|
||||
export function writeCsv(name, csv) {
|
||||
mkdirSync(SHARE_DIR, { recursive: true });
|
||||
const file = `${SHARE_DIR}/${name}`;
|
||||
writeFileSync(file, csv, 'utf8');
|
||||
return file;
|
||||
}
|
||||
158
src/runner/job.js
Normal file
158
src/runner/job.js
Normal file
@@ -0,0 +1,158 @@
|
||||
import { randomUUID } from 'crypto';
|
||||
import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime } from '../db.js';
|
||||
|
||||
// cogs:Job + task:Task, written by us. No other service is involved.
|
||||
// Triples follow §6 of plan-opus-5.iteration-1-4.md; the shape follows
|
||||
// config/resources/master-job-domain.lisp.
|
||||
// Every write carries mu-call-scope-id (REPORT_SCOPE) so the delta notifier
|
||||
// lets jobs-controller ignore our statuses — it throws on unknown operations.
|
||||
// All functions take the update function to use: scopedUpdate(sparql, extra),
|
||||
// which sends the captured groups plus this scope header.
|
||||
|
||||
const REPORT_SCOPE = 'http://redpencil.data.gift/id/concept/muScope/report-generation';
|
||||
const JOB_BASE = process.env.JOB_URI_BASE || 'http://redpencil.data.gift/id/job/';
|
||||
const TASK_BASE = process.env.TASK_URI_BASE || 'http://redpencil.data.gift/id/task/';
|
||||
const STATUS = {
|
||||
scheduled: 'http://redpencil.data.gift/id/concept/JobStatus/scheduled',
|
||||
busy: 'http://redpencil.data.gift/id/concept/JobStatus/busy',
|
||||
success: 'http://redpencil.data.gift/id/concept/JobStatus/success',
|
||||
failed: 'http://redpencil.data.gift/id/concept/JobStatus/failed',
|
||||
};
|
||||
const JOB_OPERATION = 'http://lblod.data.gift/id/jobs/concept/JobOperation/reportGeneration';
|
||||
const TASK_OPERATION = 'http://lblod.data.gift/id/jobs/concept/TaskOperation/reportGeneration';
|
||||
|
||||
const COGS_JOB = 'http://vocab.deri.ie/cogs#Job';
|
||||
const TASK_TASK = 'http://redpencil.data.gift/vocabularies/tasks/Task';
|
||||
const DATA_CONTAINER = 'http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#DataContainer';
|
||||
|
||||
const MU_UUID = 'http://mu.semte.ch/vocabularies/core/uuid';
|
||||
const ADMS_STATUS = 'http://www.w3.org/ns/adms#status';
|
||||
const DCT = 'http://purl.org/dc/terms/';
|
||||
const TASK = 'http://redpencil.data.gift/vocabularies/tasks/';
|
||||
const OSLC_MESSAGE = 'http://open-services.net/ns/core#message';
|
||||
const OSLC_ERROR_URI = 'http://open-services.net/ns/core#Error';
|
||||
|
||||
// updateFn(sparql, extraHeaders) — app.js passes scopedUpdate, which sends
|
||||
// the captured groups; the scope header is added here so every job write is
|
||||
// opt-out-able for the delta notifier.
|
||||
function withScope(updateFn) {
|
||||
return async sparql => updateFn(sparql, { 'mu-call-scope-id': REPORT_SCOPE });
|
||||
}
|
||||
|
||||
// Create job + task + result container, all at status scheduled.
|
||||
// Returns { jobUri, taskUri, containerUri }.
|
||||
export async function createJob(updateFn, title, question, creator) {
|
||||
const upd = withScope(updateFn);
|
||||
const jobId = randomUUID();
|
||||
const taskId = randomUUID();
|
||||
const containerId = randomUUID();
|
||||
const jobUri = `${JOB_BASE}${jobId}`;
|
||||
const taskUri = `${TASK_BASE}${taskId}`;
|
||||
const containerUri = `http://redpencil.data.gift/id/dataContainers/${containerId}`;
|
||||
const now = sparqlEscapeDateTime(new Date());
|
||||
|
||||
const creatorTriple = creator
|
||||
? `${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(DCT + 'creator')} ${sparqlEscapeUri(creator)} .`
|
||||
: '';
|
||||
const questionTriple = question
|
||||
? `${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri('http://mu.semte.ch/vocabularies/ext/question')} ${sparqlEscapeString(question)} .`
|
||||
: '';
|
||||
|
||||
await upd(`
|
||||
INSERT DATA {
|
||||
${sparqlEscapeUri(jobUri)} a ${sparqlEscapeUri(COGS_JOB)} ;
|
||||
${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(jobId)} ;
|
||||
${sparqlEscapeUri(DCT + 'title')} ${sparqlEscapeString(title)} ;
|
||||
${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.scheduled)} ;
|
||||
${sparqlEscapeUri(TASK + 'operation')} ${sparqlEscapeUri(JOB_OPERATION)} ;
|
||||
${sparqlEscapeUri(DCT + 'created')} ${now} ;
|
||||
${sparqlEscapeUri(DCT + 'modified')} ${now} .
|
||||
${creatorTriple}
|
||||
${questionTriple}
|
||||
${sparqlEscapeUri(taskUri)} a ${sparqlEscapeUri(TASK_TASK)} ;
|
||||
${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(taskId)} ;
|
||||
${sparqlEscapeUri(DCT + 'isPartOf')} ${sparqlEscapeUri(jobUri)} ;
|
||||
${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.scheduled)} ;
|
||||
${sparqlEscapeUri(TASK + 'operation')} ${sparqlEscapeUri(TASK_OPERATION)} ;
|
||||
${sparqlEscapeUri(TASK + 'index')} ${sparqlEscapeString('0')} ;
|
||||
${sparqlEscapeUri(DCT + 'created')} ${now} ;
|
||||
${sparqlEscapeUri(DCT + 'modified')} ${now} ;
|
||||
${sparqlEscapeUri(TASK + 'resultsContainer')} ${sparqlEscapeUri(containerUri)} .
|
||||
${sparqlEscapeUri(containerUri)} a ${sparqlEscapeUri(DATA_CONTAINER)} ;
|
||||
${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(containerId)} .
|
||||
}`);
|
||||
|
||||
return { jobUri, taskUri, containerUri };
|
||||
}
|
||||
|
||||
export async function setStatus(updateFn, uri, status) {
|
||||
const s = STATUS[status];
|
||||
if (!s) throw new Error(`unknown job status "${status}"`);
|
||||
const now = sparqlEscapeDateTime(new Date());
|
||||
await withScope(updateFn)(`
|
||||
DELETE {
|
||||
${sparqlEscapeUri(uri)} ${sparqlEscapeUri(ADMS_STATUS)} ?old ;
|
||||
${sparqlEscapeUri(DCT + 'modified')} ?modified .
|
||||
}
|
||||
INSERT {
|
||||
${sparqlEscapeUri(uri)} ${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(s)} ;
|
||||
${sparqlEscapeUri(DCT + 'modified')} ${now} .
|
||||
}
|
||||
WHERE {
|
||||
${sparqlEscapeUri(uri)} ${sparqlEscapeUri(ADMS_STATUS)} ?old .
|
||||
OPTIONAL { ${sparqlEscapeUri(uri)} ${sparqlEscapeUri(DCT + 'modified')} ?modified . }
|
||||
}`);
|
||||
}
|
||||
|
||||
export async function touch(updateFn, taskUri) {
|
||||
const now = sparqlEscapeDateTime(new Date());
|
||||
await withScope(updateFn)(`
|
||||
DELETE {
|
||||
${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(DCT + 'modified')} ?modified .
|
||||
}
|
||||
INSERT {
|
||||
${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(DCT + 'modified')} ${now} .
|
||||
}
|
||||
WHERE {
|
||||
OPTIONAL { ${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(DCT + 'modified')} ?modified . }
|
||||
}`);
|
||||
}
|
||||
|
||||
// Link the finished CSV to the task's result container. The container is
|
||||
// created with the job; the file URI is the logical file from registerFile.
|
||||
export async function attachResult(updateFn, taskUri, fileUri) {
|
||||
await withScope(updateFn)(`
|
||||
INSERT {
|
||||
?container ${sparqlEscapeUri(TASK + 'hasFile')} ${sparqlEscapeUri(fileUri)} .
|
||||
}
|
||||
WHERE {
|
||||
${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(TASK + 'resultsContainer')} ?container .
|
||||
}`);
|
||||
}
|
||||
|
||||
// An oslc:Error with the message, job and task both failed.
|
||||
export async function fail(updateFn, jobUri, taskUri, message) {
|
||||
const errorId = randomUUID();
|
||||
const errorUri = `http://redpencil.data.gift/id/jobs/error/${errorId}`;
|
||||
const now = sparqlEscapeDateTime(new Date());
|
||||
await withScope(updateFn)(`
|
||||
DELETE {
|
||||
${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?js .
|
||||
${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?ts .
|
||||
}
|
||||
INSERT DATA {
|
||||
${sparqlEscapeUri(errorUri)} a ${sparqlEscapeUri(OSLC_ERROR_URI)} ;
|
||||
${sparqlEscapeUri(MU_UUID)} ${sparqlEscapeString(errorId)} ;
|
||||
${sparqlEscapeUri(OSLC_MESSAGE)} ${sparqlEscapeString(message)} ;
|
||||
${sparqlEscapeUri(DCT + 'created')} ${now} .
|
||||
${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(TASK + 'error')} ${sparqlEscapeUri(errorUri)} ;
|
||||
${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.failed)} ;
|
||||
${sparqlEscapeUri(DCT + 'modified')} ${now} .
|
||||
${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(ADMS_STATUS)} ${sparqlEscapeUri(STATUS.failed)} ;
|
||||
${sparqlEscapeUri(DCT + 'modified')} ${now} .
|
||||
}
|
||||
WHERE {
|
||||
${sparqlEscapeUri(jobUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?js .
|
||||
${sparqlEscapeUri(taskUri)} ${sparqlEscapeUri(ADMS_STATUS)} ?ts .
|
||||
}`);
|
||||
}
|
||||
106
src/runner/profile.js
Normal file
106
src/runner/profile.js
Normal file
@@ -0,0 +1,106 @@
|
||||
import N3 from 'n3';
|
||||
import { readFileSync } from 'fs';
|
||||
|
||||
// Parse a profile (a SHACL shapes graph) and answer questions about it.
|
||||
|
||||
const SH = 'http://www.w3.org/ns/shacl#';
|
||||
const RDF = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#';
|
||||
const OWL_ONTOLOGY = 'http://www.w3.org/2002/07/owl#Ontology';
|
||||
|
||||
export async function loadProfiles(dir) {
|
||||
// eslint-disable-next-line no-undef
|
||||
const fs = await import('fs');
|
||||
// eslint-disable-next-line no-undef
|
||||
const path = await import('path');
|
||||
const files = fs.readdirSync(dir).filter(f => f.endsWith('.ttl')).sort();
|
||||
const profiles = new Map();
|
||||
for (const file of files) {
|
||||
const profile = await loadProfile(path.join(dir, file));
|
||||
profiles.set(profile.uri, profile);
|
||||
console.log(`[profile] "${profile.title}" (${file}): ${profile.shapes.length} entities, ` +
|
||||
`${profile.shapes.reduce((n, s) => n + s.fields.length, 0)} fields`);
|
||||
}
|
||||
if (!profiles.size) throw new Error(`no profiles found in ${dir}`);
|
||||
return profiles;
|
||||
}
|
||||
|
||||
export async function loadProfile(file) {
|
||||
const parser = new N3.Parser();
|
||||
const store = new N3.Store();
|
||||
await new Promise((resolve, reject) => {
|
||||
parser.parse(readFileSync(file, 'utf8'), (err, quad) => {
|
||||
if (err) return reject(new Error(`${file}: ${err.message}`));
|
||||
if (quad) store.addQuad(quad); else resolve();
|
||||
});
|
||||
});
|
||||
|
||||
const ontology = store.getQuads(null, RDF + 'type', OWL_ONTOLOGY)[0];
|
||||
if (!ontology) throw new Error(`${file}: no owl:Ontology. A profile file must name itself.`);
|
||||
const uri = ontology.subject.value;
|
||||
const title = store.getQuads(ontology.subject, 'http://purl.org/dc/terms/title', null)[0]?.object.value || uri;
|
||||
|
||||
const prefixQuads = store.getQuads(null, SH + 'declare', null);
|
||||
const prefixes = {};
|
||||
for (const q of prefixQuads) {
|
||||
const p = store.getQuads(q.object, SH + 'prefix', null)[0]?.object.value;
|
||||
const ns = store.getQuads(q.object, SH + 'namespace', null)[0]?.object.value;
|
||||
if (p && ns) prefixes[p] = ns;
|
||||
}
|
||||
|
||||
const shapes = store.getQuads(null, RDF + 'type', SH + 'NodeShape').map(q => q.subject);
|
||||
if (!shapes.length) throw new Error(`${file}: no sh:NodeShape. A profile without entities is useless.`);
|
||||
|
||||
const shapeList = shapes.map(shape => {
|
||||
const targetClass = store.getQuads(shape, SH + 'targetClass', null)[0]?.object.value || null;
|
||||
const label = store.getQuads(shape, 'http://www.w3.org/2000/01/rdf-schema#label', null)[0]?.object.value || null;
|
||||
const fields = store.getQuads(shape, SH + 'property', null).map(q => parseField(store, q.object));
|
||||
return { uri: shape.value, targetClass, label, fields };
|
||||
});
|
||||
|
||||
return { uri, title, prefixes, shapes: shapeList, store };
|
||||
}
|
||||
|
||||
function parseField(store, p) {
|
||||
let path = null, inverse = false;
|
||||
const pathQuad = store.getQuads(p, SH + 'path', null)[0]?.object;
|
||||
if (pathQuad) {
|
||||
if (pathQuad.termType === 'BlankNode') {
|
||||
const inv = store.getQuads(pathQuad, SH + 'inversePath', null)[0]?.object;
|
||||
if (inv) { path = inv.value; inverse = true; }
|
||||
} else {
|
||||
path = pathQuad.value;
|
||||
}
|
||||
}
|
||||
const one = (pred) => store.getQuads(p, pred, null)[0]?.object || null;
|
||||
return {
|
||||
path, inverse,
|
||||
name: one(SH + 'name')?.value || null,
|
||||
description: one(SH + 'description')?.value || null,
|
||||
datatype: one(SH + 'datatype')?.value || null,
|
||||
class: one(SH + 'class')?.value || null,
|
||||
node: one(SH + 'node')?.value || null,
|
||||
nodeKind: one(SH + 'nodeKind')?.value || null,
|
||||
maxCount: one(SH + 'maxCount')?.value || null,
|
||||
};
|
||||
}
|
||||
|
||||
export function entityForClass(profile, cls) {
|
||||
const shape = profile.shapes.find(s => s.targetClass === cls);
|
||||
return shape ? shape.uri : null;
|
||||
}
|
||||
|
||||
export function shape(profile, shapeUri) {
|
||||
return profile.shapes.find(s => s.uri === shapeUri) || null;
|
||||
}
|
||||
|
||||
export function fieldsOf(profile, shapeUri) {
|
||||
return shape(profile, shapeUri)?.fields || [];
|
||||
}
|
||||
|
||||
export function field(profile, shapeUri, predicate, inverse) {
|
||||
return fieldsOf(profile, shapeUri).find(f => f.path === predicate && !!f.inverse === !!inverse) || null;
|
||||
}
|
||||
|
||||
export function prefixes(profile) {
|
||||
return profile.prefixes;
|
||||
}
|
||||
62
src/runner/report.js
Normal file
62
src/runner/report.js
Normal file
@@ -0,0 +1,62 @@
|
||||
import { randomUUID } from 'crypto';
|
||||
import { statSync } from 'fs';
|
||||
import { sessionUpdate, sparqlEscapeUri, sparqlEscapeString, sparqlEscapeInt,
|
||||
sparqlEscapeDateTime } from '../db.js';
|
||||
|
||||
const REPORT_CLASS = process.env.REPORT_CLASS || 'http://lblod.data.gift/vocabularies/reporting/Report';
|
||||
|
||||
// Write the file to /share, then register it once: a logical file
|
||||
// (http://data.lblod.info/files/<uuid>) plus a physical share:// file
|
||||
// pointing back with nie:dataSource. The logical URI is what you link to.
|
||||
export async function registerFile(sessionUpdateFn, fileName, filePath) {
|
||||
const size = statSync(filePath).size;
|
||||
const uuid = randomUUID();
|
||||
const uuid2 = randomUUID();
|
||||
const now = new Date();
|
||||
const logicalUri = `http://data.lblod.info/files/${uuid}`;
|
||||
const physicalUri = `share://${fileName}`;
|
||||
|
||||
await (sessionUpdateFn || sessionUpdate)(`
|
||||
INSERT DATA {
|
||||
${sparqlEscapeUri(logicalUri)} a ${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#FileDataObject')} ;
|
||||
${sparqlEscapeUri('http://mu.semte.ch/vocabularies/core/uuid')} ${sparqlEscapeString(uuid)} ;
|
||||
${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#fileName')} ${sparqlEscapeString(fileName)} ;
|
||||
${sparqlEscapeUri('http://purl.org/dc/terms/format')} ${sparqlEscapeString('text/csv')} ;
|
||||
${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#fileSize')} ${sparqlEscapeInt(size)} ;
|
||||
${sparqlEscapeUri('http://purl.org/dc/terms/created')} ${sparqlEscapeDateTime(now)} .
|
||||
${sparqlEscapeUri(physicalUri)} a ${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#FileDataObject')} ;
|
||||
${sparqlEscapeUri('http://mu.semte.ch/vocabularies/core/uuid')} ${sparqlEscapeString(uuid2)} ;
|
||||
${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#fileName')} ${sparqlEscapeString(fileName)} ;
|
||||
${sparqlEscapeUri('http://purl.org/dc/terms/format')} ${sparqlEscapeString('text/csv')} ;
|
||||
${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#fileSize')} ${sparqlEscapeInt(size)} ;
|
||||
${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/03/22/nfo#fileCreated')} ${sparqlEscapeDateTime(now)} ;
|
||||
${sparqlEscapeUri('http://www.semanticdesktop.org/ontologies/2007/01/19/nie#dataSource')} ${sparqlEscapeUri(logicalUri)} .
|
||||
}`);
|
||||
|
||||
return logicalUri;
|
||||
}
|
||||
|
||||
// One report pointing at the logical file URI. Insert without naming a graph;
|
||||
// the auth layer places the triples from the caller's session.
|
||||
export async function registerReport(sessionUpdateFn, title, fileUri, extra = {}) {
|
||||
const uuid = randomUUID();
|
||||
const now = new Date();
|
||||
const reportUri = `http://data.lblod.info/id/reports/${uuid}`;
|
||||
|
||||
const extraTriples = [];
|
||||
if (extra.question) extraTriples.push(`${sparqlEscapeUri(reportUri)} ${sparqlEscapeUri('http://mu.semte.ch/vocabularies/ext/question')} ${sparqlEscapeString(extra.question)} .`);
|
||||
if (extra.spec) extraTriples.push(`${sparqlEscapeUri(reportUri)} ${sparqlEscapeUri('http://mu.semte.ch/vocabularies/ext/spec')} ${sparqlEscapeString(extra.spec)} .`);
|
||||
if (extra.creator) extraTriples.push(`${sparqlEscapeUri(reportUri)} ${sparqlEscapeUri('http://purl.org/dc/terms/creator')} ${sparqlEscapeUri(extra.creator)} .`);
|
||||
|
||||
await (sessionUpdateFn || sessionUpdate)(`
|
||||
INSERT DATA {
|
||||
${sparqlEscapeUri(reportUri)} a ${sparqlEscapeUri(REPORT_CLASS)} ;
|
||||
${sparqlEscapeUri('http://mu.semte.ch/vocabularies/core/uuid')} ${sparqlEscapeString(uuid)} ;
|
||||
${sparqlEscapeUri('http://purl.org/dc/terms/title')} ${sparqlEscapeString(title)} ;
|
||||
${sparqlEscapeUri('http://purl.org/dc/terms/created')} ${sparqlEscapeDateTime(now)} ;
|
||||
${sparqlEscapeUri('http://www.w3.org/ns/prov#generated')} ${sparqlEscapeUri(fileUri)} .
|
||||
${extraTriples.join('\n ')}
|
||||
}`);
|
||||
|
||||
return reportUri;
|
||||
}
|
||||
41
src/runner/run.js
Normal file
41
src/runner/run.js
Normal file
@@ -0,0 +1,41 @@
|
||||
import { seed } from './seed.js';
|
||||
import { fetchColumns } from './columns.js';
|
||||
import { assemble } from './assemble.js';
|
||||
import { rowsToCsv, writeCsv } from './csv.js';
|
||||
import { registerFile, registerReport } from './report.js';
|
||||
import { touch } from './job.js';
|
||||
import { groupsQuery, groupsUpdate } from '../db.js';
|
||||
|
||||
// Ties the six steps together. Runs with the captured groups (see
|
||||
// captureGroups), returns { reportUri, fileUri, rowCount }. The task is
|
||||
// touched once per column group so the dashboard shows movement.
|
||||
|
||||
export async function run(allowedGroups, parsed, profile, title, taskUri, extra = {}) {
|
||||
const queryFn = sparql => groupsQuery(sparql, allowedGroups);
|
||||
const updateFn = (sparql, extra) => groupsUpdate(sparql, allowedGroups, extra);
|
||||
|
||||
const subjects = await seed(queryFn, parsed);
|
||||
|
||||
const values = new Map();
|
||||
await fetchColumns(queryFn, subjects, parsed, values,
|
||||
taskUri ? () => touch(updateFn, taskUri) : null);
|
||||
|
||||
const rows = assemble(subjects, values, parsed);
|
||||
const fileName = `${slug(title)}.csv`;
|
||||
const csv = rowsToCsv(rows);
|
||||
const filePath = writeCsv(fileName, csv);
|
||||
|
||||
const fileUri = await registerFile((sparql) => updateFn(sparql), fileName, filePath);
|
||||
const reportUri = await registerReport((sparql) => updateFn(sparql), title, fileUri, extra);
|
||||
|
||||
return { reportUri, fileUri, rowCount: rows.length - 1, filePath };
|
||||
}
|
||||
|
||||
// Exported for run_report: the file name must be known before the run starts,
|
||||
// so the chat can show a pending card named after the report.
|
||||
export function slug(title) {
|
||||
return (title || 'report').toLowerCase()
|
||||
.replace(/[^a-z0-9]+/g, '-')
|
||||
.replace(/^-+|-+$/g, '')
|
||||
.slice(0, 80) || 'report';
|
||||
}
|
||||
121
src/runner/seed.js
Normal file
121
src/runner/seed.js
Normal file
@@ -0,0 +1,121 @@
|
||||
import { sparqlEscapeUri, sparqlEscapeString, sparqlEscapeDateTime, sparqlEscapeDate } from '../db.js';
|
||||
|
||||
// Subject selection. One paged query per page; stops at ROW_LIMIT.
|
||||
// No GRAPH, no property paths, every hop its own variable.
|
||||
|
||||
const ROW_LIMIT = Number(process.env.ROW_LIMIT || 200000);
|
||||
const PAGE_SIZE = Number(process.env.SEED_PAGE_SIZE || 5000);
|
||||
|
||||
export async function seed(sessionQuery, spec) {
|
||||
const subjects = [];
|
||||
let offset = 0;
|
||||
for (;;) {
|
||||
const query = seedPageQuery(spec, PAGE_SIZE, offset);
|
||||
const result = await sessionQuery(query);
|
||||
const rows = result.results.bindings.map(b => b.s.value);
|
||||
subjects.push(...rows);
|
||||
if (subjects.length > ROW_LIMIT) {
|
||||
throw new Error(`the question matches more than ${ROW_LIMIT} subjects. Add a filter, for example a date range or a bestuurseenheid.`);
|
||||
}
|
||||
if (rows.length < PAGE_SIZE) break;
|
||||
offset += PAGE_SIZE;
|
||||
}
|
||||
return subjects;
|
||||
}
|
||||
|
||||
export function seedPageQuery(spec, limit, offset) {
|
||||
const parts = [];
|
||||
parts.push(`?s a ${sparqlEscapeUri(spec.targetClass)} .`);
|
||||
spec.filters.forEach((filter, fi) => {
|
||||
parts.push(hopsPattern(filter.path, `f${fi}`, '?s'));
|
||||
parts.push(constraintsPattern(filter.constraints, `f${fi}`));
|
||||
});
|
||||
return `SELECT DISTINCT ?s WHERE {
|
||||
${parts.join('\n ')}
|
||||
}
|
||||
LIMIT ${limit} OFFSET ${offset}`;
|
||||
}
|
||||
|
||||
// Writes out every hop as its own triple pattern.
|
||||
// Forward: ?s <p1> ?f0_1 . ?f0_1 <p2> ?f0_v
|
||||
// Inverse: ?s ^<p1> → ?f0_1 <p1> ?s
|
||||
export function hopsPattern(hops, varPrefix, startVar) {
|
||||
let prev = startVar;
|
||||
const patterns = [];
|
||||
for (let i = 0; i < hops.length; i++) {
|
||||
const hop = hops[i];
|
||||
const isLast = i === hops.length - 1;
|
||||
const v = isLast ? `?${varPrefix}_v` : `?${varPrefix}_${i + 1}`;
|
||||
if (hop.inverse) {
|
||||
patterns.push(`${v} ${sparqlEscapeUri(hop.predicate)} ${prev} .`);
|
||||
} else {
|
||||
patterns.push(`${prev} ${sparqlEscapeUri(hop.predicate)} ${v} .`);
|
||||
}
|
||||
prev = v;
|
||||
}
|
||||
return patterns.join('\n ');
|
||||
}
|
||||
|
||||
export function constraintsPattern(c, varPrefix) {
|
||||
const parts = [];
|
||||
const v = `?${varPrefix}_v`;
|
||||
|
||||
if (c.hasValue) parts.push(`FILTER(${v} = ${escapeTerm(c.hasValue)})`);
|
||||
if (c.in) parts.push(`FILTER(${v} IN (${c.in.map(escapeTerm).join(', ')}))`);
|
||||
if (c.minInclusive) parts.push(`FILTER(${v} >= ${escapeTerm(c.minInclusive)})`);
|
||||
if (c.maxInclusive) parts.push(`FILTER(${v} <= ${escapeTerm(c.maxInclusive)})`);
|
||||
if (c.minExclusive) parts.push(`FILTER(${v} > ${escapeTerm(c.minExclusive)})`);
|
||||
if (c.maxExclusive) parts.push(`FILTER(${v} < ${escapeTerm(c.maxExclusive)})`);
|
||||
if (c.pattern) {
|
||||
const flags = c.flags ? `, ${sparqlEscapeString(c.flags)}` : '';
|
||||
parts.push(`FILTER(REGEX(str(${v}), ${sparqlEscapeString(c.pattern)}${flags}))`);
|
||||
}
|
||||
if (c.anyOf) {
|
||||
const alternation = c.anyOf.map(t => escapeRegexTerm(t.value)).join('|');
|
||||
const flags = c.flags ? `, ${sparqlEscapeString(c.flags)}` : ', "i"';
|
||||
parts.push(`FILTER(REGEX(str(${v}), ${sparqlEscapeString(alternation)}${flags}))`);
|
||||
}
|
||||
if (c.minCount === 2) {
|
||||
// two distinct leaf values on the same path
|
||||
const w = `?${varPrefix}_w`;
|
||||
parts.push(hopsPattern2(c, varPrefix, w));
|
||||
parts.push(`FILTER(${v} != ${w})`);
|
||||
}
|
||||
if (c.maxCount === 0) {
|
||||
parts.push(`FILTER NOT EXISTS { ${hopsPattern(c.path, varPrefix + '_nex', '?s')} }`);
|
||||
}
|
||||
return parts.join('\n ');
|
||||
}
|
||||
|
||||
function hopsPattern2(filter, varPrefix, wVar) {
|
||||
// repeat the path to a second leaf variable
|
||||
const hops = filter.path.map(h => ({ ...h }));
|
||||
let prev = '?s';
|
||||
const patterns = [];
|
||||
for (let i = 0; i < hops.length; i++) {
|
||||
const hop = hops[i];
|
||||
const isLast = i === hops.length - 1;
|
||||
const v = isLast ? wVar : `?${varPrefix}_2_${i + 1}`;
|
||||
if (hop.inverse) patterns.push(`${v} ${sparqlEscapeUri(hop.predicate)} ${prev} .`);
|
||||
else patterns.push(`${prev} ${sparqlEscapeUri(hop.predicate)} ${v} .`);
|
||||
prev = v;
|
||||
}
|
||||
return patterns.join('\n ');
|
||||
}
|
||||
|
||||
function escapeTerm(t) {
|
||||
if (t.type === 'uri') return sparqlEscapeUri(t.value);
|
||||
if (t.datatype === 'http://www.w3.org/2001/XMLSchema#dateTime')
|
||||
return sparqlEscapeDateTime(new Date(t.value));
|
||||
if (t.datatype === 'http://www.w3.org/2001/XMLSchema#date')
|
||||
return sparqlEscapeDate(new Date(t.value));
|
||||
if (t.datatype === 'http://www.w3.org/2001/XMLSchema#integer')
|
||||
return Number(t.value);
|
||||
if (t.datatype === 'http://www.w3.org/2001/XMLSchema#boolean')
|
||||
return t.value === 'true';
|
||||
return sparqlEscapeString(t.value);
|
||||
}
|
||||
|
||||
function escapeRegexTerm(s) {
|
||||
return s.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
|
||||
}
|
||||
153
src/runner/spec.js
Normal file
153
src/runner/spec.js
Normal file
@@ -0,0 +1,153 @@
|
||||
import N3 from 'n3';
|
||||
|
||||
// Parse a spec (rep:ReportSpec + sh:NodeShape) into a plain object.
|
||||
// No validation here; that is check.js.
|
||||
|
||||
const SH = 'http://www.w3.org/ns/shacl#';
|
||||
const RDF = 'http://www.w3.org/1999/02/22-rdf-syntax-ns#';
|
||||
const REP = 'http://mu.semte.ch/vocabularies/reporting/';
|
||||
const RDFS_LABEL = 'http://www.w3.org/2000/01/rdf-schema#label';
|
||||
|
||||
export function parseSpec(turtle) {
|
||||
const parser = new N3.Parser();
|
||||
const quads = parser.parse(turtle);
|
||||
const store = new N3.Store(quads);
|
||||
|
||||
const specNode = store.getQuads(null, RDF + 'type', REP + 'ReportSpec')[0]?.subject;
|
||||
if (!specNode) throw new Error('spec has no rep:ReportSpec');
|
||||
|
||||
const one = (s, pred) => store.getQuads(s, pred, null)[0]?.object || null;
|
||||
|
||||
const targetClass = one(specNode, SH + 'targetClass')?.value || null;
|
||||
const profileUri = one(specNode, REP + 'profile')?.value || null;
|
||||
const title = one(specNode, 'http://purl.org/dc/terms/title')?.value || null;
|
||||
|
||||
const filters = store.getQuads(specNode, SH + 'property', null).map(q =>
|
||||
parseFilter(store, q.object, { withLabel: true }));
|
||||
|
||||
const columnsQuad = one(specNode, REP + 'columns');
|
||||
const columns = columnsQuad ? parseColumnList(store, columnsQuad) : [];
|
||||
|
||||
return { uri: specNode.value, title, profileUri, targetClass, filters, columns };
|
||||
}
|
||||
|
||||
function parseFilter(store, p, { withLabel = false } = {}) {
|
||||
let path = [];
|
||||
const pathQuad = store.getQuads(p, SH + 'path', null)[0]?.object;
|
||||
if (pathQuad) path = parsePath(store, pathQuad);
|
||||
return {
|
||||
path,
|
||||
constraints: readConstraints(store, p, { withLabel }),
|
||||
};
|
||||
}
|
||||
|
||||
function parseColumnList(store, listNode) {
|
||||
const columns = [];
|
||||
let cur = listNode;
|
||||
while (cur && cur.value !== RDF + 'nil') {
|
||||
const item = store.getQuads(cur, RDF + 'first', null)[0]?.object;
|
||||
if (item) columns.push(parseColumn(store, item));
|
||||
cur = store.getQuads(cur, RDF + 'rest', null)[0]?.object;
|
||||
if (!cur) throw new Error('rep:columns list is broken (no rdf:rest terminator)');
|
||||
}
|
||||
return columns;
|
||||
}
|
||||
|
||||
function parseColumn(store, col) {
|
||||
let path = [];
|
||||
const pathQuad = store.getQuads(col, SH + 'path', null)[0]?.object;
|
||||
if (pathQuad) {
|
||||
if (pathQuad.termType === 'NamedNode' && pathQuad.value === REP + 'self') {
|
||||
path = []; // rep:self: the row's own subject
|
||||
} else {
|
||||
path = parsePath(store, pathQuad);
|
||||
}
|
||||
}
|
||||
return {
|
||||
path,
|
||||
label: store.getQuads(col, RDFS_LABEL, null)[0]?.object.value || null,
|
||||
collect: store.getQuads(col, REP + 'collect', null)[0]?.object.value || null,
|
||||
separator: store.getQuads(col, SH + 'separator', null)[0]?.object.value || null,
|
||||
constraints: readConstraints(store, col), // kept so check.js can refuse it
|
||||
};
|
||||
}
|
||||
|
||||
// A path is a single node (predicate, or a blank with sh:inversePath)
|
||||
// or an RDF list of predicates. Returns Hop[] = { predicate, inverse }.
|
||||
function parsePath(store, pathQuad) {
|
||||
if (pathQuad.termType === 'BlankNode') {
|
||||
const inv = store.getQuads(pathQuad, SH + 'inversePath', null)[0]?.object;
|
||||
if (inv) return [{ predicate: inv.value, inverse: true }];
|
||||
const first = store.getQuads(pathQuad, RDF + 'first', null)[0]?.object;
|
||||
if (!first) throw new Error('sh:path blank node without sh:inversePath is not supported');
|
||||
}
|
||||
if (pathQuad.termType === 'NamedNode') {
|
||||
return [{ predicate: pathQuad.value, inverse: false }];
|
||||
}
|
||||
// RDF list of predicates
|
||||
const hops = [];
|
||||
let cur = pathQuad;
|
||||
while (cur && cur.value !== RDF + 'nil') {
|
||||
const item = store.getQuads(cur, RDF + 'first', null)[0]?.object;
|
||||
if (!item) throw new Error('sh:path list is broken');
|
||||
if (item.termType !== 'NamedNode') throw new Error('sh:path list items must be URIs');
|
||||
hops.push({ predicate: item.value, inverse: false });
|
||||
cur = store.getQuads(cur, RDF + 'rest', null)[0]?.object;
|
||||
}
|
||||
return hops;
|
||||
}
|
||||
|
||||
function readConstraints(store, p, { withLabel = false } = {}) {
|
||||
const one = (pred) => store.getQuads(p, pred, null)[0]?.object || null;
|
||||
const constraints = {};
|
||||
const minCount = one(SH + 'minCount');
|
||||
if (minCount) constraints.minCount = Number(minCount.value);
|
||||
const maxCount = one(SH + 'maxCount');
|
||||
if (maxCount) constraints.maxCount = Number(maxCount.value);
|
||||
const hasValue = one(SH + 'hasValue');
|
||||
if (hasValue) constraints.hasValue = term(hasValue);
|
||||
const inList = one(SH + 'in');
|
||||
if (inList) constraints.in = readRdfList(store, inList).map(term);
|
||||
for (const [key, pred] of [
|
||||
['minInclusive', SH + 'minInclusive'], ['maxInclusive', SH + 'maxInclusive'],
|
||||
['minExclusive', SH + 'minExclusive'], ['maxExclusive', SH + 'maxExclusive'],
|
||||
]) {
|
||||
const q = one(pred);
|
||||
if (q) constraints[key] = term(q);
|
||||
}
|
||||
const pattern = one(SH + 'pattern');
|
||||
if (pattern) constraints.pattern = pattern.value;
|
||||
const flags = one(SH + 'flags');
|
||||
if (flags) constraints.flags = flags.value;
|
||||
const anyOf = one(REP + 'anyOf');
|
||||
if (anyOf) constraints.anyOf = readRdfList(store, anyOf).map(term)
|
||||
.map(t => ({ type: t.type, value: t.value }));
|
||||
// read but never used by the runner; check.js refuses it (a filter has no label)
|
||||
if (withLabel) {
|
||||
const label = one(RDFS_LABEL);
|
||||
if (label) constraints.label = label.value;
|
||||
}
|
||||
return constraints;
|
||||
}
|
||||
|
||||
function readRdfList(store, node) {
|
||||
const items = [];
|
||||
let cur = node;
|
||||
while (cur && cur.value !== RDF + 'nil') {
|
||||
const item = store.getQuads(cur, RDF + 'first', null)[0]?.object;
|
||||
if (!item) throw new Error('RDF list is broken');
|
||||
items.push(item);
|
||||
cur = store.getQuads(cur, RDF + 'rest', null)[0]?.object;
|
||||
}
|
||||
return items;
|
||||
}
|
||||
|
||||
function term(object) {
|
||||
if (object.termType === 'NamedNode') return { type: 'uri', value: object.value };
|
||||
return {
|
||||
type: 'literal',
|
||||
value: object.value,
|
||||
datatype: object.datatype ? object.datatype.value : null,
|
||||
language: object.language || null,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user