def create_user_profile(realm: Realm, email: str, password: Optional[str],
                        active: bool, bot_type: Optional[int], full_name: str,
                        short_name: str, bot_owner: Optional[UserProfile],
                        is_mirror_dummy: bool, tos_version: Optional[str],
                        timezone: Optional[str],
                        tutorial_status: Optional[str] = UserProfile.TUTORIAL_WAITING,
                        enter_sends: bool = False) -> UserProfile:
    now = timezone_now()
    email = UserManager.normalize_email(email)

    user_profile = UserProfile(is_staff=False, is_active=active,
                               full_name=full_name, short_name=short_name,
                               last_login=now, date_joined=now, realm=realm,
                               pointer=-1, is_bot=bool(bot_type), bot_type=bot_type,
                               bot_owner=bot_owner, is_mirror_dummy=is_mirror_dummy,
                               tos_version=tos_version, timezone=timezone,
                               tutorial_status=tutorial_status,
                               enter_sends=enter_sends,
                               onboarding_steps=ujson.dumps([]),
                               default_language=realm.default_language,
                               twenty_four_hour_time=realm.default_twenty_four_hour_time,
                               delivery_email=email)

    if bot_type or not active:
        password = None
    if realm.email_address_visibility == Realm.EMAIL_ADDRESS_VISIBILITY_EVERYONE:
        # If emails are visible to everyone, we can set this here and save a DB query
        user_profile.email = get_display_email_address(user_profile, realm)
    user_profile.set_password(password)
    user_profile.api_key = generate_api_key()
    return user_profile

def create_user_profile(realm, email, password, active, bot_type, full_name,
                        short_name, bot_owner, is_mirror_dummy, tos_version,
                        timezone, tutorial_status=UserProfile.TUTORIAL_WAITING,
                        enter_sends=False):
    # type: (Realm, Text, Optional[Text], bool, Optional[int], Text, Text, Optional[UserProfile], bool, Text, Optional[Text], Optional[Text], bool) -> UserProfile
    now = timezone_now()
    email = UserManager.normalize_email(email)

    user_profile = UserProfile(email=email, is_staff=False, is_active=active,
                               full_name=full_name, short_name=short_name,
                               last_login=now, date_joined=now, realm=realm,
                               pointer=-1, is_bot=bool(bot_type), bot_type=bot_type,
                               bot_owner=bot_owner, is_mirror_dummy=is_mirror_dummy,
                               tos_version=tos_version, timezone=timezone,
                               tutorial_status=tutorial_status,
                               enter_sends=enter_sends,
                               onboarding_steps=ujson.dumps([]),
                               default_language=realm.default_language)

    if bot_type or not active:
        password = None

    user_profile.set_password(password)

    user_profile.api_key = random_api_key()
    return user_profile

def create_user_profile(realm: Realm, email: str, password: Optional[str],
                        active: bool, bot_type: Optional[int], full_name: str,
                        short_name: str, bot_owner: Optional[UserProfile],
                        is_mirror_dummy: bool, tos_version: Optional[str],
                        timezone: Optional[str],
                        tutorial_status: Optional[str] = UserProfile.TUTORIAL_WAITING,
                        enter_sends: bool = False) -> UserProfile:
    now = timezone_now()
    email = UserManager.normalize_email(email)

    user_profile = UserProfile(email=email, is_staff=False, is_active=active,
                               full_name=full_name, short_name=short_name,
                               last_login=now, date_joined=now, realm=realm,
                               pointer=-1, is_bot=bool(bot_type), bot_type=bot_type,
                               bot_owner=bot_owner, is_mirror_dummy=is_mirror_dummy,
                               tos_version=tos_version, timezone=timezone,
                               tutorial_status=tutorial_status,
                               enter_sends=enter_sends,
                               onboarding_steps=ujson.dumps([]),
                               default_language=realm.default_language,
                               twenty_four_hour_time=realm.default_twenty_four_hour_time,
                               delivery_email=email)

    if bot_type or not active:
        password = None

    user_profile.set_password(password)

    user_profile.api_key = generate_api_key()
    return user_profile

def do_create_customer(user: UserProfile, stripe_token: Optional[str]=None,
                       coupon: Optional[Coupon]=None) -> stripe.Customer:
    realm = user.realm
    stripe_coupon_id = None
    if coupon is not None:
        stripe_coupon_id = coupon.stripe_coupon_id
    # We could do a better job of handling race conditions here, but if two
    # people from a realm try to upgrade at exactly the same time, the main
    # bad thing that will happen is that we will create an extra stripe
    # customer that we can delete or ignore.
    stripe_customer = stripe.Customer.create(
        description="%s (%s)" % (realm.string_id, realm.name),
        email=user.email,
        metadata={'realm_id': realm.id, 'realm_str': realm.string_id},
        source=stripe_token,
        coupon=stripe_coupon_id)
    if PRINT_STRIPE_FIXTURE_DATA:
        print(''.join(['"create_customer": ', str(stripe_customer), ',']))  # nocoverage
    event_time = timestamp_to_datetime(stripe_customer.created)
    with transaction.atomic():
        RealmAuditLog.objects.create(
            realm=user.realm, acting_user=user, event_type=RealmAuditLog.STRIPE_CUSTOMER_CREATED,
            event_time=event_time)
        if stripe_token is not None:
            RealmAuditLog.objects.create(
                realm=user.realm, acting_user=user, event_type=RealmAuditLog.STRIPE_CARD_CHANGED,
                event_time=event_time)
        Customer.objects.create(realm=realm, stripe_customer_id=stripe_customer.id)
        user.is_billing_admin = True
        user.save(update_fields=["is_billing_admin"])
    return stripe_customer

def create_user_profile(realm, email, password, active, bot_type, full_name,
                        short_name, bot_owner, is_mirror_dummy, tos_version):
    # type: (Realm, text_type, text_type, bool, Optional[int], text_type, text_type, Optional[UserProfile], bool, Optional[text_type]) -> UserProfile
    now = timezone.now()
    email = UserManager.normalize_email(email)

    enable_stream_desktop_notifications = (realm.domain != 'zulip.com')

    user_profile = UserProfile(email=email, is_staff=False, is_active=active,
                               full_name=full_name, short_name=short_name,
                               last_login=now, date_joined=now, realm=realm,
                               pointer=-1, is_bot=bool(bot_type), bot_type=bot_type,
                               is_mirror_dummy=is_mirror_dummy,
                               tos_version=tos_version,
                               enable_stream_desktop_notifications=enable_stream_desktop_notifications,
                               onboarding_steps=ujson.dumps([]),
                               default_language=realm.default_language)
    if bot_owner is not None:
        # `user_profile.bot_owner = bot_owner` doesn't work on python 3.4
        user_profile.bot_owner_id = bot_owner.id

    if bot_type or not active:
        password = None

    user_profile.set_password(password)

    user_profile.api_key = random_api_key()
    return user_profile

def get_next_hotspots(user: UserProfile) -> List[Dict[str, object]]:
    # For manual testing, it can be convenient to set
    # ALWAYS_SEND_ALL_HOTSPOTS=True in `zproject/dev_settings.py` to
    # make it easy to click on all of the hotspots.
    if settings.ALWAYS_SEND_ALL_HOTSPOTS:
        return [{
            'name': hotspot,
            'title': ALL_HOTSPOTS[hotspot]['title'],
            'description': ALL_HOTSPOTS[hotspot]['description'],
            'delay': 0,
        } for hotspot in ALL_HOTSPOTS]

    if user.tutorial_status == UserProfile.TUTORIAL_FINISHED:
        return []

    seen_hotspots = frozenset(UserHotspot.objects.filter(user=user).values_list('hotspot', flat=True))
    for hotspot in ['intro_reply', 'intro_streams', 'intro_topics', 'intro_compose']:
        if hotspot not in seen_hotspots:
            return [{
                'name': hotspot,
                'title': ALL_HOTSPOTS[hotspot]['title'],
                'description': ALL_HOTSPOTS[hotspot]['description'],
                'delay': 0.5,
            }]

    user.tutorial_status = UserProfile.TUTORIAL_FINISHED
    user.save(update_fields=['tutorial_status'])
    return []

def do_create_stripe_customer(user: UserProfile, stripe_token: Optional[str]=None) -> Customer:
    realm = user.realm
    # We could do a better job of handling race conditions here, but if two
    # people from a realm try to upgrade at exactly the same time, the main
    # bad thing that will happen is that we will create an extra stripe
    # customer that we can delete or ignore.
    stripe_customer = stripe.Customer.create(
        description="%s (%s)" % (realm.string_id, realm.name),
        email=user.email,
        metadata={'realm_id': realm.id, 'realm_str': realm.string_id},
        source=stripe_token)
    event_time = timestamp_to_datetime(stripe_customer.created)
    with transaction.atomic():
        RealmAuditLog.objects.create(
            realm=user.realm, acting_user=user, event_type=RealmAuditLog.STRIPE_CUSTOMER_CREATED,
            event_time=event_time)
        if stripe_token is not None:
            RealmAuditLog.objects.create(
                realm=user.realm, acting_user=user, event_type=RealmAuditLog.STRIPE_CARD_CHANGED,
                event_time=event_time)
        customer, created = Customer.objects.update_or_create(realm=realm, defaults={
            'stripe_customer_id': stripe_customer.id})
        user.is_billing_admin = True
        user.save(update_fields=["is_billing_admin"])
    return customer

def copy_hotpots(source_profile: UserProfile, target_profile: UserProfile) -> None:
    for userhotspot in frozenset(UserHotspot.objects.filter(user=source_profile)):
        UserHotspot.objects.create(user=target_profile, hotspot=userhotspot.hotspot,
                                   timestamp=userhotspot.timestamp)

    target_profile.tutorial_status = source_profile.tutorial_status
    target_profile.onboarding_steps = source_profile.onboarding_steps
    target_profile.save(update_fields=['tutorial_status', 'onboarding_steps'])

def set_tutorial_status(request: HttpRequest, user_profile: UserProfile,
                        status: str=REQ(validator=check_string)) -> HttpResponse:
    if status == 'started':
        user_profile.tutorial_status = UserProfile.TUTORIAL_STARTED
    elif status == 'finished':
        user_profile.tutorial_status = UserProfile.TUTORIAL_FINISHED
    user_profile.save(update_fields=["tutorial_status"])

    return json_success()

def do_soft_deactivate_user(user_profile: UserProfile) -> None:
    user_profile.last_active_message_id = UserMessage.objects.filter(
        user_profile=user_profile).order_by(
        '-message__id')[0].message_id
    user_profile.long_term_idle = True
    user_profile.save(update_fields=[
        'long_term_idle',
        'last_active_message_id'])
    logger.info('Soft Deactivated user %s (%s)' %
                (user_profile.id, user_profile.email))

def maybe_catch_up_soft_deactivated_user(user_profile: UserProfile) -> Union[UserProfile, None]:
    if user_profile.long_term_idle:
        add_missing_messages(user_profile)
        user_profile.long_term_idle = False
        user_profile.save(update_fields=['long_term_idle'])
        RealmAuditLog.objects.create(
            realm=user_profile.realm,
            modified_user=user_profile,
            event_type=RealmAuditLog.USER_SOFT_ACTIVATED,
            event_time=timezone_now()
        )
        logger.info('Soft Reactivated user %s (%s)' %
                    (user_profile.id, user_profile.email))
        return user_profile
    return None

def do_soft_deactivate_user(user_profile: UserProfile) -> None:
    try:
        user_profile.last_active_message_id = UserMessage.objects.filter(
            user_profile=user_profile).order_by(
                '-message__id')[0].message_id
    except IndexError:  # nocoverage
        # In the unlikely event that a user somehow has never received
        # a message, we just use the overall max message ID.
        user_profile.last_active_message_id = Message.objects.max().id
    user_profile.long_term_idle = True
    user_profile.save(update_fields=[
        'long_term_idle',
        'last_active_message_id'])
    logger.info('Soft Deactivated user %s (%s)' %
                (user_profile.id, user_profile.email))

def list_to_streams(streams_raw: Iterable[Mapping[str, Any]],
                    user_profile: UserProfile,
                    autocreate: bool=False) -> Tuple[List[Stream], List[Stream]]:
    """Converts list of dicts to a list of Streams, validating input in the process

    For each stream name, we validate it to ensure it meets our
    requirements for a proper stream name using check_stream_name.

    This function in autocreate mode should be atomic: either an exception will be raised
    during a precheck, or all the streams specified will have been created if applicable.

    @param streams_raw The list of stream dictionaries to process;
      names should already be stripped of whitespace by the caller.
    @param user_profile The user for whom we are retreiving the streams
    @param autocreate Whether we should create streams if they don't already exist
    """
    # Validate all streams, getting extant ones, then get-or-creating the rest.

    stream_set = set(stream_dict["name"] for stream_dict in streams_raw)

    for stream_name in stream_set:
        # Stream names should already have been stripped by the
        # caller, but it makes sense to verify anyway.
        assert stream_name == stream_name.strip()
        check_stream_name(stream_name)

    existing_streams = []  # type: List[Stream]
    missing_stream_dicts = []  # type: List[Mapping[str, Any]]
    existing_stream_map = bulk_get_streams(user_profile.realm, stream_set)

    for stream_dict in streams_raw:
        stream_name = stream_dict["name"]
        stream = existing_stream_map.get(stream_name.lower())
        if stream is None:
            missing_stream_dicts.append(stream_dict)
        else:
            existing_streams.append(stream)

    if len(missing_stream_dicts) == 0:
        # This is the happy path for callers who expected all of these
        # streams to exist already.
        created_streams = []  # type: List[Stream]
    else:
        # autocreate=True path starts here
        if not user_profile.can_create_streams():
            raise JsonableError(_('User cannot create streams.'))
        elif not autocreate:
            raise JsonableError(_("Stream(s) (%s) do not exist") % ", ".join(
                stream_dict["name"] for stream_dict in missing_stream_dicts))

        # We already filtered out existing streams, so dup_streams
        # will normally be an empty list below, but we protect against somebody
        # else racing to create the same stream.  (This is not an entirely
        # paranoid approach, since often on Zulip two people will discuss
        # creating a new stream, and both people eagerly do it.)
        created_streams, dup_streams = create_streams_if_needed(realm=user_profile.realm,
                                                                stream_dicts=missing_stream_dicts)
        existing_streams += dup_streams

    return existing_streams, created_streams

def update_display_settings_backend(
        request: HttpRequest, user_profile: UserProfile,
        twenty_four_hour_time: Optional[bool]=REQ(validator=check_bool, default=None),
        dense_mode: Optional[bool]=REQ(validator=check_bool, default=None),
        starred_message_counts: Optional[bool]=REQ(validator=check_bool, default=None),
        high_contrast_mode: Optional[bool]=REQ(validator=check_bool, default=None),
        night_mode: Optional[bool]=REQ(validator=check_bool, default=None),
        translate_emoticons: Optional[bool]=REQ(validator=check_bool, default=None),
        default_language: Optional[bool]=REQ(validator=check_string, default=None),
        left_side_userlist: Optional[bool]=REQ(validator=check_bool, default=None),
        emojiset: Optional[str]=REQ(validator=check_string, default=None),
        timezone: Optional[str]=REQ(validator=check_string, default=None)) -> HttpResponse:

    if (default_language is not None and
            default_language not in get_available_language_codes()):
        raise JsonableError(_("Invalid language '%s'" % (default_language,)))

    if (timezone is not None and
            timezone not in get_all_timezones()):
        raise JsonableError(_("Invalid timezone '%s'" % (timezone,)))

    if (emojiset is not None and
            emojiset not in UserProfile.emojiset_choices()):
        raise JsonableError(_("Invalid emojiset '%s'" % (emojiset,)))

    request_settings = {k: v for k, v in list(locals().items()) if k in user_profile.property_types}
    result = {}  # type: Dict[str, Any]
    for k, v in list(request_settings.items()):
        if v is not None and getattr(user_profile, k) != v:
            do_set_user_display_setting(user_profile, k, v)
            result[k] = v

    return json_success(result)

def _deactivate_user_profile_backend(request: HttpRequest, user_profile: UserProfile,
                                     target: UserProfile) -> HttpResponse:
    if not user_profile.can_admin_user(target):
        return json_error(_('Insufficient permission'))

    do_deactivate_user(target, acting_user=user_profile)
    return json_success()

def copy_user_settings(source_profile: UserProfile, target_profile: UserProfile) -> None:
    """Warning: Does not save, to avoid extra database queries"""
    for settings_name in UserProfile.property_types:
        value = getattr(source_profile, settings_name)
        setattr(target_profile, settings_name, value)

    for settings_name in UserProfile.notification_setting_types:
        value = getattr(source_profile, settings_name)
        setattr(target_profile, settings_name, value)

    setattr(target_profile, "full_name", source_profile.full_name)
    target_profile.save()

    if source_profile.avatar_source == UserProfile.AVATAR_FROM_USER:
        from zerver.lib.actions import do_change_avatar_fields
        do_change_avatar_fields(target_profile, UserProfile.AVATAR_FROM_USER)
        copy_avatar(source_profile, target_profile)

def access_bot_by_id(user_profile: UserProfile, user_id: int) -> UserProfile:
    try:
        target = get_user_profile_by_id_in_realm(user_id, user_profile.realm)
    except UserProfile.DoesNotExist:
        raise JsonableError(_("No such bot"))
    if not target.is_bot:
        raise JsonableError(_("No such bot"))
    if not user_profile.can_admin_user(target):
        raise JsonableError(_("Insufficient permission"))
    return target

    def test(user: UserProfile) -> bool:
        """
        :if_configured: If ``True``, an authenticated user with no confirmed
        OTP devices will be allowed. Default is ``False``. If ``False``,
        2FA will not do any authentication.
        """
        if_configured = settings.TWO_FACTOR_AUTHENTICATION_ENABLED
        if not if_configured:
            return True

        return user.is_verified() or (_user_is_authenticated(user)
                                      and not user_has_device(user))

def reactivate_user_backend(request: HttpRequest, user_profile: UserProfile,
                            email: Text) -> HttpResponse:
    try:
        target = get_user(email, user_profile.realm)
    except UserProfile.DoesNotExist:
        return json_error(_('No such user'))

    if not user_profile.can_admin_user(target):
        return json_error(_('Insufficient permission'))

    do_reactivate_user(target, acting_user=user_profile)
    return json_success()

def gather_new_users(user_profile: UserProfile, threshold: datetime.datetime) -> Tuple[int, List[str]]:
    # Gather information on users in the realm who have recently
    # joined.
    if not user_profile.can_access_all_realm_members():
        new_users = []  # type: List[UserProfile]
    else:
        new_users = list(UserProfile.objects.filter(
            realm=user_profile.realm, date_joined__gt=threshold,
            is_bot=False))
    user_names = [user.full_name for user in new_users]

    return len(user_names), user_names