从零开始学HTTP协议网络编程用Python实战模拟浏览器发送请求处理网页数据理解状态码与请求头让程序员彻底掌握HTTP通信原理

你好呀！今天咱们来聊聊一个每个程序员都必须搞懂的硬核知识——HTTP协议。别被这个名字吓到，其实它比你想象的要亲切得多。想象一下，你在浏览网页时，浏览器就像一个超级热情的信使，帮你把请求送出去，再把结果带回来。这个过程背后，就是HTTP协议在默默工作。

咱们今天就用Python，一点一点把这个神秘的面纱揭开，让你真正理解HTTP通信的每一个环节。准备好了吗？咱们开始！

一、HTTP到底是个啥？

HTTP，全称是HyperText Transfer Protocol（超文本传输协议）。听起来很高大上，其实它就是互联网上最基础的通信规则。你输入一个网址，按回车，浏览器怎么知道去哪拿数据？怎么告诉服务器你想要什么？这一切都是通过HTTP协议完成的。

可以把HTTP想象成你和餐厅服务员之间的对话：

• 你（客户端）对服务员说：”我想要一份宫保鸡丁” → 这就是请求
• 服务员（服务器）回来后告诉你：”好的，请稍等” → 这就是响应

就这么简单！不过HTTP比这个复杂得多，因为它要处理各种各样的场景。

二、HTTP请求的完整结构

一个HTTP请求长什么样呢？咱们来看一个真实例子：

POST /api/login HTTP/1.1
Host: www.example.com
User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36
Accept: text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8
Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
Content-Type: application/json
Content-Length: 52
Authorization: Bearer eyJhbGciOiJIUzI1NiIs...

{"username":"zhangsan","password":"123456"}

看到没？一个完整的请求分为三个部分：

1. 请求行：POST /api/login HTTP/1.1 —— 告诉服务器用什么方法、访问哪个路径、用哪个版本
2. 请求头：一堆Key: Value对，告诉服务器各种信息，比如浏览器类型、接受什么格式的数据等
3. 请求体：实际要发送的数据，这里是一个JSON字符串

咱们用Python来模拟这个过程：

import http.client

# 建立连接
conn = http.client.HTTPSConnection("www.example.com")

# 构造请求头
headers = {
    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
    'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
    'Content-Type': 'application/json',
    'Authorization': 'Bearer eyJhbGciOiJIUzI1NiIs...'
}

# 构造请求体
body = '{"username":"zhangsan","password":"123456"}'

# 发送请求
conn.request("POST", "/api/login", body=body, headers=headers)

# 获取响应
response = conn.getresponse()
print(f"状态码: {response.status}")
print(f"响应头: {response.getheaders()}")
print(f"响应体: {response.read().decode()}")

# 关闭连接
conn.close()

是不是很简单？不过这只是冰山一角，咱们继续深入。

三、HTTP方法：GET、POST、PUT、DELETE…

HTTP定义了多种方法（也叫动词），每种方法都有不同的含义：

• GET：获取数据，最常用，比如访问一个网页
• POST：提交数据，比如登录、注册
• PUT：更新数据，比如修改个人资料
• DELETE：删除数据
• HEAD：只获取响应头，不获取 body
• OPTIONS：查询服务器支持的方法

让我用代码演示一下这些方法的区别：

import requests

BASE_URL = "https://api.example.com"

# GET请求 - 获取用户列表
def get_users():
    response = requests.get(f"{BASE_URL}/users")
    print(f"获取用户列表: {response.status_code}")
    print(f"数据: {response.json()}")
    return response.json()

# POST请求 - 创建新用户
def create_user(username, email):
    payload = {
        "username": username,
        "email": email
    }
    response = requests.post(f"{BASE_URL}/users", json=payload)
    print(f"创建用户: {response.status_code}")
    print(f"返回数据: {response.json()}")
    return response.json()

# PUT请求 - 更新用户信息
def update_user(user_id, new_email):
    response = requests.put(f"{BASE_URL}/users/{user_id}", json={"email": new_email})
    print(f"更新用户: {response.status_code}")
    return response.json()

# DELETE请求 - 删除用户
def delete_user(user_id):
    response = requests.delete(f"{BASE_URL}/users/{user_id}")
    print(f"删除用户: {response.status_code}")
    return response.status_code

# 测试
if __name__ == "__main__":
    users = get_users()
    new_user = create_user("lisi", "lisi@example.com")
    update_user(new_user["id"], "lisi_new@example.com")
    delete_user(new_user["id"])

四、状态码：服务器给咱们的”表情”

每次服务器响应请求时，都会返回一个状态码。这个状态码就像是服务器在跟你说话：

200 OK          → 一切正常，请求成功
201 Created     → 创建成功
204 No Content  → 成功，但没有返回数据
301 Moved       → 永久重定向
302 Found       → 临时重定向
304 Not Modified → 数据未改变，可以使用缓存
400 Bad Request → 请求有误
401 Unauthorized → 未授权，需要登录
403 Forbidden   → 禁止访问
404 Not Found   → 找不到资源
405 Method Not Allowed → 方法不允许
500 Internal Server Error → 服务器内部错误
502 Bad Gateway → 网关错误
503 Service Unavailable → 服务不可用
504 Gateway Timeout → 网关超时

咱们用Python来测试各种状态码：

import requests

def test_status_codes():
    """测试不同的状态码"""
    urls = [
        ("正常页面", "https://httpbin.org/get"),
        ("重定向", "https://httpbin.org/redirect/1"),
        ("404错误", "https://httpbin.org/status/404"),
        ("403错误", "https://httpbin.org/status/403"),
        ("500错误", "https://httpbin.org/status/500"),
    ]
    
    for name, url in urls:
        try:
            response = requests.get(url, timeout=5)
            print(f"{name}: 状态码 {response.status_code}")
        except requests.exceptions.RequestException as e:
            print(f"{name}: 请求异常 - {e}")

test_status_codes()

输出结果：

正常页面: 状态码 200
重定向: 状态码 302
404错误: 状态码 404
403错误: 状态码 403
500错误: 状态码 500

五、请求头：告诉服务器你想怎么接收数据

请求头包含了很多重要信息，咱们来一个一个看：

import requests

def show_request_headers():
    """展示完整的请求头信息"""
    # 访问httpbin.org，它会回显我们发送的headers
    response = requests.get(
        "https://httpbin.org/headers",
        headers={
            "User-Agent": "MyCustomAgent/1.0",
            "Accept": "application/json",
            "Accept-Language": "zh-CN,zh;q=0.9",
            "Custom-Header": "MyValue"
        }
    )
    
    data = response.json()
    print("服务器收到的Headers:")
    for key, value in data["headers"].items():
        print(f"  {key}: {value}")

show_request_headers()

常见的请求头：

请求头	作用	示例
Host	指定服务器域名	Host: www.example.com
User-Agent	浏览器标识	User-Agent: Mozilla/5.0...
Accept	客户端能接受的数据类型	Accept: application/json
Accept-Language	接受的语言	Accept-Language: zh-CN
Content-Type	请求体的数据类型	Content-Type: application/json
Authorization	认证信息	Authorization: Bearer token
Cookie	Cookie信息	Cookie: session_id=abc123
Referer	来源页面	Referer: https://www.example.com
Cache-Control	缓存控制	Cache-Control: no-cache

六、响应头：服务器告诉咱们的信息

响应头和请求头结构类似，但包含的是服务器端的信息：

import requests

def show_response_headers():
    """查看响应头信息"""
    response = requests.get("https://httpbin.org/get")
    
    print("响应头信息:")
    for key, value in response.headers.items():
        print(f"  {key}: {value}")

show_response_headers()

常见响应头：

响应头	作用
Content-Type	响应体的数据类型
Content-Length	响应体的长度
Set-Cookie	设置Cookie
Location	重定向目标
Cache-Control	缓存控制
Access-Control-Allow-Origin	CORS跨域配置
Server	服务器软件信息

七、用Python从零实现一个HTTP客户端

学到这里，你可能会想：requests库确实方便，但底层是怎么工作的呢？咱们用Python的socket自己实现一个HTTP客户端：

import socket
import ssl

def raw_http_request(host, path="/", method="GET", headers=None, body=None):
    """
    使用原始socket实现HTTP请求
    这是理解HTTP协议最底层的方式
    """
    # 构建请求行和请求头
    request_lines = [f"{method} {path} HTTP/1.1"]
    
    # 默认请求头
    default_headers = {
        "Host": host,
        "User-Agent": "Python-RawHTTP/1.0",
        "Accept": "*/*",
        "Connection": "close"
    }
    
    # 合并请求头
    if headers:
        default_headers.update(headers)
    
    # 如果有body，添加Content-Length
    if body:
        body_bytes = body.encode('utf-8') if isinstance(body, str) else body
        default_headers["Content-Length"] = str(len(body_bytes))
    
    # 添加请求头
    for key, value in default_headers.items():
        request_lines.append(f"{key}: {value}")
    
    # 空行分隔头和body
    request_lines.append("")
    
    # 如果有body，追加到请求中
    if body:
        request_lines.append(body if isinstance(body, str) else body.decode('utf-8'))
    
    # 拼接完整请求
    request = "\r\n".join(request_lines) + "\r\n"
    
    # 创建socket连接
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    
    try:
        # 如果是HTTPS，需要SSL包装
        if host.startswith("https://"):
            host = host.replace("https://", "")
            context = ssl.create_default_context()
            sock = context.wrap_socket(sock, server_hostname=host)
        else:
            host = host.replace("http://", "")
        
        # 连接服务器
        sock.connect((host, 443 if "https" in host else 80))
        
        # 发送请求
        sock.sendall(request.encode('utf-8'))
        
        # 接收响应
        response = b""
        while True:
            chunk = sock.recv(4096)
            if not chunk:
                break
            response += chunk
            # 如果已经接收完所有内容，可以提前退出
            if b"\r\n\r\n" in response:
                header_end = response.index(b"\r\n\r\n") + 4
                content_length = 0
                headers_text = response[:header_end].decode('utf-8', errors='ignore')
                for line in headers_text.split('\r\n'):
                    if line.lower().startswith('content-length:'):
                        content_length = int(line.split(':')[1].strip())
                        break
                if content_length > 0 and len(response) - header_end >= content_length:
                    break
        
        return response
    finally:
        sock.close()


def parse_response(response_bytes):
    """解析HTTP响应"""
    # 分离头部和body
    if b"\r\n\r\n" in response_bytes:
        header_bytes, body_bytes = response_bytes.split(b"\r\n\r\n", 1)
    else:
        header_bytes = response_bytes
        body_bytes = b""
    
    # 解析响应行
    lines = header_bytes.decode('utf-8', errors='ignore').split('\r\n')
    status_line = lines[0]
    
    # 解析状态码
    parts = status_line.split(' ')
    status_code = int(parts[1]) if len(parts) > 1 else 0
    status_text = ' '.join(parts[2:]) if len(parts) > 2 else ""
    
    # 解析响应头
    headers = {}
    for line in lines[1:]:
        if ':' in line:
            key, value = line.split(':', 1)
            headers[key.strip()] = value.strip()
    
    return {
        "status_code": status_code,
        "status_text": status_text,
        "headers": headers,
        "body": body_bytes.decode('utf-8', errors='ignore')
    }


# 测试
if __name__ == "__main__":
    # GET请求
    response = raw_http_request("httpbin.org", "/get")
    parsed = parse_response(response)
    print(f"状态码: {parsed['status_code']}")
    print(f"响应头: {parsed['headers']}")
    print(f"响应体长度: {len(parsed['body'])}")
    
    # POST请求
    response = raw_http_request(
        "httpbin.org", 
        "/post", 
        method="POST",
        body='{"username": "test", "password": "123456"}'
    )
    parsed = parse_response(response)
    print(f"\nPOST请求状态码: {parsed['status_code']}")

运行这个代码，你会看到：

状态码: 200
响应头: {'Accept-Encoding': 'gzip, deflate, br', 'Content-Length': '253', 'Content-Type': 'application/json', ...}
响应体长度: 253

这个原始实现展示了HTTP协议最本质的样子——就是TCP socket上发送和接收字符串！

八、模拟浏览器发送请求：伪装成Chrome

很多网站会检测User-Agent，如果不是真实浏览器，可能会拒绝服务。咱们来模拟一个真实的Chrome浏览器：

import requests
import random

# 真实的浏览器User-Agent列表
USER_AGENTS = [
    # Chrome on Windows
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    # Chrome on Mac
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
    # Firefox on Windows
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
    # Safari on Mac
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15",
    # Edge on Windows
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0",
]

def simulate_browser(url, method="GET", headers=None, cookies=None, proxy=None):
    """
    模拟真实浏览器发送请求
    """
    session = requests.Session()
    
    # 随机选择User-Agent
    ua = random.choice(USER_AGENTS)
    
    # 构建完整的浏览器请求头
    browser_headers = {
        "User-Agent": ua,
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
        "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,zh-TW;q=0.7",
        "Accept-Encoding": "gzip, deflate, br",
        "Connection": "keep-alive",
        "Upgrade-Insecure-Requests": "1",
        "Sec-Ch-Ua": '"Chromium";v="120", "Not_A Brand";v="24"',
        "Sec-Ch-Ua-Mobile": "?0",
        "Sec-Ch-Ua-Platform": '"Windows"',
        "Sec-Fetch-Dest": "document",
        "Sec-Fetch-Mode": "navigate",
        "Sec-Fetch-Site": "none",
        "Sec-Fetch-User": "?1",
        "Cache-Control": "max-age=0",
        "Pragma": "no-cache"
    }
    
    # 合并自定义headers
    if headers:
        browser_headers.update(headers)
    
    # 设置cookies
    if cookies:
        session.cookies.update(cookies)
    
    # 设置代理
    proxies = None
    if proxy:
        proxies = {
            "http": proxy,
            "https": proxy
        }
    
    # 发送请求
    try:
        response = session.request(
            method=method,
            url=url,
            headers=browser_headers,
            proxies=proxies,
            timeout=10,
            allow_redirects=True
        )
        
        # 打印请求详情
        print(f"\n{'='*50}")
        print(f"请求URL: {response.request.url}")
        print(f"请求方法: {response.request.method}")
        print(f"请求头:")
        for k, v in response.request.headers.items():
            print(f"  {k}: {v}")
        print(f"\n响应状态码: {response.status_code}")
        print(f"响应头:")
        for k, v in response.headers.items():
            print(f"  {k}: {v}")
        print(f"响应体长度: {len(response.content)} 字节")
        print(f"{'='*50}\n")
        
        return response
        
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return None


# 测试
if __name__ == "__main__":
    # 模拟访问百度
    simulate_browser("https://www.baidu.com")
    
    # 模拟访问Google
    simulate_browser("https://www.google.com")
    
    # 模拟访问API
    simulate_browser("https://httpbin.org/get")

九、处理网页数据：解析HTML和JSON

拿到响应数据后，咱们还需要解析它。常用的有HTML解析和JSON解析：

import requests
import json
from bs4 import BeautifulSoup

def parse_json_response(url):
    """解析JSON响应"""
    response = requests.get(url)
    
    if response.status_code == 200:
        # requests自动解析JSON
        data = response.json()
        print("JSON数据:")
        print(json.dumps(data, indent=2, ensure_ascii=False))
        
        # 提取特定字段
        if "origin" in data:
            print(f"\n客户端IP: {data['origin']}")
        if "headers" in data:
            print(f"请求头数量: {len(data['headers'])}")
        
        return data
    else:
        print(f"请求失败: {response.status_code}")
        return None


def parse_html_response(url):
    """解析HTML页面"""
    response = requests.get(url)
    
    if response.status_code == 200:
        soup = BeautifulSoup(response.text, 'html.parser')
        
        # 提取标题
        title = soup.title.string if soup.title else "无标题"
        print(f"页面标题: {title}")
        
        # 提取所有链接
        links = []
        for a in soup.find_all('a', href=True):
            links.append({
                "text": a.get_text(strip=True),
                "href": a['href']
            })
        
        print(f"\n页面链接数量: {len(links)}")
        for i, link in enumerate(links[:10]):  # 只显示前10个
            print(f"  {i+1}. {link['text']} -> {link['href']}")
        
        # 提取所有图片
        images = []
        for img in soup.find_all('img'):
            src = img.get('src') or img.get('data-src')
            if src:
                images.append(src)
        
        print(f"\n图片数量: {len(images)}")
        
        # 提取文本内容（去掉标签）
        text = soup.get_text(separator='\n', strip=True)
        print(f"\n页面文本预览（前500字符）:")
        print(text[:500])
        
        return {
            "title": title,
            "links": links,
            "images": images,
            "text": text
        }
    else:
        print(f"请求失败: {response.status_code}")
        return None


# 测试
if __name__ == "__main__":
    print("=== 解析JSON ===")
    parse_json_response("https://httpbin.org/get")
    
    print("\n\n=== 解析HTML ===")
    parse_html_response("https://httpbin.org/html")

十、处理Cookie和Session

Cookie和Session是保持登录状态的关键：

import requests
from http.cookiejar import CookieJar

def handle_cookies_example():
    """处理Cookie的完整示例"""
    session = requests.Session()
    
    # 1. 自动管理Cookie
    print("=== 1. 自动Cookie管理 ===")
    response1 = session.get("https://httpbin.org/cookies/set/name/value")
    print(f"设置Cookie响应: {response1.status_code}")
    
    response2 = session.get("https://httpbin.org/cookies")
    print(f"获取Cookie: {response2.json()}")
    
    # 2. 手动设置Cookie
    print("\n=== 2. 手动设置Cookie ===")
    session.cookies.set("session_id", "abc123xyz")
    session.cookies.set("user_name", "test_user", domain="httpbin.org")
    
    response3 = session.get("https://httpbin.org/cookies")
    print(f"手动设置的Cookie: {response3.json()}")
    
    # 3. 导入Cookie
    print("\n=== 3. 导入Cookie ===")
    cookie_jar = session.cookies
    cookie_jar.set_cookie(requests.cookies.create_cookie(
        name="imported_cookie",
        value="imported_value",
        domain="httpbin.org",
        path="/"
    ))
    
    response4 = session.get("https://httpbin.org/cookies")
    print(f"导入的Cookie: {response4.json()}")
    
    # 4. 处理Set-Cookie响应头
    print("\n=== 4. 处理Set-Cookie ===")
    response5 = session.get("https://httpbin.org/cookies/set/foo/bar")
    print(f"Set-Cookie响应头: {response5.headers.get('Set-Cookie')}")
    print(f"当前Cookie: {session.cookies.get_dict()}")


def simulate_login_flow():
    """模拟完整登录流程"""
    session = requests.Session()
    
    print("\n=== 模拟登录流程 ===")
    
    # Step 1: 访问登录页面，获取CSRF token
    login_page = session.get("https://httpbin.org/cookies")
    print(f"Step 1 - 访问登录页: {login_page.status_code}")
    
    # Step 2: 提交登录表单
    login_data = {
        "username": "myuser",
        "password": "mypassword",
        # 实际项目中需要从页面提取CSRF token
    }
    
    login_response = session.post(
        "https://httpbin.org/post",
        data=login_data,
        headers={
            "Content-Type": "application/x-www-form-urlencoded"
        }
    )
    print(f"Step 2 - 提交登录: {login_response.status_code}")
    
    # Step 3: 检查是否登录成功
    if login_response.status_code == 200:
        print("Step 3 - 登录成功！")
        print(f"Session Cookie: {session.cookies.get_dict()}")
    
    # Step 4: 访问需要登录的页面
    protected_page = session.get("https://httpbin.org/headers")
    print(f"Step 4 - 访问受保护页面: {protected_page.status_code}")
    print(f"响应数据: {protected_page.json()}")


# 运行
if __name__ == "__main__":
    handle_cookies_example()
    simulate_login_flow()

十一、处理重定向

网站经常使用重定向，咱们需要理解并处理它们：

import requests

def handle_redirects():
    """处理重定向的完整示例"""
    
    # 1. 自动跟随重定向
    print("=== 1. 自动跟随重定向 ===")
    response = requests.get(
        "https://httpbin.org/redirect/3",
        allow_redirects=True
    )
    print(f"最终URL: {response.url}")
    print(f"状态码: {response.status_code}")
    print(f"重定向历史: {[r.url for r in response.history]}")
    
    # 2. 不跟随重定向
    print("\n=== 2. 不跟随重定向 ===")
    response = requests.get(
        "https://httpbin.org/redirect/1",
        allow_redirects=False
    )
    print(f"状态码: {response.status_code}")
    print(f"Location头: {response.headers.get('Location')}")
    
    # 3. 处理循环重定向
    print("\n=== 3. 处理循环重定向 ===")
    try:
        response = requests.get(
            "https://httpbin.org/redirect/5",
            allow_redirects=True,
            timeout=10
        )
        print(f"成功! 最终URL: {response.url}")
    except requests.exceptions.TooManyRedirects as e:
        print(f"重定向次数过多: {e}")
    
    # 4. 自定义重定向处理
    print("\n=== 4. 自定义重定向处理 ===")
    
    class CustomRedirectSession(requests.Session):
        def send(self, request, **kwargs):
            # 在发送前修改请求
            print(f"发送请求到: {request.url}")
            return super().send(request, **kwargs)
        
        def resolve_redirects(self, resp, req, **kwargs):
            # 自定义重定向逻辑
            print(f"检测到重定向: {resp.status_code} -> {resp.headers.get('Location')}")
            
            # 只跟随301和302重定向
            if resp.status_code not in (301, 302):
                print("不支持的重定向类型，停止跟随")
                return []
            
            return super().resolve_redirects(resp, req, **kwargs)
    
    session = CustomRedirectSession()
    response = session.get("https://httpbin.org/redirect/2")
    print(f"最终响应URL: {response.url}")


if __name__ == "__main__":
    handle_redirects()

十二、实战：爬取网页数据

现在咱们把学到的知识整合起来，写一个完整的网页爬虫：

import requests
from bs4 import BeautifulSoup
import time
import json
import re
from urllib.parse import urljoin, urlparse

class WebCrawler:
    """一个简单的网页爬虫"""
    
    def __init__(self, delay=1, max_pages=100, user_agent=None):
        self.session = requests.Session()
        self.delay = delay  # 请求间隔，避免过于频繁
        self.max_pages = max_pages
        self.visited_urls = set()
        self.results = []
        
        # 设置默认User-Agent
        self.session.headers.update({
            "User-Agent": user_agent or "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
            "Accept-Encoding": "gzip, deflate, br",
            "Connection": "keep-alive",
        })
    
    def fetch_page(self, url):
        """获取单个页面"""
        if url in self.visited_urls:
            return None
        
        self.visited_urls.add(url)
        
        try:
            response = self.session.get(url, timeout=10)
            response.raise_for_status()
            
            # 检查Content-Type
            content_type = response.headers.get('Content-Type', '')
            if 'text/html' not in content_type and 'html' not in content_type:
                print(f"跳过非HTML页面: {url}")
                return None
            
            time.sleep(self.delay)
            return response
        
        except requests.exceptions.RequestException as e:
            print(f"请求失败: {url} - {e}")
            return None
    
    def parse_page(self, response, base_url):
        """解析页面内容"""
        soup = BeautifulSoup(response.text, 'html.parser')
        
        # 提取标题
        title = soup.title.string if soup.title else "无标题"
        
        # 提取所有链接
        links = []
        for a in soup.find_all('a', href=True):
            href = a['href']
            # 处理相对路径
            full_url = urljoin(base_url, href)
            # 过滤掉锚点和JavaScript链接
            if full_url.startswith(('http://', 'https://')) and '#' not in href:
                links.append(full_url)
        
        # 提取图片
        images = []
        for img in soup.find_all('img'):
            src = img.get('src') or img.get('data-src')
            if src:
                full_src = urljoin(base_url, src)
                images.append(full_src)
        
        # 提取文本内容
        # 移除脚本和样式
        for tag in soup(['script', 'style', 'nav', 'footer', 'header']):
            tag.decompose()
        text = soup.get_text(separator='\n', strip=True)
        
        # 提取元数据
        meta_description = ""
        meta_keywords = ""
        desc_tag = soup.find('meta', attrs={'name': 'description'})
        if desc_tag:
            meta_description = desc_tag.get('content', '')
        keywords_tag = soup.find('meta', attrs={'name': 'keywords'})
        if keywords_tag:
            meta_keywords = keywords_tag.get('content', '')
        
        return {
            "url": base_url,
            "title": title,
            "links": list(set(links)),  # 去重
            "images": list(set(images)),
            "text": text[:1000],  # 限制长度
            "meta_description": meta_description,
            "meta_keywords": meta_keywords,
        }
    
    def crawl(self, start_url, max_depth=2):
        """开始爬取"""
        print(f"开始爬取: {start_url}")
        print(f"最大深度: {max_depth}, 最大页数: {self.max_pages}")
        print("=" * 60)
        
        # 使用队列进行BFS爬取
        from collections import deque
        queue = deque([(start_url, 0)])  # (url, depth)
        
        while queue and len(self.visited_urls) < self.max_pages:
            url, depth = queue.popleft()
            
            if depth > max_depth:
                continue
            
            print(f"\n[{len(self.visited_urls)}] 爬取: {url} (深度: {depth})")
            
            response = self.fetch_page(url)
            if not response:
                continue
            
            data = self.parse_page(response, url)
            self.results.append(data)
            
            # 打印解析结果
            print(f"  标题: {data['title']}")
            print(f"  链接数: {len(data['links'])}")
            print(f"  图片数: {len(data['images'])}")
            
            # 将新链接加入队列
            if depth < max_depth:
                for link in data['links'][:10]:  # 限制每页最多10个新链接
                    if link not in self.visited_urls:
                        queue.append((link, depth + 1))
            
            # 进度显示
            print(f"  已爬取: {len(self.visited_urls)}/{self.max_pages}")
        
        print("\n" + "=" * 60)
        print(f"爬取完成! 共爬取 {len(self.results)} 个页面")
        return self.results
    
    def save_results(self, filename="crawl_results.json"):
        """保存结果到JSON文件"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.results, f, ensure_ascii=False, indent=2)
        print(f"结果已保存到: {filename}")


# 使用示例
if __name__ == "__main__":
    crawler = WebCrawler(delay=0.5, max_pages=20)
    results = crawler.crawl("https://httpbin.org", max_depth=1)
    crawler.save_results()

十三、处理HTTPS和SSL

HTTPS是加密的HTTP，咱们来看看如何处理：

import requests
import ssl
import urllib3

# 禁用SSL警告（仅用于测试环境）
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

def handle_https():
    """HTTPS相关操作"""
    
    # 1. 正常请求（验证证书）
    print("=== 1. 正常HTTPS请求 ===")
    response = requests.get("https://www.baidu.com")
    print(f"状态码: {response.status_code}")
    print(f"证书验证: 成功")
    
    # 2. 忽略证书验证（不推荐用于生产环境）
    print("\n=== 2. 忽略证书验证 ===")
    response = requests.get("https://self-signed.badssl.com/", verify=False)
    print(f"状态码: {response.status_code}")
    
    # 3. 使用自定义证书
    print("\n=== 3. 使用自定义证书 ===")
    try:
        response = requests.get("https://www.baidu.com", verify="/path/to/ca-bundle.crt")
        print(f"状态码: {response.status_code}")
    except Exception as e:
        print(f"自定义证书错误: {e}")
    
    # 4. 查看SSL证书信息
    print("\n=== 4. 查看SSL证书信息 ===")
    response = requests.get("https://www.baidu.com")
    cert = response.raw.cert
    if cert:
        print(f"证书主题: {cert.get('subject')}")
        print(f"证书有效期: {cert.get('validity')}")
    else:
        # 获取证书详细信息
        import ssl
        context = ssl.create_default_context()
        with context.wrap_socket(socket.socket(), server_hostname='www.baidu.com') as s:
            s.connect(('www.baidu.com', 443))
            cert = s.getpeercert()
            print(f"证书颁发者: {cert.get('issuer')}")
            print(f"证书有效期: {cert.get('notAfter')}")


def handle_certificate_pinning():
    """证书锁定（高级安全技巧）"""
    print("\n=== 5. 证书锁定 ===")
    
    # 这里演示概念，实际使用时需要获取目标服务器的证书指纹
    # pinned_cert_pem = "MIICyDCCAbCgAwIBAgIQ..."
    
    # 简单示例：检查证书是否匹配
    response = requests.get("https://www.baidu.com")
    cert_pem = requests.get("https://www.baidu.com").raw.cert
    print(f"证书指纹: {cert_pem}")


import socket

if __name__ == "__main__":
    handle_https()

十四、HTTP缓存机制

理解缓存可以大大提升爬虫效率：

import requests
from datetime import datetime

def handle_cache():
    """处理HTTP缓存"""
    
    print("=== 1. 使用Cache-Control ===")
    
    # no-cache: 每次都向服务器验证
    response1 = requests.get(
        "https://httpbin.org/cache",
        headers={"Cache-Control": "no-cache"}
    )
    print(f"no-cache响应: {response1.status_code}")
    
    # no-store: 不存储任何缓存
    response2 = requests.get(
        "https://httpbin.org/cache",
        headers={"Cache-Control": "no-store"}
    )
    print(f"no-store响应: {response2.status_code}")
    
    # max-age: 指定缓存时间
    response3 = requests.get(
        "https://httpbin.org/cache/60",
        headers={"Cache-Control": "max-age=60"}
    )
    print(f"max-age=60响应: {response3.status_code}")
    
    print("\n=== 2. 使用ETag和Last-Modified ===")
    
    # 第一次请求
    response_a = requests.get("https://httpbin.org/etag")
    etag = response_a.headers.get('ETag')
    last_modified = response_a.headers.get('Last-Modified')
    print(f"ETag: {etag}")
    print(f"Last-Modified: {last_modified}")
    
    # 使用ETag进行条件请求
    if etag:
        response_b = requests.get(
            "https://httpbin.org/etag",
            headers={"If-None-Match": etag}
        )
        print(f"条件请求(ETag)状态码: {response_b.status_code}")
        if response_b.status_code == 304:
            print("数据未变化，使用缓存")
    
    # 使用Last-Modified进行条件请求
    if last_modified:
        response_c = requests.get(
            "https://httpbin.org/response-headers",
            headers={"If-Modified-Since": last_modified}
        )
        print(f"条件请求(Last-Modified)状态码: {response_c.status_code}")
    
    print("\n=== 3. 实现简单的缓存系统 ===")
    
    class SimpleCache:
        def __init__(self, ttl=300):
            self.cache = {}
            self.ttl = ttl  # 缓存过期时间（秒）
        
        def get(self, url):
            if url in self.cache:
                entry = self.cache[url]
                if datetime.now().timestamp() - entry['timestamp'] < self.ttl:
                    print(f"命中缓存: {url}")
                    return entry['response']
            print(f"缓存未命中: {url}")
            return None
        
        def set(self, url, response):
            self.cache[url] = {
                'response': response,
                'timestamp': datetime.now().timestamp()
            }
    
    cache = SimpleCache(ttl=60)
    
    url = "https://httpbin.org/get"
    
    # 第一次请求
    resp1 = cache.get(url)
    if not resp1:
        resp1 = requests.get(url)
        cache.set(url, resp1)
    print(f"第一次请求: {resp1.status_code}")
    
    # 第二次请求（应该命中缓存）
    resp2 = cache.get(url)
    if not resp2:
        resp2 = requests.get(url)
        cache.set(url, resp2)
    print(f"第二次请求: {resp2.status_code}")


if __name__ == "__main__":
    handle_cache()

十五、错误处理和重试机制

健壮的网络代码必须有错误处理：

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

def robust_request():
    """带重试和错误处理的请求"""
    
    # 1. 使用Session和Retry
    session = requests.Session()
    
    # 配置重试策略
    retry_strategy = Retry(
        total=3,              # 最大重试次数
        backoff_factor=1,     # 重试间隔因子
        status_forcelist=[429, 500, 502, 503, 504],  # 需要重试的状态码
        allowed_methods=["GET", "POST"]  # 允许重试的方法
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("http://", adapter)
    session.mount("https://", adapter)
    
    try:
        response = session.get(
            "https://httpbin.org/status/503",  # 模拟503错误
            timeout=5
        )
        print(f"请求成功: {response.status_code}")
    except requests.exceptions.HTTPError as e:
        print(f"HTTP错误: {e}")
    except requests.exceptions.ConnectionError as e:
        print(f"连接错误: {e}")
    except requests.exceptions.Timeout as e:
        print(f"超时: {e}")
    except requests.exceptions.RequestException as e:
        print(f"其他请求错误: {e}")
    
    # 2. 完整的错误处理封装
    print("\n=== 完整的请求封装 ===")
    
    def smart_request(url, method="GET", **kwargs):
        """智能请求函数"""
        default_kwargs = {
            "timeout": 10,
            "headers": {
                "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
            }
        }
        default_kwargs.update(kwargs)
        
        max_retries = 3
        retry_delay = 1
        
        for attempt in range(max_retries):
            try:
                response = requests.request(method, url, **default_kwargs)
                
                # 处理特殊状态码
                if response.status_code == 429:  # 限流
                    retry_after = int(response.headers.get('Retry-After', retry_delay * (2 ** attempt)))
                    print(f"被限流，等待 {retry_after} 秒后重试...")
                    time.sleep(retry_after)
                    continue
                
                if response.status_code == 401:  # 未授权
                    print("未授权，请检查认证信息")
                    return None
                
                if response.status_code == 403:  # 禁止访问
                    print("禁止访问")
                    return None
                
                if response.status_code == 404:  # 未找到
                    print("页面不存在")
                    return None
                
                response.raise_for_status()
                return response
                
            except requests.exceptions.Timeout:
                if attempt < max_retries - 1:
                    wait_time = retry_delay * (2 ** attempt)
                    print(f"超时，{wait_time}秒后重试 ({attempt + 1}/{max_retries})")
                    time.sleep(wait_time)
                else:
                    print(f"超时，已达最大重试次数")
                    return None
                    
            except requests.exceptions.ConnectionError:
                if attempt < max_retries - 1:
                    wait_time = retry_delay * (2 ** attempt)
                    print(f"连接错误，{wait_time}秒后重试 ({attempt + 1}/{max_retries})")
                    time.sleep(wait_time)
                else:
                    print(f"连接错误，已达最大重试次数")
                    return None
            
            except requests.exceptions.HTTPError as e:
                print(f"HTTP错误: {e}")
                return None
        
        return None
    
    # 测试
    result = smart_request("https://httpbin.org/get")
    if result:
        print(f"请求成功，状态码: {result.status_code}")
        print(f"响应数据: {result.json()}")


if __name__ == "__main__":
    robust_request()

十六、实战项目：构建一个简单的API客户端

最后，咱们来做一个完整的实战项目：

import requests
import json
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, asdict
import time

@dataclass
class APIResponse:
    """统一的API响应结构"""
    success: bool
    status_code: int
    data: Any = None
    error: str = None
    headers: Dict = None
    
    def to_dict(self):
        return asdict(self)
    
    def __bool__(self):
        return self.success


class APIClient:
    """
    一个完整的API客户端示例
    
    使用方法:
        client = APIClient(base_url="https://api.example.com")
        response = client.get("/users")
        if response:
            print(response.data)
    """
    
    def __init__(self, base_url: str, api_key: str = None, timeout: int = 30):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self.session = requests.Session()
        
        # 设置默认请求头
        self.session.headers.update({
            "Content-Type": "application/json",
            "Accept": "application/json",
            "User-Agent": "Python-API-Client/1.0"
        })
        
        # 设置API密钥（如果需要）
        if api_key:
            self.session.headers["Authorization"] = f"Bearer {api_key}"
    
    def _build_url(self, path: str) -> str:
        """构建完整URL"""
        if path.startswith(('http://', 'https://')):
            return path
        return f"{self.base_url}{path}"
    
    def _handle_response(self, response: requests.Response) -> APIResponse:
        """处理响应，统一返回格式"""
        try:
            data = response.json() if response.content else None
        except json.JSONDecodeError:
            data = response.text
            
        return APIResponse(
            success=200 <= response.status_code < 300,
            status_code=response.status_code,
            data=data,
            error=response.text if response.status_code >= 400 else None,
            headers=dict(response.headers)
        )
    
    def get(self, path: str, params: Dict = None, **kwargs) -> APIResponse:
        """发送GET请求"""
        url = self._build_url(path)
        try:
            response = self.session.get(
                url,
                params=params,
                timeout=self.timeout,
                **kwargs
            )
            return self._handle_response(response)
        except requests.exceptions.Timeout:
            return APIResponse(success=False, status_code=0, error="请求超时")
        except requests.exceptions.ConnectionError:
            return APIResponse(success=False, status_code=0, error="连接失败")
    
    def post(self, path: str, data: Any = None, json: Dict = None, **kwargs) -> APIResponse:
        """发送POST请求"""
        url = self._build_url(path)
        try:
            response = self.session.post(
                url,
                json=json,
                data=data,
                timeout=self.timeout,
                **kwargs
            )
            return self._handle_response(response)
        except requests.exceptions.Timeout:
            return APIResponse(success=False, status_code=0, error="请求超时")
        except requests.exceptions.ConnectionError:
            return APIResponse(success=False, status_code=0, error="连接失败")
    
    def put(self, path: str, json: Dict = None, **kwargs) -> APIResponse:
        """发送PUT请求"""
        url = self._build_url(path)
        try:
            response = self.session.put(
                url,
                json=json,
                timeout=self.timeout,
                **kwargs
            )
            return self._handle_response(response)
        except requests.exceptions.Timeout:
            return APIResponse(success=False, status_code=0, error="请求超时")
    
    def delete(self, path: str, **kwargs) -> APIResponse:
        """发送DELETE请求"""
        url = self._build_url(path)
        try:
            response = self.session.delete(
                url,
                timeout=self.timeout,
                **kwargs
            )
            return self._handle_response(response)
        except requests.exceptions.Timeout:
            return APIResponse(success=False, status_code=0, error="请求超时")
    
    def batch_request(self, requests_list: List[Dict]) -> List[APIResponse]:
        """
        批量发送请求
        
        requests_list: [
            {"method": "GET", "path": "/users"},
            {"method": "POST", "path": "/users", "json": {"name": "test"}},
        ]
        """
        results = []
        for req in requests_list:
            method = req.get("method", "GET").upper()
            path = req.get("path")
            
            if method == "GET":
                result = self.get(path, **{k: v for k, v in req.items() if k != "method" and k != "path"})
            elif method == "POST":
                result = self.post(path, **{k: v for k, v in req.items() if k != "method" and k != "path"})
            elif method == "PUT":
                result = self.put(path, **{k: v for k, v in req.items() if k != "method" and k != "path"})
            elif method == "DELETE":
                result = self.delete(path, **{k: v for k, v in req.items() if k != "method" and k != "path"})
            else:
                result = APIResponse(success=False, status_code=0, error=f"不支持的方法: {method}")
            
            results.append(result)
            time.sleep(0.1)  # 避免请求过于频繁
        
        return results


# 使用示例
if __name__ == "__main__":
    # 创建客户端
    client = APIClient(
        base_url="https://jsonplaceholder.typicode.com",
        timeout=10
    )
    
    # 测试GET请求
    print("=== GET请求 ===")
    response = client.get("/posts/1")
    print(f"成功: {response.success}")
    print(f"状态码: {response.status_code}")
    print(f"数据: {json.dumps(response.data, indent=2, ensure_ascii=False)}")
    
    # 测试POST请求
    print("\n=== POST请求 ===")
    response = client.post("/posts", json={
        "title": "foo",
        "body": "bar",
        "userId": 1
    })
    print(f"成功: {response.success}")
    print(f"状态码: {response.status_code}")
    print(f"数据: {json.dumps(response.data, indent=2, ensure_ascii=False)}")
    
    # 测试批量请求
    print("\n=== 批量请求 ===")
    batch_requests = [
        {"method": "GET", "path": "/posts/1"},
        {"method": "GET", "path": "/posts/2"},
        {"method": "GET", "path": "/posts/3"},
    ]
    results = client.batch_request(batch_requests)
    for i, result in enumerate(results, 1):
        print(f"请求{i}: 成功={result.success}, 状态码={result.status_code}")

总结

好了！咱们今天一起走过了HTTP协议的完整旅程：

1. 理解了HTTP的基本概念 —— 客户端和服务器如何通信
2. 学习了请求和响应的结构 —— 请求行、请求头、请求体
3. 掌握了各种HTTP方法 —— GET、POST、PUT、DELETE等
4. 理解了状态码的含义 —— 2xx成功、3xx重定向、4xx客户端错误、5xx服务器错误
5. 学会了处理Cookie和Session —— 保持登录状态
6. 掌握了HTML和JSON解析 —— 处理网页数据
7. 实现了带重试的错误处理 —— 让代码更健壮
8. 构建了一个完整的API客户端 —— 实战应用

HTTP协议其实是互联网最基础也最重要的协议之一。当你理解了它，你就理解了互联网通信的本质。从浏览网页到调用API，从爬虫到微服务，到处都可见HTTP的身影。

记住，理论很重要，但动手实践更重要。多写代码，多调试，多查看响应头和请求头，你会越来越熟练的！

有什么问题随时问我，咱们一起进步！加油！