258 lines
9.7 KiB
Python
258 lines
9.7 KiB
Python
# coding: utf-8
|
|
# -------------------------------------------------------------------
|
|
# YakPanel
|
|
# -------------------------------------------------------------------
|
|
# Copyright (c) 2015-2099 YakPanel(www.yakpanel.com) All rights reserved.
|
|
# -------------------------------------------------------------------
|
|
# Author: csj <csj@yakpanel.com>
|
|
# -------------------------------------------------------------------
|
|
# ------------------------------
|
|
# docker应用商店 apphub 业务类
|
|
# ------------------------------
|
|
import json
|
|
import os
|
|
import sys
|
|
import re
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
|
|
if "/www/server/panel/class" not in sys.path:
|
|
sys.path.insert(0, "/www/server/panel/class")
|
|
|
|
import public
|
|
from mod.base.git_tool import GitTool
|
|
from mod.project.docker.app.base import App
|
|
import mod.base.git_tool.install as GitInstall
|
|
|
|
class AppHub():
|
|
_instance = None
|
|
hub_config_path = os.path.join(App.dk_project_path,'dk_app','apphub_config.json') #/www/dk_project/dk_app/apphub_config.json
|
|
hub_home_path = os.path.join(App.dk_project_path,'dk_app','apphub','apphub') #/www/dk_project/dk_app/apphub/apphub
|
|
hub_apps_path = os.path.join(hub_home_path,'apps.json') #/www/dk_project/dk_app/apphub/apphub/apps.json
|
|
|
|
def __new__(cls, *args, **kwargs):
|
|
if not hasattr(cls, "_instance") or cls._instance is None:
|
|
cls._instance = super(AppHub, cls).__new__(cls)
|
|
return cls._instance
|
|
|
|
@classmethod
|
|
def get_config(cls):
|
|
'''
|
|
@name 获取外部应用配置
|
|
'''
|
|
if not os.path.exists(cls.hub_config_path):
|
|
apphub_config = {
|
|
"git_config": {
|
|
"git_url": "",
|
|
"git_branch": "main",
|
|
"user_config": {
|
|
"name": "",
|
|
"password": ""
|
|
}
|
|
}
|
|
}
|
|
public.writeFile(cls.hub_config_path, json.dumps(apphub_config))
|
|
return json.loads(public.readFile(cls.hub_config_path))
|
|
|
|
def install_apphub(self,get):
|
|
|
|
git_install = GitInstall.install_git()
|
|
if not git_install:
|
|
return public.return_message(-1, 0, public.lang("If the installation fails, check the network or install git manually"))
|
|
|
|
return public.return_message(0, 0, public.lang("The environment was successfully installed"))
|
|
|
|
def get_hub_apps(self):
|
|
'''
|
|
@name 获取外部应用列表
|
|
'''
|
|
res = []
|
|
try:
|
|
if os.path.exists(AppHub.hub_apps_path):
|
|
res=json.loads(public.readFile(self.hub_apps_path))
|
|
except:
|
|
pass
|
|
return res
|
|
|
|
def generate_apphub(self, get):
|
|
'''
|
|
@name 解析外部应用列表
|
|
'''
|
|
apps = []
|
|
if not os.path.isdir(self.hub_home_path):
|
|
return public.return_message(-1, 0, public.lang("The apphub directory does not exist"))
|
|
for name in os.listdir(self.hub_home_path):
|
|
app_dir = os.path.join(self.hub_home_path, name)
|
|
if not os.path.isdir(app_dir): continue
|
|
|
|
app_info=public.readFile(os.path.join(app_dir, 'app.json'))
|
|
if not app_info: continue
|
|
|
|
try:
|
|
app_info = json.loads(app_info)
|
|
except Exception as e:
|
|
continue
|
|
|
|
if "reuse" not in app_info: app_info["reuse"] = True
|
|
if "icon" not in app_info: app_info["icon"] = ""
|
|
if "sort" not in app_info: app_info["sort"] = 999
|
|
if "cpu" not in app_info: app_info["cpu"] = 0
|
|
if "mem" not in app_info: app_info["mem"] = 0
|
|
if "disk" not in app_info: app_info["disk"] = 10240
|
|
if "installed" not in app_info: app_info["installed"] = False
|
|
if "updateat" not in app_info: app_info["updateat"] = 0
|
|
|
|
apps.append(app_info)
|
|
|
|
self.apphub_apps = apps
|
|
|
|
public.writeFile(self.hub_apps_path, json.dumps(apps, indent=4, ensure_ascii=False))
|
|
|
|
self.generate_apphub_icon()
|
|
|
|
return public.return_message(0, 0, public.lang("The resolution was successful"))
|
|
|
|
def generate_apphub_icon(self):
|
|
'''
|
|
@name 创建外部应用图标
|
|
#/static/img/soft_ico/apphub/ico-apphub_xxx.png
|
|
'''
|
|
apphub_ico_path = "{}/YakPanel/static/img/soft_ico/apphub/".format(public.get_panel_path())
|
|
if os.path.exists(apphub_ico_path):
|
|
public.ExecShell("rm -rf {}".format(apphub_ico_path))
|
|
public.ExecShell("mkdir -p {}".format(apphub_ico_path))
|
|
|
|
for name in os.listdir(self.hub_home_path):
|
|
app_dir = os.path.join(self.hub_home_path, name,'icon.png')
|
|
if not os.path.exists(app_dir): continue
|
|
|
|
app_icon_path = os.path.join(apphub_ico_path, "ico-apphub_{}.png".format(name))
|
|
public.ExecShell("cp {} {}".format(app_dir,app_icon_path))
|
|
return True
|
|
|
|
def set_apphub_git(self, get):
|
|
'''
|
|
@name 设置git配置
|
|
@param get: git_url, git_branch, user, password
|
|
'''
|
|
config = self.get_config()
|
|
git_config = config.get("git_config", {})
|
|
git_config["git_url"] = get.git_url.strip()
|
|
git_config["git_branch"] = get.git_branch.strip()
|
|
if "name" in get and "password" in get:
|
|
git_config["user_config"] = {
|
|
"name": get.get("name", ""),
|
|
"password": get.get("password", "")
|
|
}
|
|
config["git_config"] = git_config
|
|
public.writeFile(self.hub_config_path, json.dumps(config, indent=4, ensure_ascii=False))
|
|
return public.return_message(0, 0, public.lang("GIT INFORMATION IS SUCCESSFULLY CONFIGURED"))
|
|
|
|
def import_git_apphub(self,get):
|
|
'''
|
|
@name 从git导入外部应用
|
|
@param None
|
|
'''
|
|
if not GitInstall.installed():
|
|
return public.return_message(-1, 0, public.lang("If you don't have a git environment, please install git first"))
|
|
|
|
abs_path = os.path.dirname(self.hub_home_path)
|
|
if not os.path.exists(abs_path): os.makedirs(abs_path)
|
|
|
|
gitconfig = self.get_config()
|
|
if not gitconfig or not gitconfig.get("git_config", {}).get("git_url", ""):
|
|
return public.return_message(-1, 0, public.lang("Please set the git information first"))
|
|
|
|
git_url = gitconfig.get("git_config", {}).get("git_url", "")
|
|
git_user = gitconfig.get("git_config", {}).get("user_config", {})
|
|
git_branch = gitconfig.get("git_config", {}).get("git_branch", {})
|
|
|
|
git = GitTool(project_path=abs_path, git_url=git_url,user_config=git_user,git_id="-1")
|
|
public.ExecShell("rm -rf /tmp/git_-1_log.log")
|
|
res = git.pull(git_branch)
|
|
|
|
if res is not None:
|
|
return public.return_message(-1, 0, public.lang("Import from git failed"))
|
|
|
|
#解析全部应用
|
|
res = self.generate_apphub(get)
|
|
if not res["status"]:
|
|
return public.return_message(-1, 0, res["msg"])
|
|
#删除模板
|
|
public.ExecShell("rm -rf {}".format(os.path.join(AppHub.hub_home_path, "templates")))
|
|
|
|
public.set_module_logs('apphub', 'import_git_apphub', 1)
|
|
return public.return_message(0, 0, public.lang("Import from git successful"))
|
|
|
|
def import_zip_apphub(self, get):
|
|
'''
|
|
@name 从压缩到包导入外部应用
|
|
@param get: sfile: zip文件路径
|
|
'''
|
|
sfile = get.sfile.strip()
|
|
files = sfile.split(",")
|
|
|
|
for sfile in files:
|
|
|
|
if not sfile.endswith(('.zip', '.gz')):
|
|
return public.return_message(-1, 0, public.lang("The file format is incorrect, please select the zip or gz file"))
|
|
|
|
if not os.path.exists(self.hub_home_path):
|
|
os.makedirs(self.hub_home_path)
|
|
|
|
if sfile.endswith('.zip'):
|
|
res, err = public.ExecShell("unzip -o {} -d {}".format(sfile, self.hub_home_path))
|
|
elif sfile.endswith('.gz'):
|
|
res, err = public.ExecShell("tar -xzvf {} -C {}".format(sfile, self.hub_home_path))
|
|
else:
|
|
err = "{},Unsupported file formats".format(sfile)
|
|
|
|
if err:
|
|
return public.return_message(-1, 0, public.lang("Import failure:{}", str(err)))
|
|
|
|
res = self.generate_apphub(get)
|
|
if not res["status"]:
|
|
return public.return_message(-1, 0, res["msg"])
|
|
|
|
public.set_module_logs('apphub', 'import_zip_apphub', 1)
|
|
|
|
return public.return_message(0, 0, public.lang("Successful import"))
|
|
|
|
def parser_zip_apphub(self, get):
|
|
'''
|
|
@name 解析zip包
|
|
@param get: sfile: zip文件路径
|
|
@return app_list: 外部应用列表
|
|
'''
|
|
sfile = get.sfile.strip()
|
|
|
|
app_list = []
|
|
|
|
from mod.project.docker.apphub.tool import GzHandler, ZipHandler
|
|
if sfile.endswith(".gz"):
|
|
handler = GzHandler()
|
|
else:
|
|
handler = ZipHandler()
|
|
|
|
files = handler.get_files(sfile)
|
|
|
|
if 'status' in files:
|
|
return public.return_message(-1, 0, files['msg'])
|
|
|
|
for file, file_struck in files.items():
|
|
if 'app.json' in file_struck and file_struck['app.json']['is_dir'] == 0:
|
|
|
|
filename = file_struck['app.json']['fullpath']
|
|
|
|
appinfo = handler.get_file_info(sfile, filename)
|
|
if 'status' in appinfo and appinfo['status'] == False:
|
|
return public.return_message(-1, 0, appinfo['msg'])
|
|
|
|
try:
|
|
appinfo = json.loads(appinfo['data'])
|
|
appinfo["parser_from"] = sfile
|
|
app_list.append(appinfo)
|
|
except:
|
|
pass
|
|
|
|
return app_list |