import functools import random from datetime import datetime from unify_api.modules.zhiwei_u import config from pot_libs.logger import log from pot_libs.utils.pendulum_wrapper import my_pendulum from unify_api.modules.common.dao.common_dao import user_by_user_id, \ get_all_username from pot_libs.qingstor_util.qs_client import QsClient def get_order_num(w_origin): now = datetime.now().strftime("%Y%m%d%H%M%S") if w_origin: order_num = 'S' + now[2:] + str(random.randint(10, 99)) else: order_num = 'Z' + now[2:] + str(random.randint(10, 99)) return order_num def get_time(): """返回所有的时间戳""" t = [] for h in range(24): for s in range(60): t.append("%02d:%02d" % (h, s)) return t async def check_zhiweiu_auth(request): try: user_id = request.ctx.user_id except: user_id = request.json.get("user_id") user_dic = await user_by_user_id(user_id) # 运维用户 if user_dic.get("zhiweiu_auth") == 2: return True return False def zhiweiu_required(): """运维用户权限验证,装饰器""" def decorator(f, factory=2): @functools.wraps(f) async def wrapper(request, *args, **kwargs): is_authed = await check_zhiweiu_auth(request) if is_authed: return await f(request, *args, **kwargs) else: fail_body = { "code": 5003, "message": "用户没有该操作权限", "data": {}, "srv_time": my_pendulum.now().to_datetime_string() } return [fail_body, 200] return wrapper return decorator def allowed_file(filename): return '.' in filename and \ filename.rsplit('.', 1)[1] in config.ALLOWED_EXTENSIONS async def upload_file(file, qs_url): file_name = file.name file_body = file.body if file_body and allowed_file(file_name): async with QsClient() as client: resp = await client.put_file(key=qs_url, data=file_body) log.info(f"upload to qingstore the resp: {resp}") return 1 else: return None async def get_all_user(): user_dic = await get_all_username() return user_dic