第 16 章 · 网络编程入门

# 手写一个 HTTP 请求，直接体验第 3~7 步（不要在生产用，仅教学）
raw_request = (
    "GET / HTTP/1.1\r\n"
    "Host: example.com\r\n"
    "User-Agent: python-tutorial/1.0\r\n"
    "Accept: text/html\r\n"
    "Connection: close\r\n"
    "\r\n"                       # 空行，标志请求头结束
)
print(raw_request.encode())      # 报文其实是 bytes，不是 str

# tcp_chat_server.py —— 线程版广播聊天服务器
"""
运行：python tcp_chat_server.py
客户端：telnet 127.0.0.1 9000
"""
from __future__ import annotations

import socket
import threading
from datetime import datetime

HOST, PORT = "0.0.0.0", 9000

clients: list[socket.socket] = []
lock = threading.Lock()          # 保护 clients 列表


def log(msg: str) -> None:
    print(f"[{datetime.now():%H:%M:%S}] {msg}")


def broadcast(data: bytes, sender: socket.socket) -> None:
    """把数据发给除 sender 外的所有客户端。"""
    with lock:
        targets = [c for c in clients if c is not sender]
    dead = []
    for c in targets:
        try:
            c.sendall(data)
        except OSError:
            dead.append(c)
    for c in dead:                       # 清理已断开的连接
        with lock:
            if c in clients:
                clients.remove(c)
        c.close()


def handle(conn: socket.socket, addr: tuple[str, int]) -> None:
    log(f"新连接 {addr}")
    try:
        while True:
            data = conn.recv(4096)
            if not data:                 # 对端正常关闭
                break
            text = data.decode("utf-8", errors="replace").rstrip("\n")
            log(f"{addr} 说: {text}")
            payload = f"[{addr[1]}] {text}\n".encode("utf-8")
            broadcast(payload, conn)
            conn.sendall(b"OK\n")       # 简单 ACK，演示半双工
    except ConnectionResetError:
        log(f"{addr} 强制断开")
    finally:
        with lock:
            if conn in clients:
                clients.remove(conn)
        conn.close()
        log(f"关闭 {addr}，剩余客户端 {len(clients)}")


def main() -> None:
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((HOST, PORT))
    srv.listen(64)
    srv.settimeout(1.0)                 # 每秒检查一次退出标志
    log(f"服务器启动，监听 {HOST}:{PORT}")

    running = True
    try:
        while running:
            try:
                conn, addr = srv.accept()
            except socket.timeout:
                continue
            with lock:
                clients.append(conn)
            threading.Thread(target=handle, args=(conn, addr), daemon=True).start()
    except KeyboardInterrupt:
        running = False
        log("收到 Ctrl+C，正在关闭……")
    finally:
        srv.close()
        with lock:
            for c in clients:
                c.close()
        log("服务器已关闭")


if __name__ == "__main__":
    main()

# tcp_chat_client.py
"""
运行：python tcp_chat_client.py
"""
import socket
import sys

HOST, PORT = "127.0.0.1", 9000


def main() -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(5)
    try:
        sock.connect((HOST, PORT))
    except OSError as exc:
        print(f"连接失败: {exc}")
        sys.exit(1)

    print(f"已连接到 {HOST}:{PORT}，输入内容回车发送，quit 退出")
    try:
        while True:
            line = input("> ")
            if line.strip().lower() == "quit":
                break
            sock.sendall((line + "\n").encode("utf-8"))
            try:
                ack = sock.recv(4096).decode("utf-8", errors="replace")
                print("服务器:", ack.strip())
            except socket.timeout:
                print("等待 ACK 超时")
    except KeyboardInterrupt:
        print("\n再见")
    finally:
        sock.close()


if __name__ == "__main__":
    main()

# framed_protocol.py —— 长度前缀协议，客户端与服务端通用
"""
协议格式：| 4 字节无符号大端长度 N | N 字节 UTF-8 报文 |
"""
from __future__ import annotations

import socket
import struct

HEADER = struct.Struct("!I")          # ! = 网络字节序（big-endian），I = 4 字节无符号整数
MAX_BODY = 8 * 1024 * 1024            # 8MB 上限，防内存攻击


def encode(msg: str) -> bytes:
    """把一条消息编码成字节。"""
    body = msg.encode("utf-8")
    if len(body) > MAX_BODY:
        raise ValueError("消息过大")
    return HEADER.pack(len(body)) + body


def recv_exactly(sock: socket.socket, n: int) -> bytes | None:
    """精确读取 n 个字节；连接关闭返回 None。"""
    chunks: list[bytes] = []
    remaining = n
    while remaining > 0:
        chunk = sock.recv(remaining)       # 可能一次读不到全部
        if not chunk:
            return None                    # 对端关闭
        chunks.append(chunk)
        remaining -= len(chunk)
    return b"".join(chunks)


def recv_message(sock: socket.socket) -> str | None:
    """读取一条完整消息。"""
    header = recv_exactly(sock, HEADER.size)
    if header is None:
        return None
    (length,) = HEADER.unpack(header)
    if length > MAX_BODY:
        raise ValueError(f"长度非法: {length}")
    body = recv_exactly(sock, length)
    if body is None:
        return None
    return body.decode("utf-8")


def serve(host: str = "127.0.0.1", port: int = 9001) -> None:
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((host, port))
    srv.listen(5)
    print(f"[服务端] 监听 {host}:{port}")

    while True:
        conn, addr = srv.accept()
        with conn:
            print(f"[服务端] {addr} 已连接")
            while True:
                try:
                    msg = recv_message(conn)
                except ValueError as exc:
                    print("[服务端] 协议错误:", exc)
                    break
                if msg is None:
                    print(f"[服务端] {addr} 断开")
                    break
                print(f"[服务端] 收到完整消息: {msg}")
                # 回一条更长的消息，演示大报文
                conn.sendall(encode("回显：" + msg))


def client(host: str = "127.0.0.1", port: int = 9001, messages: list[str] | None = None) -> None:
    with socket.create_connection((host, port), timeout=5) as sock:
        for m in messages or ["你好", "second message", "第三条"]:
            payload = encode(m)
            print(f"[客户端] 发送 {len(payload)} 字节: {m}")
            sock.sendall(payload)
            reply = recv_message(sock)
            print(f"[客户端] 收到: {reply}")


if __name__ == "__main__":
    import threading

    t = threading.Thread(target=serve, daemon=True)
    t.start()
    import time; time.sleep(0.3)
    client()
    print("演示完成")

常见坑「用 recv(1024) 读，然后按行 split」在消息体可能包含换行或高频小包时必然出错。TCP 的语义里没有任何「一次 recv 对应一次 send」的保证。所有自研 TCP 协议都必须自己定界。

# epoll_note.py —— 说明：C10K 的正确解法是交给成熟的服务器
"""
如果你需要真正处理上万个并发连接：
  1. 不要用 raw socket 手写 epoll 事件循环；
  2. 写一个 WSGI/ASGI 应用，用 uvicorn（异步）或 gunicorn（多进程）跑；
  3. 需要极端性能时用 Go / Rust 重写边缘层，Python 只做业务。
示例：第 18 章会用 Flask/FastAPI 走这条路线。
"""
import socket

# 仅演示 epoll 的注册与等待
if __name__ == "__main__":
    ep = __import__("select").epoll()
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("127.0.0.1", 9002))
    srv.listen(16)
    srv.setblocking(False)
    ep.register(srv, __import__("select").EPOLLIN)
    print("epoll 已注册监听 socket（生产请直接用 asyncio 或 Web 框架）")
    ep.close()
    srv.close()

# udp_echo.py —— 一个进程同时跑服务端与客户端，便于直接运行
"""
运行：python udp_echo.py
"""
from __future__ import annotations

import socket
import threading
import time
from datetime import datetime


def udp_server(host: str = "0.0.0.0", port: int = 9003) -> None:
    """回显服务：收到什么就回什么。"""
    srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind((host, port))
    print(f"[UDP 服务端] 监听 {host}:{port}")

    while True:
        try:
            data, addr = srv.recvfrom(65535)     # UDP 没有 accept
        except OSError as exc:
            print("服务端出错:", exc)
            break

        if data == b"__quit__":
            print("收到退出指令")
            srv.close()
            break

        text = data.decode("utf-8", errors="replace")
        print(f"[UDP 服务端] {datetime.now():%H:%M:%S} 来自 {addr}: {text}")
        reply = f"echo({len(text)} 字): {text}".encode("utf-8")
        srv.sendto(reply, addr)                  # 必须带目标地址


def udp_client(host: str = "127.0.0.1", port: int = 9003, count: int = 3) -> None:
    cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    cli.settimeout(2.0)                         # UDP 也应该设超时
    for i in range(count):
        msg = f"第 {i} 次心跳 {datetime.now():%H:%M:%S}".encode("utf-8")
        try:
            cli.sendto(msg, (host, port))
            data, _ = cli.recvfrom(65535)
            print(f"[UDP 客户端] 收到: {data.decode('utf-8')}")
        except socket.timeout:
            print("[UDP 客户端] 超时：可能丢包了（UDP 允许丢）")
        time.sleep(0.1)
    cli.sendto(b"__quit__", (host, port))
    cli.close()


if __name__ == "__main__":
    t = threading.Thread(target=udp_server, daemon=True)
    t.start()
    time.sleep(0.3)
    udp_client()
    t.join(timeout=2)
    print("演示结束")

[UDP 服务端] 监听 0.0.0.0:9003
[UDP 客户端] 收到: echo(16 字): 第 0 次心跳 10:22:31
[UDP 服务端] 10:22:31 来自 ('127.0.0.1', 53124): 第 0 次心跳 10:22:31
...
演示结束

# udp_reliable.py —— 极简可靠 UDP（重传 + 序列号去重）
from __future__ import annotations

import socket
import struct

SEQ = struct.Struct("!I")          # 序列号


def send_reliably(sock: socket.socket, addr, msg: str,
                  seq: int, retry: int = 3, timeout: float = 0.5) -> bool:
    payload = SEQ.pack(seq) + msg.encode("utf-8")
    sock.sendto(payload, addr)
    for _ in range(retry):
        try:
            sock.settimeout(timeout)
            ack, _ = sock.recvfrom(4096)
            return SEQ.unpack(ack[:SEQ.size])[0] == seq
        except socket.timeout:
            sock.sendto(payload, addr)     # 超时则重传
    return False                            # 重试耗尽，判定失败


def main() -> None:
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(2)
    peer = ("127.0.0.1", 9004)

    # 这里用一个极简"服务端"来演示，实际请双开进程
    threading_srv = __import__("threading").Thread(
        target=lambda: _mini_server(sock), daemon=True)
    threading_srv.start()

    ok = send_reliably(sock, peer, "可靠消息", seq=1)
    print("发送成功 =", ok)
    sock.close()


def _mini_server(sock: socket.socket) -> None:
    try:
        while True:
            data, addr = sock.recvfrom(4096)
            (seq,) = SEQ.unpack(data[:SEQ.size])
            body = data[SEQ.size:].decode("utf-8", errors="replace")
            print(f"[服务端] seq={seq} body={body}")
            sock.sendto(SEQ.pack(seq), addr)     # 回 ACK
    except OSError:
        pass


if __name__ == "__main__":
    main()

# urllib_get.py
import urllib.request
import json

URL = "https://httpbin.org/get"


def main() -> None:
    # 最朴素：urlopen 直接读
    with urllib.request.urlopen(URL, timeout=10) as resp:
        print("状态码:", resp.status)                 # 等价于 HTTPResponse.status
        print("Content-Type:", resp.headers.get("Content-Type"))
        raw = resp.read()                            # 返回 bytes
        print("原始字节数:", len(raw))

        data = json.loads(raw.decode("utf-8"))
        print("解析后 keys:", sorted(data.keys()))

        # headers 支持大小写不敏感访问
        print("所有响应头:")
        for k, v in resp.headers.items():
            print(f"  {k}: {v}")


if __name__ == "__main__":
    main()

# urllib_request.py —— POST JSON + 自定义头 + 认证 + 错误处理
import json
import urllib.error
import urllib.parse
import urllib.request

BASE = "https://httpbin.org"


def post_json(path: str, payload: dict, headers: dict | None = None) -> dict:
    """POST 一个 JSON 请求。"""
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    hdrs = {
        "Content-Type": "application/json; charset=utf-8",   # 告诉服务端编码
        "Accept": "application/json",
        "User-Agent": "python-tutorial/1.0",
    }
    hdrs.update(headers or {})

    req = urllib.request.Request(
        url=BASE + path,
        data=body,            # 传 bytes 即为非 GET/HEAD 请求
        headers=hdrs,
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        print(f"HTTP 错误 {exc.code}: {exc.read()[:200]!r}")
        raise
    except urllib.error.URLError as exc:
        print(f"网络错误 {exc.reason}")
        raise


def post_form(path: str, fields: dict) -> dict:
    """POST 表单：application/x-www-form-urlencoded。"""
    body = urllib.parse.urlencode(fields).encode("ascii")
    req = urllib.request.Request(
        BASE + path, data=body, method="POST",
        headers={"Content-Type": "application/x-www-form-urlencoded"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        return json.loads(resp.read().decode("utf-8"))


def get_with_query(path: str, params: dict) -> dict:
    """GET 带查询参数——必须自己 urlencode。"""
    qs = urllib.parse.urlencode(params, doseq=True)
    url = f"{BASE}{path}?{qs}"
    print("请求 URL:", url)
    with urllib.request.urlopen(url, timeout=10) as resp:
        return json.loads(resp.read().decode("utf-8"))


def get_basic_auth(host: str, user: str, password: str, path: str) -> bytes:
    """HTTP Basic 认证：base64(user:password) 放进 Authorization 头。"""
    import base64
    token = base64.b64encode(f"{user}:{password}".encode()).decode()
    req = urllib.request.Request(
        f"{host}{path}",
        headers={"Authorization": f"Basic {token}"},
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        return resp.read()


if __name__ == "__main__":
    r = post_json("/post", {"name": "张三", "age": 30})
    print("POST JSON ->", r["json"])

    r = post_form("/post", {"user": "tom", "pwd": "s3cr3t"})
    print("POST Form ->", r["form"])

    r = get_with_query("/get", {"page": 2, "size": 10, "tags": ["a", "b"]})
    print("GET args ->", r["args"])

    try:
        post_json("/status/404", {})
    except urllib.error.HTTPError as exc:
        print("捕获到 404，异常对象可用:", exc.code, exc.reason)

常见坑urllib.request.urlopen() 对 4xx/5xx 会抛 HTTPError（它是 URLError 的子类），不是返回一个带 404 状态的对象。想要「无论什么状态码都拿到响应」，必须用 build_opener 装一个 不抛异常 的处理器：

import urllib.request


class NoRedirect(urllib.request.HTTPRedirectHandler):
    """覆盖默认的自动重定向，返回原始 3xx 响应。"""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


opener = urllib.request.build_opener(NoRedirect)
try:
    resp = opener.open("https://httpbin.org/status/301", timeout=10)
except urllib.error.HTTPError as exc:
    print(exc.code, exc.headers.get("Location"))

4.4 文件下载与流式读取

# urllib_download.py
import os
import urllib.request
from pathlib import Path


def download(url: str, dest: str | Path, chunk: int = 8192) -> Path:
    """流式下载到文件，避免一次性读进内存。"""
    dest = Path(dest)
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_suffix(dest.suffix + ".part")     # 先写 .part，成功后改名

    req = urllib.request.Request(url, headers={"User-Agent": "python-tutorial/1.0"})
    with urllib.request.urlopen(req, timeout=30) as resp:
        total = int(resp.headers.get("Content-Length") or 0)
        done = 0
        with open(tmp, "wb") as f:
            while True:
                block = resp.read(chunk)              # 每次读 8KB
                if not block:
                    break
                f.write(block)
                done += len(block)
                pct = f"{done / total:6.1%}" if total else "  ?  "
                print(f"\r下载中 {pct} ({done}/{total} 字节)", end="")
    print()
    tmp.replace(dest)                                # 原子替换，避免半成品
    return dest


if __name__ == "__main__":
    p = download("https://httpbin.org/bytes/65536", "/tmp/sample.bin")
    print("完成:", p, p.stat().st_size, "字节")
    assert os.path.exists(p)

4.5 urllib 打开一个 TCP 隧道（进阶）

# proxy_demo.py —— 用 urllib 走 HTTP 代理
import urllib.request

PROXY = "http://127.0.0.1:8888"      # 需自备代理（如 mitmproxy / Charles / v2ray）
opener = urllib.request.build_opener(
    urllib.request.ProxyHandler({"http": PROXY, "https": PROXY})
)
try:
    with opener.open("https://httpbin.org/ip", timeout=10) as resp:
        print("经代理出口 IP:", resp.read().decode())
except Exception as exc:
    print("代理不可用:", exc)

requests：事实上的标准 HTTP 客户端

5.1 安装与最小示例

pip install requests requests-toolbelt urllib3[secure]

# requests_basics.py
import json

import requests

BASE = "https://httpbin.org"


def main() -> None:
    # 1) 极简 GET —— requests 默认带一堆好用的头
    r = requests.get(f"{BASE}/get", timeout=10)
    print(r.status_code, r.reason)                 # 200 OK
    print(r.headers["Content-Type"])
    print(r.json()["url"])                          # 自动解析 JSON

    # 2) 常用属性
    print("elapsed:", r.elapsed.total_seconds(), "秒")
    print("history:", len(r.history), "次重定向")
    print("请求头中的 UA:", r.request.headers.get("User-Agent"))

    # 3) POST JSON
    r = requests.post(f"{BASE}/post", json={"name": "李四", "score": 99})
    print("POST 收到:", r.json()["json"])

    # 4) POST 表单
    r = requests.post(f"{BASE}/post", data={"user": "amy", "pwd": "hunter2"})
    print("Form 收到:", r.json()["form"])

    # 5) 查询参数（自动编码）
    r = requests.get(f"{BASE}/get", params={"page": 2, "q": "中文 关键词"})
    print("请求 URL:", r.request.url)

    # 6) 自定义头
    r = requests.get(f"{BASE}/headers",
                     headers={"Authorization": "Bearer abc123",
                              "X-Trace-Id": "trace-001"})
    print("服务端看到的头:", r.json()["headers"]["X-Trace-Id"])

    # 7) 原始响应体与字节
    r = requests.get(f"{BASE}/bytes/32")
    print("bytes:", len(r.content), "编码:", r.encoding,
          "文本预览:", repr(r.text[:16]))


if __name__ == "__main__":
    main()

5.2 requests 与 urllib 的全面对照

能力	urllib.request	requests	httpx
GET 带参	需手动 urlencode	params={...}	params={...}
JSON 请求体	手动 json.dumps + 设 Content-Type	json={...}	json={...}
超时设置	urlopen(timeout=t)（总超时）	timeout=(connect, read)	timeout=(c, r) 或细粒度对象
异常类型	HTTPError / URLError	ConnectionError / Timeout / TooManyRedirects	细分的 TransportError 家族
会话/连接复用	需自己写 opener + Cookie 处理	Session 内置	Client 内置
Cookie	http.cookiejar.CookieJar	session.cookies	client.cookies
流式下载	resp.read(n)	stream=True + iter_content	client.stream()（异步）
重试	无	需 HTTPAdapter(max_retries=)	内置 transport=AsyncHTTPTransport(retries=n)
HTTP/2	不支持	需 httpx[http2] 或 urllib3 未来版	httpx.Client(http2=True)
异步	不支持	不支持	原生支持（AsyncClient）
依赖	标准库	第三方	第三方

# side_by_side.py —— 同一件事的两种写法
import json
import urllib.request

import requests

URL = "https://httpbin.org/post"
DATA = {"key": "值", "n": 1}

# --- urllib 版本 ---
body = json.dumps(DATA, ensure_ascii=False).encode("utf-8")
req = urllib.request.Request(
    URL, data=body, method="POST",
    headers={"Content-Type": "application/json"},
)
with urllib.request.urlopen(req, timeout=10) as resp:
    urllib_result = json.loads(resp.read().decode("utf-8"))

# --- requests 版本 ---
r = requests.post(URL, json=DATA, timeout=10)
requests_result = r.json()

print("两者解析结果一致:", urllib_result["json"] == requests_result["json"])

5.3 流式下载与大文件

# requests_stream.py
from pathlib import Path

import requests

CHUNK = 64 * 1024


def stream_download(url: str, dest: str) -> Path:
    """流式下载：不把整个文件读进内存。"""
    dest = Path(dest)
    tmp = dest.with_suffix(dest.suffix + ".part")

    with requests.get(url, stream=True, timeout=(5, 60)) as resp:
        resp.raise_for_status()                 # 4xx/5xx 直接抛异常
        total = int(resp.headers.get("Content-Length") or 0)
        got = 0
        with open(tmp, "wb") as f:
            for block in resp.iter_content(chunk_size=CHUNK):
                if not block:                   # 过滤 keep-alive 空块
                    continue
                f.write(block)
                got += len(block)
                if total:
                    print(f"\r{got / total:6.1%}", end="")
    print()
    tmp.replace(dest)
    return dest


def iter_lines_api(url: str) -> list[str]:
    """逐行读取流式响应（SSE / 大 JSONL）。"""
    with requests.get(url, stream=True, timeout=30) as resp:
        resp.raise_for_status()
        return [line for line in resp.iter_lines(decode_unicode=True) if line]


if __name__ == "__main__":
    p = stream_download("https://httpbin.org/bytes/1048576", "/tmp/big.bin")
    print("下载完成:", p.stat().st_size, "字节")
    lines = iter_lines_api("https://httpbin.org/stream-bytes/2048")
    print("读到", len(lines), "行")

5.4 代理

# requests_proxy.py
import os
import requests

# 环境变量方式（requests 自动读取，最省心）
os.environ["HTTP_PROXY"] = "http://127.0.0.1:7890"
os.environ["HTTPS_PROXY"] = "http://127.0.0.1:7890"

# 显式方式：三个字典分别是 http / https / no_proxy
proxies = {
    "http": "http://127.0.0.1:7890",
    "https": "http://127.0.0.1:7890",
}
session = requests.Session()
session.proxies.update(proxies)

# no_proxy 让内网地址绕过代理
session.trust_env = True          # 是否读取环境变量

r = session.get("https://httpbin.org/ip", timeout=10)
print(r.json())                   # {'origin': '你的代理出口 IP'}

Session、Cookie 与连接池

6.1 为什么必须用 Session

直接调用 requests.get() 每次都会新建 TCP 连接 + TLS 握手，并丢失 Cookie。抓取 100 个页面时，这是巨大的额外开销。Session 复用连接池并持有 Cookie：

# session_basics.py —— Cookie 持久化与连接复用
import requests

s = requests.Session()
s.headers.update({
    "User-Agent": "Mozilla/5.0 (compatible; TutorialBot/1.0)",
    "Accept-Language": "zh-CN,zh;q=0.9",
})

BASE = "https://httpbin.org"

# 第一次请求设置 Cookie（服务端会记录并回显）
r1 = s.get(f"{BASE}/cookies/set", params={"token": "abc123", "uid": "42"}, timeout=10)
print("状态:", r1.status_code)                 # 302 重定向

# Session 自动携带了上一步拿到的 Cookie
r2 = s.get(f"{BASE}/cookies", timeout=10)
print("后续请求自动带上的 Cookie:", r2.json()["cookies"])

# 手工设置/删除
s.cookies.set("manual", "yes", domain="httpbin.org", path="/")
print("手工 Cookie:", s.cookies.get_dict())

s.close()

6.2 登录态保持的完整流程

# login_flow.py —— 表单登录 + 会话保持 + CSRF token 处理
"""
注意：真实站点请遵守其 robots.txt 与服务条款，仅用于学习。
"""
import re
from typing import Any

import requests

LOGIN_PAGE = "https://httpbin.org/forms/post"


def login(base: str, username: str, password: str) -> requests.Session:
    """返回一个已登录的 Session。真实站点流程通常分三步。"""
    s = requests.Session()
    s.headers.update({
        "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) TutorialBot/1.0",
        "Accept": "text/html,application/xhtml+xml",
    })

    # 第 1 步：GET 登录页，拿到 CSRF token 与初始 Cookie
    resp = s.get(f"{base}/login", timeout=10)
    resp.raise_for_status()
    token = extract_csrf(resp.text)
    print("CSRF token:", token)

    # 第 2 步：POST 提交凭据
    payload: dict[str, Any] = {
        "username": username,
        "password": password,
    }
    if token:
        payload["csrf_token"] = token
    resp = s.post(f"{base}/login", data=payload, timeout=10,
                  allow_redirects=True)
    print("登录后状态:", resp.status_code, "最终 URL:", resp.url)

    # 第 3 步：验证登录态
    probe = s.get(f"{base}/me", timeout=10)
    print("是否已登录:", probe.status_code == 200 and "data" in probe.text)
    return s


def extract_csrf(html: str) -> str | None:
    """从 HTML 中抽取隐藏域的 csrf token（示例正则）。"""
    m = re.search(r'name=["\']csrf_token["\']\s+value=["\']([^"\']+)["\']', html)
    return m.group(1) if m else None


def paged_fetch(s: requests.Session, base: str) -> list[dict]:
    """带 Session 的分页抓取示例。"""
    out: list[dict] = []
    url = f"{base}/users"
    for page in range(1, 4):
        r = s.get(url, params={"page": page}, timeout=10)
        if r.status_code != 200:
            print("停止翻页:", r.status_code)
            break
        items = r.json()
        if not items:
            break
        out.extend(items)
    return out


if __name__ == "__main__":
    session = login("https://httpbin.org", "demo", "demo123")
    session.close()

6.3 连接池调优

# pool_tuning.py —— 显式配置连接池
from urllib3.util.retry import Retry

import requests
from requests.adapters import HTTPAdapter


def build_session(pool_maxsize: int = 20, retries: int = 3) -> requests.Session:
    s = requests.Session()

    # 指数退避的重试策略
    retry = Retry(
        total=retries,
        connect=retries,                     # 连接失败重试
        read=retries,                        # 读取失败重试
        status=retries,                      # 状态码重试
        backoff_factor=0.5,                  # 退避：0s, 1s, 2s, 4s...
        status_forcelist=[429, 500, 502, 503, 504],   # 只重试这些状态
        allowed_methods=["GET", "HEAD", "OPTIONS", "PUT", "DELETE"],  # POST 不自动重试
        raise_on_status=False,
    )

    adapter = HTTPAdapter(
        max_retries=retry,
        pool_connections=pool_maxsize,       # 保留的连接池个数（每个 host）
        pool_maxsize=pool_maxsize,           # 每个 host 的最大连接数（重点）
        pool_block=True,                     # 池满时阻塞等待，避免 ConnectionError
    )
    s.mount("http://", adapter)
    s.mount("https://", adapter)
    return s


if __name__ == "__main__":
    with build_session() as s:
        r = s.get("https://httpbin.org/get", timeout=(5, 15))
        print(r.status_code)
        print("连接池配置:", s.get_adapter("https://")._pool_maxsize)

参数	含义	建议
pool_connections	缓存的 host 连接池数	默认 10，够用
pool_maxsize	每个 host 保存的连接数	默认 10；高并发调到 20–100
pool_block	池满时等待而非新建连接	配合线程池设为 True
max_retries	重试次数	只对幂等方法开
backoff_factor	退避基数	0.5 表示 0/1/2/4 秒

超时、重试与错误处理

7.1 超时的四种含义

超时类型	触发条件	常见设置
连接超时 connect timeout	TCP 握手 + TLS 握手完成前	2–5 秒
读取超时 read timeout	已连上，等待服务器响应数据	10–60 秒（按接口 SLA）
写超时 write timeout	发送大 body 时	与 read 同
总超时 wall-clock	整个请求从发起到结束	业务级兜底

# timeout_handling.py —— 全套超时与重试
import itertools
import random
import time
from dataclasses import dataclass

import requests
from requests.exceptions import (
    ConnectionError,
    ConnectTimeout,
    ReadTimeout,
    TooManyRedirects,
)


@dataclass
class FetchResult:
    ok: bool
    status: int | None
    data: object = None
    error: str | None = None
    attempts: int = 0


def fetch(
    url: str,
    *,
    connect_timeout: float = 3.0,
    read_timeout: float = 10.0,
    max_retry: int = 3,
    retry_status: set[int] | None = None,
) -> FetchResult:
    """带完整超时与重试的 GET。"""
    retry_status = retry_status or {429, 500, 502, 503, 504}
    last_err = "未知"

    for attempt in itertools.count(1):
        t0 = time.perf_counter()
        try:
            r = requests.get(url, timeout=(connect_timeout, read_timeout))

            if r.status_code in retry_status and attempt <= max_retry:
                # 尊重 Retry-After，否则指数退避
                ra = float(r.headers.get("Retry-After", 2 ** attempt))
                print(f"  状态 {r.status_code}，{ra:.1f}s 后重试")
                time.sleep(ra)
                last_err = f"HTTP {r.status_code}"
                continue

            if r.status_code >= 400:
                return FetchResult(False, r.status_code,
                                   error=f"HTTP {r.status_code}",
                                   attempts=attempt)

            try:
                data = r.json()
            except ValueError:
                data = r.text[:500]

            cost = time.perf_counter() - t0
            print(f"  成功（第 {attempt} 次，{cost:.3f}s）")
            return FetchResult(True, r.status_code, data, attempts=attempt)

        except (ConnectTimeout, ConnectionError) as exc:
            last_err = f"连接失败: {exc}"
        except ReadTimeout as exc:
            last_err = f"读取超时: {exc}"
        except TooManyRedirects as exc:
            return FetchResult(False, None, error=f"重定向过多: {exc}", attempts=attempt)
        except requests.RequestException as exc:
            return FetchResult(False, None, error=f"请求异常: {exc}", attempts=attempt)

        if attempt > max_retry:
            break
        backoff = min(2 ** (attempt - 1), 8) + random.uniform(0, 0.3)
        print(f"  {last_err}，{backoff:.1f}s 后重试")
        time.sleep(backoff)

    return FetchResult(False, None, error=last_err, attempts=max_retry + 1)


if __name__ == "__main__":
    print("1) 正常请求")
    print(fetch("https://httpbin.org/json", max_retry=2))

    print("2) 服务端 500，带重试")
    print(fetch("https://httpbin.org/status/500", max_retry=2))

    print("3) 404，不重试直接返回")
    print(fetch("https://httpbin.org/status/404", max_retry=2))

7.2 异常处理金字塔

异常	继承自	何时发生	是否该重试
requests.exceptions.ConnectionError	RequestException	DNS 失败、连接被拒、SSL 握手失败	可重试（可能是瞬时）
requests.exceptions.ConnectTimeout	ConnectionError, Timeout	连接超时	可重试
requests.exceptions.ReadTimeout	Timeout	连接建立后读取超时	可重试（非幂等方法慎用）
requests.exceptions.Timeout	RequestException	超时的基类	按子类处理
requests.exceptions.TooManyRedirects	RequestException	重定向超过上限	不该重试（配置问题）
requests.exceptions.HTTPError	RequestException	调用 raise_for_status() 后抛出	按状态码判断
requests.exceptions.InvalidURL	RequestException	URL 非法	不该重试（代码 bug）

# exception_hierarchy.py —— 结构化处理
import requests


def robust_get(url: str) -> None:
    try:
        r = requests.get(url, timeout=(3, 10))
        r.raise_for_status()                  # 4xx/5xx 转成 HTTPError
    except requests.exceptions.ConnectTimeout:
        print("网络不通：连接超时，检查代理/防火墙")
    except requests.exceptions.ReadTimeout:
        print("服务器太慢或响应体太大")
    except requests.exceptions.TooManyRedirects:
        print("重定向异常，可能是被劫持或配置错误")
    except requests.exceptions.SSLError:
        print("证书校验失败：可能是中间人代理或系统时间不对")
    except requests.exceptions.ConnectionError as exc:
        print(f"DNS 解析失败或连接被拒: {exc}")
    except requests.exceptions.HTTPError as exc:
        print(f"HTTP 错误码 {exc.response.status_code}，"
              f"响应体: {exc.response.text[:100]}")
    except requests.exceptions.RequestException as exc:
        print(f"其它请求异常: {type(exc).__name__}: {exc}")
    else:
        print("成功，长度", len(r.content))
    finally:
        print("--- 本次请求结束 ---")


if __name__ == "__main__":
    robust_get("https://httpbin.org/status/404")
    robust_get("https://httpbin.org/status/500")
    robust_get("https://this-domain-really-does-not-exist-xyz123.invalid/")

REST API 调用实战

8.1 REST 的六个约定

约定	含义	正确示例	错误示例
资源用名词 URL	URL 标识资源，动作交给 HTTP 方法	GET /users/42	GET /getUser?id=42
用 HTTP 方法表达动作	GET 查 / POST 建 / PUT 全改 / PATCH 局部改 / DELETE 删	PATCH /users/42	POST /updateUser
无状态	每次请求自带全部信息，服务器不存会话	Token 在请求头里	依赖服务器端 session 变量
统一接口	所有资源返回同样的 JSON 结构	统一 {"code":0,"data":...}	A 接口返数组、B 返对象
用状态码表达结果	201 已创建、204 无内容、400 参数错、409 冲突	201 Created	全部返回 200，用 body 里的 flag 表示
可缓存	GET 响应设 Cache-Control	max-age=60	所有接口禁用缓存

8.2 一个可直接用的 REST 客户端

# rest_client.py —— 可直接改造使用的 API 客户端
"""
运行：python rest_client.py
"""
from __future__ import annotations

import json
import logging
import time
from dataclasses import dataclass, field
from typing import Any
from urllib.parse import urljoin

import requests

log = logging.getLogger("rest")


@dataclass
class ApiError(Exception):
    """统一的 API 异常，携带状态码与业务错误码。"""
    status: int
    message: str
    payload: Any = None

    def __str__(self) -> str:
        return f"[{self.status}] {self.message}"


@dataclass
class RestClient:
    base_url: str
    timeout: tuple[float, float] = (3.05, 30.0)
    max_retry: int = 3
    session: requests.Session = field(default_factory=requests.Session)

    def __post_init__(self) -> None:
        self.session.headers.update({
            "Accept": "application/json",
            "User-Agent": "TutorialRestClient/1.0",
        })

    # ---- 底层请求 ----
    def request(
        self,
        method: str,
        path: str,
        *,
        params: dict | None = None,
        json_body: Any = None,
        data_body: Any = None,
        files: dict | None = None,
        headers: dict | None = None,
        expect_json: bool = True,
    ) -> Any:
        url = urljoin(self.base_url.rstrip("/") + "/", path.lstrip("/"))
        last: Exception | None = None

        for attempt in range(1, self.max_retry + 1):
            try:
                resp = self.session.request(
                    method=method, url=url, params=params,
                    json=json_body, data=data_body, files=files,
                    headers=headers, timeout=self.timeout,
                )
            except requests.RequestException as exc:
                last = exc
                if attempt == self.max_retry:
                    raise
                time.sleep(2 ** (attempt - 1) * 0.5)
                continue

            # 业务成功
            if resp.status_code < 400:
                if resp.status_code == 204 or not resp.content:
                    return None
                return resp.json() if expect_json else resp.text

            # 可重试的状态码
            if resp.status_code in (429, 500, 502, 503, 504) and attempt < self.max_retry:
                delay = float(resp.headers.get("Retry-After", 2 ** attempt))
                log.warning("HTTP %s，%.1fs 后重试", resp.status_code, delay)
                time.sleep(delay)
                continue

            # 提取错误信息
            try:
                payload = resp.json()
                msg = payload.get("message") or payload.get("error") or str(payload)
            except ValueError:
                payload, msg = None, resp.text[:200]
            raise ApiError(resp.status_code, msg, payload)

        raise ApiError(0, str(last))

    # ---- 语义化方法 ----
    def get(self, path: str, **kw) -> Any:
        return self.request("GET", path, **kw)

    def post(self, path: str, **kw) -> Any:
        return self.request("POST", path, **kw)

    def put(self, path: str, **kw) -> Any:
        return self.request("PUT", path, **kw)

    def patch(self, path: str, **kw) -> Any:
        return self.request("PATCH", path, **kw)

    def delete(self, path: str, **kw) -> Any:
        return self.request("DELETE", path, **kw)


# ---- 分页工具 ----
@dataclass
class Pager:
    """同时兼容 page/page_size 与 cursor/next 三种主流分页风格。"""

    session: requests.Session

    def by_page(self, url: str, total_pages: int = 5,
                **params) -> list[dict]:
        out: list[dict] = []
        for p in range(1, total_pages + 1):
            r = self.session.get(url, params={**params, "page": p,
                                               "page_size": 50}, timeout=10)
            r.raise_for_status()
            items = r.json().get("data", [])
            if not items:
                break
            out.extend(items)
        return out

    def by_cursor(self, url: str, max_rounds: int = 20, **params) -> list[dict]:
        out: list[dict] = []
        cursor = None
        for _ in range(max_rounds):
            p = {**params, "limit": 50}
            if cursor:
                p["cursor"] = cursor
            r = self.session.get(url, params=p, timeout=10)
            r.raise_for_status()
            body = r.json()
            out.extend(body.get("data", []))
            cursor = body.get("next_cursor")
            if not cursor:
                break
        return out


# ---- 鉴权方式 ----
def bearer(session: requests.Session, token: str) -> None:
    session.headers["Authorization"] = f"Bearer {token}"


def basic(session: requests.Session, user: str, pwd: str) -> None:
    session.auth = (user, pwd)           # requests 内置处理 base64


def apikey_header(session: requests.Session, key: str) -> None:
    session.headers["X-API-Key"] = key


def apikey_query(client: RestClient, key: str) -> dict:
    return {"params": {"api_key": key}}


# ---- 演示 ----
if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
    c = RestClient("https://httpbin.org", max_retry=2)

    print("GET  /get ->", type(c.get("/get", params={"a": 1})))
    print("POST /post ->", c.post("/post", json_body={"k": "v"})["json"])

    try:
        c.get("/status/403")
    except ApiError as exc:
        print("统一异常:", exc, "| payload =", exc.payload)

    # 分页演示
    pager = Pager(requests.Session())
    items = pager.by_page("https://httpbin.org/anything", total_pages=2, page=0)
    print(f"分页取到 {len(items)} 条")
    c.session.close()

8.3 文件上传

# upload.py —— 单文件 / 多文件 / 分块上传
from pathlib import Path

import requests

BASE = "https://httpbin.org"


def upload_single(path: str, remote_field: str = "file") -> dict:
    """单文件上传（multipart/form-data）。"""
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(p)
    with open(p, "rb") as f:
        r = requests.post(
            f"{BASE}/post",
            files={remote_field: (p.name, f, "application/octet-stream")},
            data={"description": "示例文件"},
            timeout=30,
        )
    r.raise_for_status()
    return r.json()["form"]


def upload_multi(paths: list[str], field: str = "files") -> dict:
    """多文件上传：field 对应一个列表。"""
    handles = [open(p, "rb") for p in paths]
    try:
        files = [(field, (Path(p).name, h)) for p, h in zip(paths, handles)]
        r = requests.post(f"{BASE}/post", files=files, timeout=60)
        r.raise_for_status()
        return r.json()["form"]
    finally:
        for h in handles:
            h.close()


def upload_chunks(path: str, chunk_size: int = 1024 * 1024,
                  total_chunks: int = 3) -> list[dict]:
    """分块上传：每块一次请求，服务端用 upload_id 关联（断点续传思路）。"""
    p = Path(path)
    upload_id = f"up-{abs(hash(str(p))) % 100000}"
    results = []
    with open(p, "rb") as f:
        for i in range(total_chunks):
            block = f.read(chunk_size)
            if not block:
                break
            r = requests.post(
                f"{BASE}/post",
                files={"chunk": (None, block)},
                data={"upload_id": upload_id,
                      "chunk_index": str(i),
                      "total_chunks": str(total_chunks)},
                timeout=60,
            )
            results.append(r.json()["form"])
            print(f"上传分块 {i + 1}/{total_chunks}：{len(block)} 字节")
    return results


if __name__ == "__main__":
    Path("/tmp/demo.txt").write_text("hello http\n" * 10, encoding="utf-8")
    print("单文件:", upload_single("/tmp/demo.txt"))
    print("分块:", upload_chunks("/tmp/demo.txt", chunk_size=64, total_chunks=3))

8.4 轮询长任务（异步 API 的标准处理方式）

# long_task_poll.py —— 提交任务 + 轮询状态 + 取结果
import time

import requests


def submit_job(client: requests.Session, payload: dict) -> str:
    """提交异步任务，返回 job_id。"""
    r = client.post("https://httpbin.org/anything",
                    json=payload, timeout=15)
    r.raise_for_status()
    # 真实 API 常见：{"job_id": "...", "status": "pending"}
    return r.json()["json"].get("job_id", "demo-job")


def poll_job(client: requests.Session, job_id: str,
             interval: float = 2.0, max_wait: float = 60.0) -> dict:
    """轮询直到任务结束。

    真实项目里 Status URL 通常形如：
        GET /jobs/{job_id}
    返回 {"status": "running|succeeded|failed", "result": ..., "progress": 0.42}
    """
    deadline = time.monotonic() + max_wait
    attempt = 0
    while time.monotonic() < deadline:
        attempt += 1
        r = client.get(f"https://httpbin.org/anything/{job_id}", timeout=10)
        r.raise_for_status()
        state = r.json()
        status = state.get("status", "succeeded")     # httpbin 无状态机，视为成功
        print(f"第 {attempt} 次轮询: status={status}")
        if status in ("succeeded", "failed", "cancelled"):
            return state
        time.sleep(interval)
    raise TimeoutError(f"任务 {job_id} 在 {max_wait}s 内未完成")


if __name__ == "__main__":
    s = requests.Session()
    job_id = submit_job(s, {"task": "生成报表", "rows": 100000})
    result = poll_job(s, job_id, interval=1.0, max_wait=5.0)
    print("最终结果:", result["json"])
    s.close()

文件下载：完整实现

9.1 断点续传的原理

断点续传（HTTP Range）依赖两个请求头和一个响应头：

方向	头	作用
请求	Range: bytes=1000000-	从第 1000000 字节开始要
响应	206 Partial Content	服务端同意返回部分内容
响应	Content-Range: bytes 1000000-4999999/5000000	本次范围 + 总大小
响应	Accept-Ranges: bytes	服务端声明支持续传

# resume_download.py —— 支持断点续传、并发分块、限速的下载器
"""
运行：python resume_download.py
"""
from __future__ import annotations

import os
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from typing import Callable

import requests


@dataclass
class DownloadConfig:
    url: str
    dest: str
    chunk_size: int = 1024 * 1024            # 1MB
    connections: int = 4                     # 并发分块数
    max_retry: int = 5
    rate_limit: float | None = None           # 字节/秒，None 为不限速
    timeout: tuple[float, float] = (10, 60)
    progress: Callable[[int, int], None] | None = None


def _human(n: float) -> str:
    for unit in ("B", "KB", "MB", "GB"):
        if n < 1024 or unit == "GB":
            return f"{n:.1f}{unit}"
        n /= 1024
    return f"{n:.1f}TB"


class Downloader:
    """把大文件切成 N 块并发下载，失败自动重试，支持续传。"""

    def __init__(self, cfg: DownloadConfig) -> None:
        self.cfg = cfg
        self.dest = Path(cfg.dest)
        self.dest.parent.mkdir(parents=True, exist_ok=True)
        # 用 .part 存分块进度，.data 存最终内容
        self.part_dir = self.dest.with_suffix(self.dest.suffix + ".parts")
        self.part_dir.mkdir(exist_ok=True)
        self.total = 0
        self.downloaded = 0
        self._lock = __import__("threading").Lock()

    # ---- 探测文件信息 ----
    def probe(self) -> tuple[int, bool]:
        """返回 (总大小, 是否支持续传)。"""
        r = requests.head(self.cfg.url,
                          allow_redirects=True,
                          timeout=self.cfg.timeout)
        r.raise_for_status()
        self.total = int(r.headers.get("Content-Length") or 0)
        accepts = r.headers.get("Accept-Ranges", "")
        return self.total, ("bytes" in accepts)

    # ---- 单块下载（带 Range 与重试） ----
    def _download_range(self, index: int, start: int, end: int) -> int:
        """下载 [start, end] 到第 index 块，返回写入字节数。"""
        part_file = self.part_dir / f"{index:05d}.part"
        if part_file.exists() and part_file.stat().st_size == end - start + 1:
            print(f"[块 {index}] 已存在，跳过")
            return 0

        headers = {"Range": f"bytes={start}-{end}"}
        for attempt in range(1, self.cfg.max_retry + 1):
            try:
                with requests.get(self.cfg.url, headers=headers,
                                  stream=True, timeout=self.cfg.timeout) as r:
                    if r.status_code not in (200, 206):
                        raise requests.HTTPError(f"状态码 {r.status_code}")
                    mode = "ab" if r.status_code == 206 else "wb"
                    written = 0
                    with open(part_file, mode) as f:
                        for block in r.iter_content(self.cfg.chunk_size):
                            f.write(block)
                            written += len(block)
                            with self._lock:
                                self.downloaded += len(block)
                                if self.cfg.progress:
                                    self.cfg.progress(self.downloaded, self.total)
                            if self.cfg.rate_limit:
                                time.sleep(len(block) / self.cfg.rate_limit)
                    if written == end - start + 1:
                        return written
            except requests.RequestException as exc:
                print(f"[块 {index}] 第 {attempt} 次失败: {exc}")
                if attempt == self.cfg.max_retry:
                    raise
                time.sleep(min(2 ** attempt, 10))
        return 0

    # ---- 主流程 ----
    def run(self) -> Path:
        total, resumable = self.probe()
        if total == 0:
            # 不给 Content-Length：退化为单连接顺序下载
            return self._single_stream()
        if not resumable:
            print("服务端不支持续传，退化为单连接")
            return self._single_stream()

        n = self.cfg.connections
        size = total // n
        ranges: list[tuple[int, int, int]] = []
        for i in range(n):
            start = i * size
            end = total - 1 if i == n - 1 else (i + 1) * size - 1
            ranges.append((i, start, end))

        print(f"共 {_human(total)}，分 {n} 块并发下载")
        t0 = time.perf_counter()
        with ThreadPoolExecutor(max_workers=n) as pool:
            list(pool.map(lambda t: self._download_range(*t), ranges))
        cost = time.perf_counter() - t0

        # 合并分块
        tmp = self.dest.with_suffix(self.dest.suffix + ".tmp")
        with open(tmp, "wb") as out:
            for i, _, _ in ranges:
                out.write((self.part_dir / f"{i:05d}.part").read_bytes())
        tmp.replace(self.dest)

        for p in self.part_dir.iterdir():
            p.unlink()
        self.part_dir.rmdir()

        speed = total / cost if cost else 0
        print(f"完成：{_human(total)} 用时 {cost:.2f}s 平均 { _human(speed) }/s")
        return self.dest

    # ---- 不支持续传时的简单实现 ----
    def _single_stream(self) -> Path:
        tmp = self.dest.with_suffix(self.dest.suffix + ".tmp")
        with requests.get(self.cfg.url, stream=True,
                          timeout=self.cfg.timeout) as r:
            r.raise_for_status()
            with open(tmp, "wb") as f:
                for block in r.iter_content(self.cfg.chunk_size):
                    f.write(block)
                    if self.cfg.progress and self.total:
                        self.downloaded += len(block)
                        self.cfg.progress(self.downloaded, self.total)
        tmp.replace(self.dest)
        return self.dest


def show_progress(done: int, total: int) -> None:
    pct = done / total if total else 0
    bar = "#" * int(pct * 30)
    print(f"\r[{bar:<30}] {pct:6.1%}", end="")


if __name__ == "__main__":
    cfg = DownloadConfig(
        url="https://speed.hetzner.de/1MB.bin",   # 换成你自己的地址
        dest="/tmp/hetzner.bin",
        connections=4,
        progress=show_progress,
    )
    try:
        p = Downloader(cfg).run()
        print("\n已保存:", p, os.path.getsize(p), "字节")
    except requests.RequestException as exc:
        print("下载失败:", exc)

9.2 校验与完整性

# verify.py —— 下载后校验：大小、哈希、类型
import hashlib
from pathlib import Path

import requests


def sha256_of(path: str | Path, block: int = 1 << 20) -> str:
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(block), b""):
            h.update(chunk)
    return h.hexdigest()


def download_and_verify(url: str, dest: str, expect_sha256: str | None = None) -> bool:
    with requests.get(url, stream=True, timeout=60) as r:
        r.raise_for_status()
        declared = r.headers.get("Content-Length")
        digest_hdr = r.headers.get("Digest")     # 形如 sha-256=base64
        with open(dest, "wb") as f:
            for b in r.iter_content(1 << 16):
                f.write(b)

    size = Path(dest).stat().st_size
    if declared and int(declared) != size:
        print(f"大小不符：声明 {declared}，实际 {size}")
        return False

    if expect_sha256:
        got = sha256_of(dest)
        if got != expect_sha256:
            print(f"哈希不符：期望 {expect_sha256}，实际 {got}")
            return False
        print("哈希校验通过:", got)
    print("校验通过:", size, "字节")
    return True


if __name__ == "__main__":
    download_and_verify("https://httpbin.org/bytes/16384", "/tmp/verify.bin")

进阶：代理、抓包、WebSocket、SSE

10.1 抓包与调试工具

工具	类型	用途	快速上手
浏览器 DevTools	内置	看请求/响应、耗时、Cookie	F12 → Network → Fetch/XHR
Charles / Fiddler	GUI 代理	HTTPS 抓包、改包、重放	安装根证书，配置 8888 端口
mitmproxy	命令行代理	脚本化抓包与改写（可写 Python 插件）	mitmproxy -p 8080
Wireshark	抓包	四层协议全量分析	抓 lo0 网卡，Follow TCP Stream
curl -v	命令行	最轻量的 HTTP 调试	curl -v -H "..." URL
httpbin.org	在线服务	回显你发的任何东西	requests.get("https://httpbin.org/get")

# curl 等价对照
curl -v "https://httpbin.org/get?a=1" -H "Authorization: Bearer xxx"

# 等价 Python
import requests
requests.get("https://httpbin.org/get", params={"a": 1},
            headers={"Authorization": "Bearer xxx"}, timeout=10)

10.2 SSE（Server-Sent Events）服务端推送

# sse_client.py —— 消费 SSE 流（同步与异步两种写法）
import json

import requests


def consume_sse_sync(url: str, max_events: int = 5) -> list[dict]:
    """同步版：用 stream=True 逐行读。"""
    events: list[dict] = []
    with requests.get(url, stream=True, timeout=(5, 60),
                      headers={"Accept": "text/event-stream"}) as resp:
        resp.raise_for_status()
        event: dict = {}
        for raw in resp.iter_lines(decode_unicode=True):
            if raw is None:
                continue
            line = raw.strip()
            if line == "":
                # 空行代表一个事件结束
                if event:
                    events.append(event)
                    event = {}
                    if len(events) >= max_events:
                        break
            elif line.startswith(":"):
                continue                    # 注释行，用于心跳保活
            elif ":" in line:
                field, _, value = line.partition(":")
                value = value.lstrip()
                if field == "data":
                    try:
                        event["data"] = json.loads(value)
                    except ValueError:
                        event["data"] = value
                elif field == "event":
                    event["event"] = value
                elif field == "id":
                    event["id"] = value
    return events


async def consume_sse_async(url: str, max_events: int = 5) -> list[dict]:
    """异步版（推荐，可与其它协程并发）。"""
    import httpx

    events: list[dict] = []
    buf = ""
    async with httpx.AsyncClient(timeout=60) as client:
        async with client.stream("GET", url) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                line = line.rstrip()
                if not line:
                    if buf:
                        events.append({"data": buf})
                        buf = ""
                        if len(events) >= max_events:
                            break
                    continue
                if line.startswith("data:"):
                    buf = line[5:].strip()
    return events


if __name__ == "__main__":
    print(consume_sse_sync("https://httpbin.org/stream/3", max_events=3))

10.3 WebSocket 客户端

# ws_client.py —— 需要 pip install websocket-client
"""
运行：python ws_client.py wss://echo.websocket.org
"""
import json
import sys
import time

import websocket          # websocket-client 库


def main(url: str, send_count: int = 3) -> None:
    # enableTrace=True 可以打印握手细节，调试协议时非常有用
    ws = websocket.create_connection(
        url,
        timeout=10,
        header=["User-Agent: TutorialBot/1.0"],
    )
    print("已连接:", ws.url)

    try:
        for i in range(send_count):
            payload = json.dumps({"type": "ping", "seq": i})
            ws.send(payload)
            reply = ws.recv()
            print("收到:", reply)
            time.sleep(0.5)
    except websocket.WebSocketTimeoutException:
        print("接收超时")
    except websocket.WebSocketConnectionClosedException:
        print("连接被对端关闭")
    finally:
        ws.close()
        print("连接已关闭")


if __name__ == "__main__":
    main(sys.argv[1] if len(sys.argv) > 1 else "wss://echo.websocket.org")

10.4 让 Python 服务能被外部访问

# dev_server.py —— 最小 HTTP 服务器的几种写法
"""
启动后用手机/别的机器访问 http://<本机IP>:8000
"""
import json
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer


class Handler(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"          # 保持连接，配合 Content-Length

    def _send(self, status: int, body: bytes, ctype: str) -> None:
        self.send_response(status)
        self.send_header("Content-Type", ctype)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Access-Control-Allow-Origin", "*")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:              # noqa: N802 (BaseHTTPRequestHandler 约定)
        if self.path.startswith("/api/health"):
            body = json.dumps({"status": "ok"}).encode()
            self._send(200, body, "application/json; charset=utf-8")
        elif self.path == "/":
            self._send(200, b"Hello from Python socket server", "text/plain; charset=utf-8")
        else:
            self._send(404, b"not found", "text/plain; charset=utf-8")

    def do_POST(self) -> None:             # noqa: N802
        length = int(self.headers.get("Content-Length", 0))
        raw = self.rfile.read(length)       # 必须读干净，否则连接错乱
        try:
            payload = json.loads(raw.decode("utf-8"))
        except ValueError:
            self._send(400, b'{"error":"invalid json"}', "application/json")
            return
        self._send(200, json.dumps({"echo": payload}).encode(),
                   "application/json; charset=utf-8")

    def log_message(self, fmt: str, *args) -> None:
        print(f"[HTTP] {self.address_string()} {fmt % args}")   # 简化访问日志


def get_local_ip() -> str:
    """取本机在局域网中的 IP。"""
    import socket
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        s.connect(("8.8.8.8", 80))          # 不会真的发包
        return s.getsockname()[0]
    finally:
        s.close()


if __name__ == "__main__":
    server = ThreadingHTTPServer(("0.0.0.0", 8000), Handler)
    ip = get_local_ip()
    print(f"本机测试: http://127.0.0.1:8000")
    print(f"局域网测试: http://{ip}:8000")
    print("Ctrl+C 停止")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n正在关闭……")
    finally:
        server.shutdown()
        server.server_close()

# 测试服务端
curl -s http://127.0.0.1:8000/api/health
curl -s -X POST http://127.0.0.1:8000/ -H "Content-Type: application/json" -d '{"a":1}'

安全、常见错误与速查

11.1 安全清单

风险	错误做法	正确做法
证书校验被关	verify=False	永远 verify=True；内网自签证书用 REQUESTS_CA_BUNDLE 指定 CA
Token 泄露	写进日志 / 提交到 git	用环境变量或密钥管理服务；日志中脱敏
SSRF	把用户传入的 URL 直接请求	校验域名白名单、禁止内网地址、限制协议为 http/https
路径穿越	把用户路径直接拼到文件系统	Path(...).resolve() 后校验是否在允许目录内
解压炸弹	无限制解压 zip	限制解压后总大小与文件数量
超大响应	无上限读 body	Content-Length 校验 + 流式读取 + 主动截断
明文传输	用 http://	只用 https://；对 API Key 强制 TLS
代理泄密	把带 Authorization 的请求发到不可信代理	校验代理证书

# ssrf_guard.py —— 简易 SSRF 防护
import ipaddress
import socket
from urllib.parse import urlparse

ALLOWED_SCHEMES = {"http", "https"}


def is_safe_url(url: str, allow_private: bool = False) -> bool:
    parsed = urlparse(url)
    if parsed.scheme not in ALLOWED_SCHEMES:
        return False
    host = parsed.hostname
    if not host:
        return False

    try:
        ip = ipaddress.ip_address(socket.gethostbyname(host))
    except (socket.gaierror, ValueError):
        return False

    if allow_private:
        return True
    return not (ip.is_private or ip.is_loopback or ip.is_link_local
                or ip.is_reserved or ip.is_multicast)


if __name__ == "__main__":
    for u in ["https://example.com/a",
              "http://127.0.0.1/admin",
              "http://169.254.169.254/latest/meta-data/",   # 云元数据
              "file:///etc/passwd",
              "http://192.168.1.1/"]:
        print(f"{'允许' if is_safe_url(u) else '拒绝'}  {u}")

11.2 常见错误速查

报错 / 现象	根因	解决
ConnectionError: HTTPSConnectionPool(... Connection refused)	目标端口没服务或被防火墙拒绝	检查 ss -lntp；核对端口；容器内注意绑定 127.0.0.1 无法从外部访问
Max retries exceeded ... Name or service not known	DNS 解析失败	换 DNS、检查拼写、在容器里配 --dns
Read timed out	已连接但服务端迟迟不返回	区分是网络慢还是服务端慢；必要时走异步 + 更长 read timeout
SSLError: CERTIFICATE_VERIFY_FAILED	证书链不受信 / 系统时间不对 / 中间人代理	更新根证书（certifi）、校准时间、不要直接关 verify
ProxyError	环境变量里残留代理	requests.Session().trust_env = False
ChunkedEncodingError	服务端中途断开了流式响应	加重试与断点续传
JSONDecodeError	返回的不是 JSON（多半是 HTML 错误页）	先 resp.status_code 与 resp.headers['Content-Type']，再解析
中文乱码	未声明编码，requests 猜测错误	服务器发 charset=utf-8；或 resp.encoding = "utf-8"
请求体丢失	传了 dict 却没指定编码	明确用 data=（表单）、json=（JSON）或 files=
POST 被重定向成 GET	301/302/303 的历史语义	服务端改用 307/308 保持方法与请求体
文件下载得到 0 字节	多写了 requests.get(...) 后未加 stream=True 却读了两次	用 with requests.get(..., stream=True) 上下文管理器
并发请求偶发失败	连接池 pool_maxsize 太小 + pool_block=False	提高 pool_maxsize 并设 pool_block=True
UDP 服务收不到	本机防火墙未放行 / 绑定在 127.0.0.1	绑定 0.0.0.0 并放行端口
TCP 服务只能本机访问	绑定了 127.0.0.1	改绑 0.0.0.0
大量 CLOSE_WAIT	对端已关，自己没 close()	用 with socket 或 finally: close()

11.3 一次性自检脚本

# netcheck.py —— 快速诊断网络问题
import socket
import ssl
import time
from urllib.parse import urlparse


def dns_ok(host: str) -> tuple[bool, str]:
    t0 = time.perf_counter()
    try:
        infos = socket.getaddrinfo(host, None)
        ip = infos[0][4][0]
        return True, f"{ip}（{time.perf_counter() - t0:.3f}s）"
    except socket.gaierror as exc:
        return False, f"DNS 失败: {exc}"


def tcp_ok(host: str, port: int, timeout: float = 3) -> tuple[bool, str]:
    t0 = time.perf_counter()
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True, f"连通（{time.perf_counter() - t0:.3f}s）"
    except OSError as exc:
        return False, f"失败: {exc}"


def tls_ok(host: str, port: int = 443) -> tuple[bool, str]:
    ctx = ssl.create_default_context()       # 默认会校验证书
    t0 = time.perf_counter()
    try:
        with socket.create_connection((host, port), timeout=5) as sock:
            with ctx.wrap_socket(sock, server_hostname=host) as ssock:
                cert = ssock.getpeercert()
                return True, (f"协议 {ssock.version()}，证书 CN="
                              f"{dict(x[0] for x in cert.get('subject', [])).get('commonName')}"
                              f"（{time.perf_counter() - t0:.3f}s）")
    except ssl.SSLError as exc:
        return False, f"TLS 失败: {exc}"
    except OSError as exc:
        return False, f"连接失败: {exc}"


def check(url: str) -> None:
    p = urlparse(url if "://" in url else f"https://{url}")
    host = p.hostname or ""
    port = p.port or (443 if p.scheme == "https" else 80)
    print(f"=== {host}:{port} ({p.scheme}) ===")
    for name, ok, msg in [
        ("DNS", *dns_ok(host)),
        ("TCP", *tcp_ok(host, port)),
        ("TLS", *tls_ok(host, 443 if p.scheme == "https" else port)),
    ]:
        print(f"  {name:4s} {'OK ' if ok else 'ERR'}  {msg}")


if __name__ == "__main__":
    import requests
    for target in ["example.com", "httpbin.org", "api.github.com"]:
        check(target)
        try:
            r = requests.get(f"https://{target}", timeout=8)
            print(f"  HTTP  OK   状态 {r.status_code}，耗时 {r.elapsed.total_seconds():.3f}s")
        except requests.RequestException as exc:
            print(f"  HTTP  ERR   {exc}")
        print()

本章小结

• 分层：应用层（HTTP）、传输层（TCP/UDP）、网络层（IP）。Python 代码几乎都工作在应用层，但必须懂传输层的超时与连接语义。
• socket：bind / listen / accept / recv / sendall / close 是 TCP 骨架。优先用 sendall，recv 返回 b"" 表示对端关闭。
• 粘包拆包：TCP 是字节流，没有消息边界。必须自定义分帧协议——长度前缀（struct.pack("!I", n)）最通用；分隔符次之。
• UDP：无连接、保留消息边界、8 字节首部、无握手开销，适合 DNS / 实时音视频 / 局域网发现；需要可靠性要自己补重传与序列号，或用 QUIC。
• 并发：生产中不要用 raw socket 手写 epoll，写 WSGI/ASGI 应用交给 uvicorn/gunicorn（第 15、18 章）。
• urllib：标准库零依赖，但需手动 urlencode、JSON 编码、异常处理与 Cookie；4xx/5xx 抛 HTTPError；build_opener 可定制行为。
• requests：params / json / data / files / stream / timeout / proxies 一应俱全，新项目默认选它；异步或需 HTTP/2 选 httpx。
• Session：必须用。Session 复用 TCP/TLS 连接池（省 1~3 个 RTT）、保持 Cookie、共享默认头；HTTPAdapter 可调 pool_maxsize 与重试策略。
• 超时与重试：区分连接超时与读取超时；timeout=(3, 10) 形式最常用；重试只对幂等方法开，退避用指数 + 抖动，遵守 Retry-After。
• 异常处理：从具体到宽泛——ConnectTimeout → ReadTimeout → TooManyRedirects → SSLError → ConnectionError → HTTPError → RequestException。
• REST：名词 URL + HTTP 方法表达动作 + 无状态 + 统一返回结构 + 用状态码表达结果 + 可缓存；分页、鉴权（Bearer / Basic / API Key）、长任务轮询（指数退避 + 抖动）都要标准化。
• 下载：stream=True + iter_content 流式落盘；先写 .part 再原子改名；断点续传靠 Range + 206 Partial Content；并发分块下载要断言 206，避免流量暴增；下载后校验大小与 SHA-256。
• 调试：浏览器 DevTools 看请求、curl -v 做命令行对照、httpbin.org 回显、mitmproxy 脚本化抓包、netcheck.py 分层定位（DNS → TCP → TLS → HTTP）。
• 安全：永远不关 verify；Token 走环境变量并脱敏；防 SSRF（协议白名单 + 内网地址拦截）；防路径穿越；限制响应体大小与解压规模。

练习题

1. 实现带长度前缀的 TCP 客户端/服务端：实现 send_msg / recv_msg 两个函数，协议为「4 字节大端长度 + UTF-8 正文」。服务端能同时处理 5 个客户端连接，每个连接可以连续发多条消息，服务端把每条消息原样加前缀后回显。要求：正文长度不能超过 1MB；连接关闭时服务端要能正确退出接收循环；加入 10 秒空闲超时。

   参考答案要点：HEADER = struct.Struct("!I")；recv_exactly(sock, n) 循环 recv 直到读满；recv 返回 b"" 时返回 None 表示对端关闭；长度字段要做上界校验防内存攻击；服务端用线程处理每个连接（threading.Thread(target=handle, daemon=True)）。测试时先发一条 10 万字的长消息验证不会被截断。

2. 多线程 TCP 聊天室（带用户列表与私聊）：在 s2 的基础上升级：客户端登录时输入昵称；服务器维护用户字典；支持命令 /who（列出在线用户）、/msg 昵称 内容（私聊）、/quit；任何异常都不能让服务器退出；退出时广播「XXX 离开了」。

   参考答案要点：用户字典用 {sock: nickname} 并用同一把 threading.Lock 保护（查找与删除要在同一临界区）；命令解析用 line.split(maxsplit=2)；私聊要把目标昵称从字典反查 socket，找不到提示「用户不存在」；finally 中统一从字典移除并广播；给 recv 设 60 秒超时并在超时后发心跳或断开，避免僵死连接。

3. UDP 丢包率实验：写两个程序，客户端向服务端连发 1000 个带序号的包，服务端统计收到的序号并计算丢包率与乱序率。分别在 localhost 与局域网（两台机器）下运行，比较结果并解释差异。再实现一个简单的「序号 + 超时重传」版本，把丢包率降到 0（允许重复，但服务端要去重）。

   参考答案要点：丢包率 = 1 - len(received)/1000；乱序 = 收到的序号非单调递增的次数。localhost 的 UDP 几乎不丢包（内核直接投递），局域网也低，出现明显丢包通常是接收端 socket 缓冲区不足——解决办法是服务端尽快 recvfrom 并调大 SO_RCVBUF（sock.setsockopt(SOL_SOCKET, SO_RCVBUF, 1<<20)）。可靠版用 set 记录已收序号去重，ACK 回传序号，客户端超时未收到 ACK 则重发。

4. 用 requests 重写所有请求并对比：把本章所有 urllib 示例（GET 带参、POST JSON、POST 表单、Basic 认证、下载）用 requests 各写一遍，输出一份对比表：代码行数、是否能自动解析 JSON、Cookie 是否自动保留、异常类型差异、连接是否复用。

   参考答案要点：urllib 需要 urllib.parse.urlencode + json.dumps + Content-Type 三件事，requests 只要 params= / json=。urllib 用 build_opener(HTTPCookieProcessor(CookieJar())) 才能维持 Cookie，requests 的 Session 天生带。连接复用上 requests 每次调用都新建连接（除非用 Session），urllib 同样如此——结论：两边都必须显式用 Session/opener 才有连接复用，这是最容易答错的一点。

5. 会话保持与登录态抓取：用 requests.Session() 访问一个需要登录才能看的页面（如 httpbin.org 的 /cookies/set 与 /cookies，或自建 Flask 登录页），验证 Cookie 在多次请求间保持；然后模拟「过期」：手动清空 session.cookies，观察接口行为变化。再实现一个「刷新 Token」的装饰器：收到 401 时自动重新登录并重放一次请求。

   参考答案要点：session.cookies.clear() 后再次请求会变成未登录。装饰器核心：resp = func(...); if resp.status_code == 401: login(session); return func(...)；注意只重放一次以防死循环，且要判断请求是否幂等（POST 重放可能造成重复下单）。

6. 实现可靠的下载器（断点续传）：实现 download(url, dest)：先 HEAD 拿 Content-Length 与 Accept-Ranges；用 Range: bytes=start- 从已下载长度继续；写入 .part，完成后 os.replace 改名。测试方法：先下载一半手动中断，再运行一次程序验证能续传而不是从头下。最后加一个 sha256 校验。

   参考答案要点：existing = Path(dest + ".part").stat().st_size if exists else 0；若 Accept-Ranges 不含 bytes 则删除 .part 从头下；resp.status_code 必须是 206 才是真续传，200 说明服务端忽略了 Range（要覆盖写入模式）；用 os.replace 保证原子性；最后 hashlib.sha256 全文件比对。

7. REST 客户端类（CRUD 全实现）：封装一个 ApiClient，支持 get/post/put/patch/delete，内置：统一 base_url、统一 User-Agent、timeout=(3,10)、3 次指数退避重试（仅幂等方法）、自定义 ApiError(status, message, payload) 异常、Bearer Token 鉴权、分页迭代器 iterate(path)。用 httpbin.org 验证全部方法可用。

   参考答案要点：核心是一个私有 _request(method, path, **kw)，其余方法只做转发；重试判定条件为「抛 RequestException 或状态码 ∈ {429,500,502,503,504} 且 attempt < max_retry」；raise_for_status() 之前先把错误 body 解析出来塞进 ApiError.payload，否则 4xx 的排查信息全丢；分页迭代器要同时兼容「page 参数」与「next_cursor」两种风格，且要有最大轮次保护防死循环。

8. 分层故障排查演练：制造并定位以下每种故障，记录现象与修复过程：①DNS 解析失败（改 hosts 到不存在的 IP）；②连接被拒（访问没监听的端口）；③连接超时（防火墙 DROP 规则或访问内网地址）；④TLS 证书错误（访问自签名站点）；⑤HTTP 500（自建 Flask 抛异常）；⑥读取超时（Flask 里 time.sleep(30)，客户端 read timeout 设 2）。要求每种情况都先跑 netcheck.py 判定卡在哪一层。

   参考答案要点：① dns_ok 报 gaierror；② tcp_ok 报 Connection refused（RST 立刻返回）；③ tcp_ok 超时（无响应）；④ tls_ok 报 CERTIFICATE_VERIFY_FAILED，修法是导入自签 CA 到 REQUESTS_CA_BUNDLE 而不是 verify=False；⑤ DNS/TCP/TLS 全 OK 但 status_code=500；⑥ 抛 ReadTimeout，注意它属于「已连接」阶段，说明前面的层都通了。结论：分层定位能把「网络问题」和「应用问题」彻底分开，这是排障效率的关键。

9. 并发分块下载并校验（挑战题）：在第 6 题基础上升级：把文件切成 N 块用 ThreadPoolExecutor 并发下载，每块独立 .part{i} 文件；任一块失败只重试该块；全部完成后按序合并；合并后校验总字节数与 SHA-256。用 requests.Session + 调大 pool_maxsize 避免连接池成为瓶颈。给出你实测的加速比（4 线程 / 8 线程）。

   参考答案要点：加速比通常在 2~6 倍，不随线程数线性增长——因为服务端/链路带宽是上限。必须断言每块返回 206，若某块返回 200 说明 Range 被忽略，该块会拿到整个文件（iter_content 写满就停，事后靠大小校验发现）。pool_maxsize 至少等于并发块数并设 pool_block=True，否则会出现 ConnectionError: Connection pool is full。合并不用读进内存，用 shutil.copyfileobj 流式写。