from typing import Any, Dict, List, Optional, Union

from django.conf import settings
from django.http import HttpRequest, HttpResponse
from django.shortcuts import redirect
from django.utils.translation import ugettext as _

from zerver.decorator import require_member_or_admin, require_realm_admin
from zerver.forms import PASSWORD_TOO_WEAK_ERROR, CreateUserForm
from zerver.lib.actions import (
    check_change_bot_full_name,
    check_change_full_name,
    check_remove_custom_profile_field_value,
    do_change_avatar_fields,
    do_change_bot_owner,
    do_change_default_all_public_streams,
    do_change_default_events_register_stream,
    do_change_default_sending_stream,
    do_change_user_role,
    do_create_user,
    do_deactivate_user,
    do_reactivate_user,
    do_regenerate_api_key,
    do_update_bot_config_data,
    do_update_outgoing_webhook_service,
    do_update_user_custom_profile_data_if_changed,
    notify_created_bot,
)
from zerver.lib.avatar import avatar_url, get_gravatar_url
from zerver.lib.bot_config import set_bot_config
from zerver.lib.email_validation import email_allowed_for_realm
from zerver.lib.exceptions import CannotDeactivateLastUserError, OrganizationOwnerRequired
from zerver.lib.integrations import EMBEDDED_BOTS
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_error, json_success
from zerver.lib.streams import access_stream_by_id, access_stream_by_name, subscribed_to_stream
from zerver.lib.types import Validator
from zerver.lib.upload import upload_avatar_image
from zerver.lib.url_encoding import add_query_arg_to_redirect_url
from zerver.lib.users import (
    access_bot_by_id,
    access_user_by_id,
    add_service,
    check_bot_creation_policy,
    check_bot_name_available,
    check_full_name,
    check_short_name,
    check_valid_bot_config,
    check_valid_bot_type,
    check_valid_interface_type,
    get_api_key,
    get_raw_user_data,
    validate_user_custom_profile_data,
)
from zerver.lib.utils import generate_api_key
from zerver.lib.validator import (
    check_bool,
    check_dict,
    check_dict_only,
    check_int,
    check_int_in,
    check_list,
    check_none_or,
    check_string,
    check_union,
    check_url,
)
from zerver.models import (
    DisposableEmailError,
    DomainNotAllowedForRealmError,
    EmailContainsPlusError,
    InvalidFakeEmailDomain,
    Message,
    Realm,
    Service,
    Stream,
    UserProfile,
    get_user_by_delivery_email,
    get_user_by_id_in_realm_including_cross_realm,
    get_user_including_cross_realm,
    get_user_profile_by_id_in_realm,
)
from zproject.backends import check_password_strength


def check_last_owner(user_profile: UserProfile) -> bool:
    owners = set(user_profile.realm.get_human_owner_users())
    return user_profile.is_realm_owner and not user_profile.is_bot and len(owners) == 1

def deactivate_user_backend(request: HttpRequest, user_profile: UserProfile,
                            user_id: int) -> HttpResponse:
    target = access_user_by_id(user_profile, user_id)
    if target.is_realm_owner and not user_profile.is_realm_owner:
        raise OrganizationOwnerRequired()
    if check_last_owner(target):
        return json_error(_('Cannot deactivate the only organization owner'))
    return _deactivate_user_profile_backend(request, user_profile, target)

def deactivate_user_own_backend(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
    if UserProfile.objects.filter(realm=user_profile.realm, is_active=True).count() == 1:
        raise CannotDeactivateLastUserError(is_last_owner=False)
    if user_profile.is_realm_owner and check_last_owner(user_profile):
        raise CannotDeactivateLastUserError(is_last_owner=True)

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

def deactivate_bot_backend(request: HttpRequest, user_profile: UserProfile,
                           bot_id: int) -> HttpResponse:
    target = access_bot_by_id(user_profile, bot_id)
    return _deactivate_user_profile_backend(request, user_profile, target)

def _deactivate_user_profile_backend(request: HttpRequest, user_profile: UserProfile,
                                     target: UserProfile) -> HttpResponse:
    do_deactivate_user(target, acting_user=user_profile)
    return json_success()

def reactivate_user_backend(request: HttpRequest, user_profile: UserProfile,
                            user_id: int) -> HttpResponse:
    target = access_user_by_id(user_profile, user_id, allow_deactivated=True, allow_bots=True)
    if target.is_bot:
        assert target.bot_type is not None
        check_bot_creation_policy(user_profile, target.bot_type)
    do_reactivate_user(target, acting_user=user_profile)
    return json_success()

check_profile_data: Validator[List[Dict[str, Optional[Union[int, str, List[int]]]]]] = check_list(
    check_dict_only([
        ('id', check_int),
        ('value', check_none_or(
            check_union([check_int, check_string, check_list(check_int)]),
        )),
    ]),
)

@has_request_variables
def update_user_backend(
    request: HttpRequest, user_profile: UserProfile, user_id: int,
    full_name: Optional[str] = REQ(default=None, validator=check_string),
    role: Optional[int] = REQ(default=None, validator=check_int_in(
        UserProfile.ROLE_TYPES,
    )),
    profile_data: Optional[List[Dict[str, Optional[Union[int, str, List[int]]]]]] = REQ(
        default=None, validator=check_profile_data,
    ),
) -> HttpResponse:
    target = access_user_by_id(user_profile, user_id, allow_deactivated=True, allow_bots=True)

    if role is not None and target.role != role:
        if target.role == UserProfile.ROLE_REALM_OWNER and check_last_owner(user_profile):
            return json_error(_('The owner permission cannot be removed from the only organization owner.'))
        if UserProfile.ROLE_REALM_OWNER in [role, target.role] and not user_profile.is_realm_owner:
            raise OrganizationOwnerRequired()
        do_change_user_role(target, role)

    if (full_name is not None and target.full_name != full_name and
            full_name.strip() != ""):
        # We don't respect `name_changes_disabled` here because the request
        # is on behalf of the administrator.
        check_change_full_name(target, full_name, user_profile)

    if profile_data is not None:
        clean_profile_data = []
        for entry in profile_data:
            assert isinstance(entry["id"], int)
            if entry["value"] is None or not entry["value"]:
                field_id = entry["id"]
                check_remove_custom_profile_field_value(target, field_id)
            else:
                clean_profile_data.append({
                    "id": entry["id"],
                    "value": entry["value"],
                })
        validate_user_custom_profile_data(target.realm.id, clean_profile_data)
        do_update_user_custom_profile_data_if_changed(target, clean_profile_data)

    return json_success()

def avatar(request: HttpRequest, user_profile: UserProfile,
           email_or_id: str, medium: bool=False) -> HttpResponse:
    """Accepts an email address or user ID and returns the avatar"""
    is_email = False
    try:
        int(email_or_id)
    except ValueError:
        is_email = True

    try:
        realm = user_profile.realm
        if is_email:
            avatar_user_profile = get_user_including_cross_realm(email_or_id, realm)
        else:
            avatar_user_profile = get_user_by_id_in_realm_including_cross_realm(int(email_or_id), realm)
        # If there is a valid user account passed in, use its avatar
        url = avatar_url(avatar_user_profile, medium=medium)
    except UserProfile.DoesNotExist:
        # If there is no such user, treat it as a new gravatar
        email = email_or_id
        avatar_version = 1
        url = get_gravatar_url(email, avatar_version, medium)

    # We can rely on the url already having query parameters. Because
    # our templates depend on being able to use the ampersand to
    # add query parameters to our url, get_avatar_url does '?x=x'
    # hacks to prevent us from having to jump through decode/encode hoops.
    assert url is not None
    url = add_query_arg_to_redirect_url(url, request.META['QUERY_STRING'])
    return redirect(url)

def get_stream_name(stream: Optional[Stream]) -> Optional[str]:
    if stream:
        return stream.name
    return None

@require_member_or_admin
@has_request_variables
def patch_bot_backend(
        request: HttpRequest, user_profile: UserProfile, bot_id: int,
        full_name: Optional[str]=REQ(default=None),
        bot_owner_id: Optional[int]=REQ(validator=check_int, default=None),
        config_data: Optional[Dict[str, str]]=REQ(default=None,
                                                  validator=check_dict(value_validator=check_string)),
        service_payload_url: Optional[str]=REQ(validator=check_url, default=None),
        service_interface: int=REQ(validator=check_int, default=1),
        default_sending_stream: Optional[str]=REQ(default=None),
        default_events_register_stream: Optional[str]=REQ(default=None),
        default_all_public_streams: Optional[bool]=REQ(default=None, validator=check_bool),
) -> HttpResponse:
    bot = access_bot_by_id(user_profile, bot_id)

    if full_name is not None:
        check_change_bot_full_name(bot, full_name, user_profile)
    if bot_owner_id is not None:
        try:
            owner = get_user_profile_by_id_in_realm(bot_owner_id, user_profile.realm)
        except UserProfile.DoesNotExist:
            return json_error(_('Failed to change owner, no such user'))
        if not owner.is_active:
            return json_error(_('Failed to change owner, user is deactivated'))
        if owner.is_bot:
            return json_error(_("Failed to change owner, bots can't own other bots"))

        previous_owner = bot.bot_owner
        if previous_owner != owner:
            do_change_bot_owner(bot, owner, user_profile)

    if default_sending_stream is not None:
        if default_sending_stream == "":
            stream: Optional[Stream] = None
        else:
            (stream, recipient, sub) = access_stream_by_name(
                user_profile, default_sending_stream)
        do_change_default_sending_stream(bot, stream)
    if default_events_register_stream is not None:
        if default_events_register_stream == "":
            stream = None
        else:
            (stream, recipient, sub) = access_stream_by_name(
                user_profile, default_events_register_stream)
        do_change_default_events_register_stream(bot, stream)
    if default_all_public_streams is not None:
        do_change_default_all_public_streams(bot, default_all_public_streams)

    if service_payload_url is not None:
        check_valid_interface_type(service_interface)