157 lines
5.1 KiB
Python
157 lines
5.1 KiB
Python
# -*- coding: UTF-8 -*-
|
|
'''
|
|
@Project :domainScanDemo
|
|
@File :http_utils.py
|
|
@IDE :PyCharm
|
|
@Author :梦伴
|
|
@Date :2026/4/9 0:06
|
|
@explain : HTTP工具类
|
|
'''
|
|
|
|
import requests
|
|
from curl_cffi import requests as curl_requests
|
|
from loguru import logger
|
|
|
|
|
|
class HTTPUtils:
|
|
"""
|
|
HTTP工具类
|
|
"""
|
|
|
|
@staticmethod
|
|
def get(url, headers=None, params=None, timeout=10, proxies=None, use_curl=False):
|
|
"""
|
|
发送GET请求
|
|
|
|
:param url: 请求URL
|
|
:param headers: 请求头
|
|
:param params: 查询参数
|
|
:param timeout: 超时时间
|
|
:param proxies: 代理
|
|
:param use_curl: 是否使用curl_cffi
|
|
:return: requests.Response - 响应对象
|
|
"""
|
|
try:
|
|
if use_curl:
|
|
# 使用curl_cffi模拟浏览器
|
|
response = curl_requests.get(url, headers=headers, params=params, timeout=timeout, proxies=proxies, impersonate='chrome')
|
|
else:
|
|
# 使用requests
|
|
response = requests.get(url, headers=headers, params=params, timeout=timeout, proxies=proxies)
|
|
|
|
response.raise_for_status() # 检查状态码
|
|
return response
|
|
except Exception as e:
|
|
logger.error(f"GET请求失败: {url}, 错误: {e}")
|
|
return None
|
|
|
|
@staticmethod
|
|
def post(url, headers=None, data=None, json=None, timeout=10, proxies=None, use_curl=False):
|
|
"""
|
|
发送POST请求
|
|
|
|
:param url: 请求URL
|
|
:param headers: 请求头
|
|
:param data: 表单数据
|
|
:param json: JSON数据
|
|
:param timeout: 超时时间
|
|
:param proxies: 代理
|
|
:param use_curl: 是否使用curl_cffi
|
|
:return: requests.Response - 响应对象
|
|
"""
|
|
try:
|
|
if use_curl:
|
|
# 使用curl_cffi模拟浏览器
|
|
response = curl_requests.post(url, headers=headers, data=data, json=json, timeout=timeout, proxies=proxies, impersonate='chrome')
|
|
else:
|
|
# 使用requests
|
|
response = requests.post(url, headers=headers, data=data, json=json, timeout=timeout, proxies=proxies)
|
|
|
|
response.raise_for_status() # 检查状态码
|
|
return response
|
|
except Exception as e:
|
|
logger.error(f"POST请求失败: {url}, 错误: {e}")
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_random_user_agent():
|
|
"""
|
|
获取随机用户代理
|
|
|
|
:return: str - 用户代理
|
|
"""
|
|
user_agents = [
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/146.0.0.0 Safari/537.36',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/145.0.0.0 Safari/537.36',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/138.0',
|
|
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Edge/146.0.0.0',
|
|
'Mozilla/5.0 (Macintosh; Intel Mac OS X 14_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.6 Safari/605.1.15'
|
|
]
|
|
|
|
import random
|
|
return random.choice(user_agents)
|
|
|
|
@staticmethod
|
|
def get_default_headers():
|
|
"""
|
|
获取默认请求头
|
|
|
|
:return: dict - 请求头
|
|
"""
|
|
return {
|
|
'User-Agent': HTTPUtils.get_random_user_agent(),
|
|
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
|
|
'Accept-Language': 'zh-CN,zh;q=0.9',
|
|
'Accept-Encoding': 'gzip, deflate, br',
|
|
'Connection': 'keep-alive',
|
|
'Upgrade-Insecure-Requests': '1',
|
|
'Sec-Fetch-Dest': 'document',
|
|
'Sec-Fetch-Mode': 'navigate',
|
|
'Sec-Fetch-Site': 'none',
|
|
'Sec-Fetch-User': '?1'
|
|
}
|
|
|
|
@staticmethod
|
|
def retry_request(func, max_retries=3, delay=1):
|
|
"""
|
|
重试请求
|
|
|
|
:param func: 请求函数
|
|
:param max_retries: 最大重试次数
|
|
:param delay: 重试延迟
|
|
:return: 函数返回值
|
|
"""
|
|
import time
|
|
|
|
for i in range(max_retries):
|
|
try:
|
|
result = func()
|
|
if result:
|
|
return result
|
|
except Exception as e:
|
|
logger.warning(f"请求失败，第 {i+1} 次重试: {e}")
|
|
|
|
if i < max_retries - 1:
|
|
time.sleep(delay)
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def check_proxy(proxy):
|
|
"""
|
|
检查代理是否可用
|
|
|
|
:param proxy: 代理URL
|
|
:return: bool - 是否可用
|
|
"""
|
|
try:
|
|
proxies = {
|
|
'http': proxy,
|
|
'https': proxy
|
|
}
|
|
|
|
response = requests.get('https://www.baidu.com', proxies=proxies, timeout=5)
|
|
return response.status_code == 200
|
|
except Exception as e:
|
|
logger.error(f"代理检查失败: {proxy}, 错误: {e}")
|
|
return False |