'use strict'; import { execFileSync } from 'child_process'; import multer from 'multer'; import { LLMOrbiter, OpenRouterProvider, defineTool, DirectReturnError } from 'libllmorbiter'; import * as dbCompanySetting from '../../db/companysetting.js'; import * as dbSupplier from '../../db/supplier.js'; import env from '../../config/env.js'; import { resolveDefaultModel } from '../../tools/llm-key.js'; // ─── Multer — memory storage (no temp files); accepts PDF or text ─────────────── const upload = multer({ storage: multer.memoryStorage(), limits: { fileSize: 50 * 1024 * 1024 }, // 50 MB fileFilter: (_req, file, cb) => { const name = file.originalname?.toLowerCase() || ''; const isPdf = file.mimetype === 'application/pdf' || name.endsWith('.pdf'); const isText = file.mimetype.startsWith('text/') || /\.(txt|text|md|csv|log|json)$/.test(name); if (isPdf || isText) { cb(null, true); } else { cb(new Error('Only PDF or text files are accepted')); } }, }); /** Detects whether an uploaded file is a PDF (else treated as text). */ function isPdfUpload(file) { const name = file.originalname?.toLowerCase() || ''; return file.mimetype === 'application/pdf' || name.endsWith('.pdf'); } // ─── SSE helpers ──────────────────────────────────────────────────────────────── function sseWrite(res, event, data) { if (res.writableEnded || res.destroyed) return; res.write(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`); } // ─── PDF → pages via pdftotext ────────────────────────────────────────────────── function readPdfPages(buffer) { // Write buffer to stdin by passing '-' as output and reading from stdin via // execFileSync input option. pdftotext reads from '-' for stdin. const raw = execFileSync('pdftotext', ['-layout', '-', '-'], { input: buffer, encoding: 'utf8', maxBuffer: 50 * 1024 * 1024, }); return raw .split('\f') .map((p) => p.trim()) .filter((p) => p.length > 0); } // ─── Text → pages split by form feed (same convention as pdftotext) ───────────── function readTxtPages(textOrBuffer) { const text = Buffer.isBuffer(textOrBuffer) ? textOrBuffer.toString('utf8') : String(textOrBuffer); return text .split('\f') .map((p) => p.trim()) .filter((p) => p.length > 0); } // ─── Extraction prompt ────────────────────────────────────────────────────────── const BASE_PROMPT = `Extract metadata and product line items from the provided invoice or delivery note. Call the output_manifest tool with the extracted data. Do not answer with plain text once extraction is complete. Rules: - manifestType: "rechnung" or "lieferschein". For "lieferschein", prices and taxes must be null. - Exclude non-product lines (shipping, handling, etc.). - Line items: Extract count, vpe (default: 1), liefArtNr, mwst (7 or 19), and eknetto per piece(euro cents). - Match supplier against the provided list for lieferant_id and lieferantName. For unknown suppliers, extract lieferantName and set lieferant_id to null. - Write notes strictly in German. - The output_manifest tool has three strictly optional components: basedata, notes, items. Only include a component when it carries genuinely new data for this batch; omit the rest. - Include basedata ONLY when at least one of its fields is currently null/empty and you can fill it in, or when a value actually changes. If all basedata fields are already set and unchanged, omit basedata. - Include items ONLY when this batch contains new product line items not already present. If there are no new items, omit items. - Include notes ONLY when there is a genuinely new noteworthy detail not already in the notes list. These notes are for a human to double-check the extraction result, so record anything ambiguous, unusual, or worth verifying (e.g. packaging interpretations, excluded lines, unavailable articles). VPE (Verpackungseinheit) Logic: VPE represents the number of individually sellable units inside a bulk item. If not explicitly split for individual sale, default to 1. - "20 Boxen á 10 Stück" -> VPE = 20. - "10 Gläser á 100 Filter" -> VPE = 10 (we sell the 10 glasses individually). - "100 Stück" -> VPE = 1 (sold as one bulk pack). - We don't sell the individual filters or Pieces, but typical VPEs like Beutel, Glässer, Boxen, Hefte - That means that in "Y Hefte á X Blatt" the VPE is NOT X but: Y - a so called "Display" also means VPE. So "20er Display" means VPE = 20 - eknetto is the price of one full vpe Unit (ME) rules: - The unit (ME) must ALWAYS refer to the individual sellable item (e.g. Stk., Beutel, Flasche, Box, Set). It must NEVER be "VE" or "VPE". - A packaging size like "VE100" means the bulk pack contains 100 individual items: set vpe = 100 and unit = the individual item (e.g. Stk.). - Example: "Einhängesiebe Stahl, VE100" -> vpe = 100, unit = "Stk.", count = 1 (one bulk pack of 100 sieves).`; function buildSystemPrompt(suppliers = []) { const supplierList = suppliers.length > 0 ? suppliers.map((s) => `${s.id} | ${s.name}`).join('\n') : '(no suppliers registered)'; return `${BASE_PROMPT} Suppliers (lieferant_id | lieferant_name): ${supplierList}`; } // ─── Shared SSE manifest extraction (POST /delivery/parse) ───────────────────── /** * Fetches the OpenRouter API key, opens the SSE stream, and runs the LLM * manifest extraction over the given pages. Closes the stream when done. * * @param {object} opts * @param {import('express').Request} opts.req * @param {import('express').Response} opts.res * @param {string[]} opts.pages Trimmed, non-empty document pages * @param {string} opts.sourceLabel e.g. 'PDF' or 'Text' */ async function parseDocumentManifest({ req, res, pages, sourceLabel }) { // ── Fetch OpenRouter API key from company settings (or fallback) ── let apiKey; try { const setting = await dbCompanySetting.getByKey(req.user.id, 'keys.openrouter', { maskWriteonly: false }); apiKey = setting?.value?.trim(); } catch (err) { console.error('delivery parse: failed to read OpenRouter key:', err); } if (!apiKey) { apiKey = env.openrouterFreeKey?.trim(); if (apiKey) { console.log('delivery parse: using fallback OPENROUTER_FREE_KEY'); } } if (!apiKey) { return res.status(400).json({ error: 'OpenRouter API key not configured (companysetting: keys.openrouter)' }); } // ── Fetch suppliers from Lieferanten group ──────────────────────── let suppliers = []; try { suppliers = await dbSupplier.list(req.user.id); } catch (err) { console.warn('delivery parse: failed to list suppliers:', err?.message || err); } // ── Open SSE stream ────────────────────────────────────────────── res.setHeader('Content-Type', 'text/event-stream'); res.setHeader('Cache-Control', 'no-cache'); res.setHeader('Connection', 'keep-alive'); res.flushHeaders(); sseWrite(res, 'status', { message: `${sourceLabel} parsed: ${pages.length} page(s)` }); // ── Accumulated state ──────────────────────────────────────────── const manifestBasedata = { date: null, lieferant_id: null, lieferantName: null, manifestNr: null, manifestType: null, }; const manifestNotes = []; const manifestItems = { items: [], pagesFrom: null, pagesTo: null }; let currentPage = 0; // ── Tool definition ────────────────────────────────────────────── const outputManifestTool = defineTool({ name: 'output_manifest', description: 'Stores extracted invoice/delivery-note data: header metadata (basedata), notes, and product line items. Each component is strictly optional — only include the ones that carry new data.', strict: false, parameters: { type: 'object', additionalProperties: false, required: [], properties: { basedata: { anyOf: [ { type: 'object', additionalProperties: false, properties: { date: { type: ['string', 'null'], description: 'Document date (ISO 8601)' }, lieferant_id: { type: ['number', 'null'], description: 'Lieferant ID' }, lieferantName: { type: 'string', description: 'Lieferant name' }, manifestNr: { type: ['string', 'null'], description: 'Lieferscheinnummer or Rechnungsnummer' }, manifestType: { anyOf: [ { type: 'string', enum: ['rechnung', 'lieferschein'] }, { type: 'null' }, ], description: 'Document type', }, }, }, { type: 'null' }, ], description: 'Header/metadata. Only include fields that are currently missing or changed.', }, notes: { anyOf: [ { type: 'array', items: { type: 'string' } }, { type: 'null' }, ], description: 'New noteworthy details (in German), for a human to double-check the result.', }, items: { anyOf: [ { type: 'object', additionalProperties: false, properties: { numberOfItemsToAdd: { type: 'number', description: 'Number of NEW line items being added in this call.' }, items: { type: 'array', description: 'Invoice/delivery product line items', items: { type: 'object', additionalProperties: false, required: ['position', 'count', 'vpe', 'unit', 'articleName', 'liefArtNr', 'mwst', 'eknetto'], properties: { position: { type: 'number', description: 'Position' }, count: { type: 'number', description: 'Quantity (Menge)' }, vpe: { type: 'number', description: 'Packaging unit size (VPE). Default: 1.' }, unit: { type: 'string', description: 'Unit of the individual sellable item. NEVER "VE" or "VPE".' }, articleName: { type: 'string', description: 'Article name' }, liefArtNr: { type: 'string', description: 'Supplier article number' }, mwst: { anyOf: [{ type: 'string', enum: ['7', '19'] }, { type: 'null' }], description: 'VAT rate; null if missing', }, eknetto: { type: ['number', 'null'], description: 'Net price per vpe in euro cents; null if missing' }, }, }, }, }, }, { type: 'null' }, ], description: 'New product line items.', }, }, }, invoke: async (args) => { const basedataKeys = args.basedata && typeof args.basedata === 'object' ? Object.keys(args.basedata) : []; const notesCount = Array.isArray(args.notes) ? args.notes.length : 0; const itemsCount = args.items?.items && Array.isArray(args.items.items) ? args.items.items.length : 0; console.log(`[llmorbiter] Tool invoke output_manifest: basedata=[${basedataKeys.join(', ')}], notes=${notesCount}, items=${itemsCount}`); // Merge basedata if (args.basedata && typeof args.basedata === 'object') { for (const key of ['date', 'lieferant_id', 'lieferantName', 'manifestNr', 'manifestType']) { if (args.basedata[key] != null) manifestBasedata[key] = args.basedata[key]; } if (args.basedata.kLieferant != null && manifestBasedata.lieferant_id == null) { manifestBasedata.lieferant_id = args.basedata.kLieferant; } if (args.basedata.lieferant_name != null && manifestBasedata.lieferantName == null) { manifestBasedata.lieferantName = args.basedata.lieferant_name; } } // Merge notes if (Array.isArray(args.notes)) { for (const n of args.notes) { if (n && !manifestNotes.includes(n)) manifestNotes.push(n); } } // Merge items if (args.items && typeof args.items === 'object' && Array.isArray(args.items.items)) { const existing = new Set(manifestItems.items.map((it) => it.position)); for (const it of args.items.items) { if (!existing.has(it.position)) { manifestItems.items.push({ ...it, page: currentPage }); existing.add(it.position); } } const itemPages = manifestItems.items.map((it) => it.page).filter((p) => p != null); if (itemPages.length) { manifestItems.pagesFrom = Math.min(...itemPages); manifestItems.pagesTo = Math.max(...itemPages); } } // Emit intermediate progress sseWrite(res, 'progress', { basedata: manifestBasedata, notes: manifestNotes, items: manifestItems, }); throw new DirectReturnError({ basedata: manifestBasedata, notes: manifestNotes, items: manifestItems, }); }, }); // ── LLMOrbiter setup ───────────────────────────────────────────── const model = await resolveDefaultModel(req.user.id); const provider = new OpenRouterProvider({ apiKey, apiUrl: env.openrouterApiUrl }); const orbiter = new LLMOrbiter(provider, { onStatus: (status) => { console.log(`[llmorbiter] Status: ${status}`); sseWrite(res, 'status', { message: status }); }, onLLMCall: (call) => { console.log( `[llmorbiter] LLM call (iteration ${call.iteration}): model=${call.model}, promptSize=${call.promptSize} chars, messages=${call.messageCount || call.messages}` ); sseWrite(res, 'llm_call', { model: call.model, promptSize: call.promptSize, messages: call.messages, }); }, onToolCall: (call) => { const details = []; if (call.args?.basedata) details.push(`basedata: [${Object.keys(call.args.basedata).join(', ')}]`); if (call.args?.items?.items) details.push(`items: ${call.args.items.items.length}`); if (call.args?.notes) details.push(`notes: ${call.args.notes.length}`); const suffix = details.length ? ` (${details.join(', ')})` : ''; console.log(`[llmorbiter] Tool call: ${call.name}${suffix}`); sseWrite(res, 'tool_call', { name: call.name }); }, onToolResult: (call) => { console.log(`[llmorbiter] Tool result: ${call.name}`); }, onToolError: (call) => { console.error(`[llmorbiter] Tool error in ${call.name}:`, call.error); }, onUsage: (usage) => { const costStr = usage.cost != null ? `, cost=$${usage.cost.toFixed(4)}` : ''; console.log( `[llmorbiter] Usage: input=${usage.input}, cached=${usage.cached}, output=${usage.output}${costStr}` ); sseWrite(res, 'usage', { input: usage.input, cached: usage.cached, output: usage.output, cost: usage.cost, }); }, }); const abortController = new AbortController(); abortController.signal.addEventListener('abort', () => { const reason = abortController.signal.reason; const msg = reason instanceof Error ? reason.message : String(reason || 'aborted'); console.log(`[llmorbiter] Abort signal triggered: ${msg}`); }); const originalAbort = orbiter.abort.bind(orbiter); orbiter.abort = (reason) => { const reasonMsg = reason instanceof Error ? reason.message : String(reason || 'aborted'); console.log(`[llmorbiter] orbiter.abort() called (reason: ${reasonMsg})`); return originalAbort(reason); }; const onDisconnect = () => { if (!res.writableEnded && !abortController.signal.aborted) { console.log('[llmorbiter] Client connection closed, aborting request'); abortController.abort('Client connection closed'); orbiter.abort('Client connection closed'); } }; res.on('close', onDisconnect); // ── Process all pages in one batch ─────────────────────────────── try { if (res.destroyed || res.writableEnded) { console.log('[llmorbiter] Response already closed or destroyed before run, aborting'); abortController.abort('Response closed before run'); orbiter.abort('Response closed before run'); return; } const pageContent = pages .map((p, idx) => `--- Page ${idx + 1} of ${pages.length} ---\n${p}`) .join('\n\n'); const systemPrompt = buildSystemPrompt(suppliers); const prompt = `You are processing all ${pages.length} page(s) of this document. Current manifest_basedata (fill in what is still missing): ${JSON.stringify(manifestBasedata, null, 2)} Current notes: ${JSON.stringify(manifestNotes, null, 2)} Current manifest_items: ${JSON.stringify(manifestItems, null, 2)} Page content: ${pageContent}`; console.log(`[llmorbiter] Starting run: model=${model}, pages=${pages.length} (${sourceLabel}), suppliers=${suppliers.length}`); currentPage = 1; await orbiter.run({ prompt, systemPrompt, model, tools: [outputManifestTool], signal: abortController.signal, }); console.log(`[llmorbiter] Run completed successfully. Extracted ${manifestItems.items.length} item(s), ${manifestNotes.length} note(s)`); } catch (err) { if (err.name === 'AbortError' || abortController.signal.aborted) { console.log(`[llmorbiter] Run aborted: ${err.message || abortController.signal.reason?.message || 'AbortError'}`); } else { console.error('[llmorbiter] Run failed with error:', err); } if (!res.writableEnded && !res.destroyed) { sseWrite(res, 'error', { message: 'LLM processing failed: ' + err.message }); } } finally { res.off('close', onDisconnect); } if (res.writableEnded || res.destroyed) { console.log('[llmorbiter] Response ended or destroyed, skipping final result'); return; } // ── Final result ───────────────────────────────────────────────── console.log(`[llmorbiter] Emitting final result event and closing stream`); sseWrite(res, 'result', { basedata: manifestBasedata, notes: manifestNotes, items: manifestItems, }); res.end(); } // ─── Route ─────────────────────────────────────────────────────────────────────── /** * @param {import('express').Application} app * @param {string} [basePath=''] */ export function register(app, basePath = '') { /** * @openapi * /delivery/parse: * post: * tags: [Delivery] * summary: Parse document (SSE) * description: | * Accepts an invoice or delivery note as PDF or text file * (`multipart/form-data`, field `file`) and streams extraction results as * Server-Sent Events until parsing is complete. * * PDFs are read via `pdftotext -layout`; text files are decoded as UTF-8. * Pages are separated by the form feed character (`\f`). If no form feed * is present, the input is treated as a single page. * * The extracted pages are fed to an LLM (OpenRouter) which fills a structured * manifest of header data and line items. The OpenRouter API key is read from * the authenticated user's company setting `keys.openrouter` (or the server's * `OPENROUTER_FREE_KEY` if not configured); the endpoint * returns `400` if the key is not configured. * * The stream ends after the final `result` event (or an `error` event on failure). * security: * - cookieAuth: [] * x-markdown: delivery/parse.md * x-sse-events: * - name: status * description: Document parse confirmation and LLM status messages * payload: '{ "message": "…" }' * - name: llm_call * description: Fired before each LLM request * payload: '{ "model": "…", "promptSize": 0, "messages": 0 }' * - name: tool_call * description: Fired when the output_manifest tool is invoked * payload: '{ "name": "output_manifest" }' * - name: usage * description: Token usage after each LLM response * payload: '{ "input": 0, "cached": 0, "output": 0, "cost": 0 }' * - name: progress * description: Intermediate accumulated extraction state (basedata / notes / items) * payload: ManifestResult * - name: result * description: Final accumulated extraction — emitted just before the stream closes * payload: ManifestResult * - name: error * description: Emitted on document parse or LLM failure; stream closes immediately after * payload: '{ "message": "…" }' * requestBody: * required: true * content: * multipart/form-data: * schema: * type: object * required: [file] * properties: * file: * type: string * format: binary * description: Document to parse (max 50 MB, application/pdf or text file) * responses: * 200: * description: | * `text/event-stream` — emits `status`, `llm_call`, `tool_call`, `usage`, * `progress`, `result` (and `error` on failure). See SSE Events table below. * * **ManifestResult shape** (used by `progress` and `result`): * * ```json * { * "basedata": { * "date": "2026-01-15", * "lieferant_id": 42, * "lieferantName": "Muster GmbH", * "manifestNr": "LS-2026-001", * "manifestType": "lieferschein" * }, * "notes": ["Pos. 3: VPE-Angabe fehlte, Standardwert 1 angenommen."], * "items": { * "pagesFrom": 1, * "pagesTo": 2, * "items": [ * { * "position": 1, * "count": 10, * "vpe": 1, * "unit": "Stk.", * "articleName": "Einhängesieb Stahl", * "liefArtNr": "ES-100", * "mwst": "19", * "eknetto": 450, * "page": 1 * } * ] * } * } * ``` * content: * text/event-stream: * schema: * type: string * 400: * description: No file uploaded, unsupported file type, or OpenRouter key not configured * content: * application/json: * schema: * $ref: '#/components/schemas/Error' * 401: * description: Unauthorized * content: * application/json: * schema: * $ref: '#/components/schemas/Error' */ app.post( `${basePath}/delivery/parse`, upload.single('file'), async (req, res) => { if (!req.file?.buffer) { return res.status(400).json({ error: 'A document file is required (field name: file)' }); } const sourceLabel = isPdfUpload(req.file) ? 'PDF' : 'Text'; console.log(`[deliveryParse] Received file "${req.file.originalname}" (${req.file.size} bytes, type=${sourceLabel})`); let pages; try { pages = sourceLabel === 'PDF' ? readPdfPages(req.file.buffer) : readTxtPages(req.file.buffer); } catch (err) { console.error(`[deliveryParse] Failed to parse document: ${err.message}`); return res.status(400).json({ error: 'Failed to parse document: ' + err.message }); } if (pages.length === 0) { console.warn('[deliveryParse] Document content is empty'); return res.status(400).json({ error: 'Document content is empty' }); } await parseDocumentManifest({ req, res, pages, sourceLabel, }); }, ); }