import requests

try:
    # 連接超時(shí)3秒（TCP握手階段）
    # 讀取超時(shí)10秒（服務(wù)器處理階段）
    response = requests.get(
        'https://api.sf-express.com/track',
        params={'trackingNumber': 'SF123456789'},
        timeout=(3, 10)  # 元組形式分別設(shè)置
    )
    print(response.json())
except requests.exceptions.ConnectTimeout:
    print("連接服務(wù)器失敗，請(qǐng)檢查網(wǎng)絡(luò)")
except requests.exceptions.ReadTimeout:
    print("服務(wù)器處理超時(shí)，請(qǐng)稍后重試")

def get_dynamic_timeout(retry_count):
    base_timeout = 3  # 基礎(chǔ)超時(shí)
    if retry_count > 0:
        return min(base_timeout * (2 ** retry_count), 30)  # 指數(shù)退避，最大30秒
    return base_timeout

# 使用示例
for i in range(3):
    try:
        timeout = get_dynamic_timeout(i)
        response = requests.get(url, timeout=timeout)
        break
    except Exception as e:
        print(f"第{i+1}次嘗試失敗，超時(shí)時(shí)間調(diào)整為{timeout}秒")

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_retry_session(retries=3, backoff_factor=1, status_forcelist=(500, 502, 503, 504)):
    session = requests.Session()
    retry = Retry(
        total=retries,
        read=True,  # 允許讀取超時(shí)重試
        connect=True,  # 允許連接超時(shí)重試
        backoff_factor=backoff_factor,
        status_forcelist=status_forcelist,
        allowed_methods=["GET", "POST"]  # 支持POST請(qǐng)求重試
    )
    adapter = HTTPAdapter(max_retries=retry)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    return session

# 使用示例
session = create_retry_session()
response = session.get('https://api.sf-express.com/track', params={'trackingNumber': 'SF123456789'})

from collections import deque
import time

class CircuitBreaker:
    def __init__(self, max_failures=3, reset_timeout=60):
        self.failures = deque(maxlen=max_failures)
        self.reset_timeout = reset_timeout

    def is_open(self):
        if len(self.failures) < self.failures.maxlen:
            return False
        # 如果最近max_failures次請(qǐng)求都失敗，且最后一次失敗在reset_timeout秒內(nèi)
        return (time.time() - self.failures[-1]) < self.reset_timeout

    def record_failure(self):
        self.failures.append(time.time())

    def record_success(self):
        self.failures.clear()

# 結(jié)合重試使用
breaker = CircuitBreaker(max_failures=3, reset_timeout=30)

def safe_request():
    if breaker.is_open():
        raise Exception("Service unavailable, circuit breaker open")
    
    try:
        response = create_retry_session().get(url)
        if response.status_code == 200:
            breaker.record_success()
            return response
        else:
            breaker.record_failure()
            raise Exception("API request failed")
    except Exception as e:
        breaker.record_failure()
        raise e

import random
from requests.adapters import HTTPAdapter

class ProxyPool:
    def __init__(self):
        self.proxies = [
            {"http": "http://1.1.1.1:8080", "https": "http://1.1.1.1:8080"},
            {"http": "http://2.2.2.2:8080", "https": "http://2.2.2.2:8080"},
            # 更多代理...
        ]
        self.failed_proxies = set()

    def get_proxy(self):
        available_proxies = [p for p in self.proxies if p not in self.failed_proxies]
        if not available_proxies:
            raise Exception("No available proxies")
        return random.choice(available_proxies)

    def mark_failed(self, proxy):
        self.failed_proxies.add(proxy)

# 使用示例
proxy_pool = ProxyPool()
session = requests.Session()

for _ in range(3):  # 嘗試3個(gè)不同代理
    try:
        proxy = proxy_pool.get_proxy()
        response = session.get(
            'https://api.sf-express.com/track',
            proxies=proxy,
            timeout=(3, 10)
        )
        if response.status_code == 200:
            print("Success with proxy:", proxy)
            break
        else:
            proxy_pool.mark_failed(proxy)
    except Exception:
        proxy_pool.mark_failed(proxy)
else:
    print("All proxies failed")

import requests

def make_tor_request(url):
    proxies = {
        'http': 'socks5h://127.0.0.1:9050',
        'https': 'socks5h://127.0.0.1:9050'
    }
    try:
        response = requests.get(url, proxies=proxies, timeout=(5, 15))
        print("Tor出口IP:", response.json()['origin'])
        return response
    except Exception as e:
        print("Tor請(qǐng)求失敗:", e)

# 需提前安裝Tor服務(wù)并啟動(dòng)
# sudo apt install tor  # Ubuntu系統(tǒng)
# sudo service tor start

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import logging
from datetime import datetime

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)

class SFExpressTracker:
    def __init__(self, app_key, app_secret):
        self.app_key = app_key
        self.app_secret = app_secret
        self.session = self._create_session()

    def _create_session(self):
        """創(chuàng)建帶重試和超時(shí)的會(huì)話"""
        retry_strategy = Retry(
            total=3,
            backoff_factor=1,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST"]
        )
        adapter = HTTPAdapter(max_retries=retry_strategy)
        session = requests.Session()
        session.mount("http://", adapter)
        session.mount("https://", adapter)
        return session

    def _generate_sign(self, params):
        """生成API簽名（簡化版）"""
        import hashlib
        sorted_params = sorted(params.items(), key=lambda x: x[0])
        raw_str = self.app_secret + ''.join([f"{k}{v}" for k, v in sorted_params]) + self.app_secret
        return hashlib.md5(raw_str.encode()).hexdigest().upper()

    def query_track(self, tracking_number):
        """查詢物流軌跡"""
        url = "https://bsp-ois.sf-express.com/bsp-ois/express/service/queryTrack"
        params = {
            "appKey": self.app_key,
            "trackNumber": tracking_number,
            "timestamp": str(int(datetime.now().timestamp()))
        }
        params["sign"] = self._generate_sign(params)

        try:
            response = self.session.get(
                url,
                params=params,
                timeout=(3, 10)  # 連接3秒，讀取10秒
            )
            response.raise_for_status()
            data = response.json()
            
            if data.get('success'):
                return data['data']['tracks']
            else:
                logging.warning(f"API返回錯(cuò)誤: {data.get('errorMsg', '未知錯(cuò)誤')}")
                return None
                
        except requests.exceptions.RequestException as e:
            logging.error(f"請(qǐng)求失敗: {str(e)}")
            return None

# 使用示例
if __name__ == "__main__":
    tracker = SFExpressTracker(app_key="YOUR_APP_KEY", app_secret="YOUR_APP_SECRET")
    tracks = tracker.query_track("SF123456789")
    if tracks:
        for step in tracks:
            print(f"{step['acceptTime']} {step['acceptAddress']} - {step['remark']}")

from urllib3.util.retry import Retry
import logging

class LoggingRetry(Retry):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.logger = logging.getLogger(__name__)

    def new(self, **kw):
        self.logger.debug(f"Creating new retry adapter with params: {kw}")
        return super().new(**kw)

    def increment(self, method, url, response=None, error=None, **kwargs):
        self.logger.warning(f"Retry attempt {self.total - self._remaining + 1} for {method} {url}")
        return super().increment(method, url, response, error, **kwargs)