Spaces:

SZFSPC
/

page

Paused

page / bt-source /panel /class /sslModel /autodeployModel.py

feat: deploy to hf space | model=qwen-80241b

ab7d893 verified 5 months ago

14.8 kB

import os, sys, json, psutil
import time

os.chdir('/www/server/panel')
if not 'class/' in sys.path:
    sys.path.insert(0,'class/')

import public

from sslModel.base import sslBase


class main(sslBase):
    __APIURL2 = public.GetConfigValue('home') + '/api/v2/synchron'
    # __APIURL2 = "https://dev.bt.cn" + '/api/v2/synchron'
    __PDATA = None
    _check_url = None
    __request_url = None
    __UPATH = 'data/userInfo.json'
    __userInfo = None
    def __init__(self):
        super().__init__()
        if not os.path.exists(public.get_panel_path() + '/data/auto_deploy'):
            os.makedirs(public.get_panel_path() + '/data/auto_deploy')

        # 本地任务表
        self.M("").execute("""CREATE TABLE IF NOT EXISTS `autodeploy_task` (
          `id` INTEGER  PRIMARY KEY AUTOINCREMENT,        
          `task_name` varchar(320) NOT NULL,    -- 任务名称
          `cloud_id` INTEGER NOT NULL,    -- 云端任务ID
          `access_id` INTEGER NOT NULL,    -- 接入ID
          `private_key` varchar(320) NOT NULL,    -- 密钥
          `sites` varchar(320) NOT NULL,    -- 网站
          `cycle` INTEGER NOT NULL DEFAULT 12,    -- 检测周期
          `deploy_time` INTEGER NOT NULL DEFAULT 0,    -- 最近部署时间
          `deploy_status` INTEGER NOT NULL DEFAULT 0,    -- 最近部署状态 0等待部署 1=部署成功 -1部署失败
          `ssl_hash` varchar(320) NOT NULL DEFAULT '',   -- 证书hash
          `ssl_cloud_id` varchar(320) NOT NULL DEFAULT ''   -- 云端证书id
          );""", ())

        self.get_user_info()

        # 云端任务表
    def request(self, dname):

        request_url = self.__APIURL2 + '/' + dname
        self.__request_url = public.get_home_node(request_url)
        msg = '接口请求失败（{}）'.format(self.__request_url)
        try:

            res = public.httpPost(request_url, self.__PDATA)
            if res == False:
                return public.returnMsg(False, '宝塔云端服务器不可达（{}），请检查网络或稍后重试。'.format(self.__request_url))
        except Exception as ex:
            return public.returnMsg(False, '宝塔云端服务器不可达: ' + str(ex))

        result = public.returnMsg(False, msg)
        try:
            result = json.loads(res)
        except:
            pass
        return result

    def get_user_info(self):
        upath = public.get_panel_path() + "/" + self.__UPATH
        try:
            self.__userInfo = json.loads(public.readFile(upath))
        except:
            self.__userInfo = {}
        self.__PDATA = self.__userInfo

    def M(self, table_name):
        import db
        sql = db.Sql()
        sql._Sql__DB_FILE = public.get_panel_path() + '/data/db/ssl_data.db'
        sql._Sql__encrypt_keys = []
        return sql.table(table_name)

    def plane_synchron_list(self, get):
        res = self.request('plane_synchron_list')
        if not res["success"]:
            public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def plane_synchron_access_list(self):
        # path = public.get_panel_path() + '/data/auto_deploy/plane_synchron_list.json'
        # try:
        #     data = json.loads(public.readFile(path))
        #     return public.returnMsg(True, data)
        # except:
        #     pass
        res = self.request('plane_synchron_access_list')

        if not res["success"]:
            public.returnMsg(False, res["res"])
        # public.writeFile(path, json.dumps(res["res"]))
        return public.returnMsg(True, res["res"])

    def plane_synchron_access_set(self, get):
        self.__PDATA["key"] = get.private_key
        self.__PDATA["synid"] = int(get.cloud_id)
        # self.__PDATA["url"] = "https://www.bt.cn"
        res = self.request('plane_synchron_access_set')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def get_task_list(self, get=None):

        p = 1
        if 'p' in get:
            p = int(get.p)
        collback = ''
        if 'collback' in get:
            collback = get.collback
        limit = 999999999
        if 'limit' in get:
            limit = int(get.limit)
        f_sql = '1=1'
        if 'search' in get and get.search:
            f_sql = "task_name like '%{}%' or sites like '%{}%' ".format(get.search, get.search)

        count = self.M("autodeploy_task").where(f_sql, ()).count()
        page_data = public.get_page(count, p, limit, collback)

        data = self.M("autodeploy_task").where(f_sql, ()).limit(page_data["shift"]+","+page_data["row"]).order("id desc").select()
        # public.print_log(data)

        try:
            cloud_data = self.plane_synchron_access_list()
        except:
            cloud_data = None
        # public.print_log(cloud_data)

        deploy = []
        for i in data:
            i["access_state"] = 0
            i["synchron_info"] = {}
            if not cloud_data:
                continue
            for j in cloud_data["msg"]:
                if i["access_id"] == j["id"]:
                    i["access_state"] = j["access_state"]
                    if j["access_state"] == 2 and str(i["ssl_cloud_id"]) != str(j["synchron_info"]["ca_id"]):
                        self.M("autodeploy_task").where("cloud_id=?", (i["cloud_id"],)).update({"ssl_cloud_id": j["synchron_info"]["ca_id"]})
                        # 写需要更新证书的任务到队列
                        deploy.append(i["id"])
                    i["synchron_info"] = j["synchron_info"]
        if deploy:
            public.ExecShell("nohup btpython {}/class/sslModel/autodeployModel.py --task_ids={} > /dev/null 2>&1 &".format(public.get_panel_path(), ",".join(str(i) for i in deploy)))
        page_data["data"] = data

        return {"status": True, "msg": page_data, "deploy": deploy}

    def add_task(self, get):
        task_name = get.task_name
        cloud_id = get.cloud_id
        private_key = get.private_key
        sites = get.sites
        # cycle = get.cycle
        cycle = 1
        data = self.M("autodeploy_task").where("cloud_id=?", (cloud_id,)).find()
        if data:
            return public.returnMsg(False, "选择的证书部署任务在本服务器已存在对应的任务【{}】请勿重复添加！".format(data["task_name"]))

        res = self.plane_synchron_access_set(get)
        if not res["status"]:
            return public.returnMsg(False, res["msg"])

        self.M("autodeploy_task").add('task_name,cloud_id,access_id,private_key,sites,cycle', (task_name, cloud_id, res["msg"]["id"], private_key, sites, cycle))
        if not res["status"]:
            return public.returnMsg(False, res["msg"])
        return public.returnMsg(True, "添加成功")

    def plane_synchron_access_delte(self, get):
        self.__PDATA["id"] = int(get.access_id)
        res = self.request('plane_synchron_access_delte')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def del_task(self, get):
        data = self.M("autodeploy_task").where("id=?", (get.task_id,)).find()
        if not data:
            return public.returnMsg(False, "未找到任务")
        self.M("autodeploy_task").where("id=?", (get.task_id,)).delete()
        self.plane_synchron_access_delte(public.to_dict_obj({"access_id": data["access_id"]}))
        return public.returnMsg(True, "删除成功")

    def batch_del_task(self, get):
        task_id_list = get.task_ids.split(",")
        data = self.M("autodeploy_task").where("id in ({})".format(get.task_ids), ()).select()
        self.M("autodeploy_task").where("id in ({})".format(get.task_ids), ()).delete()

        total = len(data)
        faild = 0
        success = 0
        faildList = []
        successList = []
        for i in data:
            if str(i["id"]) not in task_id_list:
                faild += 1
                faildList.append({"task_id": i["id"], "task_name":i["task_name"],"error_msg": "任务不存在", "status": False})
                continue
            self.plane_synchron_access_delte(public.to_dict_obj({"access_id": i["access_id"]}))
            success += 1
            successList.append({"task_id": i["id"], "task_name":i["task_name"],"error_msg": "删除成功", "status": True})

        return public.returnMsg(True, {"total": total, "success": success, "faild": faild, "faildList": faildList, "successList": successList})

    def edit_task(self, get):
        task_id = get.task_id
        task_name = get.task_name
        cloud_id = int(get.cloud_id)
        private_key = get.private_key
        sites = get.sites
        # cycle = get.cycle
        cycle = 1

        data = self.M("autodeploy_task").where("id=?", (task_id,)).find()
        if not data:
            return public.returnMsg(False, "未找到任务")
        update_data = {"task_name": task_name, "cloud_id": cloud_id, "private_key": private_key, "sites": sites, "cycle": cycle}

        if data["cloud_id"] != cloud_id:
            cloud_data = self.M("autodeploy_task").where("cloud_id=?", (cloud_id)).find()
            if cloud_data:
                return public.returnMsg(False, "选择的证书部署任务在本服务器已存在对应的任务【{}】请勿重复添加！".format(cloud_data["task_name"]))
            res = self.plane_synchron_access_set(get)
            if not res["status"]:
                return public.returnMsg(False, res["msg"])
            update_data["access_id"] = res["msg"]["id"]
        elif data["private_key"] != private_key:
            res = self.plane_synchron_access_set(get)
            if not res["status"]:
                return public.returnMsg(False, res["msg"])

        if data["sites"] != sites:
            update_data["ssl_cloud_id"] = ""

        self.M("autodeploy_task").where("id=?", (get.task_id,)).update(update_data)
        return public.returnMsg(True, "修改成功")

    def plane_synchron_get_cert(self, access_id):
        self.__PDATA["id"] = int(access_id)
        res = self.request('plane_synchron_get_cert')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def set_cert_to_database(self, access_id):
        cert_data = self.plane_synchron_get_cert(access_id)
        if not cert_data["status"]:
            return public.returnMsg(False, cert_data["msg"])
        cert_data = cert_data["msg"]["cert"]

        from sslModel import certModel
        get = public.dict_obj()
        get.key = cert_data["privateCert"]
        get.csr = cert_data["cert"]

        try:
            cert = certModel.main().save_cert(get)
            if not cert["status"]:
                return False
        except:
            return False
        self.M("autodeploy_task").where("access_id=?", (access_id,)).update({"ssl_hash": cert["ssl_hash"]})
        return True

    def deploy_cert(self, get):
        data = self.M("autodeploy_task").where("id=?", (int(get.task_id),)).find()
        if not data:
            return public.returnMsg(False, "未找到任务")
        sites = data["sites"].split(",")
        ssl_hash = data["ssl_hash"]
        get.BatchInfo = json.dumps([
            {"ssl_hash": ssl_hash, "siteName": i}
            for i in sites
        ])
        import panelSSL
        res = panelSSL.panelSSL().SetBatchCertToSite(get)

        deploy_time = int(time.time())
        if res["total"] == res["success"]:
            self.write_deploy_detail(get.task_id, json.dumps(res), data["access_id"], 1, 1)
            deploy_status = 1
        else:
            self.write_deploy_detail(get.task_id, json.dumps(res), data["access_id"], -1, 1)
            deploy_status = -1
        self.M("autodeploy_task").where("id=?", (int(get.task_id),)).update({"deploy_time": deploy_time, "deploy_status": deploy_status})

    def upload_deploy_msg(self, get):
        self.__PDATA["msg"] = get.error_msg
        self.__PDATA["id"] = get.access_id
        self.__PDATA["state"] = get.state
        self.__PDATA["type"] = get.msg_type

        # print(self.__PDATA)

        res = self.request('plane_synchron_deploy_state')
        if not res["success"]:
            return public.returnMsg(False, res["res"])
        return public.returnMsg(True, res["res"])

    def write_deploy_detail(self, task_id, error_msg, access_id, state, msg_type):
        path = public.get_panel_path() + '/data/auto_deploy/deploy_detail'
        if not os.path.exists(path):
            os.makedirs(path)
        public.writeFile(path + "/{}.json".format(task_id), error_msg)
        # print(access_id, state, msg_type)
        res = self.upload_deploy_msg(public.to_dict_obj({"access_id": access_id, "error_msg": error_msg, "state": state, "msg_type": msg_type}))
        # print(res)

    def get_task_deploy_detail(self, get):
        try:
            res = json.loads(public.readFile(public.get_panel_path() + '/data/auto_deploy/deploy_detail/{}.json'.format(get.task_id)))
        except:
            res = None
        return public.returnMsg(True, res)



if __name__ == '__main__':
    try:
        if os.path.exists("/www/server/panel/data/auto_deploy.pl"):
            exit()
        public.writeFile("/www/server/panel/data/auto_deploy.pl", "")
        m = main()
        import argparse
        p = argparse.ArgumentParser()
        p.add_argument('--task_ids', default=None, help="任务id列表，用逗号分隔", dest="task_ids")
        args = p.parse_args()
        if not args.task_ids:
            exit()
        task_ids = args.task_ids.split(",")
        for i in task_ids:
            task_data = m.M("autodeploy_task").where("id=?", (int(i),)).find()
            if not task_data:
                continue
            set_cert_status = m.set_cert_to_database(task_data["access_id"])
            if not set_cert_status:
                sites = task_data["sites"].split(",")
                error_msg = json.dumps({
                    "total": len(sites),
                    "success": 0,
                    "faild": len(sites),
                    "faildList": [{"siteName": site, "error_msg": "证书保存失败", "status": False} for site in sites],
                    "successList": []
                })
                m.write_deploy_detail(i, error_msg, task_data["access_id"], -1, 1)
                m.M("autodeploy_task").where("id=?", (int(i),)).update(
                    {"deploy_time": int(time.time()), "deploy_status": -1})
                continue
            m.deploy_cert(public.to_dict_obj({"task_id": i}))
            print("部署成功：", task_data["task_name"])
        exit()
    finally:
        os.remove("/www/server/panel/data/auto_deploy.pl")

	import os, sys, json, psutil
	import time

	os.chdir('/www/server/panel')
	if not 'class/' in sys.path:
	sys.path.insert(0,'class/')

	import public

	from sslModel.base import sslBase


	class main(sslBase):
	__APIURL2 = public.GetConfigValue('home') + '/api/v2/synchron'
	# __APIURL2 = "https://dev.bt.cn" + '/api/v2/synchron'
	__PDATA = None
	_check_url = None
	__request_url = None
	__UPATH = 'data/userInfo.json'
	__userInfo = None
	def __init__(self):
	super().__init__()
	if not os.path.exists(public.get_panel_path() + '/data/auto_deploy'):
	os.makedirs(public.get_panel_path() + '/data/auto_deploy')

	# 本地任务表
	self.M("").execute("""CREATE TABLE IF NOT EXISTS `autodeploy_task` (
	`id` INTEGER PRIMARY KEY AUTOINCREMENT,
	`task_name` varchar(320) NOT NULL, -- 任务名称
	`cloud_id` INTEGER NOT NULL, -- 云端任务ID
	`access_id` INTEGER NOT NULL, -- 接入ID
	`private_key` varchar(320) NOT NULL, -- 密钥
	`sites` varchar(320) NOT NULL, -- 网站
	`cycle` INTEGER NOT NULL DEFAULT 12, -- 检测周期
	`deploy_time` INTEGER NOT NULL DEFAULT 0, -- 最近部署时间
	`deploy_status` INTEGER NOT NULL DEFAULT 0, -- 最近部署状态 0等待部署 1=部署成功 -1部署失败
	`ssl_hash` varchar(320) NOT NULL DEFAULT '', -- 证书hash
	`ssl_cloud_id` varchar(320) NOT NULL DEFAULT '' -- 云端证书id
	);""", ())

	self.get_user_info()

	# 云端任务表
	def request(self, dname):

	request_url = self.__APIURL2 + '/' + dname
	self.__request_url = public.get_home_node(request_url)
	msg = '接口请求失败（{}）'.format(self.__request_url)
	try:

	res = public.httpPost(request_url, self.__PDATA)
	if res == False:
	return public.returnMsg(False, '宝塔云端服务器不可达（{}），请检查网络或稍后重试。'.format(self.__request_url))
	except Exception as ex:
	return public.returnMsg(False, '宝塔云端服务器不可达: ' + str(ex))

	result = public.returnMsg(False, msg)
	try:
	result = json.loads(res)
	except:
	pass
	return result

	def get_user_info(self):
	upath = public.get_panel_path() + "/" + self.__UPATH
	try:
	self.__userInfo = json.loads(public.readFile(upath))
	except:
	self.__userInfo = {}
	self.__PDATA = self.__userInfo

	def M(self, table_name):
	import db
	sql = db.Sql()
	sql._Sql__DB_FILE = public.get_panel_path() + '/data/db/ssl_data.db'
	sql._Sql__encrypt_keys = []
	return sql.table(table_name)

	def plane_synchron_list(self, get):
	res = self.request('plane_synchron_list')
	if not res["success"]:
	public.returnMsg(False, res["res"])
	return public.returnMsg(True, res["res"])

	def plane_synchron_access_list(self):
	# path = public.get_panel_path() + '/data/auto_deploy/plane_synchron_list.json'
	# try:
	# data = json.loads(public.readFile(path))
	# return public.returnMsg(True, data)
	# except:
	# pass
	res = self.request('plane_synchron_access_list')

	if not res["success"]:
	public.returnMsg(False, res["res"])
	# public.writeFile(path, json.dumps(res["res"]))
	return public.returnMsg(True, res["res"])

	def plane_synchron_access_set(self, get):
	self.__PDATA["key"] = get.private_key
	self.__PDATA["synid"] = int(get.cloud_id)
	# self.__PDATA["url"] = "https://www.bt.cn"
	res = self.request('plane_synchron_access_set')
	if not res["success"]:
	return public.returnMsg(False, res["res"])
	return public.returnMsg(True, res["res"])

	def get_task_list(self, get=None):

	p = 1
	if 'p' in get:
	p = int(get.p)
	collback = ''
	if 'collback' in get:
	collback = get.collback
	limit = 999999999
	if 'limit' in get:
	limit = int(get.limit)
	f_sql = '1=1'
	if 'search' in get and get.search:
	f_sql = "task_name like '%{}%' or sites like '%{}%' ".format(get.search, get.search)

	count = self.M("autodeploy_task").where(f_sql, ()).count()
	page_data = public.get_page(count, p, limit, collback)

	data = self.M("autodeploy_task").where(f_sql, ()).limit(page_data["shift"]+","+page_data["row"]).order("id desc").select()
	# public.print_log(data)

	try:
	cloud_data = self.plane_synchron_access_list()
	except:
	cloud_data = None
	# public.print_log(cloud_data)

	deploy = []
	for i in data:
	i["access_state"] = 0
	i["synchron_info"] = {}
	if not cloud_data:
	continue
	for j in cloud_data["msg"]:
	if i["access_id"] == j["id"]:
	i["access_state"] = j["access_state"]
	if j["access_state"] == 2 and str(i["ssl_cloud_id"]) != str(j["synchron_info"]["ca_id"]):
	self.M("autodeploy_task").where("cloud_id=?", (i["cloud_id"],)).update({"ssl_cloud_id": j["synchron_info"]["ca_id"]})
	# 写需要更新证书的任务到队列
	deploy.append(i["id"])
	i["synchron_info"] = j["synchron_info"]
	if deploy:
	public.ExecShell("nohup btpython {}/class/sslModel/autodeployModel.py --task_ids={} > /dev/null 2>&1 &".format(public.get_panel_path(), ",".join(str(i) for i in deploy)))
	page_data["data"] = data

	return {"status": True, "msg": page_data, "deploy": deploy}

	def add_task(self, get):
	task_name = get.task_name
	cloud_id = get.cloud_id
	private_key = get.private_key
	sites = get.sites
	# cycle = get.cycle
	cycle = 1
	data = self.M("autodeploy_task").where("cloud_id=?", (cloud_id,)).find()
	if data:
	return public.returnMsg(False, "选择的证书部署任务在本服务器已存在对应的任务【{}】请勿重复添加！".format(data["task_name"]))

	res = self.plane_synchron_access_set(get)
	if not res["status"]:
	return public.returnMsg(False, res["msg"])

	self.M("autodeploy_task").add('task_name,cloud_id,access_id,private_key,sites,cycle', (task_name, cloud_id, res["msg"]["id"], private_key, sites, cycle))
	if not res["status"]:
	return public.returnMsg(False, res["msg"])
	return public.returnMsg(True, "添加成功")

	def plane_synchron_access_delte(self, get):
	self.__PDATA["id"] = int(get.access_id)
	res = self.request('plane_synchron_access_delte')
	if not res["success"]:
	return public.returnMsg(False, res["res"])
	return public.returnMsg(True, res["res"])

	def del_task(self, get):
	data = self.M("autodeploy_task").where("id=?", (get.task_id,)).find()
	if not data:
	return public.returnMsg(False, "未找到任务")
	self.M("autodeploy_task").where("id=?", (get.task_id,)).delete()
	self.plane_synchron_access_delte(public.to_dict_obj({"access_id": data["access_id"]}))
	return public.returnMsg(True, "删除成功")

	def batch_del_task(self, get):
	task_id_list = get.task_ids.split(",")
	data = self.M("autodeploy_task").where("id in ({})".format(get.task_ids), ()).select()
	self.M("autodeploy_task").where("id in ({})".format(get.task_ids), ()).delete()

	total = len(data)
	faild = 0
	success = 0
	faildList = []
	successList = []
	for i in data:
	if str(i["id"]) not in task_id_list:
	faild += 1
	faildList.append({"task_id": i["id"], "task_name":i["task_name"],"error_msg": "任务不存在", "status": False})
	continue
	self.plane_synchron_access_delte(public.to_dict_obj({"access_id": i["access_id"]}))
	success += 1
	successList.append({"task_id": i["id"], "task_name":i["task_name"],"error_msg": "删除成功", "status": True})

	return public.returnMsg(True, {"total": total, "success": success, "faild": faild, "faildList": faildList, "successList": successList})

	def edit_task(self, get):
	task_id = get.task_id
	task_name = get.task_name
	cloud_id = int(get.cloud_id)
	private_key = get.private_key
	sites = get.sites
	# cycle = get.cycle
	cycle = 1

	data = self.M("autodeploy_task").where("id=?", (task_id,)).find()
	if not data:
	return public.returnMsg(False, "未找到任务")
	update_data = {"task_name": task_name, "cloud_id": cloud_id, "private_key": private_key, "sites": sites, "cycle": cycle}

	if data["cloud_id"] != cloud_id:
	cloud_data = self.M("autodeploy_task").where("cloud_id=?", (cloud_id)).find()
	if cloud_data:
	return public.returnMsg(False, "选择的证书部署任务在本服务器已存在对应的任务【{}】请勿重复添加！".format(cloud_data["task_name"]))
	res = self.plane_synchron_access_set(get)
	if not res["status"]:
	return public.returnMsg(False, res["msg"])
	update_data["access_id"] = res["msg"]["id"]
	elif data["private_key"] != private_key:
	res = self.plane_synchron_access_set(get)
	if not res["status"]:
	return public.returnMsg(False, res["msg"])

	if data["sites"] != sites:
	update_data["ssl_cloud_id"] = ""

	self.M("autodeploy_task").where("id=?", (get.task_id,)).update(update_data)
	return public.returnMsg(True, "修改成功")

	def plane_synchron_get_cert(self, access_id):
	self.__PDATA["id"] = int(access_id)
	res = self.request('plane_synchron_get_cert')
	if not res["success"]:
	return public.returnMsg(False, res["res"])
	return public.returnMsg(True, res["res"])

	def set_cert_to_database(self, access_id):
	cert_data = self.plane_synchron_get_cert(access_id)
	if not cert_data["status"]:
	return public.returnMsg(False, cert_data["msg"])
	cert_data = cert_data["msg"]["cert"]

	from sslModel import certModel
	get = public.dict_obj()
	get.key = cert_data["privateCert"]
	get.csr = cert_data["cert"]

	try:
	cert = certModel.main().save_cert(get)
	if not cert["status"]:
	return False
	except:
	return False
	self.M("autodeploy_task").where("access_id=?", (access_id,)).update({"ssl_hash": cert["ssl_hash"]})
	return True

	def deploy_cert(self, get):
	data = self.M("autodeploy_task").where("id=?", (int(get.task_id),)).find()
	if not data:
	return public.returnMsg(False, "未找到任务")
	sites = data["sites"].split(",")
	ssl_hash = data["ssl_hash"]
	get.BatchInfo = json.dumps([
	{"ssl_hash": ssl_hash, "siteName": i}
	for i in sites
	])
	import panelSSL
	res = panelSSL.panelSSL().SetBatchCertToSite(get)

	deploy_time = int(time.time())
	if res["total"] == res["success"]:
	self.write_deploy_detail(get.task_id, json.dumps(res), data["access_id"], 1, 1)
	deploy_status = 1
	else:
	self.write_deploy_detail(get.task_id, json.dumps(res), data["access_id"], -1, 1)
	deploy_status = -1
	self.M("autodeploy_task").where("id=?", (int(get.task_id),)).update({"deploy_time": deploy_time, "deploy_status": deploy_status})

	def upload_deploy_msg(self, get):
	self.__PDATA["msg"] = get.error_msg
	self.__PDATA["id"] = get.access_id
	self.__PDATA["state"] = get.state
	self.__PDATA["type"] = get.msg_type

	# print(self.__PDATA)

	res = self.request('plane_synchron_deploy_state')
	if not res["success"]:
	return public.returnMsg(False, res["res"])
	return public.returnMsg(True, res["res"])

	def write_deploy_detail(self, task_id, error_msg, access_id, state, msg_type):
	path = public.get_panel_path() + '/data/auto_deploy/deploy_detail'
	if not os.path.exists(path):
	os.makedirs(path)
	public.writeFile(path + "/{}.json".format(task_id), error_msg)
	# print(access_id, state, msg_type)
	res = self.upload_deploy_msg(public.to_dict_obj({"access_id": access_id, "error_msg": error_msg, "state": state, "msg_type": msg_type}))
	# print(res)

	def get_task_deploy_detail(self, get):
	try:
	res = json.loads(public.readFile(public.get_panel_path() + '/data/auto_deploy/deploy_detail/{}.json'.format(get.task_id)))
	except:
	res = None
	return public.returnMsg(True, res)



	if __name__ == '__main__':
	try:
	if os.path.exists("/www/server/panel/data/auto_deploy.pl"):
	exit()
	public.writeFile("/www/server/panel/data/auto_deploy.pl", "")
	m = main()
	import argparse
	p = argparse.ArgumentParser()
	p.add_argument('--task_ids', default=None, help="任务id列表，用逗号分隔", dest="task_ids")
	args = p.parse_args()
	if not args.task_ids:
	exit()
	task_ids = args.task_ids.split(",")
	for i in task_ids:
	task_data = m.M("autodeploy_task").where("id=?", (int(i),)).find()
	if not task_data:
	continue
	set_cert_status = m.set_cert_to_database(task_data["access_id"])
	if not set_cert_status:
	sites = task_data["sites"].split(",")
	error_msg = json.dumps({
	"total": len(sites),
	"success": 0,
	"faild": len(sites),
	"faildList": [{"siteName": site, "error_msg": "证书保存失败", "status": False} for site in sites],
	"successList": []
	})
	m.write_deploy_detail(i, error_msg, task_data["access_id"], -1, 1)
	m.M("autodeploy_task").where("id=?", (int(i),)).update(
	{"deploy_time": int(time.time()), "deploy_status": -1})
	continue
	m.deploy_cert(public.to_dict_obj({"task_id": i}))
	print("部署成功：", task_data["task_name"])
	exit()
	finally:
	os.remove("/www/server/panel/data/auto_deploy.pl")