def send_to_missed_message_address(address, message):
    # type: (text_type, message.Message) -> None
    token = get_missed_message_token_from_address(address)
    key = missed_message_redis_key(token)
    result = redis_client.hmget(key, 'user_profile_id', 'recipient_id', 'subject')
    if not all(val is not None for val in result):
        raise ZulipEmailForwardError('Missing missed message address data')
    user_profile_id, recipient_id, subject = result

    user_profile = get_user_profile_by_id(user_profile_id)
    recipient = Recipient.objects.get(id=recipient_id)
    display_recipient = get_display_recipient(recipient)

    # Testing with basestring so we don't depend on the list return type from
    # get_display_recipient
    if not isinstance(display_recipient, six.string_types):
        recipient_str = ','.join([user['email'] for user in display_recipient])
    else:
        recipient_str = display_recipient

    body = filter_footer(extract_body(message))
    body += extract_and_upload_attachments(message, user_profile.realm)
    if not body:
        body = '(No email body)'

    if recipient.type == Recipient.STREAM:
        recipient_type_name = 'stream'
    else:
        recipient_type_name = 'private'

    internal_send_message(user_profile.email, recipient_type_name,
                          recipient_str, subject, body)

def send_to_missed_message_address(address, message):
    # type: (text_type, message.Message) -> None
    token = get_missed_message_token_from_address(address)
    key = missed_message_redis_key(token)
    result = redis_client.hmget(key, "user_profile_id", "recipient_id", "subject")
    if not all(val is not None for val in result):
        raise ZulipEmailForwardError("Missing missed message address data")
    user_profile_id, recipient_id, subject = result

    user_profile = get_user_profile_by_id(user_profile_id)
    recipient = Recipient.objects.get(id=recipient_id)
    display_recipient = get_display_recipient(recipient)

    # Testing with basestring so we don't depend on the list return type from
    # get_display_recipient
    if not isinstance(display_recipient, six.string_types):
        recipient_str = ",".join([user["email"] for user in display_recipient])
    else:
        recipient_str = display_recipient

    body = filter_footer(extract_body(message))
    body += extract_and_upload_attachments(message, user_profile.realm)
    if not body:
        body = "(No email body)"

    if recipient.type == Recipient.STREAM:
        recipient_type_name = "stream"
    else:
        recipient_type_name = "private"

    internal_send_message(user_profile.email, recipient_type_name, recipient_str, subject, body)
    logging.info("Successfully processed email from %s to %s" % (user_profile.email, recipient_str))

def zulip_server_error(report: Dict[str, Any]) -> None:
    email_subject = '%(node)s: %(message)s' % (report)

    logger_str = logger_repr(report)
    user_info = user_info_str(report)
    deployment = deployment_repr(report)

    if report['has_request']:
        request_repr = (
            "Request info:\n~~~~\n"
            "- path: %(path)s\n"
            "- %(method)s: %(data)s\n") % (report)
        for field in ["REMOTE_ADDR", "QUERY_STRING", "SERVER_NAME"]:
            val = report.get(field.lower())
            if field == "QUERY_STRING":
                val = clean_data_from_query_parameters(str(val))
            request_repr += "- %s: \"%s\"\n" % (field, val)
        request_repr += "~~~~"
    else:
        request_repr = "Request info: none"

    message = ("%s\nError generated by %s\n\n~~~~ pytb\n%s\n\n~~~~\n%s\n%s"
               % (logger_str, user_info, report['stack_trace'], deployment, request_repr))

    realm = get_system_bot(settings.ERROR_BOT).realm
    internal_send_message(realm, settings.ERROR_BOT, "stream", "errors",
                          format_email_subject(email_subject), message)

def send_to_missed_message_address(address, message):
    # type: (Text, message.Message) -> None
    token = get_missed_message_token_from_address(address)
    key = missed_message_redis_key(token)
    result = redis_client.hmget(key, 'user_profile_id', 'recipient_id', 'subject')
    if not all(val is not None for val in result):
        raise ZulipEmailForwardError('Missing missed message address data')
    user_profile_id, recipient_id, subject_b = result  # type: (bytes, bytes, bytes)

    user_profile = get_user_profile_by_id(user_profile_id)
    recipient = Recipient.objects.get(id=recipient_id)
    display_recipient = get_display_recipient(recipient)

    # Testing with basestring so we don't depend on the list return type from
    # get_display_recipient
    if not isinstance(display_recipient, str):
        recipient_str = u','.join([user['email'] for user in display_recipient])
    else:
        recipient_str = display_recipient

    body = construct_zulip_body(message, user_profile.realm)

    if recipient.type == Recipient.STREAM:
        recipient_type_name = 'stream'
    else:
        recipient_type_name = 'private'

    internal_send_message(user_profile.realm, user_profile.email,
                          recipient_type_name, recipient_str,
                          subject_b.decode('utf-8'), body)
    logger.info("Successfully processed email from %s to %s" % (
        user_profile.email, recipient_str))

def deliver_feedback_by_zulip(message: Mapping[str, Any]) -> None:
    subject = "%s" % (message["sender_email"],)

    if len(subject) > 60:
        subject = subject[:57].rstrip() + "..."

    content = ''
    sender_email = message['sender_email']

    # We generate ticket numbers if it's been more than a few minutes
    # since their last message.  This avoids some noise when people use
    # enter-send.
    need_ticket = has_enough_time_expired_since_last_message(sender_email, 180)

    if need_ticket:
        ticket_number = get_ticket_number()
        content += '\n~~~'
        content += '\nticket Z%03d (@support please ack)' % (ticket_number,)
        content += '\nsender: %s' % (message['sender_full_name'],)
        content += '\nemail: %s' % (sender_email,)
        if 'sender_realm_str' in message:
            content += '\nrealm: %s' % (message['sender_realm_str'],)
        content += '\n~~~'
        content += '\n\n'

    content += message['content']

    user_profile = get_system_bot(settings.FEEDBACK_BOT)
    internal_send_message(user_profile.realm, settings.FEEDBACK_BOT,
                          "stream", settings.FEEDBACK_STREAM, subject, content)

def send_zulip(stream, topic, content):
    internal_send_message(
            settings.EMAIL_GATEWAY_BOT,
            "stream",
            stream.name,
            topic[:60],
            content[:2000],
            stream.realm)

 def send_message(self, message):
     # type: (Dict[str, Any]) -> None
     if self._rate_limit.is_legal():
         internal_send_message(realm=self.user_profile.realm, sender_email=message['sender'],
                               recipient_type_name=message['type'], recipients=message['to'],
                               subject=message['subject'], content=message['content'])
     else:
         self._rate_limit.show_error_and_exit()

def send_zulip(sender: str, stream: Stream, topic: str, content: str) -> None:
    internal_send_message(
        stream.realm,
        sender,
        "stream",
        stream.name,
        topic[:MAX_TOPIC_NAME_LENGTH],
        content[:MAX_MESSAGE_LENGTH],
        email_gateway=True)

def send_zulip(sender: Text, stream: Stream, topic: Text, content: Text) -> None:
    internal_send_message(
        stream.realm,
        sender,
        "stream",
        stream.name,
        topic[:60],
        content[:2000],
        email_gateway=True)

def send_zulip(sender, stream, topic, content):
    # type: (text_type, Stream, text_type, text_type) -> None
    internal_send_message(
            sender,
            "stream",
            stream.name,
            topic[:60],
            content[:2000],
            stream.realm)

def send_zulip(sender: str, stream: Stream, topic: str, content: str) -> None:
    internal_send_message(
        stream.realm,
        sender,
        "stream",
        stream.name,
        truncate_topic(topic),
        truncate_body(content),
        email_gateway=True)

def send_zulip(sender, stream, topic, content):
    # type: (Text, Stream, Text, Text) -> None
    internal_send_message(
        stream.realm,
        sender,
        "stream",
        stream.name,
        topic[:60],
        content[:2000])

 def send_message(self, message):
     # type: (Dict[str, Any]) -> None
     if self._rate_limit.is_legal():
         recipients = message['to'] if message['type'] == 'stream' else ','.join(message['to'])
         internal_send_message(realm=self.user_profile.realm, sender_email=self.user_profile.email,
                               recipient_type_name=message['type'], recipients=recipients,
                               topic_name=message.get('subject', None), content=message['content'])
     else:
         self._rate_limit.show_error_and_exit()

def do_convert(md, realm_domain=None, message=None, possible_words=None):
    # type: (markdown.Markdown, Optional[text_type], Optional[Message], Optional[Set[text_type]]) -> Optional[text_type]
    """Convert Markdown to HTML, with Zulip-specific settings and hacks."""
    from zerver.models import get_active_user_dicts_in_realm, UserProfile

    if message:
        maybe_update_realm_filters(message.get_realm().domain)

    if realm_domain in md_engines:
        _md_engine = md_engines[realm_domain]
    else:
        _md_engine = md_engines["default"]
    # Reset the parser; otherwise it will get slower over time.
    _md_engine.reset()

    global current_message
    current_message = message

    # Pre-fetch data from the DB that is used in the bugdown thread
    global db_data
    if message:
        realm_users = get_active_user_dicts_in_realm(message.get_realm())

        if possible_words is None:
            possible_words = set() # Set[text_type]

        db_data = {'possible_words':    possible_words,
                   'full_names':        dict((user['full_name'].lower(), user) for user in realm_users),
                   'short_names':       dict((user['short_name'].lower(), user) for user in realm_users),
                   'emoji':             message.get_realm().get_emoji()}

    try:
        # Spend at most 5 seconds rendering.
        # Sometimes Python-Markdown is really slow; see
        # https://trac.zulip.net/ticket/345
        return timeout(5, _md_engine.convert, md)
    except:
        from zerver.lib.actions import internal_send_message

        cleaned = _sanitize_for_log(md)

        # Output error to log as well as sending a zulip and email
        log_bugdown_error('Exception in Markdown parser: %sInput (sanitized) was: %s'
            % (traceback.format_exc(), cleaned))
        subject = "Markdown parser failure on %s" % (platform.node(),)
        if settings.ERROR_BOT is not None:
            internal_send_message(settings.ERROR_BOT, "stream",
                    "errors", subject, "Markdown parser failed, email sent with details.")
        mail.mail_admins(subject, "Failed message: %s\n\n%s\n\n" % (
                                    cleaned, traceback.format_exc()),
                         fail_silently=False)
        raise BugdownRenderingException()
    finally:
        current_message = None
        db_data = None

def zulip_browser_error(report: Dict[str, Any]) -> None:
    subject = "JS error: %s" % (report['user_email'],)

    user_info = user_info_str(report)

    body = "User: %s\n" % (user_info,)
    body += ("Message: %(message)s\n"
             % (report))

    realm = get_system_bot(settings.ERROR_BOT).realm
    internal_send_message(realm, settings.ERROR_BOT,
                          "stream", "errors", format_subject(subject), body)

def highlight_html_differences(s1, s2):
    # type: (Text, Text) -> Text
    differ = diff_match_patch()
    ops = differ.diff_main(s1, s2)
    differ.diff_cleanupSemantic(ops)
    retval = u''
    in_tag = False

    idx = 0
    while idx < len(ops):
        op, text = ops[idx]
        next_op = None
        if idx != len(ops) - 1:
            next_op, next_text = ops[idx + 1]
        if op == diff_match_patch.DIFF_DELETE and next_op == diff_match_patch.DIFF_INSERT:
            # Replace operation
            chunks, in_tag = chunkize(next_text, in_tag)
            retval += highlight_chunks(chunks, highlight_replaced)
            idx += 1
        elif op == diff_match_patch.DIFF_INSERT and next_op == diff_match_patch.DIFF_DELETE:
            # Replace operation
            # I have no idea whether diff_match_patch generates inserts followed
            # by deletes, but it doesn't hurt to handle them
            chunks, in_tag = chunkize(text, in_tag)
            retval += highlight_chunks(chunks, highlight_replaced)
            idx += 1
        elif op == diff_match_patch.DIFF_DELETE:
            retval += highlight_deleted('&nbsp;')
        elif op == diff_match_patch.DIFF_INSERT:
            chunks, in_tag = chunkize(text, in_tag)
            retval += highlight_chunks(chunks, highlight_inserted)
        elif op == diff_match_patch.DIFF_EQUAL:
            chunks, in_tag = chunkize(text, in_tag)
            retval += text
        idx += 1

    if not verify_html(retval):
        from zerver.lib.actions import internal_send_message
        from zerver.models import get_user_profile_by_email
        # We probably want more information here
        logging.getLogger('').error('HTML diff produced mal-formed HTML')

        if settings.ERROR_BOT is not None:
            subject = "HTML diff failure on %s" % (platform.node(),)
            realm = get_user_profile_by_email(settings.ERROR_BOT).realm
            internal_send_message(realm, settings.ERROR_BOT, "stream",
                                  "errors", subject, "HTML diff produced malformed HTML")
        return s2

    return retval

def json_tutorial_send_message(request, user_profile, type=REQ(validator=check_string),
                               recipient=REQ(validator=check_string), topic=REQ(validator=check_string),
                               content=REQ(validator=check_string)):
    """
    This function, used by the onboarding tutorial, causes the Tutorial Bot to
    send you the message you pass in here. (That way, the Tutorial Bot's
    messages to you get rendered by the server and therefore look like any other
    message.)
    """
    sender_name = "[email protected]"
    if type == 'stream':
        internal_send_message(sender_name, "stream", recipient, topic, content,
                              realm=user_profile.realm)
        return json_success()
    # For now, there are no PM cases.
    return json_error('Bad data passed in to tutorial_send_message')

    def process_one_batch(self):
        slow_queries = self.q.drain_queue("slow_queries", json=True)

        if settings.ERROR_BOT is None:
            return

        if len(slow_queries) > 0:
            topic = "%s: slow queries" % (settings.EXTERNAL_HOST,)

            content = ""
            for query in slow_queries:
                content += "    %s\n" % (query,)

            internal_send_message(settings.ERROR_BOT, "stream", "logs", topic, content)

        reset_queries()

def highlight_html_differences(s1, s2):
    # type: (Text, Text) -> Text
    differ = diff_match_patch()
    ops = differ.diff_main(s1, s2)
    differ.diff_cleanupSemantic(ops)
    retval = u''
    in_tag = False

    idx = 0
    while idx < len(ops):
        op, text = ops[idx]
        text = check_tags(text)
        if idx != 0:
            prev_op, prev_text = ops[idx - 1]
            prev_text = check_tags(prev_text)
            # Remove visual offset from editing newlines
            if '<p><br>' in text:
                text = text.replace('<p><br>', '<p>')
            elif prev_text.endswith('<p>') and text.startswith('<br>'):
                text = text[4:]
        if op == diff_match_patch.DIFF_DELETE:
            chunks, in_tag = chunkize(text, in_tag)
            retval += highlight_chunks(chunks, highlight_deleted)
        elif op == diff_match_patch.DIFF_INSERT:
            chunks, in_tag = chunkize(text, in_tag)
            retval += highlight_chunks(chunks, highlight_inserted)
        elif op == diff_match_patch.DIFF_EQUAL:
            chunks, in_tag = chunkize(text, in_tag)
            retval += text
        idx += 1

    if not verify_html(retval):
        from zerver.lib.actions import internal_send_message
        from zerver.models import get_system_bot
        # We probably want more information here
        logging.getLogger('').error('HTML diff produced mal-formed HTML')

        if settings.ERROR_BOT is not None:
            subject = "HTML diff failure on %s" % (platform.node(),)
            realm = get_system_bot(settings.ERROR_BOT).realm
            internal_send_message(realm, settings.ERROR_BOT, "stream",
                                  "errors", subject, "HTML diff produced malformed HTML")
        return s2

    return retval

    def consume_batch(self, slow_queries):
        # type: (List[Dict[str, Any]]) -> None
        for query in slow_queries:
            logging.info("Slow query: %s" % (query))

        if settings.ERROR_BOT is None:
            return

        if len(slow_queries) > 0:
            topic = "%s: slow queries" % (settings.EXTERNAL_HOST,)

            content = ""
            for query in slow_queries:
                content += "    %s\n" % (query,)

            error_bot_realm = get_system_bot(settings.ERROR_BOT).realm
            internal_send_message(error_bot_realm, settings.ERROR_BOT,
                                  "stream", "logs", topic, content)