/** * AI lead assessment ("KI-Analyse") — collects EVERYTHING the ERP knows about one lead * into a dossier, sends it to Claude and stores a structured assessment: * - AD_User record (status, source, notes), C_ContactActivity log * - e-mail history from the mailpiler archive INCLUDING the bodies (sent + received) * - offer / contract / forecast / questionnaire conditions (offer_conditions JSON), * signing / confirmation links, Strapi attachments (file names) * - call recordings (FreePBX CDR) + whisper transcripts (cached; optionally the * missing ones are transcribed first) * Jobs run in the background (the edge nginx cuts requests at 60 s). State and the * last result are persisted in data/lead-assessments/.json so that a poll may * land on any PM2 worker and the last assessment reopens instantly. The iDempiere * reads happen BEFORE the request answers (token refresh needs the response headers); * the background part only talks to the archive, the PBX and Claude. * Config: runtimeConfig.anthropic (ANTHROPIC_API_KEY, ANTHROPIC_MODEL). Staff only. */ import Anthropic from '@anthropic-ai/sdk' import { existsSync } from 'node:fs' import { mkdir, readFile, writeFile } from 'node:fs/promises' import { join } from 'node:path' import { string } from 'alga-js' import fetchHelper from '../fetchHelper' import refreshTokenHelper from '../refreshTokenHelper' import strapiHelper from '../strapiHelper' import { resolveTableId } from '../inbox/strapiAttach' import { readOfferConditions } from '../offers/offerConditions' import { writeLeadQualification } from './leadQualification' import { listSigningRequestsForRecord, effectiveStatus, isSigningAvailable } from '../offers/contractSigningDb' import { archiveAvailable, searchArchive, fetchArchiveMessage } from '../mail/mailArchive' import { isValidEmail } from '../mail/leadMailbox' import { findRecordings, statRecordings, readTranscript, hasTranscript, pbxAvailable, digitsOf, startTranscription, waitForTranscription } from '../telephony/recordings' // ---- limits (keep the dossier well inside the context window, ~150k tokens max) ---- const MAX_MAILS = 80 // newest mails whose bodies are read const MAX_MAIL_LIST = 400 // mails listed by subject only (beyond the bodies) const MAIL_BODY_CHARS = 7000 const MAIL_CONCURRENCY = 4 const MAX_TRANSCRIPTS = 20 const TRANSCRIPT_CHARS = 25000 const MAX_TRANSCRIBE_MISSING = 3 // newest untranscribed recordings transcribed on request const TRANSCRIBE_BUDGET_MS = 6 * 60000 // total wall-clock for the on-demand transcriptions const MAX_ACTIVITIES = 300 const SECTION_JSON_CHARS = 40000 const JOB_STALE_MS = 20 * 60000 // a pending job older than this is considered dead const DEFAULT_MODEL = 'claude-opus-5-5' export type StepState = 'pending' | 'running' | 'done' | 'skipped' | 'error' export interface AssessmentStep { key: string; label: string; state: StepState; count?: number; ms?: number; message?: string } export interface AssessmentJob { leadId: number state: 'pending' | 'done' | 'failed' startedAt: string finishedAt?: string startedBy?: number language: string transcribeMissing: boolean lead?: { name: string; email: string } steps: AssessmentStep[] sources?: Record warnings: string[] model?: string usage?: { input: number; output: number; cacheRead: number; cacheWrite: number } result?: any // parsed assessment JSON (draftMessage may carry user edits + sentAt / sentTo) rawText?: string // model text when the JSON could not be parsed error?: string promptChars?: number } export const cfg = () => { const c: any = (useRuntimeConfig() as any).anthropic || {} return { apiKey: String(c.apiKey || ''), workspaceId: String(c.workspaceId || ''), model: String(c.model || DEFAULT_MODEL), chatMaxTokens: Number(c.chatMaxTokens) || 0 } } export const assessmentAvailable = (): boolean => !!cfg().apiKey // ---- persistence ------------------------------------------------------------------ const dir = () => join(process.cwd(), 'data', 'lead-assessments') const filePath = (leadId: number) => join(dir(), `${leadId}.json`) export const readAssessmentJob = async (leadId: number): Promise => { try { const p = filePath(leadId) if (!existsSync(p)) return null const j = JSON.parse(await readFile(p, 'utf8')) if (j?.state === 'pending' && Date.now() - new Date(j.startedAt).getTime() > JOB_STALE_MS) { j.state = 'failed'; j.error = 'Analyse abgebrochen (Zeitüberschreitung)'; j.finishedAt = new Date().toISOString() } return j } catch { return null } } export const saveAssessmentJob = async (job: AssessmentJob) => saveJob(job) const saveJob = async (job: AssessmentJob) => { // A job owned by another store (the lead chat console keeps its context job inside // data/lead-chats/.json) persists through its own hook; functions are skipped by // JSON.stringify, so the hook never lands in a file. const persist = (job as any).__persist if (typeof persist === 'function') { try { await persist() } catch (err: any) { console.warn('[lead-assessment] persist hook failed:', err?.message || err) } return } try { if (!existsSync(dir())) await mkdir(dir(), { recursive: true }) await writeFile(filePath(job.leadId), JSON.stringify(job), 'utf8') } catch (err: any) { console.warn('[lead-assessment] persist failed:', err?.message || err) } } const STEP_LABELS: Record = { lead: 'Lead-Datensatz', activities: 'Aktivitäten', offers: 'Angebote / Vertrag', attachments: 'Anhänge', mails: 'E-Mails (Mailpiler)', calls: 'Anrufe / Transkripte', transcribe: 'Transkription', ai: 'KI-Analyse' } /** Fresh step list; `without` drops steps a caller does not run (the chat console has no 'ai' step). */ export const newSteps = (without: string[] = []): AssessmentStep[] => Object.keys(STEP_LABELS).filter((k) => !without.includes(k)).map((key) => ({ key, label: STEP_LABELS[key], state: 'pending' })) const newJob = (leadId: number, language: string, transcribeMissing: boolean, userId: number): AssessmentJob => ({ leadId, state: 'pending', startedAt: new Date().toISOString(), startedBy: userId, language, transcribeMissing, steps: newSteps(), warnings: [] }) const step = (job: AssessmentJob, key: string) => job.steps.find((s) => s.key === key)! const setStep = async (job: AssessmentJob, key: string, patch: Partial) => { Object.assign(step(job, key), patch); await saveJob(job) } // ---- helpers ---------------------------------------------------------------------- export const val = (r: any, ...keys: string[]) => { for (const k of keys) { const v = r?.[k]; if (v !== undefined && v !== null && v !== '') return v } return '' } const ident = (v: any) => (v && typeof v === 'object') ? String(v.identifier ?? v.id ?? '') : String(v ?? '') const day = (v: any) => { const s = String(v || ''); return s ? s.slice(0, 10) : '' } const clip = (s: any, n: number) => { const t = String(s ?? ''); return t.length > n ? t.slice(0, n) + `\n[… gekürzt, ${t.length - n} Zeichen weggelassen]` : t } const htmlToText = (html: string): string => String(html || '') .replace(//gi, ' ').replace(//gi, ' ') .replace(//gi, '\n').replace(/<\/(p|div|tr|li|h[1-6]|blockquote)>/gi, '\n') .replace(/<[^>]+>/g, ' ').replace(/ /g, ' ').replace(/&/g, '&').replace(/</g, '<').replace(/>/g, '>').replace(/"/g, '"').replace(/'/g, "'") .replace(/[ \t]+/g, ' ').replace(/\n\s*\n\s*\n+/g, '\n\n').trim() /** Strip quoted replies so a thread isn't included five times over. */ const stripQuotes = (text: string): string => { const lines = String(text || '').split('\n') const cut = lines.findIndex((l, i) => i > 3 && (/^(Am .{6,80} schrieb .*:|On .{6,80} wrote:|-----\s*(Original|Ursprüngliche) (Message|Nachricht)\s*-----|Von: .*|From: .*)$/i.test(l.trim()) && lines.slice(i, i + 4).some((x) => /^(Gesendet|Sent|An|To|Betreff|Subject):/i.test(x.trim()) || /^>/.test(x)))) const body = (cut > 0 ? lines.slice(0, cut) : lines).filter((l) => !/^>/.test(l)) return body.join('\n').trim() } const limitRun = async (items: T[], n: number, fn: (it: T) => Promise) => { let i = 0 await Promise.all(Array.from({ length: Math.min(n, items.length) }, async () => { while (i < items.length) await fn(items[i++]) })) } // ---- 1) iDempiere part (synchronous, before the response) -------------------------- export interface ErpContext { lead: any activities: any[] conditions: any signing: any[] attachments: { name: string; date: string; size: number }[] } export const collectErpContext = async (event: any, token: string, leadId: number, job: AssessmentJob): Promise => { const get = async (url: string) => { try { return await fetchHelper(event, url, 'GET', token, null) } catch (err: any) { const s = Number(err?.statusCode ?? err?.status ?? err?.response?.status) if (s !== 401 && s !== 403) throw err const fresh = await refreshTokenHelper(event); token = fresh return await fetchHelper(event, url, 'GET', fresh, null) } } const out: ErpContext = { lead: null, activities: [], conditions: {}, signing: [], attachments: [] } let t0 = Date.now(); await setStep(job, 'lead', { state: 'running' }) out.lead = await get(`models/ad_user/${leadId}`) if (!out.lead?.id) throw new Error('Lead not found') await setStep(job, 'lead', { state: 'done', ms: Date.now() - t0, count: 1 }) t0 = Date.now(); await setStep(job, 'activities', { state: 'running' }) try { const res: any = await get(`models/c_contactactivity?$filter=${string.urlEncode('AD_User_ID eq ' + leadId)}&$orderby=${string.urlEncode('StartDate desc')}&$top=${MAX_ACTIVITIES}`) out.activities = res?.records || [] await setStep(job, 'activities', { state: 'done', ms: Date.now() - t0, count: out.activities.length }) } catch (err: any) { job.warnings.push('Aktivitäten: ' + (err?.message || err)); await setStep(job, 'activities', { state: 'error', message: err?.message || String(err) }) } t0 = Date.now(); await setStep(job, 'offers', { state: 'running' }) try { out.conditions = await readOfferConditions(event, token, 'lead', leadId) if (isSigningAvailable()) { out.signing = listSigningRequestsForRecord('lead', leadId, 20).map((r) => ({ kind: r.kind, state: effectiveStatus(r), language: r.language, createdAt: r.createdAt, expiresAt: r.expiresAt, signedAt: r.signedAt, renewalRequestedAt: r.renewalRequestedAt, lastOpenedAt: r.lastOpenedAt, signer: r.signer ? { name: r.signer.name, company: r.signer.company } : null, documents: (r.documents || []).map((d: any) => d.title || d.filename || d.key) })) } const n = Object.keys(out.conditions || {}).length + out.signing.length await setStep(job, 'offers', { state: 'done', ms: Date.now() - t0, count: n }) } catch (err: any) { job.warnings.push('Angebote: ' + (err?.message || err)); await setStep(job, 'offers', { state: 'error', message: err?.message || String(err) }) } t0 = Date.now(); await setStep(job, 'attachments', { state: 'running' }) try { const tableId = await resolveTableId(event, token, 'Lead_User') const existing: any = await strapiHelper(event, `ad-attachments?filters[AD_Table_ID][$eq]=${tableId}&filters[Record_ID][$eq]=${leadId}`, 'GET', null) const attId = existing?.data?.[0]?.documentId || existing?.data?.[0]?.id if (attId) { const res: any = await strapiHelper(event, `ad-attachments/${attId}?populate=attachment`, 'GET', null) const files = Array.isArray(res?.data?.attachment) ? res.data.attachment : [] out.attachments = files.map((f: any) => ({ name: String(f?.name || ''), date: day(f?.createdAt), size: Number(f?.size || 0) })) } await setStep(job, 'attachments', { state: 'done', ms: Date.now() - t0, count: out.attachments.length }) } catch (err: any) { job.warnings.push('Anhänge: ' + (err?.message || err)); await setStep(job, 'attachments', { state: 'error', message: err?.message || String(err) }) } return out } // ---- 2) archive + PBX part (background) -------------------------------------------- interface MailDossierItem { date: string; direction: 'in' | 'out'; from: string; to: string; subject: string; body?: string; attachments?: string[] } export const collectMails = async (email: string, job: AssessmentJob): Promise => { const t0 = Date.now(); await setStep(job, 'mails', { state: 'running' }) if (!isValidEmail(email)) { await setStep(job, 'mails', { state: 'skipped', message: 'Keine E-Mail-Adresse am Lead' }); return [] } if (!archiveAvailable()) { job.warnings.push('Mailpiler-Archiv nicht konfiguriert'); await setStep(job, 'mails', { state: 'skipped', message: 'Archiv nicht konfiguriert' }); return [] } try { const { rows } = await searchArchive(email, MAX_MAIL_LIST) const sorted = [...rows].sort((a: any, b: any) => new Date(b.date || 0).getTime() - new Date(a.date || 0).getTime()) const items: MailDossierItem[] = sorted.map((r: any) => ({ date: String(r.date || ''), direction: r.direction, subject: r.subject, from: r.from?.address ? `${r.from.name || ''} <${r.from.address}>`.trim() : '', to: (r.to || []).map((t: any) => t.address).filter(Boolean).join(', ') })) const withBody = sorted.slice(0, MAX_MAILS) let failed = 0 await limitRun(withBody, MAIL_CONCURRENCY, async (r: any) => { const idx = sorted.indexOf(r) try { const m: any = await fetchArchiveMessage(Number(r.uid), email) if (!m) return const text = String(m.text || '').trim() || htmlToText(String(m.html || '')) items[idx].body = clip(stripQuotes(text), MAIL_BODY_CHARS) const atts = (m.attachments || []).map((a: any) => a.filename).filter(Boolean) if (atts.length) items[idx].attachments = atts } catch { failed++ } }) if (failed) job.warnings.push(`${failed} E-Mail-Inhalte konnten nicht geladen werden`) await setStep(job, 'mails', { state: 'done', ms: Date.now() - t0, count: items.length, message: `${Math.min(withBody.length, items.length)} mit Inhalt` }) return items } catch (err: any) { job.warnings.push('E-Mails: ' + (err?.message || err)) await setStep(job, 'mails', { state: 'error', message: err?.message || String(err) }) return [] } } interface CallDossierItem { at: string; direction: string; number: string; extension: string; answered: boolean; duration: number; transcript?: string; language?: string } export const collectCalls = async (numbers: string[], job: AssessmentJob): Promise => { const t0 = Date.now(); await setStep(job, 'calls', { state: 'running' }) const nums = [...new Set(numbers.map((n) => String(n || '').trim()).filter((n) => digitsOf(n).length >= 6))] if (!nums.length) { await setStep(job, 'calls', { state: 'skipped', message: 'Keine Telefonnummer am Lead' }); await setStep(job, 'transcribe', { state: 'skipped' }); return [] } try { const { available, rows } = await findRecordings(nums) if (!available) { job.warnings.push('Telefonanlage (CDR) nicht konfiguriert'); await setStep(job, 'calls', { state: 'skipped', message: 'CDR nicht konfiguriert' }); await setStep(job, 'transcribe', { state: 'skipped' }); return [] } let usable = rows if (pbxAvailable()) { try { const stat = await statRecordings(rows.map((r) => r.file)) usable = rows.filter((r) => stat[r.file] && Number(stat[r.file]!.size) > 1000) } catch (err: any) { job.warnings.push('Aufnahmen-Dienst: ' + (err?.message || err)) } } const items: CallDossierItem[] = [] for (const r of usable) { const it: CallDossierItem = { at: r.at, direction: r.direction, number: r.number, extension: r.extension, answered: r.answered, duration: r.duration } if (hasTranscript(r.file)) { const t = await readTranscript(r.file); if (t?.text) { it.transcript = clip(t.text, TRANSCRIPT_CHARS); it.language = t.language } } items.push(it) } await setStep(job, 'calls', { state: 'done', ms: Date.now() - t0, count: items.length, message: `${items.filter((i) => i.transcript).length} transkribiert` }) // Optional: transcribe the newest untranscribed recordings (bounded). const missing = usable.filter((r) => !hasTranscript(r.file) && r.duration >= 20).slice(0, MAX_TRANSCRIBE_MISSING) if (!job.transcribeMissing || !missing.length) { await setStep(job, 'transcribe', { state: 'skipped', message: job.transcribeMissing ? 'Nichts zu transkribieren' : 'Nicht angefordert' }) } else { const t1 = Date.now(); await setStep(job, 'transcribe', { state: 'running', count: missing.length }) let done = 0 for (const r of missing) { const left = TRANSCRIBE_BUDGET_MS - (Date.now() - t1) if (left < 15000) { job.warnings.push('Transkriptions-Zeitbudget erschöpft'); break } try { let st = await startTranscription(r.file, 'de', false) if (st.state === 'pending') st = await waitForTranscription(r.file, left, 2000) if (st.state === 'done' && st.transcript?.text) { const it = items.find((i) => i.at === r.at && i.number === r.number) if (it) { it.transcript = clip(st.transcript.text, TRANSCRIPT_CHARS); it.language = st.transcript.language } done++ } else if (st.state === 'failed') job.warnings.push(`Transkription fehlgeschlagen (${r.at}): ${st.message || 'keine Sprache erkannt'}`) } catch (err: any) { job.warnings.push(`Transkription (${r.at}): ${err?.message || err}`) } await saveJob(job) } await setStep(job, 'transcribe', { state: 'done', ms: Date.now() - t1, count: done }) } // Cap the number of transcripts that go into the prompt (newest first). let n = 0 for (const it of items) { if (it.transcript) { n++; if (n > MAX_TRANSCRIPTS) delete it.transcript } } return items } catch (err: any) { job.warnings.push('Anrufe: ' + (err?.message || err)) await setStep(job, 'calls', { state: 'error', message: err?.message || String(err) }) await setStep(job, 'transcribe', { state: 'skipped' }) return [] } } // ---- 3) dossier + prompt ------------------------------------------------------------- const fmtDate = (v: any) => { const s = String(v || ''); if (!s) return ''; const d = new Date(s); return Number.isNaN(d.getTime()) ? s : d.toISOString().replace('T', ' ').slice(0, 16) } export const buildDossier = (erp: ErpContext, mails: MailDossierItem[], calls: CallDossierItem[]): string => { const L = erp.lead || {} const parts: string[] = [] parts.push('# LEAD-DATENSATZ (iDempiere AD_User)') const fields: [string, any][] = [ ['Name', val(L, 'Name')], ['Firma', val(L, 'companyname', 'CompanyName')], ['Position', ident(L.C_Job_ID)], ['E-Mail', val(L, 'EMail')], ['Telefon', val(L, 'Phone')], ['Telefon 2', val(L, 'Phone2')], ['Website', val(L, 'URL')], ['Lead-Status', ident(L.LeadStatus)], ['Lead-Quelle', ident(L.LeadSource)], ['Sales-Lead', L.IsSalesLead ? 'ja' : 'nein'], ['Vendor-Lead', L.IsVendorLead ? 'ja' : 'nein'], ['Aktiv', L.IsActive === false ? 'nein' : 'ja'], ['Betreuer (SalesRep)', ident(L.SalesRep_ID) || ident(L.Supervisor_ID)], ['Organisation', ident(L.AD_Org_ID)], ['Geschäftspartner', ident(L.C_BPartner_ID)], ['Kampagne', ident(L.C_Campaign_ID)], ['Angelegt', fmtDate(L.Created)], ['Zuletzt geändert', fmtDate(L.Updated)], ['Letzter Kontakt', fmtDate(L.LastContact)], ['Letztes Ergebnis', val(L, 'LastResult')], ['Beschreibung', val(L, 'Description')], ['Kommentare / Notizen', val(L, 'Comments')], ['Lead-Quelle Beschreibung', val(L, 'LeadSourceDescription')], ['Lead-Status Beschreibung', val(L, 'LeadStatusDescription')] ] for (const [k, v] of fields) if (v !== '' && v !== undefined && v !== null) parts.push(`- ${k}: ${String(v)}`) parts.push('\n# AKTIVITÄTEN (C_ContactActivity, neueste zuerst)') if (!erp.activities.length) parts.push('(keine)') for (const a of erp.activities) { parts.push(`- ${fmtDate(a.StartDate || a.Created)} | ${ident(a.ContactActivityType)} | ${val(a, 'Description')}${a.IsComplete ? ' | erledigt' : ''}${a.SalesRep_ID ? ' | ' + ident(a.SalesRep_ID) : ''}`) const c = val(a, 'Comments'); if (c) parts.push(' ' + clip(c, 2000).replace(/\n/g, '\n ')) } parts.push('\n# ANGEBOTE / VERTRAG / PROGNOSE / FRAGEBOGEN (gespeicherte Konditionen, offer_conditions JSON)') const cond = erp.conditions || {} const keys = Object.keys(cond) if (!keys.length) parts.push('(noch kein Angebot / Vertrag / Fragebogen gespeichert)') for (const k of keys) parts.push(`## ${k} (gespeichert ${fmtDate(cond[k]?.savedAt)})\n${clip(JSON.stringify(cond[k], null, 1), SECTION_JSON_CHARS)}`) parts.push('\n# DIGITALE SIGNATUR- / BESTÄTIGUNGS-LINKS (neueste zuerst)') if (!erp.signing.length) parts.push('(keine)') for (const s of erp.signing) parts.push(`- ${s.kind} | Status: ${s.state} | erstellt ${fmtDate(s.createdAt)} | läuft ab ${fmtDate(s.expiresAt)}${s.signedAt ? ' | unterschrieben/bestätigt ' + fmtDate(s.signedAt) : ''}${s.lastOpenedAt ? ' | zuletzt geöffnet ' + fmtDate(s.lastOpenedAt) : ''}${s.renewalRequestedAt ? ' | Erneuerung angefragt ' + fmtDate(s.renewalRequestedAt) : ''} | Dokumente: ${(s.documents || []).join(', ')}`) parts.push('\n# ANHÄNGE AM LEAD (Dateinamen)') if (!erp.attachments.length) parts.push('(keine)') for (const f of erp.attachments) parts.push(`- ${f.date} ${f.name} (${Math.round(f.size)} KB)`) parts.push(`\n# E-MAIL-VERLAUF (Mailpiler-Archiv, ${mails.length} Mails, neueste zuerst; "in" = vom Lead an uns, "out" = von uns an den Lead)`) if (!mails.length) parts.push('(keine E-Mails gefunden)') for (const m of mails) { parts.push(`\n## ${fmtDate(m.date)} | ${m.direction === 'in' ? 'EINGEHEND vom Lead' : 'AUSGEHEND an den Lead'} | Von: ${m.from} | An: ${m.to}\nBetreff: ${m.subject || '(kein Betreff)'}${m.attachments?.length ? '\nAnhänge: ' + m.attachments.join(', ') : ''}`) if (m.body !== undefined) parts.push(m.body || '(kein Textinhalt)') else parts.push('(Inhalt nicht geladen – nur Betreff)') } parts.push(`\n# TELEFONATE (FreePBX, ${calls.length} Aufnahmen, neueste zuerst)`) if (!calls.length) parts.push('(keine Aufnahmen gefunden)') for (const c of calls) { parts.push(`\n## ${fmtDate(c.at)} | ${c.direction === 'in' ? 'EINGEHEND' : c.direction === 'out' ? 'AUSGEHEND' : c.direction} | Nummer ${c.number} | Nebenstelle ${c.extension} | ${c.answered ? 'angenommen' : 'nicht angenommen'} | ${Math.round(c.duration)} s`) if (c.transcript) parts.push(`Transkript (${c.language || 'de'}, Whisper, automatisch – kann Fehler enthalten, keine Sprecherzuordnung):\n${c.transcript}`) else parts.push('(kein Transkript)') } return parts.join('\n') } export const LANG_NAMES: Record = { de: 'Deutsch', en: 'English', es: 'Español' } const SYSTEM_PROMPT = `Du bist ein erfahrener B2B-Vertriebscoach und Analyst für LogYou (LogShip), einen deutschen Fulfillment-/3PL-Dienstleister (Lagerung, Kommissionierung, Versand für Online-Händler; Angebote mit Lagergeld, Pick & Pack, Versandkonditionen, Wareneingang, Retouren; Rahmenvertrag mit AGB/ADSp/AVV, digitale Signatur, SEPA-Mandat, Fragebogen Erstgespräch, individuelle Prognose). Du bekommst ein vollständiges Dossier zu EINEM Lead: ERP-Datensatz, Aktivitäten, gespeicherte Angebots-/Vertragskonditionen, Signatur-Links, Anhänge, den kompletten E-Mail-Verlauf (gesendet und empfangen) und Transkripte der Telefonate. Deine Aufgabe: Analysiere alles gründlich und ehrlich wie ein Vertriebsleiter, der den Deal abschließen will. Beurteile: - Wo steht der Lead wirklich (Interesse, Entscheidungsstand, Hindernisse, wer entscheidet, Timing)? - Was lief gut, was lief schlecht (Reaktionszeit, Tonalität, fehlende Antworten, unbeantwortete Fragen des Kunden, Versprechen die nicht eingelöst wurden)? - Preis/Konditionen: zu teuer, unklar, falsch positioniert? Was fehlte im Angebot? Welche Einwände hat der Kunde geäußert oder angedeutet? - Was fehlt noch (Informationen, Dokumente, Zusagen), was ist offen? - Was ist der beste nächste Schritt, um den Lead zum Kunden zu machen – oder die ehrliche Einschätzung, dass er (vorerst) nicht zu gewinnen ist und warum. - Ein konkreter Vorschlag (Angebotsanpassung, Nachfass-Strategie, Gesprächsleitfaden) plus ein fertiger Nachrichtenentwurf. Regeln: - Stütze dich NUR auf das Dossier. Erfinde nichts. Wenn Informationen fehlen, sage das in dataGaps. - Zitiere, wo es hilft, kurz die Quelle (Datum + Mail/Anruf). - Sei konkret und handlungsorientiert, keine Allgemeinplätze. Nenne Daten, Beträge, Namen. - Transkripte sind automatisch erstellt (Whisper) und können Fehler enthalten; Sprecher sind nicht markiert – leite den Sprecher aus dem Kontext ab. - Antworte AUSSCHLIESSLICH mit einem JSON-Objekt (kein Markdown, keine Code-Fences, kein Text davor oder danach) mit genau dieser Struktur: { "headline": "Ein Satz: Kernaussage / Verdikt", "dealScore": 0-100 (Einschätzung der Abschlusswahrscheinlichkeit als ganze Zahl), "stage": "kurze Phasenbezeichnung, z. B. 'Angebot gesendet, wartet auf Entscheidung'", "summary": "Zusammenfassung des Verlaufs in 4-8 Sätzen", "timeline": [{ "date": "YYYY-MM-DD", "event": "kurz", "source": "mail|call|offer|activity|contract|other" }], "customerNeeds": ["was der Kunde will / braucht"], "positives": ["was für einen Abschluss spricht"], "risks": ["was dagegen spricht, was schiefgelaufen ist, was fehlt"], "pricing": { "assessment": "Einschätzung zu Preis/Konditionen", "observations": ["konkrete Beobachtungen"] }, "openQuestions": ["unbeantwortete Fragen – des Kunden und unsere"], "nextSteps": [{ "priority": "high|medium|low", "action": "konkret", "why": "Begründung", "when": "Zeitpunkt" }], "proposal": "Konkreter Vorschlag zum Abschluss (mehrere Absätze erlaubt, mit \\n getrennt)", "draftMessage": { "channel": "email|phone", "subject": "Betreff", "body": "fertiger Text / Gesprächsleitfaden – bei E-Mail: Anrede bis Grußformel ('Viele Grüße' + Vorname), OHNE Signaturblock (Name/Firma/Kontaktdaten werden beim Versand automatisch angehängt)" }, "dataGaps": ["welche Informationen im Dossier fehlten oder unklar waren"] } Alle Texte in der Sprache, die der Nutzer angibt.` const extractJson = (raw: string): any => { if (!raw) return null const fence = raw.match(/```(?:json)?\s*([\s\S]*?)```/i) const candidate = fence ? fence[1] : raw const start = candidate.indexOf('{'); const end = candidate.lastIndexOf('}') if (start === -1 || end === -1) return null try { return JSON.parse(candidate.slice(start, end + 1)) } catch { return null } } /** Anthropic client as configured (shared with the lead chat console). */ export const anthropicClient = (timeoutMs = 15 * 60 * 1000) => { const { apiKey, workspaceId } = cfg() if (!apiKey) throw new Error('ANTHROPIC_API_KEY ist nicht konfiguriert') // User-scoped / multi-workspace keys must name the workspace on every request. return new Anthropic({ apiKey, timeout: timeoutMs, maxRetries: 1, ...(workspaceId ? { defaultHeaders: { 'anthropic-workspace-id': workspaceId } } : {}) }) } /** Turns SDK errors into operator-readable German messages (same wording in both AI features). */ export const anthropicErrorMessage = (err: any): string => { if (err instanceof Anthropic.AuthenticationError) return 'Anthropic API-Key ungültig (ANTHROPIC_API_KEY prüfen)' if (err instanceof Anthropic.BadRequestError && /workspace/i.test(err.message || '')) return 'Der API-Key ist nicht an einen Workspace gebunden – ANTHROPIC_WORKSPACE_ID (wrkspc_…) setzen oder einen Workspace-Key anlegen' if (err instanceof Anthropic.RateLimitError) return 'Anthropic Rate-Limit erreicht – bitte in ein paar Minuten erneut versuchen' if (err instanceof Anthropic.APIError) return `Anthropic API-Fehler ${err.status || ''}: ${err.message}` return err?.message || String(err) } const askClaude = async (dossier: string, language: string, job: AssessmentJob) => { const { model } = cfg() const client = anthropicClient() const userText = `Sprache der Antwort: ${LANG_NAMES[language] || 'Deutsch'}.\nHeutiges Datum: ${new Date().toISOString().slice(0, 10)}.\n\nDOSSIER:\n\n${dossier}` job.promptChars = SYSTEM_PROMPT.length + userText.length const stream = client.beta.messages.stream({ model, max_tokens: 32000, betas: ['server-side-fallback-2026-07-01'], fallbacks: 'default', output_config: { effort: 'high' }, system: [{ type: 'text', text: SYSTEM_PROMPT, cache_control: { type: 'ephemeral' } }], messages: [{ role: 'user', content: userText }] } as any) let msg: any try { msg = await stream.finalMessage() } catch (err: any) { throw new Error(anthropicErrorMessage(err)) } job.model = String(msg?.model || model) job.usage = { input: Number(msg?.usage?.input_tokens || 0), output: Number(msg?.usage?.output_tokens || 0), cacheRead: Number(msg?.usage?.cache_read_input_tokens || 0), cacheWrite: Number(msg?.usage?.cache_creation_input_tokens || 0) } if (msg?.stop_reason === 'refusal') throw new Error('Das Modell hat die Analyse abgelehnt' + (msg?.stop_details?.explanation ? ': ' + msg.stop_details.explanation : '')) const text = (msg?.content || []).filter((b: any) => b.type === 'text').map((b: any) => b.text).join('\n') if (msg?.stop_reason === 'max_tokens') job.warnings.push('Antwort wurde am Token-Limit abgeschnitten') const parsed = extractJson(text) if (!parsed) { job.rawText = text; job.warnings.push('Antwort konnte nicht als JSON gelesen werden – Rohtext wird angezeigt') } return parsed } // ---- 4) orchestration ---------------------------------------------------------------- const running = new Map>() /** * Starts an assessment. Returns the pending job immediately after the iDempiere reads; * the archive / PBX / Claude work continues in the background and is polled via * readAssessmentJob(). A still-running job of the same lead is returned as-is. */ export const startAssessment = async (event: any, token: string, leadId: number, userId: number, language: string, transcribeMissing: boolean): Promise => { const existing = await readAssessmentJob(leadId) if (existing?.state === 'pending') return existing const job = newJob(leadId, language, transcribeMissing, userId) await saveJob(job) let erp: ErpContext try { erp = await collectErpContext(event, token, leadId, job) } catch (err: any) { job.state = 'failed'; job.error = err?.message || String(err); job.finishedAt = new Date().toISOString() await setStep(job, 'lead', { state: 'error', message: job.error }) await saveJob(job) return job } const email = String(val(erp.lead, 'EMail')).trim().toLowerCase() const numbers = [val(erp.lead, 'Phone'), val(erp.lead, 'Phone2')].map(String) job.lead = { name: String(val(erp.lead, 'Name')), email } await saveJob(job) const run = (async () => { try { const [mails, calls] = await Promise.all([collectMails(email, job), collectCalls(numbers, job)]) const dossier = buildDossier(erp, mails, calls) job.sources = { activities: erp.activities.length, conditions: Object.keys(erp.conditions || {}).length, signing: erp.signing.length, attachments: erp.attachments.length, mails: mails.length, mailBodies: mails.filter((m) => m.body !== undefined).length, calls: calls.length, transcripts: calls.filter((c) => c.transcript).length } const t0 = Date.now(); await setStep(job, 'ai', { state: 'running' }) job.result = await askClaude(dossier, language, job) await setStep(job, 'ai', { state: 'done', ms: Date.now() - t0 }) job.state = 'done' // Persist the deal score on the lead (AD_User.AiScore) so the list can sort by it. // Background → session token may be gone: use the service token. Fail-soft. const score = Math.round(Number(job.result?.dealScore)) if (Number.isFinite(score) && score >= 0 && score <= 100) { const serviceToken: string = (useRuntimeConfig() as any).api?.idempieretoken || '' if (serviceToken) await writeLeadQualification(event, serviceToken, leadId, { aiScore: score }, 'ai-score') } } catch (err: any) { const m = err?.message || String(err) console.warn('[lead-assessment] failed:', leadId, m) job.state = 'failed'; job.error = m const ai = step(job, 'ai'); if (ai.state !== 'done') { ai.state = 'error'; ai.message = m } } finally { job.finishedAt = new Date().toISOString() await saveJob(job) running.delete(leadId) } })() running.set(leadId, run) return job }