Python商品详情爬虫：签名请求、限速重试与断点续爬

简介面向希望了解电商商品详情采集流程的开发者与爬虫初学者这份压缩包提供一套轻量示例围绕淘宝、天猫商品页的请求发送、页面或接口数据解析、字段提取与结果保存展开涉及动态加载、反爬限制和数据结构化等常见问题便于读者建立Web爬虫的基本流程认知并作为课程实验或小型采集任务的参考。包内共2个文件均为Java源码压缩后约4KB体量虽小但可对照阅读采集逻辑与关键处理思路。目前已有2988人学习说明该案例在入门实践中具有一定参考价值。读者可从中了解请求构造、页面元素定位、JSON字段读取及数据落盘等环节的组织方式并据此改造成自己的采集脚本适合具备基础编程能力、想快速理解商品详情抓取整体链路的学习者。1. 商品详情爬取先把「能跑通一条」放在第一位上周有个做选品的朋友丢给我七十多条链接问我能不能把标题、到手价、SKU 组合、销量、主图和详情图文整理成一张表。商品详情爬取这件事听起来简单实际推进起来真正难处理的地方几乎都在体系层面登录态过期、签名参数随时变化、价格藏在两层嵌套的 JSON 里、请求太密集就容易收到限制提示。我见过太多人一上来就搭分布式框架、写调度中心结果第一条商品都没解析出来。这份资源适合两类人一类是手里有一批商品 ID 或链接、想尽快把数据写入结构化表格的运营和数据分析同学另一类是想拿一个完整工程当骨架、往上加类目遍历、价格监控、竞品对比的开发者。下面按我实际拆代码的顺序讲先看结构再看核心链路最后讲怎么改。2. 项目结构与选型请求、解析、存储三层怎么切一份采集脚本能不能长期维护八成取决于第一天的目录怎么分。我拆过不少把「发请求、解 JSON、写文件」全塞进一个main.py的项目改到第三周就没人敢动了——因为加一个字段要在四百行里找位置。这个项目的骨架是按「职责边界」切的每个文件只干一件事互相之间靠数据结构通信不靠全局变量。2.1 目录骨架与模块职责先看目录树这是整个项目的物理结构后面所有代码都能落到具体文件上。item-spider/ ├── config/ │ └── settings.yaml # 接口地址、并发数、间隔、重试次数等 ├── spider/ │ ├── __init__.py │ ├── session.py # 会话管理Cookie、UA、公共请求头 │ ├── sign.py # 签名参数计算单独一个文件便于随时替换 │ ├── fetcher.py # 发请求 限速 重试只负责拿回原始响应 │ ├── parser.py # 纯函数JSON 进结构化 dict 出 │ └── storage.py # SQLite 建表、UPSERT、断点续爬状态维护 ├── data/ │ └── items.db # 落库文件随项目走不依赖外部服务 ├── logs/ │ └── spider.log ├── run.py # 命令行入口读参数、串流程 └── requirements.txt这个划分有两个明显好处。一是parser.py是纯函数输入一段 JSON、输出一个 dict不碰网络也不碰数据库可以拿历史响应离线跑单测字段改错了立刻能发现不用重新发请求。二是sign.py单独抽出来签名算法是整个项目里最容易失效的一环平台哪天改个拼接顺序你只需要动这一个文件不必在业务代码里翻找。常见的错误做法是把签名写在请求函数内部一旦参数顺序变化改动会散落在十几个地方。session.py里维护的是一个requests.Session()实例统一管理 Cookie 和请求头。很多人图省事每次requests.get()结果每次都是新会话登录态带不上服务端看到的就是一批「陌生浏览器」在密集访问这种情况很容易被识别为异常。会话复用的另一个价值是底层连接池复用几百条商品跑下来握手开销能省掉一大截。fetcher.py的定位要守住它只负责「把响应拿回来」不解析、不判断业务字段。限速、重试、超时、异常归类都在这层做。这样run.py的流程就非常直白——取 ID、调 fetcher、传给 parser、交给 storage中间没有隐式状态。我一般还会在 fetcher 里留一个last_response_hook方便临时把原始响应写到logs/raw/下排查这个钩子在排查线上问题时往往比任何日志框架都更有效。2.2 技术选型为什么是 requests 本地签名 SQLite选型这块没什么玄学按「采集量级」和「是否要长期维护」两个维度决定。下面是这个项目实际用的组合以及为什么不选另外几个常见方案。环节本项目选型备选不选的原因HTTP 请求requestshttpx / aiohttp单机几百到几万条同步足够异步调试成本高出问题难定位调度自写队列 固定间隔Scrapy 框架只有详情页一种任务上框架属于过度设计解析直接读 JSON 字典正则 / XPath详情数据在接口里正则一改版就全崩存储SQLiteMySQL / MongoDB单机文件型零部署天然支持 UPSERT 和事务导出pandas手写 CSV字段多、有嵌套时pandas 两行搞定先说请求层。requests的优势是生态成熟、调试直观response.text打印出来就能看配合session天然处理 Cookie。异步方案在单机、有严格间隔限制的场景下收益很小——你本来就要 sleep异步的并发优势发挥不出来反而引入了事件循环、连接复用、异常传播一堆新问题。我个人的习惯是单机采集只要预计总量在十万条以内一律先上同步跑不动了再谈异步。再说存储。SQLite 在这个场景里几乎是完美匹配不需要装服务、随项目目录走、单文件方便打包给别人而且它支持INSERT ... ON CONFLICT DO UPDATE天然适合「同一商品多次采集、后一次覆盖前一次」的增量更新。用 CSV 直接追加的问题是中途失败会写出半行脏数据而且没有主键约束重复跑一遍就重复一堆记录后面还得写去重脚本。用 MySQL 倒不是不行但为了一个本地采集脚本装一套数据库服务对拿到源码想跑通的人来说门槛太高。最后是解析。详情页数据一般在两个地方接口返回的 JSON或者渲染后的 HTML。这个项目优先走接口理由很实在——JSON 有层级、有键名字段缺失时会返回null而不是让整段解析失败HTML 结构一变选择器全废。所以parser.py里对 JSON 做的是dict.get()逐层取值配默认值兜底而不是一把json[a][b][c]直接下去那种写法一个字段缺失就是KeyError整个批次都会中断。2.3 配置与启动入口把魔法数字全部外置配置文件负责把「会变的东西」集中起来。间隔、重试次数、超时、接口地址这些写在代码里是维护灾难——同事拿去用改个间隔得读三遍源码。# config/settings.yaml api: base_url: https://example.com/h5/detail/api # 详情接口地址 app_key: your_app_key # 公开的 appKey timeout: 8 # 单请求超时秒 throttle: interval: 1.5 # 每次请求之间的基础间隔秒 jitter: 0.6 # 在基础间隔上叠加的随机抖动上限秒 max_retry: 3 # 单条商品最大重试次数 backoff_base: 2 # 退避基数第 n 次重试等待 backoff_base ** n 秒 storage: db_path: data/items.db export_csv: trueinterval和jitter这两个参数是重点。固定 1.5 秒的间隔在生产上是很明显的固定节奏加一点随机抖动让请求时间点散开能显著降低被识别为自动化行为的概率。backoff_base控制重试节奏第 1 次重试等 2 秒第 2 次 4 秒第 3 次 8 秒这种指数退避比「失败就立刻重试」友好得多——后者在服务端限流时会不断加剧问题。启动入口用argparse让同一份代码支持多种跑法# run.py import argparse import yaml from spider.session import build_session from spider.fetcher import Fetcher from spider.parser import parse_detail from spider.storage import Storage def load_conf(pathconfig/settings.yaml): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def main(): ap argparse.ArgumentParser() ap.add_argument(--ids, requiredTrue, help商品ID文件每行一个) ap.add_argument(--limit, typeint, default0, help最多处理多少条0 表示不限) ap.add_argument(--resume, actionstore_true, help跳过已成功的ID断点续爬) args ap.parse_args() conf load_conf() session build_session(conf) # 建会话加载 Cookie fetcher Fetcher(session, conf) # 请求器内部带限速与重试 storage Storage(conf) # 建表返回可复用的连接 with open(args.ids, r, encodingutf-8) as f: ids [line.strip() for line in f if line.strip()] if args.limit: ids ids[: args.limit] for idx, item_id in enumerate(ids, 1): if args.resume and storage.is_done(item_id): continue # 断点续爬已成功的直接跳过 raw fetcher.fetch_detail(item_id) # 返回原始 JSON失败返回 None if raw is None: storage.mark_failed(item_id) # 记录失败下次 resume 会重试 continue record parse_detail(raw, item_id) storage.upsert_item(record) print(f[{idx}/{len(ids)}] {item_id} ok) if __name__ __main__: main()这段入口的逻辑很朴素但几个点值得说明。--resume参数配合storage.is_done()是断点续爬的核心跑一万条中途断了重新执行会直接跳过已成功的部分不必从头再来。失败的商品单独标记而不是丢弃因为失败原因往往是临时的超时、触发限流隔一段时间重跑能救回大部分。--limit用于试跑新环境第一件事永远是--limit 5跑通再放量这条血泪经验值十条代码。3. 核心链路拆解从商品 ID 到一条结构化记录这条链路是整个项目里唯一「不可替代」的部分其余都是外围保障。拆开来是三步拼参数算签名、把响应解成扁平结构、写进库并记录状态。三步里每一步都有具体的失败模式下面逐段过。3.1 参数拼装与签名计算详情接口通常不接受「裸 ID」而是要求一组带时间戳和签名的查询参数。签名的作用是证明请求来自合法页面算法一般是对若干字段按固定顺序拼接后取 MD5 或 HMAC。这个项目中把它封装成一个纯函数# spider/sign.py import hashlib import time import random import string def _token_from_cookie(cookies: dict) - str: 从 Cookie 里取签名用的 token常见键名带 _m_h5_tk 前缀 for k, v in cookies.items(): if k.startswith(_m_h5_tk) and _ in v: return v.split(_)[0] return def build_signed_params(item_id: str, app_key: str, cookies: dict) - dict: token _token_from_cookie(cookies) t str(int(time.time() * 1000)) # 毫秒时间戳 data f{{itemId:{item_id}}} # 业务参数JSON 字符串形式 # 签名原文token 时间戳 appKey data顺序不可变 raw f{token}{t}{app_key}{data} sign hashlib.md5(raw.encode(utf-8)).hexdigest() return { jsv: 2.5.1, appKey: app_key, t: t, sign: sign, api: detail.item.get, v: 1.0, type: json, dataType: json, data: data, _: str(int(time.time() * 1000)) random.randint(100, 999), }逻辑上有三处不能省。第一token来自 Cookie而不是配置里写死的常量这意味着签名和登录态是绑定的——换账号、Cookie 过期签名就跟着失效这也是「签名错误」和「登录态失效」经常同时出现的原因。第二t必须是毫秒时间戳且与服务端时差不能太大本机时间不同步会直接报签名非法这个坑排查起来相当费时间。第三data是 JSON 字符串而不是 dict拼接时要用json.dumps保证键顺序和引号一致手写 f-string 在参数变多时很容易出转义问题我建议实际项目中直接换成json.dumps(payload, separators(,, :))。_这个参数是时间戳加随机数纯粹用于避免服务端和中间层缓存不参与签名。很多项目的「同一商品两次请求返回不同内容」问题就是漏了这个参数。3.2 响应解析价格、SKU、图文详情接口返回的 JSON 通常有三层嵌套外层是状态码中层是data真正的内容在data.item和data.sku里。价格单位常见是分需要除以 100SKU 是一个数组每个元素包含规格组合和库存。解析函数要做的就是把它压平。# spider/parser.py def _to_yuan(v): 分转元缺失或非法返回 None避免污染后续统计 try: return round(int(v) / 100, 2) except (TypeError, ValueError): return None def parse_detail(raw: dict, item_id: str) - dict: data raw.get(data) or {} item data.get(item) or {} sku_block data.get(sku) or {} # SKU把 [{specId, price, stock}] 拼成可读的规格字符串 skus [] for s in sku_block.get(skuList) or []: skus.append({ sku_id: s.get(skuId), spec: /.join(s.get(specDesc) or []), # 颜色/尺码拼接 price: _to_yuan(s.get(price)), stock: s.get(quantity, 0), }) prices [x[price] for x in skus if x[price] is not None] return { item_id: str(item_id), title: (item.get(title) or ).strip(), shop_id: item.get(sellerId), price_min: min(prices) if prices else _to_yuan(item.get(price)), price_max: max(prices) if prices else None, sku_count: len(skus), skus_json: skus, # 落库时序列化为文本 main_image: (item.get(images) or [None])[0], desc_images: item.get(descImgs) or [], # 图文详情图片列表 sold_count: item.get(soldCnt, 0), }几个设计细节解释一下。_to_yuan用try/except兜住类型异常因为接口在商品下架时可能返回字符串-1或者干脆null直接int()会让整条记录失败。SKU 列表可能为空单规格商品这时用min/max会抛ValueError所以先判断prices是否为空再回退到商品级价格。skus_json保留完整列表落库时转成 JSON 文本存一列导出时再展开——这样既没有丢信息也不用为了 SKU 单独建两张关联表对单机采集来说是性价比最高的做法。sold_count这类字段经常在接口里叫soldCnt、soldCount、tradeCount各种变体甚至同一接口不同类目返回的键名不一样。稳妥的写法是封装一个_pick(d, *keys)帮助函数按优先级取第一个非空值而不是在解析函数里堆or表达式。3.3 落库与断点续爬让重跑变成安全操作存储层的目标是无论跑多少遍库里每个商品只有一条最新记录且能区分「从来没过」和「跑过但失败了」。表结构直接为此服务。# spider/storage.py import json import sqlite3 DDL CREATE TABLE IF NOT EXISTS items ( item_id TEXT PRIMARY KEY, title TEXT, shop_id TEXT, price_min REAL, price_max REAL, sku_count INTEGER, skus_json TEXT, main_image TEXT, sold_count INTEGER, status TEXT DEFAULT ok, -- ok / failed updated_at TEXT DEFAULT (datetime(now,localtime)) ); CREATE INDEX IF NOT EXISTS idx_status ON items(status); class Storage: def __init__(self, conf): self.conn sqlite3.connect(conf[storage][db_path]) self.conn.execute(PRAGMA journal_modeWAL;) # 读写并发更稳 self.conn.executescript(DDL) def is_done(self, item_id: str) - bool: cur self.conn.execute( SELECT status FROM items WHERE item_id?, (str(item_id),)) row cur.fetchone() return bool(row) and row[0] ok def upsert_item(self, rec: dict): rec dict(rec, skus_jsonjson.dumps(rec.get(skus_json), ensure_asciiFalse)) cols ,.join(rec.keys()) ph ,.join([?] * len(rec)) sql fINSERT INTO items ({cols}) VALUES ({ph}) \ fON CONFLICT(item_id) DO UPDATE SET \ ,.join([f{k}excluded.{k} for k in rec if k ! item_id]) \ , statusok, updated_atdatetime(now,localtime) self.conn.execute(sql, list(rec.values())) self.conn.commit() def mark_failed(self, item_id: str): self.conn.execute( INSERT INTO items (item_id, status) VALUES (?, failed) ON CONFLICT(item_id) DO UPDATE SET statusfailed, (str(item_id),)) self.conn.commit()ON CONFLICT ... DO UPDATE是整个存储层的核心它把「新增」和「更新」合并成一个动作跑第二遍不会产生重复行也不会误删已有字段。WAL模式是为了在导出 CSV 的同时还能继续写入不至于锁表。status字段让失败记录留在表里而不是消失配合is_done()就能实现精确的断点续爬成功的跳过失败的自动重试。有一点要注意rec.keys()的顺序在 Python 3.7 是有序的list(rec.values())与之对应所以这段拼接是安全的。但如果有人后续往rec里加字段却忘了在 DDL 里加列INSERT会直接报 no column 错误——我的习惯是在__init__里做一次PRAGMA table_info(items)校验缺列就自动ALTER TABLE ADD COLUMN省得每次改字段都要手动迁移。4. 稳定性处理限速、重试、登录态维护采集脚本写出来容易跑稳很难。前一百条顺风顺水跑到两千条突然开始大面积返回空数据这时候翻日志往往什么都看不出来是个典型的黑匣子。这一章讲的就是把黑匣子变成可观测的三件事。4.1 会话与 Cookie 维护Cookie 不是「启动时读一次就完事」的。详情接口的登录态一般有有效期短的几小时长的几天过期后接口会返回一个「需要登录」的状态码而不是 HTTP 401很多人就是卡在这里——状态码是 200body 里却是错误信息。# spider/session.py import requests DEFAULT_UA (Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36) def build_session(conf, cookie_str: str None) - requests.Session: s requests.Session() s.headers.update({ User-Agent: DEFAULT_UA, Accept: application/json, text/plain, */*, Referer: conf[api][base_url].rsplit(/api, 1)[0] /, Accept-Language: zh-CN,zh;q0.9, }) if cookie_str: # 形如 k1v1; k2v2 的字符串转成 dict for part in cookie_str.split(;): if in part: k, v part.strip().split(, 1) s.cookies.set(k, v) return sReferer不能省。不少接口会校验来源页面从空 Referer 直接打接口返回的往往是精简版数据或者干脆空壳。Accept-Language影响部分文案字段的返回语言做多语言类目时尤其要注意。Cookie 的更新策略我一般这么写把 Cookie 字符串放在配置文件或环境变量里脚本启动时读入运行过程中每 N 条检查一次响应里是否带有Set-Cookie有就session.cookies.update()合并进去。这样在长时间运行里会话能自己「续命」不用每两小时人工换一次。提示Cookie 属于个人登录凭证务必只在自己有权访问的数据范围内使用不要写进代码仓库也不要提交到公开仓库里。4.2 限速与指数退避重试限速不是「加个 sleep 就行」得区分三种情况正常请求的间隔、被限流后的退避、以及连续失败后的熔断。这个项目的做法是在 fetcher 里维护一个「上次请求时间戳」每次请求前算一下还差多少。# spider/fetcher.py import time import random import logging import requests from spider.sign import build_signed_params log logging.getLogger(__name__) class Fetcher: def __init__(self, session, conf): self.s session self.conf conf self.api conf[api] self.th conf[throttle] self._last 0.0 def _wait(self): base self.th[interval] random.uniform(0, self.th[jitter]) delta time.time() - self._last if delta base: time.sleep(base - delta) # 只补差额不无脑 sleep self._last time.time() def fetch_detail(self, item_id: str): for attempt in range(1, self.th[max_retry] 1): self._wait() params build_signed_params( item_id, self.api[app_key], self.s.cookies.get_dict()) try: r self.s.get(self.api[base_url], paramsparams, timeoutself.api[timeout]) except requests.RequestException as e: log.warning(net err %s attempt%d: %s, item_id, attempt, e) time.sleep(self.th[backoff_base] ** attempt) continue body r.json() if r.headers.get(Content-Type, ).startswith(application/json) else {} if body.get(ret) and str(body[ret][0]).startswith(SUCCESS): return body # 明确被限流的信号退避时间拉长 code str((body.get(ret) or [])[0]) log.warning(biz err %s attempt%d code%s, item_id, attempt, code) time.sleep(self.th[backoff_base] ** attempt) return None # 重试耗尽交给上层标记失败_wait里的「只补差额」是个小优化但很有用如果上一次请求本身就因为网络慢花了 3 秒那这次就不用再等 1.5 秒了。无脑time.sleep(interval)在慢网络下会把整体耗时拉得很长。退避部分把异常分成两类处理网络异常直接重试业务状态码异常则先记录再退避。backoff_base ** attempt就是 2、4、8 的节奏。如果连续多条商品都在重试耗尽后失败说明不是单条数据的问题而是会话或接口整体异常这时候应该停下来——我一般会加一个「连续失败计数」超过阈值就抛出异常终止整批而不是让它把剩下的几千条全刷成 failed。4.3 渲染兜底接口走不通时的第二条路有些类目的详情数据不在接口里或者接口对未登录用户只返回半截数据这时候只能走页面渲染。兜底方案用 Playwright重点是「只在必要时启用」不能全量走渲染——浏览器实例的开销是纯 HTTP 的几十倍。# spider/render.py可选兜底模块 from playwright.sync_api import sync_playwright def fetch_by_render(url: str, cookie_str: str, timeout_ms: int 15000) - str: with sync_playwright() as p: browser p.chromium.launch(headlessTrue) ctx browser.new_context(user_agentMozilla/5.0 ... Chrome/122.0.0.0) # 把登录态注入浏览器上下文避免渲染出来是未登录页面 for part in cookie_str.split(;): if in part: k, v part.strip().split(, 1) ctx.add_cookies([{name: k, value: v, domain: .example.com, path: /}]) page ctx.new_page() page.goto(url, timeouttimeout_ms, wait_untildomcontentloaded) page.wait_for_selector([data-detail-container], timeouttimeout_ms) html page.content() browser.close() return html关键参数是wait_until和wait_for_selector。用networkidle在详情页上经常等不到因为页面里有长轮询或埋点请求一直在发domcontentloaded配合等一个具体的选择器更可靠。add_cookies里的domain必须和实际域名匹配写错了 Cookie 就等于没注入页面会渲染成未登录态这个错误不会报异常只会让你拿到一份缺字段的数据。启用条件我建议写成接口返回的业务码属于「需要登录」或「数据为空」时才把该商品丢进渲染队列且整批里渲染占比超过 20% 就告警。渲染是兜底不是主路。5. 常见问题排查跑不通时先看这几处这一章是我自己踩过的坑按「现象 → 原因 → 解决」整理。新手照着对一遍能省掉大半天。5.1 请求层状态码 200 但数据是空的现象脚本不报错日志里全是 ok但导出的表里标题为空、价格为空sku_count是 0。原因绝大多数是登录态失效。接口返回 HTTP 200body 里是一个ret数组带错误码或者data是空对象。parser里全部用.get()取值所以不会抛异常静静地返回一堆空字段。这是最坑的一类问题因为「看起来成功」。解决在 fetcher 里做业务码判定不要只看 HTTP 状态。拿到响应后先检查ret[0]是否以成功标识开头不是就归类为失败并记录原始 body 到logs/raw/。同时给 parser 加一条校验title为空且price_min为 None 的记录直接判定为无效不计入成功数。跑完后打印「成功 / 失败 / 空数据」三个计数数字对不上立刻能发现。5.2 请求层签名错误反复出现现象所有请求返回FAIL_SYS_ILLEGAL_ACCESS之类的签名错误换 Cookie 也没用。原因三种可能。一是本机时间不准签名里的时间戳和服务端差了几分钟服务端直接判非法二是 Cookie 里的 token 提取失败_token_from_cookie返回空字符串签名自然算错三是业务参数data的 JSON 序列化方式变了比如多了空格或者键顺序不同。解决先date看一下系统时间容器环境里时钟漂移很常见配好时间同步。再单独写个三行脚本打印_token_from_cookie的返回值空的话说明 Cookie 键名变了去浏览器的 Application 面板里核对实际键名。最后确认data用的是json.dumps(payload, separators(,, :), ensure_asciiFalse)这种确定性序列化不要手拼字符串。5.3 解析层价格对不上、SKU 缺档现象导出的价格比页面上看到的高一截或者 SKU 数量比页面少。原因价格单位搞错是高频问题接口返回分、页面展示元忘了除 100 就会差两个数量级。SKU 数量对不上一般是分页——部分接口的skuList单次最多返回 50 条超过要翻页另一种情况是多规格商品的 SKU 分散在sku.core和sku.extra两个块里只读了一个。解决解析前先打一条完整原始响应出来肉眼核对。价格统一走_to_yuan()一个入口杜绝各处手写除法。SKU 如果存在分页字段类似hasMore在 fetcher 层做一个「拉全 SKU」的循环把所有页合并后再交给 parser。合并时按sku_id去重避免翻页边界重复。5.4 运行层跑到一半中断续跑后数据错乱现象脚本跑了几千条崩了重跑之后表里出现重复行或者某些商品被覆盖成了空记录。原因没用 UPSERT直接INSERT重跑就重复或者--resume用的是「文件里第几行」这种位置判断而不是查库导致跳过逻辑完全错位。解决主键item_idON CONFLICT DO UPDATE是唯一可靠方案重跑天然幂等。--resume一律查库判断statusok不要用外部状态文件。另外upsert之前先判空title和price_min都为空就不写库避免失败响应把上一次的好数据覆盖掉——这个坑我在第一次做增量更新时踩得很深几十条商品的好数据被一次会话失效全冲掉了。5.5 应用层导出 CSV 用 Excel 打开是乱码现象pandas 导出的 CSV 双击打开中文全是问号或者方块。原因Excel 默认按本地编码GBK 系解析无 BOM 的 UTF-8 文件。解决df.to_csv(path, indexFalse, encodingutf-8-sig)加 BOM 头Excel 就能正确识别。这个不算技术难点但每次交付给别人都会被问一遍干脆写进默认导出函数里。6. 二次开发从小脚本到能交付的采集服务拿到这份骨架之后最自然的改造方向有三个我按投入产出比排序。第一个是加「类目遍历」。现在是给一批 ID 跑详情往上加一层就变成「给一个类目 ID翻页拿全部商品 ID再逐条跑详情」。改动点很小新增一个fetch_id_list(category_id, page)的函数把结果写进items表时status标记为pending详情任务只取statuspending的行。这样采集和详情彻底解耦ID 列表可以反复刷详情可以慢慢跑。第二个是加「价格快照」。当前表结构只保留最新价做价格监控就不够了。加一张price_history(item_id, price, captured_at)表每次 upsert 时顺手插一条历史记录。查询时用窗口函数取每个商品的最新两条做对比涨跌一目了然。# 价格快照写入接在 storage.upsert_item 之后调用 def save_snapshot(conn, item_id: str, price: float): conn.execute( INSERT INTO price_history (item_id, price, captured_at) VALUES (?, ?, datetime(now,localtime)), (item_id, price)) conn.commit()第三个是加「配置化的字段映射」。不同类目的详情接口返回的键名差异不小把parser.py里硬编码的键名抽成 YAML 里的一张映射表换类目只改配置不改代码。这个改造的意义在于以后接口小改版你只需要调映射表不用重新读一遍解析函数。验证改造是否成功我的习惯是固定一组「基准样本」——二十条覆盖各种情况的商品 ID包含多规格、单规格、已下架、预售。每次改完跑这二十条比对输出的字段数、空值率、价格区间和改造前的结果做 diff。这个方法比写单测更接地气因为采集脚本最大的不确定性来自外部数据而不是内部逻辑。改造方向涉及文件代码改动量风险点类目遍历fetcher.py / run.py小ID 列表接口的分页终止条件价格快照storage.py小表膨胀需要定期归档字段映射parser.py settings.yaml中映射缺失时的默认值策略最后说一个我自己的习惯。这套代码我每接一个新类目第一件事永远是--limit 3跑三条然后把原始 JSON 和解析结果并排打印出来逐字段核对确认价格单位、SKU 是否完整、图片链接是不是可访问的地址再放量到全量。早些年我嫌这一步麻烦直接上五千条结果跑完才发现价格全单位错了只能清库重来那次的后悔药没处买。从那以后我每次改解析逻辑都强制走一遍这二十条基准样本跑通再谈性能。希望这篇拆解能帮你把这套骨架跑起来剩下的就是往里填你自己业务需要的字段了。本文还有配套的精品资源点击获取