Compare commits

..

25 Commits
2.2.1 ... 2.2.4

Author SHA1 Message Date
125bc693f8 Update changelog 2024-04-25 07:33:05 +02:00
8a7fdad22f Test m patterns 2024-04-25 07:31:52 +02:00
c00a9c3597 Add support for m dot urls 2024-04-25 07:29:59 +02:00
0f17d92a8c Update changelog 2024-04-24 14:54:31 +02:00
ee1e0ce853 Upgrade actions/setup-python to v5 2024-04-24 14:52:42 +02:00
1c878bbca8 Update changelog 2024-04-24 14:51:08 +02:00
941440de41 Upgrade actions/checkout to v4 2024-04-24 14:50:50 +02:00
c0eae623a4 Test clips 2024-04-24 14:45:49 +02:00
f4d0643b07 Fix flaky test 2024-04-24 14:41:48 +02:00
ea4b714343 Test videos command 2024-04-24 14:39:17 +02:00
f815934e15 Simplify and test get_game_ids 2024-04-24 14:39:12 +02:00
5aa323e3e5 Log response content 2024-04-24 14:11:22 +02:00
3d03658850 Add tests 2024-04-24 08:59:01 +02:00
69b848d341 Respect --dry-run when downloading videos 2024-04-24 08:58:58 +02:00
2422871d70 Add request logging 2024-04-24 08:43:45 +02:00
44890b4101 Pass auth_token instead of headers 2024-04-24 08:43:00 +02:00
9aa108acbf Add type hints for post 2024-04-24 08:42:01 +02:00
3fa8bcef73 Pass request as content instead of data
data works, but should be used for form data passed as a dict, while
content takes bytes or str
2024-04-24 08:41:19 +02:00
3c0f8a8ece Add some cli tests 2024-04-24 08:38:55 +02:00
7845cf6f72 Improve typing 2024-04-24 08:11:33 +02:00
6ed98fa4ef Fix tuple typing to work on py 3.8 2024-04-24 08:10:56 +02:00
d777f0e98a Set up testing on github 2024-04-24 08:10:20 +02:00
ad0f0d2a41 Update changelog 2024-04-23 18:14:22 +02:00
1057fff61a Hopefully fix python 3.8 compat 2024-04-23 18:13:40 +02:00
f1924715ed Apply formatting 2024-04-23 17:35:46 +02:00
19 changed files with 379 additions and 73 deletions

27
.github/workflows/test.yml vendored Normal file
View File

@ -0,0 +1,27 @@
name: Run tests
on: [push, pull_request]
jobs:
test:
runs-on: ubuntu-22.04
strategy:
matrix:
python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -e ".[test]"
- name: Run tests
run: |
pytest
- name: Validate minimum required version
run: |
vermin --no-tips twitchdl

View File

@ -3,9 +3,22 @@ twitch-dl changelog
<!-- Do not edit. This file is automatically generated from changelog.yaml.-->
### [2.2.4 (2024-04-25)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.4)
* Add m dot url support to video and clip regexes (thanks @localnerve)
### [2.2.3 (2024-04-24)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.3)
* Respect --dry-run option when downloading videos
* Add automated tests on github actions
### [2.2.2 (2024-04-23)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.2)
* Fix more compat issues Python < 3.10 (#152)
### [2.2.1 (2024-04-23)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.1)
* Fix compat with < Python 3.10 (#152)
* Fix compat with Python < 3.10 (#152)
* Fix division by zero in progress calculation when video duration is reported
as 0

View File

@ -24,7 +24,7 @@ publish :
twine upload dist/*.tar.gz dist/*.whl
coverage:
py.test --cov=toot --cov-report html tests/
pytest --cov=twitchdl --cov-report html tests/
man:
scdoc < twitch-dl.1.scd > twitch-dl.1.man

View File

@ -1,7 +1,23 @@
2.2.4:
date: 2024-04-25
changes:
- "Add m dot url support to video and clip regexes (thanks @localnerve)"
2.2.3:
date: 2024-04-24
changes:
- "Respect --dry-run option when downloading videos"
- "Add automated tests on github actions"
2.2.2:
date: 2024-04-23
changes:
- "Fix more compat issues Python < 3.10 (#152)"
2.2.1:
date: 2024-04-23
changes:
- "Fix compat with < Python 3.10 (#152)"
- "Fix compat with Python < 3.10 (#152)"
- "Fix division by zero in progress calculation when video duration is reported as 0"
2.2.0:

View File

@ -3,9 +3,22 @@ twitch-dl changelog
<!-- Do not edit. This file is automatically generated from changelog.yaml.-->
### [2.2.4 (2024-04-25)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.4)
* Add m dot url support to video and clip regexes (thanks @localnerve)
### [2.2.3 (2024-04-24)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.3)
* Respect --dry-run option when downloading videos
* Add automated tests on github actions
### [2.2.2 (2024-04-23)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.2)
* Fix more compat issues Python < 3.10 (#152)
### [2.2.1 (2024-04-23)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.1)
* Fix compat with < Python 3.10 (#152)
* Fix compat with Python < 3.10 (#152)
* Fix division by zero in progress calculation when video duration is reported
as 0

View File

@ -43,6 +43,11 @@ dev = [
"vermin",
]
test = [
"pytest",
"vermin",
]
[project.urls]
"Homepage" = "https://twitch-dl.bezdomni.net/"
"Source" = "https://github.com/ihabunek/twitch-dl"
@ -51,8 +56,8 @@ dev = [
twitch-dl = "twitchdl.cli:cli"
[tool.pyright]
include = ["twitchdl"]
typeCheckingMode = "strict"
pythonVersion = "3.8"
[tool.ruff]
line-length = 100

View File

@ -3,9 +3,12 @@ These tests depend on the channel having some videos and clips published.
"""
import httpx
import pytest
from twitchdl import twitch
from twitchdl.commands.download import get_clip_authenticated_url
from twitchdl.commands.videos import get_game_ids
from twitchdl.exceptions import ConsoleError
from twitchdl.playlists import enumerate_vods, load_m3u8, parse_playlists
TEST_CHANNEL = "bananasaurus_rex"
@ -53,3 +56,15 @@ def test_get_clips():
assert clip["slug"] == slug
assert get_clip_authenticated_url(slug, "source")
def test_get_games():
assert get_game_ids([]) == []
assert get_game_ids(["Bioshock"]) == ["15866"]
assert get_game_ids(["Bioshock", "Portal"]) == ["15866", "6187"]
def test_get_games_not_found():
with pytest.raises(ConsoleError) as ex:
get_game_ids(["the game which does not exist"])
assert str(ex.value) == "Game 'the game which does not exist' not found"

156
tests/test_cli.py Normal file
View File

@ -0,0 +1,156 @@
import json
import pytest
from click.testing import CliRunner, Result
from twitchdl import cli
@pytest.fixture(scope="session")
def runner():
return CliRunner(mix_stderr=False)
def assert_ok(result: Result):
if result.exit_code != 0:
raise AssertionError(
f"Command failed with exit code {result.exit_code}\nStderr: {result.stderr}"
)
def test_info_video(runner: CliRunner):
result = runner.invoke(cli.info, ["2090131595"])
assert_ok(result)
assert "Frost Fatales 2024 Day 1" in result.stdout
assert "frozenflygone playing Tomb Raider" in result.stdout
def test_info_video_json(runner: CliRunner):
result = runner.invoke(cli.info, ["2090131595", "--json"])
assert_ok(result)
video = json.loads(result.stdout)
assert video["title"] == "Frost Fatales 2024 Day 1"
assert video["game"] == {"id": "2770", "name": "Tomb Raider"}
assert video["creator"] == {"login": "frozenflygone", "displayName": "frozenflygone"}
def test_info_clip(runner: CliRunner):
result = runner.invoke(cli.info, ["PoisedTalentedPuddingChefFrank"])
assert_ok(result)
assert "AGDQ Crashes during Bioshock run" in result.stdout
assert "GamesDoneQuick playing BioShock" in result.stdout
def test_info_clip_json(runner: CliRunner):
result = runner.invoke(cli.info, ["PoisedTalentedPuddingChefFrank", "--json"])
assert_ok(result)
clip = json.loads(result.stdout)
assert clip["slug"] == "PoisedTalentedPuddingChefFrank"
assert clip["title"] == "AGDQ Crashes during Bioshock run"
assert clip["game"] == {"id": "15866", "name": "BioShock"}
assert clip["broadcaster"] == {"displayName": "GamesDoneQuick", "login": "gamesdonequick"}
def test_info_not_found(runner: CliRunner):
result = runner.invoke(cli.info, ["banana"])
assert result.exit_code == 1
assert "Clip banana not found" in result.stderr
result = runner.invoke(cli.info, ["12345"])
assert result.exit_code == 1
assert "Video 12345 not found" in result.stderr
result = runner.invoke(cli.info, [""])
assert result.exit_code == 1
assert "Invalid input" in result.stderr
def test_download_clip(runner: CliRunner):
result = runner.invoke(
cli.download,
[
"PoisedTalentedPuddingChefFrank",
"-q",
"source",
"--dry-run",
],
)
assert_ok(result)
assert (
"Found: AGDQ Crashes during Bioshock run by GamesDoneQuick, playing BioShock (30 sec)"
in result.stdout
)
assert (
"Target: 2020-01-10_3099545841_gamesdonequick_agdq_crashes_during_bioshock_run.mp4"
in result.stdout
)
assert "Dry run, clip not downloaded." in result.stdout
def test_download_video(runner: CliRunner):
result = runner.invoke(
cli.download,
[
"2090131595",
"-q",
"source",
"--dry-run",
],
)
assert_ok(result)
assert "Found: Frost Fatales 2024 Day 1 by frozenflygone" in result.stdout
assert (
"Output: 2024-03-14_2090131595_frozenflygone_frost_fatales_2024_day_1.mkv" in result.stdout
)
assert "Dry run, video not downloaded." in result.stdout
def test_videos(runner: CliRunner):
result = runner.invoke(cli.videos, ["gamesdonequick", "--json"])
assert_ok(result)
videos = json.loads(result.stdout)
assert videos["count"] == 10
assert videos["totalCount"] > 0
video = videos["videos"][0]
result = runner.invoke(cli.videos, "gamesdonequick")
assert_ok(result)
assert f"Video {video['id']}" in result.stdout
assert video["title"] in result.stdout
result = runner.invoke(cli.videos, ["gamesdonequick", "--compact"])
assert_ok(result)
assert video["id"] in result.stdout
assert video["title"][:60] in result.stdout
def test_videos_channel_not_found(runner: CliRunner):
result = runner.invoke(cli.videos, ["doesnotexisthopefully"])
assert result.exit_code == 1
assert result.stderr.strip() == "Error: Channel doesnotexisthopefully not found"
def test_clips(runner: CliRunner):
result = runner.invoke(cli.clips, ["gamesdonequick", "--json"])
assert_ok(result)
clips = json.loads(result.stdout)
clip = clips[0]
result = runner.invoke(cli.clips, "gamesdonequick")
assert_ok(result)
assert f"Clip {clip['slug']}" in result.stdout
assert clip["title"] in result.stdout
result = runner.invoke(cli.clips, ["gamesdonequick", "--compact"])
assert_ok(result)
assert clip["slug"] in result.stdout
assert clip["title"][:60] in result.stdout

View File

@ -1,35 +1,38 @@
import pytest
from twitchdl.utils import parse_video_identifier, parse_clip_identifier
from twitchdl.utils import parse_clip_identifier, parse_video_identifier
TEST_VIDEO_PATTERNS = [
("702689313", "702689313"),
("702689313", "https://twitch.tv/videos/702689313"),
("702689313", "https://www.twitch.tv/videos/702689313"),
("702689313", "https://m.twitch.tv/videos/702689313"),
]
TEST_CLIP_PATTERNS = {
("AbrasivePlayfulMangoMau5", "AbrasivePlayfulMangoMau5"),
("AbrasivePlayfulMangoMau5", "https://clips.twitch.tv/AbrasivePlayfulMangoMau5"),
("AbrasivePlayfulMangoMau5", "https://www.twitch.tv/dracul1nx/clip/AbrasivePlayfulMangoMau5"),
("AbrasivePlayfulMangoMau5", "https://m.twitch.tv/dracul1nx/clip/AbrasivePlayfulMangoMau5"),
("AbrasivePlayfulMangoMau5", "https://twitch.tv/dracul1nx/clip/AbrasivePlayfulMangoMau5"),
("HungryProudRadicchioDoggo", "HungryProudRadicchioDoggo"),
("HungryProudRadicchioDoggo", "https://clips.twitch.tv/HungryProudRadicchioDoggo"),
("HungryProudRadicchioDoggo", "https://www.twitch.tv/bananasaurus_rex/clip/HungryProudRadicchioDoggo?filter=clips&range=7d&sort=time"),
("HungryProudRadicchioDoggo", "https://m.twitch.tv/bananasaurus_rex/clip/HungryProudRadicchioDoggo?filter=clips&range=7d&sort=time"),
("HungryProudRadicchioDoggo", "https://twitch.tv/bananasaurus_rex/clip/HungryProudRadicchioDoggo?filter=clips&range=7d&sort=time"),
("GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ", "GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ"),
("GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ", "https://twitch.tv/dracul1nx/clip/GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ"),
("GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ", "https://twitch.tv/dracul1nx/clip/GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ?filter=clips&range=7d&sort=time"),
("GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ", "https://www.twitch.tv/dracul1nx/clip/GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ?filter=clips&range=7d&sort=time"),
("GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ", "https://m.twitch.tv/dracul1nx/clip/GloriousColdbloodedTortoiseRuleFive-E017utJ4DZmHVpfQ?filter=clips&range=7d&sort=time"),
}
@pytest.mark.parametrize("expected,input", TEST_VIDEO_PATTERNS)
def test_video_patterns(expected, input):
def test_video_patterns(expected: str, input: str):
assert parse_video_identifier(input) == expected
@pytest.mark.parametrize("expected,input", TEST_CLIP_PATTERNS)
def test_clip_patterns(expected, input):
def test_clip_patterns(expected: str, input: str):
assert parse_clip_identifier(input) == expected

View File

@ -2,7 +2,7 @@ import logging
import platform
import re
import sys
from typing import Optional
from typing import Optional, Tuple
import click
@ -91,7 +91,9 @@ def cli(ctx: click.Context, color: bool, debug: bool):
ctx.color = color
if debug:
logging.basicConfig(level=logging.INFO)
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("httpx").setLevel(logging.WARN)
logging.getLogger("httpcore").setLevel(logging.WARN)
@cli.command()
@ -255,7 +257,7 @@ def clips(
default=5,
)
def download(
ids: tuple[str, ...],
ids: Tuple[str, ...],
auth_token: Optional[str],
chapter: Optional[int],
concat: bool,
@ -374,7 +376,7 @@ def videos(
channel_name: str,
all: bool,
compact: bool,
games_tuple: tuple[str, ...],
games_tuple: Tuple[str, ...],
json: bool,
limit: Optional[int],
pager: Optional[int],

View File

@ -7,6 +7,7 @@ import subprocess
import tempfile
from os import path
from pathlib import Path
from typing import Dict, List
from urllib.parse import urlencode, urlparse
import click
@ -28,7 +29,7 @@ from twitchdl.playlists import (
from twitchdl.twitch import Chapter, Clip, ClipAccessToken, Video
def download(ids: list[str], args: DownloadOptions):
def download(ids: List[str], args: DownloadOptions):
for video_id in ids:
download_one(video_id, args)
@ -78,7 +79,7 @@ def _join_vods(playlist_path: str, target: str, overwrite: bool, video: Video):
raise ConsoleError("Joining files failed")
def _concat_vods(vod_paths: list[str], target: str):
def _concat_vods(vod_paths: List[str], target: str):
tool = "type" if platform.system() == "Windows" else "cat"
command = [tool] + vod_paths
@ -88,7 +89,7 @@ def _concat_vods(vod_paths: list[str], target: str):
raise ConsoleError(f"Joining files failed: {result.stderr}")
def get_video_placeholders(video: Video, format: str) -> dict[str, str]:
def get_video_placeholders(video: Video, format: str) -> Dict[str, str]:
date, time = video["publishedAt"].split("T")
game = video["game"]["name"] if video["game"] else "Unknown"
@ -272,6 +273,10 @@ def _download_video(video_id: str, args: DownloadOptions) -> None:
vods_m3u8 = load_m3u8(vods_text)
vods = enumerate_vods(vods_m3u8, start, end)
if args.dry_run:
click.echo("Dry run, video not downloaded.")
return
base_uri = re.sub("/[^/]+$", "/", playlist.url)
target_dir = _crete_temp_dir(base_uri)
@ -350,7 +355,7 @@ def _determine_time_range(video_id: str, args: DownloadOptions):
return None, None
def _choose_chapter_interactive(chapters: list[Chapter]):
def _choose_chapter_interactive(chapters: List[Chapter]):
click.echo("\nChapters:")
for index, chapter in enumerate(chapters):
duration = utils.format_time(chapter["durationMilliseconds"] // 1000)

View File

@ -1,3 +1,5 @@
from typing import List
import click
import m3u8
@ -5,6 +7,7 @@ from twitchdl import twitch, utils
from twitchdl.commands.download import get_video_placeholders
from twitchdl.exceptions import ConsoleError
from twitchdl.output import bold, print_clip, print_json, print_log, print_table, print_video
from twitchdl.playlists import parse_playlists
from twitchdl.twitch import Chapter, Clip, Video
@ -48,13 +51,13 @@ def info(id: str, *, json: bool = False):
raise ConsoleError(f"Invalid input: {id}")
def video_info(video: Video, playlists, chapters: list[Chapter]):
def video_info(video: Video, playlists: str, chapters: List[Chapter]):
click.echo()
print_video(video)
click.echo("Playlists:")
for p in m3u8.loads(playlists).playlists:
click.echo(f"{bold(p.stream_info.video)} {p.uri}")
for p in parse_playlists(playlists):
click.echo(f"{bold(p.name)} {p.url}")
if chapters:
click.echo()
@ -70,7 +73,7 @@ def video_info(video: Video, playlists, chapters: list[Chapter]):
print_table(["Placeholder", "Value"], placeholders)
def video_json(video, playlists, chapters):
def video_json(video: Video, playlists: str, chapters: List[Chapter]):
playlists = m3u8.loads(playlists).playlists
video["playlists"] = [

View File

@ -1,5 +1,5 @@
import sys
from typing import Optional
from typing import List, Optional
import click
@ -13,14 +13,14 @@ def videos(
*,
all: bool,
compact: bool,
games: list[str],
games: List[str],
json: bool,
limit: Optional[int],
pager: Optional[int],
sort: twitch.VideosSort,
type: twitch.VideosType,
):
game_ids = _get_game_ids(games)
game_ids = get_game_ids(games)
# Set different defaults for limit for compact display
limit = limit or (40 if compact else 10)
@ -67,16 +67,13 @@ def videos(
)
def _get_game_ids(names: list[str]) -> list[str]:
if not names:
return []
def get_game_ids(names: List[str]) -> List[str]:
return [get_game_id(name) for name in names]
game_ids = []
for name in names:
print_log(f"Looking up game '{name}'...")
game_id = twitch.get_game_id(name)
if not game_id:
raise ConsoleError(f"Game '{name}' not found")
game_ids.append(int(game_id))
return game_ids
def get_game_id(name: str) -> str:
print_log(f"Looking up game '{name}'...")
game_id = twitch.get_game_id(name)
if not game_id:
raise ConsoleError(f"Game '{name}' not found")
return game_id

View File

@ -1,4 +1,5 @@
import os
import httpx
from twitchdl.exceptions import ConsoleError

View File

@ -1,5 +1,5 @@
from dataclasses import dataclass
from typing import Any, Optional
from typing import Any, Mapping, Optional
@dataclass
@ -22,4 +22,4 @@ class DownloadOptions:
# Type for annotating decoded JSON
# TODO: make data classes for common structs
Data = dict[str, Any]
Data = Mapping[str, Any]

View File

@ -1,6 +1,6 @@
import json
from itertools import islice
from typing import Any, Callable, Generator, Optional, TypeVar
from typing import Any, Callable, Generator, List, Optional, TypeVar
import click
@ -25,12 +25,12 @@ def print_log(message: Any):
click.secho(message, err=True, dim=True)
def print_table(headers: list[str], data: list[list[str]]):
def print_table(headers: List[str], data: List[List[str]]):
widths = [[len(cell) for cell in row] for row in data + [headers]]
widths = [max(width) for width in zip(*widths)]
underlines = ["-" * width for width in widths]
def print_row(row: list[str]):
def print_row(row: List[str]):
for idx, cell in enumerate(row):
width = widths[idx]
click.echo(cell.ljust(width), nl=False)

View File

@ -3,7 +3,7 @@ Parse and manipulate m3u8 playlists.
"""
from dataclasses import dataclass
from typing import Generator, Optional, OrderedDict
from typing import Generator, List, Optional, OrderedDict
import click
import m3u8
@ -29,7 +29,7 @@ class Vod:
"""Segment duration in seconds"""
def parse_playlists(playlists_m3u8: str):
def parse_playlists(playlists_m3u8: str) -> List[Playlist]:
def _parse(source: str) -> Generator[Playlist, None, None]:
document = load_m3u8(source)
@ -55,7 +55,7 @@ def enumerate_vods(
document: m3u8.M3U8,
start: Optional[int] = None,
end: Optional[int] = None,
) -> list[Vod]:
) -> List[Vod]:
"""Extract VODs for download from document."""
vods = []
vod_start = 0
@ -78,8 +78,8 @@ def enumerate_vods(
def make_join_playlist(
playlist: m3u8.M3U8,
vods: list[Vod],
targets: list[str],
vods: List[Vod],
targets: List[str],
) -> m3u8.Playlist:
"""
Make a modified playlist which references downloaded VODs
@ -97,7 +97,7 @@ def make_join_playlist(
return playlist
def select_playlist(playlists: list[Playlist], quality: Optional[str]) -> Playlist:
def select_playlist(playlists: List[Playlist], quality: Optional[str]) -> Playlist:
return (
select_playlist_by_name(playlists, quality)
if quality is not None
@ -105,7 +105,7 @@ def select_playlist(playlists: list[Playlist], quality: Optional[str]) -> Playli
)
def select_playlist_by_name(playlists: list[Playlist], quality: str) -> Playlist:
def select_playlist_by_name(playlists: List[Playlist], quality: str) -> Playlist:
if quality == "source":
return playlists[0]
@ -118,7 +118,7 @@ def select_playlist_by_name(playlists: list[Playlist], quality: str) -> Playlist
raise click.ClickException(msg)
def select_playlist_interactive(playlists: list[Playlist]) -> Playlist:
def select_playlist_interactive(playlists: List[Playlist]) -> Playlist:
click.echo("\nAvailable qualities:")
for n, playlist in enumerate(playlists):
if playlist.resolution:

View File

@ -3,7 +3,9 @@ Twitch API access.
"""
import json
from typing import Dict, Generator, Literal, TypedDict, Optional
import logging
import time
from typing import Any, Dict, Generator, List, Literal, Mapping, Optional, Tuple, TypedDict, Union
import click
import httpx
@ -41,7 +43,7 @@ class VideoQuality(TypedDict):
class ClipAccessToken(TypedDict):
id: str
playbackAccessToken: AccessToken
videoQualities: list[VideoQuality]
videoQualities: List[VideoQuality]
class Clip(TypedDict):
@ -52,7 +54,7 @@ class Clip(TypedDict):
viewCount: int
durationSeconds: int
url: str
videoQualities: list[VideoQuality]
videoQualities: List[VideoQuality]
game: Game
broadcaster: User
@ -80,17 +82,29 @@ class Chapter(TypedDict):
class GQLError(click.ClickException):
def __init__(self, errors: list[str]):
def __init__(self, errors: List[str]):
message = "GraphQL query failed."
for error in errors:
message += f"\n* {error}"
super().__init__(message)
def authenticated_post(url, data=None, json=None, headers={}):
headers["Client-ID"] = CLIENT_ID
Content = Union[str, bytes]
Headers = Dict[str, str]
response = httpx.post(url, data=data, json=json, headers=headers)
def authenticated_post(
url: str,
*,
json: Any = None,
content: Optional[Content] = None,
auth_token: Optional[str] = None,
):
headers = {"Client-ID": CLIENT_ID}
if auth_token is not None:
headers["authorization"] = f"OAuth {auth_token}"
response = request("POST", url, content=content, json=json, headers=headers)
if response.status_code == 400:
data = response.json()
raise ConsoleError(data["message"])
@ -100,16 +114,50 @@ def authenticated_post(url, data=None, json=None, headers={}):
return response
def request(
method: str,
url: str,
json: Any = None,
content: Optional[Content] = None,
headers: Optional[Mapping[str, str]] = None,
):
with httpx.Client() as client:
request = client.build_request(method, url, json=json, content=content, headers=headers)
log_request(request)
start = time.time()
response = client.send(request)
duration = time.time() - start
log_response(response, duration)
return response
logger = logging.getLogger(__name__)
def log_request(request: httpx.Request):
logger.debug(f"--> {request.method} {request.url}")
if request.content:
logger.debug(f"--> {request.content}")
def log_response(response: httpx.Response, duration: float):
request = response.request
duration_ms = int(1000 * duration)
logger.debug(f"<-- {request.method} {request.url} HTTP {response.status_code} {duration_ms}ms")
if response.content:
logger.debug(f"<-- {response.content}")
def gql_post(query: str):
url = "https://gql.twitch.tv/gql"
response = authenticated_post(url, data=query)
response = authenticated_post(url, content=query)
gql_raise_on_error(response)
return response.json()
def gql_query(query: str, headers: Dict[str, str] = {}):
def gql_query(query: str, auth_token: Optional[str] = None):
url = "https://gql.twitch.tv/gql"
response = authenticated_post(url, json={"query": query}, headers=headers)
response = authenticated_post(url, json={"query": query}, auth_token=auth_token)
gql_raise_on_error(response)
return response.json()
@ -209,7 +257,12 @@ def get_clip_access_token(slug: str) -> ClipAccessToken:
return response["data"]["clip"]
def get_channel_clips(channel_id: str, period: ClipsPeriod, limit: int, after: Optional[str] = None):
def get_channel_clips(
channel_id: str,
period: ClipsPeriod,
limit: int,
after: Optional[str] = None,
):
"""
List channel clips.
@ -294,10 +347,11 @@ def get_channel_videos(
limit: int,
sort: str,
type: str = "archive",
game_ids: Optional[list[str]] = None,
game_ids: Optional[List[str]] = None,
after: Optional[str] = None,
):
game_ids = game_ids or []
game_ids_str = f"[{','.join(game_ids)}]"
query = f"""
{{
@ -308,7 +362,7 @@ def get_channel_videos(
sort: {sort.upper()},
after: "{after or ''}",
options: {{
gameIDs: {game_ids}
gameIDs: {game_ids_str}
}}
) {{
totalCount
@ -339,8 +393,8 @@ def channel_videos_generator(
max_videos: int,
sort: VideosSort,
type: VideosType,
game_ids: Optional[list[str]] = None,
) -> tuple[int, Generator[Video, None, None]]:
game_ids: Optional[List[str]] = None,
) -> Tuple[int, Generator[Video, None, None]]:
game_ids = game_ids or []
def _generator(videos: Data, max_videos: int) -> Generator[Video, None, None]:
@ -381,12 +435,8 @@ def get_access_token(video_id: str, auth_token: Optional[str] = None) -> AccessT
}}
"""
headers: dict[str, str] = {}
if auth_token is not None:
headers["authorization"] = f"OAuth {auth_token}"
try:
response = gql_query(query, headers=headers)
response = gql_query(query, auth_token=auth_token)
return response["data"]["videoPlaybackAccessToken"]
except httpx.HTTPStatusError as error:
# Provide a more useful error message when server returns HTTP 401
@ -438,7 +488,7 @@ def get_game_id(name: str):
return game["id"]
def get_video_chapters(video_id: str) -> list[Chapter]:
def get_video_chapters(video_id: str) -> List[Chapter]:
query = {
"operationName": "VideoPlayer_ChapterSelectButtonVideo",
"variables": {

View File

@ -1,6 +1,6 @@
import re
from typing import Optional, Union
import unicodedata
from typing import Optional, Union
import click
@ -84,13 +84,13 @@ def titlify(value: str) -> str:
VIDEO_PATTERNS = [
r"^(?P<id>\d+)?$",
r"^https://(www.)?twitch.tv/videos/(?P<id>\d+)(\?.+)?$",
r"^https://(www\.|m\.)?twitch\.tv/videos/(?P<id>\d+)(\?.+)?$",
]
CLIP_PATTERNS = [
r"^(?P<slug>[A-Za-z0-9]+(?:-[A-Za-z0-9_-]{16})?)$",
r"^https://(www.)?twitch.tv/\w+/clip/(?P<slug>[A-Za-z0-9]+(?:-[A-Za-z0-9_-]{16})?)(\?.+)?$",
r"^https://clips.twitch.tv/(?P<slug>[A-Za-z0-9]+(?:-[A-Za-z0-9_-]{16})?)(\?.+)?$",
r"^https://(www\.|m\.)?twitch\.tv/\w+/clip/(?P<slug>[A-Za-z0-9]+(?:-[A-Za-z0-9_-]{16})?)(\?.+)?$",
r"^https://clips\.twitch\.tv/(?P<slug>[A-Za-z0-9]+(?:-[A-Za-z0-9_-]{16})?)(\?.+)?$",
]