AutoFramework/Base/Class/Http.py

import requests, requests.utils
from typing import Any, Text, Dict, SupportsInt, SupportsFloat, SupportsBytes


def msg_digest(res, **kwargs):
    import urllib.parse
    toshort = lambda t: t[0:256] + '  ... [MORE ' + str(
        int(((len(t) - 256) / len(t)) * 100)) + '% OMITTED HERE]' if len(t) > 256 else t
    reqline = '%s %s HTTP/%.1f\n' % (res.request.method, res.request.path_url, res.raw.version * 0.1)
    reqhead = 'Host: %s\n' % urllib.parse.urlparse(res.request.url).netloc
    for k, v in dict(res.request.headers).items():
        reqhead += '%s: %s\n' % (k, v)
    else:
        reqhead += '\n'
    reqbody = ''
    if isinstance(res.request.body, bytes):
        reqbody += str(res.request.body)[2:-1]
    elif res.request.body:
        reqbody += str(res.request.body)
    resline = 'HTTP/%.1f %d %s\n' % (res.raw.version * 0.1, res.raw.status, res.raw.reason)
    reshead = ''
    for k, v in dict(res.headers).items():
        reshead += '%s: %s\n' % (k, v)
    else:
        reshead += '\n'
    resbody = ''
    if res.text:
        resbody += res.text
    return {
        "req": reqline + reqhead + toshort(reqbody),
        "res": resline + reshead + toshort(resbody)
    }


def hook_response_debug(res, **kwargs):
    info = msg_digest(res, **kwargs)
    logs = '[->]' + '\n' + info["req"] + '\n' + '[<-]' + '\n' + info["res"]
    print(logs)
    try:
        import logging
        logging.log(logging.DEBUG, logs)
    except:
        pass


class Request:
    __http__ = requests
    __method_list__ = ('GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'TRACE', 'PATCH')
    __response_object__ = None
    __response_briefs__ = None

    @staticmethod
    def _msg_digest(res, **kwargs):
        import urllib.parse
        toshort = lambda t: t[0:256] + '  ... [MORE ' + str(
            int(((len(t) - 256) / len(t)) * 100)) + '% OMITTED HERE]' if len(t) > 256 else t
        reqline = '%s %s HTTP/%.1f\n' % (res.request.method, res.request.path_url, res.raw.version * 0.1)
        reqhead = 'Host: %s\n' % urllib.parse.urlparse(res.request.url).netloc
        for k, v in dict(res.request.headers).items():
            reqhead += '%s: %s\n' % (k, v)
        else:
            reqhead += '\n'
        reqbody = ''
        if isinstance(res.request.body, bytes):
            reqbody += str(res.request.body)[2:-1]
        elif res.request.body:
            reqbody += str(res.request.body)
        resline = 'HTTP/%.1f %d %s\n' % (res.raw.version * 0.1, res.raw.status, res.raw.reason)
        reshead = ''
        for k, v in dict(res.headers).items():
            reshead += '%s: %s\n' % (k, v)
        else:
            reshead += '\n'
        resbody = ''
        if res.text:
            resbody += res.text
        return {
            "req": reqline + reqhead + toshort(reqbody),
            "res": resline + reshead + toshort(resbody)
        }

    def _response_briefs(self, res, **kwargs):
        info = msg_digest(res, **kwargs)
        logs = '[->]' + '\n' + info["req"] + '\n' + '[<-]' + '\n' + info["res"]
        self.__response_briefs__ = logs

    def http(self,
             method: Text = 'GET',
             url: Text = None,
             query: Dict | Text = None,
             data: Dict | Text | SupportsBytes = None,
             json: Any = None,
             file: Dict = None,
             header: Dict = None,
             cookie: Dict = None,
             auth: Any = None,
             timeout: SupportsInt | SupportsFloat = None,
             proxy: Text = None,
             auto_redirect=False,
             ignore_cert_error=False,
             debug=False):
        if method is None:
            raise Exception('未指定的请求方法 | Unspecified request method.')
        method = method.upper()
        if method not in self.__method_list__:
            raise Exception('不支持的请求方法 | Unsupported request method.')
        if method == 'HEAD':
            auto_redirect = False
        if isinstance(proxy, str) and len(proxy) >= 5:
            proxy = {"http": proxy, "https": proxy}
        else:
            proxy = None
        self.__response_briefs__ = None
        self.__response_object__ = self.__http__.request(
            method=method,
            url=url,
            params=query,
            data=data,
            headers=header,
            cookies=cookie,
            files=file,
            auth=auth,
            timeout=timeout,
            allow_redirects=auto_redirect,
            proxies=proxy,
            verify=not ignore_cert_error,
            json=json,
            hooks=[None, {"response": self._response_briefs}][debug]
        )
        try:
            json_content = self.__response_object__.json()
        except:
            json_content = None
        return {
            "url": self.__response_object__.url, "encoding": self.__response_object__.encoding,
            "status": self.__response_object__.status_code,
            "reason": self.__response_object__.reason,
            "header": self.__response_object__.headers,
            "duration": round(self.__response_object__.elapsed.microseconds / 1000, 1),
            "cookie": requests.utils.dict_from_cookiejar(self.__response_object__.cookies),
            "text": self.__response_object__.text,
            "json": json_content,
            "content": self.__response_object__.content,
            "brief": self.__response_briefs__
        }


class Session(Request):
    __http__ = requests.session()


if __name__ == '__main__':
    req = Session()
    print(req.http('GET', 'https://www.baidu.com/s', query={"wd": '社会主义核心价值观'},
                   header={"User-Agent": 'Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) '
                                         'Chrome/86.0.4240.198 Safari/537.36'},
                   auto_redirect=True, debug=False))