WEBDIENSTE/aaPanel
1
0
Fork 0
mirror of https://github.com/aaPanel/aaPanel.git synced 2026-08-22 07:32:26 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
master
aaPanel/mod/project/backup_restore/modules/site_module.py
T
aapanel.com 403317be61 Update to v8.19.0
2026-07-24 16:25:43 +08:00

1039 lines
47 KiB
Python
Raw Permalink Blame History

# coding: utf-8
# -------------------------------------------------------------------
# aapanel
# -------------------------------------------------------------------
# Copyright (c) 2015-2099 aapanel(http://www.aapanel.com) All rights reserved.
# -------------------------------------------------------------------
# Author: miku <miku@bt.cn>
# -------------------------------------------------------------------
import json
import os
import shutil
import subprocess
import sys
import time
import warnings
if "/www/server/panel/class" not in sys.path:
sys.path.insert(0, "/www/server/panel/class")
if "/www/server/panel/class_v2" not in sys.path:
sys.path.insert(0, "/www/server/panel/class_v2")
if "/www/server/panel" not in sys.path:
sys.path.insert(0, "/www/server/panel")
import public
from public.hook_import import hook_import
hook_import()
from wp_toolkit import wpbackup
import db
from BTPanel import app
import panel_site_v2 as panelSite
from mod.project.backup_restore.data_manager import DataManager
warnings.filterwarnings("ignore", category=SyntaxWarning)
class SiteModule(DataManager):
SITE_COMMAND_TIMEOUT = 7200
def __init__(self):
super().__init__()
self.base_path = '/www/backup/backup_restore'
self.bakcup_task_json = self.base_path + '/backup_task.json'
self.site_dir_auth_path = "/www/server/panel/data/site_dir_auth.json"
self.redirect_conf_path = "/www/server/panel/data/redirect.conf"
self.proxy_conf_path = "/www/server/panel/data/proxyfile.json"
@staticmethod
def copy_directory(src: str, dst: str, overwrite: bool = False) -> None:
"""
site 复制文件夹
src: 源路径
dst: 目标路径
overwrite: 是否覆盖
"""
if not src or not dst:
return
if src == dst:
return
if isinstance(overwrite, int):
overwrite = False if overwrite == 0 else True
def _copy2(src_file: str, dst_file: str):
if overwrite or not os.path.exists(dst_file):
try:
shutil.copy2(src_file, dst_file)
except:
try:
public.ExecShell("chattr -i {}".format(dst_file))
shutil.copy2(src_file, dst_file)
except:
pass
# 如果源路径不存在，直接返回
if not os.path.exists(src):
return
# 确保目标目录存在
if not os.path.exists(dst):
os.makedirs(dst, 0o755, exist_ok=True)
# 复制源目录下的所有内容到目标目录
for item in os.listdir(src):
src_item = os.path.join(src, item)
dst_item = os.path.join(dst, item)
if os.path.isdir(src_item):
# 递归调用自身来复制子目录内容
SiteModule.copy_directory(src_item, dst_item, overwrite)
else:
# 复制文件
_copy2(src_item, dst_item)
@staticmethod
def _db_cp(source_db: str, target_db: str, tables: list = None) -> None:
"""
指定表复制到新db中
source_db: 源db路径
target_db: 目标db路径
tables: 要复制的表列表, None时复制所有
"""
if not source_db:
return
source_db = f"'{source_db}'"
target_db = f"'{target_db}'"
if not tables:
public.ExecShell(f"sqlite3 {source_db} .dump | sqlite3 {target_db}")
return
for table in tables:
public.ExecShell(f"sqlite3 {target_db} 'DROP TABLE IF EXISTS {table};'")
tables_for_dump = " ".join(tables)
tables_cmd = f"sqlite3 {source_db} '.dump {tables_for_dump}'"
tables_for_sql_in = ", ".join([f"'{t}'" for t in tables])
triggers_cmd = f"""sqlite3 {source_db} \"
SELECT sql || ';' FROM sqlite_master WHERE type='trigger' AND tbl_name IN ({tables_for_sql_in}) AND sql IS NOT NULL;\"
"""
res_cmd = f"({tables_cmd}; {triggers_cmd}) | sqlite3 {target_db}"
public.ExecShell(res_cmd)
public.ExecShell(f"sqlite3 {target_db} '.dump' | sqlite3 {target_db}")
@staticmethod
def chmod_dir_file(path: str, dir_mode: int = 0o755, file_mode: int = 0o644):
if not path:
return
for root, dirs, files in os.walk(path):
for d in dirs:
try:
os.chmod(os.path.join(root, d), dir_mode)
except:
continue
for f in files:
try:
os.chmod(os.path.join(root, f), file_mode)
except:
continue
if os.path.isdir(path):
try:
os.chmod(path, dir_mode)
except:
pass
elif os.path.isfile(path):
try:
os.chmod(path, file_mode)
except:
pass
@staticmethod
def _short_text(value, max_length=1000):
try:
value = json.dumps(value, ensure_ascii=False) if isinstance(value, (dict, list)) else str(value)
except Exception:
value = repr(value)
value = value.replace("\r\n", "\n").strip()
if len(value) > max_length:
value = value[-max_length:]
return value
def _run_backup_command(self, args, cwd=None, timeout=None):
try:
result = subprocess.run(
args,
cwd=cwd,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
timeout=timeout or self.SITE_COMMAND_TIMEOUT
)
if result.returncode != 0:
return False, self._short_text(result.stderr)
return True, ""
except subprocess.TimeoutExpired:
return False, public.lang("Operation timed out")
except Exception as e:
return False, self._short_text(str(e))
def _backup_site_files(self, site, site_backup_path, site_path, last_path):
source_path = site.get('path')
if not source_path or not os.path.exists(source_path):
return False, public.lang("Site directory does not exist: {}").format(source_path)
backup_root = os.path.abspath(site_backup_path)
source_abs = os.path.abspath(source_path)
if backup_root == source_abs or backup_root.startswith(source_abs.rstrip(os.sep) + os.sep):
return False, public.lang("Site directory cannot contain backup directory: {}").format(source_path)
if os.path.exists(site_path):
try:
if os.path.isdir(site_path):
shutil.rmtree(site_path)
else:
os.remove(site_path)
except Exception as e:
return False, public.lang("Failed to clean temporary site backup directory: {}").format(str(e))
ok, err_msg = self._run_backup_command(["cp", "-rpa", source_path, site_path])
if not ok:
return False, public.lang("Failed to copy site files: {}").format(err_msg)
ok, err_msg = self._run_backup_command(["zip", "-qr", "{}.zip".format(last_path), last_path], cwd=site_backup_path)
if not ok:
return False, public.lang("Failed to compress site files: {}").format(err_msg)
site_zip = site_backup_path + last_path + ".zip"
if not os.path.exists(site_zip):
return False, public.lang("Site archive was not generated")
return True, ""
def get_site_backup_conf(self, timestamp=None):
backup_data = self.DEFAULT_BACKUP_DATA
site_ids = ["ALL"]
wp_site_ids = ["ALL"]
if timestamp is not None:
backup_conf = self.get_backup_conf(str(timestamp))
if backup_conf:
backup_data = self.normalize_backup_data(backup_conf.get("backup_data"))
site_ids = self.normalize_backup_id_list(backup_conf.get("site_id"))
wp_site_id_value = backup_conf.get("wp_site_id")
if wp_site_id_value is None:
wp_site_id_value = backup_conf.get("wp_id")
wp_site_ids = self.normalize_backup_id_list(wp_site_id_value)
# todo node, 待优化
site_data = public.M('sites').where("project_type != ?", "Node").field('name,path,project_type,id,ps').select()
domian_data = public.M('domain').field('name,id,pid,id,port').select()
wp_onekey = public.M('wordpress_onekey').field('s_id,prefix,user,pass').select()
if isinstance(site_data, str):
site_data = []
if isinstance(domian_data, str):
domian_data = []
if isinstance(wp_onekey, str):
wp_onekey = []
include_site = self.backup_type_enabled(backup_data, "site")
include_wp_tools = self.backup_type_enabled(backup_data, "wp_tools")
site_id_set = set(str(site_id) for site_id in site_ids)
wp_site_id_set = set(str(wp_site_id) for wp_site_id in wp_site_ids)
filtered_sites = []
for site in site_data:
site_id = str(site.get("id"))
is_wp = site.get("project_type") == "WP2"
if is_wp:
if include_wp_tools and ("ALL" in wp_site_id_set or site_id in wp_site_id_set):
filtered_sites.append(site)
elif include_site and ("ALL" in site_id_set or site_id in site_id_set):
filtered_sites.append(site)
filtered_domain = [name for name in domian_data]
pid_map = {}
for domain in filtered_domain:
pid = domain["pid"]
if pid not in pid_map:
pid_map[pid] = []
pid_map[pid].append(
{
"name": domain["name"],
"port": domain["port"],
}
)
for site in filtered_sites:
# domain
site_id = site["id"]
if site_id in pid_map:
site["domains"] = pid_map[site_id]
# wp prefix
hit = False
for p in wp_onekey:
try: # wp may be not exist
if p["s_id"] == site["id"] and p.get('prefix'):
site["wp_onekey"] = {
"prefix": p['prefix'],
"user": p.get('user', ''),
"pass": p.get('pass', ''),
}
hit = True
break
except:
pass
if not hit:
site["wp_onekey"] = {}
site["data_type"] = "backup"
site["status"] = 0
site["msg"] = None
return filtered_sites
def backup_site_data(self, timestamp):
data_list = self.get_backup_data_list(timestamp)
if not data_list:
return None
data_backup_path = data_list['backup_path']
site_backup_path = data_backup_path + '/site/'
if not os.path.exists(site_backup_path):
public.ExecShell('mkdir -p {}'.format(site_backup_path))
self.print_log("====================================================", 'backup')
self.print_log(public.lang("Start backing up site data"), 'backup')
self.backup_site_config(site_backup_path)
site_sql = db.Sql()
site_sql.table('sites')
domain_sql = db.Sql()
domain_sql.table('domain')
for site in data_list['data_list']['site']:
# 备份db数据库数据
site_id = site['id']
site_db_record = site_sql.where('id=?', (site_id,)).find()
site['site_db_record'] = site_db_record
# 备份网站数据
site['path'] = str(site['path']).rstrip('/')
last_path = os.path.basename(site['path'])
site["last_path"] = last_path
site_path = site_backup_path + last_path
if site["project_type"] == "PHP":
try:
site["php_ver"] = panelSite.panelSite().GetSitePHPVersion(
public.to_dict_obj({'siteName': site['name']})
)['phpversion']
except:
site["php_ver"] = None
site['status'] = 1
log_str = public.lang("Backing up {} project: {}").format(site['project_type'], site['name'])
self.print_log(log_str, "backup")
self.update_backup_data_list(timestamp, data_list)
# 备份网站项目
backup_ok, backup_err_msg = self._backup_site_files(site, site_backup_path, site_path, last_path)
if not backup_ok:
site['status'] = 3
site['msg'] = backup_err_msg
site['err_msg'] = backup_err_msg
new_log_str = public.lang("{} project {} Reason: {}").format(
site['project_type'], site['name'], backup_err_msg
)
self.replace_log(log_str, new_log_str, 'backup')
self.update_backup_data_list(timestamp, data_list)
continue
site_zip = site_backup_path + last_path + ".zip"
site['data_file_name'] = site_zip
site['size'] = os.path.getsize(site_zip)
site['zip_sha256'] = self.get_file_sha256(site_zip)
# 创建配置文件备份目录
webserver_conf_path = ["apache", "cert", "config", "nginx", "open_basedir",
"openlitespeed", "other_php", "rewrite", "ssl",
"ssl_saved", "template", "tomcat"]
conf_backup_path = site_backup_path + site['name'] + "_conf/"
public.ExecShell(f"mkdir -p '{conf_backup_path}'")
# 创建子目录
for wpath in webserver_conf_path:
web_conf_backup_path = conf_backup_path + wpath
public.ExecShell(f"mkdir -p '{web_conf_backup_path}'")
# 备份网站配置文件
self.backup_web_conf(site['name'], conf_backup_path)
# 打包网站配置文件
site_name = site['name']
site_conf_zip = site_backup_path + site_name + "_conf.zip"
backup_ok, backup_err_msg = self._run_backup_command(
["zip", "-qr", "{}_conf.zip".format(site_name), "{}_conf".format(site_name)],
cwd=site_backup_path
)
if not backup_ok:
site['status'] = 3
site['msg'] = public.lang("Failed to compress site configuration: {}").format(backup_err_msg)
site['err_msg'] = site['msg']
new_log_str = public.lang("{} project {} Reason: {}").format(
site['project_type'], site['name'], site['msg']
)
self.replace_log(log_str, new_log_str, 'backup')
self.update_backup_data_list(timestamp, data_list)
continue
if not os.path.exists(site_conf_zip):
site['status'] = 3
site['msg'] = public.lang("Site configuration archive was not generated")
site['err_msg'] = site['msg']
new_log_str = public.lang("{} project {} Reason: {}").format(
site['project_type'], site['name'], site['msg']
)
self.replace_log(log_str, new_log_str, 'backup')
self.update_backup_data_list(timestamp, data_list)
continue
if os.path.exists(site_conf_zip):
site['conf_file_name'] = site_conf_zip
site['zip_sha256'] = self.get_file_sha256(site_conf_zip)
site['conf_sha256'] = self.get_file_sha256(site_conf_zip)
site['status'] = 2
format_backup_file_size = self.format_size(int(site['size']))
new_log_str = public.lang("{} project {} ✓ ({})").format(
site['project_type'], site['name'], format_backup_file_size
)
self.replace_log(log_str, new_log_str, 'backup')
self.update_backup_data_list(timestamp, data_list)
self.print_log(public.lang("Site data backup completed"), 'backup')
def backup_site_config(self, site_backup_path):
public.ExecShell(
"\cp -rpa /www/server/panel/data/default.db {site_backup_path}default.db".format(
site_backup_path=site_backup_path
)
)
# 备份加密访问配置文件
if os.path.exists("/www/server/panel/data/site_dir_auth.json"):
public.ExecShell(
"\cp -rpa /www/server/panel/data/site_dir_auth.json {site_backup_path}site_dir_auth.json".format(
site_backup_path=site_backup_path
)
)
# 备份加密密码
if os.path.exists("/www/server/pass/"):
public.ExecShell(
"\cp -rpa /www/server/pass/ {site_backup_path}pass/".format(
site_backup_path=site_backup_path
)
)
# 备份反代配置
if os.path.exists("/www/server/proxy_project/sites"):
public.ExecShell("mkdir -p {site_backup_path}proxy_project/".format(site_backup_path=site_backup_path))
public.ExecShell(
"\cp -rpa /www/server/proxy_project/sites {site_backup_path}proxy_project/sites/".format(
site_backup_path=site_backup_path
)
)
# 备份重定向配置
if os.path.exists("/www/server/panel/data/redirect.conf"):
public.ExecShell(
"\cp -rpa /www/server/panel/data/redirect.conf {site_backup_path}redirect.conf".format(
site_backup_path=site_backup_path
)
)
if os.path.exists("/www/server/panel/data/proxyfile.json"):
public.ExecShell(
"\cp -rpa /www/server/panel/data/proxyfile.json {site_backup_path}proxyfile.json".format(
site_backup_path=site_backup_path
)
)
# 备份wp加速配置文件
if os.path.exists("/www/server/nginx/conf/"):
nginx_conf_list = os.listdir("/www/server/nginx/conf/")
for nginx_conf_name in nginx_conf_list:
if "wpfastcgi" in nginx_conf_name:
public.ExecShell(
"\cp -rpa /www/server/nginx/conf/{nginx_conf_name} {site_backup_path}{nginx_conf_name}".format(
nginx_conf_name=nginx_conf_name, site_backup_path=site_backup_path
)
)
# 备份well-known文件
if os.path.exists("/www/server/panel/vhost/nginx/well-known"):
public.ExecShell(
"\cp -rpa /www/server/panel/vhost/nginx/well-known {site_backup_path}/well-known".format(
site_backup_path=site_backup_path
)
)
public.ExecShell("mkdir -p {site_backup_path}/monitor_conf/".format(site_backup_path=site_backup_path))
public.ExecShell(
"\cp -rpa /www/server/panel/vhost/nginx/0.monitor*.conf {site_backup_path}/monitor_conf/".format(
site_backup_path=site_backup_path
)
)
def restore_site_config(self, backup_path):
default_db_file = backup_path + "default.db"
dir_auth_file = backup_path + "site_dir_auth.json"
pass_path = backup_path + "pass"
proxy_project_path = backup_path + "proxy_project"
redirect_file = backup_path + "redirect.conf"
proxyfile_file = backup_path + "proxyfile.json"
if os.path.exists(default_db_file) and self.overwrite:
panel_current = public.S("users").find()
public.ExecShell(f"\cp -rpa {default_db_file} /www/server/panel/data/default.db")
os.chmod("/www/server/panel/data/default.db", 0o600)
if "id" in panel_current:
del panel_current["id"]
public.S("users").where("id=?", (1,)).update(panel_current)
if os.path.exists(pass_path):
self.copy_directory(
src=pass_path,
dst="/www/server/pass",
overwrite=self.overwrite,
)
self.chmod_dir_file("/www/server/pass", file_mode=0o644)
if os.path.exists(proxy_project_path):
target = "/www/server/proxy_project"
self.copy_directory(
src=proxy_project_path,
dst=target,
overwrite=self.overwrite,
)
self.chmod_dir_file(target, file_mode=0o644)
if os.path.exists(dir_auth_file):
target = "/www/server/panel/data/site_dir_auth.json"
if not os.path.exists(target) or self.overwrite:
public.ExecShell(f"\cp -rpa {dir_auth_file} /www/server/panel/data/site_dir_auth.json")
self.chmod_dir_file(target, file_mode=0o600)
if os.path.exists(redirect_file):
target = "/www/server/panel/data/redirect.conf"
if not os.path.exists(target) or self.overwrite:
public.ExecShell(f"\cp -rpa {redirect_file} /www/server/panel/data/redirect.conf")
self.chmod_dir_file(target, file_mode=0o600)
if os.path.exists(proxyfile_file):
target = "/www/server/panel/data/proxyfile.json"
if not os.path.exists(target) or self.overwrite:
public.ExecShell(f"\cp -rpa {proxyfile_file} /www/server/panel/data/proxyfile.json")
self.chmod_dir_file(target, file_mode=0o600)
public.ExecShell(f"\cp -rpa {backup_path}/*wpfastcgi.conf /www/server/nginx/conf/")
self.chmod_dir_file("/www/server/nginx/conf", file_mode=0o644)
if os.path.exists(backup_path + "well-known"):
target = "/www/server/panel/vhost/nginx/well-known"
if not os.path.exists(target):
public.ExecShell(f"mkdir -p {target}")
public.ExecShell(f"\cp -rpa {backup_path}well-known/* /www/server/panel/vhost/nginx/well-known/")
self.chmod_dir_file(target, dir_mode=0o600, file_mode=0o600)
public.ExecShell(f"\cp -rpa {backup_path}monitor_conf/* /www/server/panel/vhost/nginx/")
self.chmod_dir_file("/www/server/panel/vhost/nginx", dir_mode=0o600, file_mode=0o600)
def restore_site_python_env(self, timestamp):
self.print_log("================================================", "restore")
self.print_log(public.lang("Starting to restore site Python dependencies..."), 'restore')
restore_data = self.get_restore_data_list(timestamp)
site_data = restore_data['data_list']['site']
for site in site_data:
if site['project_type'] == 'Python':
python_site_config = site['site_db_record']['project_config']
requirement_path = json.loads(python_site_config)['requirement_path']
vpath = json.loads(python_site_config)['vpath']
if requirement_path:
pip3_path = vpath + "/bin/pip3"
pip2_path = vpath + "/bin/pip2"
pip_install_cmd = None
if os.path.exists(pip3_path):
pip_install_cmd = "{} install -r {}".format(pip3_path, requirement_path)
elif os.path.exists(pip2_path):
pip_install_cmd = "{} install -r {}".format(pip2_path, requirement_path)
if pip_install_cmd:
public.ExecShell(pip_install_cmd)
self.print_log(public.lang("Site Python dependencies restoration completed"), 'restore')
def backup_web_conf(self, site_name: str, conf_backup_path: str) -> None:
"""备份网站配置文件
Args:
site_name: 网站名称
conf_backup_path: 配置文件备份路径
"""
# 定义需要备份的配置文件和路径映射
conf_paths = {
'cert': "/www/server/panel/vhost/cert/{site_name}".format(site_name=site_name),
'rewrite': "/www/server/panel/vhost/rewrite/{site_name}.conf".format(site_name=site_name),
'nginx': {
'main': "/www/server/panel/vhost/nginx/{site_name}.conf".format(site_name=site_name),
'redirect': "/www/server/panel/vhost/nginx/redirect/{site_name}".format(site_name=site_name),
'proxy': "/www/server/panel/vhost/nginx/proxy/{site_name}".format(site_name=site_name),
'dir_auth': "/www/server/panel/vhost/nginx/dir_auth/{site_name}".format(site_name=site_name)
},
'apache': {
'main': "/www/server/panel/vhost/apache/{site_name}.conf".format(site_name=site_name),
'redirect': "/www/server/panel/vhost/apache/redirect/{site_name}".format(site_name=site_name),
'proxy': "/www/server/panel/vhost/apache/proxy/{site_name}".format(site_name=site_name),
'dir_auth': "/www/server/panel/vhost/apache/dir_auth/{site_name}".format(site_name=site_name)
},
'openlitespeed': {
'main': '/www/server/panel/vhost/openlitespeed',
'detail': '/www/server/panel/vhost/openlitespeed/detail',
'listen': '/www/server/panel/vhost/openlitespeed/listen',
'ssl': '/www/server/panel/vhost/openlitespeed/detail/ssl',
},
}
# 备份证书
if os.path.exists(conf_paths['cert']):
public.ExecShell(f"mkdir -p {conf_backup_path}cert/")
public.ExecShell(f"\cp -rpa {conf_paths['cert']} {conf_backup_path}cert/")
# 备份伪静态
if os.path.exists(conf_paths['rewrite']):
public.ExecShell(f"\cp -rpa {conf_paths['rewrite']} {conf_backup_path}rewrite")
rewrite_file_list = os.listdir("/www/server/panel/vhost/rewrite/")
for rewrite_file in rewrite_file_list:
if rewrite_file.endswith(".conf"):
if site_name in rewrite_file:
public.ExecShell(
f"\cp -rpa /www/server/panel/vhost/rewrite/{rewrite_file} {conf_backup_path}rewrite"
)
# 备份nginx配置
nginx_paths = conf_paths['nginx']
if os.path.exists(nginx_paths['main']):
public.ExecShell(f"\cp -rpa {nginx_paths['main']} {conf_backup_path}nginx/")
if not os.path.exists(nginx_paths['main']):
web_conf_list = os.listdir("/www/server/panel/vhost/nginx/")
for web_conf_name in web_conf_list:
if web_conf_name.endswith(".conf"):
if site_name in web_conf_name:
public.ExecShell(
f"\cp -rpa /www/server/panel/vhost/nginx/{web_conf_name} {conf_backup_path}nginx/"
)
if os.path.exists(nginx_paths['redirect']):
public.ExecShell(f"mkdir -p {conf_backup_path}nginx/redirect/{site_name}/")
public.ExecShell(f"\cp -rpa {nginx_paths['redirect']}/* {conf_backup_path}nginx/redirect/{site_name}/")
if os.path.exists(nginx_paths['proxy']):
public.ExecShell(f"mkdir -p {conf_backup_path}nginx/proxy/{site_name}/")
public.ExecShell(f"\cp -rpa {nginx_paths['proxy']}/* {conf_backup_path}nginx/proxy/{site_name}/")
if os.path.exists(nginx_paths['dir_auth']):
public.ExecShell(f"mkdir -p {conf_backup_path}nginx/dir_auth/{site_name}/")
public.ExecShell(f"\cp -rpa {nginx_paths['dir_auth']}/* {conf_backup_path}nginx/dir_auth/{site_name}/")
# 备份apache配置
apache_paths = conf_paths['apache']
if os.path.exists(apache_paths['main']):
public.ExecShell(f"\cp -rpa {apache_paths['main']} {conf_backup_path}apache/")
if not os.path.exists(apache_paths['main']):
web_conf_list = os.listdir("/www/server/panel/vhost/apache/")
for web_conf_name in web_conf_list:
if web_conf_name.endswith(".conf"):
if site_name in web_conf_name:
public.ExecShell(
f"\cp -rpa /www/server/panel/vhost/apache/{web_conf_name} {conf_backup_path}apache/"
)
if os.path.exists(apache_paths['redirect']):
public.ExecShell(f"mkdir -p {conf_backup_path}apache/redirect/{site_name}/")
public.ExecShell(f"\cp -rpa {apache_paths['redirect']}/* {conf_backup_path}apache/redirect/{site_name}/")
if os.path.exists(apache_paths['proxy']):
public.ExecShell(f"mkdir -p {conf_backup_path}apache/proxy/{site_name}/")
public.ExecShell(f"\cp -rpa {apache_paths['proxy']}/* {conf_backup_path}apache/proxy/{site_name}/")
if os.path.exists(apache_paths['dir_auth']):
public.ExecShell(f"mkdir -p {conf_backup_path}apache/dir_auth/{site_name}/")
public.ExecShell(f"\cp -rpa {apache_paths['dir_auth']}/* {conf_backup_path}apache/dir_auth/{site_name}/")
# 备份openlitespeed配置
ols_paths = conf_paths['openlitespeed']
if os.path.exists(ols_paths['main']):
for web_conf_name in os.listdir(ols_paths['main']):
if site_name in web_conf_name:
public.ExecShell(
f"\cp -rpa /www/server/panel/vhost/openlitespeed/{web_conf_name} {conf_backup_path}openlitespeed/"
)
if os.path.exists(conf_paths['openlitespeed']['detail']):
public.ExecShell(f"mkdir -p {conf_backup_path}openlitespeed/detail")
for detail in os.listdir(conf_paths['openlitespeed']['detail']):
if site_name in detail:
public.ExecShell(
f"\cp -rpa {ols_paths['main']}/detail/{detail} {conf_backup_path}openlitespeed/detail/"
)
if os.path.exists(conf_paths['openlitespeed']['listen']):
public.ExecShell(
f"cp -rpa {conf_paths['openlitespeed']['listen']} {conf_backup_path}openlitespeed/listen"
)
if os.path.exists(conf_paths['openlitespeed']['ssl']):
public.ExecShell(f"mkdir -p {conf_backup_path}openlitespeed/detail/ssl")
for ssl in os.listdir(conf_paths['openlitespeed']['ssl']):
if site_name in ssl:
public.ExecShell(
f"cp -rpa {conf_paths['openlitespeed']['ssl']}/{ssl} {conf_backup_path}openlitespeed/detail/ssl/{ssl}"
)
def restore_web_conf(self, site_name: str, conf_backup_path: str) -> None:
"""还原网站配置文件
Args:
site_name: 网站名称
conf_backup_path: 配置文件备份路径
"""
# 定义需要还原的配置文件和路径映射
conf_paths = {
'cert': "/www/server/panel/vhost/cert/{site_name}".format(site_name=site_name),
'rewrite': "/www/server/panel/vhost/rewrite/{site_name}.conf".format(site_name=site_name),
'nginx': {
'main': "/www/server/panel/vhost/nginx/{site_name}.conf".format(site_name=site_name),
'redirect': "/www/server/panel/vhost/nginx/redirect/{site_name}".format(site_name=site_name),
'proxy': "/www/server/panel/vhost/nginx/proxy/{site_name}".format(site_name=site_name)
},
'apache': {
'main': "/www/server/panel/vhost/apache/{site_name}.conf".format(site_name=site_name),
'redirect': "/www/server/panel/vhost/apache/redirect/{site_name}".format(site_name=site_name),
'proxy': "/www/server/panel/vhost/apache/proxy/{site_name}".format(site_name=site_name)
},
'openlitespeed': {
'main': '/www/server/panel/vhost/openlitespeed',
'detail': '/www/server/panel/vhost/openlitespeed/detail',
'listen': '/www/server/panel/vhost/openlitespeed/listen',
'ssl': '/www/server/panel/vhost/openlitespeed/detail/ssl',
},
}
# 还原证书
if os.path.exists(f"{conf_backup_path}cert"):
public.ExecShell(f"\cp -rpa {conf_backup_path}cert {conf_paths['cert']}")
# 还原伪静态
if os.path.exists(f"{conf_backup_path}rewrite"):
public.ExecShell(f"\cp -rpa {conf_backup_path}rewrite {conf_paths['rewrite']}")
# 还原nginx配置
if os.path.exists(f"{conf_backup_path}nginx"):
public.ExecShell(f"\cp -rpa {conf_backup_path}nginx {conf_paths['nginx']['main']}")
if os.path.exists(f"{conf_backup_path}nginx/redirect"):
public.ExecShell(f"\cp -rpa {conf_backup_path}nginx/redirect {conf_paths['nginx']['redirect']}")
if os.path.exists(f"{conf_backup_path}nginx/proxy"):
public.ExecShell(f"\cp -rpa {conf_backup_path}nginx/proxy {conf_paths['nginx']['proxy']}")
# 还原apache配置
if os.path.exists(f"{conf_backup_path}apache"):
public.ExecShell(f"\cp -rpa {conf_backup_path}apache {conf_paths['apache']['main']}")
if os.path.exists(f"{conf_backup_path}apache/redirect"):
public.ExecShell(f"\cp -rpa {conf_backup_path}apache/redirect {conf_paths['apache']['redirect']}")
if os.path.exists(f"{conf_backup_path}apache/proxy"):
public.ExecShell(f"\cp -rpa {conf_backup_path}apache/proxy {conf_paths['apache']['proxy']}")
# 还原openlitespeed配置
if os.path.exists(f"{conf_backup_path}openlitespeed"):
for web_cf_name in os.listdir(f"{conf_backup_path}openlitespeed"):
if site_name in web_cf_name:
public.ExecShell(
f"\cp -rpa {conf_backup_path}openlitespeed/{web_cf_name} {conf_paths['openlitespeed']['main']}/{web_cf_name}"
)
detail_path = f"{conf_backup_path}openlitespeed/detail"
if os.path.exists(detail_path):
if not os.path.exists(conf_paths['openlitespeed']['detail']):
public.ExecShell(f"mkdir -p {conf_paths['openlitespeed']['detail']}")
for detail in os.listdir(detail_path):
if site_name in detail:
public.ExecShell(
f"\cp -rpa {detail_path}/{detail} {conf_paths['openlitespeed']['detail']}/{detail}"
)
listen_path = f"{conf_backup_path}openlitespeed/listen"
if os.path.exists(listen_path):
public.ExecShell(
f"\cp -rpa {listen_path} {conf_paths['openlitespeed']['listen']}/listen"
)
ssl_path = f"{conf_backup_path}openlitespeed/detail/ssl"
if os.path.exists(ssl_path):
if not os.path.exists(conf_paths['openlitespeed']['ssl']):
public.ExecShell(f"mkdir -p {conf_paths['openlitespeed']['ssl']}")
for ssl in os.listdir(ssl_path):
if site_name in ssl:
public.ExecShell(
f"\cp -rpa {ssl_path}/{ssl} {conf_paths['openlitespeed']['ssl']}/{ssl}"
)
def _restore_site_db_data(self, site_db_record: dict) -> int:
sql = db.Sql()
sql.table('sites')
if 'id' in site_db_record:
del site_db_record['id']
# SQL关键字字段
if 'index' in site_db_record:
del site_db_record['index']
if_exist = sql.where(
'name=? AND project_type=?',
(site_db_record['name'], site_db_record['project_type'])
).find()
if if_exist:
return if_exist['id']
# insert db record
try:
new_id = sql.insert(site_db_record)
return new_id
except Exception as e:
raise public.lang("Site database insert failed: {}").format(str(e))
def _restore_site_domian_db_data(self, pid: int, domains: list) -> None:
domain_sql = db.Sql()
domain_sql.table('domain')
for domain in domains:
try:
if not domain_sql.where('name=?', (domain['name'],)).count():
domain_sql.add(
'pid, name, port, addtime',
(pid, domain['name'], int(domain['port']), public.getDate())
)
except Exception as e:
public.print_log("Domain database insert failed: {}".format(str(e)))
continue
def _backup_site(self, site: dict, backupPath: str) -> None:
try:
if site.get("project_type", "").lower() == "php":
find = public.M('sites').where("id=?", (site['id'],)).field('name,path,id').find()
fileName = find['name'] + '_' + time.strftime(
'%Y%m%d_%H%M%S', time.localtime()
) + '.zip'
zipName = backupPath + '/' + fileName
if not (os.path.exists(backupPath)):
os.makedirs(backupPath)
tmps = '/tmp/panelExec.log'
execStr = f"cd '{find['path']}' && zip '{zipName}' . -x .user.ini > {tmps} 2>&1"
public.ExecShell(execStr)
public.M('backup').add(
'type,name,pid,filename,size,addtime',
(0, fileName, find['id'], zipName, 0, public.getDate())
)
elif "wp" in site.get("project_type", "").lower():
bak_obj = wpbackup(int(site['id']))
bak_obj.backup_full()
except:
pass
def restore_site_data(self, timestamp: str) -> None:
"""还原站点数据
Args:
timestamp: 备份时间戳
"""
restore_data = self.get_restore_data_list(timestamp)
site_backup_path = self.base_path + "/{timestamp}_backup/site/".format(timestamp=timestamp)
# 还原site环境配置, 全局配置
if not restore_data.get("restore_selection"):
self.restore_site_config(site_backup_path)
if not os.path.exists(site_backup_path):
self.print_log(public.lang("Site backup directory does not exist: {}").format(site_backup_path), 'restore')
return
self.print_log("====================================================", "restore")
self.print_log(public.lang("Start restoring site data"), 'restore')
backupPath = public.M('config').where('id=?', (1,)).getField('backup_path')
backupPath = backupPath + '/site/' if backupPath else "/www/backup/site/"
with app.app_context():
for site in restore_data['data_list']['site']:
log_str = public.lang("Restoring {} project: {}").format(site.get("project_type"), site.get("name"))
try:
site_name = site['name']
site['restore_status'] = 1
self.update_restore_data_list(timestamp, restore_data)
self.print_log(log_str, 'restore')
if self.overwrite:
# site backup if overwrite
self._backup_site(site, backupPath)
# data
if not self.overwrite and 'site_db_record' in site:
site_id = self._restore_site_db_data(site['site_db_record'])
if site_id and 'domains' in site:
# 还原域名记录
self._restore_site_domian_db_data(site_id, site['domains'])
# site file
site_path = str(site['path']).rstrip('/')
last_path: str = os.path.basename(site_path) if site['last_path'] == '' else site['last_path'] # site name
# site abs path
site_zip = site_backup_path + last_path + ".zip"
if os.path.exists(site_zip):
public.ExecShell(f"cd {site_backup_path} && unzip -o {last_path}.zip")
site_data_path = site_backup_path + last_path # site unzip file
if os.path.exists(site_data_path):
site_parent_path = os.path.dirname(site_path) # /www/wwwroot
if not os.path.exists(site_parent_path):
public.ExecShell("mkdir -p {}".format(site_parent_path))
public.ExecShell("chown -R www:www {}".format(site_parent_path))
public.ExecShell("chmod -R 755 {}".format(site_parent_path))
src_site = os.path.join(site_backup_path, last_path)
dst_site = os.path.join(site_parent_path, last_path)
public.print_log('copying site directory from {} to {}'.format(src_site, dst_site))
self.copy_directory(
src=src_site,
dst=dst_site,
overwrite=self.overwrite,
)
try:
shutil.rmtree(src_site)
except Exception as e:
public.print_log(public.lang("Failed to delete source site directory: {}").format(str(e)))
# makesure
public.ExecShell(f"chown -R www:www {dst_site}")
user_ini = dst_site + "/.user.ini"
if not os.path.exists(user_ini) or self.overwrite:
public.writeFile(user_ini, f"open_basedir={site_parent_path}/:/tmp/")
public.ExecShell("chmod 644 " + user_ini)
public.ExecShell("chown root:root " + user_ini)
public.ExecShell("chattr +i " + user_ini)
# site config
site_conf_zip = site_backup_path + site_name + "_conf.zip"
if os.path.exists(site_conf_zip):
public.ExecShell(
"cd {site_backup_path} && unzip -o {site_name}_conf.zip".format(
site_backup_path=site_backup_path, site_name=site_name
)
)
public.ExecShell(
"cd {site_backup_path} && \cp -rpa {site_name}_conf/* /www/server/panel/vhost".format(
site_backup_path=site_backup_path, site_name=site_name
)
)
new_log_str = public.lang("{} project: {} ✓").format(site['project_type'], site['name'])
self.replace_log(log_str, new_log_str, 'restore')
site['restore_status'] = 2
self.update_restore_data_list(timestamp, restore_data)
except Exception as e:
site['restore_status'] = 3
self.update_restore_data_list(timestamp, restore_data)
new_log_str = public.lang(f"{site['project_type']} project: {site['name']} Reason: {str(e)}")
self.replace_log(log_str, new_log_str, 'restore')
continue
self.print_log(public.lang("Site data restoration completed"), 'restore')
# 还原site 所有Python环境
# self.restore_site_python_env(timestamp)
def backup_site_dir_auth(self, site_name: str):
if os.path.exists(self.site_dir_auth_path):
site_dir_auth_data = json.loads(public.ReadFile(self.site_dir_auth_path))
if site_name in site_dir_auth_data:
result = {site_name: site_dir_auth_data[site_name]}
return result
return False
def restore_site_dir_auth(self, site_name: str, backup_data_path: str):
if os.path.exists(backup_data_path):
dir_auth_backup_data = json.loads(public.ReadFile(backup_data_path))
if os.path.exists(self.site_dir_auth_path):
site_dir_auth_data = json.loads(public.ReadFile(self.site_dir_auth_path))
site_dir_auth_data[site_name] = dir_auth_backup_data[site_name]
public.WriteFile(self.site_dir_auth_path, json.dumps(site_dir_auth_data))
def backup_dir_pass(self, site_name: str, backup_data_path: str):
if os.path.exists(self.site_dir_auth_path):
site_dir_auth_data = json.loads(public.ReadFile(self.site_dir_auth_path))
if site_name in site_dir_auth_data:
result = {site_name: site_dir_auth_data[site_name]}
return result
return {}
def backup_redirect_conf(self, site_name: str):
if os.path.exists(self.redirect_conf_path):
redirect_conf_data = json.loads(public.ReadFile(self.redirect_conf_path))
for item in redirect_conf_data:
if site_name in item['sitename']:
return item
return False
def restore_redirect_conf(self, site_name: str, backup_data_path: str):
if os.path.exists(backup_data_path):
redirect_conf_data = json.loads(public.ReadFile(backup_data_path))
local_redirect_conf_data = []
if os.path.exists(self.redirect_conf_path):
local_redirect_conf_data = json.loads(public.ReadFile(self.redirect_conf_path))
data_exists = None
for item in local_redirect_conf_data:
if item['sitename'] == redirect_conf_data['sitename']:
data_exists = True
if not data_exists:
local_redirect_conf_data.append(redirect_conf_data)
public.WriteFile(self.redirect_conf_path, json.dumps(local_redirect_conf_data))
return False
def backup_proxy_conf(self, site_name: str):
if os.path.exists(self.proxy_conf_path):
proxy_conf_data = json.loads(public.ReadFile(self.proxy_conf_path))
for item in proxy_conf_data:
if site_name in item['sitename']:
return item
return False
def restore_proxy_conf(self, site_name: str, backup_data_path: str):
if os.path.exists(backup_data_path):
proxy_conf_data = json.loads(public.ReadFile(backup_data_path))
local_proxy_conf_data = []
if os.path.exists(self.proxy_conf_path):
local_proxy_conf_data = json.loads(public.ReadFile(self.proxy_conf_path))
data_exists = None
for item in local_proxy_conf_data:
if item['sitename'] == proxy_conf_data['sitename']:
data_exists = True
if not data_exists:
local_proxy_conf_data.append(proxy_conf_data)
public.WriteFile(self.proxy_conf_path, json.dumps(local_proxy_conf_data))
return False
if __name__ == '__main__':
# 获取命令行参数
if len(sys.argv) < 3:
print("Usage: btpython backup_manager.py <method> <timestamp>")
sys.exit(1)
method_name = sys.argv[1] # 方法名
timestamp = sys.argv[2] # IP地址
site_module = SiteModule() # 实例化对象
if hasattr(site_module, method_name): # 检查方法是否存在
method = getattr(site_module, method_name) # 获取方法
method(timestamp) # 调用方法
else:
print(f"Error: Method '{method_name}' 'does not exist'")
Reference in New Issue View Git Blame Copy Permalink