mirror of
https://github.com/EvolutionAPI/evolution-api.git
synced 2025-12-20 20:32:23 -06:00
refactor(chatbot): unify keywordFinish type and enhance session handling
- Changed the type of `keywordFinish` from an array to a string in multiple DTOs and controller interfaces to simplify data handling. - Updated the `BaseChatbotService` to include logic for updating session status to 'opened' and managing user responses more effectively. - Refactored the media message handling in the `BaseChatbotService` to streamline the process and improve readability. - Enhanced error logging across various services to ensure better traceability during operations. This commit focuses on improving the structure and consistency of chatbot integrations while ensuring that session management is robust and user-friendly.
This commit is contained in:
Davidson Gomes
@@ -1,425 +1,111 @@
|
||||
/* eslint-disable @typescript-eslint/no-unused-vars */
|
||||
import { InstanceDto } from '@api/dto/instance.dto';
|
||||
import { PrismaRepository } from '@api/repository/repository.service';
|
||||
import { WAMonitoringService } from '@api/services/monitor.service';
|
||||
import { Integration } from '@api/types/wa.types';
|
||||
import { Auth, ConfigService, HttpServer } from '@config/env.config';
|
||||
import { Logger } from '@config/logger.config';
|
||||
import { Flowise, FlowiseSetting, IntegrationSession } from '@prisma/client';
|
||||
import { sendTelemetry } from '@utils/sendTelemetry';
|
||||
import axios from 'axios';
|
||||
|
||||
export class FlowiseService {
|
||||
constructor(
|
||||
private readonly waMonitor: WAMonitoringService,
|
||||
private readonly configService: ConfigService,
|
||||
private readonly prismaRepository: PrismaRepository,
|
||||
) {}
|
||||
import { BaseChatbotService } from '../../base-chatbot.service';
|
||||
|
||||
private readonly logger = new Logger('FlowiseService');
|
||||
export class FlowiseService extends BaseChatbotService<Flowise, FlowiseSetting> {
|
||||
constructor(waMonitor: WAMonitoringService, configService: ConfigService, prismaRepository: PrismaRepository) {
|
||||
super(waMonitor, prismaRepository, 'FlowiseService', configService);
|
||||
}
|
||||
|
||||
public async createNewSession(instance: InstanceDto, data: any) {
|
||||
/**
|
||||
* Get the bot type identifier
|
||||
*/
|
||||
protected getBotType(): string {
|
||||
return 'flowise';
|
||||
}
|
||||
|
||||
/**
|
||||
* Send a message to the Flowise API
|
||||
*/
|
||||
protected async sendMessageToBot(
|
||||
instance: any,
|
||||
session: IntegrationSession,
|
||||
settings: FlowiseSetting,
|
||||
bot: Flowise,
|
||||
remoteJid: string,
|
||||
pushName: string,
|
||||
content: string,
|
||||
msg?: any,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const session = await this.prismaRepository.integrationSession.create({
|
||||
data: {
|
||||
remoteJid: data.remoteJid,
|
||||
pushName: data.pushName,
|
||||
sessionId: data.remoteJid,
|
||||
status: 'opened',
|
||||
awaitUser: false,
|
||||
botId: data.botId,
|
||||
instanceId: instance.instanceId,
|
||||
type: 'flowise',
|
||||
},
|
||||
});
|
||||
|
||||
return { session };
|
||||
} catch (error) {
|
||||
this.logger.error(error);
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
private isImageMessage(content: string) {
|
||||
return content.includes('imageMessage');
|
||||
}
|
||||
|
||||
private async sendMessageToBot(instance: any, bot: Flowise, remoteJid: string, pushName: string, content: string) {
|
||||
const payload: any = {
|
||||
question: content,
|
||||
overrideConfig: {
|
||||
sessionId: remoteJid,
|
||||
vars: {
|
||||
remoteJid: remoteJid,
|
||||
pushName: pushName,
|
||||
instanceName: instance.instanceName,
|
||||
serverUrl: this.configService.get<HttpServer>('SERVER').URL,
|
||||
apiKey: this.configService.get<Auth>('AUTHENTICATION').API_KEY.KEY,
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
if (this.isImageMessage(content)) {
|
||||
const contentSplit = content.split('|');
|
||||
|
||||
payload.uploads = [
|
||||
{
|
||||
data: contentSplit[1].split('?')[0],
|
||||
type: 'url',
|
||||
name: 'Flowise.png',
|
||||
mime: 'image/png',
|
||||
},
|
||||
];
|
||||
payload.question = contentSplit[2] || content;
|
||||
}
|
||||
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS) {
|
||||
await instance.client.presenceSubscribe(remoteJid);
|
||||
await instance.client.sendPresenceUpdate('composing', remoteJid);
|
||||
}
|
||||
|
||||
let headers: any = {
|
||||
'Content-Type': 'application/json',
|
||||
};
|
||||
|
||||
if (bot.apiKey) {
|
||||
headers = {
|
||||
...headers,
|
||||
Authorization: `Bearer ${bot.apiKey}`,
|
||||
};
|
||||
}
|
||||
|
||||
const endpoint = bot.apiUrl;
|
||||
|
||||
if (!endpoint) return null;
|
||||
|
||||
const response = await axios.post(endpoint, payload, {
|
||||
headers,
|
||||
});
|
||||
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS)
|
||||
await instance.client.sendPresenceUpdate('paused', remoteJid);
|
||||
|
||||
const message = response?.data?.text;
|
||||
|
||||
return message;
|
||||
}
|
||||
|
||||
private async sendMessageWhatsApp(
|
||||
instance: any,
|
||||
remoteJid: string,
|
||||
session: IntegrationSession,
|
||||
settings: FlowiseSetting,
|
||||
message: string,
|
||||
) {
|
||||
const linkRegex = /(!?)\[(.*?)\]\((.*?)\)/g;
|
||||
|
||||
let textBuffer = '';
|
||||
let lastIndex = 0;
|
||||
|
||||
let match: RegExpExecArray | null;
|
||||
|
||||
const getMediaType = (url: string): string | null => {
|
||||
const extension = url.split('.').pop()?.toLowerCase();
|
||||
const imageExtensions = ['jpg', 'jpeg', 'png', 'gif', 'bmp', 'webp'];
|
||||
const audioExtensions = ['mp3', 'wav', 'aac', 'ogg'];
|
||||
const videoExtensions = ['mp4', 'avi', 'mkv', 'mov'];
|
||||
const documentExtensions = ['pdf', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx', 'txt'];
|
||||
|
||||
if (imageExtensions.includes(extension || '')) return 'image';
|
||||
if (audioExtensions.includes(extension || '')) return 'audio';
|
||||
if (videoExtensions.includes(extension || '')) return 'video';
|
||||
if (documentExtensions.includes(extension || '')) return 'document';
|
||||
return null;
|
||||
};
|
||||
|
||||
while ((match = linkRegex.exec(message)) !== null) {
|
||||
const [fullMatch, exclMark, altText, url] = match;
|
||||
const mediaType = getMediaType(url);
|
||||
|
||||
const beforeText = message.slice(lastIndex, match.index);
|
||||
if (beforeText) {
|
||||
textBuffer += beforeText;
|
||||
}
|
||||
|
||||
if (mediaType) {
|
||||
const splitMessages = settings.splitMessages ?? false;
|
||||
const timePerChar = settings.timePerChar ?? 0;
|
||||
const minDelay = 1000;
|
||||
const maxDelay = 20000;
|
||||
|
||||
if (textBuffer.trim()) {
|
||||
if (splitMessages) {
|
||||
const multipleMessages = textBuffer.trim().split('\n\n');
|
||||
|
||||
for (let index = 0; index < multipleMessages.length; index++) {
|
||||
const message = multipleMessages[index];
|
||||
|
||||
const delay = Math.min(Math.max(message.length * timePerChar, minDelay), maxDelay);
|
||||
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS) {
|
||||
await instance.client.presenceSubscribe(remoteJid);
|
||||
await instance.client.sendPresenceUpdate('composing', remoteJid);
|
||||
}
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
setTimeout(async () => {
|
||||
await instance.textMessage(
|
||||
{
|
||||
number: remoteJid.split('@')[0],
|
||||
delay: settings?.delayMessage || 1000,
|
||||
text: message,
|
||||
},
|
||||
false,
|
||||
);
|
||||
resolve();
|
||||
}, delay);
|
||||
});
|
||||
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS) {
|
||||
await instance.client.sendPresenceUpdate('paused', remoteJid);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
await instance.textMessage(
|
||||
{
|
||||
number: remoteJid.split('@')[0],
|
||||
delay: settings?.delayMessage || 1000,
|
||||
text: textBuffer.trim(),
|
||||
},
|
||||
false,
|
||||
);
|
||||
}
|
||||
textBuffer = '';
|
||||
}
|
||||
|
||||
if (mediaType === 'audio') {
|
||||
await instance.audioWhatsapp({
|
||||
number: remoteJid.split('@')[0],
|
||||
delay: settings?.delayMessage || 1000,
|
||||
audio: url,
|
||||
caption: altText,
|
||||
});
|
||||
} else {
|
||||
await instance.mediaMessage(
|
||||
{
|
||||
number: remoteJid.split('@')[0],
|
||||
delay: settings?.delayMessage || 1000,
|
||||
mediatype: mediaType,
|
||||
media: url,
|
||||
caption: altText,
|
||||
},
|
||||
null,
|
||||
false,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
textBuffer += `[${altText}](${url})`;
|
||||
}
|
||||
|
||||
lastIndex = linkRegex.lastIndex;
|
||||
}
|
||||
|
||||
if (lastIndex < message.length) {
|
||||
const remainingText = message.slice(lastIndex);
|
||||
if (remainingText.trim()) {
|
||||
textBuffer += remainingText;
|
||||
}
|
||||
}
|
||||
|
||||
const splitMessages = settings.splitMessages ?? false;
|
||||
const timePerChar = settings.timePerChar ?? 0;
|
||||
const minDelay = 1000;
|
||||
const maxDelay = 20000;
|
||||
|
||||
if (textBuffer.trim()) {
|
||||
if (splitMessages) {
|
||||
const multipleMessages = textBuffer.trim().split('\n\n');
|
||||
|
||||
for (let index = 0; index < multipleMessages.length; index++) {
|
||||
const message = multipleMessages[index];
|
||||
|
||||
const delay = Math.min(Math.max(message.length * timePerChar, minDelay), maxDelay);
|
||||
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS) {
|
||||
await instance.client.presenceSubscribe(remoteJid);
|
||||
await instance.client.sendPresenceUpdate('composing', remoteJid);
|
||||
}
|
||||
|
||||
await new Promise<void>((resolve) => {
|
||||
setTimeout(async () => {
|
||||
await instance.textMessage(
|
||||
{
|
||||
number: remoteJid.split('@')[0],
|
||||
delay: settings?.delayMessage || 1000,
|
||||
text: message,
|
||||
},
|
||||
false,
|
||||
);
|
||||
resolve();
|
||||
}, delay);
|
||||
});
|
||||
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS) {
|
||||
await instance.client.sendPresenceUpdate('paused', remoteJid);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
await instance.textMessage(
|
||||
{
|
||||
number: remoteJid.split('@')[0],
|
||||
delay: settings?.delayMessage || 1000,
|
||||
text: textBuffer.trim(),
|
||||
const payload: any = {
|
||||
question: content,
|
||||
overrideConfig: {
|
||||
sessionId: remoteJid,
|
||||
vars: {
|
||||
remoteJid: remoteJid,
|
||||
pushName: pushName,
|
||||
instanceName: instance.instanceName,
|
||||
serverUrl: this.configService.get<HttpServer>('SERVER').URL,
|
||||
apiKey: this.configService.get<Auth>('AUTHENTICATION').API_KEY.KEY,
|
||||
},
|
||||
false,
|
||||
);
|
||||
},
|
||||
};
|
||||
|
||||
if (this.isImageMessage(content)) {
|
||||
const contentSplit = content.split('|');
|
||||
|
||||
payload.uploads = [
|
||||
{
|
||||
data: contentSplit[1].split('?')[0],
|
||||
type: 'url',
|
||||
name: 'Flowise.png',
|
||||
mime: 'image/png',
|
||||
},
|
||||
];
|
||||
payload.question = contentSplit[2] || content;
|
||||
}
|
||||
textBuffer = '';
|
||||
}
|
||||
|
||||
sendTelemetry('/message/sendText');
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS) {
|
||||
await instance.client.presenceSubscribe(remoteJid);
|
||||
await instance.client.sendPresenceUpdate('composing', remoteJid);
|
||||
}
|
||||
|
||||
await this.prismaRepository.integrationSession.update({
|
||||
where: {
|
||||
id: session.id,
|
||||
},
|
||||
data: {
|
||||
status: 'opened',
|
||||
awaitUser: true,
|
||||
},
|
||||
});
|
||||
let headers: any = {
|
||||
'Content-Type': 'application/json',
|
||||
};
|
||||
|
||||
return;
|
||||
}
|
||||
if (bot.apiKey) {
|
||||
headers = {
|
||||
...headers,
|
||||
Authorization: `Bearer ${bot.apiKey}`,
|
||||
};
|
||||
}
|
||||
|
||||
private async initNewSession(
|
||||
instance: any,
|
||||
remoteJid: string,
|
||||
bot: Flowise,
|
||||
settings: FlowiseSetting,
|
||||
session: IntegrationSession,
|
||||
content: string,
|
||||
pushName?: string,
|
||||
) {
|
||||
const data = await this.createNewSession(instance, {
|
||||
remoteJid,
|
||||
pushName,
|
||||
botId: bot.id,
|
||||
});
|
||||
const endpoint = bot.apiUrl;
|
||||
|
||||
if (data.session) {
|
||||
session = data.session;
|
||||
}
|
||||
|
||||
const message = await this.sendMessageToBot(instance, bot, remoteJid, pushName, content);
|
||||
|
||||
await this.sendMessageWhatsApp(instance, remoteJid, session, settings, message);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
public async processBot(
|
||||
instance: any,
|
||||
remoteJid: string,
|
||||
bot: Flowise,
|
||||
session: IntegrationSession,
|
||||
settings: FlowiseSetting,
|
||||
content: string,
|
||||
pushName?: string,
|
||||
) {
|
||||
if (session && session.status !== 'opened') {
|
||||
return;
|
||||
}
|
||||
|
||||
if (session && settings.expire && settings.expire > 0) {
|
||||
const now = Date.now();
|
||||
|
||||
const sessionUpdatedAt = new Date(session.updatedAt).getTime();
|
||||
|
||||
const diff = now - sessionUpdatedAt;
|
||||
|
||||
const diffInMinutes = Math.floor(diff / 1000 / 60);
|
||||
|
||||
if (diffInMinutes > settings.expire) {
|
||||
if (settings.keepOpen) {
|
||||
await this.prismaRepository.integrationSession.update({
|
||||
where: {
|
||||
id: session.id,
|
||||
},
|
||||
data: {
|
||||
status: 'closed',
|
||||
},
|
||||
});
|
||||
} else {
|
||||
await this.prismaRepository.integrationSession.deleteMany({
|
||||
where: {
|
||||
botId: bot.id,
|
||||
remoteJid: remoteJid,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
await this.initNewSession(instance, remoteJid, bot, settings, session, content, pushName);
|
||||
if (!endpoint) {
|
||||
this.logger.error('No Flowise endpoint defined');
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
if (!session) {
|
||||
await this.initNewSession(instance, remoteJid, bot, settings, session, content, pushName);
|
||||
return;
|
||||
}
|
||||
const response = await axios.post(endpoint, payload, {
|
||||
headers,
|
||||
});
|
||||
|
||||
await this.prismaRepository.integrationSession.update({
|
||||
where: {
|
||||
id: session.id,
|
||||
},
|
||||
data: {
|
||||
status: 'opened',
|
||||
awaitUser: false,
|
||||
},
|
||||
});
|
||||
|
||||
if (!content) {
|
||||
if (settings.unknownMessage) {
|
||||
this.waMonitor.waInstances[instance.instanceName].textMessage(
|
||||
{
|
||||
number: remoteJid.split('@')[0],
|
||||
delay: settings.delayMessage || 1000,
|
||||
text: settings.unknownMessage,
|
||||
},
|
||||
false,
|
||||
);
|
||||
|
||||
sendTelemetry('/message/sendText');
|
||||
if (instance.integration === Integration.WHATSAPP_BAILEYS) {
|
||||
await instance.client.sendPresenceUpdate('paused', remoteJid);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
if (settings.keywordFinish && content.toLowerCase() === settings.keywordFinish.toLowerCase()) {
|
||||
if (settings.keepOpen) {
|
||||
await this.prismaRepository.integrationSession.update({
|
||||
where: {
|
||||
id: session.id,
|
||||
},
|
||||
data: {
|
||||
status: 'closed',
|
||||
},
|
||||
});
|
||||
} else {
|
||||
await this.prismaRepository.integrationSession.deleteMany({
|
||||
where: {
|
||||
botId: bot.id,
|
||||
remoteJid: remoteJid,
|
||||
},
|
||||
});
|
||||
const message = response?.data?.text;
|
||||
|
||||
if (message) {
|
||||
// Use the base class method to send the message to WhatsApp
|
||||
await this.sendMessageWhatsApp(instance, remoteJid, message, settings);
|
||||
}
|
||||
|
||||
// Send telemetry
|
||||
sendTelemetry('/message/sendText');
|
||||
} catch (error) {
|
||||
this.logger.error(`Error in sendMessageToBot: ${error.message || JSON.stringify(error)}`);
|
||||
return;
|
||||
}
|
||||
|
||||
const message = await this.sendMessageToBot(instance, bot, remoteJid, pushName, content);
|
||||
|
||||
await this.sendMessageWhatsApp(instance, remoteJid, session, settings, message);
|
||||
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user