XQLDSHelper

Source

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401
import re
import enum
import pytz
import gzip
import math
import base64
import hashlib
import dateparser
import itertools
import colorsys
import traceback
import urllib.parse
from collections import defaultdict
from collections.abc import Iterable, Hashable


DEFAULT_POLLING_INTERVAL = 10  # in seconds
DEFAULT_RETRY_INTERVAL = 10  # in seconds
DEFAULT_RETRY_MAX = 10
DEFAULT_QUERY_TIMEOUT_DURATION = 60  # in seconds


def to_float(val: Any) -> float | int:
    """Ensure the value is of type number (float or int).

    :param val: The value.
    :return: A float or int converted from `val`.
    """
    if val is None:
        return 0
    try:
        val = float(val)
        return int(val) if val.is_integer() else val
    except (ValueError, TypeError):
        return 0


def get_target_type():
    if is_platform():
        return "issues"
    elif is_xsiam():
        return "alerts"
    else:
        return "incidents"


def to_str(val: Any) -> str:
    """Ensure the value is of type string.

    :param val: The value.
    :return: A str converted from `val`.
    """
    return val if isinstance(val, str) else json.dumps(val)


class CacheType(enum.StrEnum):
    NONE = "none"
    RECORDSET = "recordset"
    ENTRY = "entry"


class DefaultEntryScope(enum.StrEnum):
    NO_RECORDSET = "no_recordset"
    QUERY_SKIPPED = "query_skipped"


class ContextData:
    def __init__(
        self,
        context: dict[str, Any] | None = None,
        incident: dict[str, Any] | None = None,
        alert: dict[str, Any] | None = None,
        issue: dict[str, Any] | None = None,
        value: dict[str, Any] | None = None,
    ) -> None:
        self.__context: dict[str, Any] = context or {}
        self.__value: dict[str, Any] = value or {}
        self.__specials: dict[str, Any] = {
            "issue": issue or {},
            "alert": alert or {},
            "incident": incident or {},
            "lists": None,
        }

    def get(self, key: str | None = None) -> Any:
        """Get the context value

        :param key: The dt expressions (string within ${}).
        :return: The value.
        """
        if not key:
            return None

        if key != "." and not key.startswith(".=") and key.startswith("."):
            dx = self.__value
            key = key[1:]
        else:
            for prefix in self.__specials:
                k = key[len(prefix) :]
                if key.startswith(prefix) and k[:1] in ("", ".", "(", "="):
                    if prefix == "lists":
                        if list_name := re.split("[.(=]", k[1:], maxsplit=1)[0]:
                            dx = {prefix: {list_name: execute_command("getList", {"listName": list_name})}}
                            break
                    else:
                        dx = self.__specials
                        break
            else:
                dx = self.__context

        return demisto.dt(dx, key)

    def inherit(
        self,
        value: dict[str, Any] | None = None,
    ) -> "ContextData":
        """Create a ContextData with the new value

        :param value: The new value.
        :return: ContextData created.
        """
        return ContextData(
            context=self.__context,
            incident=self.__specials.get("incident"),
            alert=self.__specials.get("alert"),
            issue=self.__specials.get("issue"),
            value=value,
        )


class Formatter:
    def __init__(self, variable_substitution: tuple[str, str], keep_symbol_to_null: bool) -> None:
        self.__keep_symbol_to_null = keep_symbol_to_null
        self.__var_opening, self.__var_closing = variable_substitution
        if not self.__var_opening:
            raise DemistoException("opening marker is required.")

    @staticmethod
    def __is_closure(source: str, ci: int, closure_marker: str) -> bool:
        if closure_marker:
            return source[ci : ci + len(closure_marker)] == closure_marker
        else:
            c = source[ci]
            if c.isspace():
                return True
            elif c.isascii():
                return c != "_" and not c.isalnum()
            else:
                return False

    @staticmethod
    def __extract_dt(
        dtstr: str,
        dx: ContextData | None,
    ) -> Any:
        """Extract dt expression

        :param dtstr: The dt expressions (string within ${}).
        :param dx: The context instance.
        :return: The value extracted.
        """
        try:
            return dx.get(dtstr) if dx else dtstr
        except Exception:
            return None

    def __extract(
        self,
        source: str,
        dx: ContextData | None,
        si: int,
        markers: tuple[str, str] | None,
    ) -> tuple[Any, int | None]:
        """Extract a template text, or an enclosed value within starting and ending marks

        :param source: The template text, or the enclosed value starts with the next charactor of a start marker
        :param dx: The context data
        :param si: The index of `source` to start extracting
        :param markers: The opening and closing markers to find an end position for parsing an enclosed value.
                        It must be None when the template text is given to `source`.
        :return: The extracted value and index of `source` when parsing ended.
                 The index is the next after the end marker when extracting the enclosed value.
        """
        out = None
        ci = si
        while ci < len(source):
            if markers is not None and Formatter.__is_closure(source, ci, markers[1]):
                key = source[si:ci] if out is None else str(out) + source[si:ci]
                if markers == (self.__var_opening, self.__var_closing):
                    xval = self.__extract_dt(key, dx)
                    if xval is None and self.__keep_symbol_to_null:
                        xval = markers[0] + key + markers[1]
                    else:
                        xval = self.build(xval, dx)
                else:
                    xval = markers[0] + key + markers[1]
                return xval, ci + len(markers[1])
            elif source[ci : ci + len(self.__var_opening)] == self.__var_opening:
                xval, ei = self.__extract(
                    source=source,
                    dx=dx,
                    si=ci + len(self.__var_opening),
                    markers=(self.__var_opening, self.__var_closing),
                )
                if si != ci:
                    out = source[si:ci] if out is None else str(out) + source[si:ci]

                if ei is None:
                    xval = self.__var_opening
                    ei = ci + len(self.__var_opening)

                if out is None:
                    out = xval
                elif xval is not None:
                    out = str(out) + str(xval)
                si = ci = ei
            elif markers is None:
                ci += 1
            elif endc := {"(": ")", "{": "}", "[": "]", '"': '"', "'": "'", "`": "`"}.get(source[ci]):
                xval, ei = self.__extract(
                    source=source,
                    dx=dx,
                    si=ci + 1,
                    markers=(source[ci], endc),
                )
                if si != ci:
                    out = source[si:ci] if out is None else str(out) + source[si:ci]

                if ei is None:
                    xval = source[ci]
                    ei = ci + len(source[ci])

                if out is None:
                    out = xval
                elif xval is not None:
                    out = str(out) + str(xval)

                si = ci = ei
            elif source[ci] == "\\":
                ci += 2
            else:
                ci += 1

        if markers is not None:
            # unbalanced braces, brackets, quotes, etc.
            return None, None
        elif si >= len(source):
            return out, ci
        elif out is None:
            return source[si:], ci
        else:
            return str(out) + source[si:], ci

    def build(self, template: Any, context: ContextData | None) -> Any:
        """Format a text from a template including DT expressions

        :param template: The template.
        :param context: The context instance.
        :return: The text built from the template.
        """
        if isinstance(template, dict):
            return {self.build(k, context): self.build(v, context) for k, v in template.items()}
        elif isinstance(template, list):
            return [self.build(v, context) for v in template]
        elif isinstance(template, str):
            return self.__extract(source=template, dx=context, si=0, markers=None)[0] if template else ""
        else:
            return template


class SortableValue:
    """
    The custom value object class, which enables you to sort for any types of data even in different types.
    """

    def __init__(
        self,
        value: Any,
    ) -> None:
        self.__value = value

    def __lt__(self, other: Any) -> bool:
        def __lt(obj1: Any, obj2: Any) -> bool:
            if any(
                f(obj1) and f(obj2)
                for f in [
                    lambda x: isinstance(x, int | float),
                    lambda x: isinstance(x, bool),
                    lambda x: isinstance(x, str),
                ]
            ):
                return obj1 < obj2  # type: ignore[operator]
            elif obj1 is None or obj2 is None:
                return (obj2 is None) < (obj1 is None)
            else:

                def __get_order(
                    v: Any,
                ) -> int:
                    if isinstance(v, bool):
                        return 1
                    elif isinstance(v, int | float):
                        return 2
                    elif isinstance(v, str):
                        return 3
                    else:
                        return 4

                order1 = __get_order(obj1)
                order2 = __get_order(obj2)
                if n := (order1 > order2) - (order1 < order2):
                    return n < 0
                else:
                    return json.dumps(obj1) < json.dumps(obj2)

        if not isinstance(other, SortableValue):
            return False
        return __lt(self.__value, other.__value)


class QueryParams:
    def __init__(
        self,
        query_name: str,
        query_string: str,
        earliest_time: datetime,
        latest_time: datetime,
    ) -> None:
        if not query_string:
            raise DemistoException("Query string is required.")

        if earliest_time.tzinfo is None or latest_time.tzinfo is None:
            raise DemistoException("earliest_time and latest_time must be timezone aware.")

        if earliest_time > latest_time:
            raise DemistoException(f"latest_time ({latest_time}) must be equal to or later than earliest_time ({earliest_time}).")

        self.__query_name = query_name
        self.__query_string = query_string
        self.__query_string_normalized = "\n".join(x.strip() for x in query_string.splitlines()).strip()
        self.__earliest_time = earliest_time
        self.__latest_time = latest_time

    @property
    def query_name(
        self,
    ) -> str:
        return self.__query_name

    @property
    def normalized_query_string(
        self,
    ) -> str:
        return self.__query_string_normalized

    @property
    def query_string(
        self,
    ) -> str:
        return self.__query_string

    @property
    def earliest_time(
        self,
    ) -> datetime:
        return self.__earliest_time

    @property
    def latest_time(
        self,
    ) -> datetime:
        return self.__latest_time

    def get_earliest_time_iso(
        self,
        utc: bool = False,
    ) -> str:
        if utc:
            t = self.__earliest_time.astimezone(pytz.UTC)
            return t.isoformat(timespec="milliseconds").replace("+00:00", "Z")
        else:
            return self.__earliest_time.isoformat(timespec="milliseconds")

    def get_latest_time_iso(
        self,
        utc: bool = False,
    ) -> str:
        if utc:
            t = self.__latest_time.astimezone(pytz.UTC)
            return t.isoformat(timespec="milliseconds").replace("+00:00", "Z")
        else:
            return self.__latest_time.isoformat(timespec="milliseconds")

    def query_hash(
        self,
    ) -> str:
        # The input value doesn't include `query_name` for the hash.
        return hashlib.sha256(
            json.dumps(
                {
                    "query_string": self.normalized_query_string,
                    "earliest_time": self.get_earliest_time_iso(),
                    "latest_time": self.get_latest_time_iso(),
                }
            ).encode()
        ).hexdigest()


class Cache:
    CACHE_ROOT_KEY = "XQLDSHelperCache"

    @staticmethod
    def __compress(val: str) -> dict[Hashable, str]:
        return {"type": "gz+b85x", "data": base64.b85encode(gzip.compress(val.encode())).decode().replace("$", ":")}

    @staticmethod
    def build_query_params(
        query_params: QueryParams,
    ) -> dict[Hashable, Any]:
        return {
            "query_name": query_params.query_name,
            "query_string": Cache.__compress(query_params.query_string),
            "earliest_time": query_params.get_earliest_time_iso(),
            "latest_time": query_params.get_latest_time_iso(),
        }

    def __load_data(
        self,
        query_hash: str,
        cache_node: str,
    ) -> Any:
        cache = self.__repo.get(self.__key)
        if not isinstance(cache, dict):
            return None

        if cache.get("QueryHash") != query_hash:
            return None

        _data = demisto.get(cache, f"{cache_node}.data")
        _type = demisto.get(cache, f"{cache_node}.type")
        try:
            if _type == "gz+b85":
                return json.loads(gzip.decompress(base64.b85decode(_data.encode())).decode())
            elif _type == "gz+b85x":
                return json.loads(gzip.decompress(base64.b85decode(_data.replace(":", "$").encode())).decode())
            else:
                return None
        except Exception as e:
            demisto.debug(f"Failed to load cache data [{cache_node}] - {e}")
            return None

    def __save_data(
        self,
        query_params: QueryParams,
        cache_node: str,
        data: Any,
    ) -> None:
        cache = {
            "QueryParams": self.build_query_params(query_params),
            "QueryHash": query_params.query_hash(),
            cache_node: self.__compress(json.dumps(data)),
        }
        if incident_id := demisto.incident().get("id"):
            target = get_target_type()
            demisto.executeCommand(
                "executeCommandAt",
                {
                    "command": "Set",
                    target: incident_id,
                    "arguments": {
                        "key": f"{Cache.CACHE_ROOT_KEY}.{self.__key}",
                        "value": cache,
                        "append": "false",
                    },
                },
            )

    def __init__(
        self,
        name: str,
        repo: dict[Hashable, Any] | None,
    ) -> None:
        name = urllib.parse.quote(name).replace(".", "%2E")
        self.__key = name
        self.__repo = repo or {}

    def save_recordset(
        self,
        query_params: QueryParams,
        execution_id: str,
        recordset: list[dict[Hashable, Any]],
    ) -> None:
        self.__save_data(
            query_params=query_params,
            cache_node="CacheDataset",
            data={
                "execution_id": execution_id,
                "recordset": recordset,
            },
        )

    def save_entry(
        self,
        query_params: QueryParams,
        entry: dict[Hashable, Any],
    ) -> None:
        self.__save_data(
            query_params=query_params,
            cache_node="CacheEntry",
            data=entry,
        )

    def load_recordset(
        self,
        query_hash: str,
    ) -> tuple[str, list[dict[Hashable, Any]]] | None:
        data = self.__load_data(
            query_hash=query_hash,
            cache_node="CacheDataset",
        )
        if isinstance(data, list):
            return "", data
        elif isinstance(data, dict):
            execution_id = data.get("execution_id") or ""
            recordset = data.get("recordset") or []
            return execution_id, recordset
        else:
            return None

    def load_entry(
        self,
        query_hash: str,
    ) -> dict[Hashable, Any] | None:
        entry = self.__load_data(
            query_hash=query_hash,
            cache_node="CacheEntry",
        )
        return entry if isinstance(entry, dict) else None


class CoreLock:
    """
    This class provides locking to prevent the concurrent execution of multiple tasks.
    """

    def __init__(
        self,
        module: str,
        name: str | None,
        info: str | None,
        timeout: int | None,
        using: str | None,
    ) -> None:
        """Initializes a lock instance.

        :param module: The name for the locking functions must be either core-lock or demisto-lock.
        :param name: The name to identify the lock.
        :param info: Additional information to provide for the lock.
        :param timeout: Timeout (seconds) for wait on lock to be freed.
        :param using: The name of an lock integration instance.
        """
        if module == "core-lock":
            self.__func_lock = "core-lock-get"
            self.__func_unlock = "core-lock-release"
        elif module == "demisto-lock":
            self.__func_lock = "demisto-lock-get"
            self.__func_unlock = "demisto-lock-release"
        else:
            raise DemistoException(f"locking module must be either core-lock or demisto-lock - {module}")

        self.__name = name
        self.__info = info
        self.__timeout = timeout
        self.__using = using

    def lock(
        self,
    ) -> None:
        """Acquire a specific lock via *-lock-get."""
        args = assign_params(
            name=self.__name,
            info=self.__info,
            timeout=self.__timeout,
            using=self.__using,
        )
        args_str = ", ".join(f"{k}={v}" for k, v in args.items()) or "default"

        demisto.debug(f"Attempting to acquire lock: {self.__func_lock} - {args_str}")
        execute_command(self.__func_lock, args, extract_contents=False)
        demisto.debug(f"Lock successfully acquired: {self.__func_lock} - {args_str}")

    def unlock(
        self,
    ) -> None:
        """Release a specific lock via core-lock-release."""
        args = assign_params(
            name=self.__name,
            using=self.__using,
        )
        args_str = ", ".join(f"{k}={v}" for k, v in args.items()) or "default"

        execute_command(self.__func_unlock, args, extract_contents=False)
        demisto.debug(f"Lock released: {self.__func_unlock} - {args_str}")


class XQLQuery:
    """
    This class executes XQL queries.
    """

    @staticmethod
    def __peek_response(
        res: list[dict[str, Any]],
    ) -> dict[Hashable, Any] | None:
        for ent in res:
            ec = ent.get("EntryContext") or {}
            for k, v in ec.items():
                k, _, _ = k.partition("(")
                if k == "PaloAltoNetworksXQL.GenericQuery" and isinstance(v, dict):
                    return v
        return None

    @staticmethod
    def __get_response(
        res: list[dict[str, Any]],
    ) -> dict[Hashable, Any]:
        resp = XQLQuery.__peek_response(res)
        if resp is None:
            raise DemistoException(f"Unable to get query results - {res}")
        else:
            return resp

    @staticmethod
    def __get_error_message(
        res: list[dict[str, Any]],
    ) -> str | None:
        if XQLQuery.__peek_response(res) is not None:
            return None
        elif is_error(res):
            if message := get_error(res):
                return message
        else:
            for ent in res:
                if message := ent.get("HumanReadable"):
                    return message
        return f"Unable to get query results - {res}"

    def __init__(
        self,
        xql_query_instance: str | None = None,
        polling_interval: int = DEFAULT_POLLING_INTERVAL,
        retry_interval: int = DEFAULT_RETRY_INTERVAL,
        retry_max: int = DEFAULT_RETRY_MAX,
        query_timeout_duration: int = DEFAULT_QUERY_TIMEOUT_DURATION,
        core_lock: CoreLock | None = None,
    ) -> None:
        self.__xql_query_instance = xql_query_instance
        self.__polling_interval = polling_interval
        self.__retry_interval = retry_interval
        self.__retry_max = max(retry_max, 0)
        self.__query_timeout_duration = max(query_timeout_duration, 0)
        self.__core_lock = core_lock

    def query(
        self,
        query_params: QueryParams,
    ) -> tuple[str, list[dict[Hashable, Any]]]:
        """Execute an XQL query and get results

        :param query_params: The query parameters.
        :return: The execution ID ans list of fields retrieved.
        """
        if self.__core_lock:
            self.__core_lock.lock()

        try:
            # Start the query
            time_frame = f"between {query_params.get_earliest_time_iso()} and {query_params.get_latest_time_iso()}"
            demisto.debug(f"Run XQL: {query_params.query_name} {time_frame}: {query_params.normalized_query_string}")

            for retry_count in range(self.__retry_max + 1):
                res = demisto.executeCommand(
                    "xdr-xql-generic-query",
                    assign_params(
                        query_name=query_params.query_name,
                        query=query_params.normalized_query_string,
                        time_frame=time_frame,
                        parse_result_file_to_context="true",
                        using=self.__xql_query_instance,
                    ),
                )
                error_message = self.__get_error_message(res)
                if error_message is None:
                    break

                if retry_count >= self.__retry_max or all(
                    x not in error_message
                    for x in [
                        "reached max allowed amount of parallel running queries",
                        "maximum allowed number of parallel running queries has been reached",
                    ]
                ):
                    raise DemistoException(f"Failed to execute xdr-xql-generic-query. Error details:\n{error_message}")

                time.sleep(self.__retry_interval)

            # Poll and retrieve the record set
            response = self.__get_response(res)

            execution_id = response.get("execution_id")
            if not execution_id:
                raise DemistoException("No execution_id in the response.")

            timeout_time = None

            while True:
                status = response.get("status", "")
                if status == "SUCCESS":
                    return execution_id, (response.get("results") or [])
                elif status == "PENDING":
                    current_time = time.time()
                    if timeout_time is None:
                        timeout_time = current_time + self.__query_timeout_duration

                    remaining_time = timeout_time - current_time
                    if remaining_time <= 0:
                        raise DemistoException(f"Unable to get query results within {self.__query_timeout_duration} seconds.")

                    time.sleep(min(remaining_time, self.__polling_interval))

                    res = demisto.executeCommand(
                        "xdr-xql-get-query-results",
                        assign_params(
                            query_id=execution_id,
                            parse_result_file_to_context="true",
                            using=self.__xql_query_instance,
                        ),
                    )
                    if is_error(res):
                        error_message = get_error(res)
                        raise DemistoException(f"Failed to execute xdr-xql-get-query-results. Error details:\n{error_message}")

                    response = self.__get_response(res)
                else:
                    raise DemistoException(f"Failed to get query results - {response}")
        finally:
            if self.__core_lock:
                self.__core_lock.unlock()


class Query:
    """
    This class allows you to get the query results.
    """

    def __init__(
        self,
        query_params: QueryParams,
        xql_query: XQLQuery | None,
        cache: Cache | None,
    ) -> None:
        self.__query_params = query_params
        self.__xql_query = xql_query
        self.__cache = cache

    @property
    def query_params(
        self,
    ) -> QueryParams:
        return self.__query_params

    def available(
        self,
    ) -> bool:
        return self.__xql_query is not None

    def query(
        self,
    ) -> tuple[str, list[dict[Hashable, Any]]]:
        """Get the record set by running the query. It will return an empty list if the query is not available,

        :return: The quest execution ID and list of fields retrieved by the query.
        """
        if self.__cache and (ret := self.__cache.load_recordset(self.__query_params.query_hash())):
            return ret
        elif self.__xql_query:
            execution_id, recordset = self.__xql_query.query(self.__query_params)
            if self.__cache:
                self.__cache.save_recordset(self.__query_params, execution_id, recordset)
            return execution_id, recordset
        else:
            return "", []


class EntryBuilder:
    """
    This class helps to query XQL and build an entry data.
    """

    @staticmethod
    def __enum_fields_by_group(
        recordset: Iterable[dict[Hashable, Any]],
        group_by: str,
        sort_by: str | None,
        asc: bool,
    ) -> Iterable[tuple[Hashable, Iterable[dict[Hashable, Any]]]]:
        """Enumerate fields with a group value by group

        :param recordset: The list of fields.
        :param group_by: The name of the field to make groups.
        :param sort_by: The field name by which to sort records within each group.
        :param asc: Set to True to sort the recordset in ascent order, Set to False for descent order.
        :return: Each group value with fields.
        """
        groups = itertools.groupby(
            sorted(
                recordset,
                key=lambda v: SortableValue(v.get(group_by)),
                reverse=not asc,
            ),
            key=lambda v: v.get(group_by),
        )
        if not sort_by:
            return groups
        else:
            return sorted(
                (
                    (
                        k,
                        sorted(
                            records,
                            key=lambda v: SortableValue(v.get(sort_by)),
                            reverse=not asc,
                        ),
                    )
                    for k, records in groups
                ),
                key=lambda v: SortableValue(next(iter(v[1]), {}).get(sort_by)),  # type: ignore
                reverse=not asc,
            )

    @staticmethod
    def __sum_by(
        recordset: list[dict[Hashable, Any]],
        sum_field: str,
        group_by: str,
        order_asc: bool,
    ) -> dict[Hashable, float]:
        """Sum field values by a field

        :param recordset: The list of fields.
        :param sum_field: The field name of the value to be summed.
        :param group_by: The field name to group the fields.
        :param order_asc: Set to True for ascending order, and False for descending order.
        :return: Mapping of field name with the sum value in order by the sum.
        """
        d: dict[Hashable, float] = defaultdict(float)
        for fields in recordset:
            d[fields.get(group_by)] += to_float(fields.get(sum_field))
        return dict(sorted(d.items(), key=lambda x: x[1], reverse=not order_asc))

    @staticmethod
    def __make_color_palette(
        names: list[str],
        colors: dict[Hashable, str] | list[str] | str,
    ) -> dict[Hashable, str]:
        """Build a color table

        :param names: The list of names to be mapped to colors
        :param colors: The base color mapping, or colors for 'names' in order
        :return: The color mapping. (name and color)
        """
        color_order: list[str] = []
        if isinstance(colors, str):
            color_order = [colors] * len(names)
        elif isinstance(colors, list):
            color_order = colors

        color_map: dict[Hashable, str] = dict(zip(names, color_order + EntryBuilder.list_colors(len(names))))
        if isinstance(colors, dict):
            color_map.update(colors)
        return color_map

    @staticmethod
    def __build_singley_chart(
        chart_type: str,
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a single-Y chart entry

        :param chart_type: The chart type. (bar or pie)
        :param params: The template parameters for a single-Y chart.
        :param recordset: The list of fields used to create the single-Y chart.
        :return: A single-Y chart entry.
        """

        class Template:
            class Records:
                class Sort:
                    def __init__(
                        self,
                        sort: dict[Hashable, Any],
                        default_by: str,
                    ) -> None:
                        by = sort.get("by") or default_by
                        assert isinstance(by, str) or by is None, f"sort.by must be of type str or null - {type(by)}"
                        self.__by = by or default_by
                        self.__asc = EntryBuilder.to_sort_order(sort.get("order") or "asc")

                    @property
                    def by(
                        self,
                    ) -> str:
                        return self.__by

                    @property
                    def asc(
                        self,
                    ) -> bool:
                        return self.__asc

                def __init__(
                    self,
                    records: dict[Hashable, Any],
                ) -> None:
                    name_field = records.get("name-field")
                    assert isinstance(name_field, str), f"name-field must be of type str - {type(name_field)}"
                    self.__name_field = name_field

                    data_field = records.get("data-field")
                    assert isinstance(data_field, str), f"data-field must be of type str - {type(data_field)}"
                    self.__data_field = data_field

                    colors = records.get("colors")
                    if isinstance(colors, list):
                        for color in colors:
                            assert isinstance(color, str), f"color must be of type str - {type(color)}"
                    elif isinstance(colors, dict):
                        for color in colors.values():
                            assert isinstance(color, str), f"color must be of type str - {type(color)}"
                    elif colors is None:
                        colors = []
                    elif not isinstance(colors, str):
                        raise DemistoException(f"colors must be of type dict, list or null - {type(colors)}")
                    self.__colors = colors

                    sort = records.get("sort") or {}
                    assert isinstance(sort, dict), f"sort must be of type dict or null - {type(sort)}"
                    self.__sort = self.Sort(sort, default_by=self.data_field)

                @property
                def name_field(
                    self,
                ) -> str:
                    return self.__name_field

                @property
                def data_field(
                    self,
                ) -> str:
                    return self.__data_field

                @property
                def colors(
                    self,
                ) -> dict[Hashable, str] | list[str] | str:
                    return self.__colors

                @property
                def sort(
                    self,
                ) -> Sort:
                    return self.__sort

            class Field:
                def __init__(
                    self,
                    field: dict[Hashable, Any],
                    default_color: str,
                ) -> None:
                    assert isinstance(field, dict), f"field in .fields must be of type dict - {type(field)}"
                    color = field.get("color")
                    assert isinstance(color, str) or color is None, f"field.color must be of type str or null - {type(color)}"
                    self.__color = color or default_color

                    self.__label = field.get("label")
                    assert (
                        isinstance(self.__label, str) or self.__label is None
                    ), f"field.label must be of type str or null - {type(self.__label)}"

                @property
                def color(
                    self,
                ) -> str:
                    return self.__color

                @property
                def label(
                    self,
                ) -> str | None:
                    return self.__label

            def __init__(
                self,
                template: dict[Hashable, Any],
            ) -> None:
                self.__records: Records | None = None  # pylint: disable=undefined-variable
                self.__fields: dict[Hashable, Field] | None = None  # pylint: disable=undefined-variable

                group = template.get("group")
                if group == "records":
                    records = template.get(group)
                    assert isinstance(records, dict), f"records is required and must be of type dict - {type(records)}"
                    self.__records = self.Records(records)
                elif group == "fields":
                    fields = template.get(group)
                    assert isinstance(fields, dict), f"fields is required and must be of type dict - {type(fields)}"
                    self.__fields = {
                        name: self.Field(field, default_color=color)
                        for (name, field), color in zip(fields.items(), EntryBuilder.list_colors(len(fields)))
                    }
                else:
                    raise DemistoException(f"group must be 'records' or 'fields' - {group}")

            @property
            def records(
                self,
            ) -> Records | None:
                return self.__records

            @property
            def fields(
                self,
            ) -> dict[Hashable, Field] | None:
                return self.__fields

        template = Template(params)
        if records := template.records:
            sort: Template.Records.Sort = records.sort

            names = EntryBuilder.__sum_by(
                recordset=recordset,
                sum_field=records.data_field,
                group_by=records.name_field,
                order_asc=sort.asc,
            )
            # Create color mapping
            colors = EntryBuilder.__make_color_palette(
                names=[x for x in names if isinstance(x, str)],
                colors=records.colors,
            )
            # Build stats
            stats = [
                assign_params(
                    name=to_str(name),
                    data=[to_float(value)],
                    color=colors.get(name),
                )
                for fields in sorted(
                    recordset,
                    key=lambda v: to_float(v.get(sort.by)),
                    reverse=not sort.asc,
                )
                for name, value in [(fields.get(records.name_field, ""), fields.get(records.data_field))]
            ]
        elif fields := template.fields:
            # Build stats
            stats = [
                {
                    "name": to_str(group.label or field or ""),
                    "data": [sum(to_float(x.get(field)) for x in recordset)],
                    "color": group.color,
                }
                for field, group in fields.items()
            ]
        else:
            stats = []

        return {
            "Type": EntryType.WIDGET,
            "ContentsFormat": chart_type,
            "Contents": dict(
                assign_params(params=params.get("params")),
                stats=stats,
            ),
        }

    @staticmethod
    def __build_multiy_chart(
        chart_type: str,
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a multi-Y chart entry

        :param chart_type: The chart type. (bar or line)
        :param params: The template parameters for a multi-Y chart.
        :param recordset: The list of fields used to create the multi-Y chart.
        :return: A multi-Y chart entry.
        """

        class Template:
            class X:
                def __init__(
                    self,
                    x: dict[Hashable, Any],
                ) -> None:
                    by = x.get("by")
                    assert isinstance(by, str), f"x.by must be of type str - {type(by)}"
                    self.__by = by

                    sort_by = x.get("sort-by")
                    assert isinstance(sort_by, str) or sort_by is None, f"x.sort-by must be of type str or null - {type(sort_by)}"
                    self.__sort_by = sort_by

                    self.__asc = EntryBuilder.to_sort_order(x.get("order") or "asc")
                    self.__field = x.get("field")
                    assert (
                        isinstance(self.__field, str) or self.__field is None
                    ), f"x.field must be of type str or null - {type(self.__field)}"

                @property
                def by(
                    self,
                ) -> str:
                    return self.__by

                @property
                def sort_by(
                    self,
                ) -> str | None:
                    return self.__sort_by

                @property
                def asc(
                    self,
                ) -> bool:
                    return self.__asc

                @property
                def field(
                    self,
                ) -> str | None:
                    return self.__field

            class Y:
                class Records:
                    def __init__(
                        self,
                        records: dict[Hashable, Any],
                    ) -> None:
                        name_field = records.get("name-field")
                        assert isinstance(name_field, str), f"name-field must be of type str - {type(name_field)}"
                        self.__name_field = name_field

                        data_field = records.get("data-field")
                        assert isinstance(data_field, str), f"data-field must be of type str - {type(data_field)}"
                        self.__data_field = data_field

                        colors = records.get("colors")
                        if isinstance(colors, list):
                            for color in colors:
                                assert isinstance(color, str), f"color must be of type str - {type(color)}"
                        elif isinstance(colors, dict):
                            for color in colors.values():
                                assert isinstance(color, str), f"color must be of type str - {type(color)}"
                        elif colors is None:
                            colors = []
                        elif not isinstance(colors, str):
                            raise DemistoException(f"colors must be of type dict, list or null - {type(colors)}")
                        self.__colors = colors

                    @property
                    def name_field(
                        self,
                    ) -> str:
                        return self.__name_field

                    @property
                    def data_field(
                        self,
                    ) -> str:
                        return self.__data_field

                    @property
                    def colors(
                        self,
                    ) -> dict[Hashable, str] | list[str] | str:
                        return self.__colors

                class Field:
                    def __init__(
                        self,
                        field: dict[Hashable, Any],
                        default_color: str,
                    ) -> None:
                        assert isinstance(field, dict), f"field in y.fields must be of type dict - {type(field)}"
                        color = field.get("color")
                        assert isinstance(color, str) or color is None, f"field.color must be of type str or null - {type(color)}"
                        self.__color = color or default_color

                        self.__label = field.get("label")
                        assert (
                            isinstance(self.__label, str) or self.__label is None
                        ), f"field.label must be of type str or null - {type(self.__label)}"

                    @property
                    def color(
                        self,
                    ) -> str:
                        return self.__color

                    @property
                    def label(
                        self,
                    ) -> str | None:
                        return self.__label

                def __init__(
                    self,
                    y: dict[Hashable, Any],
                ) -> None:
                    self.__records: Records | None = None  # pylint: disable=undefined-variable
                    self.__fields: dict[Hashable, Field] | None = None  # pylint: disable=undefined-variable

                    group = y.get("group")
                    if group == "records":
                        records = y.get(group)
                        assert isinstance(records, dict), f"y.records is required and must be of type dict - {type(records)}"
                        self.__records = self.Records(records)
                    elif group == "fields":
                        fields = y.get(group)
                        assert isinstance(fields, dict), f"y.fields is required and must be of type dict - {type(fields)}"
                        self.__fields = {
                            name: self.Field(field, default_color=color)
                            for (name, field), color in zip(fields.items(), EntryBuilder.list_colors(len(fields)))
                        }
                    else:
                        raise DemistoException(f"y.group must be 'records' or 'fields' - {group}")

                @property
                def records(
                    self,
                ) -> Records | None:
                    return self.__records

                @property
                def fields(
                    self,
                ) -> dict[Hashable, Field] | None:
                    return self.__fields

            def __init__(
                self,
                template: dict[Hashable, Any],
            ) -> None:
                x = template.get("x")
                assert isinstance(x, dict), f"x must be of type dict - {type(x)}"
                self.__x = self.X(x)

                y = template.get("y")
                assert isinstance(y, dict), f"y must be of type dict - {type(y)}"
                self.__y = self.Y(y)

            @property
            def x(
                self,
            ) -> X:
                return self.__x

            @property
            def y(
                self,
            ) -> Y:
                return self.__y

        template = Template(params)
        if records := template.y.records:
            ynames = EntryBuilder.__sum_by(
                recordset=recordset,
                sum_field=records.data_field,
                group_by=records.name_field,
                order_asc=False,
            )
            # Create color mapping
            ycolors = EntryBuilder.__make_color_palette(
                names=[x for x in ynames if isinstance(x, str)],
                colors=records.colors,
            )
            # Build stats
            stats = []
            for x_val, x_records in EntryBuilder.__enum_fields_by_group(
                recordset=recordset,
                group_by=template.x.by,
                sort_by=template.x.sort_by,
                asc=template.x.asc,
            ):
                groups = {k: None for k in ynames}
                xlabel = ""
                for y_fields in x_records:
                    data = y_fields.get(records.data_field)
                    name = y_fields.get(records.name_field)
                    name_str = to_str(name)
                    groups[name_str] = assign_params(
                        name=name_str,
                        data=[to_float(data)],
                        color=ycolors.get(name),
                    )
                    xlabel = xlabel or to_str(y_fields.get(template.x.field) if template.x.field else x_val)

                stats.append(
                    {
                        "name": xlabel,
                        "groups": [
                            assign_params(
                                name=k,
                                data=[0],
                                color=ycolors.get(k),
                            )
                            if v is None
                            else v
                            for k, v in groups.items()
                        ],
                    }
                )
        elif fields := template.y.fields:
            # Build stats
            stats = [
                {
                    "name": to_str(y_fields.get(template.x.field) if template.x.field else x_val),
                    "groups": [
                        {
                            "name": to_str(y_group.label or field or ""),
                            "data": [to_float(y_fields.get(field))],
                            "color": y_group.color,
                        }
                        for field, y_group in fields.items()
                    ],
                }
                for x_val, x_records in EntryBuilder.__enum_fields_by_group(
                    recordset=recordset,
                    group_by=template.x.by,
                    sort_by=template.x.sort_by,
                    asc=template.x.asc,
                )
                for y_fields in x_records
            ]
        else:
            stats = []

        return {
            "Type": EntryType.WIDGET,
            "ContentsFormat": chart_type,
            "Contents": dict(
                assign_params(params=params.get("params")),
                stats=stats,
            ),
        }

    @staticmethod
    def __build_single_bar(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a single bar entry

        :param params: The template parameters for 'single-bar'.
        :param recordset: The list of fields used to create the single-bar chart.
        :return: An single bar entry.
        """
        return EntryBuilder.__build_singley_chart(
            chart_type="bar",
            params=params,
            recordset=recordset,
        )

    @staticmethod
    def __build_stacked_bar(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a stacked bar entry

        :param params: The template parameters for 'stacked-bar'.
        :param recordset: The list of fields used to create the stacked-bar chart.
        :return: An stacked bar entry.
        """
        return EntryBuilder.__build_multiy_chart(
            chart_type="bar",
            params=params,
            recordset=recordset,
        )

    @staticmethod
    def __build_line(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a line entry

        :param params: The template parameters for 'line'.
        :param recordset: The list of fields used to create the line chart.
        :return: A line entry.
        """
        return EntryBuilder.__build_multiy_chart(
            chart_type="line",
            params=params,
            recordset=recordset,
        )

    @staticmethod
    def __build_pie(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a pie entry

        :param params: The template parameters for 'pie'.
        :param recordset: The list of fields used to create the pie chart.
        :return: A pie chart entry.
        """
        return EntryBuilder.__build_singley_chart(
            chart_type="pie",
            params=params,
            recordset=recordset,
        )

    @staticmethod
    def __build_markdown_table(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a markdown table

        :param params: The template parameters for 'markdown-table'.
        :param recordset: The list of fields used to create the markdown table.
        :return: A markdown table.
        """

        class Template:
            class Sort:
                def __init__(
                    self,
                    sort: dict[Hashable, Any],
                ) -> None:
                    self.__by = sort.get("by")
                    assert (
                        isinstance(self.__by, str) or self.__by is None
                    ), f"sort.by must be of type str or null - {type(self.__by)}"
                    self.__asc = EntryBuilder.to_sort_order(sort.get("order") or "asc")

                @property
                def by(
                    self,
                ) -> str | None:
                    return self.__by

                @property
                def asc(
                    self,
                ) -> bool:
                    return self.__asc

            class Column:
                def __init__(
                    self,
                    column: dict[Hashable, Any],
                ) -> None:
                    assert isinstance(column, dict), f"column in columns must be of type dict - {type(column)}"
                    self.field = column.get("field")
                    self.label = column.get("label") or self.field or ""

            def __init__(
                self,
                template: dict[Hashable, Any],
            ) -> None:
                self.__title = template.get("title") or ""
                assert isinstance(self.__title, str), f"title must be of type str or null - {type(self.__title)}"
                self.__columns: list[Column] | None = None  # pylint: disable=undefined-variable
                if columns := template.get("columns"):
                    assert isinstance(columns, list), f"columns must be list or null - {type(columns)}"
                    self.__columns = [self.Column(c) for c in columns]

                sort = template.get("sort") or {}
                assert isinstance(sort, dict), f"sort must be of type dict or null - {type(sort)}"
                self.__sort = self.Sort(sort)

            @property
            def title(
                self,
            ) -> str:
                return self.__title

            @property
            def columns(
                self,
            ) -> list[Column] | None:
                return self.__columns

            @property
            def sort(
                self,
            ) -> Sort:
                return self.__sort

        template = Template(params)

        if not recordset:
            md = ""
        else:
            if template.sort.by:
                recordset = sorted(
                    recordset,
                    key=lambda v: SortableValue(v.get(template.sort.by)),
                    reverse=not template.sort.asc,
                )

            # Build markdown
            md = tableToMarkdown(
                template.title,
                recordset,
                headers=[c.field for c in template.columns] if template.columns else None,
                headerTransform=lambda field: next(
                    (c.label for c in template.columns or [] if c.field == field), pascalToSpace(field.replace("_", " "))
                ),
                sort_headers=False,
            )

        return {
            "Type": EntryType.NOTE,
            "ContentsFormat": EntryFormat.MARKDOWN,
            "HumanReadable": md,
            "Contents": None,
        }

    @staticmethod
    def __build_duration(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a duration entry

        :param params: The template parameters for 'duration'.
        :param recordset: The list of fields used to create the duration entry.
        :return: A duration entry.
        """
        field = params.get("field")
        if not field or not isinstance(field, str):
            raise DemistoException(f"field must be of type str - {type(field)}")

        if len(recordset) > 1:
            raise DemistoException("The duration entry allows at most one record.")

        return {
            "Type": EntryType.WIDGET,
            "ContentsFormat": "duration",
            "Contents": dict(
                assign_params(params=params.get("params")),
                stats=int(to_float((recordset or [{}])[0].get(field))),
            ),
        }

    @staticmethod
    def __build_number(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a number entry

        :param params: The template parameters for 'number'.
        :param recordset: The list of fields used to create the number entry.
        :return: A number entry.
        """
        field = params.get("field")
        if not field or not isinstance(field, str):
            raise DemistoException(f"field must be of type str - {type(field)}")

        if len(recordset) > 1:
            raise DemistoException("The number entry allows at most one record.")

        return {
            "Type": EntryType.WIDGET,
            "ContentsFormat": "number",
            "Contents": dict(
                assign_params(params=params.get("params")),
                stats=to_float((recordset or [{}])[0].get(field)),
            ),
        }

    @staticmethod
    def __build_number_trend(
        params: dict[Hashable, Any],
        recordset: list[dict[Hashable, Any]],
    ) -> dict[Hashable, Any]:
        """Build a number trend entry

        :param params: The template parameters for 'number-trend'.
        :param recordset: The list of fields used to create the number trend entry.
        :return: A number trend entry.
        """
        prev_field = params.get("prev-field")
        if not prev_field or not isinstance(prev_field, str):
            raise DemistoException(f"prev-field must be of type str - {type(prev_field)}")

        curr_field = params.get("curr-field")
        if not curr_field or not isinstance(curr_field, str):
            raise DemistoException(f"curr-field must be of type str - {type(curr_field)}")

        if len(recordset) > 1:
            raise DemistoException("The number-trend entry allows at most one record.")

        fields = (recordset or [{}])[0]

        return {
            "Type": EntryType.WIDGET,
            "ContentsFormat": "number",
            "Contents": dict(
                assign_params(params=params.get("params")),
                stats={
                    "prevSum": to_float(fields.get(prev_field)),
                    "currSum": to_float(fields.get(curr_field)),
                },
            ),
        }

    @staticmethod
    def __build_markdown(
        params: dict[Hashable, Any],
        **kwargs,
    ) -> dict[Hashable, Any]:
        """Build a markdown entry

        :param params: The template parameters for 'markdown'.
        :return: A markdown entry.
        """
        return {
            "Type": EntryType.NOTE,
            "ContentsFormat": EntryFormat.MARKDOWN,
            "HumanReadable": str(params.get("text") or ""),
            "Contents": None,
        }

    @staticmethod
    def to_sort_order(
        order: str,
    ) -> bool:
        if order == "asc":
            return True
        elif order == "desc":
            return False
        else:
            raise DemistoException(f"Invalid sort order - {order}")

    @staticmethod
    def list_colors(
        n: int,
    ) -> list[str]:
        colors = [
            f"rgb({int(r * 256)}, {int(g * 256)}, {int(b * 256)})"
            for r, g, b in [colorsys.hsv_to_rgb(i / n, 1, 0.9) for i in range(max(0, n))]
        ]
        return list(reversed(colors[1::2])) + colors[::2]

    def __get_default_entry(
        self,
        scope: DefaultEntryScope,
        entry_params: dict[Hashable, Any],
    ) -> dict[Hashable, Any] | None:
        default = entry_params.get("default")
        assert default is None or isinstance(default, dict), f"entry.default must be of type dict or null - {type(default)}"
        if not default:
            return None

        scopes = default.get("scope")
        assert scopes is None or isinstance(
            scopes, str | list
        ), f"default.scope must be of type null, str, or list - {type(scopes)}"
        scopes = [str(x) for x in list(DefaultEntryScope)] if scopes is None else scopes
        scopes = scopes if isinstance(scopes, list) else [scopes]
        if scope not in scopes:
            return None

        entry_key = str(scope) if str(scope) in default else "entry"
        entry = default.get(entry_key)
        if isinstance(entry, dict):
            return entry
        elif isinstance(entry, str):
            return {
                "Type": EntryType.NOTE,
                "ContentsFormat": EntryFormat.MARKDOWN,
                "HumanReadable": entry,
                "Contents": None,
            }
        else:
            raise DemistoException(f"default.{entry_key} must be of type stror dict - {type(entry)}")

    def __init__(
        self,
        formatter: Formatter,
        context: ContextData,
    ) -> None:
        self.__formatter = formatter
        self.__context = context

    def build(
        self,
        query: Query,
        entry_params: dict[Hashable, Any],
    ) -> tuple[dict[str, Any], dict[Hashable, Any]]:
        """Build an entry by querying data.

        :param query: The query instance.
        :param entry_params: Template parameters for an entry.
        :return: A pair of extra context and entry.
        """
        query_params = query.query_params
        extra_context: dict[str, Any] = {
            "query": {
                "string": query_params.query_string,
                "timeframe": {
                    "from": query_params.get_earliest_time_iso(utc=True),
                    "to": query_params.get_latest_time_iso(utc=True),
                },
                "request_url": (
                    "/xql/xql-search?phrase="
                    + urllib.parse.quote(query_params.query_string)
                    + "&timeframe="
                    + urllib.parse.quote(
                        json.dumps(
                            {
                                "from": int(query_params.earliest_time.timestamp() * 1000),
                                "to": int(query_params.latest_time.timestamp() * 1000),
                            }
                        )
                    )
                ),
            }
        }

        if not query.available() and (
            entry := self.__get_default_entry(
                scope=DefaultEntryScope.QUERY_SKIPPED,
                entry_params=self.__formatter.build(template=entry_params, context=self.__context.inherit(extra_context)),
            )
        ):
            return extra_context, entry

        entry_type = entry_params.get("type")
        if not entry_type:
            raise DemistoException("type is required.")

        params = entry_params.get(entry_type, {})
        if not isinstance(params, dict):
            raise DemistoException(f"{entry_type} must be of type dict - {type(params)}.")

        build_entry = {
            "single-bar": lambda params, recordset: self.__build_single_bar(
                params=params,
                recordset=recordset,
            ),
            "stacked-bar": lambda params, recordset: self.__build_stacked_bar(
                params=params,
                recordset=recordset,
            ),
            "line": lambda params, recordset: self.__build_line(
                params=params,
                recordset=recordset,
            ),
            "pie": lambda params, recordset: self.__build_pie(
                params=params,
                recordset=recordset,
            ),
            "markdown-table": lambda params, recordset: self.__build_markdown_table(
                params=params,
                recordset=recordset,
            ),
            "duration": lambda params, recordset: self.__build_duration(
                params=params,
                recordset=recordset,
            ),
            "number": lambda params, recordset: self.__build_number(
                params=params,
                recordset=recordset,
            ),
            "number-trend": lambda params, recordset: self.__build_number_trend(
                params=params,
                recordset=recordset,
            ),
            "markdown": lambda params, recordset: self.__build_markdown(
                params=params,
                recordset=recordset,
            ),
        }.get(entry_type)

        if not build_entry:
            raise DemistoException(f"Invalid type - {entry_type}")

        execution_id, recordset = query.query()
        extra_context.update(
            {
                "recordset": recordset,
                "query": dict(
                    extra_context.get("query") or {},
                    execution_id=execution_id,
                    result_url=f"/xql/xql-search/{urllib.parse.quote(execution_id)}",
                ),
            }
        )

        if not recordset and (
            entry := self.__get_default_entry(
                scope=DefaultEntryScope.NO_RECORDSET,
                entry_params=self.__formatter.build(template=entry_params, context=self.__context.inherit(extra_context)),
            )
        ):
            return extra_context, entry
        else:
            return extra_context, build_entry(
                self.__formatter.build(template=params, context=self.__context.inherit(extra_context)), recordset
            )


""" MAIN FUNCTION """


class Main:
    @staticmethod
    def __create_base_context(
        args: dict[Hashable, Any],
    ) -> dict[str, Any]:
        """Create the base context data.

        :param args: The argument parameters.
        :return: The base context data.
        """
        context = args.get("context_data")
        if isinstance(context, str):
            context = json.loads(context)

        assert context is None or isinstance(context, dict), f"Context data must be of type str, dict, or null - {type(context)}"
        return dict(demisto.context(), **(context or {}))

    @staticmethod
    def __create_context(
        base_context: dict[str, Any],
        template: dict[Hashable, Any],
    ) -> ContextData:
        """Create the context data.

        :param base_context: The base context data.
        :param template: The template.
        :return: The context data.
        """

        def __child_paths(
            parent: str,
            paths: list[str],
        ) -> list[str]:
            """Get a list of child node paths.

            :param parent: The name of the parent node.
            :param paths: A list of node paths.
            :return: A list of relative paths under the given parent node.
            """
            cpaths = []
            for path in paths:
                pit = iter(path)
                nit = (next(pit, "\\") if c == "\\" else "" if c == "." else c for c in pit)
                if parent == "".join(itertools.takewhile(lambda x: x != "", iter(nit))):
                    cpaths.append("".join(list(pit)))
            return cpaths

        def __filter_context(
            context: Any,
            filters: list[str],
        ) -> Any:
            """Keep specific values in the context based on the given filters.

            :param context: The context node.
            :param filters: A list of node paths to be removed from the context.
            :return: The filtered context node.
            """
            if any(v == "" for v in filters):
                return context
            elif isinstance(context, dict):
                return {
                    k: v
                    for k, cpaths in ((k, __child_paths(k, filters)) for k in context)
                    if (v := __filter_context(context[k], cpaths)) or any(c == "" for c in cpaths)
                } or None
            elif isinstance(context, list):
                return [v for v in (__filter_context(v, filters) for v in context) if v] or None
            else:
                return None

        def __remove_null(
            context: Any,
            paths: list[str] | None,
            entries_only: bool,
        ) -> Any:
            """Remove key-value pairs from the context where the value is None.

            :param context: The context node.
            :param paths: A list of node paths to be removed from the context where the value is None.
            :param entries_only: Set to True to keep `None` values when they are not associated with keys,
                                 otherwise set to False to remove them.
            :return: The updated context node with key-value pairs removed.
            """
            if isinstance(context, dict):
                if paths is None:
                    return {k: v for k, v in context.items() if v is not None}
                else:
                    return {
                        k: __remove_null(v, cpaths, entries_only)
                        for k, v in context.items()
                        if (not (cpaths := __child_paths(k, paths)) or v is not None or all(path != "" for path in cpaths))
                    }
            elif isinstance(context, list):
                return [__remove_null(v, paths, entries_only) for v in context if entries_only or v is not None]
            else:
                return context

        def __apply_default(
            context: Any,
            default: dict[str, Any],
        ) -> Any:
            """Create a context node with default parameters.

            :param context: The context node.
            :param default: The default parameters to apply.
            :return: The context node with the applied default parameters.
            """
            if isinstance(context, dict):
                context = dict(context)
                for k, dv in default.items():
                    if k in context:
                        if isinstance(dv, dict):
                            context[k] = __apply_default(context.get(k), dv)
                    else:
                        context[k] = dv
                return context
            elif isinstance(context, list):
                return [__apply_default(x, default) for x in context]
            else:
                return context

        fields = demisto.incident()
        fields = dict(fields, **(fields.get("CustomFields") or {}))
        fields.pop("CustomFields", None)

        entries = {
            "context": base_context,
            "alert": fields if (is_xsiam() or is_platform()) else None,
            "issue": fields if is_platform() else None,
            "incident": None if (is_xsiam() or is_platform()) else fields,
        }
        for name, context in dict(entries).items():
            filters = demisto.get(template, f"config.{name}.filters")
            if filters is not None:
                assert isinstance(filters, list), f"config.{name}.filters must be of type null or list - {type(filters)}"
                filters = [f for f in filters if isinstance(f, str)]
                context = __filter_context(context, filters) or {}

            remove_null = demisto.get(template, f"config.{name}.remove-null")
            if remove_null is not None:
                assert isinstance(
                    remove_null, dict
                ), f"config.{name}.remove-null must be of type null or dict - {type(remove_null)}"
                entries_only = argToBoolean(remove_null.get("entries-only", "true"))
                paths = remove_null.get("paths")
                if paths is not None:
                    assert isinstance(
                        paths, list
                    ), f"config.{name}.remove-null.paths must be of type null or list - {type(paths)}"
                    paths = [x for x in paths if isinstance(x, str)]

                context = __remove_null(context, paths, entries_only) or {}

            default = demisto.get(template, f"config.{name}.default")
            if default is not None:
                assert isinstance(default, dict), f"config.{name}.default must be of type dict - {type(default)}"
                context = __apply_default(context, default) or {}

            entries[name] = context

        return ContextData(
            context=entries.get("context"),
            alert=entries.get("alert"),
            issue=entries.get("issue"),
            incident=entries.get("incident"),
        )

    @staticmethod
    def __get_template(
        args: dict[Hashable, Any],
    ) -> tuple[str, dict[Hashable, Any]]:
        """Get the templates with its name

        :param args: The argument parameters.
        :return: The template name and template.
        """
        templates_type = args.get("templates_type") or "raw"
        if templates_type in ("raw", "base64"):
            templates = args.get("templates")
            if templates_type == "base64":
                assert isinstance(
                    templates, str
                ), f"'templates' must be of type str in templates_type = 'base64' - {type(templates)}"
                templates = base64.b64decode(templates.encode()).decode()
        elif templates_type == "list":
            templates = execute_command("getList", {"listName": args.get("templates")})
        else:
            raise DemistoException(f"Invalid template type - {templates_type}")

        if isinstance(templates, str):
            if argToBoolean(args.get("triple_quotes_to_string", "true")):
                templates = re.sub(r"""(\"{3}|'{3}|`{3})(.*?)\1""", lambda m: json.dumps(m.group(2)), templates, flags=re.DOTALL)

            templates = json.loads(templates)
        if not isinstance(templates, dict):
            raise DemistoException(f"Invalid templates - {templates}")

        template_name = args.get("template_name") or ""
        if template := templates.get(template_name):
            if not isinstance(template, dict):
                raise DemistoException(f"Invalid template - {template}")
        else:
            raise DemistoException(f"No templates were found - {template_name}")

        return template_name, template

    @staticmethod
    def __parse_date_time(
        value: Any,
        base_time: datetime | None,
    ) -> datetime:
        """Parse a date time value

        :param value: The date or time to parse
        :param base_time: The base time for the relative time.
        :return: aware datetime object
        """
        if value in (None, ""):
            return datetime.now(pytz.UTC)

        if isinstance(value, int):
            # Parse as time stamp
            try:
                # Considered the value as seconds > milliseconds > microseconds when it's too large (> uint max).
                # (Currently later than 2106-02-07 06:28:15)
                while value > 4294967295:
                    value /= 1000

                return datetime.fromtimestamp(value).astimezone(pytz.UTC)
            except Exception as e:
                raise DemistoException(f"Error with input date / time - {e}")

        if isinstance(value, str):
            value = value.strip()
            try:
                # Parse as time stamp
                return Main.__parse_date_time(int(value), base_time)
            except (TypeError, ValueError):
                pass

        try:
            date_time = dateparser.parse(value, settings=assign_params(RELATIVE_BASE=base_time))
            assert date_time is not None, f"could not parse {value}"

            if date_time.tzinfo is not None:
                return date_time

            date_time = dateparser.parse(
                value,
                settings=assign_params(
                    TIMEZONE="UTC",
                    RETURN_AS_TIMEZONE_AWARE=True,
                    RELATIVE_BASE=base_time,
                ),
            )
            assert date_time is not None, f"could not parse {value}"
            return date_time
        except Exception as e:
            raise DemistoException(f"Error with input date / time - {e}")

    @staticmethod
    def __get_variable_substitution(
        args: dict[Hashable, Any],
        template: dict[Hashable, Any],
    ) -> tuple[str, str]:
        vs = args.get("variable_substitution") or "${,}"
        if isinstance(vs, str):
            vs = vs.split(",", maxsplit=1)
        elif not isinstance(vs, list):
            raise DemistoException(f"Invalid variable substitution - {vs}")

        if not vs or not vs[0]:
            raise DemistoException("variable_substitution must have a opening marker.")
        elif len(vs) == 1:
            assert isinstance(vs[0], str), f"opening marker must be of type str - {vs[0]}"
            vs = [vs[0], ""]
        elif len(vs) == 2:
            assert isinstance(vs[0], str), f"opening marker must be of type str - {vs[0]}"
            assert isinstance(vs[1], str), f"closing marker must be of type str - {vs[1]}"
        else:
            raise DemistoException(f"too many values for variable_substitution - {vs}")

        var_opening = demisto.get(template, "config.variable_substitution.opening")
        if var_opening is not None:
            assert isinstance(var_opening, str), f"config.variable_substitution.opening must be of type str - {var_opening}"
            assert var_opening, "config.variable_substitution.opening cannot be empty when provided"
        else:
            var_opening = vs[0]

        var_closing = demisto.get(template, "config.variable_substitution.closing")
        if var_closing is not None:
            assert isinstance(var_closing, str), f"config.variable_substitution.closing must be of type str - {var_opening}"
        else:
            var_closing = vs[1]

        return str(var_opening), str(var_closing)

    @staticmethod
    def __get_base_time(
        args: dict[Hashable, Any],
        query_node: dict[Hashable, Any],
        context: ContextData,
    ) -> tuple[datetime, datetime]:
        """Get the base time of earliest_time and latest_time

        :param args: The argument parameters.
        :param query_node: The `.query` node of the template.
        :param context: The context data.
        :return: The base time of earliest_time and latest_time.
        """
        round_time = demisto.get(query_node, "time_range.round_time")
        if round_time is None:
            round_time = argToList(args.get("round_time"))
            if len(round_time) == 0:
                round_time_latest = round_time_earliest = 0
            elif len(round_time) == 1:
                round_time_latest = round_time_earliest = arg_to_number(round_time[0]) or 0
            elif len(round_time) == 2:
                round_time[0] = arg_to_number(round_time[0])
                assert isinstance(round_time[0], int), f"list of round_time must be number - {round_time}"
                round_time_earliest = round_time[0]

                round_time[1] = arg_to_number(round_time[1])
                assert isinstance(round_time[1], int), f"list of round_time must be number - {round_time}"
                round_time_latest = round_time[1]
            else:
                raise DemistoException(f"Too many round_time - {round_time}")

        elif isinstance(round_time, dict):
            round_time_earliest = arg_to_number(round_time.get("earliest_time")) or 0
            round_time_latest = arg_to_number(round_time.get("latest_time")) or 0
        elif isinstance(round_time, str | int):
            round_time_latest = round_time_earliest = arg_to_number(round_time) or 0
        else:
            raise DemistoException(f"query.time_range.round_time must be of type str or dict - {round_time}")

        base_time = args.get("base_time")
        if not base_time:
            # Set default base time
            for k in [
                "issue.occurred",
                "alert.occurred",
                "incident.occurred",
                "issue.created",
                "alert.created",
                "incident.created",
            ]:
                base_time = context.get(k)
                if base_time and base_time != "0001-01-01T00:00:00Z":
                    break
            else:
                base_time = "now"

        base_time = Main.__parse_date_time(base_time, None)

        return (
            base_time
            if not round_time_earliest
            else datetime.fromtimestamp(
                math.floor(base_time.timestamp() / round_time_earliest) * round_time_earliest, base_time.tzinfo
            ),
            base_time
            if not round_time_latest
            else datetime.fromtimestamp(
                math.floor(base_time.timestamp() / round_time_latest) * round_time_latest, base_time.tzinfo
            ),
        )

    @staticmethod
    def __build_query_params(
        args: dict[Hashable, Any],
        query_name: str,
        query_node: dict[Hashable, Any],
        context: ContextData,
    ) -> QueryParams:
        """Build query parameters

        :param args: The argument parameters.
        :param query_name: The name of the query.
        :param query_node: The node of the query information.
        :param context: The context data.
        :return: Query parameters.
        """
        earliest_time_base, latest_time_base = Main.__get_base_time(
            args=args,
            query_node=query_node,
            context=context,
        )
        if not (query_string := query_node.get("xql")):
            raise DemistoException("Query string is required.")

        return QueryParams(
            query_name=query_name,
            query_string=query_string,
            earliest_time=Main.__parse_date_time(
                demisto.get(query_node, "time_range.earliest_time", args.get("earliest_time", "24 hours ago")), earliest_time_base
            ),
            latest_time=Main.__parse_date_time(
                demisto.get(query_node, "time_range.latest_time", args.get("latest_time", "now")), latest_time_base
            ),
        )

    @staticmethod
    def __create_lock(
        query_node: dict[Hashable, Any],
    ) -> CoreLock | None:
        """Create a locking instance if possible.

        :param query_node: The node of the query information.
        :return: A lock instance.
        """
        if locking := demisto.get(query_node, "locking"):
            module = locking.get("module") or "core-lock"
            timeout = locking.get("timeout")
            return CoreLock(
                module=module,
                name=locking.get("name"),
                info=locking.get("info"),
                timeout=None if timeout is None else arg_to_number(timeout),
                using=locking.get("using"),
            )
        return None

    def __arg_to_int(
        self,
        name: str,
        default_value: int,
    ) -> int:
        arg = self.__args.get(name)
        if arg is None or arg == "":
            return default_value
        elif isinstance(arg, str | int | float):
            try:
                return int(arg)
            except Exception:
                raise DemistoException(f"Invalid {name} - {arg}")
        else:
            raise DemistoException(f"Invalid {name} - {arg}")

    def __is_query_executable(
        self,
    ) -> bool:
        class Evaluation:
            def __init__(
                self,
                formatter: Formatter,
                context: ContextData,
            ) -> None:
                self.__formatter = formatter
                self.__context = context

            def evaluate(
                self,
                conds: Any,
            ) -> bool:
                if isinstance(conds, str):
                    conds = self.__formatter.build(
                        template=conds,
                        context=self.__context,
                    )

                if isinstance(conds, dict):
                    return any(self.evaluate(k) and self.evaluate(v) for k, v in conds.items())
                elif isinstance(conds, list):
                    return any(self.evaluate(v) for v in conds)
                elif conds is None:
                    return False
                elif isinstance(conds, bool):
                    return conds
                elif isinstance(conds, int | float):
                    return conds != 0.0
                elif isinstance(conds, str):
                    return conds.lower() not in ("", "false")
                else:
                    return bool(conds)

        query = self.__template.get("query") or {}
        if "conditions" not in query:
            # Queries are executable when 'query.conditions' is not specified
            return True
        else:
            conditions = query.get("conditions")

            return Evaluation(
                formatter=Formatter(
                    variable_substitution=self.__variable_substitution,
                    keep_symbol_to_null=False,
                ),
                context=self.__context,
            ).evaluate(conditions)

    def __init__(
        self,
        args: dict[Hashable, Any],
    ) -> None:
        self.__args = args
        self.__template_name, self.__template = self.__get_template(args)
        self.__variable_substitution = self.__get_variable_substitution(args, self.__template)

        context = self.__create_base_context(args)
        self.__context: ContextData = self.__create_context(context, self.__template)

        self.__cache_repo = demisto.get(context, Cache.CACHE_ROOT_KEY)
        self.__cache_type: str = args.get("cache_type") or CacheType.RECORDSET
        if self.__cache_type not in [str(x) for x in list(CacheType)]:
            raise DemistoException(f"Invalid cache_type - {self.__cache_type}")

        self.__max_retries: int = self.__arg_to_int("max_retries", DEFAULT_RETRY_MAX)  # max_retries accepts 0

        self.__retry_interval: int = self.__arg_to_int("retry_interval", DEFAULT_RETRY_INTERVAL) or DEFAULT_RETRY_INTERVAL

        self.__polling_interval: int = self.__arg_to_int("polling_interval", DEFAULT_POLLING_INTERVAL) or DEFAULT_POLLING_INTERVAL

        self.__query_timeout_duration: int = (
            self.__arg_to_int("query_timeout_duration", DEFAULT_QUERY_TIMEOUT_DURATION) or DEFAULT_QUERY_TIMEOUT_DURATION
        )
        self.__output_recordset = argToBoolean(args.get("output_recordset", "false"))

        self.__xql_query_instance: str | None = demisto.get(self.__template, "query.command.using") or args.get(
            "xql_query_instance"
        )

    def create(
        self,
    ) -> CommandResults:
        """Create a graph entry

        :return: The command results.
        """
        formatter = Formatter(
            variable_substitution=self.__variable_substitution,
            keep_symbol_to_null=True,
        )
        query_node = formatter.build(
            template=demisto.get(self.__template, "query"),
            context=self.__context,
        )
        query_params = self.__build_query_params(
            args=self.__args,
            query_name=self.__template_name,
            query_node=query_node,
            context=self.__context,
        )
        cache = Cache(name=self.__template_name, repo=self.__cache_repo)
        entry = cache.load_entry(query_params.query_hash()) if self.__cache_type == CacheType.ENTRY else None

        need_query = not entry and self.__is_query_executable()

        extra_context: dict[str, Any] = {}
        if not entry:
            extra_context, entry = EntryBuilder(
                formatter=formatter,
                context=self.__context,
            ).build(
                query=Query(
                    query_params=query_params,
                    xql_query=XQLQuery(
                        xql_query_instance=self.__xql_query_instance,
                        polling_interval=self.__polling_interval,
                        retry_interval=self.__retry_interval,
                        retry_max=self.__max_retries,
                        query_timeout_duration=self.__query_timeout_duration,
                        core_lock=self.__create_lock(query_node),
                    )
                    if need_query
                    else None,
                    cache=cache if self.__cache_type != CacheType.NONE else None,
                ),
                entry_params=self.__template.get("entry") or {},
            )

        res = {
            "QueryParams": {
                "query_name": query_params.query_name,
                "query_string": query_params.query_string,
                "earliest_time": query_params.get_earliest_time_iso(),
                "latest_time": query_params.get_latest_time_iso(),
            },
            "QueryHash": query_params.query_hash(),
            "RequestURL": demisto.get(extra_context, "query.request_url"),
            "ResultURL": demisto.get(extra_context, "query.result_url"),
            "ExecutionID": demisto.get(extra_context, "query.execution_id"),
            "RecordSet": demisto.get(extra_context, "recordset") or [] if self.__output_recordset else [],
            "Entry": entry,
        }
        if need_query and self.__cache_type == CacheType.ENTRY:
            cache.save_entry(query_params, entry)

        return CommandResults(
            readable_output="Done.",
            outputs={"XQLDSHelper": res},
            raw_response=res,
        )


def main():
    try:
        return_results(Main(demisto.args()).create())
    except Exception as e:
        return_error(f"{e}\n\n{traceback.format_exc()}")


""" ENTRY POINT """


if __name__ in ("__main__", "__builtin__", "builtins"):
    main()