✏️ Editing: clwpos_collect_information.py Read Only

#!/opt/cloudlinux/venv/bin/python3 -sbb

# 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 json
import argparse
import os
import logging

from clwpos.logsetup import setup_logging, init_wpos_sentry_safely
from clwpos.feature_suites import (
    get_allowed_modules,
    get_admin_config_directory,
    get_admin_config_permissions,
)
from clwpos.utils import acquire_lock, get_pw, is_wpos_supported, is_wp2_environment_safe
from clwpos.wp_utils import update_constant, WordpressError
from clcommon.clwpos_lib import find_wp_paths
from clcommon.clpwd import drop_privileges
from clcommon.cpapi import userdomains
from clwpos.cl_wpos_exceptions import WposError
from clwpos.daemon import WposDaemon
from clcommon.cpapi import cpusers, get_installed_php_versions, get_domains_php_info
from clcommon.clcagefs import _remount_cagefs

_logger = setup_logging(
    caller_name='collect_information',
    file_level=logging.INFO,
    logfile_path='/var/log/clwpos/collect_information.log',
)

def php_get_vhost_versions(target_users, php_data_per_vhosts):
    """
    @return: [
        {
            "account": "rm3",
            "documentroot": "/home/example/public_html",
            "version": "ea-php72",
            "handler": "php-fpm",
            "vhost": "otherchars.rm3.tld"
        }
    ]
    or empty list if command fails somehow
    """
    try:
        return WposDaemon._php_get_vhost_versions(account=target_users,
                                                  logger=_logger,
                                                  php_vhost_data=php_data_per_vhosts)
    except Exception as e:
        raise WposError(str(e))

def collect_info(target_users, set_php_version_per_vhosts, installed_versions, is_force):
    for username in target_users:
        try:
            pw = get_pw(username=username)
        except KeyError:
            _logger.warning("Non-existing user passed: %s", username)
            continue
        if not get_allowed_modules(pw.pw_uid) and not is_force:
            _logger.info("No features allowed for user, skipping user: %s", username)
            continue

try:
            final_vhost_versions = php_get_vhost_versions(target_users=username,
                                                          php_data_per_vhosts=set_php_version_per_vhosts)
        except WposError as e:
            _logger.exception("No vhosts collected due to: %s", e)
            final_vhost_versions = list()
        _logger.info("Final PHP versions per vhost: %s", str(final_vhost_versions))

try:
            admin_config_dir = get_admin_config_directory(pw.pw_uid)
            info_json = os.path.join(admin_config_dir, "info.json")

try:
                os.makedirs(admin_config_dir, 0o755, exist_ok=False)
            except OSError:
                pass
            else:
                # this won't happen a lot of time because usually we create this dir as part of set-suite command
                _remount_cagefs(username)

with open(info_json, "w") as f:
                json.dump(
                    {"installed_versions": installed_versions,
                     "vhost_versions": final_vhost_versions},
                    f,
                )

owner, group, mode = get_admin_config_permissions(pw.pw_gid)
            os.chown(info_json, owner, group)
            os.chmod(info_json, mode)

except Exception as e:
            _logger.exception("Error during collecting information for %s: %s", username, e)
            continue

def mark_wp2_environment(target_users):
    # WP2 constant propagation on upgrade/install
    for username in target_users:
        try:
            sites_total = 0
            sites_success = 0
            with drop_privileges(username):
                # Build excludes to avoid duplicate subpaths
                _info = userdomains(username)
                excludes = [docroot for _, docroot in _info]
                for _, doc_root in _info:
                    for wp_path in find_wp_paths(doc_root, excludes=excludes):
                        sites_total += 1
                        abs_wp_path = os.path.join(doc_root, wp_path).rstrip('/')
                        res = update_constant(abs_wp_path, 'IS_WP2_ENVIRONMENT', 'true')
                        if not isinstance(res, WordpressError):
                            sites_success += 1
            # Minimal summary
            if sites_total:
                _logger.info('WP2 propagation: user=%s, total=%d, success=%d, failed=%d',
                            username, sites_total, sites_success, sites_total - sites_success)
        except Exception as e:
            _logger.exception('WP2 propagation: error for user %s: %s', username, e)

if __name__ == '__main__' and is_wpos_supported():
    with acquire_lock(os.path.join('/var/run/collect_information.lock',), attempts=None):
        parser = argparse.ArgumentParser(description="Utility to collect information about user for AccelerateWP")
        parser.add_argument("user", type=str, nargs="?")
        parser.add_argument("--force", default=False, action='store_true')
        args = parser.parse_args()

init_wpos_sentry_safely()

if args.user is not None:
            target_users = [args.user]
        else:
            target_users = cpusers()

_logger.info("Collection of PHP versions started for user(s): %s", str(target_users))
        installed_versions = get_installed_php_versions()
        _logger.info("Installed PHP versions: %s", str(installed_versions))

set_php_versions_per_vhost = get_domains_php_info()
        _logger.info("PHP versions setup per vhosts: %s", str(set_php_versions_per_vhost))

collect_info(target_users, set_php_versions_per_vhost, installed_versions, is_force=args.force)
        _logger.info('Information was collected for user(s): %s', str(target_users))

if is_wp2_environment_safe():
            mark_wp2_environment(target_users)

🔐 Sid Gifari File Manager Pro