pip install requests tenacity

from net_helper import NetHelper, ApiResponse  # 假設(shè)工具保存為 net_helper.py

# 初始化 NetHelper
client = NetHelper(
    base_url="https://com.text.com",
    headers={
        "Accept": "application/json, text/plain, */*",
        "Content-Type": "application/json;charset=UTF-8",
    },
)

response = client.post(
    url="/login",
    json={
        "name": "bin",
        "pwd": "123456",
    }
)

if response.is_success():
    print("登錄成功！")
    print("響應(yīng)數(shù)據(jù):", response.data)
    print("狀態(tài)碼:", response.status_code)
else:
    print(f"登錄失??！錯誤信息: {response.error}")
    print("狀態(tài)碼:", response.status_code)
    print("原始響應(yīng):", response.data)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import logging
import json as jsonlib
from typing import Dict, Optional, Union, Any
import requests
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger(__name__)


class ApiResponse:
    """API響應(yīng)包裝類"""
    def __init__(self, data: Any, status_code: int = 200, error: Optional[str] = None):
        self.data = data
        self.status_code = status_code
        self.error = error
        self.message = data.get('message', '') if isinstance(data, dict) else ''
    
    def is_success(self) -> bool:
        """檢查請求是否成功"""
        return 200 <= self.status_code < 300 and self.error is None


class NetHelper:
    def __init__(self,
                 base_url: str,
                 headers: Optional[Dict[str, str]],
                 timeout: int = 15,
                 verify_ssl: bool = True):
        self.base_url = base_url.rstrip("/")
        self.default_headers = headers
        self.timeout = timeout
        self.verify_ssl = verify_ssl
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((
            requests.exceptions.ConnectionError,
            requests.exceptions.Timeout
        )),
        reraise=True
    )
    def _send_request(
        self,
        method: str = "GET",
        url: str = "",
        headers: Optional[Dict[str, str]] = None,
        json: Optional[Union[Dict[str, Any], str]] = None,
        data: Optional[Union[Dict[str, Any], str, bytes]] = None,
        params: Optional[Dict[str, Any]] = None
    ) -> ApiResponse:
        """內(nèi)部請求方法"""
        # 拼接完整URL
        full_url = f"{self.base_url}/{url.lstrip('/')}"
        
        # 合并請求頭
        merged_headers = {**self.default_headers, **(headers or {})}
        
        # 移除空值頭
        merged_headers = {k: v for k, v in merged_headers.items() if v is not None}

        try:
            # 打印請求體
            print("請求方法:", method.upper())
            print("請求URL:", full_url)
            if params:
                print("請求參數(shù):")
                print(jsonlib.dumps(params, ensure_ascii=False, indent=2))
            if json:
                print("請求JSON數(shù)據(jù):")
                print(jsonlib.dumps(json, ensure_ascii=False, indent=2))
            if data:
                print("請求表單數(shù)據(jù):")
                if isinstance(data, (dict, list)):
                    print(jsonlib.dumps(data, ensure_ascii=False, indent=2))
                else:
                    print(data)
            
            # 發(fā)送請求
            response = requests.request(
                method=method.upper(),
                url=full_url,
                headers=merged_headers,
                json=json,
                data=data,
                params=params,
                timeout=self.timeout,
                verify=self.verify_ssl
            )

            # 解析響應(yīng)
            try:
                result_data = response.json()
            except ValueError:
                result_data = response.text if response.encoding else response.content

            print("響應(yīng)數(shù)據(jù):")
            print(jsonlib.dumps(result_data, ensure_ascii=False, indent=2))
            
            # 檢查HTTP狀態(tài)碼
            if 200 <= response.status_code < 300:
                return ApiResponse(result_data, response.status_code)
            else:
                error_msg = f"HTTP錯誤: 狀態(tài)碼 {response.status_code}"
                logger.error(f"{error_msg}, 響應(yīng): {result_data}")
                return ApiResponse(result_data, response.status_code, error_msg)

        except requests.exceptions.ConnectionError as e:
            logger.error(f"連接失敗: 無法連接到 {full_url}")
            return ApiResponse({}, 0, str(e))
        except requests.exceptions.Timeout as e:
            logger.error(f"請求超時: 超過 {self.timeout} 秒")
            return ApiResponse({}, 0, str(e))
        except Exception as e:
            logger.error(f"未知錯誤: {str(e)}")
            return ApiResponse({}, 0, str(e))
    
    def post(self, url: str, data: Optional[Any] = None, params: Optional[Dict[str, Any]] = None) -> ApiResponse:
        return self._send_request(method="POST", url=url, json=data, params=params)
    
    def get(self, url: str, params: Optional[Dict[str, Any]] = None) -> ApiResponse:
        return self._send_request(method="GET", url=url, params=params)
    
    def put(self, url: str, data: Optional[Any] = None, params: Optional[Dict[str, Any]] = None) -> ApiResponse:
        return self._send_request(method="PUT", url=url, json=data, params=params)
    
    def delete(self, url: str, params: Optional[Dict[str, Any]] = None) -> ApiResponse:
        return self._send_request(method="DELETE", url=url, params=params)