84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
|
import inspect
|
||
|
|
import json, requests , zlib
|
||
|
|
from functools import wraps
|
||
|
|
from http.cookiejar import LWPCookieJar
|
||
|
|
import http.cookiejar as cookielib
|
||
|
|
default_headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit\
|
||
|
|
/537.36 (KHTML, like Gecko) Chrome/54.0.2840.99 Safari/537.36"}
|
||
|
|
class Session(requests.Session):
|
||
|
|
def __init__(self, headers={}, cookie_file=None):
|
||
|
|
requests.Session.__init__(self)
|
||
|
|
if cookie_file:
|
||
|
|
self.cookies = LWPCookieJar(filename = cookie_file)
|
||
|
|
self.cookies.load(ignore_discard = True)
|
||
|
|
self.update(headers)
|
||
|
|
self.auth = ('user', 'pass')
|
||
|
|
def save(self, filename = None):
|
||
|
|
self.cookies.save()
|
||
|
|
def update(self, headers):
|
||
|
|
self.headers.update(headers)
|
||
|
|
def http_error(code):
|
||
|
|
def _with(func):
|
||
|
|
@wraps(func)
|
||
|
|
def _deco(self, url):
|
||
|
|
return func(self, url)
|
||
|
|
_deco.__error_code__ = code
|
||
|
|
return _deco
|
||
|
|
return _with
|
||
|
|
class Http():
|
||
|
|
def __init__(self, headers = default_headers, session = None):
|
||
|
|
session = session or Session()
|
||
|
|
session.update(headers)
|
||
|
|
self.session = session
|
||
|
|
self._type = 'str'
|
||
|
|
self._error_dict = self._get_error_dict()
|
||
|
|
def _gets(self, url, data = None):
|
||
|
|
res = self.session.get(url, params = data)
|
||
|
|
self._error(res, url)
|
||
|
|
return self._res_data(res, self._type)
|
||
|
|
def _posts(self, url, data = None):
|
||
|
|
res = self.session.get(url, params = data)
|
||
|
|
self._error(res, url)
|
||
|
|
return self._res_data(res, self._type)
|
||
|
|
def _get(self, url, data = None):
|
||
|
|
print("_get", url)
|
||
|
|
res = requests.get(url, params = data, headers = self.session.headers)
|
||
|
|
self._error(res, url)
|
||
|
|
return self._res_data(res, self._type)
|
||
|
|
def _post(self, url, data):
|
||
|
|
print("_post", url)
|
||
|
|
res = requests.get(url, data = data, headers = self.session.headers)
|
||
|
|
self._error(res, url)
|
||
|
|
return self._res_data(res, self._type)
|
||
|
|
def _get_error_dict(self):
|
||
|
|
error_dict = {}
|
||
|
|
methods = inspect.getmembers(self, predicate=inspect.ismethod)
|
||
|
|
for method in methods:
|
||
|
|
error_code = getattr(method[1], '__error_code__', None)
|
||
|
|
if error_code:
|
||
|
|
error_dict[error_code] = method[1]
|
||
|
|
return error_dict
|
||
|
|
def _error(self, res, url):
|
||
|
|
code = res.status_code
|
||
|
|
if code == 200:
|
||
|
|
return
|
||
|
|
print(res.content)
|
||
|
|
if code in self._error_dict:
|
||
|
|
self._error_dict[code](url)
|
||
|
|
return
|
||
|
|
raise RuntimeError(code + '- unknown error ' + url)
|
||
|
|
@http_error(403)
|
||
|
|
def _error_403(self, url):
|
||
|
|
raise RuntimeError('error:403 - Forbidden ' + url)
|
||
|
|
@http_error(404)
|
||
|
|
def _error_404(self, url):
|
||
|
|
raise RuntimeError('error:404 - Not Found ' + url)
|
||
|
|
def _res_data(self, res, _type):
|
||
|
|
# encoding = None
|
||
|
|
# if 'Content-Encoding' in res.headers:
|
||
|
|
# encoding = res.headers['Content-Encoding']
|
||
|
|
# if encoding == 'gzip':
|
||
|
|
# res.content = zlib.decompress(res.content, 16 + zlib.MAX_WBITS)
|
||
|
|
# if _type == 'json':
|
||
|
|
# return json.loads(res.content)
|
||
|
|
return res.content
|