Mail channel initial

kairon/__init__.py

@@ -44,7 +44,8 @@
 
 
 def create_argument_parser():
-    from kairon.cli import importer, training, testing, conversations_deletion, translator, delete_logs, message_broadcast,content_importer
+    from kairon.cli import importer, training, testing, conversations_deletion, translator, delete_logs,\
+        message_broadcast,content_importer, mail_channel
 
     parser = ArgumentParser(
         prog="kairon",
@@ -62,6 +63,7 @@ def create_argument_parser():
     delete_logs.add_subparser(subparsers, parents=parent_parsers)
     message_broadcast.add_subparser(subparsers, parents=parent_parsers)
     content_importer.add_subparser(subparsers, parents=parent_parsers)
+    mail_channel.add_subparser(subparsers, parents=parent_parsers)
     return parser
 
 

kairon/api/app/routers/bot/bot.py

@@ -25,6 +25,7 @@
 from kairon.shared.account.activity_log import UserActivityLogger
 from kairon.shared.actions.data_objects import ActionServerLogs
 from kairon.shared.auth import Authentication
+from kairon.shared.channels.mail.data_objects import MailClassificationConfig
 from kairon.shared.constants import TESTER_ACCESS, DESIGNER_ACCESS, CHAT_ACCESS, UserActivityType, ADMIN_ACCESS, \
     EventClass, AGENT_ACCESS
 from kairon.shared.content_importer.content_processor import ContentImporterLogProcessor
@@ -33,6 +34,7 @@
 from kairon.shared.data.audit.processor import AuditDataProcessor
 from kairon.shared.data.constant import ENDPOINT_TYPE, ModelTestType, \
     AuditlogActions
+from kairon.shared.data.data_models import MailConfigRequest
 from kairon.shared.data.data_objects import TrainingExamples, ModelTraining, Rules
 from kairon.shared.data.model_processor import ModelProcessor
 from kairon.shared.data.processor import MongoProcessor
@@ -1658,3 +1660,60 @@ async def get_slot_actions(
     llm_models = MongoProcessor.get_slot_mapped_actions(current_user.get_bot(), slot_name)
     return Response(data=llm_models)
 
+
+@router.get("/mail/config", response_model=Response)
+async def get_all_mail_configs(
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=TESTER_ACCESS)):
+    """
+    Fetches mail config
+    """
+    data = MailClassificationConfig.objects(bot=current_user.get_bot(), status=True)
+    formatted_data = [
+        {key: value for key, value in item.to_mongo().items() if key not in {"_id", "user"}}
+        for item in data
+    ]
+
+    return {"data": formatted_data}
+
+
+
+@router.post("/mail/config", response_model=Response)
+async def set_mail_config(
+        request_data: MailConfigRequest,
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=DESIGNER_ACCESS)
+):
+    """
+    Applies the mail config
+    """
+    request_dict = request_data.dict()
+    MailClassificationConfig.create_doc(**request_dict, bot=current_user.get_bot(), user=current_user.get_user())
+    return {"message": "Config applied!"}
+
+
+@router.put("/mail/config", response_model=Response)
+async def update_mail_config(
+        request_data: MailConfigRequest,
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=DESIGNER_ACCESS)
+):
+    """
+    update the mail config
+    """
+    request_dict = request_data.dict()
+    MailClassificationConfig.update_doc(**request_dict, bot=current_user.get_bot())
+    return {"message": "Config updated!"}
+
+
+
+@router.delete("/mail/config/{intent}", response_model=Response)
+async def del_soft_mail_config(
+        intent: str,
+        current_user: User = Security(Authentication.get_current_user_and_bot, scopes=DESIGNER_ACCESS)
+):
+    """
+    delete the mail config
+    """
+    MailClassificationConfig.soft_delete_doc(current_user.get_bot(), intent)
+    return {"message": "Config deleted!"}
+
+
+

kairon/chat/agent_processor.py

@@ -78,3 +78,22 @@ async def handle_channel_message(bot: Text, userdata: UserMessage):
         if not is_live_agent_enabled:
             return await AgentProcessor.get_agent(bot).handle_message(userdata)
         return await LiveAgentHandler.process_live_agent(bot, userdata)
+
+    @staticmethod
+    def get_agent_without_cache(bot: str, use_store: bool = True) -> Agent:
+        endpoint = AgentProcessor.mongo_processor.get_endpoints(
+            bot, raise_exception=False
+        )
+        action_endpoint = Utility.get_action_url(endpoint)
+        model_path = Utility.get_latest_model(bot)
+        domain = AgentProcessor.mongo_processor.load_domain(bot)
+        if use_store:
+            mongo_store = Utility.get_local_mongo_store(bot, domain)
+            lock_store_endpoint = Utility.get_lock_store(bot)
+            agent = KaironAgent.load(model_path, action_endpoint=action_endpoint, tracker_store=mongo_store,
+                                     lock_store=lock_store_endpoint)
+        else:
+            agent = KaironAgent.load(model_path, action_endpoint=action_endpoint)
+
+        agent.model_ver = model_path.split("/")[-1]
+        return agent

kairon/chat/utils.py

@@ -43,6 +43,37 @@ async def chat(
         )
         return chat_response
 
+    @staticmethod
+    async  def process_messages_via_bot(
+            messages: [str],
+            account: int,
+            bot: str,
+            user: str,
+            is_integration_user: bool = False,
+            metadata: Dict = None,
+    ):
+        """Process a list of messages through the bot.
+         Args:
+             messages: List of messages to process
+             account: Account ID
+             bot: Bot ID
+             user: User ID
+             is_integration_user: Flag indicating if user is integration user
+             metadata: Additional metadata
+
+         Returns:
+             List of responses from the bot
+        """
+        responses = []
+        uncached_model = AgentProcessor.get_agent_without_cache(bot, False)
+        metadata = ChatUtils.get_metadata(account, bot, is_integration_user, metadata)
+        for message in messages:
+            msg = UserMessage(message, sender_id=user, metadata=metadata)
+            chat_response = await uncached_model.handle_message(msg)
+            responses.append(chat_response)
+        return responses
+
+
     @staticmethod
     def reload(bot: Text, user: Text):
         exc = None

kairon/cli/mail_channel.py

@@ -0,0 +1,36 @@
+import json
+from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
+from typing import List
+from rasa.cli import SubParsersAction
+
+from kairon.events.definitions.mail_channel_schedule import MailChannelScheduleEvent
+
+
+def process_channel_mails(args):
+    mails = json.loads(args.mails)
+    if not isinstance(mails, list):
+        raise ValueError("Mails should be a list")
+    MailChannelScheduleEvent(args.bot, args.user).execute(mails=mails)
+
+
+def add_subparser(subparsers: SubParsersAction, parents: List[ArgumentParser]):
+    mail_parser = subparsers.add_parser(
+        "mail-channel",
+        conflict_handler="resolve",
+        formatter_class=ArgumentDefaultsHelpFormatter,
+        parents=parents,
+        help="Mail channel"
+    )
+    mail_parser.add_argument('bot',
+                             type=str,
+                             help="Bot id for which command is executed", action='store')
+
+    mail_parser.add_argument('user',
+                            type=str,
+                            help="Kairon user who is initiating the command", action='store')
+
+    mail_parser.add_argument('mails',
+                            type=str,
+                            help="json representing List of mails to be processed", action='store')
+
+    mail_parser.set_defaults(func=process_channel_mails)

kairon/events/definitions/factory.py

@@ -2,6 +2,7 @@
 from kairon.events.definitions.data_importer import TrainingDataImporterEvent
 from kairon.events.definitions.faq_importer import FaqDataImporterEvent
 from kairon.events.definitions.history_delete import DeleteHistoryEvent
+from kairon.events.definitions.mail_channel_schedule import MailChannelScheduleEvent
 from kairon.events.definitions.message_broadcast import MessageBroadcastEvent
 from kairon.events.definitions.model_testing import ModelTestingEvent
 from kairon.events.definitions.model_training import ModelTrainingEvent
@@ -20,7 +21,8 @@ class EventFactory:
         EventClass.multilingual: MultilingualEvent,
         EventClass.faq_importer: FaqDataImporterEvent,
         EventClass.message_broadcast: MessageBroadcastEvent,
-        EventClass.content_importer: DocContentImporterEvent
+        EventClass.content_importer: DocContentImporterEvent,
+        EventClass.email_channel_scheduler: MailChannelScheduleEvent
     }
 
     @staticmethod

kairon/events/definitions/mail_channel_schedule.py

@@ -0,0 +1,51 @@
+from typing import Text
+from loguru import logger
+from kairon import Utility
+from kairon.events.definitions.base import EventsBase
+from kairon.exceptions import AppException
+from kairon.shared.channels.mail.processor import MailProcessor
+from kairon.shared.constants import EventClass
+
+
+class MailChannelScheduleEvent(EventsBase):
+    """
+    Event to start mail channel scheduler if not already running.
+    """
+
+    def __init__(self, bot: Text, user: Text, **kwargs):
+        """
+        Initialise event.
+        """
+        self.bot = bot
+        self.user = user
+
+    def validate(self):
+        """
+        validate mail channel exists
+        """
+        return MailProcessor.validate_smpt_connection(self.bot)
+
+
+    def enqueue(self, **kwargs):
+        """
+        Send event to event server.
+        """
+        try:
+            mails: list = kwargs.get('mails', [])
+            payload = {'bot': self.bot, 'user': self.user, 'mails': mails}
+            Utility.request_event_server(EventClass.email_channel_scheduler, payload)
+        except Exception as e:
+            logger.error(str(e))
+            raise AppException(e)
+
+    def execute(self, **kwargs):
+        """
+        Execute the event.
+        """
+        try:
+            mails = kwargs.get('mails')
+            if mails:
+                MailProcessor.process_message_task(self.bot, mails)
+        except Exception as e:
+            logger.error(str(e))
+            raise AppException(e)

kairon/events/server.py

@@ -56,6 +56,8 @@ async def lifespan(app: FastAPI):
     """ MongoDB is connected on the bot trainer startup """
     config: dict = Utility.mongoengine_connection(Utility.environment['database']["url"])
     connect(**config)
+    from kairon.shared.channels.mail.scheduler import MailScheduler
+    MailScheduler.epoch()
     yield
     disconnect()
 
@@ -144,3 +146,10 @@ def delete_scheduled_event(event_id: Text = Path(description="Event id")):
 def dispatch_scheduled_event(event_id: Text = Path(description="Event id")):
     KScheduler().dispatch_event(event_id)
     return {"data": None, "message": "Scheduled event dispatch!"}
+
+
+@app.get('/api/mail/request_epoch', response_model=Response)
+def request_epoch():
+    from kairon.shared.channels.mail.scheduler import MailScheduler
+    MailScheduler.epoch()
+    return {"data": None, "message": "Mail scheduler epoch request!"}

kairon/shared/channels/mail/constants.py

@@ -0,0 +1,25 @@
+
+
+class MailConstants:
+    DEFAULT_SMTP_SERVER = 'smtp.gmail.com'
+    DEFAULT_IMAP_SERVER = 'imap.gmail.com'
+    DEFAULT_SMTP_PORT = 587
+    DEFAULT_LLM_TYPE = "openai"
+    DEFAULT_HYPERPARAMETERS = {
+        "frequency_penalty": 0,
+        "logit_bias": {},
+        "max_tokens": 300,
+        "model": "gpt-4o-mini",
+        "n": 1,
+        "presence_penalty": 0,
+        "stop": None,
+        "stream": False,
+        "temperature": 0,
+        "top_p": 0
+    }
+    DEFAULT_TEMPLATE = "<p>Dear {name},</p> <p>{bot_response}</p> <br/><br/><span style='color:#999;'> Generated by kAIron AI.</span>\n"
+    DEFAULT_SYSTEM_PROMPT = 'Classify into one of the intents and extract entities as given in the context.' \
+                            'If the mail does not belong to any of the intents, classify intent as null.'
+
+    PROCESS_MESSAGE_BATCH_SIZE = 4
+