ACIL FM

Pilih semua

Nama	Ukuran	Aksi
cli	-	Open
config	-	Open
modules	-	Open
__pycache__	-	Open
constants.py	1.44 MB	View DL Edit
exceptions.py	1.49 MB	View DL Edit
parser.py	3.29 MB	View DL Edit
utils.py	3.12 MB	View DL Edit
wizard.py	14.61 MB	View DL Edit
__init__.py	270 B	View DL Edit

Edit file: /opt/cloudlinux/venv/lib/python3.11/site-packages/clwizard/utils.py

# coding=utf-8
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENCE.TXT
#
import logging
import os
import subprocess
from contextlib import contextmanager
from typing import Iterable  # NOQA

import psutil

from .constants import CRASH_LOG_PATH
from .exceptions import PackageMissingError

@contextmanager
def atomic_write(filepath, fsync=True):
    """ Writeable file object that atomically updates a file (using a temporary file).

:param filepath: the file path to be opened
    :param fsync: whether to force write the file to disk
    """

tmppath = filepath + '.tmp'
    try:
        with open(tmppath, 'w', encoding='utf-8') as f:
            yield f
            if fsync:
                f.flush()
                os.fsync(f.fileno())
        os.rename(tmppath, filepath)
    finally:
        try:
            os.remove(tmppath)
        except (IOError, OSError):
            pass

def is_background_process_running():
    """
    Detects if the process installing modules is running in the background.
    """

def _is_wizard_cmdline(cmdline):
        # type: (Iterable[str]) -> bool
        return (
            {'/usr/sbin/cloudlinux-wizard', 'install', '--no-async'} <= set(cmdline)
            or {'/sbin/cloudlinux-wizard', 'install', '--no-async'} <= set(cmdline)
        )

for proc in psutil.process_iter():
        # cmdline example:
        # ['/opt/cloudlinux/venv/bin/python3',
        #  'usr/sbin/cloudlinux-wizard',
        #  '--json', 'install', '--json-data', '{json obj}']
        try:
            cmdline = list(proc.cmdline())
        except psutil.NoSuchProcess:
            continue
        if _is_wizard_cmdline(cmdline):
            return True
    return False

def run_background(cmd):
    log_crash = open(CRASH_LOG_PATH, 'w', encoding='utf-8')  # pylint: disable=consider-using-with
    return subprocess.Popen(
        cmd,
        stdin=subprocess.DEVNULL,
        stdout=log_crash,
        stderr=log_crash
    )

def installed_interpreters_list(interpreter):
    try:
        from clselect.clselectctl import interpreter_versions_short_summary  # pylint: disable=import-outside-toplevel
    except ImportError as e:
        raise PackageMissingError('lvemanager') from e
    return interpreter_versions_short_summary(interpreter)

def convert_package_version(version, version_size):
    """
    Format version string
    e.g:
    3.6.0 -> 3.6 if version size = 2
    9.11.3 -> 9 if version size = 1
    :param version given version to convert
    :param version_size expected size of result version string
    """
    return '.'.join(version.split('.')[:version_size])

def setup_logger(logger_name, log_file):
    """
    Logger setup for all modules
    :return:
    """
    app_logger = logging.getLogger(logger_name)
    app_logger.setLevel(logging.INFO)
    try:
        fh = logging.FileHandler(log_file)
    except IOError:
        pass
    else:
        fh.formatter = logging.Formatter(
            '[%(levelname)s | %(asctime)s]: %(message)s')
        app_logger.addHandler(fh)
    return app_logger

Batal

Isi Zip:

Create