def handle(self, *args, **options):
        # type: (*Any, **str) -> None
        realm = self.get_realm(options)
        old_name = options['old_name']
        new_name = options['new_name']
        encoding = sys.getfilesystemencoding()

        stream = get_stream(force_text(old_name, encoding), realm)
        do_rename_stream(stream, force_text(new_name, encoding))

    def handle(self, *args: Any, **options: str) -> None:
        realm = self.get_realm(options)
        assert realm is not None  # Should be ensured by parser
        old_name = options['old_name']
        new_name = options['new_name']
        encoding = sys.getfilesystemencoding()

        stream = get_stream(force_text(old_name, encoding), realm)
        do_rename_stream(stream, force_text(new_name, encoding))

    def handle(self, *args, **options):
        # type: (*Any, **str) -> None
        domain = options['domain']
        encoding = sys.getfilesystemencoding()
        stream_name = options['stream_name']

        realm = get_realm(force_text(domain, encoding))
        if realm is None:
            print("Unknown domain %s" % (domain,))
            exit(1)
        else:
            do_create_stream(realm, force_text(stream_name, encoding))

    def handle(self, *args, **options):
        # type: (*Any, **str) -> None
        string_id = options['realm']
        encoding = sys.getfilesystemencoding()
        stream_name = options['stream_name']

        realm = get_realm_by_string_id(force_text(string_id, encoding))
        if realm is None:
            print("Unknown string_id %s" % (string_id,))
            exit(1)
        else:
            do_create_stream(realm, force_text(stream_name, encoding))

    def handle(self, *args, **options):
        # type: (*Any, **str) -> None
        string_id = options['string_id']
        old_name = options['old_name']
        new_name = options['new_name']
        encoding = sys.getfilesystemencoding()

        realm = get_realm(force_text(string_id, encoding))
        if realm is None:
            print("Unknown subdomain or string_id %s" % (string_id,))
            exit(1)

        do_rename_stream(realm, force_text(old_name, encoding),
                         force_text(new_name, encoding))

    def handle(self, *args: Any, **options: str) -> None:
        realm = self.get_realm(options)
        assert realm is not None  # Should be ensured by parser

        encoding = sys.getfilesystemencoding()
        stream_name = options['stream_name']
        create_stream_if_needed(realm, force_text(stream_name, encoding))

def process_message(message: message.Message, rcpt_to: Optional[str]=None, pre_checked: bool=False) -> None:
    subject_header = str(message.get("Subject", "")).strip()
    if subject_header == "":
        subject_header = "(no topic)"
    encoded_subject, encoding = decode_header(subject_header)[0]
    if encoding is None:
        subject = force_text(encoded_subject)  # encoded_subject has type str when encoding is None
    else:
        try:
            subject = encoded_subject.decode(encoding)
        except (UnicodeDecodeError, LookupError):
            subject = "(unreadable subject)"

    debug_info = {}

    try:
        if rcpt_to is not None:
            to = rcpt_to
        else:
            to = find_emailgateway_recipient(message)
        debug_info["to"] = to

        if is_missed_message_address(to):
            process_missed_message(to, message, pre_checked)
        else:
            process_stream_message(to, subject, message, debug_info)
    except ZulipEmailForwardError as e:
        # TODO: notify sender of error, retry if appropriate.
        log_and_report(message, str(e), debug_info)

 def get_url_data(self, e):
     # type: (Element) -> Optional[Tuple[text_type, text_type]]
     if e.tag == "a":
         if e.text is not None:
             return (e.get("href"), force_text(e.text))
         return (e.get("href"), e.get("href"))
     return None

def process_message(message, rcpt_to=None, pre_checked=False):
    # type: (message.Message, Optional[text_type], bool) -> None
    subject_header = message.get("Subject", "(no subject)")
    encoded_subject, encoding = decode_header(subject_header)[0] # type: ignore # https://github.com/python/typeshed/pull/333
    if encoding is None:
        subject = force_text(encoded_subject) # encoded_subject has type str when encoding is None
    else:
        try:
            subject = encoded_subject.decode(encoding)
        except (UnicodeDecodeError, LookupError):
            subject = u"(unreadable subject)"

    debug_info = {}

    try:
        if rcpt_to is not None:
            to = rcpt_to
        else:
            to = find_emailgateway_recipient(message)
        debug_info["to"] = to

        if is_missed_message_address(to):
            process_missed_message(to, message, pre_checked)
        else:
            process_stream_message(to, subject, message, debug_info)
    except ZulipEmailForwardError as e:
        # TODO: notify sender of error, retry if appropriate.
        log_and_report(message, str(e), debug_info)

def make_safe_digest(string, hash_func=hashlib.sha1):
    # type: (text_type, Callable[[binary_type], Any]) -> text_type
    """
    return a hex digest of `string`.
    """
    # hashlib.sha1, md5, etc. expect bytes, so non-ASCII strings must
    # be encoded.
    return force_text(hash_func(string.encode('utf-8')).hexdigest())

def list_of_tlds():
    # type: () -> List[text_type]
    # HACK we manually blacklist .py
    blacklist = [u'PY\n', ]

    # tlds-alpha-by-domain.txt comes from http://data.iana.org/TLD/tlds-alpha-by-domain.txt
    tlds_file = os.path.join(os.path.dirname(__file__), 'tlds-alpha-by-domain.txt')
    tlds = [force_text(tld).lower().strip() for tld in open(tlds_file, 'r')
                if tld not in blacklist and not tld[0].startswith('#')]
    tlds.sort(key=len, reverse=True)
    return tlds

def get_file_info(request, user_file):
    # type: (HttpRequest, File) -> Tuple[text_type, text_type]

    uploaded_file_name = user_file.name
    content_type = request.GET.get('mimetype')
    if content_type is None:
        content_type = force_text(guess_type(uploaded_file_name)[0])
    else:
        uploaded_file_name = uploaded_file_name + guess_extension(content_type)

    uploaded_file_name = urllib.parse.unquote(uploaded_file_name)
    return uploaded_file_name, content_type

 def run(self, root):
     # type: (ElementTree) -> None
     """ Find code blocks and store in htmlStash. """
     blocks = root.getiterator('pre')
     for block in blocks:
         children = block.getchildren()
         tag = force_text(children[0].tag)
         if len(children) == 1 and tag == 'code':
             text = force_text(children[0].text)
             code = CodeHilite(text,
                         force_linenos=self.config['force_linenos'],
                         guess_lang=self.config['guess_lang'],
                         css_class=self.config['css_class'],
                         style=self.config['pygments_style'],
                         noclasses=self.config['noclasses'],
                         tab_length=self.markdown.tab_length)
             placeholder = self.markdown.htmlStash.store(code.hilite(),
                                                         safe=True)
             # Clear codeblock in etree instance
             block.clear()
             # Change to p element which will later
             # be removed when inserting raw html
             block.tag = 'p'
             block.text = placeholder

    def handle(self, *args, **options):
        # type: (*Any, **str) -> None

        string_id = options['string_id']
        bot_email = options['bot_email']
        service_name = options['service_name']
        base_url = options['base_url']

        encoding = sys.getfilesystemencoding()
        realm = get_realm(force_text(string_id, encoding))
        if realm is None:
            print('Unknown subdomain or string_id %s' % (string_id,))
            exit(1)

        if not bot_email:
            print('Email of existing bot must be provided')
            exit(1)

        if not service_name:
            print('Name for Service object must be provided')
            exit(1)

        if not base_url:
            print('Base URL of outgoing webhook must be provided')
            exit(1)

        # TODO: Normalize email?
        bot_profile = UserProfile.objects.get(email=bot_email)
        if not bot_profile:
            print('User %s does not exist' % (bot_email,))
            exit(1)
        if not bot_profile.is_bot:
            print('User %s is not a bot' % (bot_email,))
            exit(1)
        if bot_profile.is_outgoing_webhook_bot:
            print('%s is already marked as an outgoing webhook bot' % (bot_email,))
            exit(1)

        Service.objects.create(name=service_name,
                               user_profile=bot_profile,
                               base_url=base_url,
                               token='',
                               interface=1)

        bot_profile.bot_type = UserProfile.OUTGOING_WEBHOOK_BOT
        bot_profile.save()

        print('Successfully converted %s into an outgoing webhook bot' % (bot_email,))

def highlight_string_bytes_offsets(text, locs):
    # type: (AnyStr, Iterable[Tuple[int, int]]) -> Text
    string = force_bytes(text)
    highlight_start = b'<span class="highlight">'
    highlight_stop = b'</span>'
    pos = 0
    result = b''
    for loc in locs:
        (offset, length) = loc
        result += string[pos:offset]
        result += highlight_start
        result += string[offset:offset + length]
        result += highlight_stop
        pos = offset + length
    result += string[pos:]
    return force_text(result)

def get_file_info(request, user_file):
    # type: (HttpRequest, File) -> Tuple[text_type, Optional[text_type]]

    uploaded_file_name = user_file.name
    assert isinstance(uploaded_file_name, str)

    content_type = request.GET.get('mimetype')
    if content_type is None:
        guessed_type = guess_type(uploaded_file_name)[0]
        if guessed_type is not None:
            content_type = force_text(guessed_type)
    else:
        uploaded_file_name = uploaded_file_name + guess_extension(content_type)

    uploaded_file_name = urllib.parse.unquote(uploaded_file_name)
    return uploaded_file_name, content_type

def highlight_string_text_offsets(text, locs):
    # type: (AnyStr, Iterable[Tuple[int, int]]) -> text_type
    string = force_text(text)
    highlight_start = u'<span class="highlight">'
    highlight_stop = u"</span>"
    pos = 0
    result = u""
    for loc in locs:
        (offset, length) = loc
        result += string[pos:offset]
        result += highlight_start
        result += string[offset : offset + length]
        result += highlight_stop
        pos = offset + length
    result += string[pos:]
    return result

    def write_translation_strings(self, translation_strings):
        # type: (Iterable[str]) -> None
        for locale, output_path in zip(self.get_locales(), self.get_output_paths()):
            self.stdout.write("[frontend] processing locale {}".format(locale))
            try:
                with open(output_path, 'r') as reader:
                    old_strings = json.load(reader)
            except (IOError, ValueError):
                old_strings = {}

            new_strings = {
                force_text(k): v
                for k, v in self.get_new_strings(old_strings,
                                                 translation_strings).items()
            }
            with open(output_path, 'w') as writer:
                json.dump(new_strings, writer, indent=2, sort_keys=True)

def sanitize_name(raw_value):
    # type: (NonBinaryStr) -> text_type
    """
    Sanitizes a value to be safe to store in a Linux filesystem, in
    S3, and in a URL.  So unicode is allowed, but not special
    characters other than ".", "-", and "_".

    This implementation is based on django.utils.text.slugify; it is
    modified by:
    * hardcoding allow_unicode=True.
    * adding '.' and '_' to the list of allowed characters.
    * preserving the case of the value.
    """
    value = force_text(raw_value)
    value = unicodedata.normalize('NFKC', value)
    value = re.sub('[^\w\s._-]', '', value, flags=re.U).strip()
    return mark_safe(re.sub('[-\s]+', '-', value, flags=re.U))

    def handle(self, *args, **options):
        # type: (*Any, **str) -> None
        rcpt_to = force_text(os.environ.get("ORIGINAL_RECIPIENT", options['recipient']))
        if rcpt_to is not None:
            if is_missed_message_address(rcpt_to):
                try:
                    mark_missed_message_address_as_used(rcpt_to)
                except ZulipEmailForwardError:
                    print("5.1.1 Bad destination mailbox address: Bad or expired missed message address.")
                    exit(posix.EX_NOUSER) # type: ignore # There are no stubs for posix in python 3
            else:
                try:
                    extract_and_validate(rcpt_to)
                except ZulipEmailForwardError:
                    print("5.1.1 Bad destination mailbox address: Please use the address specified "
                          "in your Streams page.")
                    exit(posix.EX_NOUSER) # type: ignore # There are no stubs for posix in python 3

            # Read in the message, at most 25MiB. This is the limit enforced by
            # Gmail, which we use here as a decent metric.
            msg_text = sys.stdin.read(25*1024*1024)

            if len(sys.stdin.read(1)) != 0:
                # We're not at EOF, reject large mail.
                print("5.3.4 Message too big for system: Max size is 25MiB")
                exit(posix.EX_DATAERR) # type: ignore # There are no stubs for posix in python 3

            queue_json_publish(
                    "email_mirror",
                    {
                        "message": msg_text,
                        "rcpt_to": rcpt_to
                    },
                    lambda x: None
            )
        else:
            # We're probably running from cron, try to batch-process mail
            if (not settings.EMAIL_GATEWAY_BOT or not settings.EMAIL_GATEWAY_LOGIN or
                not settings.EMAIL_GATEWAY_PASSWORD or not settings.EMAIL_GATEWAY_IMAP_SERVER or
                    not settings.EMAIL_GATEWAY_IMAP_PORT or not settings.EMAIL_GATEWAY_IMAP_FOLDER):
                print("Please configure the Email Mirror Gateway in /etc/zulip/, "
                      "or specify $ORIGINAL_RECIPIENT if piping a single mail.")
                exit(1)
            for message in get_imap_messages():
                process_message(message)