#Python 异步爬虫教程：aiohttp 详解

import aiohttp
import asyncio

async def fetch(session: aiohttp.ClientSession, url: str) -> str:
    # 用 async with 发起请求，确保响应结束后自动释放连接
    async with session.get(url) as response:
        # 读取响应文本也需要 await，因为数据是异步传输的
        return await response.text()

async def main():
    # 使用 async with 创建并自动关闭 ClientSession
    # Session 内部会复用 Cookie、请求头、连接池，不要为每个请求都新建一个！
    async with aiohttp.ClientSession() as session:
        html = await fetch(session, "https://example.com")
        print(html[:200])

# Python 3.7+ 推荐直接用 asyncio.run() 启动事件循环
if __name__ == "__main__":
    asyncio.run(main())

async def search_baidu(session: aiohttp.ClientSession, keyword: str) -> str:
    params = {"wd": keyword, "ie": "utf-8"}
    async with session.get("https://www.baidu.com/s", params=params) as response:
        return await response.text()

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Safari/537.36"
}
async with session.get(url, headers=headers) as response:
    ...

# 总超时 10s，连接阶段最多 2s，逐块读取最多 3s
timeout = aiohttp.ClientTimeout(total=10, connect=2, sock_read=3)
async with session.get(url, timeout=timeout) as response:
    ...

form_data = {"username": "test", "password": "123456"}
async with session.post("https://example.com/login", data=form_data) as response:
    ...

json_payload = {"name": "Alice", "age": 25, "city": "Beijing"}
async with session.post("https://example.com/api/users", json=json_payload) as response:
    ...

# PUT 更新资源
async with session.put("https://example.com/api/users/1", json={"age": 26}) as response:
    ...

# DELETE 删除资源
async with session.delete("https://example.com/api/users/1") as response:
    ...

async with session.get(url) as response:
    print(f"状态码: {response.status}")                      # 例如 200
    print(f"响应头: {response.headers}")                     # dict 类型
    print(f"Content-Type: {response.headers.get('Content-Type')}")

    # 三种最常用的读取方式
    text = await response.text()          # 文本内容
    bytes_data = await response.read()    # 二进制内容（适合图片、PDF）
    json_data = await response.json()     # 自动解析 JSON

async def download_large_file(session: aiohttp.ClientSession, url: str, save_path: str) -> None:
    async with session.get(url) as response:
        response.raise_for_status()   # 检查状态码，不是 2xx 就抛异常
        with open(save_path, "wb") as f:
            # 每次读取 1 MB，避免一次性占用大量内存
            while chunk := await response.content.read(1024 * 1024):
                f.write(chunk)

# 最多允许 10 个请求同时运行
sem = asyncio.Semaphore(10)

async def safe_fetch(session: aiohttp.ClientSession, url: str) -> str:
    # 用 async with 自动获取和释放信号量
    async with sem:
        async with session.get(url) as response:
            return await response.text()

async def persistent_session_demo() -> None:
    # 统一设置默认 cookie、UA 和超时
    session = aiohttp.ClientSession(
        cookies={"session_id": "abc123"},
        headers={"User-Agent": "MyCrawler/1.0"},
        timeout=aiohttp.ClientTimeout(total=20)
    )
    try:
        # 两次请求会自动复用 cookie 和已有连接
        async with session.get("https://example.com/page1") as r1:
            ...
        async with session.get("https://example.com/page2") as r2:
            ...
    finally:
        # 如果没用 async with 创建 session，必须手动关闭
        await session.close()

# 普通代理
async with session.get(url, proxy="http://127.0.0.1:7890") as response:
    ...

# 带用户名密码的代理
async with session.get(url, proxy="http://user:pass@proxy.example.com:8080") as response:
    ...

async def fetch_with_error_handling(session: aiohttp.ClientSession, url: str) -> str | None:
    try:
        async with session.get(url, timeout=10) as response:
            response.raise_for_status()   # 自动抛出 4xx/5xx 错误
            return await response.text()
    except aiohttp.ClientConnectorError:
        print(f"无法连接到服务器: {url}")
    except aiohttp.ClientTimeoutError:
        print(f"请求超时: {url}")
    except aiohttp.HTTPError as e:
        print(f"HTTP 错误: {e.status} - {url}")
    except aiohttp.ClientError as e:
        print(f"其他请求错误: {e} - {url}")
    return None

import aiohttp
import asyncio

# 批量爬取的目标 URL
URLS = [
    "https://example.com",
    "https://httpbin.org/status/404",  # 故意测试 404 错误
    "https://httpbin.org/delay/2",     # 模拟 2 秒延迟
]

MAX_CONCURRENCY = 2   # 最大并发数
TIMEOUT = aiohttp.ClientTimeout(total=5)   # 单次请求总超时 5 秒

async def fetch(session: aiohttp.ClientSession, url: str, sem: asyncio.Semaphore) -> tuple[str, int | None]:
    async with sem:
        try:
            async with session.get(url, timeout=TIMEOUT) as response:
                response.raise_for_status()
                content = await response.text()
                return (url, len(content))
        except Exception as e:
            print(f"❌ 处理 {url} 失败: {str(e)}")
            return (url, None)

async def main():
    sem = asyncio.Semaphore(MAX_CONCURRENCY)
    async with aiohttp.ClientSession() as session:
        # 创建所有任务
        tasks = [fetch(session, url, sem) for url in URLS]
        # 并发执行，无论成败都会返回结果
        results = await asyncio.gather(*tasks)
        
        # 统计结果
        print("\n📊 爬取结果统计:")
        for url, length in results:
            if length:
                print(f"✅ {url}: 成功获取 {length} 字符")
            else:
                print(f"❌ {url}: 爬取失败")

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