Skip to content

Commit c82bf4c

Browse files
FoamyGuyFoamyGuy
authored
Merge pull request #121 from DJDevon3/main
Add Twitch API to requests examples
2 parents 640bfac + 0c12ada commit c82bf4c

File tree

2 files changed

+295
-0
lines changed

2 files changed

+295
-0
lines changed

‎examples/requests_api_mastodon.py

+119
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
# SPDX-FileCopyrightText: 2022 DJDevon3
2+
# SPDX-License-Identifier: MIT
3+
# Coded for Circuit Python 8.0
4+
"""DJDevon3 Adafruit Feather ESP32-S2 Mastodon_API_Example"""
5+
import gc
6+
import time
7+
import ssl
8+
import wifi
9+
import socketpool
10+
import adafruit_requests
11+
12+
# Mastodon V1 API - Public access (no dev creds or app required)
13+
# Visit https://docs.joinmastodon.org/client/public/ for API docs
14+
# For finding your Mastodon User ID
15+
# Login to your mastodon server in a browser, visit your profile, UserID is in the URL.
16+
# Example: https://mastodon.YOURSERVER/web/accounts/YOURUSERIDISHERE
17+
18+
Mastodon_Server = "mastodon.social" # Set server instance
19+
Mastodon_UserID = "000000000000000000" # Set User ID you want endpoints from
20+
# Test in browser first, this will pull up a JSON webpage
21+
# https://mastodon.YOURSERVER/api/v1/accounts/YOURUSERIDHERE/statuses?limit=1
22+
23+
# Initialize WiFi Pool (There can be only 1 pool & top of script)
24+
pool = socketpool.SocketPool(wifi.radio)
25+
26+
# Time between API refreshes
27+
# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour
28+
sleep_time = 900
29+
30+
try:
31+
from secrets import secrets
32+
except ImportError:
33+
print("Secrets File Import Error")
34+
raise
35+
36+
# Converts seconds in minutes/hours/days
37+
def time_calc(input_time):
38+
if input_time < 60:
39+
sleep_int = input_time
40+
time_output = f"{sleep_int:.0f} seconds"
41+
elif 60 <= input_time < 3600:
42+
sleep_int = input_time / 60
43+
time_output = f"{sleep_int:.0f} minutes"
44+
elif 3600 <= input_time < 86400:
45+
sleep_int = input_time / 60 / 60
46+
time_output = f"{sleep_int:.0f} hours"
47+
elif 86400 <= input_time < 432000:
48+
sleep_int = input_time / 60 / 60 / 24
49+
time_output = f"{sleep_int:.1f} days"
50+
else: # if > 5 days convert float to int & display whole days
51+
sleep_int = input_time / 60 / 60 / 24
52+
time_output = f"{sleep_int:.0f} days"
53+
return time_output
54+
55+
56+
# Publicly available data no header required
57+
MAST_SOURCE = (
58+
"https://"
59+
+ Mastodon_Server
60+
+ "/api/v1/accounts/"
61+
+ Mastodon_UserID
62+
+ "/statuses?limit=1"
63+
)
64+
65+
# Connect to Wi-Fi
66+
print("\n===============================")
67+
print("Connecting to WiFi...")
68+
requests = adafruit_requests.Session(pool, ssl.create_default_context())
69+
while not wifi.radio.ipv4_address:
70+
try:
71+
wifi.radio.connect(secrets["ssid"], secrets["password"])
72+
except ConnectionError as e:
73+
print("Connection Error:", e)
74+
print("Retrying in 10 seconds")
75+
time.sleep(10)
76+
gc.collect()
77+
print("Connected!\n")
78+
79+
while True:
80+
try:
81+
print("\nAttempting to GET MASTODON Stats!") # -----------------------------
82+
# Print Request to Serial
83+
debug_mastodon_full_response = (
84+
False # STREAMER WARNING: your client secret will be viewable
85+
)
86+
print("===============================")
87+
mastodon_response = requests.get(url=MAST_SOURCE)
88+
try:
89+
mastodon_json = mastodon_response.json()
90+
except ConnectionError as e:
91+
print("Connection Error:", e)
92+
print("Retrying in 10 seconds")
93+
mastodon_json = mastodon_json[0]
94+
if debug_mastodon_full_response:
95+
print("Full API GET URL: ", MAST_SOURCE)
96+
print(mastodon_json)
97+
mastodon_userid = mastodon_json["account"]["id"]
98+
print("User ID: ", mastodon_userid)
99+
100+
mastodon_username = mastodon_json["account"]["display_name"]
101+
print("Name: ", mastodon_username)
102+
mastodon_join_date = mastodon_json["account"]["created_at"]
103+
print("Member Since: ", mastodon_join_date)
104+
mastodon_toot_count = mastodon_json["account"]["statuses_count"]
105+
print("Toots: ", mastodon_toot_count)
106+
mastodon_follower_count = mastodon_json["account"]["followers_count"]
107+
print("Followers: ", mastodon_follower_count)
108+
print("Monotonic: ", time.monotonic())
109+
110+
print("\nFinished!")
111+
print("Next Update in: ", time_calc(sleep_time))
112+
print("===============================")
113+
gc.collect()
114+
115+
except (ValueError, RuntimeError) as e:
116+
print("Failed to get data, retrying\n", e)
117+
time.sleep(60)
118+
continue
119+
time.sleep(sleep_time)

‎examples/requests_api_twitch.py

+176
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
# SPDX-FileCopyrightText: 2022 DJDevon3 for Adafruit Industries
2+
# SPDX-License-Identifier: MIT
3+
# Coded for Circuit Python 8.0
4+
"""DJDevon3 Adafruit Feather ESP32-S2 Twitch_API_Example"""
5+
import gc
6+
import time
7+
import ssl
8+
import wifi
9+
import socketpool
10+
import adafruit_requests
11+
12+
# Twitch Developer Account & 0Auth App Required:
13+
# Visit https://dev.twitch.tv/console to create an app
14+
# Ensure Twitch_ClientID & Twitch_Client_Secret are in secrets.py or .env
15+
16+
# "Twitch_ClientID": "Your Developer APP ID Here",
17+
# "Twitch_Client_Secret": "APP ID secret here",
18+
19+
# For finding your Twitch User ID
20+
# https://www.streamweasels.com/tools/convert-twitch-username-to-user-id/
21+
Twitch_UserID = "0000000" # Set User ID you want endpoints from
22+
23+
# Initialize WiFi Pool (There can be only 1 pool & top of script)
24+
pool = socketpool.SocketPool(wifi.radio)
25+
26+
# Time between API refreshes
27+
# 900 = 15 mins, 1800 = 30 mins, 3600 = 1 hour
28+
sleep_time = 900
29+
30+
try:
31+
from secrets import secrets
32+
except ImportError:
33+
print("Secrets File Import Error")
34+
raise
35+
36+
# Converts seconds in minutes/hours/days
37+
def time_calc(input_time):
38+
if input_time < 60:
39+
sleep_int = input_time
40+
time_output = f"{sleep_int:.0f} seconds"
41+
elif 60 <= input_time < 3600:
42+
sleep_int = input_time / 60
43+
time_output = f"{sleep_int:.0f} minutes"
44+
elif 3600 <= input_time < 86400:
45+
sleep_int = input_time / 60 / 60
46+
time_output = f"{sleep_int:.0f} hours"
47+
elif 86400 <= input_time < 432000:
48+
sleep_int = input_time / 60 / 60 / 24
49+
time_output = f"{sleep_int:.1f} days"
50+
else: # if > 5 days convert float to int & display whole days
51+
sleep_int = input_time / 60 / 60 / 24
52+
time_output = f"{sleep_int:.0f} days"
53+
return time_output
54+
55+
56+
# First we use Client ID & Client Secret to create a token with POST
57+
# No user interaction is required for this type of scope (implicit grant flow)
58+
twitch_0auth_header = {"Content-Type": "application/x-www-form-urlencoded"}
59+
TWITCH_0AUTH_TOKEN = "https://id.twitch.tv/oauth2/token"
60+
61+
# Connect to Wi-Fi
62+
print("\n===============================")
63+
print("Connecting to WiFi...")
64+
requests = adafruit_requests.Session(pool, ssl.create_default_context())
65+
while not wifi.radio.ipv4_address:
66+
try:
67+
wifi.radio.connect(secrets["ssid"], secrets["password"])
68+
except ConnectionError as e:
69+
print("Connection Error:", e)
70+
print("Retrying in 10 seconds")
71+
time.sleep(10)
72+
gc.collect()
73+
print("Connected!\n")
74+
75+
while True:
76+
try:
77+
# ----------------------------- POST FOR BEARER TOKEN -----------------------
78+
print(
79+
"\nAttempting to GENERATE Twitch Bearer Token!"
80+
) # ---------------------------------------
81+
# Print Request to Serial
82+
debug_bearer_request = (
83+
False # STREAMER WARNING: your client secret will be viewable
84+
)
85+
if debug_bearer_request:
86+
print("Full API GET URL: ", TWITCH_0AUTH_TOKEN)
87+
print("===============================")
88+
twitch_0auth_data = (
89+
"&client_id="
90+
+ secrets["Twitch_ClientID"]
91+
+ "&client_secret="
92+
+ secrets["Twitch_Client_Secret"]
93+
+ "&grant_type=client_credentials"
94+
)
95+
96+
# POST REQUEST
97+
twitch_0auth_response = requests.post(
98+
url=TWITCH_0AUTH_TOKEN, data=twitch_0auth_data, headers=twitch_0auth_header
99+
)
100+
try:
101+
twitch_0auth_json = twitch_0auth_response.json()
102+
twitch_access_token = twitch_0auth_json["access_token"]
103+
except ConnectionError as e:
104+
print("Connection Error:", e)
105+
print("Retrying in 10 seconds")
106+
107+
# Print Response to Serial
108+
debug_bearer_response = (
109+
False # STREAMER WARNING: your client secret will be viewable
110+
)
111+
if debug_bearer_response:
112+
print("JSON Dump: ", twitch_0auth_json)
113+
print("Header: ", twitch_0auth_header)
114+
print("Access Token: ", twitch_access_token)
115+
116+
twitch_token_expiration = twitch_0auth_json["expires_in"]
117+
print("Token Expires in: ", time_calc(twitch_token_expiration))
118+
twitch_token_type = twitch_0auth_json["token_type"]
119+
print("Token Type: ", twitch_token_type)
120+
print("Monotonic: ", time.monotonic())
121+
122+
# ----------------------------- GET DATA -------------------------------------
123+
# Bearer token is refreshed every time script runs :)
124+
# Twitch sets token expiration to about 64 days
125+
# Helix is the name of the current Twitch API
126+
# Now that we have POST bearer token we can do a GET for data
127+
# ----------------------------------------------------------------------------
128+
twitch_header = {
129+
"Authorization": "Bearer " + twitch_access_token + "",
130+
"Client-Id": "" + secrets["Twitch_ClientID"] + "",
131+
}
132+
TWITCH_FOLLOWERS_SOURCE = (
133+
"https://api.twitch.tv/helix/users"
134+
+ "/follows?"
135+
+ "to_id="
136+
+ Twitch_UserID
137+
+ "&first=1"
138+
)
139+
print(
140+
"\nAttempting to GET TWITCH Stats!"
141+
) # ------------------------------------------------
142+
print("===============================")
143+
twitch_followers_response = requests.get(
144+
url=TWITCH_FOLLOWERS_SOURCE, headers=twitch_header
145+
)
146+
try:
147+
twitch_followers_json = twitch_followers_response.json()
148+
except ConnectionError as e:
149+
print("Connection Error:", e)
150+
print("Retrying in 10 seconds")
151+
152+
# Print Response to Serial
153+
debug_bearer_response = (
154+
False # STREAMER WARNING: your bearer token will be viewable
155+
)
156+
if debug_bearer_response:
157+
print("Full API GET URL: ", TWITCH_FOLLOWERS_SOURCE)
158+
print("Header: ", twitch_header)
159+
print("JSON Full Response: ", twitch_followers_json)
160+
161+
twitch_username = twitch_followers_json["data"][0]["to_name"]
162+
print("Username: ", twitch_username)
163+
twitch_followers = twitch_followers_json["total"]
164+
print("Followers: ", twitch_followers)
165+
print("Monotonic: ", time.monotonic()) # Board Up-Time seconds
166+
167+
print("\nFinished!")
168+
print("Next Update in: ", time_calc(sleep_time))
169+
print("===============================")
170+
gc.collect()
171+
172+
except (ValueError, RuntimeError) as e:
173+
print("Failed to get data, retrying\n", e)
174+
time.sleep(60)
175+
continue
176+
time.sleep(sleep_time)

0 commit comments

Comments
 (0)