yt_dlp/update.py

   1 import atexit
   2 import contextlib
   3 import hashlib
   4 import json
   5 import os
   6 import platform
   7 import re
   8 import subprocess
   9 import sys
  10 from zipimport import zipimporter
  11
  12 from .compat import functools  # isort: split
  13 from .compat import compat_realpath, compat_shlex_quote
  14 from .utils import (
  15     Popen,
  16     cached_method,
  17     deprecation_warning,
  18     remove_end,
  19     shell_quote,
  20     system_identifier,
  21     traverse_obj,
  22     version_tuple,
  23 )
  24 from .version import UPDATE_HINT, VARIANT, __version__
  25
  26 REPOSITORY = 'yt-dlp/yt-dlp'
  27 API_URL = f'https://api.github.com/repos/{REPOSITORY}/releases'
  28
  29
  30 @functools.cache
  31 def _get_variant_and_executable_path():
  32     """@returns (variant, executable_path)"""
  33     if getattr(sys, 'frozen', False):
  34         path = sys.executable
  35         if not hasattr(sys, '_MEIPASS'):
  36             return 'py2exe', path
  37         elif sys._MEIPASS == os.path.dirname(path):
  38             return f'{sys.platform}_dir', path
  39         elif sys.platform == 'darwin':
  40             machine = '_legacy' if version_tuple(platform.mac_ver()[0]) < (10, 15) else ''
  41         else:
  42             machine = f'_{platform.machine().lower()}'
  43             # Ref: https://en.wikipedia.org/wiki/Uname#Examples
  44             if machine[1:] in ('x86', 'x86_64', 'amd64', 'i386', 'i686'):
  45                 machine = '_x86' if platform.architecture()[0][:2] == '32' else ''
  46         return f'{remove_end(sys.platform, "32")}{machine}_exe', path
  47
  48     path = os.path.dirname(__file__)
  49     if isinstance(__loader__, zipimporter):
  50         return 'zip', os.path.join(path, '..')
  51     elif (os.path.basename(sys.argv[0]) in ('__main__.py', '-m')
  52           and os.path.exists(os.path.join(path, '../.git/HEAD'))):
  53         return 'source', path
  54     return 'unknown', path
  55
  56
  57 def detect_variant():
  58     return VARIANT or _get_variant_and_executable_path()[0]
  59
  60
  61 @functools.cache
  62 def current_git_head():
  63     if detect_variant() != 'source':
  64         return
  65     with contextlib.suppress(Exception):
  66         stdout, _, _ = Popen.run(
  67             ['git', 'rev-parse', '--short', 'HEAD'],
  68             text=True, cwd=os.path.dirname(os.path.abspath(__file__)),
  69             stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  70         if re.fullmatch('[0-9a-f]+', stdout.strip()):
  71             return stdout.strip()
  72
  73
  74 _FILE_SUFFIXES = {
  75     'zip': '',
  76     'py2exe': '_min.exe',
  77     'win_exe': '.exe',
  78     'win_x86_exe': '_x86.exe',
  79     'darwin_exe': '_macos',
  80     'darwin_legacy_exe': '_macos_legacy',
  81     'linux_exe': '_linux',
  82     'linux_aarch64_exe': '_linux_aarch64',
  83     'linux_armv7l_exe': '_linux_armv7l',
  84 }
  85
  86 _NON_UPDATEABLE_REASONS = {
  87     **{variant: None for variant in _FILE_SUFFIXES},  # Updatable
  88     **{variant: f'Auto-update is not supported for unpackaged {name} executable; Re-download the latest release'
  89        for variant, name in {'win32_dir': 'Windows', 'darwin_dir': 'MacOS', 'linux_dir': 'Linux'}.items()},
  90     'source': 'You cannot update when running from source code; Use git to pull the latest changes',
  91     'unknown': 'You installed yt-dlp with a package manager or setup.py; Use that to update',
  92     'other': 'You are using an unofficial build of yt-dlp; Build the executable again',
  93 }
  94
  95
  96 def is_non_updateable():
  97     if UPDATE_HINT:
  98         return UPDATE_HINT
  99     return _NON_UPDATEABLE_REASONS.get(
 100         detect_variant(), _NON_UPDATEABLE_REASONS['unknown' if VARIANT else 'other'])
 101
 102
 103 def _sha256_file(path):
 104     h = hashlib.sha256()
 105     mv = memoryview(bytearray(128 * 1024))
 106     with open(os.path.realpath(path), 'rb', buffering=0) as f:
 107         for n in iter(lambda: f.readinto(mv), 0):
 108             h.update(mv[:n])
 109     return h.hexdigest()
 110
 111
 112 class Updater:
 113     def __init__(self, ydl):
 114         self.ydl = ydl
 115
 116     @functools.cached_property
 117     def _tag(self):
 118         if version_tuple(__version__) >= version_tuple(self.latest_version):
 119             return 'latest'
 120
 121         identifier = f'{detect_variant()} {system_identifier()}'
 122         for line in self._download('_update_spec', 'latest').decode().splitlines():
 123             if not line.startswith('lock '):
 124                 continue
 125             _, tag, pattern = line.split(' ', 2)
 126             if re.match(pattern, identifier):
 127                 return f'tags/{tag}'
 128         return 'latest'
 129
 130     @cached_method
 131     def _get_version_info(self, tag):
 132         self.ydl.write_debug(f'Fetching release info: {API_URL}/{tag}')
 133         return json.loads(self.ydl.urlopen(f'{API_URL}/{tag}').read().decode())
 134
 135     @property
 136     def current_version(self):
 137         """Current version"""
 138         return __version__
 139
 140     @property
 141     def new_version(self):
 142         """Version of the latest release we can update to"""
 143         if self._tag.startswith('tags/'):
 144             return self._tag[5:]
 145         return self._get_version_info(self._tag)['tag_name']
 146
 147     @property
 148     def latest_version(self):
 149         """Version of the latest release"""
 150         return self._get_version_info('latest')['tag_name']
 151
 152     @property
 153     def has_update(self):
 154         """Whether there is an update available"""
 155         return version_tuple(__version__) < version_tuple(self.new_version)
 156
 157     @functools.cached_property
 158     def filename(self):
 159         """Filename of the executable"""
 160         return compat_realpath(_get_variant_and_executable_path()[1])
 161
 162     def _download(self, name, tag):
 163         url = traverse_obj(self._get_version_info(tag), (
 164             'assets', lambda _, v: v['name'] == name, 'browser_download_url'), get_all=False)
 165         if not url:
 166             raise Exception('Unable to find download URL')
 167         self.ydl.write_debug(f'Downloading {name} from {url}')
 168         return self.ydl.urlopen(url).read()
 169
 170     @functools.cached_property
 171     def release_name(self):
 172         """The release filename"""
 173         return f'yt-dlp{_FILE_SUFFIXES[detect_variant()]}'
 174
 175     @functools.cached_property
 176     def release_hash(self):
 177         """Hash of the latest release"""
 178         hash_data = dict(ln.split()[::-1] for ln in self._download('SHA2-256SUMS', self._tag).decode().splitlines())
 179         return hash_data[self.release_name]
 180
 181     def _report_error(self, msg, expected=False):
 182         self.ydl.report_error(msg, tb=False if expected else None)
 183         self.ydl._download_retcode = 100
 184
 185     def _report_permission_error(self, file):
 186         self._report_error(f'Unable to write to {file}; Try running as administrator', True)
 187
 188     def _report_network_error(self, action, delim=';'):
 189         self._report_error(f'Unable to {action}{delim} Visit  https://github.com/{REPOSITORY}/releases/latest', True)
 190
 191     def check_update(self):
 192         """Report whether there is an update available"""
 193         try:
 194             self.ydl.to_screen(
 195                 f'Latest version: {self.latest_version}, Current version: {self.current_version}')
 196             if not self.has_update:
 197                 if self._tag == 'latest':
 198                     return self.ydl.to_screen(f'yt-dlp is up to date ({__version__})')
 199                 return self.ydl.report_warning(
 200                     'yt-dlp cannot be updated any further since you are on an older Python version')
 201         except Exception:
 202             return self._report_network_error('obtain version info', delim='; Please try again later or')
 203
 204         if not is_non_updateable():
 205             self.ydl.to_screen(f'Current Build Hash {_sha256_file(self.filename)}')
 206         return True
 207
 208     def update(self):
 209         """Update yt-dlp executable to the latest version"""
 210         if not self.check_update():
 211             return
 212         err = is_non_updateable()
 213         if err:
 214             return self._report_error(err, True)
 215         self.ydl.to_screen(f'Updating to version {self.new_version} ...')
 216
 217         directory = os.path.dirname(self.filename)
 218         if not os.access(self.filename, os.W_OK):
 219             return self._report_permission_error(self.filename)
 220         elif not os.access(directory, os.W_OK):
 221             return self._report_permission_error(directory)
 222
 223         new_filename, old_filename = f'{self.filename}.new', f'{self.filename}.old'
 224         if detect_variant() == 'zip':  # Can be replaced in-place
 225             new_filename, old_filename = self.filename, None
 226
 227         try:
 228             if os.path.exists(old_filename or ''):
 229                 os.remove(old_filename)
 230         except OSError:
 231             return self._report_error('Unable to remove the old version')
 232
 233         try:
 234             newcontent = self._download(self.release_name, self._tag)
 235         except OSError:
 236             return self._report_network_error('download latest version')
 237         except Exception:
 238             return self._report_network_error('fetch updates')
 239
 240         try:
 241             expected_hash = self.release_hash
 242         except Exception:
 243             self.ydl.report_warning('no hash information found for the release')
 244         else:
 245             if hashlib.sha256(newcontent).hexdigest() != expected_hash:
 246                 return self._report_network_error('verify the new executable')
 247
 248         try:
 249             with open(new_filename, 'wb') as outf:
 250                 outf.write(newcontent)
 251         except OSError:
 252             return self._report_permission_error(new_filename)
 253
 254         if old_filename:
 255             mask = os.stat(self.filename).st_mode
 256             try:
 257                 os.rename(self.filename, old_filename)
 258             except OSError:
 259                 return self._report_error('Unable to move current version')
 260
 261             try:
 262                 os.rename(new_filename, self.filename)
 263             except OSError:
 264                 self._report_error('Unable to overwrite current version')
 265                 return os.rename(old_filename, self.filename)
 266
 267         if detect_variant() in ('win32_exe', 'py2exe'):
 268             atexit.register(Popen, f'ping 127.0.0.1 -n 5 -w 1000 & del /F "{old_filename}"',
 269                             shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 270         elif old_filename:
 271             try:
 272                 os.remove(old_filename)
 273             except OSError:
 274                 self._report_error('Unable to remove the old version')
 275
 276             try:
 277                 os.chmod(self.filename, mask)
 278             except OSError:
 279                 return self._report_error(
 280                     f'Unable to set permissions. Run: sudo chmod a+rx {compat_shlex_quote(self.filename)}')
 281
 282         self.ydl.to_screen(f'Updated yt-dlp to version {self.new_version}')
 283         return True
 284
 285     @functools.cached_property
 286     def cmd(self):
 287         """The command-line to run the executable, if known"""
 288         # There is no sys.orig_argv in py < 3.10. Also, it can be [] when frozen
 289         if getattr(sys, 'orig_argv', None):
 290             return sys.orig_argv
 291         elif getattr(sys, 'frozen', False):
 292             return sys.argv
 293
 294     def restart(self):
 295         """Restart the executable"""
 296         assert self.cmd, 'Must be frozen or Py >= 3.10'
 297         self.ydl.write_debug(f'Restarting: {shell_quote(self.cmd)}')
 298         _, _, returncode = Popen.run(self.cmd)
 299         return returncode
 300
 301
 302 def run_update(ydl):
 303     """Update the program file with the latest version from the repository
 304     @returns    Whether there was a successful update (No update = False)
 305     """
 306     return Updater(ydl).update()
 307
 308
 309 # Deprecated
 310 def update_self(to_screen, verbose, opener):
 311     import traceback
 312
 313     deprecation_warning(f'"{__name__}.update_self" is deprecated and may be removed '
 314                         f'in a future version. Use "{__name__}.run_update(ydl)" instead')
 315
 316     printfn = to_screen
 317
 318     class FakeYDL():
 319         to_screen = printfn
 320
 321         def report_warning(self, msg, *args, **kwargs):
 322             return printfn(f'WARNING: {msg}', *args, **kwargs)
 323
 324         def report_error(self, msg, tb=None):
 325             printfn(f'ERROR: {msg}')
 326             if not verbose:
 327                 return
 328             if tb is None:
 329                 # Copied from YoutubeDL.trouble
 330                 if sys.exc_info()[0]:
 331                     tb = ''
 332                     if hasattr(sys.exc_info()[1], 'exc_info') and sys.exc_info()[1].exc_info[0]:
 333                         tb += ''.join(traceback.format_exception(*sys.exc_info()[1].exc_info))
 334                     tb += traceback.format_exc()
 335                 else:
 336                     tb_data = traceback.format_list(traceback.extract_stack())
 337                     tb = ''.join(tb_data)
 338             if tb:
 339                 printfn(tb)
 340
 341         def write_debug(self, msg, *args, **kwargs):
 342             printfn(f'[debug] {msg}', *args, **kwargs)
 343
 344         def urlopen(self, url):
 345             return opener.open(url)
 346
 347     return run_update(FakeYDL())