25849035942

Committed 14 May 2026 07:59AM UTC coverage: 91.212% (-0.004%) from 91.216%

Build # 25849035942

Build Type

Pull #2373

github

Committed by

web-flow

Commit Message

Merge c5fdafc5e into ec639b034

Pull Request Pull Request #2373: Added validation in analytics pipeline for creating vector collection

Coverage Stats

6 of 8 new or added lines in 1 file covered. (75.0%)

13 existing lines in 1 file now uncovered.

30764 of 33728 relevant lines covered (91.21%)

0.91 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.4

/kairon/shared/pyscript/callback_pyscript_utils.py

import pickle
from calendar import timegm
from datetime import datetime, date
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from smtplib import SMTP
from typing import Text, Dict, Callable, List
import base64
from apscheduler.triggers.date import DateTrigger
from apscheduler.util import obj_to_ref, astimezone
from mongoengine import DoesNotExist
from pymongo import MongoClient
from tzlocal import get_localzone
from uuid6 import uuid7
from loguru import logger
from kairon import Utility
from kairon.events.executors.factory import ExecutorFactory
from kairon.exceptions import AppException
from kairon.shared.actions.data_objects import EmailActionConfig
from kairon.shared.actions.utils import ActionUtility
from bson import Binary
from types import ModuleType
from requests import Response
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.asymmetric import padding as asym_padding
from cryptography.hazmat.primitives.serialization import load_pem_private_key
from kairon.shared.callback.data_objects import CallbackConfig, CallbackData
import json as jsond

from mongoengine.errors import DoesNotExist
from kairon.shared.data.data_objects import BotSettings
from kairon.shared.chat.user_media import UserMedia
from kairon.shared.cognition.data_objects import AnalyticsCollectionData


class CallbackScriptUtility:

    @staticmethod
    def generate_id():
        return uuid7().hex


    @staticmethod
    def datetime_to_utc_timestamp(timeval):
        """
        Converts a datetime instance to a timestamp.

        :type timeval: datetime
        :rtype: float

        """
        if timeval is not None:
            return timegm(timeval.utctimetuple()) + timeval.microsecond / 1000000


    @staticmethod
    def add_schedule_job(schedule_action: Text, date_time: datetime, data: Dict, timezone: Text, _id: Text = None,
                         bot: Text = None, kwargs=None):
        if not bot:
            raise AppException("Missing bot id")

        if not _id:
            _id = uuid7().hex

        if not data:
            data = {}

        data['bot'] = bot
        data['event'] = _id

        callback_config = CallbackConfig.get_entry(bot=bot, name=schedule_action)

        script = callback_config.get('pyscript_code')

        func = obj_to_ref(ExecutorFactory.get_executor_for_data(data).execute_task)

        schedule_data = {
            'source_code': script,
            'predefined_objects': data
        }

        args = (func, "scheduler_evaluator", schedule_data,)
        kwargs = {'task_type': "Callback"} if kwargs is None else {**kwargs, 'task_type': "Callback"}
        trigger = DateTrigger(run_date=date_time, timezone=timezone)

        next_run_time = trigger.get_next_fire_time(None, datetime.now(astimezone(timezone) or get_localzone()))

        job_kwargs = {
            'version': 1,
            'trigger': trigger,
            'executor': "default",
            'func': func,
            'args': tuple(args) if args is not None else (),
            'kwargs': kwargs,
            'id': _id,
            'name': "execute_task",
            'misfire_grace_time': 7200,
            'coalesce': True,
            'next_run_time': next_run_time,
            'max_instances': 1,
        }

        logger.info(job_kwargs)

        client = MongoClient(Utility.environment['database']['url'])
        events_db_name = Utility.environment["events"]["queue"]["name"]
        events_db = client.get_database(events_db_name)
        scheduler_collection = Utility.environment["events"]["scheduler"]["collection"]
        job_store_name = events_db.get_collection(scheduler_collection)
        event_server = Utility.environment['events']['server_url']

        job_store_name.insert_one({
            '_id': _id,
            'next_run_time': CallbackScriptUtility.datetime_to_utc_timestamp(next_run_time),
            'job_state': Binary(pickle.dumps(job_kwargs, pickle.HIGHEST_PROTOCOL))
        })

        http_response = ActionUtility.execute_http_request(
            f"{event_server}/api/events/dispatch/{_id}",
            "GET")

        if not http_response.get("success"):
            raise AppException(http_response)
        else:
            logger.info(http_response)

    @staticmethod
    def trigger_email(
                email: List[str],
                subject: str,
                body: str,
                smtp_url: str,
                smtp_port: int,
                sender_email: str,
                smtp_password: str,
                smtp_userid: str = None,
                tls: bool = False,
                content_type="html",
        ):
            """
            This is a sync email trigger.
            Sends an email to the mail id of the recipient

            :param smtp_userid:
            :param sender_email:
            :param tls:
            :param smtp_port:
            :param smtp_url:
            :param email: the mail id of the recipient
            :param smtp_password:
            :param subject: the subject of the mail
            :param body: the body of the mail
            :param content_type: "plain" or "html" content
            :return: None
            """
            smtp = None
            try:
                smtp = SMTP(smtp_url, port=smtp_port, timeout=10)
                smtp.connect(smtp_url, smtp_port)
                if tls:
                    smtp.starttls()
                smtp.login(smtp_userid if smtp_userid else sender_email, smtp_password)

                from_addr = sender_email
                mime_body = MIMEText(body, content_type)
                msg = MIMEMultipart("alternative")
                msg["Subject"] = subject
                msg["From"] = from_addr
                msg["To"] = ",".join(email)
                msg.attach(mime_body)

                smtp.sendmail(from_addr, email, msg.as_string())

            except Exception as e:
                print(f"Failed to send email: {e}")
                raise
            finally:
                if smtp:
                    try:
                        smtp.quit()
                    except Exception as quit_error:
                        print(f"Failed to quit SMTP connection cleanly: {quit_error}")


    @staticmethod
    def send_email(email_action: Text,
                   from_email: Text,
                   to_email: Text,
                   subject:  Text,
                   body: Text,
                   bot: Text):
        if not bot:
            raise AppException("Missing bot id")

        email_action_config = EmailActionConfig.objects(bot=bot, action_name=email_action).first()
        if not email_action_config:
            raise AppException(f"Email action '{email_action}' not configured for bot {bot}")
        action_config = email_action_config.to_mongo().to_dict()

        smtp_password = action_config.get('smtp_password').get("value")
        smtp_userid = action_config.get('smtp_userid').get("value")

        CallbackScriptUtility.trigger_email(
            email=[to_email],
            subject=subject,
            body=body,
            smtp_url=action_config['smtp_url'],
            smtp_port=action_config['smtp_port'],
            sender_email=from_email,
            smtp_password=smtp_password,
            smtp_userid=smtp_userid,
            tls=action_config['tls']
        )

    @staticmethod
    def perform_cleanup(local_vars: Dict):
        logger.info(f"local_vars: {local_vars}")
        filtered_locals = {}
        if local_vars:
            for key, value in local_vars.items():
                if not isinstance(value, Callable) and not isinstance(value, ModuleType):
                    if isinstance(value, datetime):
                        value = value.strftime("%m/%d/%Y, %H:%M:%S")
                    elif isinstance(value, date):
                        value = value.strftime("%Y-%m-%d")
                    elif isinstance(value, Response):
                        value = value.text
                    filtered_locals[key] = value
        logger.info(f"filtered_vars: {filtered_locals}")
        return filtered_locals


    @staticmethod
    def decrypt_request(request_body, private_key_pem):
        try:
            encrypted_data_b64 = request_body.get("encrypted_flow_data")
            encrypted_aes_key_b64 = request_body.get("encrypted_aes_key")
            iv_b64 = request_body.get("initial_vector")

            if not (encrypted_data_b64 and encrypted_aes_key_b64 and iv_b64):
                raise ValueError("Missing required encrypted data fields")

            # Decode base64 inputs
            encrypted_aes_key = base64.b64decode(encrypted_aes_key_b64)
            encrypted_data = base64.b64decode(encrypted_data_b64)
            iv = base64.b64decode(iv_b64)[:16]  # Ensure IV is exactly 16 bytes

            private_key = load_pem_private_key(private_key_pem.encode(), password=None)

            # Decrypt AES key using RSA and OAEP padding
            aes_key = private_key.decrypt(
                encrypted_aes_key,
                asym_padding.OAEP(
                    mgf=asym_padding.MGF1(algorithm=hashes.SHA256()),
                    algorithm=hashes.SHA256(),
                    label=None,
                ),
            )

            if len(aes_key) not in (16, 24, 32):
                raise ValueError(f"Invalid AES key size: {len(aes_key)} bytes")

            # Extract GCM tag (last 16 bytes)
            encrypted_body = encrypted_data[:-16]
            tag = encrypted_data[-16:]

            # Decrypt AES-GCM
            cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv, tag))
            decryptor = cipher.decryptor()
            decrypted_bytes = decryptor.update(encrypted_body) + decryptor.finalize()
            decrypted_data = jsond.loads(decrypted_bytes.decode("utf-8"))

            response_dict = {
                "decryptedBody": decrypted_data,
                "aesKeyBuffer": aes_key,
                "initialVectorBuffer": iv,
            }

            return response_dict

        except Exception as e:
            raise Exception(f"decryption failed-{str(e)}")


    @staticmethod
    def encrypt_response(response_body, aes_key_buffer, initial_vector_buffer):
        try:
            if aes_key_buffer is None:
                raise ValueError("AES key cannot be None")

            if initial_vector_buffer is None:
                raise ValueError("Initialization vector (IV) cannot be None")

            # Flip the IV
            flipped_iv = bytes(byte ^ 0xFF for byte in initial_vector_buffer)

            # Encrypt using AES-GCM
            encryptor = Cipher(algorithms.AES(aes_key_buffer), modes.GCM(flipped_iv)).encryptor()
            encrypted_bytes = encryptor.update(jsond.dumps(response_body).encode("utf-8")) + encryptor.finalize()
            encrypted_data_with_tag = encrypted_bytes + encryptor.tag

            # Encode result as base64
            encoded_data = base64.b64encode(encrypted_data_with_tag).decode("utf-8")
            return encoded_data
        except Exception as e:
            raise Exception(f"encryption failed-{str(e)}")


    @staticmethod
    def create_callback(callback_name: str, metadata: dict, bot: str, sender_id: str, channel: str,
                        name: str = None):
        if not callback_name:
            raise AppException("'callback name' must be provided and cannot be empty")

        if not name:
            name=callback_name
        callback_url, identifier, standalone = CallbackData.create_entry(
            name=name,
            callback_config_name=callback_name,
            bot=bot,
            sender_id=sender_id,
            channel=channel,
            metadata=metadata,
        )
        if standalone:
            return identifier
        else:
            return callback_url

    @staticmethod
    def save_as_pdf(text: str, bot: str, sender_id:str):
        try:
            _, media_id = UserMedia.save_markdown_as_pdf(
                bot=bot,
                sender_id=sender_id,
                text=text,
                filepath="report.pdf"
            )
            return media_id
        except Exception as e:
            raise Exception(f"encryption failed-{str(e)}")

    @staticmethod
    def get_data_analytics(collection_name: str, data_filters:dict, bot: str):
        if not bot:
            raise Exception("Missing bot id")

        normalized_name = collection_name.lower()
        match_filter = {
            "bot": bot,
            "collection_name": normalized_name
        }

        # Merge dictionary filter
        if data_filters:
            match_filter.update(data_filters)
        cursor = AnalyticsCollectionData._get_collection().aggregate([
            {"$match": match_filter},
            {"$project": {
                "_id": {"$toString": "$_id"},
                "collection_name": 1,
                "received_at": {
                "$dateToString": {
                    "format": "%Y-%m-%dT%H:%M:%S.%LZ",
                    "date": "$received_at"
                    }
                },
                "source": 1,
                "is_data_processed": 1,
                "data": 1
            }}
        ])

        return {"data": list(cursor)}

    @staticmethod
    def add_data_analytics(user: str, payload, bot: str = None):
        if not bot:
            raise Exception("Missing bot id")

        if not isinstance(payload, list):
            raise Exception("Payload must be a list of dicts")

        docs = []

        for item in payload:
            docs.append({
                "bot": bot,
                "user": user,
                "collection_name": item.get("collection_name", "").lower().strip(),
                "data": item.get("data"),
                "source": item.get("source", ""),
                "received_at": item.get("received_at") or datetime.utcnow(),
                "is_data_processed": False
            })

        AnalyticsCollectionData._get_collection().insert_many(docs)

        return {
            "message": "Records saved!"
        }

    @staticmethod
    def mark_as_processed(user: str, collection_name: str, bot: str = None):
        if not bot:
            raise Exception("Missing bot id")

        collection_name = collection_name.lower().strip()

        result = AnalyticsCollectionData.objects(
            bot=bot,
            collection_name=collection_name
        ).update(
            set__user=user,
            set__is_data_processed=True,
            multi=True
        )

        if result == 0:
            raise AppException("No records found for given bot and collection_name")

        return {
            "message": "Records updated!"
        }

    @staticmethod
    def delete_data_analytics(collection_id: str, bot: Text = None):
        if not bot:
            raise Exception("Missing bot id")

        try:
            AnalyticsCollectionData.objects(bot=bot, id=collection_id).delete()
        except DoesNotExist:
            raise AppException("Analytics Collection Data does not exists!")

        return {
            "message": f"Analytics Collection with ID {collection_id} has been successfully deleted.",
            "data": {"_id": collection_id}
        }

    @staticmethod
    def update_data_analytics(collection_id: str, user: str, payload: dict, bot: str = None):
        if not bot:
            raise Exception("Missing bot id")

        collection_name = payload.get("collection_name")
        data = payload.get("data", {})
        received_at=payload.get("received_at", datetime.utcnow())
        source=payload.get("source", "")
        is_data_processed=payload.get("is_data_processed", False)

        try:
            collection_obj = AnalyticsCollectionData.objects(bot=bot, id=collection_id, collection_name=collection_name).get()
            filtered_data = {
                k: v for k, v in data.items()
            }

            collection_obj.data.update(filtered_data)
            collection_obj.collection_name = collection_name
            collection_obj.user = user
            collection_obj.received_at = received_at
            collection_obj.source=source
            collection_obj.is_data_processed=is_data_processed
            collection_obj.save()
        except DoesNotExist:
            raise AppException("Analytics Collection Data with given id and collection_name not found!")

        return {
            "message": "Record updated!",
            "data": {"_id": collection_id}
        }

    @staticmethod
    def extract_data(input_source: str,
                     prompt: str = None,
                     result_type: str="markdown",
                     llm_type: str = "openrouter",
                     high_res_ocr: bool = False,
                     language: str = "en",
                     bot: str = None,
                     user: str = None):

        import requests

        llm_server_url = Utility.environment['llm']['url']

        payload = {
            "input_source": input_source,
            "llama_parser_api_key": Utility.environment['llama_parse']['key'],
            "result_type": result_type,
            "high_res_ocr": high_res_ocr,
            "language": language,
            "parsing_instruction": prompt,
            "user": user,
            "llm_type": llm_type
        }

        response = requests.post(
            f"{llm_server_url}/{bot}/parse/{llm_type}",
            json=payload
        )

        if response.status_code != 200:
            raise Exception(response.text)

        response = response.json()

        if not response.get("success"):
            raise Exception(response)

        result = response.get("data")

        return {
            "full_text": result.get("full_text"),
            "extracted_data": result.get("extracted_data")
        }


    @staticmethod
    def process_instruction(data_list, prompt, operation_type, model_id, llm_type: str = "openrouter",
                            bot: str = None, user: str = None):
        import requests
        from kairon.shared.admin.data_objects import LLMSecret

        doc = LLMSecret.objects(llm_type="openrouter").first()
        api_key = Utility.decrypt_message(doc.api_key)

        if operation_type == "embedding":

            llm_server_url = Utility.environment['llm']['url']
            payload = {
                "text": data_list,
                "user": user,
                "kwargs": {
                    "model": model_id,
                    "api_key": api_key
                }
            }

            response = requests.request(method="POST",
                                        url=f"{llm_server_url}/{bot}/aembedding/{llm_type}",
                                        json=payload)
            response.raise_for_status()
            response = response.json()
            logger.info(response)

            return {
                "embeddings": response
            }

        else:
            text_input = data_list[0]
            final_prompt = prompt.format(document=text_input)
            payload = {
                "user": user,
                "hyperparameters": {"temperature": 0, "model": model_id},
                "messages": [{"role": "user", "content": final_prompt}]
            }
            llm_server_url = Utility.environment['llm']['url']
            response = requests.request(method="POST",
                                        url=f"{llm_server_url}/{bot}/completion/{llm_type}",
                                        json=payload)

            response.raise_for_status()
            response = response.json()
            extracted_data = response['formatted_response']

            logger.info(response)
            logger.info(extracted_data)

            return extracted_data


    @staticmethod
    def create_vector_collection(collection_name, model_id: str, user: str, emb_size: int = 3072,
                                 overwrite: bool = False, metadata: list = None, bot: str = None):
        from kairon.shared.cognition.data_objects import CognitionSchema, ColumnMetadata, SchemaMetadata
        from qdrant_client.models import VectorParams, Distance
        from qdrant_client import QdrantClient

        db_url = Utility.environment['vector']['db']
        try:
            bot_settings = BotSettings.objects(bot=bot, status=True).get()
        except DoesNotExist:
            raise Exception("Bot settings not found")

        if not bot_settings.llm_settings.enable_faq:
            raise Exception(
                "Please enable FAQ/LLM before creating vector collection"
            )
        knowledge_vault_name = collection_name
        collection_name = f"{bot}_{collection_name}_faq_embd"
        schema = {
            "metadata": metadata,
            "collection_name": knowledge_vault_name
        }

        client = QdrantClient(url=db_url)

        collections = client.get_collections().collections
        exists = any(c.name == collection_name for c in collections)
        embed_config = {
            "size": emb_size,
            "distance": Distance.COSINE
        }
        vector_config = VectorParams(**embed_config)
        if exists and overwrite:
            client.delete_collection(collection_name=collection_name)
            exist = CognitionSchema.objects(bot=bot, collection_name=knowledge_vault_name).first()
            if exist:
                exist.delete()

        if not exists or overwrite:
            client.create_collection(
                collection_name=collection_name,
                vectors_config=vector_config
            )
            metadata_obj = CognitionSchema(bot=bot, user=user)
            metadata_obj.metadata = [ColumnMetadata(**meta) for meta in schema.get("metadata") or []]
            metadata_obj.collection_name = schema.get("collection_name")
            metadata_obj.schema_metadata = SchemaMetadata(
                training_needed = False,
                provider = "openrouter",
                model_id = model_id,
                size = emb_size,
                IsVisible = True
            )
            metadata_obj.save()
        else:
            return {
                "message": "collection already exists"
            }

        return {
            "message": "collection created successfully"
        }

1	import pickle	1✔
2	from calendar import timegm	1✔
3	from datetime import datetime, date	1✔
4	from email.mime.multipart import MIMEMultipart	1✔
5	from email.mime.text import MIMEText	1✔
6	from smtplib import SMTP	1✔
7	from typing import Text, Dict, Callable, List	1✔
8	import base64	1✔
9	from apscheduler.triggers.date import DateTrigger	1✔
10	from apscheduler.util import obj_to_ref, astimezone	1✔
11	from mongoengine import DoesNotExist	1✔
12	from pymongo import MongoClient	1✔
13	from tzlocal import get_localzone	1✔
14	from uuid6 import uuid7	1✔
15	from loguru import logger	1✔
16	from kairon import Utility	1✔
17	from kairon.events.executors.factory import ExecutorFactory	1✔
18	from kairon.exceptions import AppException	1✔
19	from kairon.shared.actions.data_objects import EmailActionConfig	1✔
20	from kairon.shared.actions.utils import ActionUtility	1✔
21	from bson import Binary	1✔
22	from types import ModuleType	1✔
23	from requests import Response	1✔
24	from cryptography.hazmat.primitives import hashes	1✔
25	from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes	1✔
26	from cryptography.hazmat.primitives.asymmetric import padding as asym_padding	1✔
27	from cryptography.hazmat.primitives.serialization import load_pem_private_key	1✔
28	from kairon.shared.callback.data_objects import CallbackConfig, CallbackData	1✔
29	import json as jsond	1✔
30
31	from mongoengine.errors import DoesNotExist	1✔
32	from kairon.shared.data.data_objects import BotSettings	1✔
33	from kairon.shared.chat.user_media import UserMedia	1✔
34	from kairon.shared.cognition.data_objects import AnalyticsCollectionData	1✔
35
36
37	class CallbackScriptUtility:	1✔
38
39	@staticmethod	1✔
40	def generate_id():	1✔
41	return uuid7().hex	1✔
42
43
44	@staticmethod	1✔
45	def datetime_to_utc_timestamp(timeval):	1✔
46	"""
47	Converts a datetime instance to a timestamp.
48
49	:type timeval: datetime
50	:rtype: float
51
52	"""
53	if timeval is not None:	1✔
54	return timegm(timeval.utctimetuple()) + timeval.microsecond / 1000000	1✔
55
56
57	@staticmethod	1✔
58	def add_schedule_job(schedule_action: Text, date_time: datetime, data: Dict, timezone: Text, _id: Text = None,	1✔
59	bot: Text = None, kwargs=None):
60	if not bot:	1✔
61	raise AppException("Missing bot id")	1✔
62
63	if not _id:	1✔
64	_id = uuid7().hex	1✔
65
66	if not data:	1✔
67	data = {}	1✔
68
69	data['bot'] = bot	1✔
70	data['event'] = _id	1✔
71
72	callback_config = CallbackConfig.get_entry(bot=bot, name=schedule_action)	1✔
73
74	script = callback_config.get('pyscript_code')	1✔
75
76	func = obj_to_ref(ExecutorFactory.get_executor_for_data(data).execute_task)	1✔
77
78	schedule_data = {	1✔
79	'source_code': script,
80	'predefined_objects': data
81	}
82
83	args = (func, "scheduler_evaluator", schedule_data,)	1✔
84	kwargs = {'task_type': "Callback"} if kwargs is None else {**kwargs, 'task_type': "Callback"}	1✔
85	trigger = DateTrigger(run_date=date_time, timezone=timezone)	1✔
86
87	next_run_time = trigger.get_next_fire_time(None, datetime.now(astimezone(timezone) or get_localzone()))	1✔
88
89	job_kwargs = {	1✔
90	'version': 1,
91	'trigger': trigger,
92	'executor': "default",
93	'func': func,
94	'args': tuple(args) if args is not None else (),
95	'kwargs': kwargs,
96	'id': _id,
97	'name': "execute_task",
98	'misfire_grace_time': 7200,
99	'coalesce': True,
100	'next_run_time': next_run_time,
101	'max_instances': 1,
102	}
103
104	logger.info(job_kwargs)	1✔
105
106	client = MongoClient(Utility.environment['database']['url'])	1✔
107	events_db_name = Utility.environment["events"]["queue"]["name"]	1✔
108	events_db = client.get_database(events_db_name)	1✔
109	scheduler_collection = Utility.environment["events"]["scheduler"]["collection"]	1✔
110	job_store_name = events_db.get_collection(scheduler_collection)	1✔
111	event_server = Utility.environment['events']['server_url']	1✔
112
113	job_store_name.insert_one({	1✔
114	'_id': _id,
115	'next_run_time': CallbackScriptUtility.datetime_to_utc_timestamp(next_run_time),
116	'job_state': Binary(pickle.dumps(job_kwargs, pickle.HIGHEST_PROTOCOL))
117	})
118
119	http_response = ActionUtility.execute_http_request(	1✔
120	f"{event_server}/api/events/dispatch/{_id}",
121	"GET")
122
123	if not http_response.get("success"):	1✔
124	raise AppException(http_response)	1✔
125	else:
126	logger.info(http_response)	1✔
127
128	@staticmethod	1✔
129	def trigger_email(	1✔
130	email: List[str],
131	subject: str,
132	body: str,
133	smtp_url: str,
134	smtp_port: int,
135	sender_email: str,
136	smtp_password: str,
137	smtp_userid: str = None,
138	tls: bool = False,
139	content_type="html",
140	):
141	"""
142	This is a sync email trigger.
143	Sends an email to the mail id of the recipient
144
145	:param smtp_userid:
146	:param sender_email:
147	:param tls:
148	:param smtp_port:
149	:param smtp_url:
150	:param email: the mail id of the recipient
151	:param smtp_password:
152	:param subject: the subject of the mail
153	:param body: the body of the mail
154	:param content_type: "plain" or "html" content
155	:return: None
156	"""
157	smtp = None	1✔
158	try:	1✔
159	smtp = SMTP(smtp_url, port=smtp_port, timeout=10)	1✔
160	smtp.connect(smtp_url, smtp_port)	1✔
161	if tls:	1✔
162	smtp.starttls()	1✔
163	smtp.login(smtp_userid if smtp_userid else sender_email, smtp_password)	1✔
164
165	from_addr = sender_email	1✔
166	mime_body = MIMEText(body, content_type)	1✔
167	msg = MIMEMultipart("alternative")	1✔
168	msg["Subject"] = subject	1✔
169	msg["From"] = from_addr	1✔
170	msg["To"] = ",".join(email)	1✔
171	msg.attach(mime_body)	1✔
172
173	smtp.sendmail(from_addr, email, msg.as_string())	1✔
174
175	except Exception as e:	×
UNCOV 176	print(f"Failed to send email: {e}")	×
UNCOV 177	raise	×
178	finally:
179	if smtp:	1✔
180	try:	1✔
181	smtp.quit()	1✔
UNCOV 182	except Exception as quit_error:	×
UNCOV 183	print(f"Failed to quit SMTP connection cleanly: {quit_error}")	×
184
185
186	@staticmethod	1✔
187	def send_email(email_action: Text,	1✔
188	from_email: Text,
189	to_email: Text,
190	subject: Text,
191	body: Text,
192	bot: Text):
193	if not bot:	1✔
194	raise AppException("Missing bot id")	1✔
195
196	email_action_config = EmailActionConfig.objects(bot=bot, action_name=email_action).first()	1✔
197	if not email_action_config:	1✔
198	raise AppException(f"Email action '{email_action}' not configured for bot {bot}")	1✔
199	action_config = email_action_config.to_mongo().to_dict()	1✔
200
201	smtp_password = action_config.get('smtp_password').get("value")	1✔
202	smtp_userid = action_config.get('smtp_userid').get("value")	1✔
203
204	CallbackScriptUtility.trigger_email(	1✔
205	email=[to_email],
206	subject=subject,
207	body=body,
208	smtp_url=action_config['smtp_url'],
209	smtp_port=action_config['smtp_port'],
210	sender_email=from_email,
211	smtp_password=smtp_password,
212	smtp_userid=smtp_userid,
213	tls=action_config['tls']
214	)
215
216	@staticmethod	1✔
217	def perform_cleanup(local_vars: Dict):	1✔
218	logger.info(f"local_vars: {local_vars}")	×
219	filtered_locals = {}	×
220	if local_vars:	×
221	for key, value in local_vars.items():	×
222	if not isinstance(value, Callable) and not isinstance(value, ModuleType):	×
223	if isinstance(value, datetime):	×
224	value = value.strftime("%m/%d/%Y, %H:%M:%S")	×
225	elif isinstance(value, date):	×
226	value = value.strftime("%Y-%m-%d")	×
227	elif isinstance(value, Response):	×
228	value = value.text	×
229	filtered_locals[key] = value	×
UNCOV 230	logger.info(f"filtered_vars: {filtered_locals}")	×
UNCOV 231	return filtered_locals	×
232
233
234	@staticmethod	1✔
235	def decrypt_request(request_body, private_key_pem):	1✔
236	try:	1✔
237	encrypted_data_b64 = request_body.get("encrypted_flow_data")	1✔
238	encrypted_aes_key_b64 = request_body.get("encrypted_aes_key")	1✔
239	iv_b64 = request_body.get("initial_vector")	1✔
240
241	if not (encrypted_data_b64 and encrypted_aes_key_b64 and iv_b64):	1✔
242	raise ValueError("Missing required encrypted data fields")	1✔
243
244	# Decode base64 inputs
245	encrypted_aes_key = base64.b64decode(encrypted_aes_key_b64)	1✔
246	encrypted_data = base64.b64decode(encrypted_data_b64)	1✔
247	iv = base64.b64decode(iv_b64)[:16] # Ensure IV is exactly 16 bytes	1✔
248
249	private_key = load_pem_private_key(private_key_pem.encode(), password=None)	1✔
250
251	# Decrypt AES key using RSA and OAEP padding
252	aes_key = private_key.decrypt(	1✔
253	encrypted_aes_key,
254	asym_padding.OAEP(
255	mgf=asym_padding.MGF1(algorithm=hashes.SHA256()),
256	algorithm=hashes.SHA256(),
257	label=None,
258	),
259	)
260
261	if len(aes_key) not in (16, 24, 32):	1✔
UNCOV 262	raise ValueError(f"Invalid AES key size: {len(aes_key)} bytes")	×
263
264	# Extract GCM tag (last 16 bytes)
265	encrypted_body = encrypted_data[:-16]	1✔
266	tag = encrypted_data[-16:]	1✔
267
268	# Decrypt AES-GCM
269	cipher = Cipher(algorithms.AES(aes_key), modes.GCM(iv, tag))	1✔
270	decryptor = cipher.decryptor()	1✔
271	decrypted_bytes = decryptor.update(encrypted_body) + decryptor.finalize()	1✔
272	decrypted_data = jsond.loads(decrypted_bytes.decode("utf-8"))	1✔
273
274	response_dict = {	1✔
275	"decryptedBody": decrypted_data,
276	"aesKeyBuffer": aes_key,
277	"initialVectorBuffer": iv,
278	}
279
280	return response_dict	1✔
281
282	except Exception as e:	1✔
283	raise Exception(f"decryption failed-{str(e)}")	1✔
284
285
286	@staticmethod	1✔
287	def encrypt_response(response_body, aes_key_buffer, initial_vector_buffer):	1✔
288	try:	1✔
289	if aes_key_buffer is None:	1✔
290	raise ValueError("AES key cannot be None")	1✔
291
292	if initial_vector_buffer is None:	1✔
293	raise ValueError("Initialization vector (IV) cannot be None")	1✔
294
295	# Flip the IV
296	flipped_iv = bytes(byte ^ 0xFF for byte in initial_vector_buffer)	1✔
297
298	# Encrypt using AES-GCM
299	encryptor = Cipher(algorithms.AES(aes_key_buffer), modes.GCM(flipped_iv)).encryptor()	1✔
300	encrypted_bytes = encryptor.update(jsond.dumps(response_body).encode("utf-8")) + encryptor.finalize()	1✔
301	encrypted_data_with_tag = encrypted_bytes + encryptor.tag	1✔
302
303	# Encode result as base64
304	encoded_data = base64.b64encode(encrypted_data_with_tag).decode("utf-8")	1✔
305	return encoded_data	1✔
306	except Exception as e:	1✔
307	raise Exception(f"encryption failed-{str(e)}")	1✔
308
309
310	@staticmethod	1✔
311	def create_callback(callback_name: str, metadata: dict, bot: str, sender_id: str, channel: str,	1✔
312	name: str = None):
313	if not callback_name:	1✔
314	raise AppException("'callback name' must be provided and cannot be empty")	1✔
315
316	if not name:	1✔
317	name=callback_name	1✔
318	callback_url, identifier, standalone = CallbackData.create_entry(	1✔
319	name=name,
320	callback_config_name=callback_name,
321	bot=bot,
322	sender_id=sender_id,
323	channel=channel,
324	metadata=metadata,
325	)
326	if standalone:	1✔
327	return identifier	1✔
328	else:
329	return callback_url	1✔
330
331	@staticmethod	1✔
332	def save_as_pdf(text: str, bot: str, sender_id:str):	1✔
333	try:	1✔
334	_, media_id = UserMedia.save_markdown_as_pdf(	1✔
335	bot=bot,
336	sender_id=sender_id,
337	text=text,
338	filepath="report.pdf"
339	)
340	return media_id	1✔
341	except Exception as e:	1✔
342	raise Exception(f"encryption failed-{str(e)}")	1✔
343
344	@staticmethod	1✔
345	def get_data_analytics(collection_name: str, data_filters:dict, bot: str):	1✔
346	if not bot:	1✔
UNCOV 347	raise Exception("Missing bot id")	×
348
349	normalized_name = collection_name.lower()	1✔
350	match_filter = {	1✔
351	"bot": bot,
352	"collection_name": normalized_name
353	}
354
355	# Merge dictionary filter
356	if data_filters:	1✔
357	match_filter.update(data_filters)	1✔
358	cursor = AnalyticsCollectionData._get_collection().aggregate([	1✔
359	{"$match": match_filter},
360	{"$project": {
361	"_id": {"$toString": "$_id"},
362	"collection_name": 1,
363	"received_at": {
364	"$dateToString": {
365	"format": "%Y-%m-%dT%H:%M:%S.%LZ",
366	"date": "$received_at"
367	}
368	},
369	"source": 1,
370	"is_data_processed": 1,
371	"data": 1
372	}}
373	])
374
375	return {"data": list(cursor)}	1✔
376
377	@staticmethod	1✔
378	def add_data_analytics(user: str, payload, bot: str = None):	1✔
379	if not bot:	1✔
UNCOV 380	raise Exception("Missing bot id")	×
381
382	if not isinstance(payload, list):	1✔
383	raise Exception("Payload must be a list of dicts")	1✔
384
385	docs = []	1✔
386
387	for item in payload:	1✔
388	docs.append({	1✔
389	"bot": bot,
390	"user": user,
391	"collection_name": item.get("collection_name", "").lower().strip(),
392	"data": item.get("data"),
393	"source": item.get("source", ""),
394	"received_at": item.get("received_at") or datetime.utcnow(),
395	"is_data_processed": False
396	})
397
398	AnalyticsCollectionData._get_collection().insert_many(docs)	1✔
399
400	return {	1✔
401	"message": "Records saved!"
402	}
403
404	@staticmethod	1✔
405	def mark_as_processed(user: str, collection_name: str, bot: str = None):	1✔
406	if not bot:	1✔
UNCOV 407	raise Exception("Missing bot id")	×
408
409	collection_name = collection_name.lower().strip()	1✔
410
411	result = AnalyticsCollectionData.objects(	1✔
412	bot=bot,
413	collection_name=collection_name
414	).update(
415	set__user=user,
416	set__is_data_processed=True,
417	multi=True
418	)
419
420	if result == 0:	1✔
421	raise AppException("No records found for given bot and collection_name")	1✔
422
423	return {	1✔
424	"message": "Records updated!"
425	}
426
427	@staticmethod	1✔
428	def delete_data_analytics(collection_id: str, bot: Text = None):	1✔
429	if not bot:	1✔
UNCOV 430	raise Exception("Missing bot id")	×
431
432	try:	1✔
433	AnalyticsCollectionData.objects(bot=bot, id=collection_id).delete()	1✔
434	except DoesNotExist:	1✔
435	raise AppException("Analytics Collection Data does not exists!")	1✔
436
437	return {	1✔
438	"message": f"Analytics Collection with ID {collection_id} has been successfully deleted.",
439	"data": {"_id": collection_id}
440	}
441
442	@staticmethod	1✔
443	def update_data_analytics(collection_id: str, user: str, payload: dict, bot: str = None):	1✔
444	if not bot:	1✔
UNCOV 445	raise Exception("Missing bot id")	×
446
447	collection_name = payload.get("collection_name")	1✔
448	data = payload.get("data", {})	1✔
449	received_at=payload.get("received_at", datetime.utcnow())	1✔
450	source=payload.get("source", "")	1✔
451	is_data_processed=payload.get("is_data_processed", False)	1✔
452
453	try:	1✔
454	collection_obj = AnalyticsCollectionData.objects(bot=bot, id=collection_id, collection_name=collection_name).get()	1✔
455	filtered_data = {	1✔
456	k: v for k, v in data.items()
457	}
458
459	collection_obj.data.update(filtered_data)	1✔
460	collection_obj.collection_name = collection_name	1✔
461	collection_obj.user = user	1✔
462	collection_obj.received_at = received_at	1✔
463	collection_obj.source=source	1✔
464	collection_obj.is_data_processed=is_data_processed	1✔
465	collection_obj.save()	1✔
466	except DoesNotExist:	1✔
467	raise AppException("Analytics Collection Data with given id and collection_name not found!")	1✔
468
469	return {	1✔
470	"message": "Record updated!",
471	"data": {"_id": collection_id}
472	}
473
474	@staticmethod	1✔
475	def extract_data(input_source: str,	1✔
476	prompt: str = None,
477	result_type: str="markdown",
478	llm_type: str = "openrouter",
479	high_res_ocr: bool = False,
480	language: str = "en",
481	bot: str = None,
482	user: str = None):
483
484	import requests	1✔
485
486	llm_server_url = Utility.environment['llm']['url']	1✔
487
488	payload = {	1✔
489	"input_source": input_source,
490	"llama_parser_api_key": Utility.environment['llama_parse']['key'],
491	"result_type": result_type,
492	"high_res_ocr": high_res_ocr,
493	"language": language,
494	"parsing_instruction": prompt,
495	"user": user,
496	"llm_type": llm_type
497	}
498
499	response = requests.post(	1✔
500	f"{llm_server_url}/{bot}/parse/{llm_type}",
501	json=payload
502	)
503
504	if response.status_code != 200:	1✔
505	raise Exception(response.text)	1✔
506
507	response = response.json()	1✔
508
509	if not response.get("success"):	1✔
510	raise Exception(response)	1✔
511
512	result = response.get("data")	1✔
513
514	return {	1✔
515	"full_text": result.get("full_text"),
516	"extracted_data": result.get("extracted_data")
517	}
518
519
520	@staticmethod	1✔
521	def process_instruction(data_list, prompt, operation_type, model_id, llm_type: str = "openrouter",	1✔
522	bot: str = None, user: str = None):
523	import requests	1✔
524	from kairon.shared.admin.data_objects import LLMSecret	1✔
525
526	doc = LLMSecret.objects(llm_type="openrouter").first()	1✔
527	api_key = Utility.decrypt_message(doc.api_key)	1✔
528
529	if operation_type == "embedding":	1✔
530
531	llm_server_url = Utility.environment['llm']['url']	1✔
532	payload = {	1✔
533	"text": data_list,
534	"user": user,
535	"kwargs": {
536	"model": model_id,
537	"api_key": api_key
538	}
539	}
540
541	response = requests.request(method="POST",	1✔
542	url=f"{llm_server_url}/{bot}/aembedding/{llm_type}",
543	json=payload)
544	response.raise_for_status()	1✔
545	response = response.json()	1✔
546	logger.info(response)	1✔
547
548	return {	1✔
549	"embeddings": response
550	}
551
552	else:
553	text_input = data_list[0]	1✔
554	final_prompt = prompt.format(document=text_input)	1✔
555	payload = {	1✔
556	"user": user,
557	"hyperparameters": {"temperature": 0, "model": model_id},
558	"messages": [{"role": "user", "content": final_prompt}]
559	}
560	llm_server_url = Utility.environment['llm']['url']	1✔
561	response = requests.request(method="POST",	1✔
562	url=f"{llm_server_url}/{bot}/completion/{llm_type}",
563	json=payload)
564
565	response.raise_for_status()	1✔
566	response = response.json()	1✔
567	extracted_data = response['formatted_response']	1✔
568
569	logger.info(response)	1✔
570	logger.info(extracted_data)	1✔
571
572	return extracted_data	1✔
573
574
575	@staticmethod	1✔
576	def create_vector_collection(collection_name, model_id: str, user: str, emb_size: int = 3072,	1✔
577	overwrite: bool = False, metadata: list = None, bot: str = None):
578	from kairon.shared.cognition.data_objects import CognitionSchema, ColumnMetadata, SchemaMetadata	1✔
579	from qdrant_client.models import VectorParams, Distance	1✔
580	from qdrant_client import QdrantClient	1✔
581
582	db_url = Utility.environment['vector']['db']	1✔
583	try:	1✔
584	bot_settings = BotSettings.objects(bot=bot, status=True).get()	1✔
NEW 585	except DoesNotExist:	×
NEW UNCOV 586	raise Exception("Bot settings not found")	×
587
588	if not bot_settings.llm_settings.enable_faq:	1✔
589	raise Exception(	1✔
590	"Please enable FAQ/LLM before creating vector collection"
591	)
592	knowledge_vault_name = collection_name	1✔
593	collection_name = f"{bot}_{collection_name}_faq_embd"	1✔
594	schema = {	1✔
595	"metadata": metadata,
596	"collection_name": knowledge_vault_name
597	}
598
599	client = QdrantClient(url=db_url)	1✔
600
601	collections = client.get_collections().collections	1✔
602	exists = any(c.name == collection_name for c in collections)	1✔
603	embed_config = {	1✔
604	"size": emb_size,
605	"distance": Distance.COSINE
606	}
607	vector_config = VectorParams(**embed_config)	1✔
608	if exists and overwrite:	1✔
609	client.delete_collection(collection_name=collection_name)	1✔
610	exist = CognitionSchema.objects(bot=bot, collection_name=knowledge_vault_name).first()	1✔
611	if exist:	1✔
612	exist.delete()	1✔
613
614	if not exists or overwrite:	1✔
615	client.create_collection(	1✔
616	collection_name=collection_name,
617	vectors_config=vector_config
618	)
619	metadata_obj = CognitionSchema(bot=bot, user=user)	1✔
620	metadata_obj.metadata = [ColumnMetadata(**meta) for meta in schema.get("metadata") or []]	1✔
621	metadata_obj.collection_name = schema.get("collection_name")	1✔
622	metadata_obj.schema_metadata = SchemaMetadata(	1✔
623	training_needed = False,
624	provider = "openrouter",
625	model_id = model_id,
626	size = emb_size,
627	IsVisible = True
628	)
629	metadata_obj.save()	1✔
630	else:
631	return {	1✔
632	"message": "collection already exists"
633	}
634
635	return {	1✔
636	"message": "collection created successfully"
637	}

digiteinfotech / kairon / 25849035942

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous