<commit_before>"""Twitter Authentication Views"""
from urlparse import parse_qs

import oauth2 as oauth
import requests

from pyramid.httpexceptions import HTTPFound
from pyramid.settings import asbool

from velruse.api import AuthenticationComplete
from velruse.exceptions import AuthenticationDenied
from velruse.exceptions import ThirdPartyFailure


REQUEST_URL = 'https://api.twitter.com/oauth/request_token'
ACCESS_URL = 'https://api.twitter.com/oauth/access_token'


class TwitterAuthenticationComplete(AuthenticationComplete):
    """Twitter auth complete"""


def includeme(config):
    config.add_route("twitter_login", "/twitter/login")
    config.add_route("twitter_process", "/twitter/process",
                     use_global_views=True,
                     factory=twitter_process)
    config.add_view(twitter_login, route_name="twitter_login")


def twitter_login(request):
    """Initiate a Twitter login"""
    config = request.registry.settings

    # Create the consumer and client, make the request
    consumer = oauth.Consumer(config['velruse.twitter.consumer_key'],
                              config['velruse.twitter.consumer_secret'])
    sigmethod = oauth.SignatureMethod_HMAC_SHA1()
    params = {'oauth_callback': request.route_url('twitter_process')}

    # We go through some shennanigans here to specify a callback url
    oauth_request = oauth.Request.from_consumer_and_token(consumer,
        http_url=REQUEST_URL, parameters=params)
    oauth_request.sign_request(sigmethod, consumer, None)
    r = requests.get(REQUEST_URL, headers=oauth_request.to_header())

    if r.status_code != 200:
        raise ThirdPartyFailure("Status %s: %s" % (r.status_code, r.content))
    request_token = oauth.Token.from_string(r.content)

    request.session['token'] = r.content

    # Send the user to twitter now for authorization
    if asbool(config.get('velruse.twitter.authorize')):
        req_url = 'https://api.twitter.com/oauth/authorize'
    else:
        req_url = 'https://api.twitter.com/oauth/authenticate'
    oauth_request = oauth.Request.from_token_and_callback(
        token=request_token, http_url=req_url)
    return HTTPFound(location=oauth_request.to_url())


def twitter_process(request):
    """Process the Twitter redirect"""
    if 'denied' in request.GET:
        return AuthenticationDenied("User denied authentication")

    config = request.registry.settings
    request_token = oauth.Token.from_string(request.session['token'])
    verifier = request.GET.get('oauth_verifier')
    if not verifier:
        raise ThirdPartyFailure("Oauth verifier not returned")
    request_token.set_verifier(verifier)

    # Create the consumer and client, make the request
    consumer = oauth.Consumer(config['velruse.twitter.consumer_key'],
                              config['velruse.twitter.consumer_secret'])

    client = oauth.Client(consumer, request_token)
    resp, content = client.request(ACCESS_URL, "POST")
    if resp['status'] != '200':
        raise ThirdPartyFailure("Status %s: %s" % (resp['status'], content))
    access_token = dict(parse_qs(content))

    # Setup the normalized contact info
    profile = {}
    profile['accounts'] = [{
        'domain':'twitter.com',
        'userid':access_token['user_id'][0]
    }]
    profile['displayName'] = access_token['screen_name'][0]

    cred = {'oauthAccessToken': access_token['oauth_token'][0],
            'oauthAccessTokenSecret': access_token['oauth_token_secret'][0]}
    return TwitterAuthenticationComplete(profile=profile,
                                         credentials=cred)
<commit_msg>Renamed "config" var to "settings" for clarity.<commit_after>"""Twitter Authentication Views"""
from urlparse import parse_qs

import oauth2 as oauth
import requests

from pyramid.httpexceptions import HTTPFound
from pyramid.settings import asbool

from velruse.api import AuthenticationComplete
from velruse.exceptions import AuthenticationDenied
from velruse.exceptions import ThirdPartyFailure


REQUEST_URL = 'https://api.twitter.com/oauth/request_token'
ACCESS_URL = 'https://api.twitter.com/oauth/access_token'


class TwitterAuthenticationComplete(AuthenticationComplete):
    """Twitter auth complete"""


def includeme(config):
    config.add_route("twitter_login", "/twitter/login")
    config.add_route("twitter_process", "/twitter/process",
                     use_global_views=True,
                     factory=twitter_process)
    config.add_view(twitter_login, route_name="twitter_login")


def twitter_login(request):
    """Initiate a Twitter login"""
    settings = request.registry.settings

    # Create the consumer and client, make the request
    consumer = oauth.Consumer(settings['velruse.twitter.consumer_key'],
                              settings['velruse.twitter.consumer_secret'])
    sigmethod = oauth.SignatureMethod_HMAC_SHA1()
    params = {'oauth_callback': request.route_url('twitter_process')}

    # We go through some shennanigans here to specify a callback url
    oauth_request = oauth.Request.from_consumer_and_token(consumer,
        http_url=REQUEST_URL, parameters=params)
    oauth_request.sign_request(sigmethod, consumer, None)
    r = requests.get(REQUEST_URL, headers=oauth_request.to_header())

    if r.status_code != 200:
        raise ThirdPartyFailure("Status %s: %s" % (r.status_code, r.content))
    request_token = oauth.Token.from_string(r.content)

    request.session['token'] = r.content

    # Send the user to twitter now for authorization
    if asbool(settings.get('velruse.twitter.authorize')):
        req_url = 'https://api.twitter.com/oauth/authorize'
    else:
        req_url = 'https://api.twitter.com/oauth/authenticate'
    oauth_request = oauth.Request.from_token_and_callback(
        token=request_token, http_url=req_url)
    return HTTPFound(location=oauth_request.to_url())


def twitter_process(request):
    """Process the Twitter redirect"""
    if 'denied' in request.GET:
        return AuthenticationDenied("User denied authentication")

    settings = request.registry.settings
    request_token = oauth.Token.from_string(request.session['token'])
    verifier = request.GET.get('oauth_verifier')
    if not verifier:
        raise ThirdPartyFailure("Oauth verifier not returned")
    request_token.set_verifier(verifier)

    # Create the consumer and client, make the request
    consumer = oauth.Consumer(settings['velruse.twitter.consumer_key'],
                              settings['velruse.twitter.consumer_secret'])

    client = oauth.Client(consumer, request_token)
    resp, content = client.request(ACCESS_URL, "POST")
    if resp['status'] != '200':
        raise ThirdPartyFailure("Status %s: %s" % (resp['status'], content))
    access_token = dict(parse_qs(content))

    # Setup the normalized contact info
    profile = {}
    profile['accounts'] = [{
        'domain':'twitter.com',
        'userid':access_token['user_id'][0]
    }]
    profile['displayName'] = access_token['screen_name'][0]

    cred = {'oauthAccessToken': access_token['oauth_token'][0],
            'oauthAccessTokenSecret': access_token['oauth_token_secret'][0]}
    return TwitterAuthenticationComplete(profile=profile,
                                         credentials=cred)

{'fraction_non_alphanumeric': 0.10944795385883, 'fraction_numerical': 0.0032957978577313925, 'mean_word_length': 3.82317880794702, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 8, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4095040', 'n_tokens_mistral': 2137, 'n_tokens_neox': 2049, 'n_words': 422}