105 lines
3.3 KiB
Python
105 lines
3.3 KiB
Python
|
import requests
|
||
|
|
import requests.utils
|
||
|
|
|
||
|
|
|
||
|
|
class HTTPResponseObject(dict):
|
||
|
|
def __init__(self, *args, **kwargs):
|
||
|
|
super().__init__(*args, **kwargs)
|
||
|
|
self.url = None
|
||
|
|
self.status = None
|
||
|
|
self.reason = None
|
||
|
|
self.header = None
|
||
|
|
self.cookie = None
|
||
|
|
self.binary = None
|
||
|
|
self.text, self.json = None, None
|
||
|
|
self.encoding, self.duration = None, None
|
||
|
|
|
||
|
|
def __setattr__(self, key, value):
|
||
|
|
pass
|
||
|
|
|
||
|
|
def __getitem__(self, item):
|
||
|
|
try:
|
||
|
|
return super().__getitem__(item)
|
||
|
|
except Exception:
|
||
|
|
return None
|
||
|
|
|
||
|
|
def __getattr__(self, item):
|
||
|
|
try:
|
||
|
|
return super().__getitem__(item)
|
||
|
|
except Exception:
|
||
|
|
return None
|
||
|
|
|
||
|
|
|
||
|
|
class HTTPRequest:
|
||
|
|
__http__ = requests
|
||
|
|
__accept_method__ = ('GET', 'POST', 'PUT', 'DELETE', 'HEAD', 'OPTIONS', 'TRACE', 'PATCH')
|
||
|
|
|
||
|
|
def http(self, method='GET', url: str = None, query=None, data=None, json=None, file=None, header=None, cookie=None,
|
||
|
|
ua: str = None, auth=None, timeout: int | float = None, proxy: str = None, auto_redirect=False, ignore_cert_error=False):
|
||
|
|
method = method.upper()
|
||
|
|
if method not in self.__accept_method__:
|
||
|
|
raise Exception('Unsupported request method.')
|
||
|
|
if method == 'HEAD':
|
||
|
|
auto_redirect = False
|
||
|
|
if ua:
|
||
|
|
if not isinstance(header, dict):
|
||
|
|
header = {}
|
||
|
|
header['User-Agent'] = ua
|
||
|
|
if isinstance(proxy, str) and len(proxy) >= 5:
|
||
|
|
proxy = {'http': proxy, 'https': proxy}
|
||
|
|
else:
|
||
|
|
proxy = None
|
||
|
|
try:
|
||
|
|
response = 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
|
||
|
|
)
|
||
|
|
except Exception:
|
||
|
|
response = None
|
||
|
|
try:
|
||
|
|
response_json = response.json()
|
||
|
|
except Exception:
|
||
|
|
response_json = None
|
||
|
|
if response:
|
||
|
|
return HTTPResponseObject({
|
||
|
|
'url': response.url,
|
||
|
|
'status': response.status_code,
|
||
|
|
'reason': response.reason,
|
||
|
|
'header': response.headers,
|
||
|
|
'cookie': requests.utils.dict_from_cookiejar(response.cookies),
|
||
|
|
'binary': response.content,
|
||
|
|
'text': response.text,
|
||
|
|
'json': response_json,
|
||
|
|
'encoding': response.encoding,
|
||
|
|
'duration': round(response.elapsed.microseconds / 1000, 1)
|
||
|
|
})
|
||
|
|
else:
|
||
|
|
return HTTPResponseObject({
|
||
|
|
'status': -1
|
||
|
|
})
|
||
|
|
|
||
|
|
|
||
|
|
class HTTPSession(HTTPRequest):
|
||
|
|
__http__ = requests.session()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
# e.g. Send a get request.
|
||
|
|
res = HTTPSession().http(
|
||
|
|
method='GET', url='http://pv.sohu.com/cityjson', query={'ie': 'utf-8'},
|
||
|
|
ua='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
|
||
|
|
)
|
||
|
|
print(res)
|