项目：异步网页爬虫

# fetcher.py —— 真实版本（需要 pip install aiohttp）
import aiohttp
import asyncio
 
async def fetch(session, url):
    async with session.get(url, timeout=10) as resp:
        resp.raise_for_status()
        return await resp.text()
 
async def fetch_all(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# parser.py
from typing import Any, Dict
 
def parse_post(raw: Dict[str, Any]) -> Dict[str, Any]:
    """从原始数据里挑出我们感兴趣的字段"""
    return {
        "id":   raw["id"],
        "title": raw["title"],
        "body":  raw["body"],
        "len":   len(raw["body"]),
    }

# aggregator.py
from collections import Counter
from typing import List, Dict, Any
 
def summarize(posts: List[Dict[str, Any]]) -> Dict[str, Any]:
    if not posts:
        return {"count": 0, "avg_len": 0, "longest": None}
    lengths = [p["len"] for p in posts]
    longest = max(posts, key=lambda p: p["len"])
    return {
        "count":   len(posts),
        "avg_len": sum(lengths) / len(lengths),
        "longest": {"id": longest["id"], "title": longest["title"]},
    }

# main.py
import asyncio
import json
from pathlib import Path
from fetcher import fetch_all
from parser import parse_post
from aggregator import summarize
 
URLS = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]
 
async def run():
    print("开始抓取", len(URLS), "条数据")
    raws = await fetch_all(URLS)
    posts = [parse_post(r) for r in raws if isinstance(r, dict)]
    summary = summarize(posts)
    print("汇总:", summary)
 
    out = Path("/tmp/scraper_result.json")
    out.write_text(
        json.dumps({"summary": summary, "posts": posts},
                   ensure_ascii=False, indent=2),
        encoding="utf-8"
    )
    print("已写入:", out)
 
if __name__ == "__main__":
    asyncio.run(run())

import asyncio
import json
from pathlib import Path
from typing import Any, Dict, List
 
# ---- 模拟的"远端数据库"，对应 jsonplaceholder 的 posts ----
FAKE_DB = {
    i: {
        "id": i,
        "title": f"Post #{i}: 学习 Python 异步",
        # 用 (i % 4) + 1 让长度确定，方便演示
        "body":  "在沙箱里我们用 asyncio.sleep 模拟网络延迟。" * ((i % 4) + 1),
        "userId": (i % 3) + 1,
    }
    for i in range(1, 11)
}
 
async def fetch(url: str) -> Dict[str, Any]:
    """沙箱版 fetch：解析 URL 拿 id, 模拟网络延迟, 返回数据"""
    # URL 形如 https://jsonplaceholder.typicode.com/posts/3
    post_id = int(url.rstrip("/").split("/")[-1])
    await asyncio.sleep(0.05)          # 模拟 50ms 网络
    return FAKE_DB[post_id]
 
async def fetch_all(urls: List[str]) -> List[Any]:
    """asyncio.gather 并发抓取全部"""
    tasks = [fetch(u) for u in urls]
    return await asyncio.gather(*tasks, return_exceptions=True)
 
def parse_post(raw: Dict[str, Any]) -> Dict[str, Any]:
    return {"id": raw["id"], "title": raw["title"],
            "body": raw["body"], "len": len(raw["body"])}
 
def summarize(posts: List[Dict[str, Any]]) -> Dict[str, Any]:
    if not posts:
        return {"count": 0, "avg_len": 0, "longest": None}
    longest = max(posts, key=lambda p: p["len"])
    return {
        "count":   len(posts),
        "avg_len": sum(p["len"] for p in posts) / len(posts),
        "longest": {"id": longest["id"], "title": longest["title"]},
    }
 
async def main():
    urls = [f"https://jsonplaceholder.typicode.com/posts/{i}" for i in range(1, 11)]
    print(f"开始抓取 {len(urls)} 个 URL ...")
 
    raws = await fetch_all(urls)
    print("抓取完成 (并发 50ms × 1 轮)")
 
    # 区分成功 / 失败
    posts  = [parse_post(r) for r in raws if isinstance(r, dict)]
    failed = [r for r in raws if isinstance(r, BaseException)]
    print(f"成功 {len(posts)} 条, 失败 {len(failed)} 条")
 
    summary = summarize(posts)
    print("")
    print("--- 汇总 ---")
    print(f"  count:   {summary['count']}")
    print(f"  avg_len: {summary['avg_len']:.1f}")
    print(f"  longest: #{summary['longest']['id']} {summary['longest']['title']}")
 
    out = Path("/tmp/scraper_result.json")
    out.write_text(
        json.dumps({"summary": summary, "posts": posts}, ensure_ascii=False, indent=2),
        encoding="utf-8"
    )
    print("")
    print(f"已写入: {out} ({out.stat().st_size} bytes)")
 
asyncio.run(main())

import asyncio, random
from typing import Any, Dict, List
 
FAIL_RATE = 0.2
SLOW_RATE = 0.2
 
async def raw_fetch(url: str) -> Dict[str, Any]:
    """低层：可能慢、可能失败"""
    delay = random.uniform(0.05, 0.5) if random.random() < SLOW_RATE else random.uniform(0.01, 0.05)
    await asyncio.sleep(delay)
    if random.random() < FAIL_RATE:
        raise RuntimeError(f"网络抖动: {url}")
    return {"id": int(url.rsplit("/", 1)[1]), "ok": True}
 
async def fetch_with_retry(url: str, *, max_retries=3, timeout=0.2) -> Dict[str, Any]:
    """带超时和重试的高层 fetch"""
    last = None
    for i in range(max_retries):
        try:
            return await asyncio.wait_for(raw_fetch(url), timeout=timeout)
        except (asyncio.TimeoutError, RuntimeError) as e:
            last = e
            print(f"  {url} 第 {i+1} 次失败: {type(e).__name__}: {e}")
    raise last
 
async def main():
    random.seed(7)
    urls = [f"http://api.example.com/posts/{i}" for i in range(8)]
    results = await asyncio.gather(
        *(fetch_with_retry(u) for u in urls),
        return_exceptions=True
    )
    ok = [r for r in results if isinstance(r, dict)]
    bad = [r for r in results if isinstance(r, BaseException)]
    print(f"成功 {len(ok)} / 失败 {len(bad)}")
 
asyncio.run(main())

import asyncio, time
 
# 同时最多 3 个任务
sem = asyncio.Semaphore(3)
 
async def worker(i):
    async with sem:        # 抢到信号量才执行
        await asyncio.sleep(0.1)
        return f"done-{i}"
 
async def main():
    start = time.perf_counter()
    # 10 个任务，但同时只有 3 个在跑
    results = await asyncio.gather(*(worker(i) for i in range(10)))
    print(f"完成 {len(results)} 个, 总耗时 {time.perf_counter()-start:.2f}s")
 
asyncio.run(main())

import asyncio, random
from typing import List, Dict, Any
 
async def raw_fetch(url: str) -> Dict[str, Any]:
    await asyncio.sleep(random.uniform(0.05, 0.3))
    if random.random() < 0.2:
        raise RuntimeError("网络错误")
    return {"id": int(url.rsplit("/", 1)[1])}
 
async def fetch_many(urls: List[str], concurrency: int = 5, timeout: float = 0.2):
    # 你的实现
    pass
 
async def main():
    random.seed(1)
    urls = [f"http://api.example.com/{i}" for i in range(20)]
    result = await fetch_many(urls)
    print(result)
 
asyncio.run(main())