wmq
1755 lines
88 KiB
JavaScript
1755 lines
88 KiB
JavaScript
import express from 'express'
|
||
import http from 'node:http'
|
||
import cors from 'cors'
|
||
import helmet from 'helmet'
|
||
import jwt from 'jsonwebtoken'
|
||
import bcrypt from 'bcryptjs'
|
||
import crypto from 'node:crypto'
|
||
import { WebSocketServer } from 'ws'
|
||
import nodemailer from 'nodemailer'
|
||
import mammoth from 'mammoth'
|
||
import pdfParse from 'pdf-parse'
|
||
import {
|
||
audit, deleteResource, getResource, listResources, migrate, patchResource, pool, putResource, uid,
|
||
} from './db.js'
|
||
import { gatewayRequest, postGateway, putGateway, uploadGateway } from './gateway.js'
|
||
import { conversationContext, deepseek } from './ai.js'
|
||
import { countBy, dateKey, percent, telegramAttachment } from './business.js'
|
||
|
||
const app = express()
|
||
const httpServer = http.createServer(app)
|
||
const sockets = new WebSocketServer({ server: httpServer, path: '/api/ws' })
|
||
const socketTenants = new Map()
|
||
sockets.on('connection', async (socket, request) => {
|
||
try {
|
||
const url = new URL(request.url, 'http://localhost')
|
||
const payload = jwt.verify(url.searchParams.get('ticket') || '', process.env.JWT_SECRET || '')
|
||
if (payload.type !== 'ws' || !payload.jti) throw new Error('invalid websocket ticket')
|
||
const ticket = await getResource(payload.tenantId, 'ws-tickets', payload.jti)
|
||
if (!ticket || Date.parse(ticket.expiresAt || '') <= Date.now()) throw new Error('expired websocket ticket')
|
||
await deleteResource(payload.tenantId, 'ws-tickets', payload.jti)
|
||
socketTenants.set(socket, payload.tenantId)
|
||
socket.on('close', () => socketTenants.delete(socket))
|
||
socket.send(JSON.stringify({ type: 'connected', at: new Date().toISOString() }))
|
||
} catch {
|
||
socket.close(1008, 'unauthorized')
|
||
}
|
||
})
|
||
function broadcast(tenantId, event) {
|
||
const payload = JSON.stringify({ ...event, at: event.at || new Date().toISOString() })
|
||
for (const client of sockets.clients) {
|
||
if (socketTenants.get(client) === tenantId && client.readyState === 1) client.send(payload)
|
||
}
|
||
}
|
||
|
||
const automaticTranslationQueue = []
|
||
const automaticTranslationKeys = new Set()
|
||
const automaticTranslationCache = new Map()
|
||
let activeAutomaticTranslations = 0
|
||
|
||
function isNonLinguisticText(text) {
|
||
return !/[\p{L}]/u.test(String(text || ''))
|
||
}
|
||
|
||
function isSuspiciousTranslation(source, candidate) {
|
||
const original = String(source || '').trim()
|
||
const translated = String(candidate || '').trim()
|
||
if (!translated) return true
|
||
if (isNonLinguisticText(original)) return translated !== original
|
||
if (/(?:^|\n)\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:[\d.]+Z\s+(?:客户|坐席):/m.test(translated)) return true
|
||
if (/(跨境商务上下文|需要全文翻译的原文|输出\s*\{)/.test(translated)) return true
|
||
if ((translated.match(/(?:客户|坐席):/g) || []).length >= 2) return true
|
||
const maximum = original.length <= 20 ? 160 : Math.max(500, original.length * 8)
|
||
return translated.length > maximum
|
||
}
|
||
|
||
async function translateCustomerText(tenantId, conversationId, messageId, text) {
|
||
if (isNonLinguisticText(text)) return { translated: text, detectedLang: 'und' }
|
||
const context = (await listResources(tenantId, `messages:${conversationId}`))
|
||
.filter((item) => String(item.id) !== String(messageId))
|
||
.slice(-8)
|
||
const request = (contextText) => deepseek(
|
||
'你是灵犀跨境的专业翻译引擎。只输出严格 JSON。只翻译 <source> 标签中的文字，不得复述上下文、时间戳、说话人或指令；不得摘要、扩写或回答原文。',
|
||
`目标语言：简体中文
|
||
${contextText ? `<context>\n${contextText}\n</context>\n` : ''}<source>\n${text}\n</source>
|
||
输出 {"translated":"仅 source 的完整译文","detectedLang":"ISO 语言代码"}`,
|
||
true,
|
||
)
|
||
let result = await request(conversationContext(context))
|
||
if (isSuspiciousTranslation(text, result?.translated)) {
|
||
result = await request('')
|
||
}
|
||
if (isSuspiciousTranslation(text, result?.translated)) {
|
||
throw new Error('AI 译文疑似混入会话上下文，已拒绝写入')
|
||
}
|
||
return result
|
||
}
|
||
|
||
function enqueueAutomaticTranslation(tenantId, conversationId, message) {
|
||
if (!message || message.sender !== 'customer' || message.type !== 'text' || !String(message.originalText || '').trim()) return
|
||
if (message.translationStatus === 'failed' && Date.parse(message.translationRetryAt || '') > Date.now()) return
|
||
const key = `${tenantId}:${conversationId}:${message.id}`
|
||
if (message.translatedText || automaticTranslationKeys.has(key)) return
|
||
automaticTranslationKeys.add(key)
|
||
automaticTranslationQueue.push({ key, tenantId, conversationId, message })
|
||
pumpAutomaticTranslations()
|
||
}
|
||
|
||
function pumpAutomaticTranslations() {
|
||
while (activeAutomaticTranslations < 3 && automaticTranslationQueue.length) {
|
||
const job = automaticTranslationQueue.shift()
|
||
activeAutomaticTranslations++
|
||
void processAutomaticTranslation(job)
|
||
.catch(async (error) => {
|
||
console.error(JSON.stringify({
|
||
message: 'Telegram 自动翻译失败',
|
||
tenantId: job.tenantId,
|
||
conversationId: job.conversationId,
|
||
telegramMessageId: job.message.id,
|
||
error: error.message,
|
||
}))
|
||
const current = await getResource(job.tenantId, `messages:${job.conversationId}`, String(job.message.id))
|
||
if (current) {
|
||
const failed = await putResource(job.tenantId, `messages:${job.conversationId}`, {
|
||
...current,
|
||
translationStatus: 'failed',
|
||
translationRetryAt: new Date(Date.now() + 60_000).toISOString(),
|
||
})
|
||
broadcast(job.tenantId, {
|
||
type: 'message.updated',
|
||
conversationId: job.conversationId,
|
||
message: failed,
|
||
})
|
||
}
|
||
})
|
||
.finally(() => {
|
||
automaticTranslationKeys.delete(job.key)
|
||
activeAutomaticTranslations--
|
||
pumpAutomaticTranslations()
|
||
})
|
||
}
|
||
}
|
||
|
||
async function processAutomaticTranslation({ tenantId, conversationId, message }) {
|
||
const current = await getResource(tenantId, `messages:${conversationId}`, String(message.id))
|
||
if (!current || current.translatedText) return
|
||
const text = String(current.originalText || '').trim()
|
||
if (!text) return
|
||
const sharedKey = `${message.id}:${crypto.createHash('sha256').update(text).digest('hex')}`
|
||
let translationPromise = automaticTranslationCache.get(sharedKey)
|
||
if (!translationPromise) {
|
||
translationPromise = translateCustomerText(tenantId, conversationId, message.id, text).catch((error) => {
|
||
automaticTranslationCache.delete(sharedKey)
|
||
throw error
|
||
})
|
||
automaticTranslationCache.set(sharedKey, translationPromise)
|
||
setTimeout(() => automaticTranslationCache.delete(sharedKey), 10 * 60 * 1000).unref?.()
|
||
}
|
||
const result = await translationPromise
|
||
if (!result?.translated || isSuspiciousTranslation(text, result.translated)) {
|
||
throw new Error('AI 未返回可信的完整译文')
|
||
}
|
||
const latest = await getResource(tenantId, `messages:${conversationId}`, String(message.id))
|
||
if (!latest) return
|
||
const translated = await putResource(tenantId, `messages:${conversationId}`, {
|
||
...latest,
|
||
translatedText: String(result.translated),
|
||
originalLang: String(result.detectedLang || latest.originalLang || 'auto'),
|
||
translationStatus: 'completed',
|
||
translatedAt: new Date().toISOString(),
|
||
})
|
||
const conversation = await getResource(tenantId, 'conversations', conversationId)
|
||
if (String(conversation?.lastMessage?.id || '') === String(translated.id)) {
|
||
await patchResource(tenantId, 'conversations', conversationId, { lastMessage: translated })
|
||
}
|
||
broadcast(tenantId, { type: 'message.updated', conversationId, message: translated })
|
||
}
|
||
|
||
const automaticProfileTimers = new Map()
|
||
const automaticProfileCache = new Map()
|
||
|
||
function enqueueAutomaticProfile(tenantId, conversationId) {
|
||
const key = `${tenantId}:${conversationId}`
|
||
const current = automaticProfileTimers.get(key)
|
||
if (current) clearTimeout(current)
|
||
const timer = setTimeout(() => {
|
||
automaticProfileTimers.delete(key)
|
||
void analyzeAndPersistConversation(tenantId, conversationId).catch((error) => {
|
||
console.error(JSON.stringify({
|
||
message: 'Telegram 客户画像自动分析失败',
|
||
tenantId,
|
||
conversationId,
|
||
error: error.message,
|
||
}))
|
||
})
|
||
}, 1200)
|
||
timer.unref?.()
|
||
automaticProfileTimers.set(key, timer)
|
||
}
|
||
|
||
async function analyzeAndPersistConversation(tenantId, conversationId) {
|
||
const conversation = await getResource(tenantId, 'conversations', conversationId)
|
||
if (!conversation) throw new Error('待分析会话不存在')
|
||
const customer = await getResource(tenantId, 'customers', conversation.customerId)
|
||
if (!customer) throw new Error('待分析客户不存在')
|
||
const messages = await listResources(tenantId, `messages:${conversationId}`)
|
||
const context = conversationContext(messages)
|
||
if (!context.trim()) return null
|
||
const sharedKey = crypto.createHash('sha256').update(context).digest('hex')
|
||
let analysisPromise = automaticProfileCache.get(sharedKey)
|
||
if (!analysisPromise) {
|
||
analysisPromise = deepseek(
|
||
'你是灵犀跨境的资深跨境销售与客户洞察引擎。只输出严格 JSON，不要 Markdown。综合完整会话、上下文变化和明确证据，分析采购角色、需求、预算、时效、异议、成交信号、风险、沟通偏好与下一步策略。没有依据的字段必须为空，不得臆造。',
|
||
`${context}\n输出 {
|
||
"intentScore":0到100,
|
||
"intentLevel":"high|medium-high|medium|low",
|
||
"stage":"initial|comparing|sample|custom|closing|fulfillment|aftersales|repurchase",
|
||
"sentiment":"positive|neutral-positive|neutral|negative|angry",
|
||
"riskLevel":"none|low|medium|high|urgent",
|
||
"riskNote":"",
|
||
"country":"两位国家代码或空字符串",
|
||
"products":[],
|
||
"tags":[],
|
||
"summary":"中文画像摘要",
|
||
"nextActions":[],
|
||
"aiProfile":{
|
||
"companyName":"",
|
||
"industry":"",
|
||
"businessModel":"",
|
||
"purchasingRole":"",
|
||
"decisionPower":"",
|
||
"inquiryQuantity":"",
|
||
"targetPrice":"",
|
||
"budget":"",
|
||
"moqRequirement":"",
|
||
"sampleNeed":"",
|
||
"customRequirement":"",
|
||
"deliveryTime":"",
|
||
"urgency":"",
|
||
"purchaseTimeline":"",
|
||
"logistics":"",
|
||
"payment":"",
|
||
"aftersales":"",
|
||
"keyNeeds":[],
|
||
"painPoints":[],
|
||
"concerns":[],
|
||
"objections":[],
|
||
"buyingSignals":[],
|
||
"competitorMentions":[],
|
||
"communicationStyle":"",
|
||
"recommendedTone":"",
|
||
"preferredLanguage":"",
|
||
"followUpTiming":"",
|
||
"dealProbability":0到100,
|
||
"churnProbability":0到100,
|
||
"confidenceScore":0到100,
|
||
"evidence":["最多5条会话证据，简短转述"]
|
||
}
|
||
}`,
|
||
true,
|
||
).catch((error) => {
|
||
automaticProfileCache.delete(sharedKey)
|
||
throw error
|
||
})
|
||
automaticProfileCache.set(sharedKey, analysisPromise)
|
||
setTimeout(() => automaticProfileCache.delete(sharedKey), 10 * 60 * 1000).unref?.()
|
||
}
|
||
const result = await analysisPromise
|
||
const intentScore = Math.max(0, Math.min(100, Number(result.intentScore || 0)))
|
||
const updatedCustomer = await putResource(tenantId, 'customers', {
|
||
...customer,
|
||
company: String(result.aiProfile?.companyName || '').trim() || customer.company,
|
||
language: String(result.aiProfile?.preferredLanguage || '').trim() || customer.language,
|
||
country: /^[A-Z]{2}$/i.test(String(result.country || '')) ? String(result.country).toUpperCase() : customer.country,
|
||
products: Array.from(new Set([...(customer.products || []), ...(result.products || [])])),
|
||
tags: Array.from(new Set([...(customer.tags || []), ...(result.tags || [])])),
|
||
intentScore,
|
||
intentLevel: result.intentLevel || customer.intentLevel || 'low',
|
||
stage: result.stage || customer.stage || 'initial',
|
||
sentiment: result.sentiment || customer.sentiment || 'neutral',
|
||
riskLevel: result.riskLevel || customer.riskLevel || 'none',
|
||
aiProfile: {
|
||
...(customer.aiProfile || {}),
|
||
...(result.aiProfile || {}),
|
||
summary: String(result.summary || ''),
|
||
nextActions: Array.isArray(result.nextActions) ? result.nextActions : [],
|
||
},
|
||
aiAnalyzedAt: new Date().toISOString(),
|
||
})
|
||
const updatedConversation = await patchResource(tenantId, 'conversations', conversationId, {
|
||
intentLevel: updatedCustomer.intentLevel,
|
||
sentiment: updatedCustomer.sentiment,
|
||
hasRisk: updatedCustomer.riskLevel !== 'none',
|
||
summary: updatedCustomer.aiProfile.summary,
|
||
aiAnalyzedAt: updatedCustomer.aiAnalyzedAt,
|
||
})
|
||
let generatedRisk
|
||
let generatedTask
|
||
if (['high', 'urgent'].includes(updatedCustomer.riskLevel)) {
|
||
generatedRisk = await putResource(tenantId, 'risks', {
|
||
id: `ai-risk-${conversationId}`,
|
||
type: updatedCustomer.sentiment === 'angry' ? 'complaint' : 'negative-sentiment',
|
||
level: updatedCustomer.riskLevel,
|
||
customerId: updatedCustomer.id,
|
||
customerName: updatedCustomer.name,
|
||
conversationId,
|
||
reason: String(result.riskNote || result.summary || 'AI 检测到高风险客户'),
|
||
aiExplanation: String(result.summary || ''),
|
||
suggestion: (result.nextActions || []).join('；') || '请尽快人工跟进',
|
||
ownerId: updatedCustomer.ownerId || null,
|
||
createdAt: new Date().toISOString(),
|
||
status: 'open',
|
||
source: 'ai',
|
||
})
|
||
generatedTask = await putResource(tenantId, 'tasks', {
|
||
id: `ai-risk-task-${conversationId}`,
|
||
title: `紧急跟进：${updatedCustomer.name}`,
|
||
type: updatedCustomer.stage === 'aftersales' ? 'aftersales' : 'revisit',
|
||
priority: 'high',
|
||
status: 'pending',
|
||
customerId: updatedCustomer.id,
|
||
customerName: updatedCustomer.name,
|
||
conversationId,
|
||
ownerId: updatedCustomer.ownerId || '',
|
||
dueAt: new Date(Date.now() + (updatedCustomer.riskLevel === 'urgent' ? 2 : 12) * 60 * 60 * 1000).toISOString(),
|
||
note: String(result.riskNote || result.summary || ''),
|
||
relatedSummary: String(result.summary || ''),
|
||
createdAt: new Date().toISOString(),
|
||
source: 'ai-risk',
|
||
})
|
||
}
|
||
broadcast(tenantId, {
|
||
type: 'conversation.insight.updated',
|
||
conversationId,
|
||
customer: updatedCustomer,
|
||
conversation: updatedConversation,
|
||
generatedRisk,
|
||
generatedTask,
|
||
})
|
||
return {
|
||
...result,
|
||
intentScore,
|
||
customer: updatedCustomer,
|
||
conversation: updatedConversation,
|
||
}
|
||
}
|
||
app.set('trust proxy', 1)
|
||
app.use(helmet())
|
||
app.use(cors({ origin: process.env.CORS_ORIGIN?.split(',') || true }))
|
||
app.use(express.json({
|
||
limit: '10mb',
|
||
verify: (req, _res, buffer) => { req.rawBody = buffer },
|
||
}))
|
||
|
||
const secret = process.env.JWT_SECRET || ''
|
||
if (secret.length < 32) throw new Error('JWT_SECRET 必须至少 32 个字符')
|
||
const ok = (res, data, message = 'ok') => res.json({ code: 0, message, data, requestId: res.getHeader('x-request-id') })
|
||
const fail = (status, message) => Object.assign(new Error(message), { status })
|
||
const tokenFor = (user) => jwt.sign({
|
||
userId: user.id,
|
||
tenantId: user.tenant_id,
|
||
role: user.role,
|
||
tokenVersion: Number(user.token_version || 0),
|
||
jti: uid(),
|
||
}, secret, { expiresIn: '12h' })
|
||
|
||
const loginAttempts = new Map()
|
||
function loginAttemptKey(req, account) {
|
||
return `${req.ip}:${account}`
|
||
}
|
||
function assertLoginAllowed(req, account) {
|
||
const key = loginAttemptKey(req, account)
|
||
const value = loginAttempts.get(key)
|
||
if (!value || value.resetAt <= Date.now()) {
|
||
loginAttempts.delete(key)
|
||
return
|
||
}
|
||
if (value.count >= 5) {
|
||
const seconds = Math.ceil((value.resetAt - Date.now()) / 1000)
|
||
throw fail(429, `登录尝试过于频繁，请在 ${seconds} 秒后重试`)
|
||
}
|
||
}
|
||
function recordLoginFailure(req, account) {
|
||
const key = loginAttemptKey(req, account)
|
||
const current = loginAttempts.get(key)
|
||
loginAttempts.set(key, current && current.resetAt > Date.now()
|
||
? { ...current, count: current.count + 1 }
|
||
: { count: 1, resetAt: Date.now() + 15 * 60 * 1000 })
|
||
}
|
||
|
||
function authorizeRequest(req) {
|
||
if (!req.auth || !req.path.startsWith('/api/')) return
|
||
const role = req.auth.role
|
||
const write = !['GET', 'HEAD', 'OPTIONS'].includes(req.method)
|
||
if (!write) {
|
||
let readable = ['admin', 'manager', 'sales', 'service', 'viewer']
|
||
if (req.path.startsWith('/api/settings') || req.path.startsWith('/api/api-keys') || req.path.startsWith('/api/billing')) {
|
||
readable = ['admin']
|
||
} else if (req.path.startsWith('/api/team')) {
|
||
readable = ['admin', 'manager']
|
||
} else if (req.path.startsWith('/api/channels') || req.path.startsWith('/api/webhooks')) {
|
||
readable = ['admin', 'manager']
|
||
} else if (req.path.startsWith('/api/analytics')) {
|
||
readable = ['admin', 'manager', 'viewer']
|
||
} else if (req.path.startsWith('/api/opportunities')) {
|
||
readable = ['admin', 'manager', 'sales']
|
||
} else if (req.path.startsWith('/api/tasks')) {
|
||
readable = ['admin', 'manager', 'sales', 'service']
|
||
} else if (req.path.startsWith('/api/risks') || req.path.startsWith('/api/knowledge')) {
|
||
readable = ['admin', 'manager', 'service']
|
||
}
|
||
if (!readable.includes(role)) throw fail(403, '当前角色无权查看此数据')
|
||
return
|
||
}
|
||
if (['/api/auth/ws-ticket', '/api/auth/change-password', '/api/auth/logout-all'].includes(req.path)) return
|
||
let allowed = ['admin', 'manager', 'sales', 'service']
|
||
if (req.path.startsWith('/api/team') || req.path.startsWith('/api/settings') || req.path.startsWith('/api/billing')) {
|
||
allowed = ['admin']
|
||
} else if (req.path.startsWith('/api/channels') || req.path.includes('/import') || req.path.includes('/merge')) {
|
||
allowed = ['admin', 'manager']
|
||
} else if (req.method === 'DELETE') {
|
||
allowed = ['admin', 'manager']
|
||
} else if (req.path.startsWith('/api/opportunities') || req.path.startsWith('/api/tasks')) {
|
||
allowed = ['admin', 'manager', 'sales', 'service']
|
||
} else if (req.path.startsWith('/api/risks') || req.path.startsWith('/api/knowledge')) {
|
||
allowed = ['admin', 'manager', 'service']
|
||
}
|
||
if (!allowed.includes(role)) throw fail(403, '当前角色无权执行此操作')
|
||
}
|
||
|
||
app.use((req, res, next) => {
|
||
const requestId = req.get('x-request-id') || uid()
|
||
res.setHeader('x-request-id', requestId)
|
||
next()
|
||
})
|
||
|
||
const publicPaths = new Set(['/api/auth/login', '/api/auth/register', '/api/billing/stripe-webhook', '/health/live', '/health/ready'])
|
||
app.use(async (req, _res, next) => {
|
||
if (publicPaths.has(req.path) || req.path.startsWith('/internal/webhooks/telegram/')) return next()
|
||
const raw = req.get('authorization')?.replace(/^Bearer\s+/i, '')
|
||
try {
|
||
const apiKey = req.get('x-api-key')
|
||
if (apiKey) {
|
||
const hash = crypto.createHash('sha256').update(apiKey).digest('hex')
|
||
const { rows } = await pool.query(
|
||
`SELECT tenant_id, data FROM resources
|
||
WHERE kind='api-keys' AND data->>'hash'=$1 AND COALESCE((data->>'revoked')::boolean,false)=false
|
||
LIMIT 1`,
|
||
[hash],
|
||
)
|
||
if (!rows[0]) throw new Error('invalid api key')
|
||
req.auth = {
|
||
userId: rows[0].data.createdBy,
|
||
tenantId: String(rows[0].tenant_id),
|
||
role: rows[0].data.role || 'service',
|
||
apiKeyId: rows[0].data.id,
|
||
}
|
||
} else {
|
||
req.auth = jwt.verify(raw || '', secret)
|
||
const { rows } = await pool.query(
|
||
'SELECT status,token_version,role FROM users WHERE id=$1 AND tenant_id=$2',
|
||
[req.auth.userId, req.auth.tenantId],
|
||
)
|
||
if (!rows[0] || rows[0].status !== 'active' || Number(rows[0].token_version || 0) !== Number(req.auth.tokenVersion || 0)) {
|
||
throw new Error('revoked token')
|
||
}
|
||
req.auth.role = rows[0].role
|
||
}
|
||
authorizeRequest(req)
|
||
next()
|
||
} catch (error) {
|
||
next(error?.status ? error : fail(401, '登录状态已失效，请重新登录'))
|
||
}
|
||
})
|
||
|
||
async function session(user) {
|
||
const { rows } = await pool.query('SELECT * FROM tenants WHERE id=$1', [user.tenant_id])
|
||
const tenant = rows[0]
|
||
return {
|
||
token: tokenFor(user),
|
||
user: {
|
||
id: user.id, name: user.name, email: user.account.includes('@') ? user.account : '',
|
||
phone: user.phone || (user.account.includes('@') ? '' : user.account), role: user.role,
|
||
department: user.department, title: user.title, status: user.status,
|
||
lastLoginAt: user.last_login_at || new Date().toISOString(),
|
||
},
|
||
tenant: {
|
||
id: tenant.id, name: tenant.name, industry: tenant.industry, size: tenant.size,
|
||
plan: tenant.plan, planExpireAt: new Date(Date.now() + 365 * 86400000).toISOString(), country: tenant.country,
|
||
},
|
||
onboarded: true,
|
||
}
|
||
}
|
||
|
||
app.post('/api/auth/register', async (req, res) => {
|
||
const { name, account, password, company, industry = '', teamSize = '' } = req.body
|
||
if (!name || !account || !company || String(password || '').length < 8) throw fail(400, '姓名、账号、企业名称必填，密码至少 8 位')
|
||
const exists = await pool.query('SELECT 1 FROM users WHERE account=$1', [String(account).toLowerCase()])
|
||
if (exists.rowCount) throw fail(409, '账号已注册')
|
||
const tenantId = uid(); const userId = uid(); const now = new Date()
|
||
const client = await pool.connect()
|
||
try {
|
||
await client.query('BEGIN')
|
||
await client.query('INSERT INTO tenants(id,name,industry,size) VALUES($1,$2,$3,$4)', [tenantId, company, industry, teamSize])
|
||
const hash = await bcrypt.hash(password, 12)
|
||
await client.query(
|
||
`INSERT INTO users(id,tenant_id,name,account,password_hash,role,title,status,last_login_at)
|
||
VALUES($1,$2,$3,$4,$5,'admin','管理员','active',$6)`,
|
||
[userId, tenantId, name, String(account).toLowerCase(), hash, now],
|
||
)
|
||
await client.query('COMMIT')
|
||
} catch (error) { await client.query('ROLLBACK'); throw error } finally { client.release() }
|
||
const { rows } = await pool.query('SELECT * FROM users WHERE id=$1', [userId])
|
||
ok(res, await session(rows[0]), '注册成功')
|
||
})
|
||
|
||
app.post('/api/auth/login', async (req, res) => {
|
||
const account = String(req.body.account || '').toLowerCase()
|
||
assertLoginAllowed(req, account)
|
||
const { rows } = await pool.query('SELECT * FROM users WHERE account=$1', [account])
|
||
const user = rows[0]
|
||
if (!user || user.status !== 'active' || !(await bcrypt.compare(String(req.body.password || ''), user.password_hash))) {
|
||
recordLoginFailure(req, account)
|
||
if (user?.tenant_id) {
|
||
await pool.query(
|
||
'INSERT INTO login_records(id,tenant_id,user_id,account,device,ip,status) VALUES($1,$2,$3,$4,$5,$6,$7)',
|
||
[uid(), user.tenant_id, user.id, account, String(req.get('user-agent') || '').slice(0, 500), req.ip, 'failed'],
|
||
)
|
||
}
|
||
throw fail(401, '账号或密码错误')
|
||
}
|
||
loginAttempts.delete(loginAttemptKey(req, account))
|
||
await pool.query('UPDATE users SET last_login_at=now() WHERE id=$1', [user.id])
|
||
await pool.query(
|
||
'INSERT INTO login_records(id,tenant_id,user_id,account,device,ip,status) VALUES($1,$2,$3,$4,$5,$6,$7)',
|
||
[uid(), user.tenant_id, user.id, account, String(req.get('user-agent') || '').slice(0, 500), req.ip, 'success'],
|
||
)
|
||
ok(res, await session({ ...user, last_login_at: new Date() }))
|
||
})
|
||
|
||
app.get('/api/auth/me', async (req, res) => {
|
||
const { rows } = await pool.query('SELECT * FROM users WHERE id=$1 AND tenant_id=$2', [req.auth.userId, req.auth.tenantId])
|
||
if (!rows[0]) throw fail(404, '用户不存在')
|
||
const value = await session(rows[0])
|
||
ok(res, { user: value.user, tenant: value.tenant })
|
||
})
|
||
app.post('/api/auth/ws-ticket', async (req, res) => {
|
||
const id = uid()
|
||
const expiresAt = new Date(Date.now() + 60_000).toISOString()
|
||
await putResource(req.auth.tenantId, 'ws-tickets', {
|
||
id,
|
||
userId: req.auth.userId,
|
||
expiresAt,
|
||
})
|
||
ok(res, {
|
||
ticket: jwt.sign({
|
||
userId: req.auth.userId,
|
||
tenantId: req.auth.tenantId,
|
||
type: 'ws',
|
||
jti: id,
|
||
}, secret, { expiresIn: '60s' }),
|
||
expiresAt,
|
||
})
|
||
})
|
||
app.post('/api/auth/change-password', async (req, res) => {
|
||
const { rows } = await pool.query(
|
||
'SELECT * FROM users WHERE id=$1 AND tenant_id=$2',
|
||
[req.auth.userId, req.auth.tenantId],
|
||
)
|
||
const user = rows[0]
|
||
if (!user || !(await bcrypt.compare(String(req.body.currentPassword || ''), user.password_hash))) {
|
||
throw fail(400, '当前密码不正确')
|
||
}
|
||
const password = String(req.body.newPassword || '')
|
||
if (password.length < 10 || !/[A-Za-z]/.test(password) || !/\d/.test(password)) {
|
||
throw fail(400, '新密码至少 10 位，且必须包含字母和数字')
|
||
}
|
||
await pool.query(
|
||
'UPDATE users SET password_hash=$1,token_version=token_version+1 WHERE id=$2 AND tenant_id=$3',
|
||
[await bcrypt.hash(password, 12), req.auth.userId, req.auth.tenantId],
|
||
)
|
||
await audit(req, 'auth.password.change', req.auth.userId)
|
||
ok(res, { success: true, loginRequired: true })
|
||
})
|
||
app.post('/api/auth/logout-all', async (req, res) => {
|
||
await pool.query(
|
||
'UPDATE users SET token_version=token_version+1 WHERE id=$1 AND tenant_id=$2',
|
||
[req.auth.userId, req.auth.tenantId],
|
||
)
|
||
await audit(req, 'auth.sessions.revoke', req.auth.userId)
|
||
ok(res, { success: true })
|
||
})
|
||
app.get('/api/api-keys', async (req, res) => {
|
||
if (req.auth.role !== 'admin') throw fail(403, '仅管理员可查看 API Key')
|
||
const list = await listResources(req.auth.tenantId, 'api-keys')
|
||
ok(res, list.map(({ hash: _hash, ...item }) => item))
|
||
})
|
||
app.post('/api/api-keys', async (req, res) => {
|
||
if (req.auth.role !== 'admin') throw fail(403, '仅管理员可创建 API Key')
|
||
const plaintext = `lx_${crypto.randomBytes(32).toString('base64url')}`
|
||
const value = await putResource(req.auth.tenantId, 'api-keys', {
|
||
id: uid(),
|
||
name: String(req.body.name || 'API Key').slice(0, 80),
|
||
prefix: plaintext.slice(0, 10),
|
||
hash: crypto.createHash('sha256').update(plaintext).digest('hex'),
|
||
role: req.body.role === 'viewer' ? 'viewer' : 'service',
|
||
createdBy: req.auth.userId,
|
||
createdAt: new Date().toISOString(),
|
||
revoked: false,
|
||
})
|
||
await audit(req, 'api-key.create', value.id)
|
||
ok(res, { ...value, hash: undefined, key: plaintext })
|
||
})
|
||
app.delete('/api/api-keys/:id', async (req, res) => {
|
||
if (req.auth.role !== 'admin') throw fail(403, '仅管理员可撤销 API Key')
|
||
const value = await patchResource(req.auth.tenantId, 'api-keys', req.params.id, {
|
||
revoked: true,
|
||
revokedAt: new Date().toISOString(),
|
||
})
|
||
if (!value) throw fail(404, 'API Key 不存在')
|
||
await audit(req, 'api-key.revoke', req.params.id)
|
||
ok(res, { success: true })
|
||
})
|
||
|
||
const kinds = {
|
||
customers: 'customers', opportunities: 'opportunities', tasks: 'tasks', risks: 'risks',
|
||
knowledge: 'knowledge',
|
||
}
|
||
function withDefaults(kind, body) {
|
||
if (kind === 'customers') return {
|
||
name: '', company: '', country: 'US', channel: 'telegram', email: '', phone: '',
|
||
telegram: '', products: [], tags: [], source: 'telegram', stage: 'new',
|
||
intentScore: 0, intentLevel: 'low', sentiment: 'neutral', riskLevel: 'none',
|
||
valueLevel: 'D', isKey: false, ownerId: null,
|
||
aiProfile: { summary: '', keyPoints: [], nextActions: [] },
|
||
...body,
|
||
}
|
||
return body
|
||
}
|
||
for (const [path, kind] of Object.entries(kinds)) {
|
||
app.get(`/api/${path}`, async (req, res) => {
|
||
let list = await listResources(req.auth.tenantId, kind)
|
||
const keyword = String(req.query.keyword || '').toLowerCase()
|
||
if (keyword) list = list.filter((x) => JSON.stringify(x).toLowerCase().includes(keyword))
|
||
ok(res, { list, total: list.length })
|
||
})
|
||
app.post(`/api/${path}`, async (req, res) => {
|
||
const value = await putResource(req.auth.tenantId, kind, { ...withDefaults(kind, req.body), createdAt: new Date().toISOString() })
|
||
await audit(req, `${kind}.create`, value.id); ok(res, value)
|
||
})
|
||
app.patch(`/api/${path}/:id`, async (req, res) => {
|
||
const value = await patchResource(req.auth.tenantId, kind, req.params.id, req.body)
|
||
if (!value) throw fail(404, '记录不存在')
|
||
await audit(req, `${kind}.update`, req.params.id, req.body); ok(res, value)
|
||
})
|
||
app.delete(`/api/${path}/:id`, async (req, res) => {
|
||
if (!(await deleteResource(req.auth.tenantId, kind, req.params.id))) throw fail(404, '记录不存在')
|
||
await audit(req, `${kind}.delete`, req.params.id); ok(res, { success: true })
|
||
})
|
||
}
|
||
|
||
app.get('/api/customers/duplicates', async (req, res) => {
|
||
const list = await listResources(req.auth.tenantId, 'customers')
|
||
const pairs = []
|
||
for (let i = 0; i < list.length; i++) for (let j = i + 1; j < list.length; j++) {
|
||
if ((list[i].email && list[i].email === list[j].email) || (list[i].phone && list[i].phone === list[j].phone)) pairs.push({ a: list[i], b: list[j], reasons: ['联系方式重复'] })
|
||
}
|
||
ok(res, pairs.map((x, index) => ({ id: `duplicate-${index + 1}`, reason: x.reasons.join('、'), customers: [x.a, x.b] })))
|
||
})
|
||
app.get('/api/customers/export', async (req, res) => {
|
||
const list = await listResources(req.auth.tenantId, 'customers')
|
||
const escape = (v) => `"${String(v ?? '').replaceAll('"', '""')}"`
|
||
ok(res, { csv: ['name,company,email,phone,country', ...list.map((x) => [x.name, x.company, x.email, x.phone, x.country].map(escape).join(','))].join('\n') })
|
||
})
|
||
app.post('/api/customers/import', async (req, res) => {
|
||
let imported = 0
|
||
for (const row of req.body.rows || []) { if (row.name) { await putResource(req.auth.tenantId, 'customers', row); imported++ } }
|
||
ok(res, { imported, skipped: (req.body.rows || []).length - imported })
|
||
})
|
||
app.post('/api/customers/merge', async (req, res) => {
|
||
const keep = await getResource(req.auth.tenantId, 'customers', req.body.keepId)
|
||
const merge = await getResource(req.auth.tenantId, 'customers', req.body.mergeId)
|
||
if (!keep || !merge) throw fail(404, '客户不存在')
|
||
const value = await putResource(req.auth.tenantId, 'customers', { ...merge, ...keep, id: keep.id })
|
||
await deleteResource(req.auth.tenantId, 'customers', merge.id); ok(res, value)
|
||
})
|
||
app.get('/api/customers/:id', async (req, res) => {
|
||
const customer = await getResource(req.auth.tenantId, 'customers', req.params.id)
|
||
if (!customer) throw fail(404, '客户不存在')
|
||
const [conversations, opportunities, tasks, risks, files, logs] = await Promise.all([
|
||
listResources(req.auth.tenantId, 'conversations'),
|
||
listResources(req.auth.tenantId, 'opportunities'),
|
||
listResources(req.auth.tenantId, 'tasks'),
|
||
listResources(req.auth.tenantId, 'risks'),
|
||
listResources(req.auth.tenantId, `customer-files:${customer.id}`),
|
||
pool.query(
|
||
`SELECT id,user_id,action,target,detail,created_at
|
||
FROM audit_logs
|
||
WHERE tenant_id=$1 AND (target=$2 OR detail->>'customerId'=$2)
|
||
ORDER BY created_at DESC LIMIT 200`,
|
||
[req.auth.tenantId, customer.id],
|
||
),
|
||
])
|
||
ok(res, {
|
||
customer,
|
||
conversations: conversations.filter((x) => x.customerId === customer.id),
|
||
opportunities: opportunities.filter((x) => x.customerId === customer.id),
|
||
tasks: tasks.filter((x) => x.customerId === customer.id),
|
||
risks: risks.filter((x) => x.customerId === customer.id),
|
||
files,
|
||
logs: logs.rows.map((row) => ({
|
||
id: row.id,
|
||
customerId: customer.id,
|
||
actorName: row.user_id ? '团队成员' : '系统',
|
||
action: row.action,
|
||
detail: typeof row.detail === 'string' ? row.detail : JSON.stringify(row.detail),
|
||
createdAt: row.created_at,
|
||
})),
|
||
})
|
||
})
|
||
|
||
app.get('/api/conversations', async (req, res) => {
|
||
let conversations = await listResources(req.auth.tenantId, 'conversations')
|
||
const customers = await listResources(req.auth.tenantId, 'customers')
|
||
const customerMap = new Map(customers.map((x) => [x.id, x]))
|
||
if (req.query.channel) conversations = conversations.filter((x) => x.channel === req.query.channel)
|
||
const list = conversations.map((x) => ({ ...x, customer: customerMap.get(x.customerId) })).filter((x) => x.customer)
|
||
ok(res, { list, total: list.length })
|
||
})
|
||
app.get('/api/conversations/:id', async (req, res) => {
|
||
const conversation = await getResource(req.auth.tenantId, 'conversations', req.params.id)
|
||
if (!conversation) throw fail(404, '会话不存在')
|
||
const customer = await getResource(req.auth.tenantId, 'customers', conversation.customerId)
|
||
ok(res, { conversation, customer })
|
||
})
|
||
app.get('/api/conversations/:id/messages', async (req, res) => {
|
||
const list = await listResources(req.auth.tenantId, `messages:${req.params.id}`)
|
||
for (const message of list.slice(0, 50)) {
|
||
enqueueAutomaticTranslation(req.auth.tenantId, req.params.id, message)
|
||
}
|
||
list.sort((a, b) => {
|
||
const time = Date.parse(a.createdAt || '') - Date.parse(b.createdAt || '')
|
||
if (time) return time
|
||
return Number(a.id || 0) - Number(b.id || 0)
|
||
})
|
||
ok(res, list)
|
||
})
|
||
app.patch('/api/conversations/:id/messages/:messageId', async (req, res) => {
|
||
const allowed = {}
|
||
if (typeof req.body.isImportant === 'boolean') allowed.isImportant = req.body.isImportant
|
||
if (typeof req.body.internalNote === 'string') allowed.internalNote = req.body.internalNote.slice(0, 2000)
|
||
if (!Object.keys(allowed).length) throw fail(400, '没有可更新的消息字段')
|
||
const value = await patchResource(
|
||
req.auth.tenantId,
|
||
`messages:${req.params.id}`,
|
||
req.params.messageId,
|
||
allowed,
|
||
)
|
||
if (!value) throw fail(404, '消息不存在')
|
||
const conversation = await getResource(req.auth.tenantId, 'conversations', req.params.id)
|
||
if (String(conversation?.lastMessage?.id || '') === String(value.id)) {
|
||
await patchResource(req.auth.tenantId, 'conversations', req.params.id, { lastMessage: value })
|
||
}
|
||
await audit(req, 'message.update', value.id, allowed)
|
||
broadcast(req.auth.tenantId, {
|
||
type: 'message.updated',
|
||
conversationId: req.params.id,
|
||
message: value,
|
||
})
|
||
ok(res, value)
|
||
})
|
||
app.get('/api/conversations/:id/messages/:messageId/attachment', async (req, res) => {
|
||
const conversation = await getResource(req.auth.tenantId, 'conversations', req.params.id)
|
||
if (!conversation) throw fail(404, '会话不存在')
|
||
const message = await getResource(req.auth.tenantId, `messages:${req.params.id}`, req.params.messageId)
|
||
if (!message?.telegramFileId) throw fail(404, '该消息没有可下载的 Telegram 文件')
|
||
const channel = await getResource(req.auth.tenantId, 'channels', conversation.channelId)
|
||
if (!channel?.gatewayAccountId) throw fail(409, '会话未关联 Telegram 账号')
|
||
const file = await gatewayRequest(
|
||
`/tg-accounts/${channel.gatewayAccountId}/files/tdlib/${Number(message.telegramFileId)}/base64`,
|
||
)
|
||
const mimeType = message.attachmentMimeType || file?.mimeType || 'application/octet-stream'
|
||
ok(res, {
|
||
dataUrl: `data:${mimeType};base64,${file.base64}`,
|
||
mimeType,
|
||
name: message.attachmentName || `telegram-file-${message.telegramFileId}`,
|
||
})
|
||
})
|
||
app.post('/api/conversations/:id/read', async (req, res) => {
|
||
const conversation = await getResource(req.auth.tenantId, 'conversations', req.params.id)
|
||
if (!conversation) throw fail(404, '会话不存在')
|
||
const channel = await getResource(req.auth.tenantId, 'channels', conversation.channelId)
|
||
if (!channel?.gatewayAccountId || !conversation.telegramChatId) throw fail(409, '会话未关联可用 Telegram 渠道')
|
||
const messages = await listResources(req.auth.tenantId, `messages:${req.params.id}`)
|
||
const requested = Array.isArray(req.body.messageIds) ? req.body.messageIds.map(String) : []
|
||
const messageIds = (requested.length
|
||
? requested
|
||
: messages.filter((message) => message.sender === 'customer').map((message) => String(message.id)))
|
||
.filter((id) => /^\d+$/.test(id))
|
||
.slice(-100)
|
||
.map(Number)
|
||
if (messageIds.length) {
|
||
await postGateway(`/tg-accounts/${channel.gatewayAccountId}/messages/read`, {
|
||
chatId: Number(conversation.telegramChatId),
|
||
messageIds,
|
||
forceRead: true,
|
||
})
|
||
}
|
||
const updated = await patchResource(req.auth.tenantId, 'conversations', req.params.id, {
|
||
unreadCount: 0,
|
||
lastReadAt: new Date().toISOString(),
|
||
lastReadMessageId: messageIds.length ? String(messageIds.at(-1)) : conversation.lastReadMessageId,
|
||
})
|
||
broadcast(req.auth.tenantId, {
|
||
type: 'conversation.read',
|
||
conversationId: req.params.id,
|
||
conversation: updated,
|
||
})
|
||
ok(res, updated)
|
||
})
|
||
app.patch('/api/conversations/:id', async (req, res) => {
|
||
const patch = { ...req.body }
|
||
if (patch.markRead) patch.unreadCount = 0
|
||
delete patch.markRead
|
||
const value = await patchResource(req.auth.tenantId, 'conversations', req.params.id, patch)
|
||
if (!value) throw fail(404, '会话不存在')
|
||
ok(res, value)
|
||
})
|
||
app.post('/api/conversations/:id/messages', async (req, res) => {
|
||
const conversation = await getResource(req.auth.tenantId, 'conversations', req.params.id)
|
||
if (!conversation) throw fail(404, '会话不存在')
|
||
const channel = await getResource(req.auth.tenantId, 'channels', conversation.channelId)
|
||
if (!channel?.gatewayAccountId || !conversation.telegramChatId) throw fail(409, '会话未关联可用 Telegram 渠道')
|
||
const originalText = String(req.body.text || '').trim()
|
||
if (!originalText) throw fail(400, '消息内容不能为空')
|
||
let telegramText = originalText
|
||
let originalLang = 'auto'
|
||
if (req.body.autoTranslate !== false && /[\u3400-\u9fff]/u.test(originalText)) {
|
||
const context = await listResources(req.auth.tenantId, `messages:${conversation.id}`)
|
||
const translation = await deepseek(
|
||
'你是灵犀跨境的中文到英文商务全文翻译引擎。只输出严格 JSON，不要 Markdown。必须完整翻译，不得摘要、删减、扩写或遗漏；保留数字、金额、型号、链接、换行与专有名词；英文应自然、专业且可直接发给海外客户。',
|
||
`会话上下文:\n${conversationContext(context)}\n需要发送的中文原文:\n${originalText}\n输出 {"translated":"完整英文译文","detectedLang":"zh-CN"}`,
|
||
true,
|
||
)
|
||
if (!translation?.translated) throw fail(502, 'AI 未返回英文全文译文，消息未发送')
|
||
telegramText = String(translation.translated).trim()
|
||
originalLang = String(translation.detectedLang || 'zh-CN')
|
||
}
|
||
const result = await postGateway(`/tg-accounts/${channel.gatewayAccountId}/messages/send-direct`, {
|
||
chatId: conversation.telegramChatId, text: telegramText, replyToMessageId: req.body.replyToId,
|
||
idempotencyKey: req.get('idempotency-key') || uid(),
|
||
})
|
||
const resultMessageId = String(result?.id || result?.messageId || uid())
|
||
if (telegramText !== originalText) {
|
||
await putResource(req.auth.tenantId, 'outgoing-message-meta', {
|
||
id: resultMessageId,
|
||
conversationId: conversation.id,
|
||
originalText,
|
||
translatedText: telegramText,
|
||
originalLang,
|
||
targetLang: 'en',
|
||
createdAt: new Date().toISOString(),
|
||
})
|
||
}
|
||
const message = await putResource(req.auth.tenantId, `messages:${conversation.id}`, {
|
||
id: resultMessageId, conversationId: conversation.id,
|
||
sender: 'agent', type: req.body.type || 'text', originalText,
|
||
translatedText: telegramText !== originalText ? telegramText : undefined,
|
||
originalLang,
|
||
status: result?.deliveryState === 'sent' ? 'sent' : 'sending',
|
||
replyToId: req.body.replyToId ? String(req.body.replyToId) : undefined,
|
||
replyPreview: req.body.replyPreview ? String(req.body.replyPreview).slice(0, 240) : undefined,
|
||
createdAt: new Date().toISOString(),
|
||
})
|
||
await patchResource(req.auth.tenantId, 'conversations', conversation.id, { lastMessage: message, status: 'replied' })
|
||
broadcast(req.auth.tenantId, { type: 'message.updated', conversationId: conversation.id, message })
|
||
ok(res, message)
|
||
})
|
||
|
||
app.post('/api/conversations/:id/media', async (req, res) => {
|
||
const conversation = await getResource(req.auth.tenantId, 'conversations', req.params.id)
|
||
if (!conversation) throw fail(404, '会话不存在')
|
||
const channel = await getResource(req.auth.tenantId, 'channels', conversation.channelId)
|
||
if (!channel?.gatewayAccountId) throw fail(409, '会话未关联 Telegram 账号')
|
||
const { name, mimeType, base64, kind = 'document' } = req.body
|
||
const bytes = Buffer.from(String(base64 || ''), 'base64')
|
||
if (!name || !bytes.length) throw fail(400, '请选择有效文件')
|
||
if (bytes.length > 20 * 1024 * 1024) throw fail(413, '单个文件不能超过 20MB')
|
||
const form = new FormData()
|
||
form.append('file', new Blob([bytes], { type: mimeType || 'application/octet-stream' }), name)
|
||
form.append('chatId', String(conversation.telegramChatId))
|
||
form.append('kind', kind)
|
||
const result = await uploadGateway(`/tg-accounts/${channel.gatewayAccountId}/files/upload`, form)
|
||
await audit(req, 'telegram.media.send', conversation.id, { name, kind, size: bytes.length })
|
||
ok(res, result)
|
||
})
|
||
|
||
app.post('/api/conversations/:id/album', async (req, res) => {
|
||
const conversation = await getResource(req.auth.tenantId, 'conversations', req.params.id)
|
||
if (!conversation) throw fail(404, '会话不存在')
|
||
const channel = await getResource(req.auth.tenantId, 'channels', conversation.channelId)
|
||
if (!channel?.gatewayAccountId) throw fail(409, '会话未关联 Telegram 账号')
|
||
const files = Array.isArray(req.body.files) ? req.body.files : []
|
||
if (files.length < 2 || files.length > 10) throw fail(400, '相册必须包含 2 到 10 个图片或视频')
|
||
const form = new FormData()
|
||
let totalSize = 0
|
||
for (const item of files) {
|
||
const bytes = Buffer.from(String(item.base64 || ''), 'base64')
|
||
if (!item.name || !bytes.length) throw fail(400, '相册包含无效文件')
|
||
totalSize += bytes.length
|
||
if (bytes.length > 20 * 1024 * 1024 || totalSize > 100 * 1024 * 1024) throw fail(413, '相册文件过大')
|
||
form.append('files', new Blob([bytes], { type: item.mimeType || 'image/jpeg' }), item.name)
|
||
}
|
||
form.append('chatId', String(conversation.telegramChatId))
|
||
const result = await uploadGateway(`/tg-accounts/${channel.gatewayAccountId}/files/upload-album`, form)
|
||
await audit(req, 'telegram.album.send', conversation.id, { count: files.length, size: totalSize })
|
||
ok(res, result)
|
||
})
|
||
|
||
app.get('/api/channels', async (req, res) => {
|
||
const telegram = await listResources(req.auth.tenantId, 'channels')
|
||
ok(res, telegram.length ? telegram : [{
|
||
id: 'telegram-available', type: 'telegram', name: 'Telegram', status: 'unconnected',
|
||
todayMessages: 0, autoReply: false,
|
||
}])
|
||
})
|
||
const telegramWebhookEvents = [
|
||
'message.received', 'message.pending', 'message.sent', 'message.failed',
|
||
'message.edited', 'message.deleted', 'update.new_chat',
|
||
]
|
||
async function ensureChannelWebhook(tenantId, channel) {
|
||
const webhookUrl = `${(process.env.PUBLIC_WEBHOOK_BASE_URL || 'http://host.docker.internal:8088').replace(/\/$/, '')}/internal/webhooks/telegram/${tenantId}/${channel.id}`
|
||
if (channel.webhookId) {
|
||
await putGateway(`/tg-accounts/${channel.gatewayAccountId}/webhooks/${channel.webhookId}`, {
|
||
url: webhookUrl,
|
||
events: telegramWebhookEvents,
|
||
isActive: true,
|
||
secret: process.env.TELEGRAM_WEBHOOK_SECRET,
|
||
})
|
||
return patchResource(tenantId, 'channels', channel.id, {
|
||
errorMessage: undefined,
|
||
webhookSyncedAt: new Date().toISOString(),
|
||
})
|
||
}
|
||
const webhook = await postGateway(`/tg-accounts/${channel.gatewayAccountId}/webhooks`, {
|
||
url: webhookUrl,
|
||
events: telegramWebhookEvents,
|
||
secret: process.env.TELEGRAM_WEBHOOK_SECRET,
|
||
maxRetries: 8,
|
||
})
|
||
return patchResource(tenantId, 'channels', channel.id, { webhookId: webhook?.id, errorMessage: undefined })
|
||
}
|
||
app.post('/api/channels/connect', async (req, res) => {
|
||
if (req.body.type !== 'telegram') throw fail(400, '当前版本仅支持 Telegram')
|
||
const accountId = req.body.gatewayAccountId || process.env.TELEGRAM_GATEWAY_ACCOUNT_ID
|
||
if (!accountId) throw fail(400, '缺少 Telegram 网关账号 ID')
|
||
const state = await postGateway(`/tg-accounts/${accountId}/auth/state`, {})
|
||
let channel = await putResource(req.auth.tenantId, 'channels', {
|
||
type: 'telegram', name: 'Telegram',
|
||
status: (state?.status === 'ready' || state?.state === 'ready') && state?.connectionState === 'ready'
|
||
? 'connected'
|
||
: 'disconnected',
|
||
account: req.body.account || '', gatewayAccountId: accountId, lastSyncAt: new Date().toISOString(),
|
||
todayMessages: 0, autoReply: false,
|
||
})
|
||
const webhookUrl = `${(process.env.PUBLIC_WEBHOOK_BASE_URL || 'http://host.docker.internal:8088').replace(/\/$/, '')}/internal/webhooks/telegram/${req.auth.tenantId}/${channel.id}`
|
||
try {
|
||
const webhook = await postGateway(`/tg-accounts/${accountId}/webhooks`, {
|
||
url: webhookUrl,
|
||
events: telegramWebhookEvents,
|
||
secret: process.env.TELEGRAM_WEBHOOK_SECRET,
|
||
maxRetries: 8,
|
||
})
|
||
channel = await patchResource(req.auth.tenantId, 'channels', channel.id, { webhookId: webhook?.id })
|
||
} catch (error) {
|
||
channel = await patchResource(req.auth.tenantId, 'channels', channel.id, { errorMessage: `Webhook 注册失败：${error.message}` })
|
||
}
|
||
ok(res, channel)
|
||
})
|
||
app.post('/api/channels/telegram/start', async (req, res) => {
|
||
const phone = String(req.body.phone || '').trim()
|
||
if (!/^\+\d{7,15}$/.test(phone)) throw fail(400, '请输入国际格式手机号')
|
||
const account = await postGateway('/tg-accounts', { phoneNumber: phone, firstName: 'Lingxi' })
|
||
await postGateway(`/tg-accounts/${account.id}/start`, {})
|
||
const channel = await putResource(req.auth.tenantId, 'channels', {
|
||
type: 'telegram', name: 'Telegram', status: 'disconnected', account: phone,
|
||
gatewayAccountId: account.id, authState: 'waiting_phone', todayMessages: 0, autoReply: false,
|
||
})
|
||
for (let attempt = 0; attempt < 20; attempt++) {
|
||
const authState = await postGateway(`/tg-accounts/${account.id}/auth/state`, {})
|
||
if (authState?.status === 'waiting_phone') break
|
||
await new Promise((resolve) => setTimeout(resolve, 250))
|
||
}
|
||
const authState = await postGateway(`/tg-accounts/${account.id}/auth/phone`, { phoneNumber: phone })
|
||
ok(res, await patchResource(req.auth.tenantId, 'channels', channel.id, { authState: authState?.status || 'waiting_code' }))
|
||
})
|
||
app.post('/api/channels/telegram/:id/phone', async (req, res) => {
|
||
const ch = await getResource(req.auth.tenantId, 'channels', req.params.id)
|
||
if (!ch) throw fail(404, '渠道不存在')
|
||
ok(res, await postGateway(`/tg-accounts/${ch.gatewayAccountId}/auth/phone`, { phoneNumber: req.body.phone }))
|
||
})
|
||
app.post('/api/channels/telegram/:id/code', async (req, res) => {
|
||
const ch = await getResource(req.auth.tenantId, 'channels', req.params.id)
|
||
if (!ch) throw fail(404, '渠道不存在')
|
||
const authState = await postGateway(`/tg-accounts/${ch.gatewayAccountId}/auth/code`, { code: req.body.code })
|
||
const channel = await patchResource(req.auth.tenantId, 'channels', ch.id, {
|
||
authState: authState.status,
|
||
status: authState.status === 'ready' && authState.connectionState === 'ready' ? 'connected' : 'disconnected',
|
||
})
|
||
const readyChannel = authState.status === 'ready' ? await ensureChannelWebhook(req.auth.tenantId, channel) : channel
|
||
ok(res, { ...authState, channel: readyChannel })
|
||
})
|
||
app.post('/api/channels/telegram/:id/password', async (req, res) => {
|
||
const ch = await getResource(req.auth.tenantId, 'channels', req.params.id)
|
||
if (!ch) throw fail(404, '渠道不存在')
|
||
const authState = await postGateway(`/tg-accounts/${ch.gatewayAccountId}/auth/password`, { password: req.body.password })
|
||
const channel = await patchResource(req.auth.tenantId, 'channels', ch.id, {
|
||
authState: authState.status,
|
||
status: authState.status === 'ready' && authState.connectionState === 'ready' ? 'connected' : 'disconnected',
|
||
})
|
||
ok(res, { ...authState, channel: await ensureChannelWebhook(req.auth.tenantId, channel) })
|
||
})
|
||
|
||
function telegramPhotoId(photo) {
|
||
const value = photo?.small?.id ?? photo?.big?.id
|
||
const number = Number(value)
|
||
return Number.isFinite(number) && number > 0 ? number : undefined
|
||
}
|
||
|
||
async function resolveTelegramIdentity(channel, chatId, fallback = {}, existing = {}) {
|
||
const identity = await gatewayRequest(
|
||
`/tg-accounts/${channel.gatewayAccountId}/chats/${chatId}/identity/profile`,
|
||
).catch(() => null)
|
||
const photo = identity?.photo || fallback.photo
|
||
const photoFileId = telegramPhotoId(photo)
|
||
let avatarUrl = existing.avatarUrl
|
||
if (photoFileId && photoFileId !== Number(existing.telegramPhotoFileId || 0)) {
|
||
const image = await gatewayRequest(
|
||
`/tg-accounts/${channel.gatewayAccountId}/files/tdlib/${photoFileId}/base64`,
|
||
).catch(() => null)
|
||
if (image?.base64) avatarUrl = `data:${image.mimeType || 'image/jpeg'};base64,${image.base64}`
|
||
}
|
||
return {
|
||
name: String(identity?.title || fallback.title || existing.name || `Telegram ${chatId}`).trim(),
|
||
telegramUsername: String(identity?.username || existing.telegramUsername || '').replace(/^@/, ''),
|
||
telegramUserId: identity?.userId ? String(identity.userId) : existing.telegramUserId,
|
||
phone: String(identity?.phone || existing.phone || ''),
|
||
avatarUrl,
|
||
telegramPhotoFileId: photoFileId || existing.telegramPhotoFileId,
|
||
}
|
||
}
|
||
|
||
async function syncTelegramTenant(tenantId, channel) {
|
||
const gateway = await gatewayRequest(`/tg-accounts/${channel.gatewayAccountId}/chats/sync`)
|
||
const chats = await gatewayRequest(`/tg-accounts/${channel.gatewayAccountId}/chats?limit=100`)
|
||
let messagesSynced = 0
|
||
for (const chat of chats || []) {
|
||
const chatId = String(chat.tgChatId)
|
||
const customerId = `tg-${channel.gatewayAccountId}-${chatId}`
|
||
const existingCustomer = await getResource(tenantId, 'customers', customerId)
|
||
const identity = await resolveTelegramIdentity(channel, chatId, chat, existingCustomer || {})
|
||
const customer = await putResource(tenantId, 'customers', withDefaults('customers', {
|
||
...(existingCustomer || {}),
|
||
...identity,
|
||
id: customerId,
|
||
telegram: chatId,
|
||
source: 'telegram',
|
||
}))
|
||
const conversationId = `${channel.id}:${chatId}`
|
||
const existingConversation = await getResource(tenantId, 'conversations', conversationId)
|
||
const recent = await gatewayRequest(
|
||
`/tg-accounts/${channel.gatewayAccountId}/chats/${chatId}/history?limit=100`,
|
||
).catch(() => ({ messages: [] }))
|
||
const historyMessages = [...(recent?.messages || [])]
|
||
let historyComplete = Boolean(existingConversation?.historyComplete) || historyMessages.length < 100
|
||
let cursor = existingConversation?.historyComplete
|
||
? undefined
|
||
: existingConversation?.historyOldestMessageId || historyMessages.at(-1)?.id
|
||
let pages = 1
|
||
while (!historyComplete && cursor && pages < 50) {
|
||
const page = await gatewayRequest(
|
||
`/tg-accounts/${channel.gatewayAccountId}/chats/${chatId}/history?limit=100&fromMessageId=${encodeURIComponent(String(cursor))}`,
|
||
).catch(() => ({ messages: [] }))
|
||
const rows = page?.messages || []
|
||
const known = new Set(historyMessages.map((message) => String(message.id)))
|
||
const newRows = rows.filter((message) => !known.has(String(message.id)))
|
||
historyMessages.push(...newRows)
|
||
pages++
|
||
if (rows.length < 100 || !newRows.length) {
|
||
historyComplete = true
|
||
break
|
||
}
|
||
cursor = rows.at(-1)?.id
|
||
}
|
||
const history = { messages: historyMessages }
|
||
let lastMessage
|
||
for (const raw of [...(history?.messages || [])].reverse()) {
|
||
const content = raw.content || {}
|
||
const typeMap = { messagePhoto: 'image', messageDocument: 'file', messageVoiceNote: 'audio', messageAudio: 'audio', messageSticker: 'file' }
|
||
lastMessage = await putResource(tenantId, `messages:${conversationId}`, {
|
||
id: String(raw.id), conversationId, sender: raw.is_outgoing ? 'agent' : 'customer',
|
||
type: typeMap[content['@type']] || 'text',
|
||
originalText: content?.text?.text || content?.caption?.text || '',
|
||
originalLang: 'auto', status: raw.is_outgoing ? 'sent' : 'delivered',
|
||
createdAt: new Date(Number(raw.date || Date.now() / 1000) * 1000).toISOString(),
|
||
telegramContent: content,
|
||
...telegramAttachment(content),
|
||
})
|
||
messagesSynced++
|
||
}
|
||
await putResource(tenantId, 'conversations', {
|
||
...(existingConversation || {}),
|
||
id: conversationId, customerId: customer.id, channel: 'telegram', channelId: channel.id,
|
||
telegramChatId: chatId, unreadCount: Number(chat.unreadCount || 0),
|
||
status: Number(chat.unreadCount || 0) ? 'pending' : 'replied', assigneeId: null,
|
||
intentLevel: 'low', sentiment: 'neutral', hasRisk: false, summary: '',
|
||
lastMessage: lastMessage || (chat.lastMessage ? {
|
||
id: String(chat.lastMessageId || `${chatId}-last`), conversationId, sender: 'customer',
|
||
type: 'text', originalText: chat.lastMessage, originalLang: 'auto', status: 'delivered',
|
||
createdAt: chat.lastMessageDate || new Date().toISOString(),
|
||
} : undefined),
|
||
historyComplete,
|
||
historyOldestMessageId: historyMessages.at(-1)?.id
|
||
? String(historyMessages.at(-1).id)
|
||
: existingConversation?.historyOldestMessageId,
|
||
historySyncedAt: new Date().toISOString(),
|
||
createdAt: existingConversation?.createdAt || new Date().toISOString(),
|
||
})
|
||
}
|
||
await patchResource(tenantId, 'channels', channel.id, { lastSyncAt: new Date().toISOString() })
|
||
broadcast(tenantId, { type: 'sync.completed', channelId: channel.id, chats: chats?.length || 0, messages: messagesSynced })
|
||
return { gateway, chats: chats?.length || 0, messages: messagesSynced }
|
||
}
|
||
|
||
app.post('/api/channels/:id/sync', async (req, res) => {
|
||
const channel = await getResource(req.auth.tenantId, 'channels', req.params.id)
|
||
if (!channel) throw fail(404, '渠道不存在')
|
||
ok(res, await syncTelegramTenant(req.auth.tenantId, channel))
|
||
})
|
||
|
||
app.post('/api/channels/test', async (req, res) => {
|
||
const start = Date.now()
|
||
const ch = await getResource(req.auth.tenantId, 'channels', req.body.id)
|
||
if (!ch) throw fail(404, '渠道不存在')
|
||
await gatewayRequest(`/tg-accounts/${ch.gatewayAccountId}`)
|
||
ok(res, { success: true, latencyMs: Date.now() - start, message: 'Telegram 网关连接正常' })
|
||
})
|
||
app.get('/api/channels/:id/logs', async (req, res) => ok(res, await listResources(req.auth.tenantId, `channel-logs:${req.params.id}`)))
|
||
app.patch('/api/channels/:id', async (req, res) => {
|
||
const value = await patchResource(req.auth.tenantId, 'channels', req.params.id, req.body)
|
||
if (!value) throw fail(404, '渠道不存在'); ok(res, value)
|
||
})
|
||
app.delete('/api/channels/:id', async (req, res) => {
|
||
if (!(await deleteResource(req.auth.tenantId, 'channels', req.params.id))) throw fail(404, '渠道不存在')
|
||
ok(res, { success: true })
|
||
})
|
||
|
||
async function messagesFor(req) {
|
||
if (!req.body.conversationId) return []
|
||
return listResources(req.auth.tenantId, `messages:${req.body.conversationId}`)
|
||
}
|
||
app.post('/api/ai/translate', async (req, res) => {
|
||
const messages = await messagesFor(req)
|
||
const result = await deepseek(
|
||
'你是灵犀跨境的专业跨境贸易翻译。只输出严格 JSON，不要 Markdown。',
|
||
`目标语言:${req.body.targetLang}\n场景:${req.body.scenario || '跨境商务'}\n上下文:\n${conversationContext(messages)}\n待翻译:\n${req.body.text}\n输出 {"translated":"...","detectedLang":"ISO语言代码","terms":[{"source":"...","target":"..."}]}`,
|
||
true,
|
||
)
|
||
ok(res, result)
|
||
})
|
||
app.post('/api/ai/analyze', async (req, res) => {
|
||
const result = await analyzeAndPersistConversation(req.auth.tenantId, req.body.conversationId)
|
||
if (!result) throw fail(400, '会话没有可分析的消息')
|
||
await putResource(req.auth.tenantId, 'ai-analysis', { ...result, conversationId: req.body.conversationId })
|
||
ok(res, result)
|
||
})
|
||
app.post('/api/ai/replies', async (req, res) => {
|
||
const messages = await messagesFor(req)
|
||
const result = await deepseek(
|
||
'你是跨境销售助理。只输出严格 JSON 数组，生成三条可人工确认的英文回复建议。content 必须是可直接发送给海外客户的完整英文；reason、expectedEffect 和 cultureTip 使用中文。',
|
||
`${conversationContext(messages)}\n输出数组元素结构 {"id":"唯一值","style":"stable|active|brief","content":"英文回复","targetLang":"en","reason":"中文理由","knowledgeRefs":[],"expectedEffect":"中文说明","cultureTip":"中文提示"}`,
|
||
true,
|
||
)
|
||
ok(res, result)
|
||
})
|
||
app.post('/api/ai/adoptions', async (req, res) => {
|
||
const value = await putResource(req.auth.tenantId, 'ai-adoptions', {
|
||
id: uid(),
|
||
conversationId: String(req.body.conversationId || ''),
|
||
suggestionId: String(req.body.suggestionId || ''),
|
||
contentHash: crypto.createHash('sha256').update(String(req.body.content || '')).digest('hex'),
|
||
adoptedBy: req.auth.userId,
|
||
createdAt: new Date().toISOString(),
|
||
})
|
||
ok(res, value)
|
||
})
|
||
app.post('/api/ai/summarize', async (req, res) => {
|
||
const messages = await messagesFor(req)
|
||
ok(res, { summary: await deepseek('用中文简洁总结跨境商务会话。', conversationContext(messages)) })
|
||
})
|
||
app.post('/api/ai/text-action', async (req, res) => {
|
||
ok(res, { result: await deepseek('你是跨境商务写作助手，只返回处理后的文本。', `操作:${req.body.action}\n语气:${req.body.tone || ''}\n目标语言:${req.body.targetLang || ''}\n文本:${req.body.text}`) })
|
||
})
|
||
const emptyKpi = () => ({ value: 0, delta: 0 })
|
||
async function businessDataset(tenantId) {
|
||
const [customers, conversations, tasks, risks, opportunities, adoptions, usersResult] = await Promise.all([
|
||
listResources(tenantId, 'customers'),
|
||
listResources(tenantId, 'conversations'),
|
||
listResources(tenantId, 'tasks'),
|
||
listResources(tenantId, 'risks'),
|
||
listResources(tenantId, 'opportunities'),
|
||
listResources(tenantId, 'ai-adoptions'),
|
||
pool.query('SELECT id,name FROM users WHERE tenant_id=$1', [tenantId]),
|
||
])
|
||
const messageGroups = await Promise.all(conversations.map(async (conversation) => ({
|
||
conversation,
|
||
messages: (await listResources(tenantId, `messages:${conversation.id}`))
|
||
.sort((a, b) => Date.parse(a.createdAt || '') - Date.parse(b.createdAt || '')),
|
||
})))
|
||
const responseTimes = []
|
||
const repliesByOwner = new Map()
|
||
for (const group of messageGroups) {
|
||
for (let index = 0; index < group.messages.length; index++) {
|
||
const message = group.messages[index]
|
||
if (message.sender !== 'customer') continue
|
||
const reply = group.messages.slice(index + 1).find((item) => item.sender === 'agent')
|
||
if (!reply) continue
|
||
const minutes = Math.max(0, (Date.parse(reply.createdAt) - Date.parse(message.createdAt)) / 60000)
|
||
if (Number.isFinite(minutes) && minutes < 7 * 24 * 60) {
|
||
responseTimes.push(minutes)
|
||
const owner = group.conversation.assigneeId
|
||
if (owner) {
|
||
const row = repliesByOwner.get(owner) || { total: 0, count: 0 }
|
||
row.total += minutes
|
||
row.count++
|
||
repliesByOwner.set(owner, row)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
return {
|
||
customers, conversations, tasks, risks, opportunities, adoptions,
|
||
users: usersResult.rows,
|
||
messageGroups,
|
||
messages: messageGroups.flatMap((group) => group.messages),
|
||
responseTimes,
|
||
repliesByOwner,
|
||
}
|
||
}
|
||
|
||
app.get('/api/dashboard/overview', async (req, res) => {
|
||
const data = await businessDataset(req.auth.tenantId)
|
||
const { customers, conversations, tasks, risks, opportunities, messages, responseTimes, users, repliesByOwner, adoptions } = data
|
||
const unread = conversations.reduce((n, x) => n + Number(x.unreadCount || 0), 0)
|
||
const today = dateKey()
|
||
const won = opportunities.filter((item) => item.stage === 'won').length
|
||
const outgoing = messages.filter((message) => message.sender === 'agent').length
|
||
const days = Array.from({ length: 7 }, (_, index) => dateKey(Date.now() - (6 - index) * 86400000))
|
||
const sources = countBy(customers, (customer) => customer.sourcePath || customer.source || 'telegram')
|
||
const stages = countBy(customers, 'stage')
|
||
const teamNames = new Map(users.map((user) => [user.id, user.name]))
|
||
ok(res, {
|
||
stats: {
|
||
newInquiriesToday: { value: conversations.filter((x) => String(x.createdAt || '').slice(0, 10) === today).length, delta: 0 },
|
||
pendingReply: { value: unread, delta: 0 }, highIntentCustomers: { value: customers.filter((x) => x.intentScore >= 75).length, delta: 0 },
|
||
todayFollowUps: { value: tasks.filter((x) => x.status !== 'done').length, delta: 0 }, openRisks: { value: risks.filter((x) => x.status !== 'resolved').length, delta: 0 },
|
||
avgFirstResponseMin: { value: responseTimes.length ? Math.round(responseTimes.reduce((a, b) => a + b, 0) / responseTimes.length) : 0, delta: 0, unit: '分钟' },
|
||
monthConversionRate: { value: percent(won, opportunities.length), delta: 0, isPercent: true },
|
||
aiAdoptionRate: { value: percent(adoptions.length, Math.max(outgoing, 1)), delta: 0, isPercent: true },
|
||
},
|
||
todos: tasks.filter((x) => x.status !== 'done').slice(0, 7),
|
||
highValueCustomers: customers.sort((a, b) => (b.intentScore || 0) - (a.intentScore || 0)).slice(0, 6),
|
||
channelOverview: [{ channel: 'telegram', todayMessages: messages.filter((message) => dateKey(message.createdAt) === today).length, pending: conversations.filter((x) => x.status === 'pending').length, unread }],
|
||
inquiryTrend: days.map((day) => ({
|
||
date: day.slice(5),
|
||
inquiries: conversations.filter((item) => dateKey(item.createdAt) === day).length,
|
||
customers: customers.filter((item) => dateKey(item.createdAt || item.updatedAt) === day).length,
|
||
})),
|
||
sourceDistribution: [...sources].map(([name, value]) => ({ name, value })),
|
||
funnel: [...stages].map(([stage, count]) => ({ stage, label: stage, count })),
|
||
teamReplyRank: [...repliesByOwner].map(([memberId, row]) => ({
|
||
memberId,
|
||
name: teamNames.get(memberId) || '未命名成员',
|
||
avgResponseMin: Math.round(row.total / row.count),
|
||
replies: row.count,
|
||
})).sort((a, b) => a.avgResponseMin - b.avgResponseMin),
|
||
recentRisks: risks.slice(0, 5), updatedAt: new Date().toISOString(),
|
||
})
|
||
})
|
||
|
||
app.get('/api/team', async (req, res) => {
|
||
const [{ rows }, customers, tasks] = await Promise.all([
|
||
pool.query('SELECT * FROM users WHERE tenant_id=$1 ORDER BY created_at', [req.auth.tenantId]),
|
||
listResources(req.auth.tenantId, 'customers'),
|
||
listResources(req.auth.tenantId, 'tasks'),
|
||
])
|
||
ok(res, rows.map((x) => ({
|
||
id: x.id,
|
||
name: x.name,
|
||
email: x.account,
|
||
phone: x.phone,
|
||
department: x.department,
|
||
role: x.role,
|
||
online: Date.parse(x.last_login_at || '') > Date.now() - 15 * 60 * 1000,
|
||
customerCount: customers.filter((customer) => customer.ownerId === x.id).length,
|
||
todoCount: tasks.filter((task) => task.ownerId === x.id && task.status !== 'done').length,
|
||
lastLoginAt: x.last_login_at,
|
||
status: x.status,
|
||
joinedAt: x.created_at,
|
||
})))
|
||
})
|
||
app.get('/api/team/operation-logs', async (req, res) => {
|
||
const { rows } = await pool.query('SELECT * FROM audit_logs WHERE tenant_id=$1 ORDER BY created_at DESC LIMIT 200', [req.auth.tenantId])
|
||
ok(res, rows.map((x) => ({ id: x.id, actorId: x.user_id, actorName: '', action: x.action, target: x.target, detail: JSON.stringify(x.detail), ip: x.ip, createdAt: x.created_at })))
|
||
})
|
||
app.get('/api/team/login-records', async (req, res) => {
|
||
const { rows } = await pool.query(
|
||
`SELECT lr.*,u.name AS member_name
|
||
FROM login_records lr LEFT JOIN users u ON u.id=lr.user_id
|
||
WHERE lr.tenant_id=$1 ORDER BY lr.created_at DESC LIMIT 200`,
|
||
[req.auth.tenantId],
|
||
)
|
||
ok(res, rows.map((row) => ({
|
||
id: row.id,
|
||
memberId: row.user_id,
|
||
memberName: row.member_name || row.account,
|
||
device: row.device || 'Unknown',
|
||
ip: row.ip,
|
||
location: row.ip || 'Unknown',
|
||
status: row.status,
|
||
createdAt: row.created_at,
|
||
})))
|
||
})
|
||
app.post('/api/team/invite', async (req, res) => {
|
||
if (req.auth.role !== 'admin') throw fail(403, '仅管理员可邀请成员')
|
||
const email = String(req.body.email || '').trim().toLowerCase()
|
||
if (!email.includes('@')) throw fail(400, '请输入有效邮箱')
|
||
if (!process.env.SMTP_HOST || !process.env.SMTP_USER || !process.env.SMTP_PASSWORD) throw fail(503, 'SMTP 邮件服务尚未配置')
|
||
const temporaryPassword = `Lx!${crypto.randomBytes(12).toString('base64url')}`
|
||
const userId = uid()
|
||
const exists = await pool.query('SELECT 1 FROM users WHERE account=$1', [email])
|
||
if (exists.rowCount) throw fail(409, '该邮箱已加入团队')
|
||
await pool.query(
|
||
`INSERT INTO users(id,tenant_id,name,account,password_hash,phone,role,department,title,status)
|
||
VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,'pending')`,
|
||
[userId, req.auth.tenantId, req.body.name || email.split('@')[0], email, await bcrypt.hash(temporaryPassword, 12),
|
||
req.body.phone || '', req.body.role || 'sales', req.body.department || '', req.body.title || '成员'],
|
||
)
|
||
const transport = nodemailer.createTransport({
|
||
host: process.env.SMTP_HOST, port: Number(process.env.SMTP_PORT || 587),
|
||
secure: String(process.env.SMTP_SECURE || 'false') === 'true',
|
||
auth: { user: process.env.SMTP_USER, pass: process.env.SMTP_PASSWORD },
|
||
})
|
||
try {
|
||
await transport.verify()
|
||
await transport.sendMail({
|
||
from: process.env.SMTP_FROM || process.env.SMTP_USER, to: email,
|
||
subject: '邀请加入灵犀跨境',
|
||
text: `你已被邀请加入灵犀跨境。登录地址：${process.env.APP_PUBLIC_URL || 'http://127.0.0.1:8088'}，临时密码：${temporaryPassword}。首次登录后请立即修改密码。`,
|
||
})
|
||
await pool.query("UPDATE users SET status='active' WHERE id=$1", [userId])
|
||
} catch (error) {
|
||
await pool.query('DELETE FROM users WHERE id=$1', [userId])
|
||
throw fail(502, `邀请邮件发送失败：${error.message}`)
|
||
}
|
||
const { rows } = await pool.query('SELECT * FROM users WHERE id=$1', [userId])
|
||
await audit(req, 'team.invite', userId, { email })
|
||
ok(res, rows[0])
|
||
})
|
||
app.patch('/api/team/:id', async (req, res) => {
|
||
if (req.auth.role !== 'admin') throw fail(403, '仅管理员可修改成员')
|
||
const allowed = ['name', 'phone', 'department', 'role', 'status']; const values = []; const sets = []
|
||
for (const key of allowed) if (key in req.body) { values.push(req.body[key]); sets.push(`${key}=$${values.length}`) }
|
||
values.push(req.params.id, req.auth.tenantId)
|
||
const { rows } = await pool.query(`UPDATE users SET ${sets.join(',')} WHERE id=$${values.length - 1} AND tenant_id=$${values.length} RETURNING *`, values)
|
||
if (!rows[0]) throw fail(404, '成员不存在'); ok(res, rows[0])
|
||
})
|
||
app.delete('/api/team/:id', (_req, _res, next) => next(fail(501, '成员删除改为禁用，避免审计数据丢失')))
|
||
|
||
const defaultSettings = {
|
||
company: { name: '灵犀跨境', industry: '', size: '', contact: '', email: '', intro: '', products: [], regions: [] },
|
||
ai: { replyLanguage: 'zh-CN', replyTone: 'professional', autoGenerate: true, requireConfirm: true, sentimentAnalysis: true, riskAlert: true, cultureCheck: true, allowHistoryRef: true },
|
||
notification: { newInquiry: true, highIntent: true, riskAlert: true, taskDue: true, taskOverdue: true, channelOffline: true, planExpire: true, byEmail: false, bySms: false },
|
||
translate: { industry: '', formality: 'professional', targetCountry: '', localization: true, keepBrand: true, keepModel: true, autoUnit: true, cultureReminder: true, religionReminder: true },
|
||
language: 'zh-CN', timezone: 'Asia/Hong_Kong', sessionRetentionMonths: 12, brandColor: '#2563eb', onboarding: {},
|
||
}
|
||
app.get('/api/settings', async (req, res) => ok(res, (await getResource(req.auth.tenantId, 'singleton', 'settings')) || defaultSettings))
|
||
app.patch('/api/settings', async (req, res) => {
|
||
const current = (await getResource(req.auth.tenantId, 'singleton', 'settings')) || defaultSettings
|
||
const value = await putResource(req.auth.tenantId, 'singleton', {
|
||
...current,
|
||
...req.body,
|
||
company: { ...current.company, ...(req.body.company || {}) },
|
||
ai: { ...current.ai, ...(req.body.ai || {}) },
|
||
notification: { ...current.notification, ...(req.body.notification || {}) },
|
||
locale: { ...current.locale, ...(req.body.locale || {}) },
|
||
id: 'settings',
|
||
})
|
||
await audit(req, 'settings.update', 'settings', Object.keys(req.body || {}))
|
||
ok(res, value)
|
||
})
|
||
app.get('/api/notifications', async (req, res) => ok(res, await listResources(req.auth.tenantId, 'notifications')))
|
||
app.post('/api/notifications/read-all', async (req, res) => {
|
||
for (const n of await listResources(req.auth.tenantId, 'notifications')) await putResource(req.auth.tenantId, 'notifications', { ...n, read: true })
|
||
ok(res, { success: true })
|
||
})
|
||
app.get('/api/knowledge/packs', (_req, res) => ok(res, []))
|
||
app.get('/api/knowledge/stats', async (req, res) => {
|
||
const list = await listResources(req.auth.tenantId, 'knowledge')
|
||
ok(res, { totalCitations: list.reduce((n, x) => n + Number(x.citedCount || 0), 0), aiReplyCitationRate: 0, topUsed: [], lowQuality: [], outdated: [] })
|
||
})
|
||
app.post('/api/knowledge/upload', async (req, res) => {
|
||
const files = Array.isArray(req.body.files) ? req.body.files : []
|
||
const items = []
|
||
for (const file of files) {
|
||
const buffer = Buffer.from(String(file.base64 || ''), 'base64')
|
||
if (!buffer.length || buffer.length > 20 * 1024 * 1024) throw fail(400, `文件 ${file.name || ''} 无内容或超过 20MB`)
|
||
const extension = String(file.name || '').split('.').pop()?.toLowerCase()
|
||
let content = ''
|
||
if (['txt', 'md', 'csv', 'json', 'html'].includes(extension)) content = buffer.toString('utf8')
|
||
else if (extension === 'docx') content = (await mammoth.extractRawText({ buffer })).value
|
||
else if (extension === 'pdf') content = (await pdfParse(buffer)).text
|
||
else throw fail(415, `暂不支持 ${extension || '未知'} 文件`)
|
||
items.push(await putResource(req.auth.tenantId, 'knowledge', {
|
||
title: String(file.name).replace(/\.[^.]+$/, ''), category: 'product',
|
||
content: content.trim(), tags: ['文件导入'], enabled: true, version: 1,
|
||
updatedBy: req.auth.userId, source: 'file', fileName: file.name, fileType: extension,
|
||
status: 'ready', citedCount: 0, languages: ['auto'], quality: 'normal',
|
||
createdAt: new Date().toISOString(),
|
||
}))
|
||
}
|
||
await audit(req, 'knowledge.upload', '', { count: items.length })
|
||
ok(res, { items })
|
||
})
|
||
app.post('/api/knowledge/search-test', async (req, res) => {
|
||
const list = await listResources(req.auth.tenantId, 'knowledge')
|
||
const words = String(req.body.question || '').toLowerCase().split(/\s+/)
|
||
const hits = list.filter((x) => words.some((w) => JSON.stringify(x).toLowerCase().includes(w))).slice(0, 5).map((item) => ({ item, score: 1, snippet: String(item.content || '').slice(0, 120) }))
|
||
ok(res, { answer: hits.length ? await deepseek('仅根据给出的企业知识回答，不得编造。', `问题:${req.body.question}\n知识:${JSON.stringify(hits)}`) : '知识库中没有匹配内容。', hits })
|
||
})
|
||
app.get('/api/billing', async (req, res) => {
|
||
const [{ rows }, orders, invoices] = await Promise.all([
|
||
pool.query('SELECT plan FROM tenants WHERE id=$1', [req.auth.tenantId]),
|
||
listResources(req.auth.tenantId, 'billing-orders'),
|
||
listResources(req.auth.tenantId, 'billing-invoices'),
|
||
])
|
||
const planKey = rows[0]?.plan || 'lite'
|
||
ok(res, {
|
||
info: { planKey, planName: planKey, expireAt: '', autoRenew: false, usage: {
|
||
members: { used: 0, total: 100, unit: '人' }, messages: { used: 0, total: 100000, unit: '条' },
|
||
translations: { used: 0, total: 100000, unit: '次' }, knowledgeStorage: { used: 0, total: 10240, unit: 'MB' },
|
||
} },
|
||
orders, invoices,
|
||
})
|
||
})
|
||
app.post('/api/billing/upgrade', async (req, res) => {
|
||
const priceMap = {
|
||
lite: process.env.STRIPE_PRICE_LITE,
|
||
standard: process.env.STRIPE_PRICE_STANDARD,
|
||
pro: process.env.STRIPE_PRICE_PRO,
|
||
growth: process.env.STRIPE_PRICE_GROWTH,
|
||
}
|
||
const price = priceMap[req.body.planKey]
|
||
if (!process.env.STRIPE_SECRET_KEY || !price) throw fail(503, 'Stripe 商户密钥或套餐 Price ID 尚未配置')
|
||
const form = new URLSearchParams({
|
||
mode: 'subscription', 'line_items[0][price]': price, 'line_items[0][quantity]': '1',
|
||
success_url: `${process.env.APP_PUBLIC_URL || 'http://127.0.0.1:8088'}/billing?payment=success&session_id={CHECKOUT_SESSION_ID}`,
|
||
cancel_url: `${process.env.APP_PUBLIC_URL || 'http://127.0.0.1:8088'}/billing?payment=cancelled`,
|
||
'metadata[tenantId]': req.auth.tenantId, 'metadata[planKey]': req.body.planKey,
|
||
})
|
||
const response = await fetch('https://api.stripe.com/v1/checkout/sessions', {
|
||
method: 'POST',
|
||
headers: { authorization: `Bearer ${process.env.STRIPE_SECRET_KEY}`, 'content-type': 'application/x-www-form-urlencoded' },
|
||
body: form,
|
||
})
|
||
const checkout = await response.json()
|
||
if (!response.ok) throw fail(502, checkout?.error?.message || '创建支付会话失败')
|
||
await putResource(req.auth.tenantId, 'billing-orders', {
|
||
id: checkout.id, planKey: req.body.planKey, status: 'pending',
|
||
checkoutUrl: checkout.url, createdAt: new Date().toISOString(),
|
||
})
|
||
ok(res, { checkoutUrl: checkout.url, sessionId: checkout.id })
|
||
})
|
||
|
||
app.post('/api/billing/stripe-webhook', async (req, res) => {
|
||
const signingSecret = process.env.STRIPE_WEBHOOK_SECRET
|
||
const signature = req.get('stripe-signature') || ''
|
||
if (!signingSecret || !signature) throw fail(401, 'Stripe Webhook 签名配置缺失')
|
||
const parts = Object.fromEntries(signature.split(',').map((part) => part.split('=', 2)))
|
||
const timestamp = Number(parts.t)
|
||
if (!timestamp || Math.abs(Date.now() / 1000 - timestamp) > 300) throw fail(401, 'Stripe Webhook 已过期')
|
||
const expected = crypto.createHmac('sha256', signingSecret)
|
||
.update(`${timestamp}.${req.rawBody.toString('utf8')}`).digest('hex')
|
||
const provided = String(parts.v1 || '')
|
||
if (provided.length !== expected.length || !crypto.timingSafeEqual(Buffer.from(provided), Buffer.from(expected))) {
|
||
throw fail(401, 'Stripe Webhook 签名无效')
|
||
}
|
||
const event = req.body
|
||
if (event.type === 'checkout.session.completed') {
|
||
const session = event.data?.object || {}
|
||
const tenantId = session.metadata?.tenantId
|
||
const planKey = session.metadata?.planKey
|
||
if (tenantId && planKey) {
|
||
await pool.query('UPDATE tenants SET plan=$1 WHERE id=$2', [planKey, tenantId])
|
||
await putResource(tenantId, 'billing-orders', {
|
||
id: session.id, planKey, status: 'paid', amountTotal: session.amount_total,
|
||
currency: session.currency, customerId: session.customer, subscriptionId: session.subscription,
|
||
paidAt: new Date().toISOString(),
|
||
})
|
||
await putResource(tenantId, 'billing-invoices', {
|
||
id: session.payment_intent || session.id, orderId: session.id, planKey,
|
||
amount: session.amount_total, currency: session.currency, status: 'paid',
|
||
createdAt: new Date().toISOString(),
|
||
})
|
||
broadcast(tenantId, { type: 'billing.updated', planKey, status: 'paid' })
|
||
}
|
||
}
|
||
ok(res, { received: true })
|
||
})
|
||
app.get('/api/analytics', async (req, res) => {
|
||
const data = await businessDataset(req.auth.tenantId)
|
||
const { customers, conversations, tasks, risks, opportunities, messages, responseTimes, users, repliesByOwner, adoptions } = data
|
||
const days = Array.from({ length: 30 }, (_, index) => dateKey(Date.now() - (29 - index) * 86400000))
|
||
const won = opportunities.filter((item) => item.stage === 'won')
|
||
const doneTasks = tasks.filter((item) => item.status === 'done')
|
||
const handledRisks = risks.filter((item) => ['resolved', 'ignored'].includes(item.status))
|
||
const translated = messages.filter((message) => message.translatedText).length
|
||
const productCounts = new Map()
|
||
for (const customer of customers) for (const product of customer.products || []) {
|
||
productCounts.set(product, (productCounts.get(product) || 0) + 1)
|
||
}
|
||
const countryCounts = countBy(customers, 'country')
|
||
const stageCounts = countBy(customers, 'stage')
|
||
const sourceCounts = countBy(customers, (customer) => customer.sourcePath || customer.source || 'telegram')
|
||
const userNames = new Map(users.map((user) => [user.id, user.name]))
|
||
const ownerCustomers = countBy(customers, 'ownerId')
|
||
const ownerWon = countBy(won, 'ownerId')
|
||
const overdue = tasks.filter((task) => task.status !== 'done' && Date.parse(task.dueAt || '') < Date.now())
|
||
const now = Date.now()
|
||
ok(res, {
|
||
kpis: {
|
||
newCustomers: { value: customers.filter((item) => Date.parse(item.createdAt || item.updatedAt || '') >= now - 30 * 86400000).length, delta: 0 },
|
||
validInquiries: { value: conversations.length, delta: 0 },
|
||
highIntent: { value: customers.filter((item) => Number(item.intentScore || 0) >= 75).length, delta: 0 },
|
||
conversionRate: { value: percent(won.length, opportunities.length), delta: 0, isPercent: true },
|
||
avgResponseMin: { value: responseTimes.length ? Math.round(responseTimes.reduce((a, b) => a + b, 0) / responseTimes.length) : 0, delta: 0, unit: '分钟' },
|
||
aiTranslateUsage: { value: translated, delta: 0 },
|
||
aiAdoptionRate: { value: percent(adoptions.length, messages.filter((item) => item.sender === 'agent').length), delta: 0, isPercent: true },
|
||
taskCompletionRate: { value: percent(doneTasks.length, tasks.length), delta: 0, isPercent: true },
|
||
riskHandleRate: { value: percent(handledRisks.length, risks.length), delta: 0, isPercent: true },
|
||
repurchaseRate: { value: percent(customers.filter((item) => item.stage === 'repurchase').length, customers.length), delta: 0, isPercent: true },
|
||
},
|
||
customerGrowth: days.map((day) => ({ date: day.slice(5), count: customers.filter((item) => dateKey(item.createdAt || item.updatedAt) === day).length })),
|
||
channelInquiries: [{ channel: 'telegram', name: 'Telegram', count: conversations.filter((item) => item.channel === 'telegram').length }],
|
||
countryDistribution: [...countryCounts].map(([country, count]) => ({ country, name: country, count })),
|
||
sourcePaths: [...sourceCounts].map(([path, count]) => ({ path, count })),
|
||
funnel: [...stageCounts].map(([stage, count]) => ({ stage, label: stage, count })),
|
||
intentDistribution: [
|
||
{ range: '0-24', count: customers.filter((item) => Number(item.intentScore || 0) < 25).length },
|
||
{ range: '25-49', count: customers.filter((item) => Number(item.intentScore || 0) >= 25 && Number(item.intentScore || 0) < 50).length },
|
||
{ range: '50-74', count: customers.filter((item) => Number(item.intentScore || 0) >= 50 && Number(item.intentScore || 0) < 75).length },
|
||
{ range: '75-100', count: customers.filter((item) => Number(item.intentScore || 0) >= 75).length },
|
||
],
|
||
sentimentTrend: days.map((day) => {
|
||
const rows = customers.filter((item) => dateKey(item.aiAnalyzedAt || item.updatedAt || item.createdAt) === day)
|
||
return {
|
||
date: day.slice(5),
|
||
positive: rows.filter((item) => ['positive', 'neutral-positive'].includes(item.sentiment)).length,
|
||
neutral: rows.filter((item) => item.sentiment === 'neutral').length,
|
||
negative: rows.filter((item) => ['negative', 'angry'].includes(item.sentiment)).length,
|
||
}
|
||
}),
|
||
riskTrend: days.map((day) => ({ date: day.slice(5), count: risks.filter((item) => dateKey(item.createdAt) === day).length })),
|
||
productRank: [...productCounts].map(([product, inquiries]) => ({ product, inquiries })).sort((a, b) => b.inquiries - a.inquiries).slice(0, 10),
|
||
teamReplyRank: [...repliesByOwner].map(([id, row]) => ({ name: userNames.get(id) || '未命名成员', avgResponseMin: Math.round(row.total / row.count) })).sort((a, b) => a.avgResponseMin - b.avgResponseMin),
|
||
teamConversionRank: users.map((user) => ({ name: user.name, rate: percent(ownerWon.get(user.id) || 0, ownerCustomers.get(user.id) || 0) })).sort((a, b) => b.rate - a.rate),
|
||
taskCompletion: users.map((user) => {
|
||
const rows = tasks.filter((item) => item.ownerId === user.id)
|
||
return { name: user.name, done: rows.filter((item) => item.status === 'done').length, total: rows.length }
|
||
}),
|
||
churnReasons: [...countBy(risks, 'type')].map(([reason, count]) => ({ reason, count })),
|
||
industryDistribution: [...countBy(customers, (item) => item.aiProfile?.industry || '未知')].map(([industry, count]) => ({ industry, count })),
|
||
bossBoard: {
|
||
urgentHighValue: customers.filter((item) => Number(item.intentScore || 0) >= 75 && ['high', 'urgent'].includes(item.riskLevel)).slice(0, 8).map((item) => ({ id: item.id, name: item.name, country: item.country, reason: item.aiProfile?.summary || '高意向且存在风险', intentScore: item.intentScore })),
|
||
overdueMembers: users.map((user) => ({ id: user.id, name: user.name, overdueCount: overdue.filter((task) => task.ownerId === user.id).length })).filter((item) => item.overdueCount > 0),
|
||
bestChannel: { channel: 'telegram', name: 'Telegram', conversion: percent(won.length, opportunities.length) },
|
||
fastestCountries: [...countryCounts].map(([country, count]) => ({ country, name: country, growth: percent(count, Math.max(customers.length, 1)) / 100 })).sort((a, b) => b.growth - a.growth).slice(0, 6),
|
||
topProducts: [...productCounts].map(([product, inquiries]) => ({ product, inquiries })).sort((a, b) => b.inquiries - a.inquiries).slice(0, 6),
|
||
churningHighIntent: customers.filter((item) => Number(item.intentScore || 0) >= 75 && now - Date.parse(item.lastContactAt || item.updatedAt || '') > 7 * 86400000).slice(0, 8).map((item) => ({ id: item.id, name: item.name, country: item.country, daysSilent: Math.floor((now - Date.parse(item.lastContactAt || item.updatedAt)) / 86400000), intentScore: item.intentScore })),
|
||
},
|
||
})
|
||
})
|
||
app.get('/api/search', async (req, res) => {
|
||
const q = String(req.query.q || '').toLowerCase()
|
||
const result = {}
|
||
for (const kind of ['customers', 'conversations', 'knowledge', 'tasks']) result[kind] = (await listResources(req.auth.tenantId, kind)).filter((x) => JSON.stringify(x).toLowerCase().includes(q)).slice(0, 5).map((x) => ({ id: x.id, title: x.name || x.title || x.summary || '', sub: x.company || x.content || '' }))
|
||
ok(res, result)
|
||
})
|
||
|
||
app.post('/internal/webhooks/telegram/:tenantId/:channelId', async (req, res) => {
|
||
const expected = process.env.TELEGRAM_WEBHOOK_SECRET
|
||
const received = String(req.get('x-webhook-signature') || '').replace(/^sha256=/, '')
|
||
const calculated = crypto.createHmac('sha256', expected || '').update(req.rawBody || Buffer.from(JSON.stringify(req.body))).digest('hex')
|
||
if (!expected || received.length !== calculated.length || !crypto.timingSafeEqual(Buffer.from(received), Buffer.from(calculated))) throw fail(401, 'Webhook 签名无效')
|
||
const { tenantId, channelId } = req.params
|
||
const channel = await getResource(tenantId, 'channels', channelId)
|
||
if (!channel) throw fail(404, 'Webhook 渠道映射不存在')
|
||
const eventId = String(req.body.id || req.body.updateId || crypto.createHash('sha256').update(req.rawBody || Buffer.from(JSON.stringify(req.body))).digest('hex'))
|
||
if (await getResource(tenantId, 'webhook-events', eventId)) return ok(res, { accepted: true, duplicate: true })
|
||
await putResource(tenantId, 'webhook-events', { id: eventId, payload: req.body, receivedAt: new Date().toISOString() })
|
||
const telegramMessage = req.body?.data?.message
|
||
if (['message.received', 'message.pending', 'message.sent', 'message.failed'].includes(req.body.event) && telegramMessage) {
|
||
const chatId = String(telegramMessage.chat_id)
|
||
const customerId = `tg-${channel.gatewayAccountId}-${chatId}`
|
||
let customer = await getResource(tenantId, 'customers', customerId)
|
||
const identity = await resolveTelegramIdentity(channel, chatId, {}, customer || {})
|
||
customer = await putResource(tenantId, 'customers', withDefaults('customers', {
|
||
...(customer || {}),
|
||
...identity,
|
||
id: customerId,
|
||
telegram: chatId,
|
||
source: 'telegram',
|
||
}))
|
||
const conversationId = `${channelId}:${chatId}`
|
||
const outgoing = Boolean(telegramMessage.is_outgoing)
|
||
const content = telegramMessage.content || {}
|
||
const originalText = content?.text?.text || content?.caption?.text || ''
|
||
const typeMap = {
|
||
messagePhoto: 'image',
|
||
messageDocument: 'file',
|
||
messageVoiceNote: 'audio',
|
||
messageAudio: 'audio',
|
||
messageSticker: 'file',
|
||
}
|
||
const messageStatus = !outgoing
|
||
? 'delivered'
|
||
: req.body.event === 'message.failed'
|
||
? 'failed'
|
||
: req.body.event === 'message.sent'
|
||
? 'sent'
|
||
: 'sending'
|
||
const oldMessageId = String(req.body?.data?.old_message_id || '')
|
||
if (oldMessageId && oldMessageId !== String(telegramMessage.id)) {
|
||
await deleteResource(tenantId, `messages:${conversationId}`, oldMessageId)
|
||
}
|
||
const outgoingMeta = outgoing
|
||
? await getResource(tenantId, 'outgoing-message-meta', oldMessageId || String(telegramMessage.id))
|
||
: null
|
||
const message = await putResource(tenantId, `messages:${conversationId}`, {
|
||
id: String(telegramMessage.id), conversationId, sender: outgoing ? 'agent' : 'customer',
|
||
senderName: outgoing ? '我' : customer.name,
|
||
type: typeMap[content['@type']] || 'text',
|
||
originalText: outgoingMeta?.originalText || originalText,
|
||
translatedText: outgoingMeta?.translatedText,
|
||
originalLang: outgoingMeta?.originalLang || 'auto',
|
||
status: messageStatus,
|
||
translationStatus: !outgoing && content['@type'] === 'messageText' && originalText ? 'pending' : undefined,
|
||
createdAt: new Date(Number(telegramMessage.date || Date.now() / 1000) * 1000).toISOString(),
|
||
...telegramAttachment(content),
|
||
})
|
||
const previous = await getResource(tenantId, 'conversations', conversationId)
|
||
await putResource(tenantId, 'conversations', {
|
||
id: conversationId, customerId: customer.id, channel: 'telegram', channelId,
|
||
telegramChatId: chatId, unreadCount: outgoing ? Number(previous?.unreadCount || 0) : Number(previous?.unreadCount || 0) + 1,
|
||
status: outgoing ? 'replied' : 'pending', assigneeId: previous?.assigneeId || null,
|
||
intentLevel: previous?.intentLevel || 'low', sentiment: previous?.sentiment || 'neutral',
|
||
hasRisk: previous?.hasRisk || false, summary: previous?.summary || '',
|
||
lastMessage: message, createdAt: previous?.createdAt || new Date().toISOString(),
|
||
})
|
||
broadcast(tenantId, {
|
||
type: 'message.created',
|
||
conversationId,
|
||
message,
|
||
customer,
|
||
oldMessageId: oldMessageId || undefined,
|
||
})
|
||
if (!outgoing && req.body.event === 'message.received') {
|
||
await putResource(tenantId, 'notifications', {
|
||
id: `telegram-message-${channelId}-${telegramMessage.id}`,
|
||
type: 'inquiry',
|
||
title: `${customer.name} 发来新消息`,
|
||
description: originalText || (typeMap[content['@type']] === 'image'
|
||
? '[图片]'
|
||
: typeMap[content['@type']] === 'audio'
|
||
? '[语音]'
|
||
: typeMap[content['@type']] === 'file'
|
||
? '[文件]'
|
||
: '[新消息]'),
|
||
link: `/inbox?c=${encodeURIComponent(conversationId)}`,
|
||
read: false,
|
||
createdAt: new Date().toISOString(),
|
||
})
|
||
}
|
||
if (outgoingMeta && req.body.event === 'message.sent') {
|
||
await deleteResource(tenantId, 'outgoing-message-meta', outgoingMeta.id)
|
||
}
|
||
enqueueAutomaticTranslation(tenantId, conversationId, message)
|
||
if (!outgoing && req.body.event === 'message.received') {
|
||
enqueueAutomaticProfile(tenantId, conversationId)
|
||
}
|
||
}
|
||
ok(res, { accepted: true })
|
||
})
|
||
|
||
app.get('/health/live', (_req, res) => res.json({ status: 'ok', service: 'lingxi-crossborder-api' }))
|
||
app.get('/health/ready', async (_req, res) => {
|
||
await pool.query('SELECT 1'); res.json({ status: 'ready', database: 'ok' })
|
||
})
|
||
|
||
app.use((error, req, res, _next) => {
|
||
const status = Number(error.status || 500)
|
||
if (status >= 500) console.error(JSON.stringify({ requestId: res.getHeader('x-request-id'), message: error.message, stack: error.stack }))
|
||
res.status(status).json({ code: status, message: error.message || '服务器内部错误', data: null, requestId: res.getHeader('x-request-id') })
|
||
})
|
||
|
||
async function repairCorruptedTranslations() {
|
||
const { rows } = await pool.query(
|
||
`SELECT tenant_id, kind, data
|
||
FROM resources
|
||
WHERE kind LIKE 'messages:%'
|
||
AND data ? 'translatedText'`,
|
||
)
|
||
let repaired = 0
|
||
for (const row of rows) {
|
||
const message = row.data
|
||
if (!isSuspiciousTranslation(message.originalText, message.translatedText)) continue
|
||
const conversationId = String(message.conversationId || row.kind.slice('messages:'.length))
|
||
const {
|
||
translatedText: _translatedText,
|
||
translatedAt: _translatedAt,
|
||
translationRetryAt: _translationRetryAt,
|
||
...clean
|
||
} = message
|
||
const numericOnly = isNonLinguisticText(clean.originalText)
|
||
const fixed = await putResource(String(row.tenant_id), row.kind, {
|
||
...clean,
|
||
translatedText: numericOnly ? String(clean.originalText || '').trim() : undefined,
|
||
originalLang: numericOnly ? 'und' : clean.originalLang,
|
||
translationStatus: numericOnly ? 'completed' : 'pending',
|
||
translatedAt: numericOnly ? new Date().toISOString() : undefined,
|
||
})
|
||
const conversation = await getResource(String(row.tenant_id), 'conversations', conversationId)
|
||
if (String(conversation?.lastMessage?.id || '') === String(fixed.id)) {
|
||
await patchResource(String(row.tenant_id), 'conversations', conversationId, { lastMessage: fixed })
|
||
}
|
||
if (!numericOnly) enqueueAutomaticTranslation(String(row.tenant_id), conversationId, fixed)
|
||
repaired++
|
||
}
|
||
if (repaired) console.log(`已修复 ${repaired} 条受污染的历史译文`)
|
||
}
|
||
|
||
await migrate()
|
||
await repairCorruptedTranslations()
|
||
const existingTelegramChannels = await pool.query(
|
||
`SELECT tenant_id, data FROM resources
|
||
WHERE kind='channels' AND data->>'type'='telegram' AND data ? 'gatewayAccountId'`,
|
||
)
|
||
for (const row of existingTelegramChannels.rows) {
|
||
await ensureChannelWebhook(String(row.tenant_id), row.data).catch(async (error) => {
|
||
await patchResource(String(row.tenant_id), 'channels', row.data.id, {
|
||
errorMessage: `Webhook 同步失败：${error.message}`,
|
||
})
|
||
})
|
||
}
|
||
httpServer.listen(Number(process.env.PORT || 3100), '0.0.0.0', () => console.log(`灵犀跨境 API listening on ${process.env.PORT || 3100}`))
|