pip install aiohttp aiofiles tqdm BeautifulSoup

import aiohttp
import asyncio
from bs4 import BeautifulSoup

async def get_latest_python_version():
    url = "https://www.python.org/downloads/"
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            html = await response.text()
            soup = BeautifulSoup(html, 'html.parser')
            # 提取最新穩(wěn)定版下載鏈接
            download_button = soup.select_one('.download-buttons a[href$=".exe"]')
            return download_button['href'] if download_button else None

import os
import aiofiles
from tqdm.asyncio import tqdm

async def download_file(session, url, filepath):
    # 檢查已下載部分
    downloaded = 0
    if os.path.exists(filepath):
        downloaded = os.path.getsize(filepath)
    
    headers = {'Range': f'bytes={downloaded}-'} if downloaded else {}
    
    async with session.get(url, headers=headers) as response:
        # 驗(yàn)證是否支持?jǐn)帱c(diǎn)續(xù)傳
        if downloaded and response.status != 206:
            print("Server doesn't support resume, restarting download")
            downloaded = 0
            headers = {}
            async with session.get(url) as new_response:
                response = new_response
        
        total_size = int(response.headers.get('content-length', 0)) + downloaded
        
        # 進(jìn)度條設(shè)置
        progress = tqdm(
            total=total_size, 
            unit='B', 
            unit_scale=True,
            desc=os.path.basename(filepath),
            initial=downloaded
        )
        
        # 異步寫入文件
        async with aiofiles.open(filepath, 'ab' if downloaded else 'wb') as f:
            while True:
                chunk = await response.content.read(1024 * 8)
                if not chunk:
                    break
                await f.write(chunk)
                progress.update(len(chunk))
        progress.close()
    
    # 校驗(yàn)文件完整性
    return await verify_download(filepath, total_size)

async def verify_download(filepath, expected_size):
    actual_size = os.path.getsize(filepath)
    if actual_size == expected_size:
        print(f"? Download verified: {actual_size} bytes")
        return True
    print(f"? Download corrupted: expected {expected_size}, got {actual_size}")
    return False

async def main():
    # 獲取最新版本下載鏈接
    download_url = await get_latest_python_version()
    if not download_url:
        print("Failed to get download URL")
        return
    
    filename = download_url.split('/')[-1]
    save_path = os.path.join(os.getcwd(), filename)
    
    # 檢查文件是否已完整存在
    if os.path.exists(save_path):
        file_size = os.path.getsize(save_path)
        async with aiohttp.ClientSession() as session:
            async with session.head(download_url) as response:
                total_size = int(response.headers.get('content-length', 0))
                if file_size == total_size:
                    print(f"File already exists and is complete: {filename}")
                    return
    
    # 執(zhí)行下載
    print(f"Starting download: {download_url}")
    async with aiohttp.ClientSession() as session:
        success = await download_file(session, download_url, save_path)
        if success:
            print(f"Download completed successfully: {save_path}")
        else:
            print("Download failed, please try again")

if __name__ == "__main__":
    asyncio.run(main())

python python_downloader.py

# 示例代碼片段
async def download_chunk(session, url, start, end, filepath):
    headers = {'Range': f'bytes={start}-{end}'}
    # ...分塊下載實(shí)現(xiàn)...

import hashlib
async def check_md5(filepath, expected_md5):
    hash_md5 = hashlib.md5()
    async with aiofiles.open(filepath, "rb") as f:
        while chunk := await f.read(8192):
            hash_md5.update(chunk)
    return hash_md5.hexdigest() == expected_md5

proxy = "http://user:pass@proxy:port"
connector = aiohttp.TCPConnector(ssl=False)
async with aiohttp.ClientSession(connector=connector, proxy=proxy) as session:
    # ...