25856517505

Committed 14 May 2026 11:02AM UTC coverage: 91.218% (+0.002%) from 91.216%

Build # 25856517505

Build Type

Pull #2373

github

Committed by

web-flow

Commit Message

Merge 8a9ed425f into ec639b034

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

Coverage Stats

5 of 5 new or added lines in 1 file covered. (100.0%)

9 existing lines in 1 file now uncovered.

30765 of 33727 relevant lines covered (91.22%)

0.91 hits per line

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

92.01

/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 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 as err:
            raise Exception("Bot settings not found") from err

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

digiteinfotech / kairon / 25856517505

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