Devo_v2

import demistomock as demisto
from CommonServerPython import *

""" IMPORTS """
import base64
import json
import time
from devo.api import Client, ClientConfig
import concurrent.futures
import tempfile
import urllib.parse
import re
import os
import pandas as pd
from datetime import datetime, UTC
from devo.sender import Lookup, SenderConfigSSL, Sender
from functools import partial


""" GLOBAL VARS """
ALLOW_INSECURE = demisto.params().get("insecure", False)
READER_ENDPOINT = demisto.params().get("reader_endpoint", None)
READER_OAUTH_TOKEN = demisto.params().get("reader_oauth_token", None)
WRITER_RELAY = demisto.params().get("writer_relay", None)
WRITER_CREDENTIALS = demisto.params().get("writer_credentials", None)
LINQ_LINK_BASE = demisto.params().get("linq_link_base", "https://us.devo.com/welcome")
FETCH_INCIDENTS_FILTER = demisto.params().get("fetch_incidents_filters", None)
FETCH_INCIDENTS_LIMIT = demisto.params().get("fetch_incidents_limit") or 50
FETCH_INCIDENTS_LOOKBACK_SECONDS = demisto.params().get("fetch_incidents_lookback_seconds") or 3600
# Deprecated: this parameter is never used
FETCH_INCIDENTS_DEDUPE = demisto.params().get("fetch_incidents_deduplication", None)
FETCH_INCIDENTS_WINDOW = demisto.params().get("fetch_incidents_window")
TIMEOUT = demisto.params().get("timeout", "60")
PORT = arg_to_number(demisto.params().get("port", "443") or "443")
ITEMS_PER_PAGE = 50
IP_AS_STRING = True
LIMIT = 100
HEALTHCHECK_WRITER_RECORD = [{"hello": "world", "from": "demisto-integration"}]
HEALTHCHECK_WRITER_TABLE = "test.keep.free"
RANGE_PATTERN = re.compile("^[0-9]+ [a-zA-Z]+")
TIMESTAMP_PATTERN = re.compile("^[0-9]+")
TIMESTAMP_PATTERN_MILLI = re.compile("^[0-9]+.[0-9]+")
PYTHON_DATETIME_OBJECT_PATTERN = re.compile("\d{4}-[01]\d-[0-3]\d [0-2]\d:[0-5]\d:[0-5]\d(?:\.\d+)?Z?")
PYTHON_DATETIME_OBJECT_FORMAT = "%Y-%m-%d %H:%M:%S.%f"
COUNT_SINGLE_TABLE = 0
COUNT_MULTI_TABLE = 0
COUNT_ALERTS = 0
USER_ALERT_TABLE = demisto.params().get("table_name", None)
USER_PREFIX = demisto.params().get("prefix", None)
INCIDENTS_FETCH_INTERVAL = demisto.params().get("incidentFetchInterval", 1) * 60
DEFAULT_ALERT_TABLE = "siem.logtrust.alert.info"
ALERTS_QUERY = """
from
    {table_name}
select
    eventdate,
    {user_prefix}alertHost,
    {user_prefix}domain,
    {user_prefix}priority,
    {user_prefix}context,
    {user_prefix}category,
    {user_prefix}status,
    {user_prefix}alertId,
    {user_prefix}srcIp,
    {user_prefix}srcPort,
    {user_prefix}srcHost,
    {user_prefix}dstIp,
    {user_prefix}dstPort,
    {user_prefix}dstHost,
    {user_prefix}application,
    {user_prefix}engine,
    {user_prefix}extraData
"""

HEALTHCHECK_QUERY = """
from
    test.keep.free
select
    *
"""

SEVERITY_LEVELS_MAP = {
    "1": 0.5,
    "2": 1,
    "3": 2,
    "4": 3,
    "5": 4,
    "informational": 0.5,
    "low": 1,
    "medium": 2,
    "high": 3,
    "critical": 4,
}

""" HELPER FUNCTIONS """


def timestamp_to_date(timestamp):
    datetime_obj = datetime.fromtimestamp(timestamp)
    return datetime_obj.strftime("%Y-%m-%d %H:%M:%S")


def alert_to_incident(alert, user_prefix):
    alert_severity = float(1)
    context = f"{user_prefix}context"
    alert_id = f"{user_prefix}alertId"
    extra_data = f"{user_prefix}extraData"
    event_date = "eventdate"
    alert_name = alert[context].split(".")[-1]
    alert_description = None
    alert_details = str(alert[alert_id])
    alert_occurred = demisto_ISO(float(alert[event_date]) / 1000)
    alert_labels = []
    try:
        if "alertPriority" in alert[extra_data]:
            priority = alert[extra_data].get("alertPriority")
            if priority and priority != "null" and priority.lower() in SEVERITY_LEVELS_MAP:
                alert_severity = SEVERITY_LEVELS_MAP[priority.lower()]

        if "alertName" in alert[extra_data]:
            name = alert[extra_data].get("alertName")
            if name and name != "null":
                alert_name = name

        if "alertDescription" in alert[extra_data]:
            description = alert[extra_data].get("alertDescription")
            if description and description != "null":
                alert_description = description

    except KeyError:
        demisto.debug("Couldn't get alertPriority, alertName, and/or alertDescription, will take default values")

    new_alert: dict = {"devo.metadata.alert": {}}
    for key in alert:
        if key == extra_data:
            continue
        new_alert["devo.metadata.alert"][key] = alert[key]
        alert_labels.append({"type": f"devo.metadata.alert.{key}", "value": str(alert[key])})
    for key in alert[extra_data]:
        new_alert[key] = alert[extra_data][key]
        alert_labels.append({"type": f"{key}", "value": str(alert[extra_data][key])})

    incident = {
        "name": alert_name,
        "severity": alert_severity,
        "details": alert_details,
        "description": alert_description,
        "occurred": alert_occurred,
        "labels": alert_labels,
        "rawJSON": json.dumps(new_alert),
    }

    return incident


def _to_unix(date, milliseconds=False):
    """
    Convert date to a unix timestamp

    date: A unix timestamp in second, a datetime object,
    pandas.Timestamp object, or string to be parsed
    by pandas.to_datetime
    """

    if date is None:
        return None

    elif date == "now":
        epoch = datetime.now().timestamp()
    elif type(date) is str:
        epoch = pd.to_datetime(date).timestamp()
    elif isinstance(date, pd.Timestamp | datetime):
        if date.tzinfo is None:
            epoch = date.replace(tzinfo=UTC).timestamp()
        else:
            epoch = date.timestamp()
    elif isinstance(date, int | float):
        epoch = date
    else:
        raise Exception("Invalid Date")

    if milliseconds:
        epoch *= 1000

    return int(epoch)


def get_types(self, linq_query, start):
    # Calculate stop time
    stop = _to_unix(start)
    start = stop - 10000

    # Query the data
    response = self.query(query=linq_query, dates={"from": timestamp_to_date(start), "to": timestamp_to_date(stop)})

    try:
        data = json.loads(response)
        if data.get("status", 1) != 0:
            raise ValueError("success value is set as false in response.")

        object_data = data.get("object", [])
    except ValueError as error:
        raise Exception("Error while fetching data : ", error)

    type_dict = {}
    for obj in object_data:
        for key, value in obj.items():
            type_dict[key] = type(value)

    return type_dict


def build_link(query, start_ts_milli, end_ts_milli, mode="queryApp", linq_base=None):
    myb64str = base64.b64encode(
        json.dumps(
            {
                "query": query,
                "mode": mode,
                "dates": {"from": start_ts_milli, "to": end_ts_milli},
            }
        ).encode("ascii")
    ).decode()

    if linq_base:
        url = f"{linq_base}/#/vapps/app.custom.queryApp_dev?&targetQuery={myb64str}"
    else:
        url = f"{LINQ_LINK_BASE}/#/vapps/app.custom.queryApp_dev?&targetQuery={myb64str}"

    return url


def check_configuration():
    api = Client(auth={"token": READER_OAUTH_TOKEN}, address=READER_ENDPOINT, config=ClientConfig(response="json", stream=False))

    # Basic functionality of integration
    demisto.debug("inside fetch from time : ", int(time.time() - 1))
    response = api.query(query=HEALTHCHECK_QUERY, dates={"from": int(time.time() - 1), "to": int(time.time())})

    if json.loads(response).get("status", 1) != 0:
        return False
    demisto.debug("check_configuration output:")
    demisto.debug((json.loads(response)).get("object", []))

    if WRITER_RELAY and WRITER_CREDENTIALS:
        creds = get_writer_creds()
        Sender(
            SenderConfigSSL(
                address=(WRITER_RELAY, PORT),
                key=creds["key"].name,
                cert=creds["crt"].name,
                chain=creds["chain"].name,
            )
        ).send(tag=HEALTHCHECK_WRITER_TABLE, msg=f"{HEALTHCHECK_WRITER_RECORD}")

    if FETCH_INCIDENTS_FILTER:
        alert_filters = check_type(FETCH_INCIDENTS_FILTER, dict)

        assert "type" in alert_filters, 'Missing key: "type" in fetch_incidents_filters'
        assert alert_filters["type"] in ["AND", "OR"], "Unsupported value in fetch_incidents_filters.type"

        filters = check_type(alert_filters.get("filters"), list)
        assert filters, 'Missing key: "filters" in fetch_incidents_filters'

        for filt in filters:
            assert "key" in filt, 'Missing key: "key" in fetch_incidents_filters.filters configuration'
            assert filt["key"], 'Empty value for "key" in fetch_incidents_filters.filters configuration'

            assert "operator" in filt, 'Missing key: "operator" in fetch_incidents_filters.filters configuration'
            assert filt["operator"] in [
                "=",
                "!=",
                "/=",
                ">",
                "<",
                ">=",
                "<=",
                "and",
                "or",
                "->",
            ], "Unsupported operator in fetch_incidents_filters.filters configuration"

            assert "value" in filt, 'Missing key: "value" in fetch_incidents_filters.filters configuration'
            assert filt["value"], 'Empty value for "value" in fetch_incidents_filters.filters configuration'

    # Deprecated: this parameter is never used
    if FETCH_INCIDENTS_DEDUPE:
        dedupe_conf = check_type(FETCH_INCIDENTS_DEDUPE, dict)
        assert isinstance(dedupe_conf["cooldown"], int | float), "Invalid fetch_incidents_deduplication configuration"

    return True


def check_type(input, tar_type):
    if tar_type is list and isinstance(input, str) and input.startswith("[") and input.endswith("]"):
        input = input.replace("[", "").replace("]", "").replace("'", "")
        input = input.split(",")

    if isinstance(input, str):
        input = json.loads(input)
        if not isinstance(input, tar_type):
            raise ValueError(f"tables to query should either be a json string of a {tar_type} or a {tar_type} input")
    elif isinstance(input, tar_type):
        pass
    else:
        raise ValueError(f"tables to query should either be a json string of a {tar_type} or a {tar_type} input")
    return input


# Converts epoch (miliseconds) to ISO string
def demisto_ISO(s_epoch):
    if s_epoch >= 0:
        return datetime.utcfromtimestamp(s_epoch).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    return s_epoch


# We will assume timestamp_from and timestamp_to will be the same format or to will be None
def get_time_range(timestamp_from, timestamp_to):
    if isinstance(timestamp_from, float | int):
        t_from = timestamp_from
        t_to = time.time() if timestamp_to is None else timestamp_to
    elif isinstance(timestamp_from, str):
        if re.fullmatch(RANGE_PATTERN, timestamp_from):
            t_range = parse_date_range(timestamp_from)
            t_from = t_range[0].timestamp()
            t_to = t_range[1].timestamp()
        elif re.fullmatch(TIMESTAMP_PATTERN, timestamp_from) or re.fullmatch(TIMESTAMP_PATTERN_MILLI, timestamp_from):
            t_from = float(timestamp_from)
            t_to = time.time() if timestamp_to is None else float(timestamp_to)
        elif re.fullmatch(PYTHON_DATETIME_OBJECT_PATTERN, timestamp_from):
            t_from = date_to_timestamp(timestamp_from, PYTHON_DATETIME_OBJECT_FORMAT) / 1000
            t_to = time.time() if timestamp_to is None else date_to_timestamp(timestamp_to, PYTHON_DATETIME_OBJECT_FORMAT) / 1000
        else:
            t_from = date_to_timestamp(timestamp_from) / 1000
            t_to = time.time() if timestamp_to is None else date_to_timestamp(timestamp_to) / 1000
    elif isinstance(timestamp_from, datetime):
        t_from = timestamp_from.timestamp()
        t_to = time.time() if timestamp_to is None else timestamp_to.timestamp()
    else:
        raise ValueError(f"To and From Dates should have values. {timestamp_from=} {timestamp_to=}")
    current_time: float = time.time()
    if t_from > current_time or t_to > current_time:
        raise ValueError("Date should not be greater than current time")
    return (t_from, t_to)


def get_writer_creds():
    if WRITER_RELAY is None:
        raise ValueError("writer_relay is not set in your Devo Integration")

    if WRITER_CREDENTIALS is None:
        raise ValueError("writer_credentials are not set in your Devo Integration")

    write_credentials = check_type(WRITER_CREDENTIALS, dict)
    assert "key" in write_credentials, 'Required key: "key" is not present in writer credentials'
    assert "crt" in write_credentials, 'Required key: "crt" is not present in writer credentials'
    assert "chain" in write_credentials, 'Required key: "chain" is not present in writer credentials'

    # Limitation in Devo DS Connector SDK. Currently require filepaths for credentials.
    # Will accept file-handler type objects in the future.
    key_tmp = tempfile.NamedTemporaryFile(mode="w")
    crt_tmp = tempfile.NamedTemporaryFile(mode="w")
    chain_tmp = tempfile.NamedTemporaryFile(mode="w")

    key_tmp.write(write_credentials["key"])
    crt_tmp.write(write_credentials["crt"])
    chain_tmp.write(write_credentials["chain"])

    key_tmp.flush()
    crt_tmp.flush()
    chain_tmp.flush()

    creds = {"key": key_tmp, "crt": crt_tmp, "chain": chain_tmp}

    return creds


def parallel_query_helper(sub_query, append_list, timestamp_from, timestamp_to):
    append_list.extend(
        list(
            json.loads(
                Client(
                    auth={"token": READER_OAUTH_TOKEN},
                    address=READER_ENDPOINT,
                    config=ClientConfig(response="json", stream=False),
                ).query(query=sub_query, dates={"from": timestamp_to_date(timestamp_from), "to": timestamp_to_date(timestamp_to)})
            ).get("object", [])
        )
    )


""" FUNCTIONS """


def fetch_incidents():
    api = Client(auth={"token": READER_OAUTH_TOKEN}, address=READER_ENDPOINT, config=ClientConfig(response="json", stream=False))

    demisto.debug(f"ts: {time.time()} | func: fetch_incidents | msg: begin fetch incidents")
    last_run = demisto.getLastRun()
    user_prefix = f"{USER_PREFIX}_" if USER_PREFIX else ""
    user_alert_table = USER_ALERT_TABLE if USER_ALERT_TABLE else DEFAULT_ALERT_TABLE
    alert_query = ALERTS_QUERY.format(table_name=user_alert_table, user_prefix=user_prefix)

    to_time = time.time()
    from_time = 0.0
    alert_id = f"{user_prefix}alertId"
    cur_events: list = []
    final_events: list = []
    new_last_run: dict = {}

    if int(FETCH_INCIDENTS_LIMIT) < 10 or int(FETCH_INCIDENTS_LIMIT) > 100:
        raise ValueError("Fetch incidents limit should be greater than or equal to 10 and smaller than or equal to 100")

    if FETCH_INCIDENTS_FILTER:
        alert_filters = check_type(FETCH_INCIDENTS_FILTER, dict)
        filter_string = ""
        if alert_filters["type"] == "AND":
            filter_string = " , ".join(
                [
                    f'{user_prefix}{filt["key"]} {filt["operator"]} "{urllib.parse.quote(filt["value"])}"'
                    for filt in alert_filters["filters"]
                ]
            )
        elif alert_filters["type"] == "OR":
            filter_string = " or ".join(
                [
                    f'{user_prefix}{filt["key"]} {filt["operator"]} "{urllib.parse.quote(filt["value"])}"'
                    for filt in alert_filters["filters"]
                ]
            )

        alert_query = f"{alert_query} where {filter_string}"

    alert_query = f"{alert_query} limit {FETCH_INCIDENTS_LIMIT}"

    if FETCH_INCIDENTS_WINDOW:
        # Use the configured window for from_time if FETCH_INCIDENTS_WINDOW is set
        from_time = to_time - float(FETCH_INCIDENTS_WINDOW)
    elif "from_time" in last_run:
        # If FETCH_INCIDENTS_WINDOW is not set and it is incremental pull
        from_time = float(last_run["from_time"])
        new_last_run["from_time"] = from_time
    else:
        # If FETCH_INCIDENTS_WINDOW is not set and if it is initial pull
        from_time = to_time - float(FETCH_INCIDENTS_LOOKBACK_SECONDS)
        new_last_run["from_time"] = from_time

    # execute the query and get the events
    from_timestamp = timestamp_to_date(from_time)
    to_timestamp = timestamp_to_date(to_time)

    demisto.debug("inside fetch_incident method: ")
    demisto.debug("from time :", from_time)
    demisto.debug("to time : ", to_time)

    # execute the query and get the events
    # reverse the list so that the most recent event timestamp event is taken when de-duping if needed.
    events = api.query(query=alert_query, dates={"from": from_timestamp, "to": to_timestamp})

    # Parse the JSON string
    data_dict = json.loads(events)

    # Access the "object" property
    object_data = data_dict.get("object", [])
    extra_data = f"{user_prefix}extraData"
    event_date = "eventdate"

    # convert the events to demisto incident
    incidents = []

    demisto.debug(f"ts: {time.time()} | func: fetch_incidents | msg: number of alerts returned {len(object_data)}")

    # de duplicate events between two consecutive fetches
    if "last_fetch_events" in last_run:
        # Retrieve the alert Ids that is stored in the dictionary
        last_events_ids = []
        for record in last_run.get("last_fetch_events"):
            for key in record:
                last_events_ids.append(key)
        demisto.debug(f"List of event ids fetched in last poll: {last_events_ids}")
        for event in object_data:
            if event[alert_id] not in last_events_ids:
                final_events.append(event)
    else:
        final_events = object_data

    demisto.debug(f"ts: {time.time()} | func: fetch_incidents | msg: number of final_events {len(final_events)}")

    # Store in a list alert_id and timestamp of the alerts. [{alert_id:timestamp},{alert_id:timestamp}]
    demisto.debug(final_events)
    for event in final_events:
        if not isinstance(event[extra_data], dict):
            event[extra_data] = json.loads(event[extra_data])
        if event[extra_data] is not None:
            for ed in event[extra_data]:
                if event[extra_data][ed] and isinstance(event[extra_data][ed], str):
                    event[extra_data][ed] = urllib.parse.unquote_plus(event[extra_data][ed])
        cur_events.append({event[alert_id]: event[event_date] / 1000})
        inc = alert_to_incident(event, user_prefix)
        incidents.append(inc)

    # Combine the previously stored alert list with newly fetched alerts
    if last_run.get("last_fetch_events"):
        for record in last_run.get("last_fetch_events"):
            demisto.debug(record)
            cur_events.append(record)
    demisto.debug(cur_events)
    # update new_last_run and add the event_date of the last event fetched
    if len(final_events) > 0:
        new_last_run["from_time"] = max(event[event_date] for event in final_events) / 1000
    else:
        # set the to_time to current to_time, if no data recieved
        new_last_run["from_time"] = to_time

    from_previous_poll: list = []

    # Before completing the run removing the expired alerts in a list based on the timestamp
    for record in cur_events:
        for timestamp in record.values():
            if timestamp > from_time:
                from_previous_poll.append(record)

    new_last_run["last_fetch_events"] = from_previous_poll

    demisto.setLastRun(new_last_run)

    demisto.debug(f"ts: {time.time()} | func: fetch_incidents | msg: last_run_set = {new_last_run}")
    demisto.debug(f"ts: {time.time()} | func: fetch_incidents | msg: number of incidents generated {len(incidents)}")

    # this command will create incidents in Demisto
    demisto.incidents(incidents)

    return incidents


def filter_results_by_fields(results, filtered_columns_string):
    """
    Filter a list of dictionaries by including only specified fields.

    Parameters:
        - results (list): A list of dictionaries representing rows of data.
        - filtered_columns_string (str): A comma-separated string containing field names to include.

    Returns:
        list: A new list of dictionaries with only the specified fields.

    Raises:
        ValueError: If the filtered_columns_string contains invalid column names.
    """
    if filtered_columns_string == "":
        raise ValueError("filtered_columns cannot be empty.")

    if not filtered_columns_string:
        return results

    if not results:
        return results

    filtered_columns_list = argToList(filtered_columns_string)

    # Check if all fields from filtered_columns_list are present in the first dictionary in results
    first_dict = results[0]
    missing_columns = set(filtered_columns_list) - set(first_dict)
    if missing_columns:
        raise ValueError(f"Fields {list(missing_columns)} not found in query result")

    filtered_results = []

    for result in results:
        filtered_result = {column: result.get(column) for column in filtered_columns_list}
        filtered_results.append(filtered_result)

    return filtered_results


def run_query_command(offset, items, ip_as_string):
    to_query = demisto.args()["query"]
    timestamp_from = demisto.args()["from"]
    timestamp_to = demisto.args().get("to", None)
    write_context = demisto.args()["writeToContext"].lower()
    query_timeout = int(demisto.args().get("queryTimeout", TIMEOUT))
    linq_base = demisto.args().get("linqLinkBase", None)
    time_range = get_time_range(timestamp_from, timestamp_to)
    to_query = f"{to_query} offset {offset} limit {items}"
    filtered_columns = demisto.args().get("filtered_columns", None)
    results: Union[str, bytes] = ""

    from_time = timestamp_to_date(time_range[0])
    to_time = timestamp_to_date(time_range[1])

    demisto.debug("inside run_query_command method: ")
    demisto.debug("from time :", from_time)
    demisto.debug("to time : ", to_time)

    try:
        api = Client(
            auth={"token": READER_OAUTH_TOKEN},
            address=READER_ENDPOINT,
            timeout=query_timeout,
            config=ClientConfig(response="json", stream=False),
        )

        results = api.query(query=to_query, dates={"from": from_time, "to": to_time}, ip_as_string=ip_as_string)
    except Exception as e:
        return_error(f"Failed to execute Devo query: {str(e)}")

    # Parse the JSON string
    data_dict = json.loads(results)

    # Access the "object" property
    object_data = data_dict.get("object", [])

    demisto.debug("result in run-query function :")
    demisto.debug(data_dict)
    global COUNT_SINGLE_TABLE
    COUNT_SINGLE_TABLE = len(object_data)
    querylink = {
        "DevoTableLink": build_link(
            to_query,
            int(1000 * float(time_range[0])),
            int(1000 * float(time_range[1])),
            linq_base=linq_base,
        )
    }
    results = filter_results_by_fields(object_data, filtered_columns)

    entry = {
        "Type": entryTypes["note"],
        "Contents": results,
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
    }
    entry_linq = {
        "Type": entryTypes["note"],
        "Contents": querylink,
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
    }

    if len(results) == 0:
        entry["HumanReadable"] = "No results found"
        entry["Devo.QueryResults"] = None
        entry["Devo.QueryLink"] = querylink
        return entry
    headers = list(results[0].keys())

    md = tableToMarkdown("Devo query results", results, headers)
    entry["HumanReadable"] = md

    md_linq = tableToMarkdown(
        "Link to Devo Query",
        {"DevoTableLink": f'[Devo Direct Link]({querylink["DevoTableLink"]})'},
    )
    entry_linq["HumanReadable"] = md_linq

    if write_context == "true":
        entry["EntryContext"] = {"Devo.QueryResults": createContext(results)}
        entry_linq["EntryContext"] = {"Devo.QueryLink": createContext(querylink)}
    return [entry, entry_linq]


def get_alerts_command(offset, items):
    timestamp_from = demisto.args()["from"]
    timestamp_to = demisto.args().get("to", None)
    alert_filters = demisto.args().get("filters", None)
    write_context = demisto.args()["writeToContext"].lower()
    linq_base = demisto.args().get("linqLinkBase", None)
    query_timeout = int(demisto.args().get("queryTimeout", TIMEOUT))
    user_alert_table = demisto.args().get("table_name", None)
    user_prefix = demisto.args().get("prefix", "")
    filtered_columns = demisto.args().get("filtered_columns", None)
    user_alert_table = user_alert_table if user_alert_table else DEFAULT_ALERT_TABLE
    if user_prefix:
        user_prefix = f"{user_prefix}_"
    alert_query = ALERTS_QUERY.format(table_name=user_alert_table, user_prefix=user_prefix)

    query = f"{alert_query} offset {offset} limit {items}"
    time_range = get_time_range(timestamp_from, timestamp_to)
    from_time = timestamp_to_date(time_range[0])
    to_time = timestamp_to_date(time_range[1])

    if alert_filters:
        alert_filters = check_type(alert_filters, dict)
        filter_string = ""
        if alert_filters["type"] == "AND":
            filter_string = ", ".join(
                [f'{filt["key"]} {filt["operator"]} "{urllib.parse.quote(filt["value"])}"' for filt in alert_filters["filters"]]
            )
        elif alert_filters["type"] == "OR":
            filter_string = " or ".join(
                [f'{filt["key"]} {filt["operator"]} "{urllib.parse.quote(filt["value"])}"' for filt in alert_filters["filters"]]
            )
        alert_query = f"{alert_query} where {filter_string}"

    api = Client(
        auth={"token": READER_OAUTH_TOKEN},
        address=READER_ENDPOINT,
        timeout=query_timeout,
        config=ClientConfig(response="json", stream=False),
    )

    results = api.query(query=query, dates={"from": from_time, "to": to_time, "timeZone": "UTC"})

    # Parse the JSON string
    data_dict = json.loads(results)

    # Access the "object" property
    object_data = data_dict.get("object", [])

    global COUNT_ALERTS
    COUNT_ALERTS = len(object_data)

    querylink = {
        "DevoTableLink": build_link(
            alert_query,
            int(1000 * float(time_range[0])),
            int(1000 * float(time_range[1])),
            linq_base=linq_base,
        )
    }

    extra_data = f"{user_prefix}extraData"

    for res in object_data:
        if not isinstance(res[extra_data], dict):
            res[extra_data] = json.loads(res[extra_data])

        if res[extra_data] is not None:
            for ed in res[extra_data]:
                if res[extra_data][ed] is not None:
                    res[extra_data][ed] = urllib.parse.unquote_plus(res[extra_data][ed])

    result = filter_results_by_fields(object_data, filtered_columns)

    entry = {
        "Type": entryTypes["note"],
        "Contents": result,
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
    }
    entry_linq = {
        "Type": entryTypes["note"],
        "Contents": querylink,
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
    }

    if len(result) == 0:
        entry["HumanReadable"] = "No results found"
        entry["Devo.AlertsResults"] = None
        entry_linq["Devo.QueryLink"] = querylink
        return entry

    headers = list(result[0].keys())

    md = tableToMarkdown("Devo query results", result, headers)
    entry["HumanReadable"] = md

    md_linq = tableToMarkdown(
        "Link to Devo Query",
        {"DevoTableLink": f'[Devo Direct Link]({querylink["DevoTableLink"]})'},
    )
    entry_linq["HumanReadable"] = md_linq

    if write_context == "true":
        entry["EntryContext"] = {"Devo.AlertsResults": createContext(result)}
        entry_linq["EntryContext"] = {"Devo.QueryLink": createContext(querylink)}

    # raise Exception("on line 530")
    return [entry, entry_linq]


def multi_table_query_command(offset, items):
    tables_to_query = check_type(demisto.args()["tables"], list)
    search_token = demisto.args()["searchToken"]
    timestamp_from = demisto.args()["from"]
    query_timeout = int(demisto.args().get("queryTimeout", TIMEOUT))
    timestamp_to = demisto.args().get("to", None)
    write_context = demisto.args()["writeToContext"].lower()
    filtered_columns = demisto.args().get("filtered_columns", None)
    global COUNT_MULTI_TABLE
    time_range = get_time_range(timestamp_from, timestamp_to)

    futures = []
    all_results: List[dict] = []
    sub_queries = []

    api = Client(
        auth={"token": READER_OAUTH_TOKEN},
        address=READER_ENDPOINT,
        timeout=query_timeout,
        config=ClientConfig(response="json", stream=False),
    )

    api.get_types = partial(get_types, api)

    for table in tables_to_query:
        fields = api.get_types(f"from {table} select *", "now").keys()
        clauses = [f'( isnotnull({field}) and str({field})->"' + search_token + '")' for field in fields]
        sub_queries.append(
            "from " + table + " where" + " or ".join(clauses) + " select *" + " offset " + str(offset) + " limit " + str(items)
        )

    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        for q in sub_queries:
            futures.append(executor.submit(parallel_query_helper, q, all_results, time_range[0], time_range[1]))

    concurrent.futures.wait(futures)

    all_results = filter_results_by_fields(all_results, filtered_columns)

    entry = {
        "Type": entryTypes["note"],
        "Contents": all_results,
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
    }

    COUNT_MULTI_TABLE = len(all_results)
    if len(all_results) == 0:
        entry["HumanReadable"] = "No results found"
        return entry

    headers: set = set().union(*(r.keys() for r in all_results))

    md = tableToMarkdown("Devo query results", all_results, headers)
    entry["HumanReadable"] = md

    if write_context == "true":
        entry["EntryContext"] = {"Devo.MultiResults": createContext(all_results)}
    return entry


def convert_to_str(value):
    if isinstance(value, list) and not value:
        return_warning("Empty list encountered.")
        return "[]"
    elif isinstance(value, dict) and not value:
        return_warning("Empty dictionary encountered.")
        return "{}"
    elif isinstance(value, list | dict):
        return json.dumps(value)
    return str(value)


def write_to_table_command():
    tag = demisto.args().get("tag", None)
    tableName = demisto.args()["tableName"]
    records_str = demisto.args()["records"]
    linq_base = demisto.args().get("linqLinkBase", None)

    final_tag = tag or tableName

    # Use json.loads to parse the records string
    try:
        records = json.loads(records_str)
    except json.JSONDecodeError:
        return_error("Error decoding JSON. Please ensure the records are valid JSON.")

    # Ensure records is a list
    if not isinstance(records, list):
        return_error("The 'records' parameter must be a list.")

    # Check if all records are empty
    if all(not record for record in records):
        return_error("All records are empty.")

    creds = get_writer_creds()
    linq = f"from {tableName}"

    sender = Sender(
        SenderConfigSSL(
            address=(WRITER_RELAY, PORT),
            key=creds["key"].name,
            cert=creds["crt"].name,
            chain=creds["chain"].name,
        )
    )

    total_events = 0
    total_bytes_sent = 0

    for r in records:
        # Convert each record to a JSON string or string
        formatted_record = convert_to_str(r)

        # If the record is empty, skip sending it
        if not formatted_record.strip():
            continue

        # Send each record to Devo with the specified tag
        sender.send(tag=final_tag, msg=formatted_record)

        # Update totals
        total_events += 1
        total_bytes_sent += len(formatted_record.encode("utf-8"))

    current_ts = int(time.time())
    start_ts = (current_ts - 30) * 1000
    end_ts = (current_ts + 30) * 1000

    querylink = {
        "DevoTableLink": build_link(
            linq,
            start_ts,
            end_ts,
            linq_base=linq_base,
        )
    }

    entry_linq = {
        "Type": entryTypes["note"],
        "Contents": querylink,
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
        "EntryContext": {"Devo.QueryLink": createContext(querylink)},
    }

    entry = {
        "Type": entryTypes["note"],
        "Contents": {"TotalRecords": total_events, "TotalBytesSent": total_bytes_sent},
        "ContentsFormat": formats["json"],
        "ReadableContentsFormat": formats["markdown"],
        "EntryContext": {"Devo.TotalRecords": total_events, "Devo.TotalBytesSent": total_bytes_sent, "Devo.LinqQuery": linq},
    }

    md_message = f"Total Records Sent: {total_events}.\nTotal Bytes Sent: {total_bytes_sent}."
    entry["HumanReadable"] = md_message

    md_linq = tableToMarkdown(
        "Link to Devo Query",
        {"DevoTableLink": f'[Devo Direct Link]({querylink["DevoTableLink"]})'},
    )
    entry_linq["HumanReadable"] = md_linq

    return [entry, entry_linq]


def write_to_lookup_table_command():
    lookup_table_name = demisto.args().get("lookupTableName")
    headers_param = demisto.args().get("headers")
    records_param = demisto.args().get("records")

    if not lookup_table_name or not headers_param or not records_param:
        return_error("Missing required arguments. Please provide lookupTableName, headers, and records.")

    # Parse JSON strings to Python objects
    try:
        headers = json.loads(headers_param)
        records = json.loads(records_param)
    except json.JSONDecodeError as e:
        return_error(f"Failed to parse JSON: {str(e)}")

    creds = get_writer_creds()

    engine_config = SenderConfigSSL(
        address=(WRITER_RELAY, PORT),
        key=creds["key"].name,
        cert=creds["crt"].name,
        chain=creds["chain"].name,
    )

    con = None
    total_events = 0
    total_bytes = 0

    try:
        # Validate headers
        if not isinstance(headers, dict) or "headers" not in headers or not isinstance(headers["headers"], list):
            raise ValueError("Invalid headers format. 'headers' must be a list.")

        columns = headers["headers"]

        # Set default values for optional parameters
        key_index = int(headers.get("key_index", 0))  # Ensure it's casted to integer
        action = headers.get("action", "INC")  # Set default value to 'INC'

        # Validate key_index
        if key_index < 0:
            raise ValueError("key_index must be a non-negative integer value.")

        # Validate action
        if action not in {"INC", "FULL"}:
            raise ValueError("action must be either 'INC' or 'FULL'.")

        con = Sender(config=engine_config, timeout=60)
        lookup = Lookup(name=lookup_table_name, con=con)

        lookup.send_headers(headers=columns, key_index=key_index, event="START", action=action)

        # Send data lines
        for r in records:
            fields = r.get("fields", [])
            delete = r.get("delete", False)
            lookup.send_data_line(key_index=key_index, fields=fields, delete=delete)
            total_events += 1
            total_bytes += len(json.dumps(r))

        # Send end event
        lookup.send_headers(headers=columns, key_index=key_index, event="END", action=action)

        # Flush buffer and shutdown connection
        con.flush_buffer()
        con.socket.shutdown(0)

        # Return three lines as output
        return f"Lookup Table Name: {lookup_table_name}.\nTotal Records Sent: {total_events}.\nTotal Bytes Sent: {total_bytes}."

    except Exception as e:
        return_error(f"Failed to execute command write-to-lookup-table. Error: {str(e)}")

    finally:
        if con:
            con.flush_buffer()
            con.socket.shutdown(0)


def main():
    """EXECUTION CODE"""
    try:
        if ALLOW_INSECURE:
            os.environ["CURL_CA_BUNDLE"] = ""
            os.environ["PYTHONWARNINGS"] = "ignore:Unverified HTTPS request"
        handle_proxy()
        if demisto.command() == "test-module":
            check_configuration()
            demisto.results("ok")
        elif demisto.command() == "fetch-incidents":
            fetch_incidents()
        elif demisto.command() == "devo-run-query":
            OFFSET = 0
            items_per_page = int(demisto.args().get("items_per_page", ITEMS_PER_PAGE))

            if items_per_page <= 0:
                raise ValueError("items_per_page should be a positive non-zero value.")
            total = 0
            ip_as_string = argToBoolean(demisto.args().get("ip_as_string", IP_AS_STRING))
            demisto.results(run_query_command(OFFSET, items_per_page, ip_as_string))
            total = total + COUNT_SINGLE_TABLE
            while items_per_page == COUNT_SINGLE_TABLE:
                OFFSET = OFFSET + items_per_page
                total = total + COUNT_SINGLE_TABLE
                demisto.results(run_query_command(OFFSET, items_per_page, ip_as_string))
        elif demisto.command() == "devo-get-alerts":
            OFFSET = 0
            items_per_page = int(demisto.args().get("items_per_page", ITEMS_PER_PAGE))
            if items_per_page <= 0:
                raise ValueError("items_per_page should be a positive non-zero value.")
            total = 0
            demisto.results(get_alerts_command(OFFSET, items_per_page))
            total = total + COUNT_ALERTS
            while items_per_page == COUNT_ALERTS:
                OFFSET = OFFSET + items_per_page
                total = total + COUNT_ALERTS
                demisto.results(get_alerts_command(OFFSET, items_per_page))
        elif demisto.command() == "devo-multi-table-query":
            OFFSET = 0
            items_per_page = int(demisto.args().get("items_per_page", ITEMS_PER_PAGE))
            if items_per_page <= 0:
                raise ValueError("items_per_page should be a positive non-zero value.")
            total = 0
            demisto.results(multi_table_query_command(OFFSET, items_per_page))
            total = total + COUNT_MULTI_TABLE
            while items_per_page * 2 == COUNT_MULTI_TABLE:
                OFFSET = OFFSET + items_per_page
                total = total + COUNT_MULTI_TABLE
                demisto.results(multi_table_query_command(OFFSET, items_per_page))
        elif demisto.command() == "devo-write-to-table":
            demisto.results(write_to_table_command())
        elif demisto.command() == "devo-write-to-lookup-table":
            demisto.results(write_to_lookup_table_command())
    except Exception as e:
        return_error(f"Failed to execute command {demisto.command()}. Error: {str(e)}.")


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