简介：面向希望入门网络爬虫与电商商品信息采集的开发者，这份淘宝天猫商品详情爬取示例聚焦商品名称、价格、描述、评价等字段的获取流程，适合具备一定编程基础、想了解请求发送、页面解析与数据落盘的学习者参考。压缩包内共2个Java源码文件，整体约4KB，体量轻便，便于直接阅读代码结构与核心实现，快速把握商品详情抓取的骨架。已有2988人学习或下载，说明该主题在爬虫入门与电商数据采集练习中具有较高关注度。内容可帮助读者理解如何构造请求、解析HTML或JSON响应、提取目标字段并保存结果，同时为处理动态加载、验证码、IP限制等反爬场景提供思路参考；代码精简，也便于按自身项目需求进行二次改造与扩展练习。

1. 商品详情爬取不是抓 HTML：先想清楚登录态、字段和合规边界

商品详情爬取最容易翻车的地方，不是正则写错，而是把登录态、字段口径和合规边界放到最后才想。你拿到的这份示例工程，核心不是一页 HTML，而是一条可回放的链路：请求参数从商品 ID 和规格维度生成，会话复用登录态，限速与重试兜住偶发失败，解析后把 SKU、价格、库存落成可对账的表。它适合做竞品巡检、活动价监控、内部选品看板的从业者；不适合想绕过验证码、硬破签名的人。常见做法是先申请合规数据接口，公开页只做低频抽样，遇到风控就停。

2. 请求层怎么搭：会话复用、限速与重试把成功率稳住

请求层是整条链路的地基。很多人一上来就写解析函数，结果跑到一半发现 Cookie 过期、连接数暴涨、偶发 429 把整批任务打挂。我的习惯是把请求层拆成三块：会话与请求头、限速与重试、参数模板。会话负责复用 TCP 连接和登录态，限速负责把频率压在可控范围，参数模板负责让每个商品 ID 都按同一套语义发请求。三块都落成独立函数，后面排查时才能定位到具体一层，而不是对着日志猜。下面按可抄作业的顺序展开，代码里的域名和字段名都是占位，替换成自己的合规接口地址即可。

2.1 从环境变量读取登录态和基础请求头

登录态不要写进代码，也不要提交到仓库。常见做法是把 Cookie 和 User-Agent 放进环境变量，启动任务前由人工登录后注入；本地调试可以用.env文件，但.env必须进.gitignore。这样做的好处是同一份代码可以在测试环境和生产环境跑，换账号只换环境变量，不需要改源码。请求头里至少要带 User-Agent、Accept、Referer 和 Cookie，缺 Referer 时有些详情接口会直接返回空数据，缺 Accept 时可能拿到 HTML 而不是 JSON。

import os import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry BASE = "https://example.com/api/item" COOKIE = os.getenv("SHOP_COOKIE", "") UA = os.getenv("SHOP_UA", "Mozilla/5.0 (兼容占位，替换成真实浏览器UA)") session = requests.Session() session.headers.update({ "User-Agent": UA, "Accept": "application/json, text/plain, */*", "Referer": "https://example.com/", "Cookie": COOKIE, }) # 连接层不自动重试，重试交给外层限速器统一控制，避免叠加放大 adapter = HTTPAdapter(max_retries=Retry( total=0, connect=0, read=0, status=0, )) session.mount("https://", adapter) def fetch_item(item_id: str, timeout=(3.05, 10)) -> requests.Response: url = f"{BASE}/{item_id}" return session.get(url, timeout=timeout)

这段代码的关键点有三个。session复用连接，避免每个商品都新建 TLS 握手；timeout=(3.05, 10)是连接超时和读取超时，连接超时短一点，读取超时长一点，防止慢响应把线程池占满；max_retries=0是故意关掉连接库自带的重试，因为外层限速器还要做退避，两层叠加会让失败请求成倍放大。Cookie 为空时不要硬跑，先抛一个明确错误，比后面解析出一堆空字段更容易排查。

提示：登录态过期后有些接口仍然返回 200，只是数据体变成空对象或登录提示页，所以状态码不是唯一判断依据。

2.2 限速器与重试器：别让偶发 429 变成整批翻车

限速不是越慢越好，而是要稳定、可解释。我一般用最小间隔限速器，把 QPS 压在 0.3 到 0.5 之间，批量任务再开 2 到 4 个并发，超过这个范围就容易触发服务端保护。重试只对连接错误、超时和明确的临时性 5xx 生效，遇到 401、403、429 不要无限重试；401 和 403 先检查登录态和权限，429 先停下来看限速参数。重试次数设 3 次，退避基数设 2 秒，再加一点随机抖动，避免同一批任务在同一秒集中重试。

import time import random from functools import wraps class RateLimiter: def __init__(self, qps=0.5): self.min_interval = 1.0 / qps self.last = 0.0 def wait(self): now = time.monotonic() delta = now - self.last if delta < self.min_interval: time.sleep(self.min_interval - delta) self.last = time.monotonic() limiter = RateLimiter(qps=0.5) def retry_on(exceptions=(ConnectionError, TimeoutError), tries=3, base=2.0): def deco(fn): @wraps(fn) def wrapper(*args, **kwargs): for i in range(tries): try: limiter.wait() return fn(*args, **kwargs) except exceptions: if i == tries - 1: raise sleep_s = base ** i + random.uniform(0, 0.3) time.sleep(sleep_s) return wrapper return deco

qps=0.5表示每 2 秒最多发一个请求，适合小批量巡检；tries=3表示最多尝试三次；base=2.0让退避间隔按 1 秒、2 秒、4 秒的节奏增长，random.uniform(0, 0.3)是抖动，避免重试风暴。这里没有把 429 放进自动重试，是因为 429 往往说明频率已经越界，继续重试只会让情况更糟。正确做法是记录命中 429 的商品 ID，暂停任务，调低 QPS，再从断点继续。

2.3 请求参数模板：商品 ID、规格维度和分页语义

参数模板的价值在于把“每次请求带什么”固化下来。商品详情通常围绕商品 ID 展开，规格维度决定要不要拉 SKU 列表，分页参数只对评价、问答这类附属数据有意义。我的习惯是用 dataclass 定义查询对象，先校验再转字典，避免字符串拼接把空值和数字类型混进去。item_id必须是纯数字，page_size设一个示例上限，scene用来区分详情页、活动页和比价页，后面排查时能按场景过滤日志。

from dataclasses import dataclass, asdict from typing import Optional @dataclass class ItemQuery: item_id: str scene: str = "detail" sku_mode: str = "all" page: int = 1 page_size: int = 20 def validate(self): if not self.item_id.isdigit(): raise ValueError("item_id 应为纯数字") if not (1 <= self.page_size <= 50): raise ValueError("page_size 超过示例上限") if self.page < 1: raise ValueError("page 从 1 开始") return self q = ItemQuery(item_id="100000000001", page=1).validate() params = {k: v for k, v in asdict(q).items() if v is not None}

item_id用字符串而不是整数，是因为长 ID 在 JSON 和 SQLite 里都可能被截断；scene和sku_mode是业务语义，不要省略；page_size上限设 50 是示例约束，实际按接口文档调整。如果详情接口需要动态签名，不要在示例工程里硬抠算法，优先走官方开放接口或让业务方提供合法数据源；签名逻辑属于边界外的东西，硬破既不稳定也不合规。

3. 详情字段怎么落：JSON 解析、SKU 合并与入库幂等

请求通了之后，下一层是字段落库。商品详情的数据形态通常有三种：接口直接返回 JSON、页面里内嵌一段 JSON、纯 DOM 结构。优先选 JSON，因为字段稳定、类型明确；内嵌 JSON 次之，需要正则或解析器定位；DOM 最不稳定，页面改版就翻车。字段口径要先定：价格用分还是元，库存是单 SKU 还是聚合，规格用字典还是拼接字符串，批次日期怎么记。口径定不下来，后面做对账时全是黑匣子。

3.1 识别详情页的数据承载方式

我一般先看响应头里的 Content-Type，再看响应体前 200 个字符。如果是application/json，直接json.loads；如果是 HTML，就在页面源码里搜window.、__INITIAL_STATE__、__ITEM__这类标记。不要一上来写一个巨大的正则匹配整页，页面结构一变就全挂。更稳的做法是用 marker 定位，只取 marker 后面的一段 JSON，再交给json.loads解析；解析失败时把原始响应落盘，方便回放。

import json import re from typing import Any def extract_embedded_json(html: str, marker: str) -> dict[str, Any]: # 用 marker 定位内嵌 JSON，不要写死整页结构 pattern = re.compile(rf"{re.escape(marker)}\s*=\s*(\{{.*?\}})\s*;", re.S) m = pattern.search(html) if not m: raise ValueError(f"未找到内嵌 JSON: {marker}") return json.loads(m.group(1)) def load_detail(raw: str) -> dict: # 优先按接口 JSON 解析，失败再退到页面内嵌 try: return json.loads(raw) except json.JSONDecodeError: return extract_embedded_json(raw, "window.__ITEM__")

marker是页面里实际存在的变量名，示例用window.__ITEM__占位；re.S让点号匹配换行；.*?是非贪婪匹配，遇到嵌套很深的 JSON 可能截断，所以生产环境更推荐用 HTML 解析器拿到 script 标签内容后再json.loads。解析失败时不要静默返回空字典，直接抛错并把原始响应写入debug/目录，后面排查才有依据。

3.2 价格、库存和 SKU 的归一化

价格和库存是最容易出错的字段。接口可能返回分，页面可能显示元；活动价、到手价、划线价混在一起；库存可能是单 SKU 库存，也可能是规格聚合库存。我的做法是统一在解析层转成“元 + 两位小数”和“非负整数库存”，SKU 规格按 key 排序后拼成字符串，保证同一规格每次生成的字符串一致，方便入库去重。

def normalize_sku(api: dict) -> list[dict]: rows = [] for sku in api.get("sku_list", []): price_cent = int(sku.get("price_cent", 0)) stock = int(sku.get("stock", 0)) specs = sku.get("specs", {}) or {} rows.append({ "sku_id": str(sku["sku_id"]), "price_yuan": round(price_cent / 100, 2), "stock": max(stock, 0), "spec": "|".join(f"{k}:{v}" for k, v in sorted(specs.items())), }) return rows

price_cent是分，除以 100 后保留两位；max(stock, 0)把负库存压成 0；sorted(specs.items())是保证规格字符串稳定，避免同一 SKU 因为字典顺序不同被当成两条。如果接口同时返回活动价和原价，不要在解析层做“哪个更便宜”的判断，先把两个字段都落下来，让业务层决定用哪个口径。

3.3 SQLite 入库与幂等更新

小批量巡检用 SQLite 足够，关键是设计好主键。我的习惯是按“商品 ID + SKU ID + 批次日期”做联合主键，同一批次重跑时走ON CONFLICT DO UPDATE，跨天保留历史快照。这样既能幂等重跑，又能按天对比价格和库存变化。原始响应不要整段塞进主表，存一个raw_hash就够，原始文件按批次落盘，数据库只保留可查询字段。

CREATE TABLE IF NOT EXISTS item_snapshot ( item_id TEXT NOT NULL, sku_id TEXT NOT NULL, batch_date TEXT NOT NULL, title TEXT, price_yuan REAL, stock INTEGER, spec TEXT, raw_hash TEXT, updated_at TEXT DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (item_id, sku_id, batch_date) );

import sqlite3 import hashlib def save_snapshot(conn: sqlite3.Connection, item_id: str, title: str, skus: list[dict], batch_date: str, raw: str): raw_hash = hashlib.sha256(raw.encode("utf-8")).hexdigest() sql = """ INSERT INTO item_snapshot(item_id, sku_id, batch_date, title, price_yuan, stock, spec, raw_hash) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(item_id, sku_id, batch_date) DO UPDATE SET title=excluded.title, price_yuan=excluded.price_yuan, stock=excluded.stock, spec=excluded.spec, raw_hash=excluded.raw_hash, updated_at=CURRENT_TIMESTAMP """ with conn: for s in skus: conn.execute(sql, ( item_id, s["sku_id"], batch_date, title, s["price_yuan"], s["stock"], s["spec"], raw_hash ))

联合主键保证同一批次重跑不会产生重复行；ON CONFLICT后面的excluded表示本次要插入的值；raw_hash用来判断原始响应有没有变化。如果某天某个商品解析出 0 个 SKU，不要直接写库，先走告警，因为这通常意味着解析路径失效或登录态过期，空结果落库会把历史数据覆盖成脏数据。

4. 避坑排查：五个高频翻车点与定位顺序

这套链路跑通不难，难的是稳定跑一周。下面五条是我在商品详情爬取任务里踩过的典型坑，每条按“现象 → 原因 → 解决”写，排查顺序也按这个来。遇到问题时不要先改解析代码，先看请求层和登录态，再看字段口径，最后才怀疑页面结构。按这个顺序走，大部分问题十分钟内能定位。

4.1 登录态过期却返回 200

现象：任务没有报错，但落库的标题为空、SKU 数量为 0，或者价格全是 0。原因：登录态过期后，有些接口不返回 401，而是返回一个空对象、登录提示页或降级数据，状态码仍然是 200。解决：在解析前加一层“业务有效性校验”，标题为空、SKU 列表为空、价格小于等于 0 都记为可疑；连续 N 个商品可疑就暂停任务，重新注入 Cookie。日志里要打印响应体的前 300 个字符和当前 Cookie 的哈希前缀，不要打印完整 Cookie。

4.2 价格单位与优惠叠加误判

现象：同一商品今天 199 元，明天 19900 元，或者活动价和原价来回跳。原因：接口有的字段是分，有的是元；有的返回原价，有的返回到手价；优惠券和满减在详情页和结算页口径不同。解决：在解析层固定字段含义，price_cent只进分字段，price_yuan只进元字段，不要混用；把原价、活动价、到手价分三列落库；对账时用“价格区间”而不是单点值，超过阈值再人工看。

4.3 解析路径随页面结构漂移

现象：昨天还能解析 20 个 SKU，今天只解析出 1 个，或者spec全是空字符串。原因：页面把内嵌 JSON 从window.__ITEM__换成了__INITIAL_STATE__，或者把 SKU 列表挪到了另一个 script 标签。解决：不要写死一个 marker，配置多个候选 marker，按顺序尝试；解析失败时把原始响应落盘，用grep搜sku、price关键字定位新路径；每次页面改版后跑一遍固定样本回放，确认字段数量没掉。

4.4 限速参数设错导致误封

现象：前 100 个商品正常，第 101 个开始连续 429，换 Cookie 也没用。原因：QPS 设得太高，或者重试层把 429 也当成可重试错误，短时间放大请求量。解决：把 QPS 降到 0.3 到 0.5，重试只对连接错误和超时生效；命中 429 后暂停 5 到 10 分钟，记录当前商品 ID 做断点续跑；不要用多账号轮换来硬扛，频率问题不是换账号能解决的。

4.5 重复入库与脏数据回滚

现象：同一批次跑了两遍，表里行数翻倍，或者某天库存全是 0。原因：主键没设对，或者空结果被当成正常数据写入。解决：联合主键必须包含item_id、sku_id、batch_date；写入前校验 SKU 数量，0 个 SKU 直接进死信表；发现脏批次时按batch_date删除后重跑，不要逐行改。下面这条 SQL 用来定位重复和空批次，跑之前先备份。

-- 查同一商品同一 SKU 在同一天是否重复 SELECT item_id, sku_id, batch_date, COUNT(*) AS c FROM item_snapshot GROUP BY item_id, sku_id, batch_date HAVING c > 1; -- 查某批次 SKU 数量异常的商品 SELECT item_id, COUNT(*) AS sku_count FROM item_snapshot WHERE batch_date = '2025-01-01' GROUP BY item_id HAVING sku_count = 0;

5. 验证与进阶：用抽样回放和字段对账把结果钉死

采集任务上线前，我会固定 10 个商品 ID 做抽样回放：保存第一次的原始响应和解析结果，之后每次改代码都重跑这 10 个，对比raw_hash、SKU 数量、价格区间和库存总量。字段对账看四列：SKU 数量是否突然掉到 0，价格是否超出历史区间，库存是否全为 0，spec空值率是否超过 10%。这四列任何一列异常，先停任务再查原因。抽样回放不追求覆盖全部商品，追求的是“改动能被看见”。

# 固定样本回放：ids.txt 放 10 个商品 ID，baseline 存第一次结果 python replay.py --ids ids.txt --out runs/$(date +%F) --compare runs/baseline

replay.py是示例脚本名，核心逻辑就是读 ID 列表、走同一套请求和解析、把结果写成 JSON、和 baseline 做 diff。--compare指向基线目录，diff 结果里重点看sku_count、price_min、price_max、stock_sum四个键。从那以后我每次上线采集任务前，都强制跑一遍抽样回放和字段对账，哪怕只改了一行请求头；希望帮到你。

本文还有配套的精品资源，点击获取