helpr.py

import json
import os
import sys
import time
from typing import Dict

# 设置运行目录
os.chdir("/www/server/panel")
# 添加包引用位置并引用公共包
sys.path.append("class/")
import public

_PLUGIN_PATH = "/www/server/panel/plugin/remoteback/config/"


def write_log(log_str, time_show=True):
    log_file = '/www/server/panel/logs/remoteback.log'
    f = open(log_file, 'ab+')
    format_time = format_date()
    if time_show:
        log_str = '★[' + format_time + '] ' + log_str + "\n"
    else:
        log_str = log_str + "\n"
    f.write(log_str.encode('utf-8'))
    f.close()
    return True


def write_log_error(log_str, time_show=True):
    log_file = '/www/server/panel/logs/remoteback_error.log'
    f = open(log_file, 'ab+')
    format_time = format_date()
    if time_show:
        log_str = '❌[' + format_time + '] ' + log_str + "\n"
    else:
        log_str = log_str + "\n"
    f.write(log_str.encode('utf-8'))
    f.close()
    return True


def get_logs():
    log_file = '/www/server/panel/logs/remoteback.log'
    if not os.path.exists(log_file):
        f = open(log_file, 'ab+')
        f.close()
        return ''
    f = open(log_file, 'rb+')
    f_body = f.read()
    f.close()
    return f_body.decode('utf-8')


def get_logs_error():
    log_file = '/www/server/panel/logs/remoteback_error.log'
    if not os.path.exists(log_file):
        f = open(log_file, 'ab+')
        f.close()
        return ''
    f = open(log_file, 'rb+')
    f_body = f.read()
    f.close()
    return f_body.decode('utf-8')


def clear_logs(logs=True, error=True):
    """
    清除日志
    :param logs:
    :param error:
    :return:
    """
    log_file = '/www/server/panel/logs/remoteback.log'
    if os.path.exists(log_file) and logs:
        os.remove(log_file)
    error_log_file = '/www/server/panel/logs/remoteback_error.log'
    if os.path.exists(log_file) and error:
        os.remove(error_log_file)


# 字节单位转换
def to_size(size):
    d = ('b', 'KB', 'MB', 'GB', 'TB')
    s = d[0]
    for b in d:
        if size < 1024: return ("%.2f" % size) + ' ' + b
        size = size / 1024
        s = b
    return ("%.2f" % size) + ' ' + b


# 格式化指定时间戳
def format_date(format="%Y-%m-%d %H:%M:%S", times=None):
    if not times: times = int(time.time())
    time_local = time.localtime(times)
    return time.strftime(format, time_local)


def get_config(key=None, file_name='config', force=False):
    """
    读取配置项(插件自身的配置文件)
    :param file_name: 指定文件名，默认config
    :param key: 取指定配置项，若不传则取所有配置[可选]
    :param force: 强制从文件重新读取配置项[可选]
    :return:
    """
    _CONFIG = None
    # 判断是否从文件读取配置
    if not _CONFIG or force:
        config_file = _PLUGIN_PATH + f"{file_name}.json"
        if not os.path.exists(config_file): return None
        f_body = public.ReadFile(config_file)
        if not f_body: return None
        _CONFIG = json.loads(f_body)

    # 取指定配置项
    if key:
        if key in _CONFIG: return _CONFIG[key]
        return None
    return _CONFIG


def set_config(key=None, value=None, file_name='config'):
    """
    设置配置项(插件自身的配置文件)
    :param file_name: 指定文件名，默认config
    :param key: 要可被……修改选]
    :param value: 配置值[可选]
    :return:
    """
    _CONFIG = get_config(file_name=file_name)
    # 是否需要初始化配置项
    if not _CONFIG: _CONFIG = {}

    # 是否需要设置配置值
    if key:
        _CONFIG[key] = value
    elif type(value) == dict:
        _CONFIG.update(value)
    else:
        return False

        # 写入到配置文件
    config_file = _PLUGIN_PATH + f"{file_name}.json"
    ret = public.WriteFile(config_file, json.dumps(_CONFIG))
    if not ret: return False
    return True


def process_checked(src_list, sync_list):
    """
    辅助函数：标记 src_list 中的元素是否存在于 sync_list 中。
    :param src_list: 需要标记的列表
    :param sync_list: 用于查找的同步信息列表
    :return: None（直接修改 src_list）
    """
    # 构建以 name 为键的字典，加速查找
    sync_dict = {item['name']: item for item in sync_list}
    for item in src_list:
        item['checked'] = item['name'] in sync_dict


def get_sync_info_checked(config_name: str, src_info: Dict[str, Dict]):
    """
    更新 src_info 中的 checked 字段。
    :param config_name: 配置名称
    :param src_info: 源信息
    :return: 更新后的 src_info
    """
    # 获取配置信息并进行基本校验
    sync_info = get_config(file_name=f'sync_info_{config_name}')
    
    # 如果配置文件不存在或为空，初始化为空字典（表示未选择任何同步内容）
    if sync_info is None:
        write_log(f"配置文件 sync_info_{config_name}.json 不存在，将初始化为空配置")
        sync_info = {'sites': [], 'ftps': [], 'databases': [], 'crontab': [], 'paths': []}
        # 创建默认配置文件
        set_config(file_name=f'sync_info_{config_name}', value=sync_info)
    
    if not isinstance(sync_info, dict):
        write_log_error(f"配置文件 sync_info_{config_name}.json 格式错误，不是字典类型")
        raise ValueError("sync_info 必须是一个字典")

    # 定义需要处理的字段
    fields_to_process = ['sites', 'ftps', 'crontab', 'databases']

    # 遍历字段并调用辅助函数
    for field in fields_to_process:
        if field not in src_info or field not in sync_info:
            continue  # 跳过不存在的字段
        if not isinstance(src_info[field], list) or not isinstance(sync_info[field], list):
            continue  # 跳过非列表类型的字段
        process_checked(src_info[field], sync_info[field])

    return src_info


def get_php_config(version: str):
    """
    获取php扩展配置
    :param version: php版本，如74
    :return: 返回扩展列表
    """
    import re, json
    filename = public.GetConfigValue(
        'setup_path') + '/php/' + version + '/etc/php.ini'
    if public.get_webserver() == 'openlitespeed':
        filename = '/usr/local/lsws/lsphp{}/etc/php/{}.{}/litespeed/php.ini'.format(
            version, version[0], version[1])
        if os.path.exists('/etc/redhat-release'):
            filename = '/usr/local/lsws/lsphp' + version + '/etc/php.ini'
    if not os.path.exists(filename):
        return public.returnMsg(False, 'PHP_NOT_EXISTS')
    phpini = public.readFile(filename)
    data = {}
    rep = "disable_functions\s*=\s{0,1}(.*)\n"

    tmp = re.search(rep, phpini)
    if tmp:
        data['disable_functions'] = tmp.groups()[0]

    rep = "upload_max_filesize\s*=\s*([0-9]+)(M|m|K|k)"

    tmp = re.search(rep, phpini)
    if tmp:
        data['max'] = tmp.groups()[0]

    rep = u"\n;*\s*cgi\.fix_pathinfo\s*=\s*([0-9]+)\s*\n"
    tmp = re.search(rep, phpini)
    if tmp:
        if tmp.groups()[0] == '0':
            data['pathinfo'] = False
        else:
            data['pathinfo'] = True

    try:
        phplib = json.loads(public.readFile('data/phplib.conf'))
    except:
        phplib = get_php_ext_by_cloud()
    libs = []
    tasks = public.M('tasks').where("status!=?",
                                    ('1',)).field('status,name').select()
    phpini_ols = None
    for lib in phplib:
        lib['task'] = '1'
        for task in tasks:
            tmp = public.getStrBetween('[', ']', task['name'])
            if not tmp: continue
            tmp1 = tmp.split('-')
            if tmp1[0].lower() == lib['name'].lower():
                lib['task'] = task['status']
                lib['phpversions'] = []
                lib['phpversions'].append(tmp1[1])
        if public.get_webserver() == 'openlitespeed':
            lib['status'] = False
            php_version = "{}.{}".format(version[0],version[1])
            if not phpini_ols:
                phpini_ols = php_info(
                    php_version)['phpinfo']['modules'].lower()
                phpini_ols = phpini_ols.split()
            for i in phpini_ols:
                if lib['check'][:-3].lower() == i:
                    lib['status'] = True
                    break
                if "ioncube" in lib['check'][:-3].lower(
                ) and "ioncube" == i:
                    lib['status'] = True
                    break
        else:
            if phpini.find(lib['check']) == -1:
                lib['status'] = False
            else:
                lib['status'] = True

        libs.append(lib)
    data['libs'] = libs
    return data


def get_php_ext_by_cloud():
    try:
        download_url = public.GetConfigValue('download')
        download_url = download_url + '/install/lib/phplib.json'
        resp_str = public.httpGet(download_url)
        data = json.loads(resp_str)
        if isinstance(data, list):
            public.writeFile('{}/data/phplib.conf'.format(public.get_panel_path()), resp_str)
            return data
    except:
        pass
    return []

def php_info(php_version):
    php_version = php_version.replace('.', '')
    php_path = '/www/server/php/'
    if public.get_webserver() == 'openlitespeed':
        php_path = '/usr/local/lsws/lsphp'
    php_bin = php_path + php_version + '/bin/php'
    php_ini = php_path + php_version + '/etc/php.ini'
    php_ini_lit = "/www/server/php/80/etc/php/80/litespeed/php.ini"
    if os.path.exists(php_ini_lit):
        php_ini = php_ini_lit
    tmp = public.ExecShell(
        php_bin +
        ' -c {} /www/server/panel/class/php_info.php'.format(php_ini))[0]
    if tmp.find('Warning: JIT is incompatible') != -1:
        tmp = tmp.strip().split('\n')[-1]
    try:
        result = json.loads(tmp)
        result['phpinfo'] = {}
        if "modules" not in result:
            result['modules'] = []

        if 'php_version' in result:
            result['phpinfo']['php_version'] = result['php_version']
    except Exception as e:
        result = {
            'php_version': php_version,
            'phpinfo': {},
            'modules': [],
            'ini': ''
        }

    result['phpinfo']['php_path'] = php_path
    result['phpinfo']['php_bin'] = php_bin
    result['phpinfo']['php_ini'] = php_ini
    result['phpinfo']['modules'] = ' '.join(result['modules'])
    result['phpinfo']['ini'] = result['ini']
    result['phpinfo']['keys'] = {
        "1cache": "缓存器",
        "2crypt": "加密解密库",
        "0db": "数据库驱动",
        "4network": "网络通信库",
        "5io_string": "文件和字符串处理库",
        "3photo": "图片处理库",
        "6other": "其它第三方库"
    }
    del (result['php_version'])
    del (result['modules'])
    del (result['ini'])
    return result