A tool to retrieve information from Twitch.
feat: popular streams option
+162
-34
+29
-1
README.md
+29
-1
README.md
···
6
6
It requires to have an application created in Twitch, with a client id and a client
7
7
secret that will be use to obtain the necessary credentials.
8
8
9
+
## Install
10
+
11
+
The recommended way to install purple is using [pipx](https://pipx.pypa.io/latest/installation/).
12
+
This could be done by using directly the repository URL:
13
+
14
+
```bash
15
+
pipx install git+https://github.com/marcosgabarda/purple-cli
16
+
```
17
+
18
+
This will add `purple` command to your shell:
19
+
20
+
```bash
21
+
$ purple -h
22
+
usage: purple [-h] [-v] [-V] [--popular [POPULAR]] [-l [LANG]]
23
+
24
+
Get the list of live streams from the list of following channels in Twitch.
25
+
26
+
options:
27
+
-h, --help show this help message and exit
28
+
-v, --verbose increase output verbosity
29
+
-V, --version show version
30
+
--popular [POPULAR] get the list of live streams with most viewers (top 20 by default).
31
+
-l, --lang [LANG] filter the list of live streams by language.
32
+
```
33
+
9
34
## Settings
10
35
11
-
This tool requires the following environment variables to work:
36
+
`purple-cli` requires uses a [Twitch application](https://dev.twitch.tv/console/apps/create),
37
+
and to take from there the credentials.
38
+
39
+
The following environment variables can be used to set the credentials:
12
40
13
41
- `PURPLE_CLIENT_ID` Twitch application client id.
14
42
- `PURPLE_CLIENT_SECRET` Twitch application client secret.
+29
-11
src/purple/api.py
+29
-11
src/purple/api.py
···
132
132
if total is None:
133
133
total = data["total"]
134
134
135
-
followed.extend(
136
-
[
137
-
{
138
-
"name": channel["broadcaster_name"],
139
-
"login": channel["broadcaster_login"],
140
-
"id": channel["broadcaster_id"],
141
-
"followed_at": channel["followed_at"],
142
-
}
143
-
for channel in data["data"]
144
-
]
145
-
)
135
+
followed.extend(data["data"])
146
136
147
137
cursor = data["pagination"].get("cursor")
148
138
if cursor:
···
179
169
params["after"] = cursor
180
170
181
171
return streams
172
+
173
+
174
+
def retrieve_live_streams(
175
+
access_token: str,
176
+
language: list[str] | None = None,
177
+
size: int | None = None,
178
+
) -> list[dict]:
179
+
"""Retrieve the list of live stream, sorted by viewers.
180
+
181
+
It doesn't relay in pagination, because the number of viewers can change between
182
+
calls, and therefore, it could generate duplicated results.
183
+
"""
184
+
params: dict[str, str | int | list[str]] = {
185
+
"type": "live",
186
+
}
187
+
if language:
188
+
params["language"] = language
189
+
if size:
190
+
if size > 100:
191
+
logger.warning("Twitch API supports maximum of 100 items.")
192
+
params["first"] = size
193
+
194
+
with twitch_api_client(access_token) as client:
195
+
response = client.get("/helix/streams", params=params)
196
+
raise_for_status(response)
197
+
data = response.json()
198
+
199
+
return data["data"]
+73
-22
src/purple/cli.py
+73
-22
src/purple/cli.py
···
5
5
import logging
6
6
import sys
7
7
8
+
from pydantic import ValidationError
9
+
8
10
import purple
9
11
10
12
from .api import (
11
-
retrieve_followed_channels,
12
13
retrieve_followed_streams,
14
+
retrieve_live_streams,
13
15
retrieve_user,
14
16
)
15
17
from .auth import obtain_access_token
18
+
from .settings import settings
16
19
17
20
logger = logging.getLogger(__name__)
18
21
19
22
20
-
def followed(args: argparse.Namespace) -> None:
23
+
def live_followed(languages: list[str] | None = None) -> str:
21
24
"""List the channels the user follows."""
22
-
# logger
23
-
if args.verbose:
24
-
logging.basicConfig(level=logging.DEBUG)
25
-
26
25
# obtain access token
27
26
access_token = obtain_access_token()
28
27
···
32
31
user_id = user["id"]
33
32
34
33
# get followed channels
35
-
followed = retrieve_followed_channels(access_token=access_token, user_id=user_id)
34
+
streams = retrieve_followed_streams(access_token=access_token, user_id=user_id)
36
35
37
-
print(json.dumps(followed, indent=2))
36
+
# filter streams by language
37
+
if languages:
38
+
streams = [stream for stream in streams if stream["language"] in languages]
38
39
40
+
return json.dumps(streams, indent=2)
39
41
40
-
def live(args: argparse.Namespace) -> None:
41
-
"""List the channels the user follows."""
42
-
# logger
43
-
if args.verbose:
44
-
logging.basicConfig(level=logging.DEBUG)
45
42
43
+
def live_popular(size: int, languages: list[str] | None = None) -> str:
44
+
"""List the most popular channels."""
46
45
# obtain access token
47
46
access_token = obtain_access_token()
48
-
49
-
# get user id
50
-
user = retrieve_user(access_token=access_token)
51
-
logger.debug(f"User data:\n{user}")
52
-
user_id = user["id"]
53
47
54
48
# get followed channels
55
-
streams = retrieve_followed_streams(access_token=access_token, user_id=user_id)
49
+
streams = retrieve_live_streams(
50
+
access_token=access_token,
51
+
size=size,
52
+
language=languages,
53
+
)
56
54
57
-
print(json.dumps(streams, indent=2))
55
+
return json.dumps(streams, indent=2)
56
+
57
+
58
+
def parse_languages(value: str | None) -> list[str] | None:
59
+
"""Parse the popular argument."""
60
+
if not value or value == "all":
61
+
return None
62
+
return value.split(",")
63
+
64
+
65
+
def do_it(args: argparse.Namespace) -> None:
66
+
"""Execute the command."""
67
+
if not args.popular:
68
+
streams = live_followed(languages=args.lang)
69
+
else:
70
+
streams = live_popular(size=args.popular, languages=args.lang)
71
+
print(streams)
58
72
59
73
60
74
def main():
61
75
"""Execute main function to handle the CLI parameters and operations."""
62
76
parser = argparse.ArgumentParser(
63
-
description="Collects data from your Twitch account to use it in other scripts."
77
+
description=(
78
+
"Get the list of live streams from the list of following channels in "
79
+
"Twitch."
80
+
)
64
81
)
65
82
parser.add_argument(
66
83
"-v",
···
69
86
action="store_true",
70
87
)
71
88
parser.add_argument("-V", "--version", help="show version", action="store_true")
72
-
parser.set_defaults(func=live)
89
+
parser.add_argument(
90
+
"--popular",
91
+
type=int,
92
+
nargs="?",
93
+
const=20,
94
+
help="get the list of live streams with most viewers (top 20 by default).",
95
+
)
96
+
parser.add_argument(
97
+
"-l",
98
+
"--lang",
99
+
nargs="?",
100
+
const="all",
101
+
default="all",
102
+
type=parse_languages,
103
+
help="filter the list of live streams by language.",
104
+
)
105
+
106
+
parser.set_defaults(func=do_it)
73
107
74
108
args = parser.parse_args()
75
109
110
+
# debug logger
111
+
if args.verbose:
112
+
logging.basicConfig(level=logging.DEBUG)
113
+
76
114
# just print version
77
115
if args.version:
78
116
print(purple.__version__)
79
117
sys.exit(0)
118
+
119
+
# check settings
120
+
try:
121
+
settings.client_id
122
+
settings.client_secret
123
+
except ValidationError:
124
+
logger.error(
125
+
"\033[93mYou need to define the client ID and the client secret to execute "
126
+
"purple. This is done using the following environment variables:\033[0m\n\n"
127
+
" - \033[96mPURPLE_CLIENT_ID\033[0m\n"
128
+
" - \033[96mPURPLE_CLIENT_SECRET\033[0m"
129
+
)
130
+
sys.exit(-1)
80
131
81
132
args.func(args)
+31
src/purple/settings.py
+31
src/purple/settings.py
···
3
3
import os
4
4
from functools import lru_cache
5
5
from pathlib import Path
6
+
from typing import Any
6
7
7
8
from pydantic_settings import BaseSettings, SettingsConfigDict
8
9
9
10
11
+
def lazy_settings(cls: type) -> type:
12
+
"""Define a decorator for use a settings instance as a lazy object.
13
+
14
+
That it is only initialized when it is called the first time.
15
+
"""
16
+
17
+
class LazySettings:
18
+
"""Wrap a Settings class, to allow on need evaluation."""
19
+
20
+
_wrapped_settings_class: type[BaseSettings] = cls
21
+
_wrapped_settings: BaseSettings | None = None
22
+
23
+
def __getattr__(self, name: str) -> Any:
24
+
if not self._wrapped_settings:
25
+
self._wrapped_settings = self._wrapped_settings_class()
26
+
return getattr(self._wrapped_settings, name)
27
+
28
+
def __setattr__(self, name: str, value: Any) -> None:
29
+
if name in ("_wrapped_settings_class", "_wrapped_settings"):
30
+
self.__dict__[name] = value
31
+
else:
32
+
if not self._wrapped_settings:
33
+
self._wrapped_settings = self._wrapped_settings_class()
34
+
setattr(self._wrapped_settings, name, value)
35
+
36
+
return LazySettings
37
+
38
+
10
39
@lru_cache
11
40
def app_data_path() -> Path:
12
41
"""Retrieve the path to the default application data path."""
···
24
53
return Path(data_home)
25
54
26
55
56
+
@lazy_settings
27
57
class Settings(BaseSettings):
28
58
"""Configuration of the purple script."""
29
59
···
47
77
return app_data_path() / "auth.json"
48
78
49
79
80
+
# Lazy instance of settings to enable on need evaluation of the settings properties
50
81
settings = Settings()