89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
import gzip
|
|
import urllib.parse
|
|
import urllib.error
|
|
import requests
|
|
import ssl
|
|
import json
|
|
import urllib.request
|
|
|
|
no_proxy_handler = urllib.request.ProxyHandler({})
|
|
opener = urllib.request.build_opener(no_proxy_handler)
|
|
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
OptionalStr = str | None
|
|
OptionalDict = dict | None
|
|
|
|
|
|
def sync_req(
|
|
url: str,
|
|
proxy_addr: OptionalStr = None,
|
|
headers: OptionalDict = None,
|
|
data: dict | bytes | None = None,
|
|
json_data: dict | list | None = None,
|
|
timeout: int = 20,
|
|
redirect_url: bool = False,
|
|
abroad: bool = False,
|
|
content_conding: str = 'utf-8'
|
|
) -> str:
|
|
if headers is None:
|
|
headers = {}
|
|
try:
|
|
if proxy_addr:
|
|
proxies = {
|
|
'http': proxy_addr,
|
|
'https': proxy_addr
|
|
}
|
|
if data or json_data:
|
|
response = requests.post(
|
|
url, data=data, json=json_data, headers=headers, proxies=proxies, timeout=timeout
|
|
)
|
|
else:
|
|
response = requests.get(url, headers=headers, proxies=proxies, timeout=timeout)
|
|
if redirect_url:
|
|
return response.url
|
|
resp_str = response.text
|
|
else:
|
|
if data and not isinstance(data, bytes):
|
|
data = urllib.parse.urlencode(data).encode(content_conding)
|
|
if json_data and isinstance(json_data, (dict, list)):
|
|
data = json.dumps(json_data).encode(content_conding)
|
|
|
|
req = urllib.request.Request(url, data=data, headers=headers)
|
|
|
|
try:
|
|
if abroad:
|
|
response = urllib.request.urlopen(req, timeout=timeout)
|
|
else:
|
|
response = opener.open(req, timeout=timeout)
|
|
if redirect_url:
|
|
return response.url
|
|
content_encoding = response.info().get('Content-Encoding')
|
|
try:
|
|
if content_encoding == 'gzip':
|
|
with gzip.open(response, 'rt', encoding=content_conding) as gzipped:
|
|
resp_str = gzipped.read()
|
|
else:
|
|
resp_str = response.read().decode(content_conding)
|
|
finally:
|
|
response.close()
|
|
|
|
except urllib.error.HTTPError as e:
|
|
if e.code == 400:
|
|
resp_str = e.read().decode(content_conding)
|
|
else:
|
|
raise
|
|
except urllib.error.URLError as e:
|
|
print(f"URL Error: {e}")
|
|
raise
|
|
except Exception as e:
|
|
print(f"An error occurred: {e}")
|
|
raise
|
|
|
|
except Exception as e:
|
|
resp_str = str(e)
|
|
|
|
return resp_str
|