登录接口示例

4. 云端登录

云端API

"""
def cloud_login_api(self, userinfo: dict):
    """
    接口名称：云端版登录
    请求参数：userinfo 
    请求方式：post
    请求路径：cloud_login_url = Prefix.UserServer + "/users/login"
    参数示例：userinfo: body {"username":xxx, "password": xxx}
    """
    return self.send(self.post, self.cloud_login_url, json=userinfo).json()

def check_username_api(self, username:str):
    """
    接口名称：用户校验
    请求参数：username       
    请求方式：post
    请求路径：cloud_login_url = Prefix.UserServer + "/users/check"
    """
    payload = {'username': username}
    return self.send(self.post, self.login_check, json=payload).json()
"""

功能实现

class LoginBusiness(RquestApi):

    def cloud_login_api(self, userinfo: dict):
        """
        云端登录API
        @param userinfo:
        @return:
        """
        return self.send(self.post, cloud_login_url, json=userinfo).json()

    def check_username_api(self, username: str):
        """
        校验用户名称是否合格
        @param username:
        @return:
        """
        payload = {'username': username}
        return self.send(self.post, login_check, json=payload).json()

    def getCheckToken(self) -> str:
        """
        校验token是否合格
        @return:
        """
        getCheckUrl = os.path.join(SimOneUrl, user_check_url)
        payload = {'username': loginData['username']}
        response = self.send(self.post, getCheckUrl, json=payload)
        checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
        return checkToken

    def pad(self, text):
        """
        填充函数，使被加密数据的字节码长度是block_size的整数倍
        @param text:
        @return:
        """
        length = AES.block_size
        count = len(text.encode('utf-8'))
        add = length - (count % length)
        entext = text + (chr(add) * add)
        return entext

    def str_aes(self, string: str, key: str):
        """
        aes 加密
        @param string:
        @param key:
        @return:
        """
        aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
        res = aes.encrypt(self.pad(string).encode("utf8"))
        msg = str(base64.b64encode(res), encoding="utf8")
        return msg

    def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
        '''
        @param string:
        @param username:
        @param key: AES加密的key
        @return:
        '''
        check_token = self.check_username_api(username)
        target_string = check_token["data"]["checkToken"] + string
        msg = self.str_aes(target_string, key)
        return msg

    @property
    def _cloud_std_login(self):
        """
        云端版本登陆方式
        @return:
        """
        try:
            new_login_data = copy.deepcopy(loginData)
            new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
                                                          username=new_login_data['username'])
            print('cloud login data:' + str(new_login_data))
            response = self.cloud_login_api(userinfo=new_login_data)
            print("-------", response)
            print('cloud login request result:' + str(response))
            token = response['data']['token']
            os.environ[EnvPath.token] = token
            print(token)
            return token
        except Exception as e:
            print(str(e))
            raise print('get token error ')


# 登录入口
class LoginPolicy():
    standalone = "standalone"
    cloud = "cloud"

    def __new__(cls, env_name):
        if env_name == cls.standalone:
            return LoginBusiness()._enterprise_std_login
        elif env_name == cls.cloud:
            return LoginBusiness()._cloud_std_login
        else:
            return print("----------input error---------------")

云端调用示例

SimOneUrl = "http://127.0.0.1:30083"
loginData = {
    "username": 'admin',
    "password": 'admin',
}

if __name__ == '__main__':
    LoginPolicy("cloud")

正确返回结果

cloud login request result:{'code': 0, 'data': {'token': 'f2e3e458-264b-44fb-9f95-319a978ddec3'}}

5. 单机版登录

单机版API

"""
def enterprise_login_api(self):
    """
    接口名称：企业单机版本登录
    请求参数：无 
    请求方式：get
    请求路径：SimOneUrl + standalone_login_url
    参数示例：userinfo: body {"username":xxx, "password": xxx}
    """
    getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
    response = self.send(self.get, url=getLoginUrl)
    return response
"""

功能实现

class LoginBusiness(RquestApi):

    def cloud_login_api(self, userinfo: dict):
        """
        云端登录API
        @param userinfo:
        @return:
        """
        return self.send(self.post, cloud_login_url, json=userinfo).json()

    def check_username_api(self, username: str):
        """
        校验用户名称是否合格
        @param username:
        @return:
        """
        payload = {'username': username}
        return self.send(self.post, login_check, json=payload).json()

    def getCheckToken(self) -> str:
        """
        校验token是否合格
        @return:
        """
        getCheckUrl = os.path.join(SimOneUrl, user_check_url)
        payload = {'username': loginData['username']}
        response = self.send(self.post, getCheckUrl, json=payload)
        checkToken = json.loads(response.content.decode("utf-8"))['data']['checkToken']
        return checkToken

    def pad(self, text):
        """
        填充函数，使被加密数据的字节码长度是block_size的整数倍
        @param text:
        @return:
        """
        length = AES.block_size
        count = len(text.encode('utf-8'))
        add = length - (count % length)
        entext = text + (chr(add) * add)
        return entext

    def str_aes(self, string: str, key: str):
        """
        aes 加密
        @param string:
        @param key:
        @return:
        """
        aes = AES.new(key.encode("utf-8"), AES.MODE_ECB)
        res = aes.encrypt(self.pad(string).encode("utf8"))
        msg = str(base64.b64encode(res), encoding="utf8")
        return msg

    def get_str_aes(self, string: str, username: str, key: str = "eGeVQh0lRyq41I41") -> str:
        '''
        @param string:
        @param username:
        @param key: AES加密的key
        @return:
        '''
        check_token = self.check_username_api(username)
        target_string = check_token["data"]["checkToken"] + string
        msg = self.str_aes(target_string, key)
        return msg

    def enterprise_login_api(self):
        """
        @return:
        """
        getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
        response = self.send(self.get, url=getLoginUrl)
        return response
    @property
    def _enterprise_std_login(self):
        """
        单机版本登录方式
        @return:
        """
        try:
            getLoginUrl = os.path.join(SimOneUrl, standalone_login_url)
            response = self.send(self.get, url=getLoginUrl)
            token = json.loads(response.content.decode("utf-8"))['data']['token']
            os.environ[EnvPath.token] = token
        except Exception as e:
            print(str(e))
            token = ""
            # raise AssertionError('get token error ')
        return token

    @property
    def _cloud_std_login(self):
        """
        云端版本登陆方式
        @return:
        """
        try:
            new_login_data = copy.deepcopy(loginData)
            new_login_data['password'] = self.get_str_aes(string=new_login_data['password'],
                                                          username=new_login_data['username'])
            print('cloud login data:' + str(new_login_data))
            response = self.cloud_login_api(userinfo=new_login_data)
            print("-------", response)
            print('cloud login request result:' + str(response))
            token = response['data']['token']
            os.environ[EnvPath.token] = token
            print(token)
            return token
        except Exception as e:
            print(str(e))
            raise print('get token error ')


class LoginPolicy():
    standalone = "standalone"
    cloud = "cloud"

    def __new__(cls, env_name):
        if env_name == cls.standalone:
            return LoginBusiness()._enterprise_std_login
        elif env_name == cls.cloud:
            return LoginBusiness()._cloud_std_login
        else:
            return print("----------input error---------------")

调用示例

if __name__ == '__main__':
    LoginPolicy("standalone")

返回结果

{'method': 'GET', 'url': 'http://172.31.9.85:30083/api-user/users/standalone', 'headers': {'Content-Type': 'application/json', 'Authorization': None}}
response:{"code":0,"data":{"token":"b557f84a-070d-4ce3-8f70-8c0021428bed"}}