1 from __future__
import annotations
13 from dataclasses
import dataclass
14 from zipimport
import zipimporter
16 from .compat
import compat_realpath
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 # sys.executable returns a /tmp/ path for staticx builds (linux_static)
73 # Ref: https://staticx.readthedocs.io/en/latest/usage.html#run-time-information
74 if static_exe_path
:= os
.getenv('STATICX_PROG_PATH'):
75 path
= static_exe_path
76 return f
'{remove_end(sys.platform, "32")}{machine}_exe', path
78 path
= os
.path
.dirname(__file__
)
79 if isinstance(__loader__
, zipimporter
):
80 return 'zip', os
.path
.join(path
, '..')
81 elif (os
.path
.basename(sys
.argv
[0]) in ('__main__.py', '-m')
82 and os
.path
.exists(os
.path
.join(path
, '../.git/HEAD'))):
84 return 'unknown', path
88 return VARIANT
or _get_variant_and_executable_path()[0]
92 def current_git_head():
93 if detect_variant() != 'source':
95 with contextlib
.suppress(Exception):
96 stdout
, _
, _
= Popen
.run(
97 ['git', 'rev-parse', '--short', 'HEAD'],
98 text
=True, cwd
=os
.path
.dirname(os
.path
.abspath(__file__
)),
99 stdout
=subprocess
.PIPE
, stderr
=subprocess
.PIPE
)
100 if re
.fullmatch('[0-9a-f]+', stdout
.strip()):
101 return stdout
.strip()
107 'win_x86_exe': '_x86.exe',
108 'darwin_exe': '_macos',
109 'darwin_legacy_exe': '_macos_legacy',
110 'linux_exe': '_linux',
111 'linux_aarch64_exe': '_linux_aarch64',
112 'linux_armv7l_exe': '_linux_armv7l',
115 _NON_UPDATEABLE_REASONS
= {
116 **{variant
: None for variant
in _FILE_SUFFIXES
}, # Updatable
117 **{variant
: f
'Auto-update is not supported for unpackaged {name} executable; Re-download the latest release'
118 for variant
, name
in {'win32_dir': 'Windows', 'darwin_dir': 'MacOS', 'linux_dir': 'Linux'}.items()},
119 'py2exe': 'py2exe is no longer supported by yt-dlp; This executable cannot be updated',
120 'source': 'You cannot update when running from source code; Use git to pull the latest changes',
121 'unknown': 'You installed yt-dlp from a manual build or with a package manager; Use that to update',
122 'other': 'You are using an unofficial build of yt-dlp; Build the executable again',
126 def is_non_updateable():
129 return _NON_UPDATEABLE_REASONS
.get(
130 detect_variant(), _NON_UPDATEABLE_REASONS
['unknown' if VARIANT
else 'other'])
133 def _get_binary_name():
134 return format_field(_FILE_SUFFIXES
, detect_variant(), template
='yt-dlp%s', ignore
=None, default
=None)
137 def _get_system_deprecation():
138 MIN_SUPPORTED
, MIN_RECOMMENDED
= (3, 9), (3, 9)
140 if sys
.version_info
> MIN_RECOMMENDED
:
143 major
, minor
= sys
.version_info
[:2]
144 PYTHON_MSG
= f
'Please update to Python {".".join(map(str, MIN_RECOMMENDED))} or above'
146 if sys
.version_info
< MIN_SUPPORTED
:
147 return f
'Python version {major}.{minor} is no longer supported! {PYTHON_MSG}'
149 return f
'Support for Python version {major}.{minor} has been deprecated. {PYTHON_MSG}'
152 def _sha256_file(path
):
154 mv
= memoryview(bytearray(128 * 1024))
155 with
open(os
.path
.realpath(path
), 'rb', buffering
=0) as f
:
156 for n
in iter(lambda: f
.readinto(mv
), 0):
161 def _make_label(origin
, tag
, version
=None):
163 channel
= _INVERSE_UPDATE_SOURCES
.get(origin
, origin
)
166 label
= f
'{channel}@{tag}'
167 if version
and version
!= tag
:
168 label
+= f
' build {version}'
169 if channel
!= origin
:
170 label
+= f
' from {origin}'
177 Update target information
179 Can be created by `query_update()` or manually.
182 tag The release tag that will be updated to. If from query_update,
183 the value is after API resolution and update spec processing.
184 The only property that is required.
185 version The actual numeric version (if available) of the binary to be updated to,
186 after API resolution and update spec processing. (default: None)
187 requested_version Numeric version of the binary being requested (if available),
188 after API resolution only. (default: None)
189 commit Commit hash (if available) of the binary to be updated to,
190 after API resolution and update spec processing. (default: None)
191 This value will only match the RELEASE_GIT_HEAD of prerelease builds.
192 binary_name Filename of the binary to be updated to. (default: current binary name)
193 checksum Expected checksum (if available) of the binary to be
194 updated to. (default: None)
197 version
: str |
None = None
198 requested_version
: str |
None = None
199 commit
: str |
None = None
201 binary_name
: str |
None = _get_binary_name() # noqa: RUF009: Always returns the same value
202 checksum
: str |
None = None
208 # XXX: use class variables to simplify testing
211 _update_sources
= UPDATE_SOURCES
213 def __init__(self
, ydl
, target
: str |
None = None):
215 # For backwards compat, target needs to be treated as if it could be None
216 self
.requested_channel
, sep
, self
.requested_tag
= (target
or self
._channel
).rpartition('@')
217 # Check if requested_tag is actually the requested repo/channel
218 if not sep
and ('/' in self
.requested_tag
or self
.requested_tag
in self
._update
_sources
):
219 self
.requested_channel
= self
.requested_tag
220 self
.requested_tag
: str = None # type: ignore (we set it later)
221 elif not self
.requested_channel
:
222 # User did not specify a channel, so we are requesting the default channel
223 self
.requested_channel
= self
._channel
.partition('@')[0]
225 # --update should not be treated as an exact tag request even if CHANNEL has a @tag
226 self
._exact
= bool(target
) and target
!= self
._channel
227 if not self
.requested_tag
:
228 # User did not specify a tag, so we request 'latest' and track that no exact tag was passed
229 self
.requested_tag
= 'latest'
232 if '/' in self
.requested_channel
:
233 # requested_channel is actually a repository
234 self
.requested_repo
= self
.requested_channel
235 if not self
.requested_repo
.startswith('yt-dlp/') and self
.requested_repo
!= self
._origin
:
236 self
.ydl
.report_warning(
237 f
'You are switching to an {self.ydl._format_err("unofficial", "red")} executable '
238 f
'from {self.ydl._format_err(self.requested_repo, self.ydl.Styles.EMPHASIS)}. '
239 f
'Run {self.ydl._format_err("at your own risk", "light red")}')
240 self
._block
_restart
('Automatically restarting into custom builds is disabled for security reasons')
242 # Check if requested_channel resolves to a known repository or else raise
243 self
.requested_repo
= self
._update
_sources
.get(self
.requested_channel
)
244 if not self
.requested_repo
:
246 f
'Invalid update channel {self.requested_channel!r} requested. '
247 f
'Valid channels are {", ".join(self._update_sources)}', True)
249 self
._identifier
= f
'{detect_variant()} {system_identifier()}'
252 def current_version(self
):
253 """Current version"""
257 def current_commit(self
):
258 """Current commit hash"""
259 return RELEASE_GIT_HEAD
261 def _download_asset(self
, name
, tag
=None):
263 tag
= self
.requested_tag
265 path
= 'latest/download' if tag
== 'latest' else f
'download/{tag}'
266 url
= f
'https://github.com/{self.requested_repo}/releases/{path}/{name}'
267 self
.ydl
.write_debug(f
'Downloading {name} from {url}')
268 return self
.ydl
.urlopen(url
).read()
270 def _call_api(self
, tag
):
271 tag
= f
'tags/{tag}' if tag
!= 'latest' else tag
272 url
= f
'{API_BASE_URL}/{self.requested_repo}/releases/{tag}'
273 self
.ydl
.write_debug(f
'Fetching release info: {url}')
274 return json
.loads(self
.ydl
.urlopen(Request(url
, headers
={
275 'Accept': 'application/vnd.github+json',
276 'User-Agent': 'yt-dlp',
277 'X-GitHub-Api-Version': '2022-11-28',
280 def _get_version_info(self
, tag
: str) -> tuple[str |
None, str |
None]:
281 if _VERSION_RE
.fullmatch(tag
):
284 api_info
= self
._call
_api
(tag
)
287 requested_version
= api_info
['tag_name']
289 match
= re
.search(rf
'\s+(?P<version>{_VERSION_RE.pattern})$', api_info
.get('name', ''))
290 requested_version
= match
.group('version') if match
else None
292 if re
.fullmatch(_HASH_PATTERN
, api_info
.get('target_commitish', '')):
293 target_commitish
= api_info
['target_commitish']
295 match
= _COMMIT_RE
.match(api_info
.get('body', ''))
296 target_commitish
= match
.group('hash') if match
else None
298 if not (requested_version
or target_commitish
):
299 self
._report
_error
('One of either version or commit hash must be available on the release', expected
=True)
301 return requested_version
, target_commitish
303 def _download_update_spec(self
, source_tags
):
304 for tag
in source_tags
:
306 return self
._download
_asset
('_update_spec', tag
=tag
).decode()
307 except network_exceptions
as error
:
308 if isinstance(error
, HTTPError
) and error
.status
== 404:
310 self
._report
_network
_error
(f
'fetch update spec: {error}')
314 f
'The requested tag {self.requested_tag} does not exist for {self.requested_repo}', True)
317 def _process_update_spec(self
, lockfile
: str, resolved_tag
: str):
318 lines
= lockfile
.splitlines()
319 is_version2
= any(line
.startswith('lockV2 ') for line
in lines
)
323 if not line
.startswith(f
'lockV2 {self.requested_repo} '):
325 _
, _
, tag
, pattern
= line
.split(' ', 3)
327 if not line
.startswith('lock '):
329 _
, tag
, pattern
= line
.split(' ', 2)
331 if re
.match(pattern
, self
._identifier
):
332 if _VERSION_RE
.fullmatch(tag
):
335 elif self
._version
_compare
(tag
, resolved_tag
):
337 elif tag
!= resolved_tag
:
341 f
'yt-dlp cannot be updated to {resolved_tag} since you are on an older Python version '
342 'or your operating system is not compatible with the requested build', True)
347 def _version_compare(self
, a
: str, b
: str):
349 Compare two version strings
351 This function SHOULD NOT be called if self._exact == True
353 if _VERSION_RE
.fullmatch(f
'{a}.{b}'):
354 return version_tuple(a
) >= version_tuple(b
)
357 def query_update(self
, *, _output
=False) -> UpdateInfo |
None:
358 """Fetches info about the available update
359 @returns An `UpdateInfo` if there is an update available, else None
361 if not self
.requested_repo
:
362 self
._report
_error
('No target repository could be determined from input')
366 requested_version
, target_commitish
= self
._get
_version
_info
(self
.requested_tag
)
367 except network_exceptions
as e
:
368 self
._report
_network
_error
(f
'obtain version info ({e})', delim
='; Please try again later or')
371 if self
._exact
and self
._origin
!= self
.requested_repo
:
373 elif requested_version
:
375 has_update
= self
.current_version
!= requested_version
377 has_update
= not self
._version
_compare
(self
.current_version
, requested_version
)
378 elif target_commitish
:
379 has_update
= target_commitish
!= self
.current_commit
383 resolved_tag
= requested_version
if self
.requested_tag
== 'latest' else self
.requested_tag
384 current_label
= _make_label(self
._origin
, self
._channel
.partition('@')[2] or self
.current_version
, self
.current_version
)
385 requested_label
= _make_label(self
.requested_repo
, resolved_tag
, requested_version
)
386 latest_or_requested
= f
'{"Latest" if self.requested_tag == "latest" else "Requested"} version: {requested_label}'
389 self
.ydl
.to_screen(f
'{latest_or_requested}\nyt-dlp is up to date ({current_label})')
392 update_spec
= self
._download
_update
_spec
(('latest', None) if requested_version
else (None,))
395 # `result_` prefixed vars == post-_process_update_spec() values
396 result_tag
= self
._process
_update
_spec
(update_spec
, resolved_tag
)
397 if not result_tag
or result_tag
== self
.current_version
:
399 elif result_tag
== resolved_tag
:
400 result_version
= requested_version
401 elif _VERSION_RE
.fullmatch(result_tag
):
402 result_version
= result_tag
403 else: # actual version being updated to is unknown
404 result_version
= None
407 # Non-updateable variants can get update_info but need to skip checksum
408 if not is_non_updateable():
410 hashes
= self
._download
_asset
('SHA2-256SUMS', result_tag
)
411 except network_exceptions
as error
:
412 if not isinstance(error
, HTTPError
) or error
.status
!= 404:
413 self
._report
_network
_error
(f
'fetch checksums: {error}')
415 self
.ydl
.report_warning('No hash information found for the release, skipping verification')
417 for ln
in hashes
.decode().splitlines():
418 if ln
.endswith(_get_binary_name()):
419 checksum
= ln
.split()[0]
422 self
.ydl
.report_warning('The hash could not be found in the checksum file, skipping verification')
425 update_label
= _make_label(self
.requested_repo
, result_tag
, result_version
)
427 f
'Current version: {current_label}\n{latest_or_requested}'
428 + (f
'\nUpgradable to: {update_label}' if update_label
!= requested_label
else ''))
432 version
=result_version
,
433 requested_version
=requested_version
,
434 commit
=target_commitish
if result_tag
== resolved_tag
else None,
437 def update(self
, update_info
=NO_DEFAULT
):
438 """Update yt-dlp executable to the latest version
439 @param update_info `UpdateInfo | None` as returned by query_update()
441 if update_info
is NO_DEFAULT
:
442 update_info
= self
.query_update(_output
=True)
446 err
= is_non_updateable()
448 self
._report
_error
(err
, True)
451 self
.ydl
.to_screen(f
'Current Build Hash: {_sha256_file(self.filename)}')
453 update_label
= _make_label(self
.requested_repo
, update_info
.tag
, update_info
.version
)
454 self
.ydl
.to_screen(f
'Updating to {update_label} ...')
456 directory
= os
.path
.dirname(self
.filename
)
457 if not os
.access(self
.filename
, os
.W_OK
):
458 return self
._report
_permission
_error
(self
.filename
)
459 elif not os
.access(directory
, os
.W_OK
):
460 return self
._report
_permission
_error
(directory
)
462 new_filename
, old_filename
= f
'{self.filename}.new', f
'{self.filename}.old'
463 if detect_variant() == 'zip': # Can be replaced in-place
464 new_filename
, old_filename
= self
.filename
, None
467 if os
.path
.exists(old_filename
or ''):
468 os
.remove(old_filename
)
470 return self
._report
_error
('Unable to remove the old version')
473 newcontent
= self
._download
_asset
(update_info
.binary_name
, update_info
.tag
)
474 except network_exceptions
as e
:
475 if isinstance(e
, HTTPError
) and e
.status
== 404:
476 return self
._report
_error
(
477 f
'The requested tag {self.requested_repo}@{update_info.tag} does not exist', True)
478 return self
._report
_network
_error
(f
'fetch updates: {e}', tag
=update_info
.tag
)
480 if not update_info
.checksum
:
481 self
._block
_restart
('Automatically restarting into unverified builds is disabled for security reasons')
482 elif hashlib
.sha256(newcontent
).hexdigest() != update_info
.checksum
:
483 return self
._report
_network
_error
('verify the new executable', tag
=update_info
.tag
)
486 with
open(new_filename
, 'wb') as outf
:
487 outf
.write(newcontent
)
489 return self
._report
_permission
_error
(new_filename
)
492 mask
= os
.stat(self
.filename
).st_mode
494 os
.rename(self
.filename
, old_filename
)
496 return self
._report
_error
('Unable to move current version')
499 os
.rename(new_filename
, self
.filename
)
501 self
._report
_error
('Unable to overwrite current version')
502 return os
.rename(old_filename
, self
.filename
)
504 variant
= detect_variant()
505 if variant
.startswith('win'):
506 atexit
.register(Popen
, f
'ping 127.0.0.1 -n 5 -w 1000 & del /F "{old_filename}"',
507 shell
=True, stdout
=subprocess
.DEVNULL
, stderr
=subprocess
.DEVNULL
)
510 os
.remove(old_filename
)
512 self
._report
_error
('Unable to remove the old version')
515 os
.chmod(self
.filename
, mask
)
517 return self
._report
_error
(
518 f
'Unable to set permissions. Run: sudo chmod a+rx {shell_quote(self.filename)}')
520 self
.ydl
.to_screen(f
'Updated yt-dlp to {update_label}')
523 @functools.cached_property
525 """Filename of the executable"""
526 return compat_realpath(_get_variant_and_executable_path()[1])
528 @functools.cached_property
530 """The command-line to run the executable, if known"""
531 # There is no sys.orig_argv in py < 3.10. Also, it can be [] when frozen
532 if getattr(sys
, 'orig_argv', None):
534 elif getattr(sys
, 'frozen', False):
538 """Restart the executable"""
539 assert self
.cmd
, 'Must be frozen or Py >= 3.10'
540 self
.ydl
.write_debug(f
'Restarting: {shell_quote(self.cmd)}')
541 _
, _
, returncode
= Popen
.run(self
.cmd
)
544 def _block_restart(self
, msg
):
546 self
._report
_error
(f
'{msg}. Restart yt-dlp to use the updated version', expected
=True)
547 return self
.ydl
._download
_retcode
548 self
.restart
= wrapper
550 def _report_error(self
, msg
, expected
=False):
551 self
.ydl
.report_error(msg
, tb
=False if expected
else None)
552 self
.ydl
._download
_retcode
= 100
554 def _report_permission_error(self
, file):
555 self
._report
_error
(f
'Unable to write to {file}; try running as administrator', True)
557 def _report_network_error(self
, action
, delim
=';', tag
=None):
559 tag
= self
.requested_tag
560 path
= tag
if tag
== 'latest' else f
'tag/{tag}'
562 f
'Unable to {action}{delim} visit '
563 f
'https://github.com/{self.requested_repo}/releases/{path}', True)
565 # XXX: Everything below this line in this class is deprecated / for compat only
567 def _target_tag(self
):
568 """Deprecated; requested tag with 'tags/' prepended when necessary for API calls"""
569 return f
'tags/{self.requested_tag}' if self
.requested_tag
!= 'latest' else self
.requested_tag
571 def _check_update(self
):
572 """Deprecated; report whether there is an update available"""
573 return bool(self
.query_update(_output
=True))
575 def __getattr__(self
, attribute
: str):
576 """Compat getter function for deprecated attributes"""
577 deprecated_props_map
= {
578 'check_update': '_check_update',
579 'target_tag': '_target_tag',
580 'target_channel': 'requested_channel',
582 update_info_props_map
= {
583 'has_update': '_has_update',
584 'new_version': 'version',
585 'latest_version': 'requested_version',
586 'release_name': 'binary_name',
587 'release_hash': 'checksum',
590 if attribute
not in deprecated_props_map
and attribute
not in update_info_props_map
:
591 raise AttributeError(f
'{type(self).__name__!r} object has no attribute {attribute!r}')
593 msg
= f
'{type(self).__name__}.{attribute} is deprecated and will be removed in a future version'
594 if attribute
in deprecated_props_map
:
595 source_name
= deprecated_props_map
[attribute
]
596 if not source_name
.startswith('_'):
597 msg
+= f
'. Please use {source_name!r} instead'
599 mapping
= deprecated_props_map
601 else: # attribute in update_info_props_map
602 msg
+= '. Please call query_update() instead'
603 source
= self
.query_update()
605 source
= UpdateInfo('', None, None, None)
606 source
._has
_update
= False
607 mapping
= update_info_props_map
609 deprecation_warning(msg
)
610 for target_name
, source_name
in mapping
.items():
611 value
= getattr(source
, source_name
)
612 setattr(self
, target_name
, value
)
614 return getattr(self
, attribute
)
618 """Update the program file with the latest version from the repository
619 @returns Whether there was a successful update (No update = False)
621 return Updater(ydl
).update()
624 __all__
= ['Updater']