streaming_installer

streaming_installer

Script di installazione del software Streaming Santa Croce

VenvPaths

Bases: NamedTuple

Raccoglie tutti i percorsi relativi al virtual environment.

Source code in streaming_installer.py

class VenvPaths(NamedTuple):
    """Raccoglie tutti i percorsi relativi al virtual environment."""
    base_dir: str
    venv_dir: str
    scripts_dir: str
    packages_dir: str
    python_path: str
    pep8_path: str
    pyinstaller_path: str
    pylint_path: str
    mkdocs_path: str

    @staticmethod
    def from_base(base_dir: str) -> 'VenvPaths':
        """
        Costruisce i percorsi del venv a partire dalla directory base.
        :param base_dir: Directory root del progetto.
        :return: Istanza di VenvPaths.
        """
        path_venv = join(base_dir, 'venv')
        path_dot_venv = join(base_dir, '.venv')
        venv_dir = path_dot_venv if isdir(path_dot_venv) else path_venv

        scripts_dir = join(venv_dir, 'Scripts')
        packages_dir = join(venv_dir, 'Lib', 'site-packages')
        return VenvPaths(
            base_dir=base_dir,
            venv_dir=venv_dir,
            scripts_dir=scripts_dir,
            packages_dir=packages_dir,
            python_path=join(scripts_dir, 'python'),
            pep8_path=join(scripts_dir, 'autopep8'),
            pyinstaller_path=join(scripts_dir, 'pyinstaller'),
            pylint_path=join(scripts_dir, 'pylint'),
            mkdocs_path=join(scripts_dir, 'mkdocs'),
        )

from_base(base_dir) staticmethod

Costruisce i percorsi del venv a partire dalla directory base. :param base_dir: Directory root del progetto. :return: Istanza di VenvPaths.

Source code in streaming_installer.py

@staticmethod
def from_base(base_dir: str) -> 'VenvPaths':
    """
    Costruisce i percorsi del venv a partire dalla directory base.
    :param base_dir: Directory root del progetto.
    :return: Istanza di VenvPaths.
    """
    path_venv = join(base_dir, 'venv')
    path_dot_venv = join(base_dir, '.venv')
    venv_dir = path_dot_venv if isdir(path_dot_venv) else path_venv

    scripts_dir = join(venv_dir, 'Scripts')
    packages_dir = join(venv_dir, 'Lib', 'site-packages')
    return VenvPaths(
        base_dir=base_dir,
        venv_dir=venv_dir,
        scripts_dir=scripts_dir,
        packages_dir=packages_dir,
        python_path=join(scripts_dir, 'python'),
        pep8_path=join(scripts_dir, 'autopep8'),
        pyinstaller_path=join(scripts_dir, 'pyinstaller'),
        pylint_path=join(scripts_dir, 'pylint'),
        mkdocs_path=join(scripts_dir, 'mkdocs'),
    )

ask_yes(prompt, default_yes=False)

Chiede conferma all'utente e restituisce True se la risposta è affermativa. :param prompt: Testo della domanda da mostrare. :param default_yes: Se True, INVIO senza testo conta come 'sì'. :return: True se l'utente ha risposto affermativamente.

Source code in streaming_installer.py

def ask_yes(prompt: str, default_yes: bool = False) -> bool:
    """
    Chiede conferma all'utente e restituisce True se la risposta è affermativa.
    :param prompt: Testo della domanda da mostrare.
    :param default_yes: Se True, INVIO senza testo conta come 'sì'.
    :return: True se l'utente ha risposto affermativamente.
    """
    answer = input(prompt).strip()
    if default_yes:
        return answer in YES_ANSWERS or answer == ''
    return answer in YES_ANSWERS

build_docs(paths)

Genera la documentazione HTML con mkdocs build usando la configurazione in mkdocs.yml. L'output viene scritto nella cartella site/. :param paths: Istanza VenvPaths.

Source code in streaming_installer.py

def build_docs(paths: VenvPaths) -> None:
    """
    Genera la documentazione HTML con mkdocs build usando la configurazione
    in ``mkdocs.yml``. L'output viene scritto nella cartella ``site/``.
    :param paths: Istanza VenvPaths.
    """
    check_and_enter_venv(paths)
    mkdocs_yml = join(paths.base_dir, 'mkdocs.yml')
    if not exists(mkdocs_yml):
        sys.exit("mkdocs.yml non trovato. Impossibile generare la documentazione.")

    log.info("Generazione documentazione con mkdocs...")
    sp.check_call(
        [paths.mkdocs_path, 'build', '--config-file', mkdocs_yml, '--clean'],
        env=environ.copy(),
    )
    log.info("Documentazione generata in: %s", join(paths.base_dir, 'site'))

build_exe(paths)

Compila l'eseguibile con PyInstaller e copia i file JSON nella cartella dist. :param paths: Istanza VenvPaths.

Source code in streaming_installer.py

def build_exe(paths: VenvPaths) -> None:
    """
    Compila l'eseguibile con PyInstaller e copia i file JSON nella cartella
    dist.
    :param paths: Istanza VenvPaths.
    """
    check_and_enter_venv(paths)
    log.info("Compilazione dell'eseguibile con PyInstaller...")
    hidden_imports: list[str] = []
    params = [
        paths.pyinstaller_path, '--noconfirm',
        '--add-data', './images;images',
        '--collect-data', 'pytablericons',
        '--onefile', '--windowed',
        '--icon', './images/yt.ico',
        '--name', EXE_NAME,
        '--runtime-tmpdir', '.',
        '--splash', './images/splash.jpg',
        'main.py',
    ]
    for hi in hidden_imports:
        params += ['--hidden-import', hi]

    sp.check_call(params, env=environ.copy())

    log.info("Copia file JSON in ./dist ...")
    for file in glob('*.json', root_dir=paths.base_dir):
        copy(file, './dist')

check_and_enter_venv(paths)

Verifica che il venv esista ed entra se non siamo già dentro. Termina lo script se il venv non esiste. :param paths: Istanza VenvPaths con i percorsi del venv.

Source code in streaming_installer.py

def check_and_enter_venv(paths: VenvPaths) -> None:
    """
    Verifica che il venv esista ed entra se non siamo già dentro.
    Termina lo script se il venv non esiste.
    :param paths: Istanza VenvPaths con i percorsi del venv.
    """
    if not in_virtualenv():
        if not exists(paths.venv_dir):
            sys.exit("Esegui lo script nel Virtual Environment")
        enter_venv(paths)

clean_build()

Rimuove i file e le directory temporanei generati dalla build (spec, log, pycache, build, dist).

Source code in streaming_installer.py

def clean_build() -> None:  # pylint: disable=unused-argument
    """
    Rimuove i file e le directory temporanei generati dalla build
    (spec, log, __pycache__, build, dist).
    """
    log.info("Pulizia dei file temporanei...")
    files_to_remove = [f'./{EXE_NAME}.spec', './streaming.log']
    for file in (f for f in files_to_remove if exists(f)):
        remove(file)
        log.debug("Rimosso: %s", file)

    dirs_to_remove = ['./__pycache__', './build', './dist', './site', './CTkScrollableDropdown/__pycache__']
    for path in (p for p in dirs_to_remove if exists(p)):
        rmtree(path, onerror=force_remove)
        log.debug("Rimossa directory: %s", path)

delete_sources(paths)

Elimina tutti i file sorgente del progetto, mantenendo solo i file elencati in SOURCES_KEEP (.json, .exe, streaming_installer.py, .env). Chiede conferma prima di procedere. :param paths: Istanza VenvPaths.

Source code in streaming_installer.py

def delete_sources(paths: VenvPaths) -> None:
    """
    Elimina tutti i file sorgente del progetto, mantenendo solo i file
    elencati in SOURCES_KEEP (.json, .exe, streaming_installer.py, .env).
    Chiede conferma prima di procedere.
    :param paths: Istanza VenvPaths.
    """
    if not ask_yes("Cancello i sorgenti? [Y/n] ", default_yes=True):
        return

    log.info("Eliminazione dei sorgenti...")
    all_entries = glob('*', root_dir=paths.base_dir, include_hidden=True)
    to_delete = [
        f for f in all_entries
        if isdir(f) or (isfile(f) and not f.endswith(SOURCES_KEEP))
    ]
    for path in to_delete:
        if isfile(path):
            remove(path)
            log.debug("Rimosso file: %s", path)
        else:
            rmtree(path, ignore_errors=True)
            log.debug("Rimossa directory: %s", path)

deploy_exe()

Copia l'eseguibile dalla cartella dist alla directory base del progetto. Non fa nulla se l'eseguibile non è stato ancora compilato.

Source code in streaming_installer.py

def deploy_exe() -> None:
    """
    Copia l'eseguibile dalla cartella dist alla directory base del progetto.
    Non fa nulla se l'eseguibile non è stato ancora compilato.
    """
    exe_path = f'./dist/{EXE_NAME}.exe'
    if exists('./dist') and exists(exe_path):
        log.info("Deploy dell'eseguibile nella directory principale...")
        copy(exe_path, '.')
    else:
        log.warning("Eseguibile non trovato in ./dist, deploy saltato.")

download_from_github(paths, branch=GITHUB_DEFAULT_BRANCH)

scarica l'ultima versione del software da GitHub ed estrae lo zip nella directory base del progetto, proteggendo i file di configurazione esistenti. :param paths: Istanza VenvPaths (serve python_path e base_dir). :param branch: Branch GitHub da scaricare (default: master). :return: Tupla con i percorsi dei file estratti.

Source code in streaming_installer.py

def download_from_github(paths: VenvPaths, branch: str = GITHUB_DEFAULT_BRANCH) -> tuple[str, ...]:
    """
    scarica l'ultima versione del software da GitHub ed estrae lo zip
    nella directory base del progetto, proteggendo i file di configurazione
    esistenti.
    :param paths: Istanza VenvPaths (serve python_path e base_dir).
    :param branch: Branch GitHub da scaricare (default: master).
    :return: Tupla con i percorsi dei file estratti.
    """
    # pylint: disable=C0415
    try:
        import requests
        from dotenv import load_dotenv
    except ImportError:
        log.info("Installazione dipendenze per il download...")
        run_package_manager('requests', paths.python_path)
        run_package_manager('python-dotenv', paths.python_path)
        import requests
        from dotenv import load_dotenv

    # Carica il .env dalla directory dello script, non dalla CWD
    env_path = join(paths.base_dir, '.env')
    load_dotenv(env_path)
    token = os.getenv("GH_TOKEN")

    if token:
        log.info("Token GitHub trovato (lunghezza: %d caratteri).", len(token))
    else:
        log.warning(
            "GH_TOKEN non trovato in '%s'. "
            "Il download funziona solo se il repository è pubblico.",
            env_path,
        )

    headers = {
        'Accept': 'application/vnd.github+json',
        'X-GitHub-Api-Version': '2022-11-28',
    }
    if token:
        headers['Authorization'] = f'Bearer {token}'

    url = github_url(branch)
    log.info("Download da GitHub (branch: %s): %s", branch, url)
    resp = requests.get(url, stream=True, headers=headers, timeout=30)
    if not resp.ok:
        if resp.status_code == 404 and not token:
            sys.exit(
                "Errore 404: repository non trovato o privato. "
                "Aggiungi GH_TOKEN al file .env per accedere a repository privati."
            )
        if resp.status_code == 401:
            sys.exit(
                "Errore 401: token non autorizzato. "
                "Verifica che GH_TOKEN in .env sia valido e abbia i permessi 'Contents: Read'."
            )
        sys.exit(f"Errore nel download da GitHub: {resp.status_code} {resp.reason}")

    total = int(resp.headers.get('Content-Length', 0))

    # scrittura su file temporaneo con progress bar ASCII (senza dipendenze esterne)
    with tempfile.NamedTemporaryFile(suffix='.zip', delete=False) as tmp:
        tmp_path = tmp.name
        downloaded = 0
        bar_width = 40
        for chunk in resp.iter_content(chunk_size=8192):
            tmp.write(chunk)
            downloaded += len(chunk)
            if total:
                filled = int(bar_width * downloaded / total)
                bar = '█' * filled + '░' * (bar_width - filled)
                pct = downloaded / total * 100
                print(f"\r  [{bar}] {pct:5.1f}%  {downloaded // 1024} / {total // 1024} KB",
                      end='', flush=True)
        print()  # newline finale dopo la barra
        if not total:
            log.info("Download completato: %.1f KB", downloaded / 1024)

    try:
        files: list[str] = []
        with ZipFile(tmp_path) as zip_file:
            root_dir = zip_file.infolist()[0].filename
            members = zip_no_root(zip_file, root_dir, files)
            zip_file.extractall(path=paths.base_dir, members=members)
        log.info("Estratti %d file.", len(files))
        return tuple(files)
    finally:
        remove(tmp_path)

enter_venv(paths)

Attiva il virtual environment modificando PATH e sys.path. :param paths: Istanza VenvPaths con i percorsi del venv.

Source code in streaming_installer.py

def enter_venv(paths: VenvPaths) -> None:
    """
    Attiva il virtual environment modificando PATH e sys.path.
    :param paths: Istanza VenvPaths con i percorsi del venv.
    """
    environ['PATH'] = pathsep.join(
        [paths.scripts_dir] + environ.get('PATH', '').split(pathsep)
    )
    environ['VIRTUAL_ENV'] = paths.venv_dir
    prev_length = len(sys.path)
    addsitedir(paths.packages_dir)
    sys.path[:] = sys.path[prev_length:] + sys.path[:prev_length]
    sys.real_prefix = sys.prefix
    sys.prefix = paths.venv_dir

format_code(paths)

Riformatta i sorgenti Python con autopep8, escludendo la cartella venv. :param paths: Istanza VenvPaths.

Source code in streaming_installer.py

def format_code(paths: VenvPaths) -> None:
    """
    Riformatta i sorgenti Python con autopep8, escludendo la cartella venv.
    :param paths: Istanza VenvPaths.
    """
    check_and_enter_venv(paths)
    log.info("Formattazione del codice sorgente...")
    sp.check_call(
        [paths.pep8_path, paths.base_dir, '-r', '-i',
         '--max-line-length', '300', '--exclude', 'venv'],
        env=environ.copy(),
    )

get_base_prefix()

Restituisce il prefisso base dell'ambiente Python corrente. :return: Prefisso base.

Source code in streaming_installer.py

def get_base_prefix() -> str:
    """
    Restituisce il prefisso base dell'ambiente Python corrente.
    :return: Prefisso base.
    """
    return (getattr(sys, 'base_prefix', None) or
            getattr(sys, 'real_prefix', None) or sys.prefix)

github_url(branch=GITHUB_DEFAULT_BRANCH)

Restituisce l'URL API per scaricare lo zip del branch indicato. L'endpoint API funziona sia per repository pubblici che privati (con token). :param branch: Nome del branch da scaricare. :return: URL completo dello zip via API GitHub.

Source code in streaming_installer.py

def github_url(branch: str = GITHUB_DEFAULT_BRANCH) -> str:
    """
    Restituisce l'URL API per scaricare lo zip del branch indicato.
    L'endpoint API funziona sia per repository pubblici che privati
    (con token).
    :param branch: Nome del branch da scaricare.
    :return: URL completo dello zip via API GitHub.
    """
    return f"{GITHUB_API_REPO}/zipball/{branch}"

handle_exit()

Su Windows, se lo script è stato aperto con doppio clic (processo console solitario), attende che l'utente prema INVIO prima di chiudere.

Source code in streaming_installer.py

def handle_exit() -> None:
    """
    Su Windows, se lo script è stato aperto con doppio clic (processo
    console solitario), attende che l'utente prema INVIO prima di chiudere.
    """
    try:
        kernel32 = WinDLL('kernel32', use_last_error=True)
        process_array = (c_uint * 1)()
        num_processes = kernel32.GetConsoleProcessList(process_array, 1)
        if num_processes == 1:
            input('\nPremi ENTER per uscire...')
    except OSError:
        # Non siamo su Windows: nessuna azione necessaria
        pass

in_virtualenv()

Controlla se Python è in esecuzione dentro un virtualenv. :return: True se siamo dentro un virtualenv.

Source code in streaming_installer.py

def in_virtualenv() -> bool:
    """
    Controlla se Python è in esecuzione dentro un virtualenv.
    :return: True se siamo dentro un virtualenv.
    """
    try:
        return get_base_prefix() != sys.prefix
    except Exception:  # pylint: disable=broad-except
        return False

install_dependencies(paths)

Installa le dipendenze del progetto (requirements.txt) e gli strumenti di sviluppo (pyinstaller, mkdocs). Gestisce il blocco antivirus su pyinstaller chiedendo all'utente di aggiungere un'eccezione. :param paths: Istanza VenvPaths.

Source code in streaming_installer.py

def install_dependencies(paths: VenvPaths) -> None:
    """
    Installa le dipendenze del progetto (requirements.txt) e gli strumenti
    di sviluppo (pyinstaller, mkdocs). Gestisce il blocco antivirus su
    pyinstaller chiedendo all'utente di aggiungere un'eccezione.
    :param paths: Istanza VenvPaths.
    """
    log.info("Installazione dipendenze da requirements.txt...")
    run_package_manager(['-r', 'requirements.txt'], paths.python_path)

    log.info("Installazione strumenti di sviluppo...")
    dev_packages = [
        'pyinstaller',
        'mkdocs',
        'mkdocstrings',
        'mkdocstrings[python]',
        'mkdocs-material',
    ]
    try:
        run_package_manager(dev_packages, paths.python_path)
    except sp.CalledProcessError:
        input("Abilita un'eccezione sull'antivirus e poi premi invio")
        run_package_manager('pyinstaller', paths.python_path)

lint_code(paths)

Esegue pylint su tutti i file .py del progetto (esclusa la cartella venv) e logga il report completo. Termina con exit code 0 anche in presenza di warning: pylint usa exit code non-zero per qualsiasi osservazione, quindi l'errore viene loggato ma non fa fallire lo script. :param paths: Istanza VenvPaths.

Source code in streaming_installer.py

def lint_code(paths: VenvPaths) -> None:
    """
    Esegue pylint su tutti i file ``.py`` del progetto (esclusa la cartella
    venv) e logga il report completo. Termina con exit code 0 anche in
    presenza di warning: pylint usa exit code non-zero per qualsiasi
    osservazione, quindi l'errore viene loggato ma non fa fallire lo script.
    :param paths: Istanza VenvPaths.
    """
    check_and_enter_venv(paths)
    py_files = [
        f for f in glob('**/*.py', root_dir=paths.base_dir, recursive=True)
        if not f.startswith('venv')
    ]
    if not py_files:
        log.warning("Nessun file .py trovato da analizzare.")
        return

    log.info("Analisi statica con pylint (%d file)...", len(py_files))
    result = sp.run(
        [paths.pylint_path] + py_files,
        env=environ.copy(),
        capture_output=True,
        text=True,
        encoding='utf-8',
        errors='replace',
    )
    for line in result.stdout.splitlines():
        log.info("  %s", line)
    if result.returncode not in (0, 4):  # 4 = solo warning, non errori
        log.warning("pylint ha rilevato problemi (exit code: %d).", result.returncode)
    else:
        log.info("pylint completato senza errori critici.")

main(args)

Punto di ingresso del flusso di installazione/build. Ogni step corrisponde a una funzione dedicata, selezionata tramite i flag passati da riga di comando.

Flag azione

(nessuno) setup venv → download opzionale → install dipendenze --install setup completo + build exe + deploy + clean + delete sorgenti --exe compila l'eseguibile con PyInstaller --format riformatta i sorgenti con autopep8 --lint analisi statica con pylint --docs genera la documentazione HTML con mkdocs --clean rimuove file e directory temporanei di build

Flag modificatori (non attivano il setup da soli): --verbose mostra l'output di debug (incluso pip riga per riga) --dry-run mostra gli step senza eseguirli --branch branch GitHub da scaricare (default: master)

:param args: Argomenti parsati da ArgumentParser.

Source code in streaming_installer.py

def main(args: Namespace) -> None:
    """
    Punto di ingresso del flusso di installazione/build.
    Ogni step corrisponde a una funzione dedicata, selezionata tramite i
    flag passati da riga di comando.

    Flag azione:
      (nessuno)   setup venv → download opzionale → install dipendenze
      --install   setup completo + build exe + deploy + clean + delete sorgenti
      --exe       compila l'eseguibile con PyInstaller
      --format    riformatta i sorgenti con autopep8
      --lint      analisi statica con pylint
      --docs      genera la documentazione HTML con mkdocs
      --clean     rimuove file e directory temporanei di build

    Flag modificatori (non attivano il setup da soli):
      --verbose   mostra l'output di debug (incluso pip riga per riga)
      --dry-run   mostra gli step senza eseguirli
      --branch    branch GitHub da scaricare (default: master)

    :param args: Argomenti parsati da ArgumentParser.
    """
    validate_args(args)

    # --verbose: porta il logging a DEBUG, rendendo visibile l'output di pip
    if args.verbose:
        logging.getLogger().setLevel(logging.DEBUG)
        log.debug("Modalità verbose attiva.")

    base_dir = dirname(realpath(str(expanduser(__file__))))
    paths = VenvPaths.from_base(str(base_dir))
    branch = getattr(args, 'branch', None) or GITHUB_DEFAULT_BRANCH

    # --dry-run: wrapper che logga ogni step senza eseguirlo
    def step(label: str, fn, *fn_args, **fn_kwargs) -> None:
        if args.dry_run:
            log.info("[DRY-RUN] Saltato: %s", label)
        else:
            fn(*fn_args, **fn_kwargs)

    # Considera solo i flag "azione" per decidere se eseguire il setup iniziale
    action_values = {k: v for k, v in args.__dict__.items()
                     if k not in _MODIFIER_FLAGS}
    no_action_flags = not any(action_values.values())

    # Setup iniziale: nessun flag azione oppure --install completo
    if no_action_flags or args.install:
        step("setup venv", setup_venv, paths)

        if args.dry_run or ask_yes("Scarico l'ultima versione del software da GitHub? [y/N] "):
            step(f"download branch '{branch}' da GitHub",
                 download_from_github, paths, branch)

        if args.dry_run or ask_yes("Installo le librerie? [Y/n] ", default_yes=True):
            step("installazione dipendenze", install_dependencies, paths)

    if args.format:
        step("formattazione codice (autopep8)", format_code, paths)

    if args.lint:
        step("analisi statica (pylint)", lint_code, paths)

    if args.docs:
        step("generazione documentazione (mkdocs)", build_docs, paths)

    if args.exe or args.install:
        step("build eseguibile (PyInstaller)", build_exe, paths)

    if args.install:
        step("deploy eseguibile", deploy_exe)

    if args.clean or args.install:
        step("pulizia file temporanei", clean_build)

    if args.install:
        step("eliminazione sorgenti", delete_sources, paths)

    if args.dry_run:
        log.info("[DRY-RUN] Simulazione completata. Nessuna modifica eseguita.")

run_package_manager(package, python_path='python', tool='pip', extra_flags=None)

Esegue pip o pipwin per installare uno o più pacchetti, mostrando l'output in tempo reale riga per riga tramite il logger. :param package: Pacchetto o lista di pacchetti da installare. :param python_path: Percorso dell'interprete Python. :param tool: Stringa 'pip' oppure 'pipwin'. :param extra_flags: Flag aggiuntivi da passare al tool (es. ['--upgrade']). :return: Codice di ritorno del sottoprocesso. :raises ValueError: Se il tool non è 'pip' o 'pipwin'. :raises subprocess.CalledProcessError: Se il comando fallisce.

Source code in streaming_installer.py

def run_package_manager(package: str | list[str], python_path: str = 'python', tool: str = 'pip',
                        extra_flags: list[str] | None = None, ) -> int:
    """
    Esegue pip o pipwin per installare uno o più pacchetti, mostrando
    l'output in tempo reale riga per riga tramite il logger.
    :param package: Pacchetto o lista di pacchetti da installare.
    :param python_path: Percorso dell'interprete Python.
    :param tool: Stringa 'pip' oppure 'pipwin'.
    :param extra_flags: Flag aggiuntivi da passare al tool (es. ['--upgrade']).
    :return: Codice di ritorno del sottoprocesso.
    :raises ValueError: Se il tool non è 'pip' o 'pipwin'.
    :raises subprocess.CalledProcessError: Se il comando fallisce.
    """
    packages = [package] if isinstance(package, str) else list(package)
    flags = extra_flags if extra_flags is not None else []

    if tool == 'pip':
        params = [python_path, '-m', 'pip', 'install', '--upgrade'] + flags + packages
    elif tool == 'pipwin':
        params = [python_path, '-m', 'pipwin', 'install'] + flags + packages
    else:
        raise ValueError(f"Tool non supportato: {tool!r}. Usa 'pip' o 'pipwin'.")

    log.info("Esecuzione: %s", ' '.join(params))

    # Legge stdout e stderr in tempo reale e li instrada al logger
    with sp.Popen(
        params,
        env=environ.copy(),
        stdout=sp.PIPE,
        stderr=sp.STDOUT,
        text=True,
        encoding='utf-8',
        errors='replace',
    ) as proc:
        for line in proc.stdout:
            stripped = line.rstrip()
            if stripped:
                log.debug("  %s", stripped)
        proc.wait()

    if proc.returncode != 0:
        raise sp.CalledProcessError(proc.returncode, params)
    return proc.returncode

setup_venv(paths)

Crea o ricrea il virtual environment nella directory del progetto. Chiede conferma prima di sovrascrivere un venv esistente. :param paths: Istanza VenvPaths.

Source code in streaming_installer.py

def setup_venv(paths: VenvPaths) -> None:
    """
    Crea o ricrea il virtual environment nella directory del progetto.
    Chiede conferma prima di sovrascrivere un venv esistente.
    :param paths: Istanza VenvPaths.
    """
    if in_virtualenv():
        sys.exit("Esegui lo script fuori dal Virtual Environment")

    if not exists(paths.venv_dir) or ask_yes("Sovrascrivere l'attuale venv? [y/N] "):
        log.info("Creazione del virtual environment in: %s", paths.venv_dir)
        EnvBuilder(clear=True, with_pip=True, upgrade_deps=True).create(paths.venv_dir)

    enter_venv(paths)

validate_args(args)

Controlla combinazioni di flag incompatibili o ridondanti e termina lo script con un messaggio chiaro se ne trova. Emette warning per combinazioni legali ma potenzialmente inaspettate (es. --branch senza un'azione che scarica).

:param args: Argomenti parsati da ArgumentParser. :raises SystemExit: Se viene rilevata una combinazione non valida.

Source code in streaming_installer.py

def validate_args(args: Namespace) -> None:
    """
    Controlla combinazioni di flag incompatibili o ridondanti e termina lo
    script con un messaggio chiaro se ne trova. Emette warning per
    combinazioni legali ma potenzialmente inaspettate (es. ``--branch``
    senza un'azione che scarica).

    :param args: Argomenti parsati da ArgumentParser.
    :raises SystemExit: Se viene rilevata una combinazione non valida.
    """
    errors: list[str] = []
    for flag_a, flag_b, msg in _INCOMPATIBLE_FLAGS:
        if getattr(args, flag_a, False) and getattr(args, flag_b, False):
            errors.append(f"  ✗ --{flag_a} + --{flag_b}: {msg}")

    if errors:
        print("Errore: combinazione di flag non valida:")
        for err in errors:
            print(err)
        print("\nUsa --help per vedere i flag disponibili.")
        sys.exit(1)

    # Warning non bloccante: --branch senza un'azione che scarica
    branch = getattr(args, 'branch', None)
    if branch and branch != GITHUB_DEFAULT_BRANCH:
        action_flags = {k for k, v in args.__dict__.items()
                        if k not in _MODIFIER_FLAGS and v}
        if not action_flags & _DOWNLOAD_FLAGS:
            log.warning(
                "--branch '%s' specificato ma nessuna azione scarica da GitHub "
                "(serve --install oppure nessun flag per il setup iniziale).",
                branch,
            )

zip_no_root(zip_file, main_dir='', extracted_files=None, protected=CONFIG_FILES_PROTECTED)

Itera i file dentro lo zip saltando la cartella root e i file protetti già presenti su disco. :param zip_file: Oggetto ZipFile da iterare. :param main_dir: Nome della cartella root da rimuovere dai percorsi. :param extracted_files: Lista in cui accumulare i nomi dei file estratti. :param protected: Tuple di suffissi di file da non sovrascrivere se già esistenti su disco. :yield: oggetti ZipInfo con il percorso relativo corretto.

Source code in streaming_installer.py

def zip_no_root(zip_file: ZipFile, main_dir: str = '', extracted_files: list | None = None,
                protected: tuple[str, ...] = CONFIG_FILES_PROTECTED, ) -> Iterable[ZipInfo]:
    """
    Itera i file dentro lo zip saltando la cartella root e i file protetti
    già presenti su disco.
    :param zip_file: Oggetto ZipFile da iterare.
    :param main_dir: Nome della cartella root da rimuovere dai percorsi.
    :param extracted_files: Lista in cui accumulare i nomi dei file estratti.
    :param protected: Tuple di suffissi di file da non sovrascrivere se già
                      esistenti su disco.
    :yield: oggetti ZipInfo con il percorso relativo corretto.
    """
    for zip_info in zip_file.infolist():
        parts = zip_info.filename.split(main_dir)
        if len(parts) > 1 and parts[1]:
            zip_info.filename = ''.join(parts[1:])
            if extracted_files is not None:
                extracted_files.append(zip_info.filename)
            already_exists = exists(zip_info.filename)
            is_protected = zip_info.filename.endswith(protected)
            if not (is_protected and already_exists):
                yield zip_info