def fetch_user_profile_cross_realm(response: TableData, config: Config, context: Context) -> None:
    realm = context['realm']

    if realm.string_id == settings.SYSTEM_BOT_REALM:
        response['zerver_userprofile_crossrealm'] = []
    else:
        response['zerver_userprofile_crossrealm'] = [dict(email=x.email, id=x.id) for x in [
            get_system_bot(settings.NOTIFICATION_BOT),
            get_system_bot(settings.EMAIL_GATEWAY_BOT),
            get_system_bot(settings.WELCOME_BOT),
        ]]

    def handle(self, *args: Any, **options: Any) -> None:
        if Realm.objects.count() > 0:
            print("Database already initialized; doing nothing.")
            return
        realm = Realm.objects.create(string_id=settings.INTERNAL_BOT_DOMAIN.split('.')[0])

        names = [(settings.FEEDBACK_BOT_NAME, settings.FEEDBACK_BOT)]
        create_users(realm, names, bot_type=UserProfile.DEFAULT_BOT)

        get_client("website")
        get_client("API")

        internal_bots = [(bot['name'], bot['email_template'] % (settings.INTERNAL_BOT_DOMAIN,))
                         for bot in settings.INTERNAL_BOTS]
        create_users(realm, internal_bots, bot_type=UserProfile.DEFAULT_BOT)
        # Set the owners for these bots to the bots themselves
        bots = UserProfile.objects.filter(email__in=[bot_info[1] for bot_info in internal_bots])
        for bot in bots:
            bot.bot_owner = bot
            bot.save()

        # Initialize the email gateway bot as an API Super User
        email_gateway_bot = get_system_bot(settings.EMAIL_GATEWAY_BOT)
        email_gateway_bot.is_api_super_user = True
        email_gateway_bot.save()

        self.stdout.write("Successfully populated database with initial data.\n")
        self.stdout.write("Please run ./manage.py generate_realm_creation_link "
                          "to generate link for creating organization")

def extract_and_upload_attachments(message: message.Message, realm: Realm) -> str:
    user_profile = get_system_bot(settings.EMAIL_GATEWAY_BOT)
    attachment_links = []

    payload = message.get_payload()
    if not isinstance(payload, list):
        # This is not a multipart message, so it can't contain attachments.
        return ""

    for part in payload:
        content_type = part.get_content_type()
        filename = part.get_filename()
        if filename:
            attachment = part.get_payload(decode=True)
            if isinstance(attachment, bytes):
                s3_url = upload_message_file(filename, len(attachment), content_type,
                                             attachment,
                                             user_profile,
                                             target_realm=realm)
                formatted_link = "[%s](%s)" % (filename, s3_url)
                attachment_links.append(formatted_link)
            else:
                logger.warning("Payload is not bytes (invalid attachment %s in message from %s)." %
                               (filename, message.get("From")))

    return "\n".join(attachment_links)

def report_to_zulip(error_message: str) -> None:
    if settings.ERROR_BOT is None:
        return
    error_bot = get_system_bot(settings.ERROR_BOT)
    error_stream = Stream.objects.get(name="errors", realm=error_bot.realm)
    send_zulip(settings.ERROR_BOT, error_stream, "email mirror error",
               """~~~\n%s\n~~~""" % (error_message,))

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 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_initial_pms(user):
    # type: (UserProfile) -> None
    content = """Welcome to Zulip!

This is a great place to test formatting, sending, and editing messages.
Click anywhere on this message to reply. A compose box will open at the bottom of the screen."""

    internal_send_private_message(user.realm, get_system_bot(settings.WELCOME_BOT),
                                  user.email, content)

 def test_stream_error_pm_to_bot_owner(self) -> None:
     # Note taht this is really just a test for check_send_webhook_message
     self.STREAM_NAME = 'nonexistent'
     self.url = self.build_webhook_url()
     notification_bot = get_system_bot(settings.NOTIFICATION_BOT)
     expected_message = "Hi there! We thought you'd like to know that your bot **Zulip Webhook Bot** just tried to send a message to stream `nonexistent`, but that stream does not yet exist. To create it, click the gear in the left-side stream list."
     self.send_and_test_private_message('goodbye', expected_message=expected_message,
                                        content_type='application/x-www-form-urlencoded',
                                        sender=notification_bot)

def export_avatars_from_local(realm: Realm, local_dir: Path, output_dir: Path) -> None:

    count = 0
    records = []

    users = list(UserProfile.objects.filter(realm=realm))
    users += [
        get_system_bot(settings.NOTIFICATION_BOT),
        get_system_bot(settings.EMAIL_GATEWAY_BOT),
        get_system_bot(settings.WELCOME_BOT),
    ]
    for user in users:
        if user.avatar_source == UserProfile.AVATAR_FROM_GRAVATAR:
            continue

        avatar_path = user_avatar_path_from_ids(user.id, realm.id)
        wildcard = os.path.join(local_dir, avatar_path + '.*')

        for local_path in glob.glob(wildcard):
            logging.info('Copying avatar file for user %s from %s' % (
                user.email, local_path))
            fn = os.path.relpath(local_path, local_dir)
            output_path = os.path.join(output_dir, fn)
            os.makedirs(str(os.path.dirname(output_path)), exist_ok=True)
            subprocess.check_call(["cp", "-a", str(local_path), str(output_path)])
            stat = os.stat(local_path)
            record = dict(realm_id=realm.id,
                          user_profile_id=user.id,
                          user_profile_email=user.email,
                          s3_path=fn,
                          path=fn,
                          size=stat.st_size,
                          last_modified=stat.st_mtime,
                          content_type=None)
            records.append(record)

            count += 1

            if (count % 100 == 0):
                logging.info("Finished %s" % (count,))

    with open(os.path.join(output_dir, "records.json"), "w") as records_file:
        ujson.dump(records, records_file, indent=4)

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_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 send_initial_pms(user):
    # type: (UserProfile) -> None
    content = (
        "Hello, and welcome to Zulip!\n\nThis is a private message from me, Welcome Bot. "
        "Here are some tips to get you started:\n"
        "* Download our [Desktop and mobile apps](/apps)\n"
        "* Customize your account and notifications on your [Settings page](#settings).\n"
        "* Check out our !modal_link(#keyboard-shortcuts, Keyboard shortcuts)\n\n"
        "The most important shortcut is `r` or `Enter` to reply.\n\n"
        "Practice sending a few messages by replying to this conversation. If you're not into "
        "keyboards, that's okay too; clicking anywhere on this message will also do the trick!")

    internal_send_private_message(user.realm, get_system_bot(settings.WELCOME_BOT),
                                  user.email, content)

def send_initial_realm_messages(realm):
    # type: (Realm) -> None
    welcome_bot = get_system_bot(settings.WELCOME_BOT)
    # Make sure each stream created in the realm creation process has at least one message below
    # Order corresponds to the ordering of the streams on the left sidebar, to make the initial Home
    # view slightly less overwhelming
    welcome_messages = [
        {'stream': Realm.DEFAULT_NOTIFICATION_STREAM_NAME,
         'topic': "welcome",
         'content': "This is a message on stream `%s` with the topic `welcome`. We'll use this stream "
         "for system-generated notifications." % (Realm.DEFAULT_NOTIFICATION_STREAM_NAME,)},
        {'stream': "core team",
         'topic': "private streams",
         'content': "This is a private stream. Only admins and people you invite "
         "to the stream will be able to see that this stream exists."},
        {'stream': "general",
         'topic': "welcome",
         'content': "Welcome to #**general**."},
        {'stream': "new members",
         'topic': "onboarding",
         'content': "A #**new members** stream is great for onboarding new members.\n\nIf you're "
         "reading this and aren't the first person here, introduce yourself in a new thread "
         "using your name as the topic! Type `c` or click on `New Topic` at the bottom of the "
         "screen to start a new topic."},
        {'stream': "zulip",
         'topic': "topic demonstration",
         'content': "Here is a message in one topic. Replies to this message will go to this topic."},
        {'stream': "zulip",
         'topic': "topic demonstration",
         'content': "A second message in this topic. With [turtles](/static/images/cute/turtle.png)!"},
        {'stream': "zulip",
         'topic': "second topic",
         'content': "This is a message in a second topic.\n\nTopics are similar to email subjects, "
         "in that each conversation should get its own topic. Keep them short, though; one "
         "or two words will do it!"},
    ]  # type: List[Dict[str, Text]]
    messages = [internal_prep_stream_message(
        realm, welcome_bot,
        message['stream'], message['topic'], message['content']) for message in welcome_messages]
    message_ids = do_send_messages(messages)

    # We find the one of our just-sent messages with turtle.png in it,
    # and react to it.  This is a bit hacky, but works and is kinda a
    # 1-off thing.
    turtle_message = Message.objects.get(
        id__in=message_ids,
        subject='topic demonstration',
        content__icontains='cute/turtle.png')
    do_add_reaction_legacy(welcome_bot, turtle_message, 'turtle')

    def test_cross_realm_file_access(self):
        # type: () -> None

        def create_user(email, realm_id):
            # type: (Text, Text) -> UserProfile
            self.register(email, 'test', subdomain=realm_id)
            return get_user(email, get_realm(realm_id))

        test_subdomain = "uploadtest.example.com"
        user1_email = '[email protected]e.com'
        user2_email = '[email protected]'
        user3_email = '[email protected]'

        r1 = Realm.objects.create(string_id=test_subdomain, invite_required=False)
        RealmDomain.objects.create(realm=r1, domain=test_subdomain)

        create_user(user1_email, test_subdomain)
        create_user(user2_email, 'zulip')
        create_user(user3_email, test_subdomain)

        # Send a message from @zulip.com -> @uploadtest.example.com
        self.login(user2_email, 'test')
        fp = StringIO("zulip!")
        fp.name = "zulip.txt"
        result = self.client_post("/json/user_uploads", {'file': fp})
        uri = result.json()['uri']
        fp_path_id = re.sub('/user_uploads/', '', uri)
        body = "First message ...[zulip.txt](http://localhost:9991/user_uploads/" + fp_path_id + ")"
        with self.settings(CROSS_REALM_BOT_EMAILS = set((user2_email, user3_email))):
            internal_send_private_message(
                realm=r1,
                sender=get_system_bot(user2_email),
                recipient_user=get_user(user1_email, r1),
                content=body,
            )

        self.login(user1_email, 'test')
        response = self.client_get(uri, subdomain=test_subdomain)
        self.assertEqual(response.status_code, 200)
        data = b"".join(response.streaming_content)
        self.assertEqual(b"zulip!", data)
        self.logout()

        # Confirm other cross-realm users can't read it.
        self.login(user3_email, 'test')
        response = self.client_get(uri, subdomain=test_subdomain)
        self.assertEqual(response.status_code, 403)
        self.assert_in_response("You are not authorized to view this file.", response)

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 __init__(self, host_url: str, sockjs_url: str, sender_email: str,
              run_on_start: Callable[..., None], validate_ssl: bool=True,
              **run_kwargs: Any) -> None:
     # NOTE: Callable should take a WebsocketClient & kwargs, but this is not standardised
     self.validate_ssl = validate_ssl
     self.auth_email = sender_email
     self.user_profile = get_system_bot(sender_email)
     self.request_id_number = 0
     self.parsed_host_url = urlparse(host_url)
     self.sockjs_url = sockjs_url
     self.cookie_dict = self._login()
     self.cookie_str = self._get_cookie_header(self.cookie_dict)
     self.events_data = self._get_queue_events(self.cookie_str)
     self.ioloop_instance = IOLoop.instance()
     self.run_on_start = run_on_start
     self.run_kwargs = run_kwargs
     self.scheme_dict = {'http': 'ws', 'https': 'wss'}
     self.ws = None  # type: Optional[WebSocketClientConnection]

    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)

    def process_one_batch(self):
        # type: () -> None
        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,)

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

        reset_queries()

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

        if settings.SLOW_QUERY_LOGS_STREAM is None:
            return

        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", settings.SLOW_QUERY_LOGS_STREAM, topic, content)

def zulip_server_error(report):
    # type: (Dict[str, Any]) -> None
    subject = '%(node)s: %(message)s' % (report)
    stack_trace = report['stack_trace'] or "No stack trace available"

    user_info = user_info_str(report)

    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"]:
        request_repr += "- %s: \"%s\"\n" % (field, report.get(field.lower()))
    request_repr += "~~~~"

    realm = get_system_bot(settings.ERROR_BOT).realm
    internal_send_message(realm, settings.ERROR_BOT,
                          "stream", "errors", format_subject(subject),
                          "Error generated by %s\n\n~~~~ pytb\n%s\n\n~~~~\n%s" % (
                              user_info, stack_trace, request_repr))