A tool to retrieve information from Twitch.
fix: changed the iteration over followed streams
+74
-48
+6
-2
README.md
+6
-2
README.md
···
1
1
# purple-cli
2
2
3
-
CLI tool to retrieve information from Twitch.
3
+
`purple-cli` is a tool to retrieve information from Twitch, in a format that can be
4
+
used to integrate with other scripts or applications.
5
+
6
+
It requires to have an application created in Twitch, with a client id and a client
7
+
secret that will be use to obtain the necessary credentials.
4
8
5
9
## Settings
6
10
7
-
This project requires the following environment variables to work:
11
+
This tool requires the following environment variables to work:
8
12
9
13
- `PURPLE_CLIENT_ID` Twitch application client id.
10
14
- `PURPLE_CLIENT_SECRET` Twitch application client secret.
+1
-1
pyproject.toml
+1
-1
pyproject.toml
+62
-41
src/purple/api.py
+62
-41
src/purple/api.py
···
10
10
logger = logging.getLogger(__name__)
11
11
12
12
13
-
def twitch_api_client(access_token: str) -> httpx.Client:
14
-
"""Create a simple client to access Twitch API."""
15
-
base_url = "https://api.twitch.tv"
16
-
headers = {
17
-
"Client-ID": settings.client_id,
18
-
"Authorization": f"Bearer {access_token}",
19
-
}
20
-
return httpx.Client(base_url=base_url, headers=headers)
13
+
# Common
14
+
# --------------------------------------------------------------------------------------
21
15
22
16
23
17
def raise_for_status(response: httpx.Response) -> None:
···
30
24
response.raise_for_status()
31
25
32
26
27
+
# Twitch ID
28
+
# --------------------------------------------------------------------------------------
29
+
30
+
31
+
def twitch_id_client() -> httpx.Client:
32
+
"""Create a simple client to access Twitch ID API."""
33
+
base_url = "https://id.twitch.tv"
34
+
return httpx.Client(base_url=base_url)
35
+
36
+
33
37
def retrieve_token(code: str) -> dict:
34
38
"""Retrieve the token information from Twitch using the given code."""
35
39
logger.debug(f"Obtaining token information using code {code}")
36
-
token_url = "https://id.twitch.tv/oauth2/token"
37
40
params = {
38
41
"client_id": settings.client_id,
39
42
"client_secret": settings.client_secret,
···
41
44
"grant_type": "authorization_code",
42
45
"redirect_uri": settings.redirect_uri,
43
46
}
44
-
45
-
response = httpx.post(token_url, params=params)
46
-
raise_for_status(response)
47
+
with twitch_id_client() as client:
48
+
response = client.post("/oauth2/token", params=params)
49
+
raise_for_status(response)
47
50
48
51
return response.json()
49
52
···
60
63
return f"{authorize_url}?{urllib.parse.urlencode(params)}"
61
64
62
65
66
+
def validate_access_token(access_token: str) -> bool:
67
+
"""Validate the given access token."""
68
+
headers = {"Authorization": f"OAuth {access_token}"}
69
+
with twitch_id_client() as client:
70
+
response = client.get("/oauth2/validate", headers=headers)
71
+
return response.status_code == 200
72
+
73
+
74
+
def refresh_access_token(refresh_token: str) -> dict:
75
+
"""Refresh the access token."""
76
+
params = {
77
+
"grant_type": "refresh_token",
78
+
"refresh_token": refresh_token,
79
+
"client_id": settings.client_id,
80
+
"client_secret": settings.client_secret,
81
+
}
82
+
83
+
with twitch_id_client() as client:
84
+
response = client.post("/oauth2/token", params=params)
85
+
raise_for_status(response)
86
+
87
+
data = response.json()
88
+
return data
89
+
90
+
91
+
# Twitch API
92
+
# --------------------------------------------------------------------------------------
93
+
94
+
95
+
def twitch_api_client(access_token: str) -> httpx.Client:
96
+
"""Create a simple client to access Twitch API."""
97
+
base_url = "https://api.twitch.tv"
98
+
headers = {
99
+
"Client-ID": settings.client_id,
100
+
"Authorization": f"Bearer {access_token}",
101
+
}
102
+
return httpx.Client(base_url=base_url, headers=headers)
103
+
104
+
63
105
def retrieve_user(access_token: str) -> dict:
64
106
"""Retrieve the user information."""
65
107
with twitch_api_client(access_token) as client:
···
109
151
return followed
110
152
111
153
112
-
def retireve_followed_streams(access_token: str, user_id: str) -> list[dict]:
154
+
def retrieve_followed_streams(access_token: str, user_id: str) -> list[dict]:
113
155
"""Retrieve the list of followed streams."""
114
156
page_size = 100
115
-
params = {
157
+
params: dict[str, str | int] = {
116
158
"user_id": user_id,
117
159
"first": page_size,
118
160
}
119
161
120
162
streams: list[dict] = []
121
-
total: int | None = None
163
+
first_query: bool = True
164
+
cursor: str | None = None
122
165
123
-
while total is None or len(streams) > total:
166
+
while first_query or cursor:
124
167
with twitch_api_client(access_token) as client:
125
-
response = client.get("/helix/streams/followed")
168
+
response = client.get("/helix/streams/followed", params=params)
126
169
raise_for_status(response)
127
170
data = response.json()
128
171
129
-
if total is None:
130
-
total = data["total"]
172
+
if first_query:
173
+
first_query = False
131
174
132
175
streams.extend(data["data"])
133
176
···
136
179
params["after"] = cursor
137
180
138
181
return streams
139
-
140
-
141
-
def validate_access_token(access_token: str) -> bool:
142
-
"""Validate the given access token."""
143
-
headers = {"Authorization": f"OAuth {access_token}"}
144
-
response = httpx.get("https://id.twitch.tv/oauth2/validate", headers=headers)
145
-
return response.status_code == 200
146
-
147
-
148
-
def refresh_access_token(refresh_token: str) -> dict:
149
-
"""Refresh the access token."""
150
-
params = {
151
-
"grant_type": "refresh_token",
152
-
"refresh_token": refresh_token,
153
-
"client_id": settings.client_id,
154
-
"client_secret": settings.client_secret,
155
-
}
156
-
response = httpx.post("https://id.twitch.tv/oauth2/token", params=params)
157
-
raise_for_status(response)
158
-
159
-
data = response.json()
160
-
return data
+1
-1
src/purple/auth.py
+1
-1
src/purple/auth.py
···
45
45
self.access_token_complete = token_data
46
46
else:
47
47
logger.error("Access token not found in response.")
48
-
return web.Response(text="⚠️ Error, no se encuentra el parametro code.")
48
+
return web.Response(text="⚠️ Error, access token can't be obtained.")
49
49
50
50
# event to close the server
51
51
logger.debug("Sending event to close web server...")
+4
-3
src/purple/cli.py
+4
-3
src/purple/cli.py
···
9
9
10
10
from .api import (
11
11
retrieve_followed_channels,
12
+
retrieve_followed_streams,
12
13
retrieve_user,
13
14
)
14
15
from .auth import obtain_access_token
···
51
52
user_id = user["id"]
52
53
53
54
# get followed channels
54
-
followed = retrieve_followed_channels(access_token=access_token, user_id=user_id)
55
+
streams = retrieve_followed_streams(access_token=access_token, user_id=user_id)
55
56
56
-
print(json.dumps(followed, indent=2))
57
+
print(json.dumps(streams, indent=2))
57
58
58
59
59
60
def main():
···
68
69
action="store_true",
69
70
)
70
71
parser.add_argument("-V", "--version", help="show version", action="store_true")
71
-
parser.set_defaults(func=followed)
72
+
parser.set_defaults(func=live)
72
73
73
74
args = parser.parse_args()
74
75