KI-AGENT: Mandantenbezogenen IMAP-Dokumentenimport umsetzen
This commit is contained in:
@@ -35,6 +35,7 @@ import communicationRoutes from "./routes/communication";
|
||||
import telephonyRoutes from "./routes/telephony";
|
||||
import instanceAgentRoutes from "./routes/instanceAgents";
|
||||
import instanceAgentGatewayRoutes from "./routes/instanceAgentGateway";
|
||||
import documentImportRoutes from "./routes/documentImports";
|
||||
|
||||
//Public Links
|
||||
import publiclinksNonAuthenticatedRoutes from "./routes/publiclinks/publiclinks-non-authenticated";
|
||||
@@ -62,6 +63,7 @@ import {initS3} from "./utils/s3";
|
||||
import { runBootstrap } from "./modules/bootstrap.service";
|
||||
import { startMatrixPushWorker } from "./modules/matrix-push-worker.service";
|
||||
import { startCentralServicesHeartbeat } from "./modules/central-services-heartbeat.service";
|
||||
import { startDocumentImportWorker } from "./modules/document-import/document-import.worker";
|
||||
|
||||
|
||||
//Services
|
||||
@@ -89,6 +91,7 @@ async function main() {
|
||||
await runBootstrap(app);
|
||||
startMatrixPushWorker(app);
|
||||
startCentralServicesHeartbeat(app);
|
||||
startDocumentImportWorker(app);
|
||||
|
||||
app.addHook('preHandler', (req, reply, done) => {
|
||||
console.log(req.method)
|
||||
@@ -167,6 +170,7 @@ async function main() {
|
||||
await subApp.register(communicationRoutes);
|
||||
await subApp.register(telephonyRoutes);
|
||||
await subApp.register(instanceAgentRoutes);
|
||||
await subApp.register(documentImportRoutes);
|
||||
|
||||
},{prefix: "/api"})
|
||||
|
||||
|
||||
307
backend/src/modules/document-import/document-import.service.ts
Normal file
307
backend/src/modules/document-import/document-import.service.ts
Normal file
@@ -0,0 +1,307 @@
|
||||
import { createHash } from "node:crypto"
|
||||
import dayjs from "dayjs"
|
||||
import { and, eq } from "drizzle-orm"
|
||||
import { FastifyInstance } from "fastify"
|
||||
import { ImapFlow } from "imapflow"
|
||||
import { simpleParser } from "mailparser"
|
||||
|
||||
import {
|
||||
documentImportItems,
|
||||
documentImportSources,
|
||||
documentImportStates,
|
||||
filetags,
|
||||
folders,
|
||||
} from "../../../db/schema"
|
||||
import { decrypt } from "../../utils/crypt"
|
||||
import { saveFile } from "../../utils/files"
|
||||
|
||||
type ImportSourceConnection = {
|
||||
id: string
|
||||
tenantId: number
|
||||
name: string
|
||||
provider: string
|
||||
enabled: boolean
|
||||
mailboxAddress: string
|
||||
password: string
|
||||
imapHost: string
|
||||
imapPort: number
|
||||
imapSecure: boolean
|
||||
mailboxPath: string
|
||||
targetFolderId: string | null
|
||||
defaultFiletypeId: string | null
|
||||
markAsSeen: boolean
|
||||
}
|
||||
|
||||
const activeSyncs = new Set<string>()
|
||||
|
||||
const decryptString = (value: unknown) => value ? decrypt(value as any) : ""
|
||||
|
||||
export function documentImportService(server: FastifyInstance) {
|
||||
const loadSource = async (tenantId: number, sourceId: string): Promise<ImportSourceConnection | null> => {
|
||||
const [row] = await server.db
|
||||
.select()
|
||||
.from(documentImportSources)
|
||||
.where(and(
|
||||
eq(documentImportSources.id, sourceId),
|
||||
eq(documentImportSources.tenantId, tenantId),
|
||||
))
|
||||
.limit(1)
|
||||
|
||||
if (!row) return null
|
||||
return {
|
||||
id: row.id,
|
||||
tenantId: row.tenantId,
|
||||
name: row.name,
|
||||
provider: row.provider,
|
||||
enabled: row.enabled,
|
||||
mailboxAddress: decryptString(row.mailboxAddressEncrypted),
|
||||
password: decryptString(row.passwordEncrypted),
|
||||
imapHost: decryptString(row.imapHostEncrypted),
|
||||
imapPort: row.imapPort,
|
||||
imapSecure: row.imapSecure,
|
||||
mailboxPath: row.mailboxPath,
|
||||
targetFolderId: row.targetFolderId,
|
||||
defaultFiletypeId: row.defaultFiletypeId,
|
||||
markAsSeen: row.markAsSeen,
|
||||
}
|
||||
}
|
||||
|
||||
const createImapClient = (source: ImportSourceConnection) => new ImapFlow({
|
||||
host: source.imapHost,
|
||||
port: source.imapPort,
|
||||
secure: source.imapSecure,
|
||||
auth: { user: source.mailboxAddress, pass: source.password },
|
||||
logger: false,
|
||||
})
|
||||
|
||||
const classify = async (source: ImportSourceConnection, subject = "") => {
|
||||
let folderId = source.targetFolderId
|
||||
let filetypeId = source.defaultFiletypeId
|
||||
|
||||
if (!folderId && /(Rechnung|Beleg|Invoice|Quittung)/i.test(subject)) {
|
||||
const [folder] = await server.db.select({ id: folders.id }).from(folders).where(and(
|
||||
eq(folders.tenant, source.tenantId),
|
||||
eq(folders.function, "incomingInvoices"),
|
||||
// @ts-ignore Das bestehende Schema typisiert das Jahr numerisch, verwendet es aber als Zeichenfolge.
|
||||
eq(folders.year, dayjs().format("YYYY")),
|
||||
)).limit(1)
|
||||
folderId = folder?.id || null
|
||||
|
||||
if (!filetypeId) {
|
||||
const [tag] = await server.db.select({ id: filetags.id }).from(filetags).where(and(
|
||||
eq(filetags.tenant, source.tenantId),
|
||||
eq(filetags.incomingDocumentType, "invoices"),
|
||||
)).limit(1)
|
||||
filetypeId = tag?.id || null
|
||||
}
|
||||
} else if (!filetypeId && /(Mahnung|Zahlungsaufforderung|Zahlungsverzug)/i.test(subject)) {
|
||||
const [tag] = await server.db.select({ id: filetags.id }).from(filetags).where(and(
|
||||
eq(filetags.tenant, source.tenantId),
|
||||
eq(filetags.incomingDocumentType, "reminders"),
|
||||
)).limit(1)
|
||||
filetypeId = tag?.id || null
|
||||
}
|
||||
|
||||
if (!folderId) {
|
||||
const [folder] = await server.db.select({ id: folders.id }).from(folders).where(and(
|
||||
eq(folders.tenant, source.tenantId),
|
||||
eq(folders.function, "deposit"),
|
||||
)).limit(1)
|
||||
folderId = folder?.id || null
|
||||
}
|
||||
|
||||
return { folderId, filetypeId }
|
||||
}
|
||||
|
||||
const importAttachment = async (
|
||||
source: ImportSourceConnection,
|
||||
remoteMessageId: string,
|
||||
subject: string,
|
||||
attachment: any,
|
||||
index: number,
|
||||
) => {
|
||||
const content = Buffer.from(attachment.content)
|
||||
const checksum = createHash("sha256").update(content).digest("hex")
|
||||
const attachmentKey = `${index}:${attachment.filename || "Anhang"}`
|
||||
|
||||
const [existing] = await server.db.select({ status: documentImportItems.status }).from(documentImportItems)
|
||||
.where(and(
|
||||
eq(documentImportItems.sourceId, source.id),
|
||||
eq(documentImportItems.remoteMessageId, remoteMessageId),
|
||||
eq(documentImportItems.attachmentKey, attachmentKey),
|
||||
)).limit(1)
|
||||
if (existing?.status === "imported" || existing?.status === "duplicate") return "duplicate"
|
||||
|
||||
const [sameContent] = await server.db.select({ id: documentImportItems.id }).from(documentImportItems)
|
||||
.where(and(
|
||||
eq(documentImportItems.sourceId, source.id),
|
||||
eq(documentImportItems.attachmentChecksum, checksum),
|
||||
eq(documentImportItems.status, "imported"),
|
||||
)).limit(1)
|
||||
|
||||
if (sameContent) {
|
||||
await server.db.insert(documentImportItems).values({
|
||||
tenantId: source.tenantId,
|
||||
sourceId: source.id,
|
||||
remoteMessageId,
|
||||
attachmentKey,
|
||||
attachmentChecksum: checksum,
|
||||
filename: attachment.filename || null,
|
||||
status: "duplicate",
|
||||
}).onConflictDoUpdate({
|
||||
target: [documentImportItems.sourceId, documentImportItems.remoteMessageId, documentImportItems.attachmentKey],
|
||||
set: { status: "duplicate", error: null },
|
||||
})
|
||||
return "duplicate"
|
||||
}
|
||||
|
||||
try {
|
||||
const target = await classify(source, subject)
|
||||
const saved = await saveFile(
|
||||
server,
|
||||
source.tenantId,
|
||||
remoteMessageId,
|
||||
attachment,
|
||||
target.folderId,
|
||||
target.filetypeId,
|
||||
)
|
||||
if (!saved) throw new Error("Datei konnte nicht gespeichert werden")
|
||||
|
||||
await server.db.insert(documentImportItems).values({
|
||||
tenantId: source.tenantId,
|
||||
sourceId: source.id,
|
||||
remoteMessageId,
|
||||
attachmentKey,
|
||||
attachmentChecksum: checksum,
|
||||
filename: attachment.filename || null,
|
||||
status: "imported",
|
||||
fileId: saved.id,
|
||||
}).onConflictDoUpdate({
|
||||
target: [documentImportItems.sourceId, documentImportItems.remoteMessageId, documentImportItems.attachmentKey],
|
||||
set: { status: "imported", error: null, fileId: saved.id },
|
||||
})
|
||||
return "imported"
|
||||
} catch (error: any) {
|
||||
await server.db.insert(documentImportItems).values({
|
||||
tenantId: source.tenantId,
|
||||
sourceId: source.id,
|
||||
remoteMessageId,
|
||||
attachmentKey,
|
||||
attachmentChecksum: checksum,
|
||||
filename: attachment.filename || null,
|
||||
status: "failed",
|
||||
error: error?.message || "Import fehlgeschlagen",
|
||||
}).onConflictDoUpdate({
|
||||
target: [documentImportItems.sourceId, documentImportItems.remoteMessageId, documentImportItems.attachmentKey],
|
||||
set: { status: "failed", error: error?.message || "Import fehlgeschlagen" },
|
||||
})
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
const testConnection = async (tenantId: number, sourceId: string) => {
|
||||
const source = await loadSource(tenantId, sourceId)
|
||||
if (!source) throw new Error("Importquelle wurde nicht gefunden")
|
||||
if (source.provider !== "imap") throw new Error("Dieser Provider wird noch nicht unterstützt")
|
||||
const client = createImapClient(source)
|
||||
try {
|
||||
await client.connect()
|
||||
const mailbox = await client.mailboxOpen(source.mailboxPath, { readOnly: true })
|
||||
return { success: true, mailbox: source.mailboxPath, messages: mailbox.exists }
|
||||
} finally {
|
||||
await client.logout().catch(() => client.close())
|
||||
}
|
||||
}
|
||||
|
||||
const syncSource = async (tenantId: number, sourceId: string) => {
|
||||
if (activeSyncs.has(sourceId)) throw new Error("Diese Importquelle wird bereits synchronisiert")
|
||||
const source = await loadSource(tenantId, sourceId)
|
||||
if (!source) throw new Error("Importquelle wurde nicht gefunden")
|
||||
if (!source.enabled) throw new Error("Importquelle ist deaktiviert")
|
||||
if (source.provider !== "imap") throw new Error("Dieser Provider wird noch nicht unterstützt")
|
||||
|
||||
activeSyncs.add(sourceId)
|
||||
const client = createImapClient(source)
|
||||
let imported = 0
|
||||
let duplicates = 0
|
||||
let messages = 0
|
||||
try {
|
||||
await client.connect()
|
||||
const lock = await client.getMailboxLock(source.mailboxPath)
|
||||
try {
|
||||
const opened: any = await client.mailboxOpen(source.mailboxPath)
|
||||
const uidValidity = Number(opened.uidValidity || 0)
|
||||
const [state] = await server.db.select().from(documentImportStates).where(and(
|
||||
eq(documentImportStates.sourceId, source.id),
|
||||
eq(documentImportStates.mailboxPath, source.mailboxPath),
|
||||
)).limit(1)
|
||||
const highestUid = state && Number(state.uidValidity) === uidValidity ? Number(state.highestUid) : 0
|
||||
let processedHighestUid = highestUid
|
||||
// Ungelesene Nachrichten werden immer berücksichtigt. So bleibt ein fehlgeschlagener
|
||||
// Import erneut verarbeitbar, auch wenn danach bereits neuere UIDs erfolgreich waren.
|
||||
const query: any = { seen: false }
|
||||
|
||||
for await (const message of client.fetch(query, { uid: true, envelope: true, source: true })) {
|
||||
messages += 1
|
||||
const parsed = await simpleParser(message.source)
|
||||
const remoteMessageId = `${uidValidity}:${message.uid}`
|
||||
let complete = true
|
||||
for (const [index, attachment] of (parsed.attachments || []).entries()) {
|
||||
if (attachment.contentDisposition === "inline" && !attachment.filename) continue
|
||||
try {
|
||||
const result = await importAttachment(source, remoteMessageId, parsed.subject || "", attachment, index)
|
||||
if (result === "imported") imported += 1
|
||||
else duplicates += 1
|
||||
} catch {
|
||||
complete = false
|
||||
}
|
||||
}
|
||||
if (complete && source.markAsSeen) await client.messageFlagsAdd({ uid: message.uid }, ["\\Seen"], { uid: true })
|
||||
if (complete) processedHighestUid = Math.max(processedHighestUid, Number(message.uid))
|
||||
}
|
||||
|
||||
await server.db.insert(documentImportStates).values({
|
||||
sourceId: source.id,
|
||||
mailboxPath: source.mailboxPath,
|
||||
uidValidity,
|
||||
highestUid: processedHighestUid,
|
||||
updatedAt: new Date(),
|
||||
}).onConflictDoUpdate({
|
||||
target: [documentImportStates.sourceId, documentImportStates.mailboxPath],
|
||||
set: { uidValidity, highestUid: processedHighestUid, updatedAt: new Date() },
|
||||
})
|
||||
} finally {
|
||||
lock.release()
|
||||
}
|
||||
|
||||
await server.db.update(documentImportSources).set({ lastSyncedAt: new Date(), lastError: null }).where(eq(documentImportSources.id, source.id))
|
||||
return { success: true, messages, imported, duplicates }
|
||||
} catch (error: any) {
|
||||
await server.db.update(documentImportSources).set({ lastError: error?.message || "Synchronisierung fehlgeschlagen" }).where(eq(documentImportSources.id, source.id))
|
||||
throw error
|
||||
} finally {
|
||||
activeSyncs.delete(sourceId)
|
||||
if (client.usable) await client.logout().catch(() => client.close())
|
||||
}
|
||||
}
|
||||
|
||||
const syncAll = async () => {
|
||||
const sources = await server.db.select({
|
||||
id: documentImportSources.id,
|
||||
tenantId: documentImportSources.tenantId,
|
||||
}).from(documentImportSources).where(eq(documentImportSources.enabled, true))
|
||||
|
||||
const results = []
|
||||
for (const source of sources) {
|
||||
try {
|
||||
results.push({ sourceId: source.id, ...(await syncSource(source.tenantId, source.id)) })
|
||||
} catch (error: any) {
|
||||
server.log.error({ sourceId: source.id, error: error?.message }, "Dokumentenimport fehlgeschlagen")
|
||||
results.push({ sourceId: source.id, success: false, error: error?.message || "Import fehlgeschlagen" })
|
||||
}
|
||||
}
|
||||
return results
|
||||
}
|
||||
|
||||
return { testConnection, syncSource, syncAll }
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
import { FastifyInstance } from "fastify"
|
||||
|
||||
const SYNC_INTERVAL_MS = 5 * 60 * 1000
|
||||
|
||||
export function startDocumentImportWorker(server: FastifyInstance) {
|
||||
const run = () => server.services.documentImports.syncAll().catch((error) => {
|
||||
server.log.error({ error }, "Automatischer Dokumentenimport fehlgeschlagen")
|
||||
})
|
||||
|
||||
const timer = setInterval(run, SYNC_INTERVAL_MS)
|
||||
timer.unref()
|
||||
server.addHook("onClose", async () => clearInterval(timer))
|
||||
}
|
||||
@@ -4,6 +4,7 @@ import { bankStatementService } from "../modules/cron/bankstatementsync.service"
|
||||
import {syncDokuboxService} from "../modules/cron/dokuboximport.service";
|
||||
import { FastifyInstance } from "fastify";
|
||||
import {prepareIncomingInvoices} from "../modules/cron/prepareIncomingInvoices";
|
||||
import {documentImportService} from "../modules/document-import/document-import.service";
|
||||
|
||||
declare module "fastify" {
|
||||
interface FastifyInstance {
|
||||
@@ -11,6 +12,7 @@ declare module "fastify" {
|
||||
bankStatements: ReturnType<typeof bankStatementService>;
|
||||
dokuboxSync: ReturnType<typeof syncDokuboxService>;
|
||||
prepareIncomingInvoices: ReturnType<typeof prepareIncomingInvoices>;
|
||||
documentImports: ReturnType<typeof documentImportService>;
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -20,5 +22,6 @@ export default fp(async function servicePlugin(server: FastifyInstance) {
|
||||
bankStatements: bankStatementService(server),
|
||||
dokuboxSync: syncDokuboxService(server),
|
||||
prepareIncomingInvoices: prepareIncomingInvoices(server),
|
||||
documentImports: documentImportService(server),
|
||||
});
|
||||
});
|
||||
|
||||
118
backend/src/routes/documentImports.ts
Normal file
118
backend/src/routes/documentImports.ts
Normal file
@@ -0,0 +1,118 @@
|
||||
import { and, desc, eq } from "drizzle-orm"
|
||||
import { FastifyInstance } from "fastify"
|
||||
|
||||
import { documentImportItems, documentImportSources } from "../../db/schema"
|
||||
import { decrypt, encrypt } from "../utils/crypt"
|
||||
|
||||
const decrypted = (value: unknown) => value ? decrypt(value as any) : null
|
||||
|
||||
export default async function documentImportRoutes(server: FastifyInstance) {
|
||||
const tenantId = (req: any) => {
|
||||
if (!req.user?.tenant_id) throw new Error("Kein aktiver Mandant")
|
||||
return Number(req.user.tenant_id)
|
||||
}
|
||||
|
||||
const response = (row: any) => ({
|
||||
id: row.id,
|
||||
name: row.name,
|
||||
provider: row.provider,
|
||||
enabled: row.enabled,
|
||||
mailboxAddress: decrypted(row.mailboxAddressEncrypted),
|
||||
imapHost: decrypted(row.imapHostEncrypted),
|
||||
imapPort: row.imapPort,
|
||||
imapSecure: row.imapSecure,
|
||||
mailboxPath: row.mailboxPath,
|
||||
targetFolderId: row.targetFolderId,
|
||||
defaultFiletypeId: row.defaultFiletypeId,
|
||||
markAsSeen: row.markAsSeen,
|
||||
hasPassword: Boolean(row.passwordEncrypted),
|
||||
lastSyncedAt: row.lastSyncedAt,
|
||||
lastError: row.lastError,
|
||||
createdAt: row.createdAt,
|
||||
})
|
||||
|
||||
server.get("/document-imports", async (req) => {
|
||||
const rows = await server.db.select().from(documentImportSources)
|
||||
.where(eq(documentImportSources.tenantId, tenantId(req)))
|
||||
.orderBy(documentImportSources.name)
|
||||
return rows.map(response)
|
||||
})
|
||||
|
||||
server.get("/document-imports/:id", async (req, reply) => {
|
||||
const { id } = req.params as { id: string }
|
||||
const [row] = await server.db.select().from(documentImportSources).where(and(
|
||||
eq(documentImportSources.id, id),
|
||||
eq(documentImportSources.tenantId, tenantId(req)),
|
||||
)).limit(1)
|
||||
if (!row) return reply.code(404).send({ error: "Importquelle wurde nicht gefunden" })
|
||||
return response(row)
|
||||
})
|
||||
|
||||
server.post("/document-imports/:id?", async (req, reply) => {
|
||||
const currentTenantId = tenantId(req)
|
||||
const { id } = req.params as { id?: string }
|
||||
const body = (req.body || {}) as any
|
||||
if (!body.name?.trim()) return reply.code(400).send({ error: "Name fehlt" })
|
||||
if ((body.provider || "imap") !== "imap") return reply.code(400).send({ error: "Aktuell wird nur IMAP unterstützt" })
|
||||
if (!body.mailboxAddress || !body.imapHost) return reply.code(400).send({ error: "Postfachadresse und IMAP-Host sind erforderlich" })
|
||||
|
||||
const values: any = {
|
||||
name: body.name.trim(),
|
||||
provider: "imap",
|
||||
enabled: body.enabled !== false,
|
||||
mailboxAddressEncrypted: encrypt(body.mailboxAddress),
|
||||
imapHostEncrypted: encrypt(body.imapHost),
|
||||
imapPort: Number(body.imapPort || 993),
|
||||
imapSecure: body.imapSecure !== false,
|
||||
mailboxPath: body.mailboxPath?.trim() || "INBOX",
|
||||
targetFolderId: body.targetFolderId || null,
|
||||
defaultFiletypeId: body.defaultFiletypeId || null,
|
||||
markAsSeen: body.markAsSeen !== false,
|
||||
updatedAt: new Date(),
|
||||
}
|
||||
if (body.password) values.passwordEncrypted = encrypt(body.password)
|
||||
|
||||
if (id) {
|
||||
const [existing] = await server.db.select({ id: documentImportSources.id }).from(documentImportSources).where(and(
|
||||
eq(documentImportSources.id, id),
|
||||
eq(documentImportSources.tenantId, currentTenantId),
|
||||
)).limit(1)
|
||||
if (!existing) return reply.code(404).send({ error: "Importquelle wurde nicht gefunden" })
|
||||
await server.db.update(documentImportSources).set(values).where(eq(documentImportSources.id, id))
|
||||
return { success: true, id }
|
||||
}
|
||||
|
||||
if (!body.password) return reply.code(400).send({ error: "Passwort fehlt" })
|
||||
const [created] = await server.db.insert(documentImportSources).values({
|
||||
...values,
|
||||
tenantId: currentTenantId,
|
||||
createdBy: req.user.user_id,
|
||||
}).returning({ id: documentImportSources.id })
|
||||
return { success: true, id: created.id }
|
||||
})
|
||||
|
||||
server.post("/document-imports/:id/test", async (req, reply) => {
|
||||
try {
|
||||
return await server.services.documentImports.testConnection(tenantId(req), (req.params as { id: string }).id)
|
||||
} catch (error: any) {
|
||||
return reply.code(400).send({ error: error?.message || "Verbindung fehlgeschlagen" })
|
||||
}
|
||||
})
|
||||
|
||||
server.post("/document-imports/:id/sync", async (req, reply) => {
|
||||
try {
|
||||
return await server.services.documentImports.syncSource(tenantId(req), (req.params as { id: string }).id)
|
||||
} catch (error: any) {
|
||||
return reply.code(400).send({ error: error?.message || "Synchronisierung fehlgeschlagen" })
|
||||
}
|
||||
})
|
||||
|
||||
server.get("/document-imports/:id/items", async (req) => {
|
||||
const currentTenantId = tenantId(req)
|
||||
const { id } = req.params as { id: string }
|
||||
return server.db.select().from(documentImportItems).where(and(
|
||||
eq(documentImportItems.tenantId, currentTenantId),
|
||||
eq(documentImportItems.sourceId, id),
|
||||
)).orderBy(desc(documentImportItems.createdAt)).limit(100)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user