unify parse api

This commit is contained in:
seb
2026-09-18 04:35:42 +02:00
parent adb529d14f
commit bbd285109d
5 changed files with 491 additions and 443 deletions

View File

@@ -0,0 +1,463 @@
'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';
// ─── 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) 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 kLieferant and lieferantName. For unknown suppliers, extract lieferantName and set kLieferant 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).`;
// ─── 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 ────────────────
let apiKey;
try {
const setting = await dbCompanySetting.getByKey(req.user.id, 'keys.openrouter', { maskWriteonly: false });
apiKey = setting?.value;
} catch (err) {
console.error('delivery parse: failed to read OpenRouter key:', err);
}
if (!apiKey) {
return res.status(400).json({ error: 'OpenRouter API key not configured (companysetting: keys.openrouter)' });
}
// ── 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,
kLieferant: 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)' },
kLieferant: { 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) => {
// Merge basedata
if (args.basedata && typeof args.basedata === 'object') {
for (const key of ['date', 'kLieferant', 'lieferantName', 'manifestNr', 'manifestType']) {
if (args.basedata[key] != null) manifestBasedata[key] = args.basedata[key];
}
}
// 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 = '~openai/gpt-luna-latest:floor';
const provider = new OpenRouterProvider({ apiKey });
const orbiter = new LLMOrbiter(provider, {
onStatus: (status) => sseWrite(res, 'status', { message: status }),
onLLMCall: (call) =>
sseWrite(res, 'llm_call', {
model: call.model,
promptSize: call.promptSize,
messages: call.messages,
}),
onToolCall: (call) => sseWrite(res, 'tool_call', { name: call.name }),
onUsage: (usage) =>
sseWrite(res, 'usage', {
input: usage.input,
cached: usage.cached,
output: usage.output,
cost: usage.cost,
}),
});
// ── Process all pages in one batch ───────────────────────────────
try {
const pageContent = pages
.map((p, idx) => `--- Page ${idx + 1} of ${pages.length} ---\n${p}`)
.join('\n\n');
const prompt = `${BASE_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}`;
currentPage = 1;
await orbiter.run({ prompt, model, tools: [outputManifestTool] });
} catch (err) {
if (!res.writableEnded) {
sseWrite(res, 'error', { message: 'LLM processing failed: ' + err.message });
}
}
// ── Final result ─────────────────────────────────────────────────
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`; 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",
* "kLieferant": 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)' });
}
let pages;
try {
pages = isPdfUpload(req.file) ? readPdfPages(req.file.buffer) : readTxtPages(req.file.buffer);
} catch (err) {
return res.status(400).json({ error: 'Failed to parse document: ' + err.message });
}
if (pages.length === 0) {
return res.status(400).json({ error: 'Document content is empty' });
}
await parseDocumentManifest({
req,
res,
pages,
sourceLabel: isPdfUpload(req.file) ? 'PDF' : 'Text',
});
},
);
}