Update changelog

data works, but should be used for form data passed as a dict, while
content takes bytes or str

.github/workflows/test.yml

@@ -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

CHANGELOG.md

@@ -3,6 +3,25 @@ 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 division by zero in progress calculation when video duration is reported
+  as 0
+
 ### [2.2.0 (2024-04-10)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.0)
 
 * **Requires Python 3.8+**

Makefile

@@ -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

changelog.yaml

@@ -1,3 +1,25 @@
+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 division by zero in progress calculation when video duration is reported as 0"
+
 2.2.0:
   date: 2024-04-10
   changes:

docs/changelog.md

@@ -3,6 +3,25 @@ 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 division by zero in progress calculation when video duration is reported
+  as 0
+
 ### [2.2.0 (2024-04-10)](https://github.com/ihabunek/twitch-dl/releases/tag/2.2.0)
 
 * **Requires Python 3.8+**

pyproject.toml

@@ -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

tests/test_api.py

@@ -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"

tests/test_cli.py

@@ -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

tests/test_patterns.py

@@ -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

twitchdl/cli.py

@@ -2,6 +2,7 @@ import logging
 import platform
 import re
 import sys
+from typing import Optional, Tuple
 
 import click
 
@@ -30,13 +31,13 @@ json_option = click.option(
 )
 
 
-def validate_positive(_ctx: click.Context, _param: click.Parameter, value: int | None):
+def validate_positive(_ctx: click.Context, _param: click.Parameter, value: Optional[int]):
     if value is not None and value <= 0:
         raise click.BadParameter("must be greater than 0")
     return value
 
 
-def validate_time(_ctx: click.Context, _param: click.Parameter, value: str) -> int | None:
+def validate_time(_ctx: click.Context, _param: click.Parameter, value: str) -> Optional[int]:
     """Parse a time string (hh:mm or hh:mm:ss) to number of seconds."""
     if not value:
         return None
@@ -56,7 +57,7 @@ def validate_time(_ctx: click.Context, _param: click.Parameter, value: str) -> i
     return hours * 3600 + minutes * 60 + seconds
 
 
-def validate_rate(_ctx: click.Context, _param: click.Parameter, value: str) -> int | None:
+def validate_rate(_ctx: click.Context, _param: click.Parameter, value: str) -> Optional[int]:
     if not value:
         return None
 
@@ -90,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()
@@ -143,8 +146,8 @@ def clips(
     compact: bool,
     download: bool,
     json: bool,
-    limit: int | None,
-    pager: int | None,
+    limit: Optional[int],
+    pager: Optional[int],
     period: ClipsPeriod,
 ):
     """List or download clips for given CHANNEL_NAME."""
@@ -254,20 +257,20 @@ def clips(
     default=5,
 )
 def download(
-    ids: tuple[str, ...],
-    auth_token: str | None,
-    chapter: int | None,
+    ids: Tuple[str, ...],
+    auth_token: Optional[str],
+    chapter: Optional[int],
     concat: bool,
     dry_run: bool,
-    end: int | None,
+    end: Optional[int],
     format: str,
     keep: bool,
     no_join: bool,
     overwrite: bool,
     output: str,
-    quality: str | None,
-    rate_limit: int | None,
-    start: int | None,
+    quality: Optional[str],
+    rate_limit: Optional[int],
+    start: Optional[int],
     max_workers: int,
 ):
     """Download videos or clips.
@@ -373,10 +376,10 @@ def videos(
     channel_name: str,
     all: bool,
     compact: bool,
-    games_tuple: tuple[str, ...],
+    games_tuple: Tuple[str, ...],
     json: bool,
-    limit: int | None,
-    pager: int | None,
+    limit: Optional[int],
+    pager: Optional[int],
     sort: VideosSort,
     type: VideosType,
 ):

twitchdl/commands/clips.py

@@ -1,7 +1,7 @@
 import re
 import sys
 from os import path
-from typing import Callable, Generator
+from typing import Callable, Generator, Optional
 
 import click
 
@@ -9,7 +9,7 @@ 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_clip_compact, print_json, print_paged, yellow
-from twitchdl.twitch import Clip
+from twitchdl.twitch import Clip, ClipsPeriod
 
 
 def clips(
@@ -19,9 +19,9 @@ def clips(
     compact: bool = False,
     download: bool = False,
     json: bool = False,
-    limit: int | None = None,
-    pager: int | None = None,
-    period: twitch.ClipsPeriod = "all_time",
+    limit: Optional[int] = None,
+    pager: Optional[int] = None,
+    period: ClipsPeriod = "all_time",
 ):
     # Set different defaults for limit for compact display
     default_limit = 40 if compact else 10

twitchdl/commands/download.py

@@ -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"
 
@@ -251,7 +252,7 @@ def _download_video(video_id: str, args: DownloadOptions) -> None:
     click.echo(f"Output: {blue(target)}")
 
     if not args.overwrite and path.exists(target):
-        response = click.prompt("File exists. Overwrite? [Y/n]: ", default="Y", show_default=False)
+        response = click.prompt("File exists. Overwrite? [Y/n]", default="Y", show_default=False)
         if response.lower().strip() != "y":
             raise click.Abort()
         args.overwrite = True
@@ -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)

twitchdl/commands/info.py

@@ -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"] = [

twitchdl/commands/videos.py

@@ -1,4 +1,5 @@
 import sys
+from typing import List, Optional
 
 import click
 
@@ -12,14 +13,14 @@ def videos(
     *,
     all: bool,
     compact: bool,
-    games: list[str],
+    games: List[str],
     json: bool,
-    limit: int | None,
-    pager: int | None,
+    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)
@@ -66,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

twitchdl/download.py

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

twitchdl/entities.py

@@ -1,25 +1,25 @@
 from dataclasses import dataclass
-from typing import Any
+from typing import Any, Mapping, Optional
 
 
 @dataclass
 class DownloadOptions:
-    auth_token: str | None
-    chapter: int | None
+    auth_token: Optional[str]
+    chapter: Optional[int]
     concat: bool
     dry_run: bool
-    end: int | None
+    end: Optional[int]
     format: str
     keep: bool
     no_join: bool
     overwrite: bool
     output: str
-    quality: str | None
-    rate_limit: int | None
-    start: int | None
+    quality: Optional[str]
+    rate_limit: Optional[int]
+    start: Optional[int]
     max_workers: int
 
 
 # Type for annotating decoded JSON
 # TODO: make data classes for common structs
-Data = dict[str, Any]
+Data = Mapping[str, Any]

twitchdl/output.py

@@ -1,6 +1,6 @@
 import json
 from itertools import islice
-from typing import Any, Callable, Generator, 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)
@@ -49,7 +49,7 @@ def print_paged(
     generator: Generator[T, Any, Any],
     print_fn: Callable[[T], None],
     page_size: int,
-    total_count: int | None = None,
+    total_count: Optional[int] = None,
 ):
     iterator = iter(generator)
     page = list(islice(iterator, page_size))

twitchdl/playlists.py

@@ -3,7 +3,7 @@ Parse and manipulate m3u8 playlists.
 """
 
 from dataclasses import dataclass
-from typing import Generator, OrderedDict
+from typing import Generator, List, Optional, OrderedDict
 
 import click
 import m3u8
@@ -15,7 +15,7 @@ from twitchdl.output import bold, dim
 @dataclass
 class Playlist:
     name: str
-    resolution: str | None
+    resolution: Optional[str]
     url: str
 
 
@@ -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)
 
@@ -53,9 +53,9 @@ def load_m3u8(playlist_m3u8: str) -> m3u8.M3U8:
 
 def enumerate_vods(
     document: m3u8.M3U8,
-    start: int | None = None,
-    end: int | None = None,
-) -> list[Vod]:
+    start: Optional[int] = None,
+    end: Optional[int] = None,
+) -> 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: str | None) -> 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: str | None) -> Playlist:
     )
 
 
-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:

twitchdl/progress.py

@@ -127,7 +127,7 @@ class Progress:
         size = last_sample.downloaded - first_sample.downloaded
         duration = last_sample.timestamp - first_sample.timestamp
 
-        return size / duration
+        return size / duration if duration > 0 else None
 
     def print(self):
         now = time.time()
@@ -136,17 +136,20 @@ class Progress:
         if now - self.last_printed < 0.1:
             return
 
-        progress = " ".join(
-            [
-                f"Downloaded {self.vod_downloaded_count}/{self.vod_count} VODs",
-                f"{blue(self.progress_perc)}%",
-                f"of ~{blue(format_size(self.estimated_total))}" if self.estimated_total else "",
-                f"at {blue(format_size(self.speed))}/s" if self.speed else "",
-                f"ETA {blue(format_time(self.remaining_time))}"
-                if self.remaining_time is not None
-                else "",
-            ]
-        )
+        click.echo(f"\rDownloaded {self.vod_downloaded_count}/{self.vod_count} VODs", nl=False)
+        click.secho(f" {self.progress_perc}%", fg="blue", nl=False)
+
+        if self.estimated_total is not None:
+            total = f"~{format_size(self.estimated_total)}"
+            click.echo(f" of {blue(total)}", nl=False)
+
+        if self.speed is not None:
+            speed = f"{format_size(self.speed)}/s"
+            click.echo(f" at {blue(speed)}", nl=False)
+
+        if self.remaining_time is not None:
+            click.echo(f" ETA {blue(format_time(self.remaining_time))}", nl=False)
+
+        click.echo("    ", nl=False)
 
-        click.echo(f"\r{progress}     ", nl=False)
         self.last_printed = now

twitchdl/twitch.py

@@ -3,7 +3,9 @@ Twitch API access.
 """
 
 import json
-from typing import Dict, Generator, Literal, TypedDict
+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()
 
@@ -163,7 +211,7 @@ CLIP_FIELDS = """
 """
 
 
-def get_video(video_id: str) -> Video | None:
+def get_video(video_id: str) -> Optional[Video]:
     query = f"""
     {{
         video(id: "{video_id}") {{
@@ -176,7 +224,7 @@ def get_video(video_id: str) -> Video | None:
     return response["data"]["video"]
 
 
-def get_clip(slug: str) -> Clip | None:
+def get_clip(slug: str) -> Optional[Clip]:
     query = f"""
     {{
         clip(slug: "{slug}") {{
@@ -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: str | None = 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: list[str] | None = None,
-    after: str | None = 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: list[str] | None = 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]:
@@ -364,7 +418,7 @@ def channel_videos_generator(
     return videos["totalCount"], _generator(videos, max_videos)
 
 
-def get_access_token(video_id: str, auth_token: str | None = None) -> AccessToken:
+def get_access_token(video_id: str, auth_token: Optional[str] = None) -> AccessToken:
     query = f"""
     {{
         videoPlaybackAccessToken(
@@ -381,12 +435,8 @@ def get_access_token(video_id: str, auth_token: str | None = None) -> AccessToke
     }}
     """
 
-    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": {

twitchdl/utils.py

@@ -1,5 +1,6 @@
 import re
 import unicodedata
+from typing import Optional, Union
 
 import click
 
@@ -11,7 +12,7 @@ def _format_size(value: float, digits: int, unit: str):
         return f"{int(value)}{unit}"
 
 
-def format_size(bytes_: int | float, digits: int = 1):
+def format_size(bytes_: Union[int, float], digits: int = 1):
     if bytes_ < 1024:
         return _format_size(bytes_, digits, "B")
 
@@ -26,7 +27,7 @@ def format_size(bytes_: int | float, digits: int = 1):
     return _format_size(mega / 1024, digits, "GB")
 
 
-def format_duration(total_seconds: int | float) -> str:
+def format_duration(total_seconds: Union[int, float]) -> str:
     total_seconds = int(total_seconds)
     hours = total_seconds // 3600
     remainder = total_seconds % 3600
@@ -42,7 +43,7 @@ def format_duration(total_seconds: int | float) -> str:
     return f"{seconds} sec"
 
 
-def format_time(total_seconds: int | float, force_hours: bool = False) -> str:
+def format_time(total_seconds: Union[int, float], force_hours: bool = False) -> str:
     total_seconds = int(total_seconds)
     hours = total_seconds // 3600
     remainder = total_seconds % 3600
@@ -55,7 +56,7 @@ def format_time(total_seconds: int | float, force_hours: bool = False) -> str:
     return f"{minutes:02}:{seconds:02}"
 
 
-def read_int(msg: str, min: int, max: int, default: int | None = None) -> int:
+def read_int(msg: str, min: int, max: int, default: Optional[int] = None) -> int:
     while True:
         try:
             val = click.prompt(msg, default=default, type=int)
@@ -83,17 +84,17 @@ 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})?)(\?.+)?$",
 ]
 
 
-def parse_video_identifier(identifier: str) -> str | None:
+def parse_video_identifier(identifier: str) -> Optional[str]:
     """Given a video ID or URL returns the video ID, or null if not matched"""
     for pattern in VIDEO_PATTERNS:
         match = re.match(pattern, identifier)
@@ -101,7 +102,7 @@ def parse_video_identifier(identifier: str) -> str | None:
             return match.group("id")
 
 
-def parse_clip_identifier(identifier: str) -> str | None:
+def parse_clip_identifier(identifier: str) -> Optional[str]:
     """Given a clip slug or URL returns the clip slug, or null if not matched"""
     for pattern in CLIP_PATTERNS:
         match = re.match(pattern, identifier)