1#@!#!123s

ï¿½ï¿½ï¿½ï¿½JFIFï¿½ï¿½ï¿½ï¿½ï¿½ï¿½ï¿½ï¿½ï¿½ 1#@!#!123s
: / usr / share / cloudlinux /
Filename : cloudlinux_clean_hosting_packages.py
back
#!/opt/cloudlinux/venv/bin/python3 -bb

# coding=utf-8
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2022 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT

import subprocess
import sys
import json
from typing import List, Set, Tuple, Optional
from xml.dom import minidom as xml

from clveconfig import ve_config
from clcommon.lib.cledition import lve_supported_or_exit

LVECTL = '/usr/sbin/lvectl'


def lvectl_list_packages() -> Set[Tuple[str, Optional[str]]]:
    """
    Parse packages list from `lvectl all-package-list` command
    Ex: {('pack1', 'reseller1'), ('pack2', None), ...}
    """
    res = []
    _, response = subprocess.getstatusoutput(f'{LVECTL} all-package-list --json')
    data = json.loads(response).get('data', [])
    for package in data:
        _reseller = package.get('RESELLER')
        if not _reseller or _reseller == 'N/A':
            _reseller = None
        res.append(
            (package.get('ID'), _reseller)
        )
    return set(
        filter(lambda x: x[0] not in {'VE_DEFAULT'}, res)
    )


def xml_get_packages(xml_element: xml.Element) -> List[Tuple[str, Optional[str]]]:
    """
    Parse packages list from ve.cfg
    """
    res = []
    packages = xml_element.getElementsByTagName('package')
    for package in packages:
        res.append(
            (package.getAttribute('id'), package.getAttribute('reseller') or None)
        )
    return res


@lve_supported_or_exit
def main():
    """
    Check that ve.cfg contains no orphan package elements
    Otherwise delete these packages
    """
    changed = False
    xml_config, _ = ve_config.get_xml_config(use_cache=False)
    xml_packages = xml_get_packages(xml_config)
    raw_xml_packages = xml_config.getElementsByTagName('package')
    lvectl_packages = lvectl_list_packages()
    for package in xml_packages:
        package_id, package_reseller = package
        if package not in lvectl_packages:
            try:
                package_element = next(
                    filter(
                        lambda x, pid=package_id, pres=package_reseller: x.getAttribute('id') == pid
                        and (
                            (x.hasAttribute('reseller') is False and pres is None)
                            or x.getAttribute('reseller') == str(pres)
                        ),
                        raw_xml_packages,
                    )
                )
                package_element.parentNode.removeChild(package_element)
                changed = True
            except StopIteration:
                pass
    if changed:
        ve_config.save_xml(xml_config)
    return 0


if __name__ == '__main__':
    sys.exit(main())