''' Author: your name Date: 2021-06-30 18:03:41 LastEditTime: 2021-07-15 17:10:22 LastEditors: Please set LastEditors Description: In User Settings Edit FilePath: \evm-store\tools\build_out\controllers\package.py ''' #!/usr/bin/env python # -*- coding: utf_8 -*- import os import re import shutil from datetime import datetime from application.app import db, config from models.annex import AnnexModel from models.app import AppModel from models.user import UserModel from models.package import PackageModel from webcreator.utils.epk import EpkApp from webcreator.log import logger from webcreator.response import ResponseCode class PackageResource(object): def __init__(self): super().__init__() def get(self, uuid, params): # handle business filters = [PackageModel.is_delete==False, PackageModel.uuid==uuid] result = PackageModel.query.filter(*filters).first() if result: return result, ResponseCode.HTTP_SUCCESS return None, ResponseCode.HTTP_NOT_FOUND def getList(self, params): # handle business logger.warn(params) filters = [PackageModel.is_delete==False] result = PackageModel.query.filter(*filters).order_by(PackageModel.create_at).paginate(params.get('page', 1), params.get('pageSize', 10), error_out=False) if result: return result, ResponseCode.HTTP_SUCCESS user = UserModel.query.filter(id=params.get('user')).onr_or_none() if not user: return False, ResponseCode.USER_NOT_EXISTS temp = {} if user.role == "administrator": temp.update({"is_delete": False}) else: temp.update({ "create_by": user.id, "is_delete": False }) if "scope_type" in params and params.get("scope_type") == "list": result = PackageModel.query.filter_by(**temp).order_by(PackageModel.create_at.desc()) temp = [] for item in result: temp.append({ "name": item.app.app_name, "uuid": str(item.uuid) }) return temp, len(temp), "get build_logs {}.".format("success" if temp else "fail") result = PackageModel.query.filter_by(**temp).order_by(PackageModel.create_at.desc()).paginate(params.get("pagenum", 1), params.get("pagesize", 10), error_out=False) if result.total and len(result.items): temp = [] for item in result.items: t = item.to_dict() t.update({ "create_at": item.create_at.strftime("%Y-%m-%d %H:%M:%S") if item.create_at else None, "update_at": item.update_at.strftime("%Y-%m-%d %H:%M:%S") if item.update_at else None, }) temp.append(t) return (temp, result.total), ResponseCode.HTTP_SUCCESS return result, ResponseCode.HTTP_NO_DATA def post(self, params, jwt={}): # 判断用户是否存在 user = UserModel.query.filter(UserModel.id==params.get('user')) if not user: return False, ResponseCode.USER_NOT_EXISTS # 判断app是否存在 app = AppModel.query.filter(AppModel.id==params.get("app")) if not app: return False, ResponseCode.HTTP_NOT_FOUND # 根据应用查找有哪些源文件 source_files = AnnexModel.query.filter(AnnexModel.app==app.id).all() if not source_files: return None, ResponseCode.HTTP_NO_DATA dir_format = "{}-{}-{}".format(app.app_name, app.app_version, datetime.now().strftime("%Y%m%d%H%M%S")) upload_dir = os.sep.join([config.UPLOAD_ROOT_DIR, "uploads", "evueapps"]) target_dir = os.sep.join([upload_dir, user.account, dir_format]) dest_dir = os.sep.join([target_dir, "src"]) if not os.path.exists(dest_dir): os.makedirs(dest_dir) app_files = [] for sf in source_files: target_file = os.sep.join([config.UPLOAD_ROOT_DIR, sf.path]) filename = os.path.basename(target_file) name, suffix = os.path.splitext(filename) name = re.sub(r"_\d{14}$", "", name) dst_file = os.path.normpath(os.sep.join([dest_dir, name + suffix])) shutil.move(os.path.normpath(target_file), dst_file) app_files.append([sf.id, dst_file]) # 打包成EPK文件 app_info = {} params = { 'appName': app.app_name, 'appDir': dest_dir, 'appVersion': app.app_version, 'output': target_dir } if user.role == "administrator" or user.role == "community": params['algorithm'] = "h" epk = EpkApp(**params) app_info = epk.pack() if app_info: app_info['md5'] = str(app_info['md5']) # 更新数据库对应文件路径 # for sf in source_files: # for af in app_files: # if sf.id == af[0]: # t = os.path.normpath(af[1].replace(config.UPLOAD_ROOT_DIR, "")).replace('\\', '/') # sf.set(path=t) # db.session.flush() # db.session.commit() epk_path = os.sep.join([target_dir.replace(config.UPLOAD_ROOT_DIR, ""), "{}.epk".format(app.app_name)]).replace('\\', '/') # handle business # result = PackageModel.query.filter(PackageModel.app == params.get('app')).first() # if result and result.is_delete: # result.is_delete = False # result.update_by = jwt.get("id", "") # result.update_date = datetime.now() # db.session.commit() # return True, ResponseCode.HTTP_SUCCESS # elif result and result.is_delete == False: # return False, ResponseCode.HTTP_INVAILD_REQUEST result = PackageModel(app=app.id, file_path=epk_path, package_info=app_info, app_version=params.get("app_version"), create_by=user.id, create_at=datetime.now(), update_by=user.id, update_at=datetime.now()) db.session.add(result) db.session.commit() return True, ResponseCode.HTTP_SUCCESS def put(self, uuid, params, jwt={}): # handle business result = PackageModel.query.filter(PackageModel.uuid==uuid).first() if not result: return None, ResponseCode.HTTP_NOT_FOUND if params: for key, value in params.items(): if value != None: setattr(result, key, value) result.update_by = jwt.get("id", "") result.update_date = datetime.now() db.session.commit() return True, ResponseCode.HTTP_SUCCESS else: return False, ResponseCode.HTTP_INVAILD_REQUEST def delete(self, uuid, jwt={}): # handle business result = PackageModel.query.filter(PackageModel.uuid==uuid).first() if not result: return False, ResponseCode.HTTP_NOT_FOUND result.update_by = jwt.get("id", "") result.update_date = datetime.now() result.is_delete = True db.session.delete(result) db.session.commit() return True, ResponseCode.HTTP_SUCCESS packageManager = PackageResource()