1
- #!/usr/bin/env python3
2
- # vim: ai ts=4 sts=4 et sw=4 nu
3
-
4
- from __future__ import annotations
5
-
6
1
import pathlib
7
2
import subprocess
8
3
from concurrent .futures import Future , ThreadPoolExecutor
9
- from typing import ClassVar
4
+ from typing import Any , ClassVar
10
5
11
6
import requests
12
7
import requests .adapters
13
8
import requests .structures
14
9
import urllib3 .util
15
- import yt_dlp as youtube_dl
10
+ import yt_dlp as youtube_dl # pyright: ignore[reportMissingTypeStubs]
16
11
17
12
from zimscraperlib import logger
18
13
from zimscraperlib .constants import DEFAULT_WEB_REQUESTS_TIMEOUT
@@ -31,24 +26,24 @@ def __init__(self, threads: int | None = 1) -> None:
31
26
def __enter__ (self ):
32
27
return self
33
28
34
- def __exit__ (self , * args ):
29
+ def __exit__ (self , * _ : Any ):
35
30
self .shutdown ()
36
31
37
32
def shutdown (self ) -> None :
38
33
"""shuts down the executor, awaiting completion"""
39
34
self .executor .shutdown (wait = True )
40
35
41
- def _run_youtube_dl (self , url : str , options : dict ) -> None :
36
+ def _run_youtube_dl (self , url : str , options : dict [ str , Any ] ) -> None :
42
37
with youtube_dl .YoutubeDL (options ) as ydl :
43
- ydl .download ([url ])
38
+ ydl .download ([url ]) # pyright: ignore[reportUnknownMemberType]
44
39
45
40
def download (
46
41
self ,
47
42
url : str ,
48
- options : dict | None ,
43
+ options : dict [ str , Any ] | None ,
49
44
* ,
50
45
wait : bool | None = True ,
51
- ) -> bool | Future :
46
+ ) -> bool | Future [ Any ] :
52
47
"""Downloads video using initialized executor.
53
48
54
49
url: URL or Video ID
@@ -66,7 +61,7 @@ def download(
66
61
return True
67
62
68
63
69
- class YoutubeConfig (dict ):
64
+ class YoutubeConfig (dict [ str , str | bool | int | None ] ):
70
65
options : ClassVar [dict [str , str | bool | int | None ]] = {}
71
66
defaults : ClassVar [dict [str , str | bool | int | None ]] = {
72
67
"writethumbnail" : True ,
@@ -82,7 +77,7 @@ class YoutubeConfig(dict):
82
77
"outtmpl" : "video.%(ext)s" ,
83
78
}
84
79
85
- def __init__ (self , ** kwargs ):
80
+ def __init__ (self , ** kwargs : str | bool | int | None ):
86
81
super ().__init__ (self , ** type (self ).defaults )
87
82
self .update (self .options )
88
83
self .update (kwargs )
@@ -92,7 +87,7 @@ def get_options(
92
87
cls ,
93
88
target_dir : pathlib .Path | None = None ,
94
89
filepath : pathlib .Path | None = None ,
95
- ** options ,
90
+ ** options : str | bool | int | None ,
96
91
):
97
92
if "outtmpl" not in options :
98
93
outtmpl = cls .options .get ("outtmpl" , cls .defaults ["outtmpl" ])
@@ -143,9 +138,10 @@ def save_large_file(url: str, fpath: pathlib.Path) -> None:
143
138
)
144
139
145
140
146
- def _get_retry_adapter (
141
+ def get_retry_adapter (
147
142
max_retries : int | None = 5 ,
148
143
) -> requests .adapters .BaseAdapter :
144
+ """A requests adapter to automatically retry on known HTTP status that can be"""
149
145
retries = urllib3 .util .retry .Retry (
150
146
total = max_retries , # total number of retries
151
147
connect = max_retries , # connection errors
@@ -169,7 +165,7 @@ def _get_retry_adapter(
169
165
def get_session (max_retries : int | None = 5 ) -> requests .Session :
170
166
"""Session to hold cookies and connection pool together"""
171
167
session = requests .Session ()
172
- session .mount ("http" , _get_retry_adapter (max_retries )) # tied to http and https
168
+ session .mount ("http" , get_retry_adapter (max_retries )) # tied to http and https
173
169
return session
174
170
175
171
0 commit comments