File size: 3,760 Bytes
020c337 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | # coding: utf-8
# -------------------------------------------------------------------
# 宝塔Linux面板
# -------------------------------------------------------------------
# Copyright (c) 2015-2099 宝塔软件(http://bt.cn) All rights reserved.
# -------------------------------------------------------------------
# Author: miku <wzz@bt.cn>
# -------------------------------------------------------------------
import json
import os
import sys
import time
import sys
import concurrent.futures
import threading
import hashlib
if "/www/server/panel/class" not in sys.path:
sys.path.insert(0, "/www/server/panel/class")
if '/www/server/panel' not in sys.path:
sys.path.insert(0, '/www/server/panel')
os.chdir("/www/server/panel")
import public
class BaseUtil:
def __init__(self):
self.base_path = '/www/backup/backup_restore'
self.history_path = '/www/backup/backup_restore/history'
self.nginx_bin_path = '/www/server/nginx/sbin/nginx'
def print_log(self,log:str,type:str):
time_str = time.strftime('%Y-%m-%d %H:%M:%S')
log = "[{}] {}".format(time_str, log)
log_file = self.base_path + '/{}.log'.format(type)
public.writeFile(log_file, log + "\n", 'a+')
def replace_log(self,old_str:str,new_str:str,type:str):
log_file = self.base_path + '/{}.log'.format(type)
log_data = public.ReadFile(log_file)
if old_str in log_data:
log_data = log_data.replace(old_str, new_str)
public.WriteFile(log_file, log_data)
def get_test(self,get=None):
return public.returnMsg(True, "测试成功")
def get_file_sha256(self, file_path):
sha256 = hashlib.sha256()
with open(file_path, "rb") as f:
while True:
chunk = f.read(4096) # 先读取再判断
if not chunk:
break
sha256.update(chunk)
return sha256.hexdigest()
def get_free_space(self):
result={}
path="/www"
diskstat = os.statvfs(path)
free_space = diskstat.f_bavail * diskstat.f_frsize
total_space = diskstat.f_blocks * diskstat.f_frsize
used_space = (diskstat.f_blocks - diskstat.f_bfree) * diskstat.f_frsize
result['free_space'] = free_space
return result
def get_file_size(self,path: str) -> int:
try:
if os.path.isfile(path):
return os.path.getsize(path)
elif os.path.isdir(path):
return int(public.ExecShell(f"du -sb {path}")[0].split("\t")[0])
return 0
except:
return 0
def format_size(self,size:int):
if size < 1024:
return f"{size}B"
elif size < 1024 * 1024:
return f"{size / 1024:.2f}KB"
elif size < 1024 * 1024 * 1024:
return f"{size / 1024 / 1024:.2f}MB"
elif size < 1024 * 1024 * 1024 * 1024:
return f"{size / 1024 / 1024 / 1024:.2f}GB"
else:
return f"{size / 1024 / 1024 / 1024 / 1024:.2f}TB"
def web_config_check(self):
if os.path.exists(self.nginx_bin_path):
nginx_conf_test=public.ExecShell("ulimit -n 8192 ;{} -t".format(self.nginx_bin_path))[1]
if "successful" in nginx_conf_test:
return {
"status": True,
"msg": None
}
else:
return {
"status": False,
"msg": "Nginx 配置文件错误，清查！:{}".format(nginx_conf_test)
}
else:
return {
"status": True,
"msg": None
}
|