Mail channel first iteration

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_process, mail_channel_read
 
     parser = ArgumentParser(
         prog="kairon",
@@ -62,6 +63,8 @@ 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_process.add_subparser(subparsers, parents=parent_parsers)
+    mail_channel_read.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.processor import MailProcessor
 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
@@ -1658,3 +1659,14 @@ 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_channel/logs", response_model=Response)
+async def get_mail_channel_logs(start_idx: int = 0, page_size: int = 10,
+                                 current_user: User = Security(Authentication.get_current_user_and_bot,
+                                                               scopes=TESTER_ACCESS)):
+    """
+    Retrieves mail channel related logs for the bot.
+    """
+    data = MailProcessor.get_log(current_user.get_bot(), start_idx, page_size)
+    return Response(data=data)

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_process.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 import MailProcessEvent
+
+
+def process_channel_mails(args):
+    mails = json.loads(args.mails)
+    if not isinstance(mails, list):
+        raise ValueError("Mails should be a list")
+    MailProcessEvent(args.bot, args.user).execute(mails=mails)
+
+
+def add_subparser(subparsers: SubParsersAction, parents: List[ArgumentParser]):
+    mail_parser = subparsers.add_parser(
+        "mail-channel-process",
+        conflict_handler="resolve",
+        formatter_class=ArgumentDefaultsHelpFormatter,
+        parents=parents,
+        help="Mail channel process mails"
+    )
+    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/cli/mail_channel_read.py

@@ -0,0 +1,28 @@
+from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
+from typing import List
+from rasa.cli import SubParsersAction
+
+from kairon.events.definitions.mail_channel import MailReadEvent
+
+
+def read_channel_mails(args):
+    MailReadEvent(args.bot, args.user).execute()
+
+
+def add_subparser(subparsers: SubParsersAction, parents: List[ArgumentParser]):
+    mail_parser = subparsers.add_parser(
+        "mail-channel-read",
+        conflict_handler="resolve",
+        formatter_class=ArgumentDefaultsHelpFormatter,
+        parents=parents,
+        help="Mail channel initiate reading"
+    )
+    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.set_defaults(func=read_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 import MailProcessEvent, MailReadEvent
 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,9 @@ class EventFactory:
         EventClass.multilingual: MultilingualEvent,
         EventClass.faq_importer: FaqDataImporterEvent,
         EventClass.message_broadcast: MessageBroadcastEvent,
-        EventClass.content_importer: DocContentImporterEvent
+        EventClass.content_importer: DocContentImporterEvent,
+        EventClass.mail_channel_read_mails: MailReadEvent,
+        EventClass.mail_channel_process_mails: MailProcessEvent
     }
 
     @staticmethod

kairon/events/definitions/mail_channel.py

@@ -0,0 +1,98 @@
+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 MailProcessEvent(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_smtp_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.mail_channel_process_mails, 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)
+
+
+
+class MailReadEvent(EventsBase):
+    """
+    Event to read mails from mail channel and create events for each mail tp process them via bot
+    """
+
+    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_imap_connection(self.bot)
+
+    def enqueue(self, **kwargs):
+        """
+        Send event to event server.
+        """
+        try:
+            payload = {'bot': self.bot, 'user': self.user}
+            Utility.request_event_server(EventClass.mail_channel_read_mails, payload)
+        except Exception as e:
+            logger.error(str(e))
+            raise AppException(e)
+
+    def execute(self, **kwargs):
+        """
+        Execute the event.
+        """
+        try:
+            vals = MailProcessor.read_mails(self.bot)
+            print(vals)
+            emails, _, _ = vals
+            for email in emails:
+                ev = MailProcessEvent(self.bot, self.user)
+                ev.validate()
+                ev.enqueue(mails=[email])
+
+        except Exception as e:
+            raise AppException(f"Failed to schedule mail reading for bot {self.bot}. Error: {str(e)}")

kairon/events/server.py

@@ -56,6 +56,7 @@ async def lifespan(app: FastAPI):
     """ MongoDB is connected on the bot trainer startup """
     config: dict = Utility.mongoengine_connection(Utility.environment['database']["url"])
     connect(**config)
+    EventUtility.reschedule_all_bots_channel_mail_reading()
     yield
     disconnect()
 
@@ -144,3 +145,9 @@ 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/schedule/{bot}', response_model=Response)
+def request_epoch(bot: Text = Path(description="Bot id")):
+    EventUtility.schedule_channel_mail_reading(bot)
+    return {"data": None, "message": "Mail scheduler epoch request!"}

kairon/events/utility.py

@@ -1,9 +1,14 @@
 from typing import Dict, Text
 
+from uuid6 import uuid7
+
 from kairon.events.executors.factory import ExecutorFactory
 from kairon.events.scheduler.kscheduler import KScheduler
 from kairon.exceptions import AppException
+from kairon.shared.chat.data_objects import Channels
+from kairon.shared.constants import EventClass, ChannelTypes
 from kairon.shared.data.constant import TASK_TYPE
+from loguru import logger
 
 
 class EventUtility:
@@ -32,3 +37,36 @@ def update_job(event_type: Text, request_data: Dict, is_scheduled: bool):
         KScheduler().update_job(event_class=event_type, event_id=event_id, task_type=TASK_TYPE.EVENT.value,
                                 **request_data)
         return None, 'Scheduled event updated!'
+
+    @staticmethod
+    def schedule_channel_mail_reading(bot: str):
+        from kairon.shared.channels.mail.processor import MailProcessor
+
+        try:
+            mail_processor = MailProcessor(bot)
+            interval = mail_processor.config.get("interval", 60)
+            event_id = mail_processor.state.event_id
+            if event_id:
+                KScheduler().update_job(event_id,
+                                        TASK_TYPE.EVENT,
+                                        f"*/{interval} * * * *",
+                                        EventClass.mail_channel_read_mails, {"bot": bot, "user": mail_processor.bot_settings.user})
+            else:
+                event_id = uuid7().hex
+                mail_processor.update_event_id(event_id)
+                KScheduler().add_job(event_id,
+                                     TASK_TYPE.EVENT,
+                                     f"*/{interval} * * * *",
+                                     EventClass.mail_channel_read_mails, {"bot": bot, "user": mail_processor.bot_settings.user})
+        except Exception as e:
+            raise AppException(f"Failed to schedule mail reading for bot {bot}. Error: {str(e)}")
+
+    @staticmethod
+    def reschedule_all_bots_channel_mail_reading():
+        try:
+            bots = list(Channels.objects(connector_type= ChannelTypes.MAIL.value).distinct("bot"))
+            for bot in bots:
+                logger.info(f"Rescheduling mail reading for bot {bot}")
+                EventUtility.schedule_channel_mail_reading(bot)
+        except Exception as e:
+            raise AppException(f"Failed to reschedule mail reading events. Error: {str(e)}")

kairon/shared/channels/mail/constants.py

@@ -0,0 +1,11 @@
+
+
+class MailConstants:
+    DEFAULT_SMTP_SERVER = 'smtp.gmail.com'
+    DEFAULT_IMAP_SERVER = 'imap.gmail.com'
+    DEFAULT_SMTP_PORT = 587
+
+    DEFAULT_TEMPLATE = "<p>{bot_response}</p> <br/><br/><span style='color:#999;'> Generated by kAIron AI.</span>\n"
+
+    PROCESS_MESSAGE_BATCH_SIZE = 4
+