#Python 异步爬虫教程 (2024 更新版)

import aiohttp
import asyncio

async def fetch(url):
    # 异步上下文管理器管理Session（类比requests.Session）
    # 会自动复用连接池、自动清理资源
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            # 等待响应文本读取完成
            return await response.text()

async def main():
    html = await fetch('https://example.com')
    print(html[:200])  # 只打印前200字，避免刷屏

# 3.7+ 标准入口
asyncio.run(main())

import aiohttp
import asyncio
from bs4 import BeautifulSoup
from typing import List, Optional, Dict

# 限制并发数，避免被目标网站封IP
CONCURRENT_LIMIT = 20  # 单总连接数
HOST_LIMIT = 5          # 单个域名的最大连接数（非常重要！防封）
TIMEOUT = aiohttp.ClientTimeout(total=30, connect=10)

async def fetch_single(
    session: aiohttp.ClientSession,
    url: str
) -> Optional[str]:
    """爬取单个URL并处理异常"""
    try:
        async with session.get(url, timeout=TIMEOUT) as resp:
            if resp.status == 200:
                # 可以根据需要改成await resp.json()/await resp.read()
                return await resp.text()
            # 记录非200状态码
            print(f"⚠️  {url} 返回状态码 {resp.status}")
            return None
    except asyncio.TimeoutError:
        print(f"⏱️  {url} 超时")
        return None
    except Exception as e:
        print(f"❌  {url} 未知错误: {str(e)[:100]}")
        return None

async def parse_single(html: str) -> Optional[Dict]:
    """异步解析单个页面（这里用bs4是同步，但小数据量不影响）"""
    if not html:
        return None
    soup = BeautifulSoup(html, 'lxml')  # lxml比html.parser快很多
    # 👇 这里替换成你的解析逻辑，比如
    title = soup.title.string if soup.title else None
    return {"title": title}

async def batch_crawl(urls: List[str]) -> List[Dict]:
    """批量爬取+解析的主协程"""
    # 配置连接池
    connector = aiohttp.TCPConnector(
        limit=CONCURRENT_LIMIT,
        limit_per_host=HOST_LIMIT,
        force_close=False,  # 开启长连接复用
        enable_cleanup_closed=True
    )
    
    # 批量执行任务
    async with aiohttp.ClientSession(connector=connector) as session:
        # 生成所有爬取任务
        fetch_tasks = [fetch_single(session, url) for url in urls]
        # 等待所有爬取任务完成（gather会收集所有结果，即使有失败）
        raw_pages = await asyncio.gather(*fetch_tasks)
        
        # 生成所有解析任务（过滤掉None的页面）
        parse_tasks = [parse_single(page) for page in raw_pages if page]
        # 等待所有解析完成
        results = await asyncio.gather(*parse_tasks)
        # 最后过滤一下解析失败的None
        return [res for res in results if res]

if __name__ == "__main__":
    # 测试用的10个带1秒延迟的URL
    test_urls = [f"https://httpbin.org/delay/1?num={i}" for i in range(10)]
    # 爬取+计时
    import time
    start = time.time()
    final_data = asyncio.run(batch_crawl(test_urls))
    end = time.time()
    # 输出结果
    print(f"\n✅ 成功爬取+解析 {len(final_data)} 条数据")
    print(f"⏱️  总耗时: {end - start:.2f} 秒")