<commit_before>import argparse
import asyncio
import importlib.util
import logging
import os
import signal
import traceback
from multiprocessing import get_context
from typing import List, Text, Optional, Tuple, Union, Iterable

import aiohttp
import ruamel.yaml as yaml

import rasa.cli.utils as cli_utils
import rasa.utils.io as io_utils
from rasa.cli.arguments import x as arguments
from rasa.constants import (
    DEFAULT_ENDPOINTS_PATH,
    DEFAULT_CREDENTIALS_PATH,
    DEFAULT_DOMAIN_PATH,
    DEFAULT_CONFIG_PATH,
    DEFAULT_LOG_LEVEL_RASA_X,
    DEFAULT_RASA_X_PORT,
    DEFAULT_RASA_PORT,
)
from rasa.core.utils import AvailableEndpoints
from rasa.utils.endpoints import EndpointConfig

logger = logging.getLogger(__name__)

DEFAULT_EVENTS_DB = "events.db"


# noinspection PyProtectedMember
def add_subparser(
    subparsers: argparse._SubParsersAction, parents: List[argparse.ArgumentParser]
):
    x_parser_args = {
        "parents": parents,
        "conflict_handler": "resolve",
        "formatter_class": argparse.ArgumentDefaultsHelpFormatter,
    }

    if is_rasa_x_installed():
        # we'll only show the help msg for the command if Rasa X is actually installed
        x_parser_args["help"] = "Starts the Rasa X interface."

    shell_parser = subparsers.add_parser("x", **x_parser_args)
    shell_parser.set_defaults(func=rasa_x)

    arguments.set_x_arguments(shell_parser)


def _rasa_service(
    args: argparse.Namespace,
    endpoints: AvailableEndpoints,
    rasa_x_url: Optional[Text] = None,
    credentials_path: Optional[Text] = None,
):
    """Starts the Rasa application."""
    from rasa.core.run import serve_application

    # needs separate logging configuration as it is started in its own process
    logging.basicConfig(level=args.loglevel)
    io_utils.configure_colored_logging(args.loglevel)
    logging.getLogger("apscheduler").setLevel(logging.WARNING)

    if not credentials_path:
        credentials_path = _prepare_credentials_for_rasa_x(
            args.credentials, rasa_x_url=rasa_x_url
        )

    serve_application(
        endpoints=endpoints,
        port=args.port,
        credentials=credentials_path,
        cors=args.cors,
        auth_token=args.auth_token,
        enable_api=True,
        jwt_secret=args.jwt_secret,
        jwt_method=args.jwt_method,
        ssl_certificate=args.ssl_certificate,
        ssl_keyfile=args.ssl_keyfile,
        ssl_password=args.ssl_password,
    )


def _prepare_credentials_for_rasa_x(
    credentials_path: Optional[Text], rasa_x_url: Optional[Text] = None
) -> Text:
    credentials_path = cli_utils.get_validated_path(
        credentials_path, "credentials", DEFAULT_CREDENTIALS_PATH, True
    )
    if credentials_path:
        credentials = io_utils.read_config_file(credentials_path)
    else:
        credentials = {}

    # this makes sure the Rasa X is properly configured no matter what
    if rasa_x_url:
        credentials["rasa"] = {"url": rasa_x_url}
    dumped_credentials = yaml.dump(credentials, default_flow_style=False)
    tmp_credentials = io_utils.create_temporary_file(dumped_credentials, "yml")

    return tmp_credentials


def _overwrite_endpoints_for_local_x(
    endpoints: AvailableEndpoints, rasa_x_token: Text, rasa_x_url: Text
):
    from rasa.utils.endpoints import EndpointConfig
    import questionary

    # Checking if endpoint.yml has existing url and wait time values set, if so give warning we are overwriting
    # the endpoint.yml file.
    custom_wait_time_pulls = endpoints.model.kwargs.get("wait_time_between_pulls")
    custom_url = endpoints.model.url


    if custom_url and custom_url != rasa_x_url:
        logger.info(
            f"Ignoring url '{custom_url}' from 'endpoints.yml' and using "
            f"{rasa_x_url}/projects/default/models/tag/production instead"
        )

    endpoints.model = EndpointConfig(
        f"{rasa_x_url}/projects/default/models/tags/production",
        token=rasa_x_token,
        wait_time_between_pulls=custom_wait_time_pulls or 2,
    )

    overwrite_existing_event_broker = False
    if endpoints.event_broker and not _is_correct_event_broker(endpoints.event_broker):
        cli_utils.print_error(
            "Rasa X currently only supports a SQLite event broker with path '{}' "
            "when running locally. You can deploy Rasa X with Docker "
            "(https://rasa.com/docs/rasa-x/deploy/) if you want to use "
            "other event broker configurations.".format(DEFAULT_EVENTS_DB)
        )
        overwrite_existing_event_broker = questionary.confirm(
            "Do you want to continue with the default SQLite event broker?"
        ).ask()<commit_msg>fixing linting error<commit_after>import argparse
import asyncio
import importlib.util
import logging
import os
import signal
import traceback
from multiprocessing import get_context
from typing import List, Text, Optional, Tuple, Union, Iterable

import aiohttp
import ruamel.yaml as yaml

import rasa.cli.utils as cli_utils
import rasa.utils.io as io_utils
from rasa.cli.arguments import x as arguments
from rasa.constants import (
    DEFAULT_ENDPOINTS_PATH,
    DEFAULT_CREDENTIALS_PATH,
    DEFAULT_DOMAIN_PATH,
    DEFAULT_CONFIG_PATH,
    DEFAULT_LOG_LEVEL_RASA_X,
    DEFAULT_RASA_X_PORT,
    DEFAULT_RASA_PORT,
)
from rasa.core.utils import AvailableEndpoints
from rasa.utils.endpoints import EndpointConfig

logger = logging.getLogger(__name__)

DEFAULT_EVENTS_DB = "events.db"


# noinspection PyProtectedMember
def add_subparser(
    subparsers: argparse._SubParsersAction, parents: List[argparse.ArgumentParser]
):
    x_parser_args = {
        "parents": parents,
        "conflict_handler": "resolve",
        "formatter_class": argparse.ArgumentDefaultsHelpFormatter,
    }

    if is_rasa_x_installed():
        # we'll only show the help msg for the command if Rasa X is actually installed
        x_parser_args["help"] = "Starts the Rasa X interface."

    shell_parser = subparsers.add_parser("x", **x_parser_args)
    shell_parser.set_defaults(func=rasa_x)

    arguments.set_x_arguments(shell_parser)


def _rasa_service(
    args: argparse.Namespace,
    endpoints: AvailableEndpoints,
    rasa_x_url: Optional[Text] = None,
    credentials_path: Optional[Text] = None,
):
    """Starts the Rasa application."""
    from rasa.core.run import serve_application

    # needs separate logging configuration as it is started in its own process
    logging.basicConfig(level=args.loglevel)
    io_utils.configure_colored_logging(args.loglevel)
    logging.getLogger("apscheduler").setLevel(logging.WARNING)

    if not credentials_path:
        credentials_path = _prepare_credentials_for_rasa_x(
            args.credentials, rasa_x_url=rasa_x_url
        )

    serve_application(
        endpoints=endpoints,
        port=args.port,
        credentials=credentials_path,
        cors=args.cors,
        auth_token=args.auth_token,
        enable_api=True,
        jwt_secret=args.jwt_secret,
        jwt_method=args.jwt_method,
        ssl_certificate=args.ssl_certificate,
        ssl_keyfile=args.ssl_keyfile,
        ssl_password=args.ssl_password,
    )


def _prepare_credentials_for_rasa_x(
    credentials_path: Optional[Text], rasa_x_url: Optional[Text] = None
) -> Text:
    credentials_path = cli_utils.get_validated_path(
        credentials_path, "credentials", DEFAULT_CREDENTIALS_PATH, True
    )
    if credentials_path:
        credentials = io_utils.read_config_file(credentials_path)
    else:
        credentials = {}

    # this makes sure the Rasa X is properly configured no matter what
    if rasa_x_url:
        credentials["rasa"] = {"url": rasa_x_url}
    dumped_credentials = yaml.dump(credentials, default_flow_style=False)
    tmp_credentials = io_utils.create_temporary_file(dumped_credentials, "yml")

    return tmp_credentials


def _overwrite_endpoints_for_local_x(
    endpoints: AvailableEndpoints, rasa_x_token: Text, rasa_x_url: Text
):
    from rasa.utils.endpoints import EndpointConfig
    import questionary

    # Checking if endpoint.yml has existing url and wait time values set, if so give warning we are overwriting
    # the endpoint.yml file.
    custom_wait_time_pulls = endpoints.model.kwargs.get("wait_time_between_pulls")
    custom_url = endpoints.model.url

    if custom_url and custom_url != rasa_x_url:
        logger.info(
            f"Ignoring url '{custom_url}' from 'endpoints.yml' and using "
            f"{rasa_x_url}/projects/default/models/tag/production instead"
        )

    endpoints.model = EndpointConfig(
        f"{rasa_x_url}/projects/default/models/tags/production",
        token=rasa_x_token,
        wait_time_between_pulls=custom_wait_time_pulls or 2,
    )

    overwrite_existing_event_broker = False
    if endpoints.event_broker and not _is_correct_event_broker(endpoints.event_broker):
        cli_utils.print_error(
            "Rasa X currently only supports a SQLite event broker with path '{}' "
            "when running locally. You can deploy Rasa X with Docker "
            "(https://rasa.com/docs/rasa-x/deploy/) if you want to use "
            "other event broker configurations.".format(DEFAULT_EVENTS_DB)
        )
        overwrite_existing_event_broker = questionary.confirm(
            "Do you want to continue with the default SQLite event broker?"
        ).ask()
{'fraction_non_alphanumeric': 0.07528257624226914, 'fraction_numerical': 0.00021326508850501172, 'mean_word_length': 3.57958984375, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '53100', 'n_tokens_mistral': 2957, 'n_tokens_neox': 2905, 'n_words': 651}