1 from __future__
import annotations
12 from dataclasses
import dataclass
13 from zipimport
import zipimporter
15 from .compat
import functools
# isort: split
16 from .compat
import compat_realpath
, compat_shlex_quote
17 from .networking
import Request
18 from .networking
.exceptions
import HTTPError
, network_exceptions
29 from .version
import (
39 'stable': 'yt-dlp/yt-dlp',
40 'nightly': 'yt-dlp/yt-dlp-nightly-builds',
41 'master': 'yt-dlp/yt-dlp-master-builds',
43 REPOSITORY
= UPDATE_SOURCES
['stable']
44 _INVERSE_UPDATE_SOURCES
= {value: key for key, value in UPDATE_SOURCES.items()}
46 _VERSION_RE
= re
.compile(r
'(\d+\.)*\d+')
47 _HASH_PATTERN
= r
'[\da-f]{40}'
48 _COMMIT_RE
= re
.compile(rf
'Generated from: https://(?:[^/?#]+/){{3}}commit/(?P<hash>{_HASH_PATTERN})')
50 API_BASE_URL
= 'https://api.github.com/repos'
52 # Backwards compatibility variables for the current channel
53 API_URL
= f
'{API_BASE_URL}/{REPOSITORY}/releases'
57 def _get_variant_and_executable_path():
58 """@returns (variant, executable_path)"""
59 if getattr(sys
, 'frozen', False):
61 if not hasattr(sys
, '_MEIPASS'):
63 elif sys
._MEIPASS
== os
.path
.dirname(path
):
64 return f
'{sys.platform}_dir', path
65 elif sys
.platform
== 'darwin':
66 machine
= '_legacy' if version_tuple(platform
.mac_ver()[0]) < (10, 15) else ''
68 machine
= f
'_{platform.machine().lower()}'
69 # Ref: https://en.wikipedia.org/wiki/Uname#Examples
70 if machine
[1:] in ('x86', 'x86_64', 'amd64', 'i386', 'i686'):
71 machine
= '_x86' if platform
.architecture()[0][:2] == '32' else ''
72 return f
'{remove_end(sys.platform, "32")}{machine}_exe', path
74 path
= os
.path
.dirname(__file__
)
75 if isinstance(__loader__
, zipimporter
):
76 return 'zip', os
.path
.join(path
, '..')
77 elif (os
.path
.basename(sys
.argv
[0]) in ('__main__.py', '-m')
78 and os
.path
.exists(os
.path
.join(path
, '../.git/HEAD'))):
80 return 'unknown', path
84 return VARIANT
or _get_variant_and_executable_path()[0]
88 def current_git_head():
89 if detect_variant() != 'source':
91 with contextlib
.suppress(Exception):
92 stdout
, _
, _
= Popen
.run(
93 ['git', 'rev-parse', '--short', 'HEAD'],
94 text
=True, cwd
=os
.path
.dirname(os
.path
.abspath(__file__
)),
95 stdout
=subprocess
.PIPE
, stderr
=subprocess
.PIPE
)
96 if re
.fullmatch('[0-9a-f]+', stdout
.strip()):
102 'py2exe': '_min.exe',
104 'win_x86_exe': '_x86.exe',
105 'darwin_exe': '_macos',
106 'darwin_legacy_exe': '_macos_legacy',
107 'linux_exe': '_linux',
108 'linux_aarch64_exe': '_linux_aarch64',
109 'linux_armv7l_exe': '_linux_armv7l',
112 _NON_UPDATEABLE_REASONS
= {
113 **{variant: None for variant in _FILE_SUFFIXES}
, # Updatable
114 **{variant: f'Auto-update is not supported for unpackaged {name} executable
; Re
-download the latest release
'
115 for variant, name in {'win32_dir': 'Windows', 'darwin_dir': 'MacOS', 'linux_dir': 'Linux'}.items()},
116 'source
': 'You cannot update when running
from source code
; Use git to pull the latest changes
',
117 'unknown
': 'You installed yt
-dlp
with a package manager
or setup
.py
; Use that to update
',
118 'other
': 'You are using an unofficial build of yt
-dlp
; Build the executable again
',
122 def is_non_updateable():
125 return _NON_UPDATEABLE_REASONS.get(
126 detect_variant(), _NON_UPDATEABLE_REASONS['unknown
' if VARIANT else 'other
'])
129 def _get_binary_name():
130 return format_field(_FILE_SUFFIXES, detect_variant(), template='yt
-dlp
%s', ignore=None, default=None)
133 def _get_system_deprecation():
134 MIN_SUPPORTED, MIN_RECOMMENDED = (3, 8), (3, 8)
136 if sys.version_info > MIN_RECOMMENDED:
139 major, minor = sys.version_info[:2]
140 if sys.version_info < MIN_SUPPORTED:
141 msg = f'Python version {major}
.{minor}
is no longer supported
'
143 msg = (f'Support
for Python version {major}
.{minor} has been deprecated
. '
144 '\nYou may stop receiving updates on this version at any time
')
146 major, minor = MIN_RECOMMENDED
147 return f'{msg}
! Please update to Python {major}
.{minor}
or above
'
150 def _sha256_file(path):
152 mv = memoryview(bytearray(128 * 1024))
153 with open(os.path.realpath(path), 'rb
', buffering=0) as f:
154 for n in iter(lambda: f.readinto(mv), 0):
159 def _make_label(origin, tag, version=None):
161 channel = _INVERSE_UPDATE_SOURCES.get(origin, origin)
164 label = f'{channel}
@{tag}
'
165 if version and version != tag:
166 label += f' build {version}
'
167 if channel != origin:
168 label += f' from {origin}
'
175 Update target information
177 Can be created by `query_update()` or manually.
180 tag The release tag that will be updated to. If from query_update,
181 the value is after API resolution and update spec processing.
182 The only property that is required.
183 version The actual numeric version (if available) of the binary to be updated to,
184 after API resolution and update spec processing. (default: None)
185 requested_version Numeric version of the binary being requested (if available),
186 after API resolution only. (default: None)
187 commit Commit hash (if available) of the binary to be updated to,
188 after API resolution and update spec processing. (default: None)
189 This value will only match the RELEASE_GIT_HEAD of prerelease builds.
190 binary_name Filename of the binary to be updated to. (default: current binary name)
191 checksum Expected checksum (if available) of the binary to be
192 updated to. (default: None)
195 version: str | None = None
196 requested_version: str | None = None
197 commit: str | None = None
199 binary_name: str | None = _get_binary_name()
200 checksum: str | None = None
206 # XXX: use class variables to simplify testing
209 _update_sources = UPDATE_SOURCES
211 def __init__(self, ydl, target: str | None = None):
213 # For backwards compat, target needs to be treated as if it could be None
214 self.requested_channel, sep, self.requested_tag = (target or self._channel).rpartition('@')
215 # Check if requested_tag is actually the requested repo/channel
216 if not sep and ('/' in self.requested_tag or self.requested_tag in self._update_sources):
217 self.requested_channel = self.requested_tag
218 self.requested_tag: str = None # type: ignore (we set it later)
219 elif not self.requested_channel:
220 # User did not specify a channel, so we are requesting the default channel
221 self.requested_channel = self._channel.partition('@')[0]
223 # --update should not be treated as an exact tag request even if CHANNEL has a @tag
224 self._exact = bool(target) and target != self._channel
225 if not self.requested_tag:
226 # User did not specify a tag, so we request 'latest
' and track that no exact tag was passed
227 self.requested_tag = 'latest
'
230 if '/' in self.requested_channel:
231 # requested_channel is actually a repository
232 self.requested_repo = self.requested_channel
233 if not self.requested_repo.startswith('yt
-dlp
/') and self.requested_repo != self._origin:
234 self.ydl.report_warning(
235 f'You are switching to an {self.ydl._format_err("unofficial", "red")} executable
'
236 f'from {self.ydl._format_err(self.requested_repo, self.ydl.Styles.EMPHASIS)}
. '
237 f'Run {self.ydl._format_err("at your own risk", "light red")}
')
238 self._block_restart('Automatically restarting into custom builds
is disabled
for security reasons
')
240 # Check if requested_channel resolves to a known repository or else raise
241 self.requested_repo = self._update_sources.get(self.requested_channel)
242 if not self.requested_repo:
244 f'Invalid update channel {self.requested_channel!r} requested
. '
245 f'Valid channels are {", ".join(self._update_sources)}
', True)
247 self._identifier = f'{detect_variant()} {system_identifier()}
'
250 def current_version(self):
251 """Current version"""
255 def current_commit(self):
256 """Current commit hash"""
257 return RELEASE_GIT_HEAD
259 def _download_asset(self, name, tag=None):
261 tag = self.requested_tag
263 path = 'latest
/download
' if tag == 'latest
' else f'download
/{tag}
'
264 url = f'https
://github
.com
/{self.requested_repo}
/releases
/{path}
/{name}
'
265 self.ydl.write_debug(f'Downloading {name}
from {url}
')
266 return self.ydl.urlopen(url).read()
268 def _call_api(self, tag):
269 tag = f'tags
/{tag}
' if tag != 'latest
' else tag
270 url = f'{API_BASE_URL}
/{self.requested_repo}
/releases
/{tag}
'
271 self.ydl.write_debug(f'Fetching release info
: {url}
')
272 return json.loads(self.ydl.urlopen(Request(url, headers={
273 'Accept
': 'application
/vnd
.github
+json
',
274 'User
-Agent
': 'yt
-dlp
',
275 'X
-GitHub
-Api
-Version
': '2022-11-28',
278 def _get_version_info(self, tag: str) -> tuple[str | None, str | None]:
279 if _VERSION_RE.fullmatch(tag):
282 api_info = self._call_api(tag)
285 requested_version = api_info['tag_name
']
287 match = re.search(rf'\s
+(?P
<version
>{_VERSION_RE.pattern}
)$
', api_info.get('name
', ''))
288 requested_version = match.group('version
') if match else None
290 if re.fullmatch(_HASH_PATTERN, api_info.get('target_commitish
', '')):
291 target_commitish = api_info['target_commitish
']
293 match = _COMMIT_RE.match(api_info.get('body
', ''))
294 target_commitish = match.group('hash') if match else None
296 if not (requested_version or target_commitish):
297 self._report_error('One of either version
or commit
hash must be available on the release
', expected=True)
299 return requested_version, target_commitish
301 def _download_update_spec(self, source_tags):
302 for tag in source_tags:
304 return self._download_asset('_update_spec
', tag=tag).decode()
305 except network_exceptions as error:
306 if isinstance(error, HTTPError) and error.status == 404:
308 self._report_network_error(f'fetch update spec
: {error}
')
311 f'The requested tag {self.requested_tag} does
not exist
for {self.requested_repo}
', True)
314 def _process_update_spec(self, lockfile: str, resolved_tag: str):
315 lines = lockfile.splitlines()
316 is_version2 = any(line.startswith('lockV2
') for line in lines)
320 if not line.startswith(f'lockV2 {self.requested_repo}
'):
322 _, _, tag, pattern = line.split(' ', 3)
324 if not line.startswith('lock
'):
326 _, tag, pattern = line.split(' ', 2)
328 if re.match(pattern, self._identifier):
329 if _VERSION_RE.fullmatch(tag):
332 elif self._version_compare(tag, resolved_tag):
334 elif tag != resolved_tag:
338 f'yt
-dlp cannot be updated to {resolved_tag} since you are on an older Python version
', True)
343 def _version_compare(self, a: str, b: str):
345 Compare two version strings
347 This function SHOULD NOT be called if self._exact == True
349 if _VERSION_RE.fullmatch(f'{a}
.{b}
'):
350 return version_tuple(a) >= version_tuple(b)
353 def query_update(self, *, _output=False) -> UpdateInfo | None:
354 """Fetches and returns info about the available update"""
355 if not self.requested_repo:
356 self._report_error('No target repository could be determined
from input')
360 requested_version, target_commitish = self._get_version_info(self.requested_tag)
361 except network_exceptions as e:
362 self._report_network_error(f'obtain version
info ({e}
)', delim='; Please
try again later
or')
365 if self._exact and self._origin != self.requested_repo:
367 elif requested_version:
369 has_update = self.current_version != requested_version
371 has_update = not self._version_compare(self.current_version, requested_version)
372 elif target_commitish:
373 has_update = target_commitish != self.current_commit
377 resolved_tag = requested_version if self.requested_tag == 'latest
' else self.requested_tag
378 current_label = _make_label(self._origin, self._channel.partition("@")[2] or self.current_version, self.current_version)
379 requested_label = _make_label(self.requested_repo, resolved_tag, requested_version)
380 latest_or_requested = f'{"Latest" if self.requested_tag == "latest" else "Requested"} version
: {requested_label}
'
383 self.ydl.to_screen(f'{latest_or_requested}
\nyt
-dlp
is up to
date ({current_label}
)')
386 update_spec = self._download_update_spec(('latest
', None) if requested_version else (None,))
389 # `result_` prefixed vars == post-_process_update_spec() values
390 result_tag = self._process_update_spec(update_spec, resolved_tag)
391 if not result_tag or result_tag == self.current_version:
393 elif result_tag == resolved_tag:
394 result_version = requested_version
395 elif _VERSION_RE.fullmatch(result_tag):
396 result_version = result_tag
397 else: # actual version being updated to is unknown
398 result_version = None
401 # Non-updateable variants can get update_info but need to skip checksum
402 if not is_non_updateable():
404 hashes = self._download_asset('SHA2
-256SUMS
', result_tag)
405 except network_exceptions as error:
406 if not isinstance(error, HTTPError) or error.status != 404:
407 self._report_network_error(f'fetch checksums
: {error}
')
409 self.ydl.report_warning('No
hash information found
for the release
, skipping verification
')
411 for ln in hashes.decode().splitlines():
412 if ln.endswith(_get_binary_name()):
413 checksum = ln.split()[0]
416 self.ydl.report_warning('The
hash could
not be found
in the checksum
file, skipping verification
')
419 update_label = _make_label(self.requested_repo, result_tag, result_version)
421 f'Current version
: {current_label}
\n{latest_or_requested}
'
422 + (f'\nUpgradable to
: {update_label}
' if update_label != requested_label else ''))
426 version=result_version,
427 requested_version=requested_version,
428 commit=target_commitish if result_tag == resolved_tag else None,
431 def update(self, update_info=NO_DEFAULT):
432 """Update yt-dlp executable to the latest version"""
433 if update_info is NO_DEFAULT:
434 update_info = self.query_update(_output=True)
438 err = is_non_updateable()
440 self._report_error(err, True)
443 self.ydl.to_screen(f'Current Build Hash
: {_sha256_file(self.filename)}
')
445 update_label = _make_label(self.requested_repo, update_info.tag, update_info.version)
446 self.ydl.to_screen(f'Updating to {update_label}
...')
448 directory = os.path.dirname(self.filename)
449 if not os.access(self.filename, os.W_OK):
450 return self._report_permission_error(self.filename)
451 elif not os.access(directory, os.W_OK):
452 return self._report_permission_error(directory)
454 new_filename, old_filename = f'{self.filename}
.new
', f'{self.filename}
.old
'
455 if detect_variant() == 'zip': # Can be replaced in-place
456 new_filename, old_filename = self.filename, None
459 if os.path.exists(old_filename or ''):
460 os.remove(old_filename)
462 return self._report_error('Unable to remove the old version
')
465 newcontent = self._download_asset(update_info.binary_name, update_info.tag)
466 except network_exceptions as e:
467 if isinstance(e, HTTPError) and e.status == 404:
468 return self._report_error(
469 f'The requested tag {self.requested_repo}
@{update_info.tag} does
not exist
', True)
470 return self._report_network_error(f'fetch updates
: {e}
', tag=update_info.tag)
472 if not update_info.checksum:
473 self._block_restart('Automatically restarting into unverified builds
is disabled
for security reasons
')
474 elif hashlib.sha256(newcontent).hexdigest() != update_info.checksum:
475 return self._report_network_error('verify the new executable
', tag=update_info.tag)
478 with open(new_filename, 'wb
') as outf:
479 outf.write(newcontent)
481 return self._report_permission_error(new_filename)
484 mask = os.stat(self.filename).st_mode
486 os.rename(self.filename, old_filename)
488 return self._report_error('Unable to move current version
')
491 os.rename(new_filename, self.filename)
493 self._report_error('Unable to overwrite current version
')
494 return os.rename(old_filename, self.filename)
496 variant = detect_variant()
497 if variant.startswith('win
') or variant == 'py2exe
':
498 atexit.register(Popen, f'ping
127.0.0.1 -n
5 -w
1000 & del /F
"{old_filename}"',
499 shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
502 os.remove(old_filename)
504 self._report_error('Unable to remove the old version
')
507 os.chmod(self.filename, mask)
509 return self._report_error(
510 f'Unable to
set permissions
. Run
: sudo chmod a
+rx {compat_shlex_quote(self.filename)}
')
512 self.ydl.to_screen(f'Updated yt
-dlp to {update_label}
')
515 @functools.cached_property
517 """Filename of the executable"""
518 return compat_realpath(_get_variant_and_executable_path()[1])
520 @functools.cached_property
522 """The command-line to run the executable, if known"""
523 # There is no sys.orig_argv in py < 3.10. Also, it can be [] when frozen
524 if getattr(sys, 'orig_argv
', None):
526 elif getattr(sys, 'frozen
', False):
530 """Restart the executable"""
531 assert self.cmd, 'Must be frozen
or Py
>= 3.10'
532 self.ydl.write_debug(f'Restarting
: {shell_quote(self.cmd)}
')
533 _, _, returncode = Popen.run(self.cmd)
536 def _block_restart(self, msg):
538 self._report_error(f'{msg}
. Restart yt
-dlp to use the updated version
', expected=True)
539 return self.ydl._download_retcode
540 self.restart = wrapper
542 def _report_error(self, msg, expected=False):
543 self.ydl.report_error(msg, tb=False if expected else None)
544 self.ydl._download_retcode = 100
546 def _report_permission_error(self, file):
547 self._report_error(f'Unable to write to {file}
; try running
as administrator
', True)
549 def _report_network_error(self, action, delim=';', tag=None):
551 tag = self.requested_tag
553 f'Unable to {action}{delim} visit https
://github
.com
/{self.requested_repo}
/releases
/'
554 + tag if tag == "latest" else f"tag/{tag}", True)
556 # XXX: Everything below this line in this class is deprecated / for compat only
558 def _target_tag(self):
559 """Deprecated; requested tag with 'tags
/' prepended when necessary for API calls"""
560 return f'tags
/{self.requested_tag}
' if self.requested_tag != 'latest
' else self.requested_tag
562 def _check_update(self):
563 """Deprecated; report whether there is an update available"""
564 return bool(self.query_update(_output=True))
566 def __getattr__(self, attribute: str):
567 """Compat getter function for deprecated attributes"""
568 deprecated_props_map = {
569 'check_update
': '_check_update
',
570 'target_tag
': '_target_tag
',
571 'target_channel
': 'requested_channel
',
573 update_info_props_map = {
574 'has_update
': '_has_update
',
575 'new_version
': 'version
',
576 'latest_version
': 'requested_version
',
577 'release_name
': 'binary_name
',
578 'release_hash
': 'checksum
',
581 if attribute not in deprecated_props_map and attribute not in update_info_props_map:
582 raise AttributeError(f'{type(self).__name__!r}
object has no attribute {attribute!r}
')
584 msg = f'{type(self).__name__}
.{attribute}
is deprecated
and will be removed
in a future version
'
585 if attribute in deprecated_props_map:
586 source_name = deprecated_props_map[attribute]
587 if not source_name.startswith('_
'):
588 msg += f'. Please use {source_name!r} instead
'
590 mapping = deprecated_props_map
592 else: # attribute in update_info_props_map
593 msg += '. Please call
query_update() instead
'
594 source = self.query_update()
596 source = UpdateInfo('', None, None, None)
597 source._has_update = False
598 mapping = update_info_props_map
600 deprecation_warning(msg)
601 for target_name, source_name in mapping.items():
602 value = getattr(source, source_name)
603 setattr(self, target_name, value)
605 return getattr(self, attribute)
609 """Update the program file with the latest version from the repository
610 @returns Whether there was a successful update (No update = False)
612 return Updater(ydl).update()
615 __all__ = ['Updater
']