yt_dlp/update.py

   1 import hashlib
   2 import json
   3 import os
   4 import platform
   5 import subprocess
   6 import sys
   7 import traceback
   8 from zipimport import zipimporter
   9
  10 from .compat import compat_realpath
  11 from .utils import Popen, encode_compat_str, write_string
  12 from .version import __version__
  13
  14
  15 def detect_variant():
  16     if hasattr(sys, 'frozen'):
  17         prefix = 'mac' if sys.platform == 'darwin' else 'win'
  18         if getattr(sys, '_MEIPASS', None):
  19             if sys._MEIPASS == os.path.dirname(sys.executable):
  20                 return f'{prefix}_dir'
  21             return f'{prefix}_exe'
  22         return 'py2exe'
  23     elif isinstance(__loader__, zipimporter):
  24         return 'zip'
  25     elif os.path.basename(sys.argv[0]) == '__main__.py':
  26         return 'source'
  27     return 'unknown'
  28
  29
  30 _NON_UPDATEABLE_REASONS = {
  31     'win_exe': None,
  32     'zip': None,
  33     'mac_exe': None,
  34     'py2exe': None,
  35     'win_dir': 'Auto-update is not supported for unpackaged windows executable; Re-download the latest release',
  36     'mac_dir': 'Auto-update is not supported for unpackaged MacOS executable; Re-download the latest release',
  37     'source': 'You cannot update when running from source code; Use git to pull the latest changes',
  38     'unknown': 'It looks like you installed yt-dlp with a package manager, pip or setup.py; Use that to update',
  39 }
  40
  41
  42 def is_non_updateable():
  43     return _NON_UPDATEABLE_REASONS.get(detect_variant(), _NON_UPDATEABLE_REASONS['unknown'])
  44
  45
  46 def run_update(ydl):
  47     """
  48     Update the program file with the latest version from the repository
  49     Returns whether the program should terminate
  50     """
  51
  52     JSON_URL = 'https://api.github.com/repos/yt-dlp/yt-dlp/releases/latest'
  53
  54     def report_error(msg, expected=False):
  55         ydl.report_error(msg, tb='' if expected else None)
  56
  57     def report_unable(action, expected=False):
  58         report_error(f'Unable to {action}', expected)
  59
  60     def report_permission_error(file):
  61         report_unable(f'write to {file}; Try running as administrator', True)
  62
  63     def report_network_error(action, delim=';'):
  64         report_unable(f'{action}{delim} Visit  https://github.com/yt-dlp/yt-dlp/releases/latest', True)
  65
  66     def calc_sha256sum(path):
  67         h = hashlib.sha256()
  68         b = bytearray(128 * 1024)
  69         mv = memoryview(b)
  70         with open(os.path.realpath(path), 'rb', buffering=0) as f:
  71             for n in iter(lambda: f.readinto(mv), 0):
  72                 h.update(mv[:n])
  73         return h.hexdigest()
  74
  75     # Download and check versions info
  76     try:
  77         version_info = ydl._opener.open(JSON_URL).read().decode()
  78         version_info = json.loads(version_info)
  79     except Exception:
  80         return report_network_error('obtain version info', delim='; Please try again later or')
  81
  82     def version_tuple(version_str):
  83         return tuple(map(int, version_str.split('.')))
  84
  85     version_id = version_info['tag_name']
  86     ydl.to_screen(f'Latest version: {version_id}, Current version: {__version__}')
  87     if version_tuple(__version__) >= version_tuple(version_id):
  88         ydl.to_screen(f'yt-dlp is up to date ({__version__})')
  89         return
  90
  91     err = is_non_updateable()
  92     if err:
  93         return report_error(err, True)
  94
  95     # sys.executable is set to the full pathname of the exe-file for py2exe
  96     # though symlinks are not followed so that we need to do this manually
  97     # with help of realpath
  98     filename = compat_realpath(sys.executable if hasattr(sys, 'frozen') else sys.argv[0])
  99     ydl.to_screen(f'Current Build Hash {calc_sha256sum(filename)}')
 100     ydl.to_screen(f'Updating to version {version_id} ...')
 101
 102     version_labels = {
 103         'zip_3': '',
 104         'win_exe_64': '.exe',
 105         'py2exe_64': '_min.exe',
 106         'win_exe_32': '_x86.exe',
 107         'mac_exe_64': '_macos',
 108     }
 109
 110     def get_bin_info(bin_or_exe, version):
 111         label = version_labels[f'{bin_or_exe}_{version}']
 112         return next((i for i in version_info['assets'] if i['name'] == 'yt-dlp%s' % label), {})
 113
 114     def get_sha256sum(bin_or_exe, version):
 115         filename = 'yt-dlp%s' % version_labels[f'{bin_or_exe}_{version}']
 116         urlh = next(
 117             (i for i in version_info['assets'] if i['name'] in ('SHA2-256SUMS')),
 118             {}).get('browser_download_url')
 119         if not urlh:
 120             return None
 121         hash_data = ydl._opener.open(urlh).read().decode()
 122         return dict(ln.split()[::-1] for ln in hash_data.splitlines()).get(filename)
 123
 124     if not os.access(filename, os.W_OK):
 125         return report_permission_error(filename)
 126
 127     # PyInstaller
 128     variant = detect_variant()
 129     if variant in ('win_exe', 'py2exe'):
 130         directory = os.path.dirname(filename)
 131         if not os.access(directory, os.W_OK):
 132             return report_permission_error(directory)
 133         try:
 134             if os.path.exists(filename + '.old'):
 135                 os.remove(filename + '.old')
 136         except OSError:
 137             return report_unable('remove the old version')
 138
 139         try:
 140             arch = platform.architecture()[0][:2]
 141             url = get_bin_info(variant, arch).get('browser_download_url')
 142             if not url:
 143                 return report_network_error('fetch updates')
 144             urlh = ydl._opener.open(url)
 145             newcontent = urlh.read()
 146             urlh.close()
 147         except OSError:
 148             return report_network_error('download latest version')
 149
 150         try:
 151             with open(filename + '.new', 'wb') as outf:
 152                 outf.write(newcontent)
 153         except OSError:
 154             return report_permission_error(f'{filename}.new')
 155
 156         expected_sum = get_sha256sum(variant, arch)
 157         if not expected_sum:
 158             ydl.report_warning('no hash information found for the release')
 159         elif calc_sha256sum(filename + '.new') != expected_sum:
 160             report_network_error('verify the new executable')
 161             try:
 162                 os.remove(filename + '.new')
 163             except OSError:
 164                 return report_unable('remove corrupt download')
 165
 166         try:
 167             os.rename(filename, filename + '.old')
 168         except OSError:
 169             return report_unable('move current version')
 170         try:
 171             os.rename(filename + '.new', filename)
 172         except OSError:
 173             report_unable('overwrite current version')
 174             os.rename(filename + '.old', filename)
 175             return
 176         try:
 177             # Continues to run in the background
 178             Popen(
 179                 'ping 127.0.0.1 -n 5 -w 1000 & del /F "%s.old"' % filename,
 180                 shell=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
 181             ydl.to_screen('Updated yt-dlp to version %s' % version_id)
 182             return True  # Exit app
 183         except OSError:
 184             report_unable('delete the old version')
 185
 186     elif variant in ('zip', 'mac_exe'):
 187         pack_type = '3' if variant == 'zip' else '64'
 188         try:
 189             url = get_bin_info(variant, pack_type).get('browser_download_url')
 190             if not url:
 191                 return report_network_error('fetch updates')
 192             urlh = ydl._opener.open(url)
 193             newcontent = urlh.read()
 194             urlh.close()
 195         except OSError:
 196             return report_network_error('download the latest version')
 197
 198         expected_sum = get_sha256sum(variant, pack_type)
 199         if not expected_sum:
 200             ydl.report_warning('no hash information found for the release')
 201         elif hashlib.sha256(newcontent).hexdigest() != expected_sum:
 202             return report_network_error('verify the new package')
 203
 204         try:
 205             with open(filename, 'wb') as outf:
 206                 outf.write(newcontent)
 207         except OSError:
 208             return report_unable('overwrite current version')
 209
 210         ydl.to_screen('Updated yt-dlp to version %s; Restart yt-dlp to use the new version' % version_id)
 211         return
 212
 213     assert False, f'Unhandled variant: {variant}'
 214
 215
 216 # Deprecated
 217 def update_self(to_screen, verbose, opener):
 218
 219     printfn = to_screen
 220
 221     write_string(
 222         'DeprecationWarning: "yt_dlp.update.update_self" is deprecated and may be removed in a future version. '
 223         'Use "yt_dlp.update.run_update(ydl)" instead\n')
 224
 225     class FakeYDL():
 226         _opener = opener
 227         to_screen = printfn
 228
 229         @staticmethod
 230         def report_warning(msg, *args, **kwargs):
 231             return printfn('WARNING: %s' % msg, *args, **kwargs)
 232
 233         @staticmethod
 234         def report_error(msg, tb=None):
 235             printfn('ERROR: %s' % msg)
 236             if not verbose:
 237                 return
 238             if tb is None:
 239                 # Copied from YoutubeDl.trouble
 240                 if sys.exc_info()[0]:
 241                     tb = ''
 242                     if hasattr(sys.exc_info()[1], 'exc_info') and sys.exc_info()[1].exc_info[0]:
 243                         tb += ''.join(traceback.format_exception(*sys.exc_info()[1].exc_info))
 244                     tb += encode_compat_str(traceback.format_exc())
 245                 else:
 246                     tb_data = traceback.format_list(traceback.extract_stack())
 247                     tb = ''.join(tb_data)
 248             if tb:
 249                 printfn(tb)
 250
 251     return run_update(FakeYDL())