-
Notifications
You must be signed in to change notification settings - Fork 22
/
Copy pathmain.py
94 lines (70 loc) · 2.37 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
import base64
import hashlib
import os
import re
import json
import requests
import redis
from requests_oauthlib import OAuth2Session
from flask import Flask, redirect, session
r = redis.from_url(os.environ["REDIS_URL_DOGS"])
app = Flask(__name__)
app.secret_key = os.urandom(50)
client_id = os.environ.get("CLIENT_ID")
client_secret = os.environ.get("CLIENT_SECRET")
auth_url = "https://twitter.com/i/oauth2/authorize"
token_url = "https://api.twitter.com/2/oauth2/token"
redirect_uri = os.environ.get("REDIRECT_URI")
# Set the scopes
scopes = ["tweet.read", "users.read", "tweet.write", "offline.access"]
# Create a code verifier
code_verifier = base64.urlsafe_b64encode(os.urandom(30)).decode("utf-8")
code_verifier = re.sub("[^a-zA-Z0-9]+", "", code_verifier)
# Create a code challenge
code_challenge = hashlib.sha256(code_verifier.encode("utf-8")).digest()
code_challenge = base64.urlsafe_b64encode(code_challenge).decode("utf-8")
code_challenge = code_challenge.replace("=", "")
def make_token():
return OAuth2Session(client_id, redirect_uri=redirect_uri, scope=scopes)
def parse_dog_fact():
url = "http://dog-api.kinduff.com/api/facts"
dog_fact = requests.request("GET", url).json()
return dog_fact["facts"][0]
def post_tweet(payload, token):
print("Tweeting!")
return requests.request(
"POST",
"https://api.twitter.com/2/tweets",
json=payload,
headers={
"Authorization": "Bearer {}".format(token["access_token"]),
"Content-Type": "application/json",
},
)
@app.route("/")
def demo():
global twitter
twitter = make_token()
authorization_url, state = twitter.authorization_url(
auth_url, code_challenge=code_challenge, code_challenge_method="S256"
)
session["oauth_state"] = state
return redirect(authorization_url)
@app.route("/oauth/callback", methods=["GET"])
def callback():
code = request.args.get("code")
token = twitter.fetch_token(
token_url=token_url,
client_secret=client_secret,
code_verifier=code_verifier,
code=code,
)
st_token = '"{}"'.format(token)
j_token = json.loads(st_token)
r.set("token", j_token)
doggie_fact = parse_dog_fact()
payload = {"text": "{}".format(doggie_fact)}
response = post_tweet(payload, token).json()
return response
if __name__ == "__main__":
app.run()