一、什么是 urllib?
urllib 是 Python 内置的 HTTP 客户端库,用于:
- 发送 HTTP/HTTPS 请求(GET、POST 等)
1.1 生活比喻
把 urllib 想象成一个快递员:
1.2 urllib 的四个子模块
| | |
|---|
urllib.request | | |
urllib.parse | | |
urllib.error | | |
urllib.robotparser | | |
1.3 导入方式
# urllib 不是直接 import urllib 就能用的# 需要导入具体的子模块import urllib.request # 发送请求import urllib.parse # URL 处理import urllib.error # 错误处理import urllib.robotparser # robots.txt
二、urllib.request —— 发送请求(核心)
2.1 最简单的 GET 请求
import urllib.request# ============ 最简写法:一行代码获取网页 ============response = urllib.request.urlopen("https://www.python.org")# response 是一个 HTTPResponse 对象print(f"状态码:{response.status}") # 200print(f"编码:{response.headers.get_content_charset()}") # utf-8print(f"URL:{response.url}") # 最终URL(可能经过重定向)# 读取响应内容(返回 bytes)html_bytes = response.read()print(f"内容长度:{len(html_bytes)} 字节")# 转为字符串html = html_bytes.decode("utf-8")print(html[:200]) # 打印前200个字符# 关闭连接response.close()
📌 注意:urlopen() 返回的是字节流(bytes),需要 .decode("utf-8") 转为字符串。
2.2 使用 with 语句(推荐)
import urllib.request# ✅ 推荐:用 with 自动关闭连接with urllib.request.urlopen("https://www.python.org") as response: html = response.read().decode("utf-8") print(f"获取到 {len(html)} 个字符") print(f"状态码:{response.status}")# with 结束后,连接自动关闭,不用手动 close()
2.3 设置超时
import urllib.request# timeout 参数:超时秒数(防止无限等待)try: with urllib.request.urlopen("https://www.python.org", timeout=10) as response: html = response.read().decode("utf-8") print(f"✅ 成功,{len(html)} 字符")except Exception as e: print(f"❌ 请求失败:{e}")# 如果服务器10秒内没响应,会抛出 socket.timeout 异常
2.4 读取响应的方式
import urllib.requestwith urllib.request.urlopen("https://www.python.org") as response: # 方式1:一次性读取全部 # data = response.read() # 方式2:读取指定字节数 # first_100 = response.read(100) # 方式3:逐行读取(适合大文件) # for line in response: # print(line) # 方式4:readline() 读一行 # first_line = response.readline() # 方式5:readlines() 读所有行(返回列表) # all_lines = response.readlines() # 实际使用 html = response.read().decode("utf-8") print(html[:500])
2.5 查看响应头信息
import urllib.requestwith urllib.request.urlopen("https://www.python.org") as response: # ============ 获取所有响应头 ============ print("=== 所有响应头 ===") print(response.headers) # ============ 获取单个响应头 ============ print(f"\nContent-Type: {response.headers.get('Content-Type')}") print(f"Server: {response.headers.get('Server')}") print(f"Content-Length: {response.headers.get('Content-Length')}") # ============ 其他信息 ============ print(f"\n状态码:{response.status}") print(f"原因:{response.reason}") # OK print(f"最终URL:{response.url}") print(f"编码:{response.headers.get_content_charset()}")
典型输出:
=== 所有响应头 ===Server: nginxContent-Type: text/html; charset=utf-8Content-Length: 52831...Content-Type: text/html; charset=utf-8Server: nginxContent-Length: 52831状态码:200原因:OK最终URL:https://www.python.org编码:utf-8
三、设置请求头(Headers)
3.1 为什么需要设置请求头?
很多网站会检测 User-Agent,如果发现是 Python 脚本就拒绝访问。
import urllib.request# ❌ 不设置 User-Agent,很多网站会返回 403# response = urllib.request.urlopen("https://www.baidu.com")# ✅ 方法1:通过 Request 对象设置url = "https://www.baidu.com"# 创建 Request 对象(可以自定义请求头)request = urllib.request.Request(url)# 添加请求头request.add_header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36")request.add_header("Accept", "text/html,application/xhtml+xml")request.add_header("Accept-Language", "zh-CN,zh;q=0.9")# 发送请求with urllib.request.urlopen(request) as response: html = response.read().decode("utf-8") print(f"✅ 成功获取 {len(html)} 字符")
3.2 创建 Request 对象时直接传入 headers
import urllib.requesturl = "https://httpbin.org/get"headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0", "Accept": "application/json", "X-Custom-Header": "my-value",}# 创建 Request 时传入 headers 字典request = urllib.request.Request(url, headers=headers)with urllib.request.urlopen(request) as response: data = response.read().decode("utf-8") print(data) # httpbin.org 会把你发送的请求头原样返回,方便调试
3.3 常用请求头说明
| | |
|---|
User-Agent | | Mozilla/5.0 ... |
Accept | | text/html, application/json |
Accept-Language | | zh-CN,zh;q=0.9 |
Content-Type | | application/json |
Authorization | | Bearer token123 |
Cookie | | session=abc123 |
Referer | | https://google.com |
四、POST 请求
4.1 发送表单数据(application/x-www-form-urlencoded)
import urllib.requestimport urllib.parseurl = "https://httpbin.org/post"# 要发送的表单数据form_data = { "username": "zhangsan", "password": "123456", "remember": "true"}# 第一步:将字典编码为 URL 编码格式# {"username": "zhangsan"} → "username=zhangsan&password=123456&remember=true"encoded_data = urllib.parse.urlencode(form_data)print(f"编码后:{encoded_data}")# 输出:username=zhangsan&password=123456&remember=true# 第二步:转为 bytes(POST 的 data 必须是 bytes!)data_bytes = encoded_data.encode("utf-8")# 第三步:创建 Request(传入 data 参数就变成 POST 了)request = urllib.request.Request(url, data=data_bytes)request.add_header("Content-Type", "application/x-www-form-urlencoded")# 第四步:发送with urllib.request.urlopen(request) as response: result = response.read().decode("utf-8") print(result)
📌 关键:urlopen() 不传 data = GET 请求;传了 data = POST 请求。
4.2 发送 JSON 数据
import urllib.requestimport jsonurl = "https://httpbin.org/post"# 要发送的 JSON 数据payload = { "name": "张三", "age": 25, "hobbies": ["编程", "读书"]}# 第一步:字典 → JSON 字符串 → bytesjson_string = json.dumps(payload, ensure_ascii=False)data_bytes = json_string.encode("utf-8")# 第二步:创建 Requestrequest = urllib.request.Request(url, data=data_bytes)request.add_header("Content-Type", "application/json; charset=utf-8")request.add_header("User-Agent", "Python-Script/1.0")# 第三步:发送with urllib.request.urlopen(request) as response: result = json.loads(response.read().decode("utf-8")) print(f"服务器收到:{result['json']}") # 输出:服务器收到:{'name': '张三', 'age': 25, 'hobbies': ['编程', '读书']}
4.3 GET 请求带参数
import urllib.requestimport urllib.parse# GET 请求的参数放在 URL 里(?key=value&key2=value2)base_url = "https://httpbin.org/get"params = { "q": "python tutorial", "page": 1, "lang": "zh-CN"}# 将参数编码并拼接到 URLquery_string = urllib.parse.urlencode(params)full_url = f"{base_url}?{query_string}"print(f"完整URL:{full_url}")# https://httpbin.org/get?q=python+tutorial&page=1&lang=zh-CN# 发送请求with urllib.request.urlopen(full_url) as response: result = response.read().decode("utf-8") print(result)
五、urllib.parse —— URL 处理
5.1 urlparse() —— 解析 URL
from urllib.parse import urlparseurl = "https://user:pass@www.example.com:8080/path/to/page?q=hello&lang=zh#section1"parsed = urlparse(url)print(f"协议(scheme):{parsed.scheme}") # httpsprint(f"网络位置(netloc):{parsed.netloc}") # user:pass@www.example.com:8080print(f"路径(path):{parsed.path}") # /path/to/pageprint(f"参数(params):{parsed.params}") # (空)print(f"查询(query):{parsed.query}") # q=hello&lang=zhprint(f"片段(fragment):{parsed.fragment}") # section1# 还可以进一步获取:print(f"用户名:{parsed.username}") # userprint(f"密码:{parsed.password}") # passprint(f"主机名:{parsed.hostname}") # www.example.comprint(f"端口:{parsed.port}") # 8080
URL 的结构图:
https://user:pass@www.example.com:8080/path/to/page?q=hello&lang=zh#section1├───┤ ├──────┤ ├──────────────┤├──┤├──────────┤├──────────────┤├──────┤scheme 用户名 主机名 端口 路径 查询参数 片段 密码
5.2 urlunparse() —— 拼接 URL
from urllib.parse import urlunparse# urlunparse 接收一个6元素的元组/列表parts = ("https", "www.example.com", "/search", "", "q=python&page=1", "")url = urlunparse(parts)print(url)# https://www.example.com/search?q=python&page=1# 6个元素分别是:(scheme, netloc, path, params, query, fragment)
5.3 urljoin() —— 拼接相对路径
from urllib.parse import urljoin# 将相对路径拼接为完整 URL(非常实用!)base = "https://www.example.com/docs/tutorial/"print(urljoin(base, "page1.html"))# https://www.example.com/docs/tutorial/page1.htmlprint(urljoin(base, "../index.html"))# https://www.example.com/docs/index.htmlprint(urljoin(base, "/about"))# https://www.example.com/about(绝对路径,忽略base的路径部分)print(urljoin(base, "https://other.com/page"))# https://other.com/page(完整URL,直接返回)# 实际应用:爬取页面中的相对链接page_url = "https://example.com/blog/post1"links_in_page = ["/about", "comments", "../archive", "https://cdn.example.com/img.png"]for link in links_in_page: full_link = urljoin(page_url, link) print(f" {link:30s} → {full_link}")
输出:
/about → https://example.com/about comments → https://example.com/blog/comments ../archive → https://example.com/archive https://cdn.example.com/img.png → https://cdn.example.com/img.png
5.4 urlencode() —— 字典转查询字符串
from urllib.parse import urlencode# 字典 → URL 编码的查询字符串params = { "q": "python 教程", "page": 1, "sort": "relevance"}encoded = urlencode(params)print(encoded)# q=python+%E6%95%99%E7%A8%8B&page=1&sort=relevance# 中文被编码为 %XX 格式(URL编码/百分号编码)# 空格变成 + 号# 列表值(多个同名参数)params2 = {"color": ["red", "blue", "green"]}print(urlencode(params2, doseq=True))# color=red&color=blue&color=green
5.5 parse_qs() 和 parse_qsl() —— 解析查询字符串
from urllib.parse import parse_qs, parse_qslquery = "name=%E5%BC%A0%E4%B8%89&age=25&hobby=%E7%BC%96%E7%A8%8B&hobby=%E8%AF%BB%E4%B9%A6"# parse_qs():返回字典(值为列表)result = parse_qs(query)print(result)# {'name': ['张三'], 'age': ['25'], 'hobby': ['编程', '读书']}# 注意:值都是列表!即使只有一个值print(result["name"][0]) # 张三# parse_qsl():返回键值对列表result2 = parse_qsl(query)print(result2)# [('name', '张三'), ('age', '25'), ('hobby', '编程'), ('hobby', '读书')]
5.6 quote() 和 unquote() —— URL 编码/解码
from urllib.parse import quote, unquote, quote_plus, unquote_plus# ============ quote:编码(特殊字符 → %XX) ============print(quote("你好世界")) # %E4%BD%A0%E5%A5%BD%E4%B8%96%E7%95%8Cprint(quote("hello world")) # hello%20world(空格→%20)print(quote("a/b/c")) # a%2Fb%2Fc(/也被编码了)print(quote("a/b/c", safe="/")) # a/b/c(safe参数:不编码/)# ============ unquote:解码(%XX → 原字符) ============print(unquote("%E4%BD%A0%E5%A5%BD")) # 你好print(unquote("hello%20world")) # hello world# ============ quote_plus:空格编码为+(用于表单) ============print(quote_plus("hello world")) # hello+worldprint(unquote_plus("hello+world")) # hello world# 实际应用:构造搜索URLkeyword = "Python 入门教程"encoded_keyword = quote(keyword)search_url = f"https://www.google.com/search?q={encoded_keyword}"print(search_url)# https://www.google.com/search?q=Python%20%E5%85%A5%E9%97%A8%E6%95%99%E7%A8%8B
5.7 urlsplit() 和 urldefrag()
from urllib.parse import urlsplit, urldefrag# urlsplit:类似 urlparse,但不分离 paramsurl = "https://example.com/path;params?query=1#frag"print(urlsplit(url))# SplitResult(scheme='https', netloc='example.com', path='/path;params', query='query=1', fragment='frag')# urldefrag:去掉片段(#后面的部分)url_with_frag = "https://example.com/page#section1"url_no_frag, fragment = urldefrag(url_with_frag)print(f"URL:{url_no_frag}") # https://example.com/pageprint(f"片段:{fragment}") # section1
六、urllib.error —— 错误处理
6.1 错误类型
urllib.error.URLError ← 所有网络错误的基类├── urllib.error.HTTPError ← HTTP 错误(404、500等)└── socket.timeout ← 超时
6.2 完整的错误处理模板
import urllib.requestimport urllib.errorimport socketdef fetch_url(url, timeout=10): """ 安全地获取网页内容(完整错误处理) """ try: request = urllib.request.Request(url) request.add_header("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0") with urllib.request.urlopen(request, timeout=timeout) as response: charset = response.headers.get_content_charset() or "utf-8" html = response.read().decode(charset) return html except urllib.error.HTTPError as e: # HTTP 错误(服务器返回了错误状态码) print(f"❌ HTTP 错误:{e.code}{e.reason}") print(f" URL:{e.url}") # 可以读取错误页面的内容 error_body = e.read().decode("utf-8", errors="ignore") print(f" 错误页面内容(前200字符):{error_body[:200]}") return None except urllib.error.URLError as e: # 网络错误(DNS解析失败、连接被拒绝等) print(f"❌ 网络错误:{e.reason}") return None except socket.timeout: # 超时 print(f"❌ 请求超时({timeout}秒)") return None except Exception as e: # 其他未知错误 print(f"❌ 未知错误:{type(e).__name__}: {e}") return None# ============ 测试 ============# 正常请求html = fetch_url("https://www.python.org")if html: print(f"✅ 成功获取 {len(html)} 字符\n")# 404 错误html = fetch_url("https://www.python.org/nonexistent-page-12345")# 网络错误(域名不存在)html = fetch_url("https://this-domain-does-not-exist-12345.com")# 超时(用一个很慢的地址测试)# html = fetch_url("https://httpbin.org/delay/30", timeout=3)
输出:
✅ 成功获取 52831 字符❌ HTTP 错误:404 Not Found URL:https://www.python.org/nonexistent-page-12345 错误页面内容(前200字符):<!DOCTYPE html>...❌ 网络错误:[Errno -2] Name or service not known
6.3 HTTPError 也是响应对象
import urllib.requestimport urllib.errortry: urllib.request.urlopen("https://httpbin.org/status/403")except urllib.error.HTTPError as e: # HTTPError 本身也是一个响应对象! print(f"状态码:{e.code}") # 403 print(f"原因:{e.reason}") # Forbidden print(f"响应头:{e.headers}") print(f"响应体:{e.read().decode()}") # 可以读取错误页面内容
七、下载文件
7.1 下载小文件(一次性读入内存)
import urllib.requesturl = "https://www.python.org/static/img/python-logo.png"filename = "python-logo.png"# 方法1:urlopen + readwith urllib.request.urlopen(url) as response: data = response.read() with open(filename, "wb") as f: # 注意:二进制模式 "wb" f.write(data)print(f"✅ 下载完成:{filename}({len(data)} 字节)")
7.2 下载大文件(分块读取,节省内存)
import urllib.requestdef download_file(url, filename, chunk_size=8192): """分块下载文件(适合大文件)""" request = urllib.request.Request(url) request.add_header("User-Agent", "Mozilla/5.0") with urllib.request.urlopen(request) as response: # 获取文件大小(如果服务器提供了) total_size = response.headers.get("Content-Length") total_size = int(total_size) if total_size else None downloaded = 0 with open(filename, "wb") as f: while True: chunk = response.read(chunk_size) # 每次读 8KB if not chunk: break f.write(chunk) downloaded += len(chunk) # 显示进度 if total_size: percent = downloaded / total_size * 100 print(f"\r 下载中:{percent:.1f}% ({downloaded}/{total_size})", end="", flush=True) else: print(f"\r 已下载:{downloaded} 字节", end="", flush=True) print(f"\n✅ 下载完成:{filename}")# 使用download_file("https://www.python.org/static/img/python-logo.png", "logo.png")
7.3 urlretrieve() —— 最简单的下载方式
import urllib.request# urlretrieve(url, filename) 一行搞定!filename, headers = urllib.request.urlretrieve( "https://www.python.org/static/img/python-logo.png", "python-logo.png")print(f"保存到:{filename}")print(f"响应头:{headers}")# ⚠️ 注意:urlretrieve 会创建临时文件,用完后建议清理import os# urllib.request.urlcleanup() # 清理临时文件
7.4 带进度条的下载(使用 reporthook)
import urllib.requestimport sysdef progress_hook(block_num, block_size, total_size): """ 下载进度回调函数 block_num: 已下载的块数 block_size: 每块大小 total_size: 文件总大小(-1表示未知) """ downloaded = block_num * block_size if total_size > 0: percent = min(downloaded / total_size * 100, 100) bar = "█" * int(percent / 2) + "░" * (50 - int(percent / 2)) sys.stdout.write(f"\r [{bar}] {percent:.1f}% ({downloaded}/{total_size})") else: sys.stdout.write(f"\r 已下载:{downloaded} 字节") sys.stdout.flush()# 使用print("开始下载...")urllib.request.urlretrieve( "https://www.python.org/static/img/python-logo.png", "logo.png", reporthook=progress_hook)print("\n✅ 完成!")
八、Cookie 处理
8.1 使用 CookieJar 自动管理 Cookie
import urllib.requestimport http.cookiejar# ============ 第一步:创建 Cookie 管理器 ============cookie_jar = http.cookiejar.CookieJar()# ============ 第二步:创建带 Cookie 处理的 opener ============cookie_handler = urllib.request.HTTPCookieProcessor(cookie_jar)opener = urllib.request.build_opener(cookie_handler)# ============ 第三步:用 opener 发送请求 ============# 第一次请求:服务器会设置 Cookie(Set-Cookie 头)response = opener.open("https://httpbin.org/cookies/set/session_id/abc123")print(response.read().decode())# 查看保存的 Cookieprint("\n=== 保存的 Cookies ===")for cookie in cookie_jar: print(f" {cookie.name} = {cookie.value} (域:{cookie.domain})")# 第二次请求:会自动带上之前保存的 Cookieresponse = opener.open("https://httpbin.org/cookies")print(f"\n服务器看到的Cookie:{response.read().decode()}")
8.2 保存 Cookie 到文件
import urllib.requestimport http.cookiejar# 使用 MozillaCookieJar 保存到文件(Netscape 格式)cookie_file = "cookies.txt"cookie_jar = http.cookiejar.MozillaCookieJar(cookie_file)# 如果文件已存在,加载之前的 Cookietry: cookie_jar.load(ignore_discard=True, ignore_expires=True) print("✅ 已加载保存的 Cookie")except FileNotFoundError: print("📝 首次运行,将创建新的 Cookie 文件")# 创建 openerhandler = urllib.request.HTTPCookieProcessor(cookie_jar)opener = urllib.request.build_opener(handler)# 发送请求(会自动使用/更新 Cookie)response = opener.open("https://httpbin.org/cookies/set/user/zhangsan")print(response.read().decode())# 保存 Cookie 到文件cookie_jar.save(ignore_discard=True, ignore_expires=True)print(f"✅ Cookie 已保存到 {cookie_file}")
8.3 手动设置 Cookie
import urllib.requesturl = "https://httpbin.org/cookies"request = urllib.request.Request(url)# 手动添加 Cookie 头request.add_header("Cookie", "session=abc123; user=zhangsan; lang=zh-CN")with urllib.request.urlopen(request) as response: print(response.read().decode()) # 服务器会看到你发送的 Cookie
九、代理设置
9.1 使用代理
import urllib.request# ============ 方法1:通过 ProxyHandler ============proxy_handler = urllib.request.ProxyHandler({ "http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890",})opener = urllib.request.build_opener(proxy_handler)response = opener.open("https://httpbin.org/ip")print(response.read().decode())# 会显示代理服务器的 IP# ============ 方法2:通过环境变量 ============import osos.environ["http_proxy"] = "http://127.0.0.1:7890"os.environ["https_proxy"] = "http://127.0.0.1:7890"# 之后 urlopen 会自动使用代理# ============ 方法3:不使用代理(绕过系统代理) ============no_proxy_handler = urllib.request.ProxyHandler({})opener = urllib.request.build_opener(no_proxy_handler)response = opener.open("https://httpbin.org/ip")
9.2 带认证的代理
import urllib.request# 代理需要用户名密码proxy_handler = urllib.request.ProxyHandler({ "http": "http://user:password@proxy.example.com:8080", "https": "http://user:password@proxy.example.com:8080",})opener = urllib.request.build_opener(proxy_handler)response = opener.open("https://httpbin.org/ip")
十、SSL/HTTPS 处理
10.1 忽略 SSL 证书验证(仅用于测试!)
import urllib.requestimport ssl# ⚠️ 仅用于开发测试,生产环境不要这样做!# 创建不验证证书的 SSL 上下文context = ssl.create_default_context()context.check_hostname = Falsecontext.verify_mode = ssl.CERT_NONE# 使用自定义 SSL 上下文response = urllib.request.urlopen("https://self-signed.example.com", context=context)
10.2 使用自定义 CA 证书
import urllib.requestimport ssl# 指定 CA 证书文件context = ssl.create_default_context(cafile="/path/to/ca-bundle.crt")response = urllib.request.urlopen("https://example.com", context=context)
十一、urllib.robotparser —— robots.txt 解析
11.1 什么是 robots.txt?
网站的 robots.txt 文件告诉爬虫哪些页面可以访问,哪些不可以。
# https://www.example.com/robots.txtUser-agent: *Disallow: /admin/Disallow: /private/Allow: /public/User-agent: BadBotDisallow: /
11.2 使用 robotparser
import urllib.robotparser# 创建解析器rp = urllib.robotparser.RobotFileParser()# 设置 robots.txt 的 URLrp.set_url("https://www.python.org/robots.txt")# 读取并解析rp.read()# 检查某个 User-Agent 是否可以访问某个 URLprint(rp.can_fetch("*", "https://www.python.org/about/"))# True(允许)print(rp.can_fetch("*", "https://www.python.org/admin/"))# 取决于 robots.txt 的规则# 获取 Crawl-delay(爬取间隔)delay = rp.crawl_delay("*")print(f"Crawl-delay: {delay}")# 手动设置 robots.txt 内容(不从网络读取)rp2 = urllib.robotparser.RobotFileParser()rp2.parse([ "User-agent: *", "Disallow: /secret/", "Allow: /secret/public/", "Crawl-delay: 10"])print(rp2.can_fetch("MyBot", "https://example.com/secret/page")) # Falseprint(rp2.can_fetch("MyBot", "https://example.com/secret/public/")) # Trueprint(rp2.can_fetch("MyBot", "https://example.com/normal/page")) # True
十二、构建自定义 Opener
12.1 什么是 Opener?
urlopen() 是简化接口。如果需要更复杂的配置(Cookie、代理、重定向等),需要构建 Opener。
import urllib.requestimport http.cookiejar# ============ 构建一个功能完整的 opener ============# 1. Cookie 处理cookie_jar = http.cookiejar.CookieJar()cookie_handler = urllib.request.HTTPCookieProcessor(cookie_jar)# 2. 代理(可选)# proxy_handler = urllib.request.ProxyHandler({"http": "http://127.0.0.1:7890"})# 3. 构建 openeropener = urllib.request.build_opener( cookie_handler, # proxy_handler, # 如果需要代理)# 4. 设置默认请求头opener.addheaders = [ ("User-Agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/120.0.0.0"), ("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"),]# 5. 使用 opener 发送请求response = opener.open("https://httpbin.org/get")print(response.read().decode())# 6.(可选)设为全局默认,之后 urlopen 也会使用这个 openerurllib.request.install_opener(opener)# 之后可以直接用 urllib.request.urlopen(),自动带 Cookie 和自定义头
12.2 常用 Handler 一览
| |
|---|
HTTPCookieProcessor | |
ProxyHandler | |
HTTPBasicAuthHandler | |
HTTPDigestAuthHandler | |
HTTPSHandler | |
HTTPRedirectHandler | |
12.3 HTTP 基本认证
import urllib.request# 访问需要用户名密码的页面password_mgr = urllib.request.HTTPPasswordMgrWithDefaultRealm()password_mgr.add_password( realm=None, # realm(通常填 None) uri="https://httpbin.org/basic-auth/user/pass", user="user", passwd="pass")auth_handler = urllib.request.HTTPBasicAuthHandler(password_mgr)opener = urllib.request.build_opener(auth_handler)response = opener.open("https://httpbin.org/basic-auth/user/pass")print(response.read().decode())# {"authenticated": true, "user": "user"}
十三、处理重定向
13.1 自动重定向(默认行为)
import urllib.request# urllib 默认自动跟随重定向(最多10次)# httpbin.org/redirect/3 会重定向3次with urllib.request.urlopen("https://httpbin.org/redirect/3") as response: print(f"最终URL:{response.url}") # https://httpbin.org/get(重定向3次后的最终地址) print(f"状态码:{response.status}") # 200
13.2 禁止自动重定向
import urllib.requestclass NoRedirectHandler(urllib.request.HTTPRedirectHandler): """禁止自动重定向""" def redirect_request(self, req, fp, code, msg, headers, newurl): return None # 返回 None 表示不跟随重定向opener = urllib.request.build_opener(NoRedirectHandler)try: response = opener.open("https://httpbin.org/redirect/1")except urllib.error.HTTPError as e: print(f"状态码:{e.code}") # 302 print(f"重定向到:{e.headers.get('Location')}") # 目标URL
十四、实战案例
14.1 简易网页爬虫
import urllib.requestimport urllib.parseimport urllib.errorimport reimport timeclass SimpleCrawler: """简易网页爬虫""" def __init__(self): self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 Chrome/120.0.0.0 Safari/537.36", "Accept": "text/html,application/xhtml+xml", "Accept-Language": "zh-CN,zh;q=0.9", } def fetch(self, url, timeout=10): """获取网页内容""" try: request = urllib.request.Request(url, headers=self.headers) with urllib.request.urlopen(request, timeout=timeout) as response: charset = response.headers.get_content_charset() or "utf-8" return response.read().decode(charset, errors="ignore") except urllib.error.HTTPError as e: print(f" ❌ HTTP {e.code}: {url}") return None except urllib.error.URLError as e: print(f" ❌ 网络错误: {e.reason}") return None except Exception as e: print(f" ❌ 错误: {e}") return None def extract_links(self, html, base_url): """提取页面中的所有链接""" # 简单的正则提取 href pattern = r'href=["\']([^"\']+)["\']' links = re.findall(pattern, html) # 转为绝对路径 absolute_links = [] for link in links: full_url = urllib.parse.urljoin(base_url, link) # 只保留 http/https 链接 if full_url.startswith(("http://", "https://")): absolute_links.append(full_url) return list(set(absolute_links)) # 去重 def extract_title(self, html): """提取页面标题""" match = re.search(r"<title>(.*?)</title>", html, re.IGNORECASE | re.DOTALL) return match.group(1).strip() if match else "(无标题)" def crawl(self, start_url, max_pages=5): """从起始URL开始爬取""" visited = set() to_visit = [start_url] print(f"🕷️ 开始爬取:{start_url}") print(f" 最大页数:{max_pages}") print("=" * 50) while to_visit and len(visited) < max_pages: url = to_visit.pop(0) if url in visited: continue print(f"\n📄 [{len(visited)+1}/{max_pages}] {url}") html = self.fetch(url) if html is None: continue visited.add(url) # 提取标题 title = self.extract_title(html) print(f" 标题:{title}") print(f" 大小:{len(html)} 字符") # 提取链接 links = self.extract_links(html, url) new_links = [l for l in links if l not in visited] print(f" 发现 {len(links)} 个链接({len(new_links)} 个新链接)") # 将新链接加入待爬队列(只爬同域的) start_domain = urllib.parse.urlparse(start_url).netloc for link in new_links[:10]: # 每页最多取10个 link_domain = urllib.parse.urlparse(link).netloc if link_domain == start_domain: to_visit.append(link) time.sleep(1) # 礼貌等待 print("\n" + "=" * 50) print(f"✅ 爬取完成!共访问 {len(visited)} 个页面")# 使用crawler = SimpleCrawler()crawler.crawl("https://www.python.org", max_pages=3)
14.2 调用 REST API
import urllib.requestimport urllib.parseimport urllib.errorimport jsonclass APIClient: """简易 REST API 客户端""" def __init__(self, base_url, headers=None): self.base_url = base_url.rstrip("/") self.default_headers = { "User-Agent": "Python-APIClient/1.0", "Accept": "application/json", } if headers: self.default_headers.update(headers) def _request(self, method, path, data=None, params=None): """发送请求""" url = f"{self.base_url}/{path.lstrip('/')}" # 添加查询参数 if params: query = urllib.parse.urlencode(params) url = f"{url}?{query}" # 准备请求体 body = None if data is not None: body = json.dumps(data, ensure_ascii=False).encode("utf-8") # 创建请求 request = urllib.request.Request(url, data=body, method=method) for key, value in self.default_headers.items(): request.add_header(key, value) if body: request.add_header("Content-Type", "application/json") # 发送 try: with urllib.request.urlopen(request, timeout=15) as response: result = json.loads(response.read().decode("utf-8")) return {"status": response.status, "data": result} except urllib.error.HTTPError as e: error_body = e.read().decode("utf-8", errors="ignore") return {"status": e.code, "error": e.reason, "body": error_body} except Exception as e: return {"status": 0, "error": str(e)} def get(self, path, params=None): return self._request("GET", path, params=params) def post(self, path, data=None): return self._request("POST", path, data=data) def put(self, path, data=None): return self._request("PUT", path, data=data) def delete(self, path): return self._request("DELETE", path)# ============ 使用示例 ============# 使用 JSONPlaceholder(免费的测试 API)api = APIClient("https://jsonplaceholder.typicode.com")# GET:获取所有帖子(带参数)print("=== 获取帖子 ===")result = api.get("/posts", params={"_limit": 3})if result["status"] == 200: for post in result["data"]: print(f" [{post['id']}] {post['title'][:40]}...")# GET:获取单个帖子print("\n=== 获取帖子 #1 ===")result = api.get("/posts/1")if result["status"] == 200: print(f" 标题:{result['data']['title']}") print(f" 内容:{result['data']['body'][:80]}...")# POST:创建帖子print("\n=== 创建帖子 ===")new_post = { "title": "我的新文章", "body": "这是文章内容...", "userId": 1}result = api.post("/posts", data=new_post)if result["status"] == 201: print(f" ✅ 创建成功!ID = {result['data']['id']}")# GET:获取用户print("\n=== 获取用户 ===")result = api.get("/users/1")if result["status"] == 200: user = result["data"] print(f" 姓名:{user['name']}") print(f" 邮箱:{user['email']}") print(f" 城市:{user['address']['city']}")
14.3 图片批量下载器
import urllib.requestimport urllib.parseimport osimport timeimport redef download_images(page_url, save_dir="images", max_images=10): """从网页中下载所有图片""" # 创建保存目录 os.makedirs(save_dir, exist_ok=True) # 获取网页 headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"} request = urllib.request.Request(page_url, headers=headers) try: with urllib.request.urlopen(request, timeout=10) as response: html = response.read().decode("utf-8", errors="ignore") except Exception as e: print(f"❌ 获取页面失败:{e}") return # 提取图片 URL img_pattern = r'<img[^>]+src=["\']([^"\']+)["\']' img_urls = re.findall(img_pattern, html, re.IGNORECASE) # 过滤和转为绝对路径 valid_urls = [] for url in img_urls: full_url = urllib.parse.urljoin(page_url, url) if full_url.startswith(("http://", "https://")): # 只下载常见图片格式 if any(ext in full_url.lower() for ext in [".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg"]): valid_urls.append(full_url) valid_urls = list(dict.fromkeys(valid_urls))[:max_images] # 去重 + 限制数量 print(f"📷 找到 {len(valid_urls)} 张图片,开始下载...\n") # 下载 success = 0 for i, img_url in enumerate(valid_urls, 1): # 生成文件名 parsed = urllib.parse.urlparse(img_url) filename = os.path.basename(parsed.path) or f"image_{i}.jpg" filepath = os.path.join(save_dir, filename) print(f" [{i}/{len(valid_urls)}] {filename}...", end=" ") try: img_request = urllib.request.Request(img_url, headers=headers) with urllib.request.urlopen(img_request, timeout=10) as response: data = response.read() with open(filepath, "wb") as f: f.write(data) print(f"✅ ({len(data)} bytes)") success += 1 except Exception as e: print(f"❌ ({e})") time.sleep(0.5) # 间隔 print(f"\n✅ 完成!成功下载 {success}/{len(valid_urls)} 张图片") print(f" 保存目录:{os.path.abspath(save_dir)}")# 使用# download_images("https://www.python.org", save_dir="python_images", max_images=5)
14.4 网页编码自动检测
import urllib.requestdef smart_fetch(url): """智能获取网页(自动处理编码)""" headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"} request = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(request, timeout=10) as response: raw_data = response.read() # 方法1:从响应头获取编码 charset = response.headers.get_content_charset() # 方法2:如果响应头没有,从 HTML meta 标签获取 if not charset: # 检查 <meta charset="utf-8"> 或 <meta http-equiv="Content-Type" content="text/html; charset=gbk"> import re meta_match = re.search( rb'charset=["\']?([a-zA-Z0-9_-]+)', raw_data[:1000], # 只搜索前1000字节 re.IGNORECASE ) if meta_match: charset = meta_match.group(1).decode("ascii") # 方法3:默认 UTF-8 if not charset: charset = "utf-8" print(f" 检测到编码:{charset}") # 解码 try: return raw_data.decode(charset) except (UnicodeDecodeError, LookupError): # 如果检测的编码不对,尝试常见编码 for enc in ["utf-8", "gbk", "gb2312", "latin-1"]: try: return raw_data.decode(enc) except UnicodeDecodeError: continue # 最后手段:忽略错误 return raw_data.decode("utf-8", errors="ignore")# 测试html = smart_fetch("https://www.baidu.com")print(f"获取到 {len(html)} 字符")print(html[:200])
十五、urllib vs requests 对比
| | |
|---|
| | pip install requests |
| | |
| | |
| | json=payload |
| | |
| timeout | timeout |
| | proxies |
| | |
| | |
# urllib 写法(获取JSON API)import urllib.requestimport jsonrequest = urllib.request.Request("https://api.github.com/users/octocat")request.add_header("User-Agent", "Python-Script")with urllib.request.urlopen(request) as response: data = json.loads(response.read().decode("utf-8")) print(data["login"])# requests 写法(同样的功能)import requestsdata = requests.get("https://api.github.com/users/octocat").json()print(data["login"])
📌 建议:学习阶段用 urllib 理解底层原理;实际项目用 requests 更高效。
十六、常见问题 FAQ
Q1:中文 URL 怎么处理?
import urllib.parse# 中文 URL 需要编码keyword = "Python 教程"encoded = urllib.parse.quote(keyword)url = f"https://www.baidu.com/s?wd={encoded}"print(url)# https://www.baidu.com/s?wd=Python%20%E6%95%99%E7%A8%8B
Q2:如何处理 gzip 压缩的响应?
import urllib.requestimport gzipimport iorequest = urllib.request.Request("https://www.python.org")request.add_header("Accept-Encoding", "gzip") # 告诉服务器我们可以接受gzipwith urllib.request.urlopen(request) as response: data = response.read() # 检查是否是 gzip 压缩 if response.headers.get("Content-Encoding") == "gzip": data = gzip.decompress(data) html = data.decode("utf-8") print(f"解压后:{len(html)} 字符")
Q3:如何限制下载速度?
import urllib.requestimport timedef slow_download(url, filename, speed_limit=10240): """限速下载(speed_limit: 每秒最大字节数)""" request = urllib.request.Request(url) request.add_header("User-Agent", "Mozilla/5.0") with urllib.request.urlopen(request) as response: with open(filename, "wb") as f: start_time = time.time() total = 0 while True: chunk = response.read(1024) # 每次1KB if not chunk: break f.write(chunk) total += len(chunk) # 限速:计算应该等待的时间 elapsed = time.time() - start_time expected_time = total / speed_limit if expected_time > elapsed: time.sleep(expected_time - elapsed) print(f"✅ 下载完成:{total} 字节")
Q4:urlopen 和 Request 的区别?
import urllib.request# urlopen(url):最简单,不能自定义请求头response = urllib.request.urlopen("https://example.com")# Request + urlopen:可以自定义请求头、方法等request = urllib.request.Request("https://example.com")request.add_header("User-Agent", "MyBot/1.0")request.method = "GET" # 可以改为 POST、PUT 等response = urllib.request.urlopen(request)
十七、速查表
═══════════════════════════════════════════════════════ urllib.request(发送请求)═══════════════════════════════════════════════════════GET 请求: response = urllib.request.urlopen(url, timeout=10) html = response.read().decode("utf-8")POST 请求: data = urllib.parse.urlencode(form).encode("utf-8") request = urllib.request.Request(url, data=data) response = urllib.request.urlopen(request)自定义请求头: request = urllib.request.Request(url, headers={...}) 或 request.add_header("User-Agent", "...")构建 Opener: opener = urllib.request.build_opener(handler1, handler2) response = opener.open(url)下载文件: urllib.request.urlretrieve(url, filename)═══════════════════════════════════════════════════════ urllib.parse(URL处理)═══════════════════════════════════════════════════════解析URL: urlparse(url) → scheme, netloc, path, query, fragment拼接URL: urljoin(base, relative)编码参数: urlencode({"key": "value"})解析参数: parse_qs("key=value&k2=v2")URL编码: quote("中文") → "%E4%B8%AD%E6%96%87"URL解码: unquote("%E4%B8%AD%E6%96%87") → "中文"═══════════════════════════════════════════════════════ urllib.error(错误处理)═══════════════════════════════════════════════════════HTTPError: 服务器返回错误状态码(404、500等)URLError: 网络层错误(DNS失败、连接拒绝等)socket.timeout:超时═══════════════════════════════════════════════════════ http.cookiejar(Cookie管理)═══════════════════════════════════════════════════════CookieJar(): 内存中管理CookieMozillaCookieJar(file): 保存到文件HTTPCookieProcessor(jar): 配合opener使用