Extract print_paged, improve types

This commit is contained in:
Ivan Habunek 2024-04-01 09:40:54 +02:00
parent 3ae99fe159
commit 3270d857b1
No known key found for this signature in database
GPG Key ID: F5F0623FF5EBCB3D
6 changed files with 62 additions and 81 deletions

View File

@ -5,8 +5,8 @@ import re
import sys
from twitchdl import __version__
from twitchdl.commands.clips import ClipsPeriod
from twitchdl.entities import DownloadOptions
from twitchdl.twitch import ClipsPeriod, VideosSort, VideosType
# Tweak the Click context
# https://click.palletsprojects.com/en/8.1.x/api/#context
@ -369,8 +369,8 @@ def videos(
json: bool,
limit: int | None,
pager: int | None,
sort: str,
type: str,
sort: VideosSort,
type: VideosType,
):
"""List or download clips for given CHANNEL_NAME."""
from twitchdl.commands.videos import videos

View File

@ -1,18 +1,17 @@
from typing import Generator
import click
import re
import sys
from typing import Literal
from itertools import islice
from os import path
import click
from twitchdl import twitch, utils
from twitchdl.commands.download import get_clip_authenticated_url
from twitchdl.download import download_file
from twitchdl.output import green, print_clip, print_json, yellow
from twitchdl.entities import Data
from twitchdl.output import green, print_clip, print_json, print_paged, yellow
ClipsPeriod = Literal["last_day", "last_week", "last_month", "all_time"]
def clips(
@ -23,7 +22,7 @@ def clips(
json: bool = False,
limit: int = 10,
pager: int | None = None,
period: ClipsPeriod = "all_time",
period: twitch.ClipsPeriod = "all_time",
):
# Ignore --limit if --pager or --all are given
limit = sys.maxsize if all or pager else limit
@ -37,7 +36,7 @@ def clips(
return _download_clips(generator)
if pager:
return _print_paged(generator, pager)
return print_paged("Clips", generator, print_clip, pager)
return _print_all(generator, all)
@ -55,7 +54,7 @@ def _continue():
return True
def _target_filename(clip):
def _target_filename(clip: Data):
url = clip["videoQualities"][0]["sourceURL"]
_, ext = path.splitext(url)
ext = ext.lstrip(".")
@ -75,7 +74,7 @@ def _target_filename(clip):
return f"{name}.{ext}"
def _download_clips(generator):
def _download_clips(generator: Generator[Data, None, None]):
for clip in generator:
target = _target_filename(clip)
@ -87,7 +86,7 @@ def _download_clips(generator):
download_file(url, target)
def _print_all(generator, all: bool):
def _print_all(generator: Generator[Data, None, None], all: bool):
for clip in generator:
click.echo()
print_clip(clip)
@ -98,31 +97,3 @@ def _print_all(generator, all: bool):
"Increase the --limit, use --all or --pager to see the rest.",
dim=True
)
def _print_paged(generator, page_size):
iterator = iter(generator)
page = list(islice(iterator, page_size))
first = 1
last = first + len(page) - 1
while True:
click.echo("-" * 80)
click.echo()
for clip in page:
print_clip(clip)
click.echo()
last = first + len(page) - 1
click.echo("-" * 80)
click.echo(f"Clips {first}-{last} of ???")
first = first + len(page)
last = first + 1
page = list(islice(iterator, page_size))
if not page or not _continue():
break

View File

@ -51,7 +51,6 @@ def video_info(video, playlists, chapters):
click.echo()
print_video(video)
click.echo()
click.echo("Playlists:")
for p in m3u8.loads(playlists).playlists:
click.echo(f"{bold(p.stream_info.video)} {p.uri}")

View File

@ -4,7 +4,7 @@ import click
from twitchdl import twitch
from twitchdl.exceptions import ConsoleError
from twitchdl.output import print_log, print_paged_videos, print_video, print_json, print_video_compact
from twitchdl.output import print_log, print_paged, print_video, print_json, print_video_compact
def videos(
@ -44,7 +44,8 @@ def videos(
return
if pager:
print_paged_videos(generator, pager, total_count)
print_fn = print_video_compact if compact else print_video
print_paged("Videos", generator, print_fn, pager, total_count)
return
count = 0

View File

@ -3,10 +3,12 @@ import json
from itertools import islice
from twitchdl import utils
from typing import Any, Generator
from typing import Any, Callable, Generator, TypeVar
from twitchdl.entities import Data
T = TypeVar("T")
def truncate(string: str, length: int) -> str:
if len(string) > length:
@ -42,6 +44,40 @@ def print_table(headers: list[str], data: list[list[str]]):
print_row(row)
def print_paged(
label: str,
generator: Generator[T, Any, Any],
print_fn: Callable[[T], None],
page_size: int,
total_count: int | None = None,
):
iterator = iter(generator)
page = list(islice(iterator, page_size))
first = 1
last = first + len(page) - 1
while True:
click.echo("-" * 80)
click.echo()
for item in page:
print_fn(item)
last = first + len(page) - 1
click.echo("-" * 80)
click.echo(f"{label} {first}-{last} of {total_count or '???'}")
first = first + len(page)
last = first + 1
page = list(islice(iterator, page_size))
if not page or not prompt_continue():
break
def print_video(video: Data):
published_at = video["publishedAt"].replace("T", " @ ").replace("Z", "")
length = utils.format_duration(video["lengthSeconds"])
@ -63,6 +99,7 @@ def print_video(video: Data):
click.echo(f"Published {blue(published_at)} Length: {blue(length)} ")
click.secho(url, italic=True)
click.echo()
def print_video_compact(video: Data):
@ -73,34 +110,6 @@ def print_video_compact(video: Data):
click.echo(f"{bold(id)} {date} {green(title)} {blue(game)}")
def print_paged_videos(generator: Generator[Data, None, None], page_size: int, total_count: int):
iterator = iter(generator)
page = list(islice(iterator, page_size))
first = 1
last = first + len(page) - 1
while True:
click.echo("-" * 80)
click.echo()
for video in page:
print_video(video)
click.echo()
last = first + len(page) - 1
click.echo("-" * 80)
click.echo(f"Videos {first}-{last} of {total_count}")
first = first + len(page)
last = first + 1
page = list(islice(iterator, page_size))
if not page or not _continue():
break
def print_clip(clip: Data):
published_at = clip["createdAt"].replace("T", " @ ").replace("Z", "")
length = utils.format_duration(clip["durationSeconds"])
@ -118,7 +127,7 @@ def print_clip(clip: Data):
click.secho(clip["url"], italic=True)
def _continue():
def prompt_continue():
enter = click.style("Enter", bold=True, fg="green")
ctrl_c = click.style("Ctrl+C", bold=True, fg="yellow")
click.echo(f"Press {enter} to continue, {ctrl_c} to break.")

View File

@ -12,6 +12,11 @@ from twitchdl.entities import Data
from twitchdl.exceptions import ConsoleError
ClipsPeriod = Literal["last_day", "last_week", "last_month", "all_time"]
VideosSort = Literal["views", "time"]
VideosType = Literal["archive", "highlight", "upload"]
class GQLError(click.ClickException):
def __init__(self, errors: list[str]):
message = "GraphQL query failed."
@ -141,7 +146,7 @@ def get_clip_access_token(slug: str):
return response["data"]["clip"]
def get_channel_clips(channel_id: str, period: str, limit: int, after: str | None= None):
def get_channel_clips(channel_id: str, period: ClipsPeriod, limit: int, after: str | None= None):
"""
List channel clips.
@ -178,8 +183,8 @@ def get_channel_clips(channel_id: str, period: str, limit: int, after: str | Non
return response["data"]["user"]["clips"]
def channel_clips_generator(channel_id, period, limit):
def _generator(clips, limit):
def channel_clips_generator(channel_id: str, period: str, limit: int) -> Generator[Data, None, None]:
def _generator(clips: Data, limit: int) -> Generator[Data, None, None]:
for clip in clips["edges"]:
if limit < 1:
return
@ -263,10 +268,6 @@ def get_channel_videos(
return response["data"]["user"]["videos"]
VideosSort = Literal["views", "time"]
VideosType = Literal["archive", "highlight", "upload"]
def channel_videos_generator(
channel_id: str,
max_videos: int,