Python 网络请求与爬虫基础

import requests

# GET（永远带 timeout!）
r = requests.get(
    "https://httpbin.org/get",
    params={"page": 2, "size": 10},        # ?page=2&size=10
    headers={"User-Agent": "my-spider/1.0"},
    timeout=5,
)
r.status_code          # 200
r.json()               # 解析响应体
r.text                 # 文本
r.headers["Content-Type"]

# POST JSON
r = requests.post(
    "https://httpbin.org/post",
    json={"name": "zy", "level": 5},       # 自动序列化 + Content-Type
    timeout=5,
)

# POST 表单 / 文件上传
requests.post(url, data={"k": "v"}, timeout=5)
requests.post(url, files={"f": open("a.png", "rb")}, timeout=5)

# 常见 HTTP 动词
requests.put(url, json={...}); requests.delete(url); requests.head(url)

with requests.Session() as s:
    s.headers.update({"Authorization": "Bearer xxx"})

    for i in range(100):
        r = s.get(f"https://api.example.com/items/{i}", timeout=5)
        process(r.json())
# TCP 连接复用: 快, 且对目标服务器友好

from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

retry = Retry(
    total=3,
    backoff_factor=1,               # 退避: 1s, 2s, 4s...
    status_forcelist=[429, 500, 502, 503, 504],
    allowed_methods=["GET"],
)

s = requests.Session()
s.mount("https://", HTTPAdapter(max_retries=retry))
s.mount("http://",  HTTPAdapter(max_retries=retry))

try:
    r = s.get("https://api.example.com/data", timeout=(3, 10))
    r.raise_for_status()           # 4xx/5xx 直接抛异常
    data = r.json()
except requests.Timeout:
    print("超时")
except requests.HTTPError as e:
    print(f"HTTP错误: {e.response.status_code}")
except requests.RequestException as e:
    print(f"请求失败: {e}")

pip install beautifulsoup4 lxml -i https://mirrors.aliyun.com/pypi/simple/

from bs4 import BeautifulSoup

html = """
<div class="post-list">
  <article class="post">
    <h2><a href="/posts/1">Linux 权限详解</a></h2>
    <span class="date">2023-03-05</span>
    <span class="tag">Linux</span>
  </article>
  <article class="post">
    <h2><a href="/posts/2">Python 虚拟环境</a></h2>
    <span class="date">2023-03-18</span>
    <span class="tag">Python</span>
  </article>
</div>
"""

soup = BeautifulSoup(html, "lxml")          # lxml 解析器最快

# CSS 选择器（推荐, 和前端知识通用）
for art in soup.select("article.post"):
    title = art.select_one("h2 a").get_text(strip=True)
    link  = art.select_one("h2 a")["href"]
    date  = art.select_one(".date").get_text()
    tag   = art.select_one(".tag").get_text()
    print(date, tag, title, link)

# 常用 API
soup.find("h2")                     # 第一个 h2
soup.find_all("span", class_="tag") # 全部匹配
soup.find(id="main")
tag.get_text(), tag["href"], tag.attrs

#!/usr/bin/env python3
"""抓取博客文章列表 -> 存 JSON"""
import json, time
import requests
from bs4 import BeautifulSoup
from pathlib import Path

BASE = "https://example-blog.cn"

def fetch_page(session: requests.Session, url: str) -> list[dict]:
    r = session.get(url, timeout=10)
    r.raise_for_status()
    r.encoding = r.apparent_encoding       # 自动修正乱码

    soup = BeautifulSoup(r.text, "lxml")
    items = []
    for art in soup.select("article.post"):
        items.append({
            "title": art.select_one("h2 a").get_text(strip=True),
            "url":   BASE + art.select_one("h2 a")["href"],
            "date":  art.select_one(".date").get_text(strip=True),
            "tags":  [t.get_text() for t in art.select(".tag")],
        })
    return items

def crawl(max_pages: int = 5) -> list[dict]:
    results = []
    with requests.Session() as s:
        s.headers["User-Agent"] = "Mozilla/5.0 (compatible; zy-crawler/1.0)"
        for page in range(1, max_pages + 1):
            url = f"{BASE}/page/{page}/"
            try:
                items = fetch_page(s, url)
            except requests.RequestException as e:
                print(f"[!] 第{page}页失败: {e}")
                continue
            if not items:
                break                       # 没有更多了
            results.extend(items)
            print(f"[+] 第{page}页: {len(items)} 条")
            time.sleep(2)                   # 爬虫礼仪: 别把人家打挂
    return results

if __name__ == "__main__":
    data = crawl()
    Path("posts.json").write_text(
        json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"共抓取 {len(data)} 条, 已存 posts.json")

# 方案1: 线程池（IO 密集型首选, 简单）
from concurrent.futures import ThreadPoolExecutor, as_completed

urls = [f"https://httpbin.org/delay/1?i={i}" for i in range(20)]

with requests.Session() as s, ThreadPoolExecutor(max_workers=8) as pool:
    futures = {pool.submit(s.get, u, timeout=10): u for u in urls}
    for f in as_completed(futures):
        try:
            print(f.result().status_code)
        except Exception as e:
            print("失败:", futures[f], e)
# 20 个耗时1s的请求: 串行20s -> 并发约3s

# 方案2: httpx 异步（大规模时更省资源）
import httpx, asyncio

async def fetch(client, url):
    r = await client.get(url, timeout=10)
    return r.status_code

async def main():
    async with httpx.AsyncClient() as client:
        results = await asyncio.gather(*[
            fetch(client, u) for u in urls[:20]
        ])
    print(results)

asyncio.run(main())