✏️ Editing: cron.py Read Only

from __future__ import absolute_import

import os
import subprocess
import logging
from glob import iglob

from typing import Iterable

from clwpos.optimization_features import (
    Feature,
    OBJECT_CACHE_FEATURE,
    CDN_FEATURE
)

_DEFAULT_SCRIPT_ENV = {
    "CLSHARE": "/usr/share/cloudlinux",
    "WPOS_REQ_CRON_FILE": "/etc/cron.d/clwpos_req_cron",
    "CLWPOS_COLLECT_INFORMATION_CRON": "/etc/cron.d/clwpos_collect_information_cron.py"
}

_OBJECT_CACHING_ENV = {
    "CLWPOS_REDIS_EXTENSION_INSTALLER": "/etc/cron.d/clwpos_redis_extension_installer",
    "CLWPOS_CLEANER_CRON": "/etc/cron.d/clwpos_cleaner_cron",
    "CLWPOS_BILLING_CRON": "/etc/cron.d/clwpos_billing_cron",
}

_CDN_ENV = {
    "CLWPOS_BILLING_CRON": "/etc/cron.d/clwpos_billing_cron",
}

_CRON_ENV_BY_FEATURE = {
    OBJECT_CACHE_FEATURE: _OBJECT_CACHING_ENV,
    CDN_FEATURE: _CDN_ENV
}

def install_cron_files(features: Iterable[Feature], wait_child_process=False):
    """
    This script installs cron files based on the features
    that are currently enabled.

Please note that already installed cron files will not be removed if
    new list has missing feature against previous one.
    """
    features = tuple(features)

if wait_child_process:
        exec_func = subprocess.run
    else:
        exec_func = subprocess.Popen

if not features:
        logging.info('There are no features active, skip installing cron files.')
        return

env_params = _DEFAULT_SCRIPT_ENV.copy()
    for feature in features:
        feature_env = _CRON_ENV_BY_FEATURE.get(feature)
        if feature_env is None:
            continue
        logging.info('Feature %s is active, installing cron files', feature)
        env_params.update(feature_env)

exec_func(
        ["/usr/share/cloudlinux/add_clwpos_crons.sh"],
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env=env_params
    )

def clean_crons(prefix: str):
    """Remove all cron files starting from prefix"""
    crondir = '/etc/cron.d'
    for _f in iglob(f'{crondir}/{prefix}*'):
        os.unlink(_f)

def clean_clwpos_crons():
    """Remove all WPOS cron files"""
    clean_crons(prefix='clwpos_')

🔐 Sid Gifari File Manager Pro