109 lines
3.7 KiB
Python
109 lines
3.7 KiB
Python
from dataclasses import asdict
|
|
from framework.main_framework import Response, main_function
|
|
from framework.requests_dataclass import SampleRequests
|
|
from dataclasses import dataclass
|
|
from flask import Blueprint
|
|
from flask import request
|
|
from framework.m_mysql_cli import MysqlCli
|
|
from framework.response_code import RequestMethods
|
|
|
|
|
|
@dataclass
|
|
class OverrideRequestsPOST(SampleRequests):
|
|
username: str
|
|
password: str
|
|
|
|
def __init__(self):
|
|
super(OverrideRequestsPOST, self).__init__(json_data=request.get_data(), methods=RequestMethods.POST)
|
|
self.username = self.json_data.get("username")
|
|
self.password = self.json_data.get("password")
|
|
if not self.username or not self.password:
|
|
self.make_exception(self.response_code_enum.RESPONSE_PARAMS_IS_VALID)
|
|
|
|
|
|
@dataclass
|
|
class ResponseInterface:
|
|
username: str
|
|
user_token: str
|
|
|
|
def convert_to_dict(self) -> dict:
|
|
return asdict(self)
|
|
|
|
|
|
class OverrideMysqlCli(MysqlCli):
|
|
"""
|
|
创建Mysql操作类
|
|
"""
|
|
|
|
def __init__(self):
|
|
super(OverrideMysqlCli, self).__init__()
|
|
self.table = "accounts_info"
|
|
|
|
def handle_search_password(self, params: OverrideRequestsPOST) -> dict:
|
|
sql: str = f"SELECT username, password FROM {self.table} WHERE username='{params.username}' AND 1=1"
|
|
self.create_sql(sql)
|
|
mysql_result: dict = self.fetchone(fetchall=True)
|
|
self.check_mysql_exception(mysql_result=mysql_result)
|
|
return mysql_result
|
|
|
|
def handle_update_token(self, data: ResponseInterface) -> None:
|
|
sql: str = f"UPDATE {self.table} SET token='{data.user_token}' WHERE username='{data.username}'"
|
|
self.create_sql(sql)
|
|
mysql_result = self.commit_all(please_commit=True)
|
|
self.check_mysql_exception(mysql_result=mysql_result)
|
|
|
|
|
|
class OverriderMainResponse(Response):
|
|
def __init__(self, connect_mysql: bool = True):
|
|
super(OverriderMainResponse, self).__init__()
|
|
self.is_connect_mysql = connect_mysql
|
|
|
|
# 是否连接数据库
|
|
if self.is_connect_mysql:
|
|
self.mysql_cli: OverrideMysqlCli = OverrideMysqlCli()
|
|
|
|
# 是否处理成功
|
|
self.handle_success: bool = False
|
|
|
|
def core(self) -> None:
|
|
params = OverrideRequestsPOST()
|
|
mysql_result = self.mysql_cli.handle_search_password(params=params)
|
|
|
|
if not self.mysql_cli.call_result(mysql_result):
|
|
self.make_exception(self.response_code_enum.RESPONSE_401_Unauthorized)
|
|
|
|
form_password = self.md5sum(params.password)
|
|
mysql_password = self.mysql_cli.call_result(mysql_result)[0][1]
|
|
|
|
if form_password != mysql_password:
|
|
self.make_exception(self.response_code_enum.RESPONSE_401_Unauthorized)
|
|
create_token: str = self.md5sum(f"{mysql_password}+{self.call_datetime(need_str=True)}")
|
|
|
|
response_interface: ResponseInterface = ResponseInterface(username=params.username,
|
|
user_token=create_token)
|
|
self.mysql_cli.handle_update_token(data=response_interface)
|
|
|
|
self.handle_success = True
|
|
if self.handle_success:
|
|
response_data: dict = response_interface.convert_to_dict()
|
|
self.build_response_data(response_data)
|
|
self.build_response_code(self.response_code_enum.RESPONSE_200_SUCCESS)
|
|
|
|
def before_response(self) -> dict:
|
|
if self.response == self.response_code_enum.RESPONSE_200_SUCCESS:
|
|
pass
|
|
|
|
if self.is_connect_mysql:
|
|
self.mysql_cli.close_connect()
|
|
|
|
return self.make_response()
|
|
|
|
|
|
login_bp = Blueprint("login_bp", __name__)
|
|
|
|
|
|
@login_bp.route("/login", methods=["POST"])
|
|
def main():
|
|
override_main_response = OverriderMainResponse()
|
|
return main_function(override_main_response)
|