89 lines
2.6 KiB
Python
89 lines
2.6 KiB
Python
|
from abc import ABC
|
||
|
|
from framework.response_code import ResponseCode
|
||
|
|
from datetime import datetime
|
||
|
|
from flask import request
|
||
|
|
import hashlib
|
||
|
|
|
||
|
|
|
||
|
|
class MinCustomException(Exception):
|
||
|
|
def __init__(self, error):
|
||
|
|
self.error = error
|
||
|
|
super().__init__(self.error)
|
||
|
|
|
||
|
|
|
||
|
|
class OpenABC(ABC):
|
||
|
|
"""
|
||
|
|
提供公共函数和功能函数
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self.response_code_enum: ResponseCode = ResponseCode
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def call_datetime(need_str: bool = True):
|
||
|
|
return str(datetime.now()) if need_str else datetime.now()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def md5sum(string: str):
|
||
|
|
md5 = hashlib.md5()
|
||
|
|
password = string.encode("utf-8")
|
||
|
|
md5.update(password)
|
||
|
|
return md5.hexdigest()
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def make_exception(error_code: ResponseCode):
|
||
|
|
raise MinCustomException(error_code)
|
||
|
|
|
||
|
|
|
||
|
|
class OpenResponse(OpenABC):
|
||
|
|
def __init__(self):
|
||
|
|
super(OpenResponse, self).__init__()
|
||
|
|
self._response_code = self.response_code_enum.RESPONSE_100_CONTINUE # 错误代码, 默认Continue
|
||
|
|
|
||
|
|
self._response_data = None # 返回的数据
|
||
|
|
|
||
|
|
self.response: dict = {}
|
||
|
|
|
||
|
|
self.echo = False # 是否打印消息
|
||
|
|
|
||
|
|
# def make_logging(self):
|
||
|
|
# request_url: str = request.url
|
||
|
|
# if self.response_code == self.response_code_enum.RESPONSE_200_SUCCESS:
|
||
|
|
# self.log_access(f"{request_url}\n{self.response}\n")
|
||
|
|
# else:
|
||
|
|
# self.log_error(f"{request_url}\n{self.response}\n")
|
||
|
|
|
||
|
|
def make_response(self) -> dict:
|
||
|
|
response_code: ResponseCode = self._response_code
|
||
|
|
|
||
|
|
err_code, err_message = response_code.return_tuple()
|
||
|
|
self.response = {"code": err_code, "message": err_message}
|
||
|
|
|
||
|
|
if self._response_data is not None:
|
||
|
|
self.response.update({"data": self._response_data})
|
||
|
|
|
||
|
|
if self.echo is True:
|
||
|
|
print("-" * 50)
|
||
|
|
print("Interface ERROR:")
|
||
|
|
print(f"Echo url: {request.url}")
|
||
|
|
print(f"Echo methods: {request.method}")
|
||
|
|
if request.method == "GET":
|
||
|
|
print(f"Echo params: {request.args}")
|
||
|
|
elif request.method == "POST":
|
||
|
|
print(f"Echo data: {request.data}")
|
||
|
|
print(f"Echo response: {self.response}")
|
||
|
|
return self.response
|
||
|
|
|
||
|
|
def build_response_data(self, data: object):
|
||
|
|
"""
|
||
|
|
生成response_data
|
||
|
|
:return:
|
||
|
|
"""
|
||
|
|
self._response_data = data
|
||
|
|
|
||
|
|
def build_response_code(self, code_enum: ResponseCode):
|
||
|
|
self._response_code = code_enum
|
||
|
|
|
||
|
|
def make_debug_response(self):
|
||
|
|
raise MinCustomException(self.response_code_enum.RESPONSE_201_DEBUG)
|