<commit_before>from typing import Any, AnyStr, Iterable, Dict, Tuple, Callable, Mapping, Optional

import requests
import json
import sys
import inspect
import logging
import re
import urllib
from functools import reduce
from requests import Response

from django.utils.translation import ugettext as _

from zerver.models import Realm, UserProfile, get_user_profile_by_id, get_client, \
    GENERIC_INTERFACE, Service, SLACK_INTERFACE, email_to_domain, get_service_profile
from zerver.lib.actions import check_send_message
from zerver.lib.queue import retry_event
from zerver.lib.url_encoding import encode_stream
from zerver.lib.validator import check_dict, check_string
from zerver.decorator import JsonableError

class OutgoingWebhookServiceInterface:

    def __init__(self, base_url: str, token: str, user_profile: UserProfile, service_name: str) -> None:
        self.base_url = base_url  # type: str
        self.token = token  # type: str
        self.user_profile = user_profile  # type: UserProfile
        self.service_name = service_name  # type: str

    # Given an event that triggers an outgoing webhook operation, returns:
    # - The REST operation that should be performed
    # - The body of the request
    #
    # The REST operation is a dictionary with the following keys:
    # - method
    # - base_url
    # - relative_url_path
    # - request_kwargs
    def process_event(self, event: Dict[str, Any]) -> Tuple[Dict[str, Any], Any]:
        raise NotImplementedError()

    # Given a successful outgoing webhook REST operation, returns two-element tuple
    # whose left-hand value contains a success message
    # to sent back to the user (or None if no success message should be sent)
    # and right-hand value contains a failure message
    # to sent back to the user (or None if no failure message should be sent)
    def process_success(self, response: Response,
                        event: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
        raise NotImplementedError()

class GenericOutgoingWebhookService(OutgoingWebhookServiceInterface):

    def process_event(self, event: Dict[str, Any]) -> Tuple[Dict[str, Any], Any]:
        rest_operation = {'method': 'POST',
                          'relative_url_path': '',
                          'base_url': self.base_url,
                          'request_kwargs': {}}
        request_data = {"data": event['command'],
                        "message": event['message'],
                        "bot_email": self.user_profile.email,
                        "token": self.token,
                        "trigger": event['trigger']}
        return rest_operation, json.dumps(request_data)

    def process_success(self, response: Response,
                        event: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
        response_json = json.loads(response.text)

        if "response_not_required" in response_json and response_json['response_not_required']:
            return None, None
        if "response_string" in response_json:
            return str(response_json['response_string']), None
        else:
            return None, None

class SlackOutgoingWebhookService(OutgoingWebhookServiceInterface):

    def process_event(self, event: Dict[str, Any]) -> Tuple[Dict[str, Any], Any]:
        rest_operation = {'method': 'POST',
                          'relative_url_path': '',
                          'base_url': self.base_url,
                          'request_kwargs': {}}

        if event['message']['type'] == 'private':
            failure_message = "Slack outgoing webhooks don't support private messages."
            fail_with_message(event, failure_message)
            return None, None

        request_data = [("token", self.token),
                        ("team_id", event['message']['sender_realm_str']),
                        ("team_domain", email_to_domain(event['message']['sender_email'])),
                        ("channel_id", event['message']['stream_id']),
                        ("channel_name", event['message']['display_recipient']),
                        ("timestamp", event['message']['timestamp']),
                        ("user_id", event['message']['sender_id']),
                        ("user_name", event['message']['sender_full_name']),
                        ("text", event['command']),
                        ("trigger_word", event['trigger']),
                        ("service_id", event['user_profile_id']),
                        ]

        return rest_operation, request_data

    def process_success(self, response: Response,
                        event: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
        response_json = json.loads(response.text)
        if "text" in response_json:
            return response_json["text"], None
        else:
            return None, None

AVAILABLE_OUTGOING_WEBHOOK_INTERFACES = {
    GENERIC_INTERFACE: GenericOutgoingWebhookService,
    SLACK_INTERFACE: SlackOutgoingWebhookService,
}   # type: Dict[str, Any]

def get_service_interface_class(interface: str) -> Any:
    if interface is None or interface not in AVAILABLE_OUTGOING_WEBHOOK_INTERFACES:
        return AVAILABLE_OUTGOING_WEBHOOK_INTERFACES[GENERIC_INTERFACE]
    else:
        return AVAILABLE_OUTGOING_WEBHOOK_INTERFACES[interface]

def get_outgoing_webhook_service_handler(service: Service) -> Any:

    service_interface_class = get_service_interface_class(service.interface_name())
    service_interface = service_interface_class(base_url=service.base_url,
                                                token=service.token,
                                                user_profile=service.user_profile,
                                                service_name=service.name)
    return service_interface

def send_response_message(bot_id: str, message: Dict[str, Any], response_message_content: str) -> None:
    recipient_type_name = message['type']
    bot_user = get_user_profile_by_id(bot_id)
    realm = bot_user.realm
    client = get_client('OutgoingWebhookResponse')

    if recipient_type_name == 'stream':
        recipients = [message['display_recipient']]
        check_send_message(bot_user, client, recipient_type_name, recipients,
                           message['subject'], response_message_content, realm)
    elif recipient_type_name == 'private':
        recipients = [recipient['email'] for recipient in message['display_recipient']]
        check_send_message(bot_user, client, recipient_type_name, recipients,
                           None, response_message_content, realm)
    else:
        raise JsonableError(_("Invalid message type"))

def succeed_with_message(event: Dict[str, Any], success_message: str) -> None:
    success_message = "Success! " + success_message
    send_response_message(event['user_profile_id'], event['message'], success_message)

def fail_with_message(event: Dict[str, Any], failure_message: str) -> None:
    failure_message = "Failure! " + failure_message
    send_response_message(event['user_profile_id'], event['message'], failure_message)

def get_message_url(event: Dict[str, Any], request_data: Dict[str, Any]) -> str:
    bot_user = get_user_profile_by_id(event['user_profile_id'])
    message = event['message']
    if message['type'] == 'stream':
        stream_url_frag = encode_stream(message.get('stream_id'), message['display_recipient'])
        message_url = ("%(server)s/#narrow/stream/%(stream)s/subject/%(subject)s/near/%(id)s"
                       % {'server': bot_user.realm.uri,
                          'stream': stream_url_frag,
                          'subject': message['subject'],<commit_msg>minor: Rename var to message_type.<commit_after>from typing import Any, AnyStr, Iterable, Dict, Tuple, Callable, Mapping, Optional

import requests
import json
import sys
import inspect
import logging
import re
import urllib
from functools import reduce
from requests import Response

from django.utils.translation import ugettext as _

from zerver.models import Realm, UserProfile, get_user_profile_by_id, get_client, \
    GENERIC_INTERFACE, Service, SLACK_INTERFACE, email_to_domain, get_service_profile
from zerver.lib.actions import check_send_message
from zerver.lib.queue import retry_event
from zerver.lib.url_encoding import encode_stream
from zerver.lib.validator import check_dict, check_string
from zerver.decorator import JsonableError

class OutgoingWebhookServiceInterface:

    def __init__(self, base_url: str, token: str, user_profile: UserProfile, service_name: str) -> None:
        self.base_url = base_url  # type: str
        self.token = token  # type: str
        self.user_profile = user_profile  # type: UserProfile
        self.service_name = service_name  # type: str

    # Given an event that triggers an outgoing webhook operation, returns:
    # - The REST operation that should be performed
    # - The body of the request
    #
    # The REST operation is a dictionary with the following keys:
    # - method
    # - base_url
    # - relative_url_path
    # - request_kwargs
    def process_event(self, event: Dict[str, Any]) -> Tuple[Dict[str, Any], Any]:
        raise NotImplementedError()

    # Given a successful outgoing webhook REST operation, returns two-element tuple
    # whose left-hand value contains a success message
    # to sent back to the user (or None if no success message should be sent)
    # and right-hand value contains a failure message
    # to sent back to the user (or None if no failure message should be sent)
    def process_success(self, response: Response,
                        event: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
        raise NotImplementedError()

class GenericOutgoingWebhookService(OutgoingWebhookServiceInterface):

    def process_event(self, event: Dict[str, Any]) -> Tuple[Dict[str, Any], Any]:
        rest_operation = {'method': 'POST',
                          'relative_url_path': '',
                          'base_url': self.base_url,
                          'request_kwargs': {}}
        request_data = {"data": event['command'],
                        "message": event['message'],
                        "bot_email": self.user_profile.email,
                        "token": self.token,
                        "trigger": event['trigger']}
        return rest_operation, json.dumps(request_data)

    def process_success(self, response: Response,
                        event: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
        response_json = json.loads(response.text)

        if "response_not_required" in response_json and response_json['response_not_required']:
            return None, None
        if "response_string" in response_json:
            return str(response_json['response_string']), None
        else:
            return None, None

class SlackOutgoingWebhookService(OutgoingWebhookServiceInterface):

    def process_event(self, event: Dict[str, Any]) -> Tuple[Dict[str, Any], Any]:
        rest_operation = {'method': 'POST',
                          'relative_url_path': '',
                          'base_url': self.base_url,
                          'request_kwargs': {}}

        if event['message']['type'] == 'private':
            failure_message = "Slack outgoing webhooks don't support private messages."
            fail_with_message(event, failure_message)
            return None, None

        request_data = [("token", self.token),
                        ("team_id", event['message']['sender_realm_str']),
                        ("team_domain", email_to_domain(event['message']['sender_email'])),
                        ("channel_id", event['message']['stream_id']),
                        ("channel_name", event['message']['display_recipient']),
                        ("timestamp", event['message']['timestamp']),
                        ("user_id", event['message']['sender_id']),
                        ("user_name", event['message']['sender_full_name']),
                        ("text", event['command']),
                        ("trigger_word", event['trigger']),
                        ("service_id", event['user_profile_id']),
                        ]

        return rest_operation, request_data

    def process_success(self, response: Response,
                        event: Dict[str, Any]) -> Tuple[Optional[str], Optional[str]]:
        response_json = json.loads(response.text)
        if "text" in response_json:
            return response_json["text"], None
        else:
            return None, None

AVAILABLE_OUTGOING_WEBHOOK_INTERFACES = {
    GENERIC_INTERFACE: GenericOutgoingWebhookService,
    SLACK_INTERFACE: SlackOutgoingWebhookService,
}   # type: Dict[str, Any]

def get_service_interface_class(interface: str) -> Any:
    if interface is None or interface not in AVAILABLE_OUTGOING_WEBHOOK_INTERFACES:
        return AVAILABLE_OUTGOING_WEBHOOK_INTERFACES[GENERIC_INTERFACE]
    else:
        return AVAILABLE_OUTGOING_WEBHOOK_INTERFACES[interface]

def get_outgoing_webhook_service_handler(service: Service) -> Any:

    service_interface_class = get_service_interface_class(service.interface_name())
    service_interface = service_interface_class(base_url=service.base_url,
                                                token=service.token,
                                                user_profile=service.user_profile,
                                                service_name=service.name)
    return service_interface

def send_response_message(bot_id: str, message: Dict[str, Any], response_message_content: str) -> None:
    message_type = message['type']
    bot_user = get_user_profile_by_id(bot_id)
    realm = bot_user.realm
    client = get_client('OutgoingWebhookResponse')

    if message_type == 'stream':
        recipients = [message['display_recipient']]
        check_send_message(bot_user, client, message_type, recipients,
                           message['subject'], response_message_content, realm)
    elif message_type == 'private':
        recipients = [recipient['email'] for recipient in message['display_recipient']]
        check_send_message(bot_user, client, message_type, recipients,
                           None, response_message_content, realm)
    else:
        raise JsonableError(_("Invalid message type"))

def succeed_with_message(event: Dict[str, Any], success_message: str) -> None:
    success_message = "Success! " + success_message
    send_response_message(event['user_profile_id'], event['message'], success_message)

def fail_with_message(event: Dict[str, Any], failure_message: str) -> None:
    failure_message = "Failure! " + failure_message
    send_response_message(event['user_profile_id'], event['message'], failure_message)

def get_message_url(event: Dict[str, Any], request_data: Dict[str, Any]) -> str:
    bot_user = get_user_profile_by_id(event['user_profile_id'])
    message = event['message']
    if message['type'] == 'stream':
        stream_url_frag = encode_stream(message.get('stream_id'), message['display_recipient'])
        message_url = ("%(server)s/#narrow/stream/%(stream)s/subject/%(subject)s/near/%(id)s"
                       % {'server': bot_user.realm.uri,
                          'stream': stream_url_frag,
                          'subject': message['subject'],
{'fraction_non_alphanumeric': 0.10197710718002082, 'fraction_numerical': 0.0, 'mean_word_length': 2.770720941638058, 'pattern_counts': {'":': 10, '<': 3, '<?xml version=': 0, '>': 29, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6589047', 'n_tokens_mistral': 4240, 'n_tokens_neox': 4135, 'n_words': 993}