mirror of
https://github.com/EstrellaXD/Auto_Bangumi.git
synced 2026-04-02 10:10:32 +08:00
105 lines
3.4 KiB
Python
105 lines
3.4 KiB
Python
import logging
|
|
import socket
|
|
import time
|
|
|
|
import requests
|
|
import socks
|
|
|
|
from module.conf import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class RequestURL:
|
|
def __init__(self):
|
|
self.header = {"user-agent": "Mozilla/5.0", "Accept": "application/xml"}
|
|
self._socks5_proxy = False
|
|
|
|
def get_url(self, url, retry=3):
|
|
try_time = 0
|
|
while True:
|
|
try:
|
|
req = self.session.get(url=url, headers=self.header, timeout=5)
|
|
req.raise_for_status()
|
|
return req
|
|
except requests.RequestException:
|
|
logger.warning(
|
|
f"[Network] Cannot connect to {url}. Wait for 5 seconds."
|
|
)
|
|
try_time += 1
|
|
if try_time >= retry:
|
|
break
|
|
time.sleep(5)
|
|
except Exception as e:
|
|
logger.debug(e)
|
|
break
|
|
logger.error(f"[Network] Failed connecting to {url}")
|
|
logger.warning("[Network] Please check DNS/Connection settings")
|
|
return None
|
|
|
|
def post_url(self, url: str, data: dict, retry=3):
|
|
try_time = 0
|
|
while True:
|
|
try:
|
|
req = self.session.post(
|
|
url=url, headers=self.header, data=data, timeout=5
|
|
)
|
|
req.raise_for_status()
|
|
return req
|
|
except requests.RequestException:
|
|
logger.warning(
|
|
f"[Network] Cannot connect to {url}. Wait for 5 seconds."
|
|
)
|
|
try_time += 1
|
|
if try_time >= retry:
|
|
break
|
|
time.sleep(5)
|
|
except Exception as e:
|
|
logger.debug(e)
|
|
break
|
|
logger.error(f"[Network] Failed connecting to {url}")
|
|
logger.warning("[Network] Please check DNS/Connection settings")
|
|
return None
|
|
|
|
def check_url(self, url: str):
|
|
if "://" not in url:
|
|
url = f"http://{url}"
|
|
try:
|
|
req = requests.head(url=url, headers=self.header, timeout=5)
|
|
req.raise_for_status()
|
|
return True
|
|
except requests.RequestException:
|
|
logger.debug(f"[Network] Cannot connect to {url}.")
|
|
return False
|
|
|
|
def __enter__(self):
|
|
self.session = requests.Session()
|
|
if settings.proxy.enable:
|
|
if "http" in settings.proxy.type:
|
|
url = f"{settings.proxy.type}://{settings.proxy.host}:{settings.proxy.port}"
|
|
self.session.proxies = {
|
|
"https": url,
|
|
"http": url,
|
|
}
|
|
elif settings.proxy.type == "socks5":
|
|
self._socks5_proxy = True
|
|
socks.set_default_proxy(
|
|
socks.SOCKS5,
|
|
addr=settings.proxy.host,
|
|
port=settings.proxy.port,
|
|
rdns=True,
|
|
username=settings.proxy.username,
|
|
password=settings.proxy.password,
|
|
)
|
|
socket.socket = socks.socksocket
|
|
else:
|
|
logger.error(f"[Network] Unsupported proxy type: {settings.proxy.type}")
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
|
if self._socks5_proxy:
|
|
socks.set_default_proxy()
|
|
socket.socket = socks.socksocket
|
|
self._socks5_proxy = False
|
|
self.session.close()
|