robots-auto-action/tools/http_util.py

import json
import pickle

import requests

from tools import loggerKit


class http_util:
    def __init__(self):
        self.api_domain = None

    @staticmethod
    def __http_get(interface_url, header_data, interface_param):
        """

        :param interface_url: 接口地址
        :param header_data: 请求头文件
        :param interface_param: 接口请求参数
        :return: 字典形式结果
        """

        result = None
        try:
            if interface_url != '':
                req_url = interface_url

                response = requests.get(url=req_url,
                                        headers=header_data,
                                        verify=False,
                                        timeout=40,
                                        params=interface_param
                                        )
                if response.status_code == 200:
                    response_time = response.elapsed.microseconds / 1000  # 发起请求和响应到达的时间,单位ms
                    result = response.text
                if response.status_code == 404:
                    result = json.dumps({'接口返回状态': response.status_code}, ensure_ascii=False)
                    loggerKit.error(result)
            elif interface_url == '':
                result = json.dumps({'接口地址参数为空': ''}, ensure_ascii=False)
                loggerKit.error(result)
            else:
                result = json.dumps({'接口地址错误': ''}, ensure_ascii=False)
                loggerKit.error(result)
        except Exception as e:
            result = json.dumps({'未知错误': e}, ensure_ascii=False)
            loggerKit.error(result)

        return result

    @staticmethod
    def __http_post(interface_url, header_data, interface_param):
        """

        :param interface_url: 接口地址
        :param header_data: 请求头文件
        :param interface_param: 接口请求参数
        :return: 字典形式结果
        """
        print("headers", type(header_data))
        print("interface_param", type(interface_param))

        result = None
        response = None
        try:
            if interface_url != '':
                if 'application/x-www-form-urlencoded' in header_data.get('Content-Type'):
                    response = requests.post(url=interface_url,
                                             headers=header_data,  # 字典
                                             data=interface_param,  # 字典
                                             verify=False,
                                             timeout=40)

                if 'application/json' in header_data.get('Content-Type'):
                    response = requests.post(url=interface_url,
                                             headers=header_data,  # 字典
                                             json=interface_param,  # 字典
                                             verify=False,
                                             timeout=40)

                if 'multipart/form-data' in header_data.get('Content-Type'):
                    # files = {'file': open('upload.txt', 'rb')}
                    response = requests.post(url=interface_url,
                                             headers=header_data,  # 字典
                                             files=interface_param,  # 字典
                                             verify=False,
                                             timeout=40)

                if response.status_code == 200:
                    response_time = response.elapsed.microseconds / 1000  # 发起请求和响应到达的时间,单位ms
                    loggerKit.debug('接口返回结果：%s' % response.text)
                    result = response.text
                if response.status_code == 404:
                    result = json.dumps({'接口返回状态:%s': response.status_code}, ensure_ascii=False)
                    loggerKit.error(result)
            elif interface_url == '':
                result = json.dumps({'接口地址参数为空': ''}, ensure_ascii=False)
                loggerKit.error(result)
            else:
                result = json.dumps({'接口地址错误': ''}, ensure_ascii=False)
                loggerKit.error(result)
        except Exception as e:
            result = json.dumps({'未知错误': e}, ensure_ascii=False)
            loggerKit.error(result)

        return result

    @staticmethod
    def __http_put(interface_url, header_data, interface_param):
        """

        :param interface_url: 请求URL
        :param header_data: 请求头参数
        :param interface_param: 请求体参数
        :return:
        """
        result = None
        response = None

        try:
            if 'application/x-www-form-urlencoded' in header_data.get('Content-Type'):
                interface_param = interface_param
            elif 'application/json' in header_data.get('Content-Type'):
                interface_param = json.dumps(interface_param)
            else:
                pass
            print(type(interface_param), interface_param)
            if interface_url != '':
                req_url = interface_url
                response = requests.put(url=req_url,
                                        headers=header_data,
                                        data=interface_param,
                                        verify=False,
                                        timeout=40)
                if response.status_code == 200:
                    result = response.text
                if response.status_code == 404:
                    result = json.dumps({'接口返回状态': response.status_code}, ensure_ascii=False)
            elif interface_url == '':
                result = json.dumps({'接口地址参数为空': ''}, ensure_ascii=False)
                loggerKit.error(result)
            else:
                result = json.dumps({'接口地址错误': ''}, ensure_ascii=False)
                loggerKit.error(result)
        except Exception as e:
            result = json.dumps({'未知错误': e}, ensure_ascii=False)
            loggerKit.error(result)

        return result

    @staticmethod
    def set_http_head(header_data):
        """
        公共设置请求头方法
        设置http请求头，参数为json格式的字符串：{"Content-Type":“application/json”}
        """
        http_head = None
        if not header_data:
            http_head = {}
        else:
            if isinstance(header_data, dict):
                http_head = header_data
            elif isinstance(header_data, str) and header_data.startswith('{') and header_data.endswith('}'):
                http_head = json.loads(header_data)
            elif isinstance(header_data, bytes):
                http_head = pickle.loads(header_data)
            else:
                pass

        return http_head

    @staticmethod
    def set_http_body(body):
        """
        公共设置请求体方法
        设置http请求体，参数为json格式的字符串：{"demo":123}
        """
        http_body = None

        try:
            if isinstance(body, str) and body.startswith('{'):
                http_body = json.loads(body)
                loggerKit.debug('设置请求体成功')
            elif isinstance(body, dict):
                http_body = body
            elif isinstance(body, bytes):
                http_body = pickle.loads(body)
            else:
                pass
        except Exception as e:
            http_body = body
            loggerKit.error('设置请求体失败:%s' % body)
        return http_body

    def set_http_url(self, interface_url):
        """
         公共设置URL方法
         设置http请求url,如：http://www.baidu.com
        """
        self.api_domain = "https://www.baidu.com/"
        if not interface_url.startswith('https://') and not interface_url.startswith('http://'):
            http_url = self.api_domain + interface_url
        else:
            http_url = interface_url

        return http_url

    def common_http_request(self, request_type, request_url=None, headers=None, request_data=None, result_type="dict"):
        """
        发送http请求
        :param request_url:
        :param interface_url: 接口地址
        :param header_data: 请求头文件
        :param interface_param: 接口请求参数
        :param request_type: 请求类型,不区分大小写,GET POST
        :return: 字典形式结果
        """

        result = None
        # if requestUrl is not None:
        http_url = self.set_http_url(request_url)
        # if headers is not None:
        http_head = self.set_http_head(headers)

        if request_data:
            http_body = self.set_http_body(request_data)
        else:
            http_body = {}
        # try:
        if request_type.lower() == 'get':
            result = self.__http_get(http_url,
                                     http_head,
                                     http_body)
            loggerKit.debug('用例发送消息URL:%s\n'
                            '用例发送消息头：%s\n'
                            '用例发送消息体：%s\n'
                            % (http_url, http_head, http_body))

        elif request_type.lower() == 'post':
            result = self.__http_post(http_url,
                                      http_head,
                                      http_body)
            loggerKit.debug('用例发送消息URL:%s\n'
                            '用例发送消息头：%s\n'
                            '用例发送消息体：%s\n'
                            % (http_url, http_head, http_body))

        elif request_type.lower() == 'put':
            result = self.__http_put(http_url,
                                     http_head,
                                     http_body)
            loggerKit.debug('用例发送消息URL:%s\n'
                            '用例发送消息头：%s\n'
                            '用例发送消息体：%s\n'
                            % (http_url, http_head, http_body))
        else:
            loggerKit.error('请求类型错误')
        # except Exception as e:
        #     logger.error(e)
        #     logger.error('用例发送消息URL:%s\n'
        #                  '用例发送消息头：%s\n'
        #                  '用例发送消息体：%s\n'
        #                  % (http_url, http_head, http_body))
        #     return {}
        if result_type == "dict":
            try:
                return pickle.dumps(json.loads(result))
            except Exception:
                return pickle.dumps(result)
            # return json.loads(result)
        else:
            return pickle.dumps(result)