# 原始代碼 - 有內(nèi)存溢出風(fēng)險(xiǎn)
def download_file(download_url, file_type):
    try:
        response = requests.get(download_url, timeout=10)
        with open(save_path, 'wb') as f:
            f.write(response.content)  # 這里出問(wèn)題了！
        return True
    except Exception as e:
        log.error(f"下載失敗: {str(e)}")
        return False

# 模擬requests內(nèi)部處理（簡(jiǎn)化版）
class Response:
    def __init__(self, raw_response):
        self.raw_response = raw_response
        self._content = None
    
    @property
    def content(self):
        if self._content is None:
            # 這里會(huì)一次性讀取所有數(shù)據(jù)到內(nèi)存！
            self._content = b"".join(self.iter_content(CHUNK_SIZE)) or b""
        return self._content

  # 使用 stream=True 進(jìn)行流式下載
   with requests.get(download_url, timeout=10, stream=True) as response

# 關(guān)鍵代碼段
with requests.get(download_url, stream=True) as response:
    with open(save_path, 'wb') as f:
        for chunk in response.iter_content(chunk_size=8192):
            if chunk:
                f.write(chunk)

import psutil
import requests

def memory_usage():
    """獲取當(dāng)前進(jìn)程內(nèi)存使用"""
    process = psutil.Process()
    return process.memory_info().rss / 1024 / 1024  # MB

# 測(cè)試傳統(tǒng)方式
print("傳統(tǒng)下載方式內(nèi)存使用:")
print(f"開(kāi)始前: {memory_usage():.1f}MB")
response = requests.get("http://speedtest.ftp.otenet.gr/files/test100Mb.db")
content = response.content  # 一次性加載到內(nèi)存
print(f"下載后: {memory_usage():.1f}MB")

# 測(cè)試流式下載
print("\n流式下載方式內(nèi)存使用:")
print(f"開(kāi)始前: {memory_usage():.1f}MB")
with requests.get("http://speedtest.ftp.ebpytes.gr/files/test100Mb.db", stream=True) as r:
    with open("test.file", "wb") as f:
        for chunk in r.iter_content(chunk_size=8192):
            f.write(chunk)
print(f"下載后: {memory_usage():.1f}MB")

# 分塊大小優(yōu)化策略
def get_optimal_chunk_size(file_size: int) -> int:
    """
    根據(jù)文件大小動(dòng)態(tài)調(diào)整分塊大小
    
    分塊大小經(jīng)驗(yàn)值：
    - 小文件(<10MB): 4KB-8KB
    - 中等文件(10MB-100MB): 32KB-64KB  
    - 大文件(>100MB): 128KB-1MB
    """
    if file_size < 10 * 1024 * 1024:  # < 10MB
        return 8 * 1024  # 8KB
    elif file_size < 100 * 1024 * 1024:  # 10MB-100MB
        return 64 * 1024  # 64KB
    else:  # > 100MB
        return 256 * 1024  # 256KB

def resume_download(self, download_url: str, save_path: str, 
                   timeout: int = 30) -> bool:
    """
    支持?jǐn)帱c(diǎn)續(xù)傳的下載
    
    原理：檢查已下載部分，設(shè)置Range頭繼續(xù)下載
    """
    # 檢查文件是否已部分下載
    downloaded_size = 0
    if os.path.exists(save_path):
        downloaded_size = os.path.getsize(save_path)
    
    headers = {}
    if downloaded_size > 0:
        headers['Range'] = f'bytes={downloaded_size}-'
    
    try:
        with requests.get(download_url, headers=headers, 
                         timeout=timeout, stream=True) as response:
            
            # 檢查服務(wù)器是否支持?jǐn)帱c(diǎn)續(xù)傳
            if response.status_code == 206:  # Partial Content
                log.info(f"從 {downloaded_size} 字節(jié)處繼續(xù)下載")
                mode = 'ab'  # 追加模式
            elif response.status_code == 200:
                log.info("開(kāi)始全新下載")
                mode = 'wb'  # 覆蓋模式
            else:
                response.raise_for_status()
            
            with open(save_path, mode) as f:
                for chunk in response.iter_content(chunk_size=self.chunk_size):
                    if chunk:
                        f.write(chunk)
            
            return True
            
    except Exception as e:
        log.error(f"斷點(diǎn)續(xù)傳失敗: {str(e)}")
        return False

import sys
from tqdm import tqdm  # 進(jìn)度條庫(kù)

def download_with_progress(self, download_url: str, save_path: str) -> bool:
    """
    帶進(jìn)度條的下載
    """
    try:
        with requests.get(download_url, stream=True) as response:
            response.raise_for_status()
            
            # 獲取文件總大小
            total_size = int(response.headers.get('content-length', 0))
            
            with open(save_path, 'wb') as f, \
                 tqdm(total=total_size, unit='B', 
                     unit_scale=True, desc=os.path.basename(save_path)) as pbar:
                
                for chunk in response.iter_content(chunk_size=self.chunk_size):
                    if chunk:
                        f.write(chunk)
                        pbar.update(len(chunk))
            
            return True
            
    except Exception as e:
        log.error(f"下載失敗: {str(e)}")
        return False

# 性能測(cè)試代碼
import time
import statistics

def benchmark_download(url, file_size_mb, method='traditional'):
    """下載性能基準(zhǔn)測(cè)試"""
    start_time = time.time()
    start_memory = memory_usage()
    
    if method == 'traditional':
        # 傳統(tǒng)方式
        response = requests.get(url)
        with open('test.tmp', 'wb') as f:
            f.write(response.content)
    else:
        # 流式方式
        with requests.get(url, stream=True) as response:
            with open('test.tmp', 'wb') as f:
                for chunk in response.iter_content(chunk_size=8192):
                    f.write(chunk)
    
    end_time = time.time()
    end_memory = memory_usage()
    
    # 清理
    os.remove('test.tmp')
    
    return {
        'method': method,
        'file_size_mb': file_size_mb,
        'time_seconds': end_time - start_time,
        'memory_peak_mb': end_memory - start_memory,
        'speed_mbps': file_size_mb / (end_time - start_time)
    }

# 測(cè)試結(jié)果示例
results = [
    {'file_size': '10MB', '傳統(tǒng)方式': '0.8s/15MB', '流式方式': '0.9s/8MB'},
    {'file_size': '100MB', '傳統(tǒng)方式': '7.2s/105MB', '流式方式': '7.5s/8MB'},
    {'file_size': '1GB', '傳統(tǒng)方式': '內(nèi)存溢出', '流式方式': '72.3s/8MB'},
    {'file_size': '5GB', '傳統(tǒng)方式': '內(nèi)存溢出', '流式方式': '361.5s/8MB'},
]

# 錯(cuò)誤做法
response = requests.get(url, stream=True)
# 如果響應(yīng)是404/500，這里也會(huì)嘗試讀取響應(yīng)體

# 正確做法
response = requests.get(url, stream=True)
response.raise_for_status()  # 非2xx狀態(tài)碼會(huì)拋出異常

# 太?。篒/O操作頻繁，性能差
for chunk in response.iter_content(chunk_size=1024):  # 1KB太小

# 太大：內(nèi)存占用高，失去流式優(yōu)勢(shì)  
for chunk in response.iter_content(chunk_size=10 * 1024 * 1024):  # 10MB太大

# 適中：根據(jù)網(wǎng)絡(luò)和磁盤(pán)性能調(diào)整
for chunk in response.iter_content(chunk_size=8192):  # 8KB適中

# 添加重試機(jī)制
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def download_with_retry(url, save_path):
    with requests.get(url, stream=True, timeout=30) as response:
        response.raise_for_status()
        # ... 下載邏輯