A tool to retrieve information from Twitch.
1"""Module with CLI application."""
2
3import argparse
4import json
5import logging
6import sys
7
8from pydantic import ValidationError
9
10import purple
11
12from .api import (
13 retrieve_followed_streams,
14 retrieve_live_streams,
15 retrieve_user,
16)
17from .auth import obtain_access_token
18from .settings import settings
19
20logger = logging.getLogger(__name__)
21
22
23def live_followed(languages: list[str] | None = None) -> str:
24 """List the channels the user follows."""
25 # obtain access token
26 access_token = obtain_access_token()
27
28 # get user id
29 user = retrieve_user(access_token=access_token)
30 logger.debug(f"User data:\n{user}")
31 user_id = user["id"]
32
33 # get followed channels
34 streams = retrieve_followed_streams(access_token=access_token, user_id=user_id)
35
36 # filter streams by language
37 if languages:
38 streams = [stream for stream in streams if stream["language"] in languages]
39
40 return json.dumps(streams, indent=2)
41
42
43def live_popular(size: int, languages: list[str] | None = None) -> str:
44 """List the most popular channels."""
45 # obtain access token
46 access_token = obtain_access_token()
47
48 # get followed channels
49 streams = retrieve_live_streams(
50 access_token=access_token,
51 size=size,
52 language=languages,
53 )
54
55 return json.dumps(streams, indent=2)
56
57
58def 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
65def 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)
72
73
74def main():
75 """Execute main function to handle the CLI parameters and operations."""
76 parser = argparse.ArgumentParser(
77 description=(
78 "Get the list of live streams from the list of following channels in "
79 "Twitch."
80 )
81 )
82 parser.add_argument(
83 "-v",
84 "--verbose",
85 help="increase output verbosity",
86 action="store_true",
87 )
88 parser.add_argument("-V", "--version", help="show version", action="store_true")
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)
107
108 args = parser.parse_args()
109
110 # debug logger
111 if args.verbose:
112 logging.basicConfig(level=logging.DEBUG)
113
114 # just print version
115 if args.version:
116 print(purple.__version__)
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)
131
132 args.func(args)