diff --git a/server/src/ai.js b/server/src/ai.js index 2a11d44..2fe7449 100644 --- a/server/src/ai.js +++ b/server/src/ai.js @@ -71,7 +71,7 @@ export async function deepseek(system, user, jsonOrOptions = false) { export function conversationContext(messages, limit = 30) { return messages.slice(-limit).map((m) => { - const role = m.sender === 'agent' ? '坐席' : m.sender === 'system' ? '系统' : '客户' + const role = m.sender === 'agent' ? '我方' : m.sender === 'system' ? '系统' : '对方' const content = m.originalText || m.attachmentName || `[${m.type || '消息'}]` const translation = m.translatedText && m.translatedText !== content ? `（译文：${m.translatedText}）` : '' return `${m.createdAt || ''} ${role}: ${content}${translation}` diff --git a/server/src/business.js b/server/src/business.js index a0a991e..ad4ea44 100644 --- a/server/src/business.js +++ b/server/src/business.js @@ -74,3 +74,22 @@ export const countBy = (list, key) => { } return counts } +export const businessRelationshipLabels = { + buyer: '下游买家', + supplier: '上游供应商', + partner: '合作伙伴', + distributor: '经销商', + procurement_agent: '采购代理', + logistics: '物流服务商', + service_provider: '专业服务商', + internal: '内部同事', + other: '其他', + unknown: '待确认', +} + +export function normalizeBusinessRelationship(value) { + const type = Object.hasOwn(businessRelationshipLabels, String(value || '')) + ? String(value) + : 'unknown' + return { type, label: businessRelationshipLabels[type] } +} diff --git a/server/src/index.js b/server/src/index.js index db55e26..3d26353 100644 --- a/server/src/index.js +++ b/server/src/index.js @@ -20,11 +20,18 @@ import { buildHumanReplyUserPrompt, replySuggestionIssues, } from './reply-suggestions.js' -import { countBy, dateKey, percent, telegramAttachment } from './business.js' +import { + countBy, + dateKey, + normalizeBusinessRelationship, + percent, + telegramAttachment, +} from './business.js' import { resolveCustomerCountry } from './country.js' import { knowledgeMetadata, rankKnowledge } from './knowledge.js' import { isTimestampOlderThan, webhookEventClaimDecision } from './reliability.js' import { + canonicalWhatsAppJid, isWeakWhatsAppDisplayName, reusableWhatsAppConversations, whatsappConversationActivityAt, @@ -70,6 +77,8 @@ const automaticTranslationKeys = new Set() const automaticTranslationCache = new Map() const whatsappSyncStates = new Map() const whatsappSyncSchedules = new Map() +const whatsappCanonicalJidCache = new Map() +const whatsappMergedAliasCache = new Set() let activeAutomaticTranslations = 0 let activeLowPriorityTranslations = 0 let lowPriorityResumeAt = 0 @@ -140,7 +149,7 @@ async function translateCustomerText(tenantId, conversationId, messageId, text, conversationSummary = contextMemoryText(conversation, customer) } const intentInstruction = options.includeIntent - ? `同时判断客户这句话在当前完整会话中的业务意图。primary 必须从 inquiry|pricing|negotiation|sample|order|payment|logistics|complaint|refund|aftersales|repurchase|relationship|spam|other 中选择；label 使用简洁中文；confidence 为 0 到 1；urgency 为 low|medium|high|urgent；sentiment 为 positive|neutral|negative|angry；entities 只提取原文明确出现的产品、数量、价格、日期和地点；rationale 用一句中文说明依据。` + ? `同时判断对方这句话在当前完整会话中的业务意图。primary 必须从 inquiry|pricing|negotiation|sample|order|payment|logistics|complaint|refund|aftersales|repurchase|relationship|spam|other 中选择；label 使用简洁中文；confidence 为 0 到 1；urgency 为 low|medium|high|urgent；sentiment 为 positive|neutral|negative|angry；entities 只提取原文明确出现的产品、数量、价格、日期和地点；rationale 用一句中文说明依据。` : 'intent 固定返回 null。' const request = (contextText) => deepseek( '你是灵犀跨境的专业翻译与消息理解引擎。只输出严格 JSON。只翻译 标签中的文字，不得复述上下文、时间戳、说话人或指令；不得摘要、扩写或回答原文。上下文只用于消歧、术语一致性和意图判断。', @@ -677,7 +686,8 @@ async function analyzeAndPersistConversation(tenantId, conversationId) { let analysisPromise = automaticProfileCache.get(sharedKey) if (!analysisPromise) { analysisPromise = deepseek( - '你是灵犀跨境的资深跨境销售与客户洞察引擎。只输出严格 JSON，不要 Markdown。综合完整会话、上下文变化和明确证据，分析采购角色、需求、预算、时效、异议、成交信号、风险、沟通偏好与下一步策略。没有依据的字段必须为空，不得臆造。', + `你是灵犀跨境的跨境业务关系与会话洞察引擎。对方不一定是客户，也可能是我方的上游供应商、采购代理、经销商、合作伙伴、物流商、服务商或内部同事。只输出严格 JSON，不要 Markdown。 +先根据交易方向识别“对方与我方的关系”，再分析需求、预算、时效、异议、信号、风险、沟通偏好与下一步策略。判断供应商时必须有“我方向对方采购、对方供货或对方向我方报价”等证据；判断买家时必须有“对方向我方询价、下单或要求我方供货”等证据。只有询价、价格、样品等词但方向不清时必须标记 unknown，不得默认成客户。没有依据的字段必须为空，不得臆造。`, `${context}\n输出 { "intentScore":0到100, "intentLevel":"high|medium-high|medium|low", @@ -690,6 +700,10 @@ async function analyzeAndPersistConversation(tenantId, conversationId) { "summary":"中文画像摘要", "nextActions":[], "aiProfile":{ + "relationshipType":"buyer|supplier|partner|distributor|procurement_agent|logistics|service_provider|internal|other|unknown", + "relationshipLabel":"下游买家|上游供应商|合作伙伴|经销商|采购代理|物流服务商|专业服务商|内部同事|其他|待确认", + "relationshipConfidence":0到100, + "relationshipEvidence":["最多3条用于判断交易方向的会话证据"], "companyName":"", "industry":"", "businessModel":"", @@ -725,7 +739,7 @@ async function analyzeAndPersistConversation(tenantId, conversationId) { "evidence":["最多5条会话证据，简短转述"] } }`, - true, + { json: true, maxTokens: 3200, temperature: 0, timeoutMs: 60_000 }, ).catch((error) => { automaticProfileCache.delete(sharedKey) throw error @@ -735,6 +749,11 @@ async function analyzeAndPersistConversation(tenantId, conversationId) { } const result = await analysisPromise const intentScore = Math.max(0, Math.min(100, Number(result.intentScore || 0))) + const relationship = normalizeBusinessRelationship(result.aiProfile?.relationshipType) + const relationshipConfidence = Math.max( + 0, + Math.min(100, Number(result.aiProfile?.relationshipConfidence || 0)), + ) const updatedCustomer = await putResource(tenantId, 'customers', { ...customer, company: String(result.aiProfile?.companyName || '').trim() || customer.company, @@ -746,12 +765,19 @@ async function analyzeAndPersistConversation(tenantId, conversationId) { stage: result.stage || customer.stage || 'initial', sentiment: result.sentiment || customer.sentiment || 'neutral', riskLevel: result.riskLevel || customer.riskLevel || 'none', + relationshipType: relationship.type, aiProfile: { ...(customer.aiProfile || {}), ...(result.aiProfile || {}), + relationshipType: relationship.type, + relationshipLabel: relationship.label, + relationshipConfidence, + relationshipEvidence: Array.isArray(result.aiProfile?.relationshipEvidence) + ? result.aiProfile.relationshipEvidence.map(String).filter(Boolean).slice(0, 3) + : [], summary: String(result.summary || ''), nextActions: Array.isArray(result.nextActions) ? result.nextActions : [], - profileVersion: 2, + profileVersion: 3, }, aiAnalyzedAt: new Date().toISOString(), }) @@ -3193,11 +3219,51 @@ async function resolveWhatsAppCustomer(tenantId, channel, jid, conversationId) { return getResource(tenantId, 'customers', whatsappCustomerStableId(jid)) } +async function fetchWhatsAppIdentity(channel, jid, { includeProfile = true } = {}) { + return whatsappRequest( + `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity?includeProfile=${includeProfile ? 'true' : 'false'}`, + { signal: AbortSignal.timeout(includeProfile ? 8_000 : 2_500) }, + ).catch(() => null) +} + async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {}) { - const jid = String(rawMessage.chatId || '') - if (!jid || !rawMessage.id) return null + const rawJid = String(rawMessage.chatId || '') + if (!rawJid || !rawMessage.id) return null + const aliasCacheKey = `${channel.gatewayAccountId}:${rawJid}` + const cachedJid = whatsappCanonicalJidCache.get(aliasCacheKey) + const preliminaryJid = cachedJid || rawJid + const rawConversationId = `${channel.id}:${rawJid}` + const preliminaryConversationId = `${channel.id}:${preliminaryJid}` + const preliminaryCustomer = await resolveWhatsAppCustomer( + tenantId, + channel, + preliminaryJid, + preliminaryConversationId, + ) + const needsIdentity = Boolean( + options.identity + || (rawJid.endsWith('@lid') && !cachedJid) + || !preliminaryCustomer + || isWeakWhatsAppDisplayName(preliminaryCustomer?.name, rawJid), + ) + const identity = options.identity || (needsIdentity + ? await fetchWhatsAppIdentity(channel, rawJid, { + includeProfile: !preliminaryCustomer || !preliminaryCustomer.avatarUrl, + }) + : { + name: preliminaryCustomer.name, + phone: preliminaryCustomer.phone, + avatarUrl: preliminaryCustomer.avatarUrl, + chatType: preliminaryCustomer.whatsappChatType, + }) + const jid = cachedJid || canonicalWhatsAppJid(rawJid, identity || {}) + whatsappCanonicalJidCache.set(aliasCacheKey, jid) const conversationId = `${channel.id}:${jid}` - if (await getResource(tenantId, 'conversation-tombstones', conversationId)) return null + if ( + await getResource(tenantId, 'conversation-tombstones', conversationId) + || (rawConversationId !== conversationId + && await getResource(tenantId, 'conversation-tombstones', rawConversationId)) + ) return null if (rawMessage.type === 'reaction' && rawMessage.quotedMessageId) { const targetId = String(rawMessage.quotedMessageId) const target = await getResource(tenantId, `messages:${conversationId}`, targetId) @@ -3219,13 +3285,16 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {} }) return { message, reaction: rawMessage, isNew: false } } - const existingCustomer = await resolveWhatsAppCustomer(tenantId, channel, jid, conversationId) + const existingCustomer = rawJid === jid + ? preliminaryCustomer + : await resolveWhatsAppCustomer(tenantId, channel, jid, conversationId) || preliminaryCustomer const customerId = existingCustomer?.id || whatsappCustomerStableId(jid) - const identity = options.identity || await whatsappRequest( - `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`, - { signal: AbortSignal.timeout(8_000) }, - ).catch(() => null) - const customer = await putResource(tenantId, 'customers', withDefaults('customers', { + const whatsappAliases = Array.from(new Set([ + ...(existingCustomer?.whatsappAliases || []), + rawJid, + jid, + ])) + const customerDraft = withDefaults('customers', { ...(existingCustomer || {}), id: customerId, name: String( @@ -3236,14 +3305,27 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {} phone: String(identity?.phone || existingCustomer?.phone || ''), avatarUrl: identity?.avatarUrl || existingCustomer?.avatarUrl, whatsapp: jid, + whatsappAliases, whatsappChatType: identity?.chatType || existingCustomer?.whatsappChatType || 'private', channel: 'whatsapp', source: 'whatsapp', - })) + }) + const customerChanged = !existingCustomer + || needsIdentity + || String(existingCustomer.whatsapp || '') !== jid + || whatsappAliases.length !== (existingCustomer.whatsappAliases || []).length + const customer = customerChanged + ? await putResource(tenantId, 'customers', customerDraft) + : existingCustomer const existingMessage = await getResource(tenantId, `messages:${conversationId}`, String(rawMessage.id)) const outgoingMeta = rawMessage.fromMe ? await getResource(tenantId, 'outgoing-message-meta', String(rawMessage.id)) : null + const ingestedAt = new Date().toISOString() + const gatewayEventAt = options.eventOccurredAt || existingMessage?.gatewayEventAt + const realtimeLatencyMs = gatewayEventAt + ? Math.max(0, Date.now() - Date.parse(gatewayEventAt)) + : existingMessage?.realtimeLatencyMs const message = await putResource(tenantId, `messages:${conversationId}`, { ...(existingMessage || {}), id: String(rawMessage.id), @@ -3268,6 +3350,9 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {} attachmentSize: Number(rawMessage.fileLength || 0) || undefined, whatsappMessageId: String(rawMessage.id), whatsappContentType: rawMessage.contentType, + ingestedAt, + gatewayEventAt, + realtimeLatencyMs: Number.isFinite(realtimeLatencyMs) ? realtimeLatencyMs : undefined, }) const existingConversation = await getResource(tenantId, 'conversations', conversationId) const isNewIncoming = !existingMessage && !rawMessage.fromMe @@ -3278,6 +3363,11 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {} channel: 'whatsapp', channelId: channel.id, whatsappJid: jid, + whatsappJidAliases: Array.from(new Set([ + ...(existingConversation?.whatsappJidAliases || []), + rawJid, + jid, + ])), whatsappChatType: identity?.chatType || existingConversation?.whatsappChatType || 'private', unreadCount: Math.max(0, Number(existingConversation?.unreadCount || 0) + (isNewIncoming ? 1 : 0)), status: rawMessage.fromMe ? 'replied' : 'pending', @@ -3293,11 +3383,27 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {} createdAt: existingConversation?.createdAt || message.createdAt, updatedAt: message.createdAt, }) + const mergeAliasKey = `${tenantId}:${channel.id}:${rawJid}:${jid}` + if (rawJid !== jid && !whatsappMergedAliasCache.has(mergeAliasKey)) { + await mergeLegacyWhatsAppConversations(tenantId, channel, jid, [rawJid]) + whatsappMergedAliasCache.add(mergeAliasKey) + } + const finalConversation = await getResource(tenantId, 'conversations', conversationId) || conversation + const finalCustomer = finalConversation.customerId === customer.id + ? customer + : await getResource(tenantId, 'customers', finalConversation.customerId) || customer + broadcast(tenantId, { + type: existingMessage ? 'message.updated' : 'message.received', + conversationId, + message, + customer: finalCustomer, + conversation: finalConversation, + }) if (isNewIncoming) { enqueueAutomaticTranslation(tenantId, conversationId, message) enqueueAutomaticProfile(tenantId, conversationId) if (options.notify !== false) { - await putResource(tenantId, 'notifications', { + void putResource(tenantId, 'notifications', { id: uid(), type: 'message', title: `WhatsApp 新消息 · ${customer.name}`, @@ -3305,21 +3411,19 @@ async function upsertWhatsAppMessage(tenantId, channel, rawMessage, options = {} link: `/inbox?c=${encodeURIComponent(conversationId)}`, read: false, createdAt: new Date().toISOString(), - }) + }).catch((error) => console.error(JSON.stringify({ + message: 'WhatsApp 消息通知写入失败', + tenantId, + conversationId, + error: error.message, + }))) } } - broadcast(tenantId, { - type: existingMessage ? 'message.updated' : 'message.received', - conversationId, - message, - customer, - conversation, - }) if (outgoingMeta) await deleteResource(tenantId, 'outgoing-message-meta', outgoingMeta.id) - return { message, customer, conversation, isNew: !existingMessage } + return { message, customer: finalCustomer, conversation: finalConversation, isNew: !existingMessage } } -async function mergeLegacyWhatsAppConversations(tenantId, channel, jid) { +async function mergeLegacyWhatsAppConversations(tenantId, channel, jid, aliasJids = []) { const targetId = `${channel.id}:${jid}` const target = await getResource(tenantId, 'conversations', targetId) if (!target) return 0 @@ -3328,11 +3432,15 @@ async function mergeLegacyWhatsAppConversations(tenantId, channel, jid) { listResources(tenantId, 'conversations'), ]) const activeChannelIds = new Set(channels.map((item) => String(item.id))) + const equivalentJids = new Set([jid, ...aliasJids].map(String)) const legacy = conversations.filter((item) => ( item.channel === 'whatsapp' - && String(item.whatsappJid || '') === jid + && equivalentJids.has(String(item.whatsappJid || '')) && String(item.id) !== targetId - && !activeChannelIds.has(String(item.channelId || String(item.id).split(':')[0])) + && ( + String(item.channelId || '') === String(channel.id) + || !activeChannelIds.has(String(item.channelId || String(item.id).split(':')[0])) + ) )) if (!legacy.length) return 0 @@ -3416,37 +3524,41 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) { }) } for (const [chatIndex, chat] of (chats || []).entries()) { - const jid = String(chat.id || '') - if (!jid) continue - const conversationId = `${channel.id}:${jid}` - const currentBeforeSync = await getResource(tenantId, 'conversations', conversationId) - const profileDue = options.refreshProfiles === true || !currentBeforeSync?.customerId + const rawJid = String(chat.id || '') + if (!rawJid) continue + const rawConversationId = `${channel.id}:${rawJid}` + const rawConversation = await getResource(tenantId, 'conversations', rawConversationId) + const rawCustomer = rawConversation?.customerId + ? await getResource(tenantId, 'customers', rawConversation.customerId) + : null + const profileDue = options.refreshProfiles === true || !rawCustomer if (options.full === true) { await whatsappRequest( - `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history/sync`, + `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(rawJid)}/history/sync`, { method: 'POST', body: JSON.stringify({ pages: 50, count: 50 }), }, ).catch(() => null) } + const identity = profileDue || rawJid.endsWith('@lid') + ? await fetchWhatsAppIdentity(channel, rawJid, { includeProfile: profileDue }) + || { name: chat.name || rawJid.split('@')[0] } + : { + name: rawCustomer.name, + phone: rawCustomer.phone, + avatarUrl: rawCustomer.avatarUrl, + chatType: rawConversation?.whatsappChatType, + } + const jid = canonicalWhatsAppJid(rawJid, identity) + const conversationId = `${channel.id}:${jid}` + const currentBeforeSync = await getResource(tenantId, 'conversations', conversationId) || rawConversation const currentCustomer = currentBeforeSync?.customerId ? await getResource(tenantId, 'customers', currentBeforeSync.customerId) - : null - const identity = profileDue || !currentCustomer - ? await whatsappRequest( - `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/identity`, - { signal: AbortSignal.timeout(8_000) }, - ).catch(() => ({ name: chat.name || jid.split('@')[0] })) - : { - name: currentCustomer.name, - phone: currentCustomer.phone, - avatarUrl: currentCustomer.avatarUrl, - chatType: currentBeforeSync.whatsappChatType, - } + : rawCustomer if (profileDue && currentCustomer) { const customer = await patchResource(tenantId, 'customers', currentCustomer.id, { - name: !isWeakWhatsAppDisplayName(currentCustomer.name, jid) + name: !isWeakWhatsAppDisplayName(currentCustomer.name, rawJid) ? currentCustomer.name : identity?.name || currentCustomer.name, phone: identity?.phone || currentCustomer.phone, @@ -3468,7 +3580,7 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) { : `&after=${encodeURIComponent(historyCursor)}` : '' const history = await whatsappRequest( - `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(jid)}/history?limit=${options.full === true ? 2000 : 500}${cursorQuery}`, + `/accounts/${channel.gatewayAccountId}/chats/${encodeURIComponent(rawJid)}/history?limit=${options.full === true ? 2000 : 500}${cursorQuery}`, ).catch(() => ({ messages: [], beforeCursor: null, hasMoreBefore: false })) for (const raw of history.messages || []) { const result = await upsertWhatsAppMessage(tenantId, channel, raw, { identity, notify: false }) @@ -3480,7 +3592,7 @@ async function performWhatsAppTenantSync(tenantId, channel, options = {}) { ? history.hasMoreBefore && history.beforeCursor ? String(history.beforeCursor) : '' : history.hasMoreAfter && history.afterCursor ? String(history.afterCursor) : '' } while (historyCursor && historyPages < (options.full === true ? 100 : 20)) - messagesSynced += await mergeLegacyWhatsAppConversations(tenantId, channel, jid) + messagesSynced += await mergeLegacyWhatsAppConversations(tenantId, channel, jid, [rawJid]) const current = await getResource(tenantId, 'conversations', conversationId) if (current) { await patchResource(tenantId, 'conversations', conversationId, { @@ -4989,6 +5101,7 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) => if (['message.received', 'message.sent', 'message.edited'].includes(req.body.event) && message) { await upsertWhatsAppMessage(tenantId, channel, message, { notify: req.body.event === 'message.received', + eventOccurredAt: req.body.occurredAt, }) } else if (req.body.event === 'message.updated') { for (const update of req.body?.data?.updates || []) { @@ -4996,7 +5109,10 @@ app.post('/internal/webhooks/whatsapp/:tenantId/:channelId', async (req, res) => const messageId = String(update.id || '') if (!jid || !messageId) continue if (update.message) { - await upsertWhatsAppMessage(tenantId, channel, update.message, { notify: false }) + await upsertWhatsAppMessage(tenantId, channel, update.message, { + notify: false, + eventOccurredAt: req.body.occurredAt, + }) continue } const conversationId = `${channel.id}:${jid}` diff --git a/server/src/whatsapp-identity.js b/server/src/whatsapp-identity.js index 90d78b9..60fbc20 100644 --- a/server/src/whatsapp-identity.js +++ b/server/src/whatsapp-identity.js @@ -1,7 +1,16 @@ import crypto from 'node:crypto' +export function canonicalWhatsAppJid(jid, identity = {}) { + const raw = String(jid || '').trim().toLowerCase() + if (!raw || raw.endsWith('@g.us') || raw.endsWith('@newsletter')) return raw + const canonical = String(identity?.canonicalJid || '').trim().toLowerCase() + if (canonical.endsWith('@s.whatsapp.net')) return canonical + const phone = String(identity?.phone || '').replace(/\D/g, '') + return phone ? `${phone}@s.whatsapp.net` : raw +} + export function whatsappCustomerStableId(jid) { - const normalized = String(jid || '').trim().toLowerCase() + const normalized = canonicalWhatsAppJid(jid) const hash = crypto.createHash('sha256').update(normalized).digest('hex').slice(0, 24) return `wa-customer-${hash}` } diff --git a/server/test/business.test.js b/server/test/business.test.js index 05c8d36..722f6f5 100644 --- a/server/test/business.test.js +++ b/server/test/business.test.js @@ -1,6 +1,12 @@ import test from 'node:test' import assert from 'node:assert/strict' -import { countBy, dateKey, percent, telegramAttachment } from '../src/business.js' +import { + countBy, + dateKey, + normalizeBusinessRelationship, + percent, + telegramAttachment, +} from '../src/business.js' import { decryptEmailPassword, encryptEmailPassword } from '../src/email.js' import { HUMAN_REPLY_SYSTEM_PROMPT, @@ -50,6 +56,11 @@ test('业务指标工具处理百分比、分组和无效日期', () => { ) }) +test('业务关系分类使用受控类型和中文标签', () => { + assert.deepEqual(normalizeBusinessRelationship('supplier'), { type: 'supplier', label: '上游供应商' }) + assert.deepEqual(normalizeBusinessRelationship('unsupported'), { type: 'unknown', label: '待确认' }) +}) + test('Email 渠道凭据使用认证加密且可正确解密', () => { process.env.EMAIL_CREDENTIAL_KEY = 'lingxi-email-test-key-with-more-than-32-bytes' const encrypted = encryptEmailPassword('app-password-123') diff --git a/server/test/whatsapp-identity.test.js b/server/test/whatsapp-identity.test.js index aea727a..d5352c6 100644 --- a/server/test/whatsapp-identity.test.js +++ b/server/test/whatsapp-identity.test.js @@ -1,11 +1,19 @@ import assert from 'node:assert/strict' import test from 'node:test' import { + canonicalWhatsAppJid, isWeakWhatsAppDisplayName, reusableWhatsAppConversations, whatsappCustomerStableId, } from '../src/whatsapp-identity.js' +test('WhatsApp LID resolves to the same canonical phone identity', () => { + const phoneJid = '15551234567@s.whatsapp.net' + assert.equal(canonicalWhatsAppJid('987654321@lid', { phone: '+1 555 123 4567' }), phoneJid) + assert.equal(canonicalWhatsAppJid('987654321@lid', { canonicalJid: phoneJid }), phoneJid) + assert.equal(whatsappCustomerStableId(canonicalWhatsAppJid('987654321@lid', { phone: '+1 555 123 4567' })), whatsappCustomerStableId(phoneJid)) +}) + test('WhatsApp customer id remains stable after gateway reauthorization', () => { assert.equal( whatsappCustomerStableId('15551234567@s.whatsapp.net'), diff --git a/src/components/inbox/ConversationInsightBar.tsx b/src/components/inbox/ConversationInsightBar.tsx index c70a3fe..2a73881 100644 --- a/src/components/inbox/ConversationInsightBar.tsx +++ b/src/components/inbox/ConversationInsightBar.tsx @@ -25,7 +25,7 @@ export function ConversationInsightBar({ aiBackfill?: AiBackfillStatus }) { const stage = validStages.has(customer.stage) ? customer.stage : 'initial' - const summary = customer.aiProfile?.summary || conversation.summary || '收到更多客户消息后将补充客户画像' + const summary = customer.aiProfile?.summary || conversation.summary || '收到更多消息后将补充联系人画像' return (
- 客户洞察 + 对话洞察

{summary}

@@ -55,12 +55,17 @@ export function ConversationInsightBar({ 意向 + {customer.aiProfile?.relationshipLabel && ( + + {customer.aiProfile.relationshipLabel} + + )}
- diff --git a/src/components/inbox/InsightPanel.tsx b/src/components/inbox/InsightPanel.tsx index 2c8a840..4ff3074 100644 --- a/src/components/inbox/InsightPanel.tsx +++ b/src/components/inbox/InsightPanel.tsx @@ -249,14 +249,15 @@ export function InsightPanel({ 正在分析会话上下文… ) : ( -

{analysis || '收到更多客户消息后将补充客户画像。'}

+

{analysis || '收到更多消息后将补充联系人画像。'}

)} {/* AI 提取关键信息 */} {Object.values(customer.aiProfile || {}).some(Boolean) && (
{Object.entries({ + 关系类型: customer.aiProfile.relationshipLabel, 行业: customer.aiProfile.industry, - 采购角色: customer.aiProfile.purchasingRole, + 业务身份: customer.aiProfile.purchasingRole, 询价数量: customer.aiProfile.inquiryQuantity, 目标价格: customer.aiProfile.targetPrice, 预算: customer.aiProfile.budget, diff --git a/src/types/crm.ts b/src/types/crm.ts index 850bbbe..7aceb8f 100644 --- a/src/types/crm.ts +++ b/src/types/crm.ts @@ -14,6 +14,10 @@ export type CustomerStage = /** AI 自动提取的客户画像 */ export interface AiCustomerProfile { + relationshipType?: 'buyer' | 'supplier' | 'partner' | 'distributor' | 'procurement_agent' | 'logistics' | 'service_provider' | 'internal' | 'other' | 'unknown' + relationshipLabel?: string + relationshipConfidence?: number + relationshipEvidence?: string[] inquiryQuantity?: string targetPrice?: string moqRequirement?: string @@ -88,6 +92,7 @@ export interface Customer { createdAt: string isKey: boolean aiProfile: AiCustomerProfile + relationshipType?: AiCustomerProfile['relationshipType'] telegramUsername?: string telegramUserId?: string avatarUrl?: string diff --git a/src/types/messaging.ts b/src/types/messaging.ts index ea618e9..72bd430 100644 --- a/src/types/messaging.ts +++ b/src/types/messaging.ts @@ -69,6 +69,9 @@ export interface Message { }> telegramFileId?: number whatsappMessageId?: string + gatewayEventAt?: string + ingestedAt?: string + realtimeLatencyMs?: number whatsappContentType?: string attachmentMimeType?: string attachmentSize?: number diff --git a/whatsapp-gateway/src/index.js b/whatsapp-gateway/src/index.js index 0494298..408544c 100644 --- a/whatsapp-gateway/src/index.js +++ b/whatsapp-gateway/src/index.js @@ -125,7 +125,7 @@ async function saveWebhookDeliveries() { const sorted = [...webhookDeliveries.values()] .sort((a, b) => Date.parse(b.createdAt || '') - Date.parse(a.createdAt || '')) const active = sorted.filter((item) => !['delivered', 'dead'].includes(item.status)) - const terminal = sorted.filter((item) => ['delivered', 'dead'].includes(item.status)).slice(0, 10_000) + const terminal = sorted.filter((item) => ['delivered', 'dead'].includes(item.status)).slice(0, 2_000) const values = [...active, ...terminal] const keep = new Set(values.map((item) => item.id)) for (const id of webhookDeliveries.keys()) { @@ -134,11 +134,20 @@ async function saveWebhookDeliveries() { await atomicJson(webhookDeliveriesFile, values) } +let webhookDeliverySaveChain = Promise.resolve() +function queueWebhookDeliveriesSave() { + webhookDeliverySaveChain = webhookDeliverySaveChain + .catch(() => undefined) + .then(() => saveWebhookDeliveries()) + .catch((error) => log.error({ err: error }, 'webhook delivery persistence failed')) + return webhookDeliverySaveChain +} + async function saveOutboundJobs() { const sorted = [...outboundJobs.values()] .sort((a, b) => Date.parse(b.createdAt || '') - Date.parse(a.createdAt || '')) const active = sorted.filter((item) => !['sent', 'dead'].includes(item.status)) - const terminal = sorted.filter((item) => ['sent', 'dead'].includes(item.status)).slice(0, 10_000) + const terminal = sorted.filter((item) => ['sent', 'dead'].includes(item.status)).slice(0, 2_000) const values = [...active, ...terminal] const keep = new Set(values.map((item) => item.id)) for (const id of outboundJobs.keys()) { @@ -147,6 +156,15 @@ async function saveOutboundJobs() { await atomicJson(outboundJobsFile, values) } +let outboundJobsSaveChain = Promise.resolve() +function queueOutboundJobsSave() { + outboundJobsSaveChain = outboundJobsSaveChain + .catch(() => undefined) + .then(() => saveOutboundJobs()) + .catch((error) => log.error({ err: error }, 'outbound job persistence failed')) + return outboundJobsSaveChain +} + async function saveAccounts() { const values = [...accounts.values()].map(({ qrDataUrl, pairingCode, ...item }) => item) await atomicJson(accountsFile, values) @@ -484,7 +502,7 @@ async function processWebhookDelivery(deliveryId) { delivery.status = 'delivering' delivery.attempts = Number(delivery.attempts || 0) + 1 delivery.updatedAt = new Date().toISOString() - await saveWebhookDeliveries() + void queueWebhookDeliveriesSave() try { const signature = crypto.createHmac('sha256', webhookSecret) .update(delivery.payload) @@ -523,13 +541,19 @@ async function processWebhookDelivery(deliveryId) { } finally { delivery.updatedAt = new Date().toISOString() webhookDeliveriesProcessing.delete(deliveryId) - await saveWebhookDeliveries() + await queueWebhookDeliveriesSave() } } async function deliverWebhook(account, event, data) { if (!account.webhookUrl || !account.webhookActive || !accountWebhookSecret(account)) return - const id = crypto.randomUUID() + const messageId = ['message.received', 'message.sent', 'message.edited'].includes(event) + ? String(data?.message?.id || '') + : '' + const id = messageId + ? `wa-${crypto.createHash('sha256').update(`${account.id}:${event}:${messageId}`).digest('hex').slice(0, 32)}` + : crypto.randomUUID() + if (webhookDeliveries.has(id)) return const payload = JSON.stringify({ id, event, @@ -547,8 +571,8 @@ async function deliverWebhook(account, event, data) { createdAt: new Date().toISOString(), updatedAt: new Date().toISOString(), }) - await saveWebhookDeliveries() scheduleWebhookDelivery(id) + await queueWebhookDeliveriesSave() } function queuedOutboundMessage(job) { @@ -611,7 +635,7 @@ async function processOutboundJob(jobId) { job.status = 'queued' job.nextAttemptAt = new Date(Date.now() + 5_000).toISOString() job.updatedAt = new Date().toISOString() - await saveOutboundJobs() + await queueOutboundJobsSave() scheduleOutboundJob(job.id, 5_000) return queuedOutboundMessage(job) } @@ -621,7 +645,7 @@ async function processOutboundJob(jobId) { job.throttled = true job.nextAttemptAt = new Date(Date.now() + throttleDelay).toISOString() job.updatedAt = new Date().toISOString() - await saveOutboundJobs() + await queueOutboundJobsSave() scheduleOutboundJob(job.id, throttleDelay) return queuedOutboundMessage(job) } @@ -629,7 +653,7 @@ async function processOutboundJob(jobId) { job.status = 'sending' job.attempts = Number(job.attempts || 0) + 1 job.updatedAt = new Date().toISOString() - await saveOutboundJobs() + void queueOutboundJobsSave() try { const store = await loadStore(job.accountId) const quoted = job.quotedMessageId @@ -713,7 +737,7 @@ async function processOutboundJob(jobId) { } finally { job.updatedAt = new Date().toISOString() outboundJobsProcessing.delete(job.id) - await saveOutboundJobs() + await queueOutboundJobsSave() } } @@ -1415,7 +1439,7 @@ app.post('/accounts/:id/backup/restore', async (req, res) => { if (delivery.accountId === account.id) webhookDeliveries.delete(id) } for (const delivery of snapshot.webhookDeliveries || []) webhookDeliveries.set(delivery.id, delivery) - await Promise.all([saveAccounts(), saveOutboundJobs(), saveWebhookDeliveries()]) + await Promise.all([saveAccounts(), queueOutboundJobsSave(), queueWebhookDeliveriesSave()]) await connectAccount(account.id) res.json({ code: 0, data: { restored: true, account: accountPublic(account) } }) }) @@ -1575,12 +1599,16 @@ app.get('/accounts/:id/chats/:jid/identity', async (req, res) => { const group = jid.endsWith('@g.us') ? await sock.groupMetadata(jid).catch(() => store.groups.get(jid)) : null - const avatarUrl = await sock.profilePictureUrl(jid, 'image') - .catch(() => phoneJid !== jid ? sock.profilePictureUrl(phoneJid, 'image').catch(() => '') : '') + const includeProfile = req.query.includeProfile !== 'false' + const avatarUrl = includeProfile + ? await sock.profilePictureUrl(jid, 'image') + .catch(() => phoneJid !== jid ? sock.profilePictureUrl(phoneJid, 'image').catch(() => '') : '') + : '' res.json({ code: 0, data: { id: jid, + canonicalJid: phoneJid.endsWith('@s.whatsapp.net') ? phoneJid : jid, name: group?.subject || contact.name || contact.notify || contact.verifiedName || jid.split('@')[0], phone: phoneJid.endsWith('@s.whatsapp.net') ? phoneJid.split('@')[0] : '', avatarUrl, @@ -1663,7 +1691,7 @@ app.post('/accounts/:id/messages/send', async (req, res) => { updatedAt: new Date().toISOString(), } outboundJobs.set(job.id, job) - await saveOutboundJobs() + await queueOutboundJobsSave() const result = await processOutboundJob(job.id) res.json({ code: 0, data: result || queuedOutboundMessage(job) }) }) @@ -1776,7 +1804,7 @@ app.post('/accounts/:id/outbound-jobs/:jobId/retry', async (req, res) => { delete job.deadAt delete job.lastError delete job.nextAttemptAt - await saveOutboundJobs() + await queueOutboundJobsSave() const result = await processOutboundJob(job.id) res.json({ code: 0, data: result || queuedOutboundMessage(job) }) }) @@ -1835,7 +1863,7 @@ app.post('/accounts/:id/messages/media', upload.single('file'), async (req, res) job.filePath = path.join(outboundMediaDir, `${job.id}.bin`) await fs.writeFile(job.filePath, req.file.buffer, { mode: 0o600 }) outboundJobs.set(job.id, job) - await saveOutboundJobs() + await queueOutboundJobsSave() const result = await processOutboundJob(job.id) res.json({ code: 0, data: result || queuedOutboundMessage(job) }) })