import Imap from "imap"; import { simpleParser, ParsedMail, Attachment } from "mailparser"; import { getImportSettingsByUser, createSourceFile, updateSourceFile, getUserSettings, getSourceFileByContentHash, findDuplicateInvoice, isInvoiceBlacklisted, createInvoice, createImportLog, } from "./db"; import { extractInvoicesWithMistral, generateMetadataJSON } from "./invoiceExtractor"; import { localStorageDelete, localStoragePut, generateStorageKey } from "./localStorage"; import { calculateFileSha256 } from "./fileFingerprint"; import { sendImportNotification } from "./notificationService"; import { getOffice365ImapToken, buildXOAuth2String } from "./office365OAuth"; import { applyAutomationRules } from "./automationEngine"; interface EmailImportConfig { userId: number; emailAddress: string; password: string; host: string; port: number; sinceDate?: number; // Unix timestamp (s) — ne pas lire les emails antérieurs à cette date authMode?: "basic" | "oauth2"; // Mode d'authentification IMAP // Credentials Azure AD pour OAuth2 azureTenantId?: string; azureClientId?: string; azureClientSecret?: string; } // Store active intervals for each user const activeIntervals = new Map(); // Une extraction IA peut dépasser la fréquence configurée : ce verrou évite // qu'un second cycle IMAP traite les mêmes messages avant la fin du premier. const activeChecks = new Map>(); function runEmailCheckExclusive(config: EmailImportConfig): Promise { const runningCheck = activeChecks.get(config.userId); if (runningCheck) { console.log(`[EmailImport] Vérification déjà en cours pour user ${config.userId}, cycle ignoré`); return runningCheck; } const check = checkEmailsForPDFs(config).finally(() => { if (activeChecks.get(config.userId) === check) activeChecks.delete(config.userId); }); activeChecks.set(config.userId, check); return check; } /** * Process a single email attachment (PDF) * Replicates the same logic as manual upload */ async function processEmailAttachment( userId: number, attachment: Attachment, emailSubject: string ): Promise<{ success: boolean; totalInvoices: number; imported: number; duplicates: number; errors: number; quotaError?: boolean }> { const fileName = attachment.filename || `email-attachment-${Date.now()}.pdf`; console.log(`[EmailImport] Processing attachment: ${fileName} from email: ${emailSubject}`); try { // Convert attachment content to Buffer const fileBuffer = attachment.content; console.log(`[EmailImport] File size: ${fileBuffer.length} bytes`); const contentHash = calculateFileSha256(fileBuffer); const existingSource = await getSourceFileByContentHash(contentHash); if (existingSource) { console.log( `[EmailImport] PDF déjà importé, extraction ignorée: ${fileName} -> source ${existingSource.id}`, ); return { success: true, totalInvoices: Math.max(existingSource.totalInvoicesDetected, 1), imported: 0, duplicates: Math.max(existingSource.totalInvoicesDetected, 1), errors: 0, }; } // Store source file const sourceFileKey = generateStorageKey(userId, fileName); console.log(`[EmailImport] Generated storage key: ${sourceFileKey}`); let sourceFileUrl: string; try { const result = await localStoragePut(sourceFileKey, fileBuffer, "application/pdf"); sourceFileUrl = result.url; console.log(`[EmailImport] File stored successfully at: ${sourceFileUrl}`); } catch (error) { console.error(`[EmailImport] FAILED to store file:`, error); throw new Error("Failed to store PDF file"); } // Create source file record let sourceFile; try { sourceFile = await createSourceFile({ userId, fileName, fileKey: sourceFileKey, fileUrl: sourceFileUrl, contentHash, processingStatus: "processing", }); } catch (error: any) { // La contrainte unique protège également contre deux imports concurrents. if (error?.code === "ER_DUP_ENTRY" || error?.errno === 1062) { await localStorageDelete(sourceFileKey).catch(() => undefined); console.log(`[EmailImport] PDF réservé par un autre traitement: ${fileName}`); return { success: true, totalInvoices: 1, imported: 0, duplicates: 1, errors: 0 }; } throw error; } console.log(`[EmailImport] Source file record created with ID: ${sourceFile.id}`); // Get user settings for custom keywords const settings = await getUserSettings(userId); const customKeywords = settings ? { invoiceNumber: settings.invoiceNumberKeywords, deliveryNote: settings.deliveryNoteKeywords, orderNumber: settings.orderNumberKeywords, supplier: settings.supplierKeywords, totalAmount: settings.totalAmountKeywords, subscription: settings.subscriptionKeywords, recipient: settings.recipientKeywords, } : undefined; const model = settings?.llmModel || "mistral-large-latest"; const aiSettings = settings ? { aiProvider: settings.aiProvider, mistralApiKey: settings.mistralApiKey, manusForgeApiKey: settings.manusForgeApiKey, manusForgeApiUrl: settings.manusForgeApiUrl, geminiApiKey: (settings as any).geminiApiKey || null, } : undefined; // Extract invoices console.log(`[EmailImport] Starting invoice extraction (provider: ${aiSettings?.aiProvider || 'mistral'})...`); const result = await extractInvoicesWithMistral( fileBuffer, userId, sourceFile.id, model, customKeywords, aiSettings ); console.log(`[EmailImport] Extraction complete: ${result.invoiceCount} invoice(s) detected`); // Update source file with total count await updateSourceFile(sourceFile.id, { totalInvoicesDetected: result.invoiceCount, processingProgress: `Extraction ${result.invoiceCount} facture(s) détectée(s)`, }); // Process each invoice let importedCount = 0; let duplicatesCount = 0; let errorsCount = 0; const duplicateDetails: any[] = []; const errorDetails: any[] = []; for (let i = 0; i < result.invoices.length; i++) { const invoiceData = result.invoices[i]!; try { // Update progress await updateSourceFile(sourceFile.id, { processingProgress: `Extraction ${i + 1}/${result.invoiceCount} factures...`, }); // Vérifier la blacklist (factures supprimées manuellement) const blacklisted = await isInvoiceBlacklisted(invoiceData.invoiceNumber, userId); if (blacklisted) { duplicatesCount++; duplicateDetails.push({ supplierName: invoiceData.supplierName, invoiceNumber: invoiceData.invoiceNumber, totalAmount: invoiceData.totalAmount, reason: 'blacklisted', }); continue; } // Check for duplicates (numéro de facture + montant) const duplicate = await findDuplicateInvoice( invoiceData.invoiceNumber, invoiceData.totalAmount?.toString() ?? null, userId ); if (duplicate) { duplicatesCount++; duplicateDetails.push({ supplierName: invoiceData.supplierName, invoiceNumber: invoiceData.invoiceNumber, totalAmount: invoiceData.totalAmount, }); continue; } // Generate metadata JSON const metadataJson = generateMetadataJSON(invoiceData); const metadataKey = generateStorageKey(userId, `${fileName}-${i + 1}-metadata.json`); const { url: metadataUrl } = await localStoragePut( metadataKey, Buffer.from(metadataJson), "application/json" ); // Create invoice record const newInvoice = await createInvoice({ userId, sourceFileId: sourceFile.id, invoiceIndexInFile: i + 1, fileName: `${fileName} - Facture ${i + 1}`, fileKey: sourceFileKey, fileUrl: sourceFileUrl, supplierName: invoiceData.supplierName, invoiceNumber: invoiceData.invoiceNumber, invoiceDate: invoiceData.invoiceDate, deliveryNoteNumber: invoiceData.deliveryNoteNumber, orderNumber: invoiceData.orderNumber, totalAmount: invoiceData.totalAmount?.toString(), recipientName: invoiceData.recipientName, pageRange: invoiceData.pageRange, qualityScore: invoiceData.qualityScore, extractedText: invoiceData.extractedText, metadataFileKey: metadataKey, metadataFileUrl: metadataUrl, status: "completed", }); // Apply automation rules (ventilation, typeAchat, serviceConcerne) try { const { updateInvoice } = await import("./db"); const automationUpdates = await applyAutomationRules(userId, newInvoice); if (Object.keys(automationUpdates).length > 0) { await updateInvoice(newInvoice.id, automationUpdates); console.log(`[EmailImport] Automation rules applied to invoice ${newInvoice.id}: ${Object.keys(automationUpdates).join(', ')}`); } } catch (autoError: any) { console.error(`[EmailImport] Error applying automation rules:`, autoError.message); } importedCount++; } catch (error: any) { errorsCount++; errorDetails.push({ invoiceIndex: i + 1, error: error.message, }); } } // Update source file status await updateSourceFile(sourceFile.id, { processingStatus: "completed", processingProgress: `Terminé: ${importedCount} importée(s), ${duplicatesCount} doublon(s)`, }); // Détecter si des erreurs de quota ont eu lieu const quotaErrors = errorDetails.filter(e => e.error && ( e.error.includes('usage exhausted') || e.error.includes('quota') || e.error.includes('rate limit') || e.error.includes('Precondition Failed') || e.error.includes('429') ) ); const warningMessage = quotaErrors.length > 0 ? `Quota IA épuisé : ${quotaErrors.length} facture(s) non extraite(s). Vérifiez votre quota Manus ou configurez une clé API externe dans Paramétrage → IA.` : null; // Create import log await createImportLog({ userId, sourceFileId: sourceFile.id, fileName, totalInvoicesDetected: result.invoiceCount, invoicesImported: importedCount, duplicatesIgnored: duplicatesCount, errors: errorsCount, duplicateDetails: duplicateDetails.length > 0 ? JSON.stringify(duplicateDetails) : null, errorDetails: errorDetails.length > 0 ? JSON.stringify(errorDetails) : null, warningMessage, importSource: "email", }); console.log(`[EmailImport] Successfully processed attachment: ${fileName}`); console.log(`[EmailImport] Results: ${importedCount} imported, ${duplicatesCount} duplicates, ${errorsCount} errors`); return { success: true, totalInvoices: result.invoiceCount, imported: importedCount, duplicates: duplicatesCount, errors: errorsCount, }; } catch (error: any) { console.error(`[EmailImport] Error processing attachment ${attachment.filename}:`, error); // Détecter l'erreur de quota IA const isQuotaError = error?.message && ( error.message.includes('usage exhausted') || error.message.includes('quota') || error.message.includes('429') || error.message.includes('rate limit') || error.message.includes('Precondition Failed') ); return { success: false, totalInvoices: 0, imported: 0, duplicates: 0, errors: 1, quotaError: isQuotaError as boolean, }; } } /** * Construit la configuration IMAP en fonction du mode d'authentification. * - basic : login/password classique * - oauth2 : obtient un token Azure AD et utilise XOAUTH2 */ async function buildImapConfig(config: EmailImportConfig): Promise { if (config.authMode === "oauth2") { if (!config.azureTenantId || !config.azureClientId || !config.azureClientSecret) { throw new Error( "OAuth2 IMAP : credentials Azure AD incomplets (Tenant ID, Client ID, Client Secret requis)" ); } console.log(`[EmailImport] Obtaining OAuth2 token for ${config.emailAddress}...`); const accessToken = await getOffice365ImapToken( config.azureTenantId, config.azureClientId, config.azureClientSecret ); const xoauth2 = buildXOAuth2String(config.emailAddress, accessToken); console.log(`[EmailImport] OAuth2 token obtained successfully`); return { user: config.emailAddress, xoauth2, host: config.host, port: config.port, tls: true, tlsOptions: { rejectUnauthorized: false }, authTimeout: 30000, } as any; } // Basic auth (par défaut) return { user: config.emailAddress, password: config.password, host: config.host, port: config.port, tls: true, tlsOptions: { rejectUnauthorized: false }, authTimeout: 30000, }; } /** * Connect to IMAP and process unread emails with PDF attachments */ async function checkEmailsForPDFs(config: EmailImportConfig): Promise { // Build IMAP config (may involve async OAuth2 token fetch) const imapConfig = await buildImapConfig(config); return new Promise((resolve, reject) => { const imap = new Imap(imapConfig); function openInbox(cb: (err: Error | null, box?: any) => void) { imap.openBox("INBOX", false, cb); } imap.once("ready", () => { console.log(`[EmailImport] Connected to IMAP server for user ${config.userId} (mode: ${config.authMode || "basic"})`); openInbox((err) => { if (err) { console.error("[EmailImport] Error opening inbox:", err); imap.end(); reject(err); return; } // Build search criteria: unread emails, optionally filtered by date const searchCriteria: any[] = ["UNSEEN"]; if (config.sinceDate) { // IMAP SINCE expects a date string like "1-Jan-2026" const since = new Date(config.sinceDate * 1000); const months = ["Jan","Feb","Mar","Apr","May","Jun","Jul","Aug","Sep","Oct","Nov","Dec"]; const sinceStr = `${since.getDate()}-${months[since.getMonth()]}-${since.getFullYear()}`; searchCriteria.push(["SINCE", sinceStr]); console.log(`[EmailImport] Filtering emails since ${sinceStr} for user ${config.userId}`); } imap.search(searchCriteria, (err, results) => { if (err) { console.error("[EmailImport] Error searching emails:", err); imap.end(); reject(err); return; } if (!results || results.length === 0) { console.log(`[EmailImport] No unread emails found for user ${config.userId}`); imap.end(); resolve(); return; } console.log(`[EmailImport] Found ${results.length} unread emails for user ${config.userId}`); const fetch = imap.fetch(results, { bodies: "", markSeen: false, // Don't mark as seen yet }); const processedEmailUids: number[] = []; const messageTasks: Promise[] = []; fetch.on("message", (msg, seqno) => { const uidPromise = new Promise((resolveUid) => { msg.once("attributes", (attributes) => resolveUid(attributes.uid)); }); msg.on("body", (stream) => { const task = (async () => { try { const parsed: ParsedMail = await simpleParser(stream as any); const pdfAttachments = parsed.attachments.filter( (attachment) => attachment.contentType === "application/pdf" || attachment.filename?.toLowerCase().endsWith(".pdf"), ); if (pdfAttachments.length === 0) return; console.log( `[EmailImport] Email ${seqno} has ${pdfAttachments.length} PDF attachment(s)`, ); let allAttachmentsSucceeded = true; for (const attachment of pdfAttachments) { try { const result = await processEmailAttachment( config.userId, attachment, parsed.subject || "No subject", ); allAttachmentsSucceeded = allAttachmentsSucceeded && result.success; if (result.success) { await sendImportNotification(config.userId, { source: "email", fileName: attachment.filename || "email-attachment.pdf", totalInvoices: result.totalInvoices, imported: result.imported, duplicates: result.duplicates, errors: result.errors, }); } } catch (error) { allAttachmentsSucceeded = false; console.error( `[EmailImport] Failed to process attachment from email ${seqno}:`, error, ); } } if (allAttachmentsSucceeded) { const uid = await uidPromise; if (!processedEmailUids.includes(uid)) processedEmailUids.push(uid); } } catch (error) { console.error(`[EmailImport] Error parsing email ${seqno}:`, error); } })(); messageTasks.push(task); }); }); fetch.once("error", (err) => { console.error("[EmailImport] Fetch error:", err); imap.end(); reject(err); }); fetch.once("end", async () => { console.log(`[EmailImport] Finished fetching emails for user ${config.userId}`); // Le flux IMAP peut se terminer avant les traitements IA asynchrones. // On attend explicitement chaque message avant de le marquer comme lu. await Promise.allSettled(messageTasks); if (processedEmailUids.length > 0) { imap.addFlags(processedEmailUids, ["\\Seen"], (err) => { if (err) { console.error("[EmailImport] Error marking emails as seen:", err); } else { console.log(`[EmailImport] Marked ${processedEmailUids.length} emails as seen`); } imap.end(); resolve(); }); } else { imap.end(); resolve(); } }); }); }); }); imap.once("error", (err) => { console.error("[EmailImport] IMAP connection error:", err); reject(err); }); imap.once("end", () => { console.log(`[EmailImport] IMAP connection ended for user ${config.userId}`); }); imap.connect(); }); } /** * Test IMAP connection for a user (without processing emails) * Returns detailed error message if connection fails */ export async function testImapConnection(config: EmailImportConfig): Promise<{ success: boolean; message: string }> { try { const imapConfig = await buildImapConfig(config); return new Promise((resolve) => { const imap = new Imap(imapConfig); let resolved = false; const done = (result: { success: boolean; message: string }) => { if (!resolved) { resolved = true; try { imap.destroy(); } catch {} resolve(result); } }; imap.once("ready", () => { console.log(`[EmailImport] Test connection successful for ${config.emailAddress}`); done({ success: true, message: `Connexion IMAP réussie pour ${config.emailAddress}` }); }); imap.once("error", (err: any) => { console.error(`[EmailImport] Test connection failed:`, err); let message = `Erreur de connexion IMAP : ${err.message || err}`; // Messages d'erreur plus clairs if (err.message?.includes("Invalid credentials") || err.message?.includes("AUTHENTICATE")) { if (config.authMode === "oauth2") { message = "Authentification OAuth2 refusée. Vérifiez que l'application Azure AD a bien la permission IMAP.AccessAsApp et que le consentement admin a été accordé."; } else { message = "Identifiants invalides. Pour Office 365, l'authentification basique est désactivée. Activez le mode OAuth2 et configurez les credentials Azure AD."; } } else if (err.message?.includes("ECONNREFUSED") || err.message?.includes("ENOTFOUND")) { message = `Impossible de se connecter au serveur ${config.host}:${config.port}. Vérifiez l'adresse et le port IMAP.`; } else if (err.message?.includes("certificate") || err.message?.includes("SSL")) { message = `Erreur SSL/TLS lors de la connexion à ${config.host}. Vérifiez le port (993 pour SSL).`; } else if (err.message?.includes("timeout") || err.message?.includes("Timeout")) { message = `Timeout de connexion à ${config.host}:${config.port}. Vérifiez l'adresse du serveur IMAP.`; } done({ success: false, message }); }); // Timeout de sécurité setTimeout(() => { done({ success: false, message: `Timeout : impossible de se connecter à ${config.host}:${config.port} dans les 15 secondes.` }); }, 15000); imap.connect(); }); } catch (error: any) { return { success: false, message: `Erreur : ${error.message || error}` }; } } /** * Start email import service for a user */ export async function startEmailImportService(userId: number): Promise { try { // Get user's import settings const settings = await getImportSettingsByUser(userId); if (!settings || settings.emailImportEnabled !== 1) { console.log(`[EmailImport] Email import not enabled for user ${userId}`); return false; } if (!settings.emailImportAddress || !settings.emailImportHost) { console.log(`[EmailImport] Email import configuration incomplete for user ${userId}`); return false; } const authMode = (settings as any).emailImportAuthMode as "basic" | "oauth2" || "basic"; // Vérification selon le mode d'auth if (authMode === "basic" && !settings.emailImportPassword) { console.log(`[EmailImport] Email import password missing for user ${userId}`); return false; } if (authMode === "oauth2" && (!settings.azureTenantId || !settings.azureClientId || !settings.azureClientSecret)) { console.log(`[EmailImport] OAuth2 credentials incomplete for user ${userId}`); return false; } // Stop existing service if running stopEmailImportService(userId); const config: EmailImportConfig = { userId, emailAddress: settings.emailImportAddress, password: settings.emailImportPassword || "", host: settings.emailImportHost, port: settings.emailImportPort || 993, sinceDate: settings.emailImportSinceDate ?? undefined, authMode, azureTenantId: settings.azureTenantId || undefined, azureClientId: settings.azureClientId || undefined, azureClientSecret: settings.azureClientSecret || undefined, }; const frequencyMs = (settings.emailImportFrequency || 30) * 60 * 1000; // Convert minutes to milliseconds console.log( `[EmailImport] Starting email import service for user ${userId} with frequency ${settings.emailImportFrequency} minutes (auth: ${authMode})` ); // Run immediately on start runEmailCheckExclusive(config).catch((error) => { console.error(`[EmailImport] Error checking emails for user ${userId}:`, error); }); // Set up interval for periodic checks const interval = setInterval(() => { runEmailCheckExclusive(config).catch((error) => { console.error(`[EmailImport] Error checking emails for user ${userId}:`, error); }); }, frequencyMs); activeIntervals.set(userId, interval); console.log(`[EmailImport] Email import service started for user ${userId}`); return true; } catch (error) { console.error(`[EmailImport] Error starting email import service for user ${userId}:`, error); return false; } } /** * Stop email import service for a user */ export function stopEmailImportService(userId: number): void { const interval = activeIntervals.get(userId); if (interval) { clearInterval(interval); activeIntervals.delete(userId); console.log(`[EmailImport] Email import service stopped for user ${userId}`); } } /** * Check if email import service is running for a user */ export function isEmailImportServiceRunning(userId: number): boolean { return activeIntervals.has(userId); } /** * Manually trigger an immediate email check for a user */ export async function triggerEmailCheck(userId: number): Promise<{ success: boolean; message: string }> { try { // Get user's import settings const settings = await getImportSettingsByUser(userId); if (!settings || settings.emailImportEnabled !== 1) { return { success: false, message: "Import par email non activé" }; } if (!settings.emailImportAddress || !settings.emailImportHost) { return { success: false, message: "Configuration IMAP incomplète" }; } const authMode = (settings as any).emailImportAuthMode as "basic" | "oauth2" || "basic"; const config: EmailImportConfig = { userId, emailAddress: settings.emailImportAddress, password: settings.emailImportPassword || "", host: settings.emailImportHost, port: settings.emailImportPort || 993, sinceDate: settings.emailImportSinceDate ?? undefined, authMode, azureTenantId: settings.azureTenantId || undefined, azureClientId: settings.azureClientId || undefined, azureClientSecret: settings.azureClientSecret || undefined, }; console.log(`[EmailImport] Manual check triggered for user ${userId}`); await runEmailCheckExclusive(config); return { success: true, message: "Vérification terminée avec succès" }; } catch (error: any) { console.error(`[EmailImport] Error during manual check for user ${userId}:`, error); return { success: false, message: `Erreur: ${error.message}` }; } } /** * Stop all email import services */ export function stopAllEmailImportServices(): void { activeIntervals.forEach((interval, userId) => { clearInterval(interval); console.log(`[EmailImport] Stopped email import service for user ${userId}`); }); activeIntervals.clear(); }