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
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()
106 'py2exe': '_min.exe',
108 'win_x86_exe': '_x86.exe',
109 'darwin_exe': '_macos',
110 'darwin_legacy_exe': '_macos_legacy',
111 'linux_exe': '_linux',
112 'linux_aarch64_exe': '_linux_aarch64',
113 'linux_armv7l_exe': '_linux_armv7l',
116 _NON_UPDATEABLE_REASONS
= {
117 **{variant
: None for variant
in _FILE_SUFFIXES
}, # Updatable
118 **{variant
: f
'Auto-update is not supported for unpackaged {name} executable; Re-download the latest release'
119 for variant
, name
in {'win32_dir': 'Windows', 'darwin_dir': 'MacOS', 'linux_dir': 'Linux'}.items()},
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, 8), (3, 8)
140 if sys
.version_info
> MIN_RECOMMENDED
:
143 major
, minor
= sys
.version_info
[:2]
144 if sys
.version_info
< MIN_SUPPORTED
:
145 msg
= f
'Python version {major}.{minor} is no longer supported'
147 msg
= (f
'Support for Python version {major}.{minor} has been deprecated. '
148 '\nYou may stop receiving updates on this version at any time')
150 major
, minor
= MIN_RECOMMENDED
151 return f
'{msg}! Please update to Python {major}.{minor} or above'
154 def _sha256_file(path
):
156 mv
= memoryview(bytearray(128 * 1024))
157 with
open(os
.path
.realpath(path
), 'rb', buffering
=0) as f
:
158 for n
in iter(lambda: f
.readinto(mv
), 0):
163 def _make_label(origin
, tag
, version
=None):
165 channel
= _INVERSE_UPDATE_SOURCES
.get(origin
, origin
)
168 label
= f
'{channel}@{tag}'
169 if version
and version
!= tag
:
170 label
+= f
' build {version}'
171 if channel
!= origin
:
172 label
+= f
' from {origin}'
179 Update target information
181 Can be created by `query_update()` or manually.
184 tag The release tag that will be updated to. If from query_update,
185 the value is after API resolution and update spec processing.
186 The only property that is required.
187 version The actual numeric version (if available) of the binary to be updated to,
188 after API resolution and update spec processing. (default: None)
189 requested_version Numeric version of the binary being requested (if available),
190 after API resolution only. (default: None)
191 commit Commit hash (if available) of the binary to be updated to,
192 after API resolution and update spec processing. (default: None)
193 This value will only match the RELEASE_GIT_HEAD of prerelease builds.
194 binary_name Filename of the binary to be updated to. (default: current binary name)
195 checksum Expected checksum (if available) of the binary to be
196 updated to. (default: None)
199 version
: str |
None = None
200 requested_version
: str |
None = None
201 commit
: str |
None = None
203 binary_name
: str |
None = _get_binary_name() # noqa: RUF009: Always returns the same value
204 checksum
: str |
None = None
210 # XXX: use class variables to simplify testing
213 _update_sources
= UPDATE_SOURCES
215 def __init__(self
, ydl
, target
: str |
None = None):
217 # For backwards compat, target needs to be treated as if it could be None
218 self
.requested_channel
, sep
, self
.requested_tag
= (target
or self
._channel
).rpartition('@')
219 # Check if requested_tag is actually the requested repo/channel
220 if not sep
and ('/' in self
.requested_tag
or self
.requested_tag
in self
._update
_sources
):
221 self
.requested_channel
= self
.requested_tag
222 self
.requested_tag
: str = None # type: ignore (we set it later)
223 elif not self
.requested_channel
:
224 # User did not specify a channel, so we are requesting the default channel
225 self
.requested_channel
= self
._channel
.partition('@')[0]
227 # --update should not be treated as an exact tag request even if CHANNEL has a @tag
228 self
._exact
= bool(target
) and target
!= self
._channel
229 if not self
.requested_tag
:
230 # User did not specify a tag, so we request 'latest' and track that no exact tag was passed
231 self
.requested_tag
= 'latest'
234 if '/' in self
.requested_channel
:
235 # requested_channel is actually a repository
236 self
.requested_repo
= self
.requested_channel
237 if not self
.requested_repo
.startswith('yt-dlp/') and self
.requested_repo
!= self
._origin
:
238 self
.ydl
.report_warning(
239 f
'You are switching to an {self.ydl._format_err("unofficial", "red")} executable '
240 f
'from {self.ydl._format_err(self.requested_repo, self.ydl.Styles.EMPHASIS)}. '
241 f
'Run {self.ydl._format_err("at your own risk", "light red")}')
242 self
._block
_restart
('Automatically restarting into custom builds is disabled for security reasons')
244 # Check if requested_channel resolves to a known repository or else raise
245 self
.requested_repo
= self
._update
_sources
.get(self
.requested_channel
)
246 if not self
.requested_repo
:
248 f
'Invalid update channel {self.requested_channel!r} requested. '
249 f
'Valid channels are {", ".join(self._update_sources)}', True)
251 self
._identifier
= f
'{detect_variant()} {system_identifier()}'
254 def current_version(self
):
255 """Current version"""
259 def current_commit(self
):
260 """Current commit hash"""
261 return RELEASE_GIT_HEAD
263 def _download_asset(self
, name
, tag
=None):
265 tag
= self
.requested_tag
267 path
= 'latest/download' if tag
== 'latest' else f
'download/{tag}'
268 url
= f
'https://github.com/{self.requested_repo}/releases/{path}/{name}'
269 self
.ydl
.write_debug(f
'Downloading {name} from {url}')
270 return self
.ydl
.urlopen(url
).read()
272 def _call_api(self
, tag
):
273 tag
= f
'tags/{tag}' if tag
!= 'latest' else tag
274 url
= f
'{API_BASE_URL}/{self.requested_repo}/releases/{tag}'
275 self
.ydl
.write_debug(f
'Fetching release info: {url}')
276 return json
.loads(self
.ydl
.urlopen(Request(url
, headers
={
277 'Accept': 'application/vnd.github+json',
278 'User-Agent': 'yt-dlp',
279 'X-GitHub-Api-Version': '2022-11-28',
282 def _get_version_info(self
, tag
: str) -> tuple[str |
None, str |
None]:
283 if _VERSION_RE
.fullmatch(tag
):
286 api_info
= self
._call
_api
(tag
)
289 requested_version
= api_info
['tag_name']
291 match
= re
.search(rf
'\s+(?P<version>{_VERSION_RE.pattern})$', api_info
.get('name', ''))
292 requested_version
= match
.group('version') if match
else None
294 if re
.fullmatch(_HASH_PATTERN
, api_info
.get('target_commitish', '')):
295 target_commitish
= api_info
['target_commitish']
297 match
= _COMMIT_RE
.match(api_info
.get('body', ''))
298 target_commitish
= match
.group('hash') if match
else None
300 if not (requested_version
or target_commitish
):
301 self
._report
_error
('One of either version or commit hash must be available on the release', expected
=True)
303 return requested_version
, target_commitish
305 def _download_update_spec(self
, source_tags
):
306 for tag
in source_tags
:
308 return self
._download
_asset
('_update_spec', tag
=tag
).decode()
309 except network_exceptions
as error
:
310 if isinstance(error
, HTTPError
) and error
.status
== 404:
312 self
._report
_network
_error
(f
'fetch update spec: {error}')
316 f
'The requested tag {self.requested_tag} does not exist for {self.requested_repo}', True)
319 def _process_update_spec(self
, lockfile
: str, resolved_tag
: str):
320 lines
= lockfile
.splitlines()
321 is_version2
= any(line
.startswith('lockV2 ') for line
in lines
)
325 if not line
.startswith(f
'lockV2 {self.requested_repo} '):
327 _
, _
, tag
, pattern
= line
.split(' ', 3)
329 if not line
.startswith('lock '):
331 _
, tag
, pattern
= line
.split(' ', 2)
333 if re
.match(pattern
, self
._identifier
):
334 if _VERSION_RE
.fullmatch(tag
):
337 elif self
._version
_compare
(tag
, resolved_tag
):
339 elif tag
!= resolved_tag
:
343 f
'yt-dlp cannot be updated to {resolved_tag} since you are on an older Python version', True)
348 def _version_compare(self
, a
: str, b
: str):
350 Compare two version strings
352 This function SHOULD NOT be called if self._exact == True
354 if _VERSION_RE
.fullmatch(f
'{a}.{b}'):
355 return version_tuple(a
) >= version_tuple(b
)
358 def query_update(self
, *, _output
=False) -> UpdateInfo |
None:
359 """Fetches info about the available update
360 @returns An `UpdateInfo` if there is an update available, else None
362 if not self
.requested_repo
:
363 self
._report
_error
('No target repository could be determined from input')
367 requested_version
, target_commitish
= self
._get
_version
_info
(self
.requested_tag
)
368 except network_exceptions
as e
:
369 self
._report
_network
_error
(f
'obtain version info ({e})', delim
='; Please try again later or')
372 if self
._exact
and self
._origin
!= self
.requested_repo
:
374 elif requested_version
:
376 has_update
= self
.current_version
!= requested_version
378 has_update
= not self
._version
_compare
(self
.current_version
, requested_version
)
379 elif target_commitish
:
380 has_update
= target_commitish
!= self
.current_commit
384 resolved_tag
= requested_version
if self
.requested_tag
== 'latest' else self
.requested_tag
385 current_label
= _make_label(self
._origin
, self
._channel
.partition('@')[2] or self
.current_version
, self
.current_version
)
386 requested_label
= _make_label(self
.requested_repo
, resolved_tag
, requested_version
)
387 latest_or_requested
= f
'{"Latest" if self.requested_tag == "latest" else "Requested"} version: {requested_label}'
390 self
.ydl
.to_screen(f
'{latest_or_requested}\nyt-dlp is up to date ({current_label})')
393 update_spec
= self
._download
_update
_spec
(('latest', None) if requested_version
else (None,))
396 # `result_` prefixed vars == post-_process_update_spec() values
397 result_tag
= self
._process
_update
_spec
(update_spec
, resolved_tag
)
398 if not result_tag
or result_tag
== self
.current_version
:
400 elif result_tag
== resolved_tag
:
401 result_version
= requested_version
402 elif _VERSION_RE
.fullmatch(result_tag
):
403 result_version
= result_tag
404 else: # actual version being updated to is unknown
405 result_version
= None
408 # Non-updateable variants can get update_info but need to skip checksum
409 if not is_non_updateable():
411 hashes
= self
._download
_asset
('SHA2-256SUMS', result_tag
)
412 except network_exceptions
as error
:
413 if not isinstance(error
, HTTPError
) or error
.status
!= 404:
414 self
._report
_network
_error
(f
'fetch checksums: {error}')
416 self
.ydl
.report_warning('No hash information found for the release, skipping verification')
418 for ln
in hashes
.decode().splitlines():
419 if ln
.endswith(_get_binary_name()):
420 checksum
= ln
.split()[0]
423 self
.ydl
.report_warning('The hash could not be found in the checksum file, skipping verification')
426 update_label
= _make_label(self
.requested_repo
, result_tag
, result_version
)
428 f
'Current version: {current_label}\n{latest_or_requested}'
429 + (f
'\nUpgradable to: {update_label}' if update_label
!= requested_label
else ''))
433 version
=result_version
,
434 requested_version
=requested_version
,
435 commit
=target_commitish
if result_tag
== resolved_tag
else None,
438 def update(self
, update_info
=NO_DEFAULT
):
439 """Update yt-dlp executable to the latest version
440 @param update_info `UpdateInfo | None` as returned by query_update()
442 if update_info
is NO_DEFAULT
:
443 update_info
= self
.query_update(_output
=True)
447 err
= is_non_updateable()
449 self
._report
_error
(err
, True)
452 self
.ydl
.to_screen(f
'Current Build Hash: {_sha256_file(self.filename)}')
454 update_label
= _make_label(self
.requested_repo
, update_info
.tag
, update_info
.version
)
455 self
.ydl
.to_screen(f
'Updating to {update_label} ...')
457 directory
= os
.path
.dirname(self
.filename
)
458 if not os
.access(self
.filename
, os
.W_OK
):
459 return self
._report
_permission
_error
(self
.filename
)
460 elif not os
.access(directory
, os
.W_OK
):
461 return self
._report
_permission
_error
(directory
)
463 new_filename
, old_filename
= f
'{self.filename}.new', f
'{self.filename}.old'
464 if detect_variant() == 'zip': # Can be replaced in-place
465 new_filename
, old_filename
= self
.filename
, None
468 if os
.path
.exists(old_filename
or ''):
469 os
.remove(old_filename
)
471 return self
._report
_error
('Unable to remove the old version')
474 newcontent
= self
._download
_asset
(update_info
.binary_name
, update_info
.tag
)
475 except network_exceptions
as e
:
476 if isinstance(e
, HTTPError
) and e
.status
== 404:
477 return self
._report
_error
(
478 f
'The requested tag {self.requested_repo}@{update_info.tag} does not exist', True)
479 return self
._report
_network
_error
(f
'fetch updates: {e}', tag
=update_info
.tag
)
481 if not update_info
.checksum
:
482 self
._block
_restart
('Automatically restarting into unverified builds is disabled for security reasons')
483 elif hashlib
.sha256(newcontent
).hexdigest() != update_info
.checksum
:
484 return self
._report
_network
_error
('verify the new executable', tag
=update_info
.tag
)
487 with
open(new_filename
, 'wb') as outf
:
488 outf
.write(newcontent
)
490 return self
._report
_permission
_error
(new_filename
)
493 mask
= os
.stat(self
.filename
).st_mode
495 os
.rename(self
.filename
, old_filename
)
497 return self
._report
_error
('Unable to move current version')
500 os
.rename(new_filename
, self
.filename
)
502 self
._report
_error
('Unable to overwrite current version')
503 return os
.rename(old_filename
, self
.filename
)
505 variant
= detect_variant()
506 if variant
.startswith('win') or variant
== 'py2exe':
507 atexit
.register(Popen
, f
'ping 127.0.0.1 -n 5 -w 1000 & del /F "{old_filename}"',
508 shell
=True, stdout
=subprocess
.DEVNULL
, stderr
=subprocess
.DEVNULL
)
511 os
.remove(old_filename
)
513 self
._report
_error
('Unable to remove the old version')
516 os
.chmod(self
.filename
, mask
)
518 return self
._report
_error
(
519 f
'Unable to set permissions. Run: sudo chmod a+rx {shell_quote(self.filename)}')
521 self
.ydl
.to_screen(f
'Updated yt-dlp to {update_label}')
524 @functools.cached_property
526 """Filename of the executable"""
527 return compat_realpath(_get_variant_and_executable_path()[1])
529 @functools.cached_property
531 """The command-line to run the executable, if known"""
532 # There is no sys.orig_argv in py < 3.10. Also, it can be [] when frozen
533 if getattr(sys
, 'orig_argv', None):
535 elif getattr(sys
, 'frozen', False):
539 """Restart the executable"""
540 assert self
.cmd
, 'Must be frozen or Py >= 3.10'
541 self
.ydl
.write_debug(f
'Restarting: {shell_quote(self.cmd)}')
542 _
, _
, returncode
= Popen
.run(self
.cmd
)
545 def _block_restart(self
, msg
):
547 self
._report
_error
(f
'{msg}. Restart yt-dlp to use the updated version', expected
=True)
548 return self
.ydl
._download
_retcode
549 self
.restart
= wrapper
551 def _report_error(self
, msg
, expected
=False):
552 self
.ydl
.report_error(msg
, tb
=False if expected
else None)
553 self
.ydl
._download
_retcode
= 100
555 def _report_permission_error(self
, file):
556 self
._report
_error
(f
'Unable to write to {file}; try running as administrator', True)
558 def _report_network_error(self
, action
, delim
=';', tag
=None):
560 tag
= self
.requested_tag
561 path
= tag
if tag
== 'latest' else f
'tag/{tag}'
563 f
'Unable to {action}{delim} visit '
564 f
'https://github.com/{self.requested_repo}/releases/{path}', True)
566 # XXX: Everything below this line in this class is deprecated / for compat only
568 def _target_tag(self
):
569 """Deprecated; requested tag with 'tags/' prepended when necessary for API calls"""
570 return f
'tags/{self.requested_tag}' if self
.requested_tag
!= 'latest' else self
.requested_tag
572 def _check_update(self
):
573 """Deprecated; report whether there is an update available"""
574 return bool(self
.query_update(_output
=True))
576 def __getattr__(self
, attribute
: str):
577 """Compat getter function for deprecated attributes"""
578 deprecated_props_map
= {
579 'check_update': '_check_update',
580 'target_tag': '_target_tag',
581 'target_channel': 'requested_channel',
583 update_info_props_map
= {
584 'has_update': '_has_update',
585 'new_version': 'version',
586 'latest_version': 'requested_version',
587 'release_name': 'binary_name',
588 'release_hash': 'checksum',
591 if attribute
not in deprecated_props_map
and attribute
not in update_info_props_map
:
592 raise AttributeError(f
'{type(self).__name__!r} object has no attribute {attribute!r}')
594 msg
= f
'{type(self).__name__}.{attribute} is deprecated and will be removed in a future version'
595 if attribute
in deprecated_props_map
:
596 source_name
= deprecated_props_map
[attribute
]
597 if not source_name
.startswith('_'):
598 msg
+= f
'. Please use {source_name!r} instead'
600 mapping
= deprecated_props_map
602 else: # attribute in update_info_props_map
603 msg
+= '. Please call query_update() instead'
604 source
= self
.query_update()
606 source
= UpdateInfo('', None, None, None)
607 source
._has
_update
= False
608 mapping
= update_info_props_map
610 deprecation_warning(msg
)
611 for target_name
, source_name
in mapping
.items():
612 value
= getattr(source
, source_name
)
613 setattr(self
, target_name
, value
)
615 return getattr(self
, attribute
)
619 """Update the program file with the latest version from the repository
620 @returns Whether there was a successful update (No update = False)
622 return Updater(ydl
).update()
625 __all__
= ['Updater']