tabby-web/backend/tabby/app/management/commands/add_version.py

import fsspec
import logging
import requests
import shutil
import subprocess
import tempfile
from django.core.management.base import BaseCommand
from django.conf import settings
from pathlib import Path
from urllib.parse import urlparse


class Command(BaseCommand):
    help = 'Downloads a new app version'

    def add_arguments(self, parser):
        parser.add_argument('version', type=str)

    def handle(self, *args, **options):
        version = options['version']
        target = f'{settings.APP_DIST_STORAGE}/{version}'

        fs = fsspec.filesystem(urlparse(settings.APP_DIST_STORAGE).scheme)

        plugin_list = [
            'tabby-web-container',
            'tabby-core',
            'tabby-settings',
            'tabby-terminal',
            'tabby-ssh',
            'tabby-community-color-schemes',
            'tabby-serial',
            'tabby-telnet',
            'tabby-web',
            'tabby-web-demo',
        ]

        with tempfile.TemporaryDirectory() as tempdir:
            tempdir = Path(tempdir)
            for plugin in plugin_list:
                logging.info(f'Resolving {plugin}@{version}')
                response = requests.get(f'{settings.NPM_REGISTRY}/{plugin}/{version}')
                response.raise_for_status()
                info = response.json()
                url = info['dist']['tarball']

                logging.info(f'Downloading {plugin}@{version} from {url}')
                response = requests.get(url)

                with tempfile.NamedTemporaryFile('wb') as f:
                    f.write(response.content)
                    plugin_final_target = Path(tempdir) / plugin

                    with tempfile.TemporaryDirectory() as extraction_tmp:
                        subprocess.check_call(
                            ['tar', '-xzf', f.name, '-C', str(extraction_tmp)]
                        )
                        shutil.move(
                            Path(extraction_tmp) / 'package', plugin_final_target
                        )

            if fs.exists(target):
                fs.rm(target, recursive=True)
            fs.mkdir(target)
            fs.put(str(tempdir), target, recursive=True)