# Tweepy
# Copyright 2009-2023 Joshua Roesslein
# See LICENSE for details.

import logging
import warnings

import requests
from requests.auth import AuthBase, HTTPBasicAuth
from requests_oauthlib import OAuth1, OAuth1Session, OAuth2Session

from tweepy.errors import TweepyException

WARNING_MESSAGE = """Warning! Due to a Twitter API bug, signin_with_twitter
and access_type don't always play nice together. Details
https://dev.twitter.com/discussions/21281"""

log = logging.getLogger(__name__)


class OAuth1UserHandler:
    """OAuth 1.0a User Context authentication handler

    .. versionchanged:: 4.5
        Renamed from :class:`OAuthHandler`
    """

    def __init__(self, consumer_key, consumer_secret, access_token=None,
                 access_token_secret=None, callback=None):
        if not isinstance(consumer_key, (str, bytes)):
            raise TypeError("Consumer key must be string or bytes, not "
                            + type(consumer_key).__name__)
        if not isinstance(consumer_secret, (str, bytes)):
            raise TypeError("Consumer secret must be string or bytes, not "
                            + type(consumer_secret).__name__)

        self.consumer_key = consumer_key
        self.consumer_secret = consumer_secret
        self.access_token = access_token
        self.access_token_secret = access_token_secret
        self.callback = callback
        self.username = None
        self.request_token = {}
        self.oauth = OAuth1Session(consumer_key, client_secret=consumer_secret,
                                   callback_uri=self.callback)

    def apply_auth(self):
        return OAuth1(
            self.consumer_key, client_secret=self.consumer_secret,
            resource_owner_key=self.access_token,
            resource_owner_secret=self.access_token_secret, decoding=None
        )

    def _get_oauth_url(self, endpoint):
        return 'https://api.twitter.com/oauth/' + endpoint

    def _get_request_token(self, access_type=None):
        try:
            url = self._get_oauth_url('request_token')
            if access_type:
                url += f'?x_auth_access_type={access_type}'
            return self.oauth.fetch_request_token(url)
        except Exception as e:
            raise TweepyException(e)

    def get_authorization_url(self, signin_with_twitter=False,
                              access_type=None):
        """Get the authorization URL to redirect the user to"""
        try:
            if signin_with_twitter:
                url = self._get_oauth_url('authenticate')
                if access_type:
                    log.warning(WARNING_MESSAGE)
            else:
                url = self._get_oauth_url('authorize')
            self.request_token = self._get_request_token(
                access_type=access_type
            )
            return self.oauth.authorization_url(url)
        except Exception as e:
            raise TweepyException(e)

    def get_access_token(self, verifier=None):
        """After user has authorized the app, get access token and secret with
        verifier
        """
        try:
            url = self._get_oauth_url('access_token')
            self.oauth = OAuth1Session(
                self.consumer_key, client_secret=self.consumer_secret,
                resource_owner_key=self.request_token['oauth_token'],
                resource_owner_secret=self.request_token['oauth_token_secret'],
                verifier=verifier, callback_uri=self.callback
            )
            resp = self.oauth.fetch_access_token(url)
            self.access_token = resp['oauth_token']
            self.access_token_secret = resp['oauth_token_secret']
            return self.access_token, self.access_token_secret
        except Exception as e:
            raise TweepyException(e)

    def set_access_token(self, key, secret):
        """
        .. deprecated:: 4.5
            Set through initialization instead.
        """
        self.access_token = key
        self.access_token_secret = secret


class OAuthHandler(OAuth1UserHandler):
    """Alias for :class:`OAuth1UserHandler`

    .. deprecated:: 4.5
        Use :class:`OAuth1UserHandler` instead.
    """

    def __init__(self, consumer_key, consumer_secret, access_token=None,
                 access_token_secret=None, callback=None):
        warnings.warn(
            "OAuthHandler is deprecated; use OAuth1UserHandler instead.",
            DeprecationWarning
        )
        super().__init__(consumer_key, consumer_secret, access_token, 
                         access_token_secret, callback)


class OAuth2AppHandler:
    """OAuth 2.0 Bearer Token (App-Only) using API / Consumer key and secret
    authentication handler

    .. versionchanged:: 4.5
        Renamed from :class:`AppAuthHandler`
    """

    def __init__(self, consumer_key, consumer_secret):
        self.consumer_key = consumer_key
        self.consumer_secret = consumer_secret
        self._bearer_token = ''

        resp = requests.post(
            'https://api.twitter.com/oauth2/token',
            auth=(self.consumer_key, self.consumer_secret),
            data={'grant_type': 'client_credentials'}
        )
        data = resp.json()
        if data.get('token_type') != 'bearer':
            raise TweepyException('Expected token_type to equal "bearer", '
                                  f'but got {data.get("token_type")} instead')

        self._bearer_token = data['access_token']

    def apply_auth(self):
        return OAuth2BearerHandler(self._bearer_token)


class AppAuthHandler(OAuth2AppHandler):
    """Alias for :class:`OAuth2AppHandler`

    .. deprecated:: 4.5
        Use :class:`OAuth2AppHandler` instead.
    """

    def __init__(self, consumer_key, consumer_secret):
        warnings.warn(
            "AppAuthHandler is deprecated; use OAuth2AppHandler instead.",
            DeprecationWarning
        )
        super().__init__(consumer_key, consumer_secret)


class OAuth2BearerHandler(AuthBase):
    """OAuth 2.0 Bearer Token (App-Only) authentication handler

    .. versionadded:: 4.5
    """

    def __init__(self, bearer_token):
        self.bearer_token = bearer_token

    def __call__(self, request):
        request.headers['Authorization'] = 'Bearer ' + self.bearer_token
        return request

    def apply_auth(self):
        return self


class OAuth2UserHandler(OAuth2Session):
    """OAuth 2.0 Authorization Code Flow with PKCE (User Context)
    authentication handler

    .. versionadded:: 4.5
    """

    def __init__(self, *, client_id, redirect_uri, scope, client_secret=None):
        super().__init__(client_id, redirect_uri=redirect_uri, scope=scope)
        if client_secret is not None:
            self.auth = HTTPBasicAuth(client_id, client_secret)
        else:
            self.auth = None

    def get_authorization_url(self):
        """Get the authorization URL to redirect the user to"""
        authorization_url, state = self.authorization_url(
            "https://twitter.com/i/oauth2/authorize",
            code_challenge=self._client.create_code_challenge(
                self._client.create_code_verifier(128), "S256"
            ), code_challenge_method="S256"
        )
        return authorization_url

    def fetch_token(self, authorization_response):
        """After user has authorized the app, fetch access token with
        authorization response URL
        """
        return super().fetch_token(
            "https://api.twitter.com/2/oauth2/token",
            authorization_response=authorization_response,
            auth=self.auth,
            include_client_id=True,
            code_verifier=self._client.code_verifier
        )