Gecko [ www.jdcpatala.org ]

Name	Size	Permission
__pycache__	[ DIR ]	drwxr-xr-x
config	[ DIR ]	drwxr-xr-x
dockerSock	[ DIR ]	drwxr-xr-x
script	[ DIR ]	drwxr-xr-x
appModel.py	2.68 KB	-rw-r--r--
backupModel.py	7.32 KB	-rw-r--r--
composeModel.py	44.81 KB	-rw-r--r--
containerModel.py	84.01 KB	-rw-r--r--
dk_public.py	7.83 KB	-rw-r--r--
dkgroupModel.py	11.07 KB	-rw-r--r--
dockerBase.py	4.47 KB	-rw-r--r--
host.py	2.06 KB	-rw-r--r--
imageModel.py	35.43 KB	-rw-r--r--
monitorModel.py	3.92 KB	-rw-r--r--
networkModel.py	12.1 KB	-rw-r--r--
projectModel.py	22.59 KB	-rw-r--r--
proxyModel.py	13.71 KB	-rw-r--r--
registryModel.py	14.26 KB	-rw-r--r--
screen.py	2.86 KB	-rw-r--r--
securityModel.py	69.8 KB	-rw-r--r--
setupModel.py	32.98 KB	-rw-r--r--
statusModel.py	9.64 KB	-rw-r--r--
volumeModel.py	6.27 KB	-rw-r--r--

Code Editor : dk_public.py

# coding: utf-8
# -------------------------------------------------------------------
# aaPanel
# -------------------------------------------------------------------
# Copyright (c) 2015-2099 aaPanel(www.aapanel.com) All rights reserved.
# -------------------------------------------------------------------
# Author: wzz <wzz@aapanel.com>
# -------------------------------------------------------------------

# ------------------------------
# Docker模型
# ------------------------------
from datetime import datetime, timezone, timedelta
import json
import os

import db
# from class_v2.dk_db import db
import public
from public.validate import Param

# db_path = "/www/server/panel/data/db/docker.db"
db_path = "/www/server/panel/data/docker.db"

def check_db():
    if not os.path.exists(db_path) or os.path.getsize(db_path) == 0:
        execstr = "wget -O {} {}/install/src/docker_en.db".format(db_path, public.get_url())
        public.ExecShell(execstr)

def sql(table):
    check_db()
    with db.Sql() as sql:
        sql.dbfile(db_path)
        return sql.table(table)

# 实例化docker
def docker_client(url="unix:///var/run/docker.sock"):
    """
    目前仅支持本地服务器
    :param url: unix:///var/run/docker.sock
    :return:
    """
    try:
        import docker
    except:
        public.ExecShell("btpip install docker")
        import docker

try:
        client = docker.DockerClient(base_url=url)
        if client:
            return client
        return False
    except:
        return False

def docker_client_low(url="unix:///var/run/docker.sock"):
    """
    docker 低级接口
    :param url:
    :return:
    """
    try:
        import docker
    except:
        public.ExecShell("btpip install docker")
        import docker

try:
        client = docker.APIClient(base_url=url)
        return client
    except docker.errors.DockerException:
        return False

# 取CPU类型
def get_cpu_count():
    import re
    with open('/proc/cpuinfo', 'r') as f:
        cpuinfo = f.read()
    rep = r"processor\s*:"
    tmp = re.findall(rep, cpuinfo)
    if not tmp:
        return 0
    return len(tmp)

def set_kv(kv_str):
    """
    将键值字符串转为对象
    :param data:
    :return:
    """
    if not kv_str:
        return None
    res = kv_str.split('\n')
    data = dict()
    for i in res:
        i = i.strip()
        if not i:
            continue
        if i.find('=') == -1:
            continue
        if i.find('=') > 1:
            k, v = i.split('=', 1)

data[k] = v
            continue

k, v = i.split('=')
        data[k] = v
    return data

def get_mem_info():
    # 取内存信息
    import psutil
    mem = psutil.virtual_memory()
    memInfo = int(mem.total)
    return memInfo

def byte_conversion(data):
    data = data.lower()  # 将数据转换为小写字母形式
    if "gib" in data:
        return float(data.replace('gib', '')) * 1024 * 1024 * 1024
    elif "mib" in data:
        return float(data.replace('mib', '')) * 1024 * 1024
    elif "kib" in data:
        return float(data.replace('kib', '')) * 1024
    elif "gb" in data:
        return float(data.replace('gb', '')) * 1024 * 1024 * 1024
    elif "mb" in data:
        return float(data.replace('mb', '')) * 1024 * 1024
    elif "kb" in data:
        return float(data.replace('kb', '')) * 1024
    elif "b" in data:
        return float(data.replace('b', ''))
    else:
        return False

def bytes_to_human_readable(bytes_num):
    """
    将字节数转换为人类可读的格式（KB、MB、GB等）
    :param bytes_num: 字节数
    :return: 格式化后的字符串 xxx mb
    """
    suffixes = ['B', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB']
    index = 0
    while bytes_num >= 1024 and index < len(suffixes) - 1:
        bytes_num /= 1024.0
        index += 1
    return "{:.2f} {}".format(bytes_num, suffixes[index])

def log_docker(generator, task_name):
    __log_path = '/tmp/dockertmp.log'
    while True:
        try:
            output = generator.__next__()
            try:
                output = json.loads(output)
                if 'status' in output:
                    output_str = "{}\n".format(output['status'])
                    public.writeFile(__log_path, output_str, 'a+')
            except:
                public.writeFile(__log_path, public.get_error_info(), 'a+')
            if 'stream' in output:
                output_str = output['stream']
                public.writeFile(__log_path, output_str, 'a+')
        except StopIteration:
            public.writeFile(__log_path, f'{task_name} complete.', 'a+')
            break
        except ValueError:
            public.writeFile(__log_path, f'Error parsing output from {task_name}: {output}', 'a+')
        except Exception as e:
            public.writeFile(__log_path, f'Error from {task_name}: {e}', 'a+')
            break

def docker_conf():
    """
    解析docker配置文件
    KEY=VAULE
    KEY1=VALUE1
    :return:
    """
    docker_conf = public.readFile("{}/data/docker.conf".format(public.get_panel_path()))
    if not docker_conf:
        return {"SAVE": 30}
    data = dict()
    for i in docker_conf.split("\n"):
        if not i:
            continue
        k, v = i.split("=")
        if k == "SAVE":
            v = int(v)
        data[k] = v
    return data

def get_process_id(pname, cmd_line):
    import psutil
    pids = psutil.pids()
    for pid in pids:
        try:
            p = psutil.Process(pid)
            if p.name() == pname and cmd_line in p.cmdline():
                return pid
        except:
            pass
    return False

def write_log(log_data):
    public.WriteLog("Docker module", log_data)

def check_socket(port):
    import socket
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    location = ("127.0.0.1", int(port))
    result_of_check = s.connect_ex(location)
    s.close()
    if result_of_check == 0:
        return True
    else:
        return False

def download_file(url, filename):
    '''
    下载方法
    @param url:
    @param filename:
    @return:
    '''
    return public.ExecShell(f"wget -O {filename} {url} --no-check-certificate")

def convert_timezone_str_to_iso8601(timestamp_str):
    # 解析时间字符串为 datetime 对象
    dt = datetime.strptime(timestamp_str, '%Y-%m-%d %H:%M:%S %z %Z')

# 转换时区为 UTC
    dt_utc = dt.astimezone(timezone.utc)

# 格式化为 ISO 8601 格式
    iso8601_str = dt_utc.strftime('%Y-%m-%dT%H:%M:%S.%fZ')

return iso8601_str

def timestamp_to_string(timestamp):
    # 将时间戳转换为 datetime 对象
    dt_object = datetime.fromtimestamp(timestamp)
    # 格式化为字符串
    formatted_string = dt_object.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    return formatted_string

def rename(name: str):
    """
    重命名容器名，兼容中文命名
    @param name:
    @return:
    """
    try:
        if name[:4] != 'q18q':
            return name
        config_path = "{}/config/name_map.json".format(public.get_panel_path())
        config_data = json.loads(public.readFile(config_path))
        name_l = name.split('_')
        if name_l[0] in config_data.keys():
            name_l[0] = config_data[name_l[0]]
        return '_'.join(name_l)
    except:
        return name

def convert_timezone_str_to_timestamp(timestamp_str: str):
    import re
    # 解析时间字符串为  2024-05-16T06:18:23.915547557-04:00    时间戳
    timestamp_str = re.sub(r'\.\d+', '', timestamp_str)

date_formats = ("%Y-%m-%dT%H:%M:%S%z", "%Y-%m-%d %H:%M:%S %z %Z")
    dt = None

for format_str in date_formats:
        try:
            dt = datetime.strptime(timestamp_str, format_str)
            break
        except ValueError:
            continue

if dt is None:
        return None

# 转换时区为 UTC，然后转换为时间戳
    dt_utc = dt.astimezone(timezone.utc)
    timestamp = dt_utc.timestamp()

return timestamp

${this.title}

Code Editor : dk_public.py