convert domainCheck to regular directory

domainCheck/app/utils/http_utils.py

@@ -0,0 +1,157 @@
+# -*- 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