企业官网建设流程全解析

简介：面向需要跨平台采集电商商品数据的Python开发者，这套源码实现了淘宝、京东、拼多多、1688与京喜的商品信息自动化爬取，可提取商品链接、价格、名称、店铺名称及店铺链接，并通过Tkinter图形界面直观监控爬虫运行状态。压缩包内含20个文件，以8个Python脚本为骨架，分别覆盖淘宝、京东、1688、拼多多、京喜等平台的爬虫主体，以及HAR数据读取、Cookie获取等辅助模块；另含5张抓包与开发工具操作截图、2个Markdown说明文档、2个txt说明文件，以及wav提示音、.gitignore和LICENSE等，整体仅1.21MB，轻量易部署。目前已有348人学习下载，适合正在搭建多平台商品采集系统、研究爬虫架构或准备相关毕设的开发者参考。从中可掌握多站点请求构造、HAR日志解析、登录Cookie复用、GUI状态监控等关键环节，配合说明文档与操作截图可快速迁移到自己的项目，显著减少重复开发与踩坑成本。

1. 基于 Python 的五大电商平台商品信息爬虫：为什么值钱的是设计而不是代码

做电商选品、竞品监控或者比价工具的朋友，大概率都搜过这类源码——“基于 Python 的淘宝、京东、拼多多、1688、京喜商品信息爬虫”。搜回来的代码包不少，免费 Python 源码一抓一大把，可真拿去跑，一天内能顺利出数据的没几个。问题几乎不在正则写错，而在设计缺位：字段没有统一模型、请求失败没有降级、数据重复没人管。这个方向真正值钱的，是一套拆开能维护的架构——表结构、平台适配器、限速与重试策略。对想入 Python 爬虫这行的开发者，以及有真实商品数据采集需求的运营或工程师，搞清楚这套设计，比拿到一百个源码都管用。下面按我落地的顺序讲：选型、定表、单平台跑通、避坑、验证。

2. 淘宝京东拼多多 1688 京喜的页面差异与采集选型：一张表看懂该怎么选

2.1 五个平台的数据藏在哪：HTML 与 JSON 接口的真实分布

开工前第一件事不是选库，而是搞清楚目标平台的数据到底落在哪里。五个平台虽然都是电商，但前端架构差异很大，数据来源大致分三类：服务端渲染的 HTML、页面内嵌的 JSON 变量、异步加载的接口返回。

淘宝搜索页是典型的重前端。搜索列表拿到的是页面骨架，商品信息要么嵌在页面脚本的初始化数据里，要么来自后续的异步搜索接口。京东相对“老实”一点，列表页 HTML 里能直接翻到商品 JSON，详情页则走独立的详情接口，而且不同端——PC、H5、App——返回的字段名都不一样。拼多多 H5 页面的数据集中在 window 下的某个全局变量里，需要先定位再 JSON 解析。1688 是老牌 B 端平台，很多页面仍是服务端渲染，直接解析 HTML 就能拿到标题、价格和起订量，但要拿到成交数据，多半得登录。京喜的玩法更接近小程序逻辑，大部分数据来自后端异步接口。

常见做法是先用浏览器开发者工具抓一轮请求，确认数据是不是直接出现在 HTML 响应里。这决定了后面用 requests 还是 Selenium，也决定了你要不要为每个平台单独写解析器。

2.2 请求库选型：requests、httpx 与 Python Selenium 的边界

选请求库不能只看熟不熟，要看场景。requests 适合页面或接口直接返回数据的场景，胜在直观、文档多、用 Session 维持 Cookie 方便，入门刷一遍网上的 requests 爬虫教程就能上手。如果你的目标是五平台通吃，requests 一定是主武器，因为大部分数据其实藏在 JSON 里，不需要渲染。

httpx 和 requests 的 API 几乎一致，但支持连接复用和异步批量请求。采集任务一旦到了“每天几十万商品”的规模，用 httpx 的异步客户端能省下大量等待时间。我个人习惯是 requests 先跑通单页，再选 httpx 或线程池做并发，而不是一上来就上异步。

Python Selenium 的定位是兜底，不是首选。页面数据全部靠 JS 动态渲染、或者必须模拟真实浏览器才能拿到完整内容时才用。很多人误以为 Selenium 是为了“反爬虫”才引入的，其实更多时候解决的是页面滚动加载、点击翻页这类交互问题。代价也很明显：每个页面要多开一个浏览器进程，单机吞吐量直线下降，而且 headless 浏览器本身更容易被识别。

第三类特殊场景是移动端数据。拼多多和京喜的部分商品字段在 PC 端拿不全——比如拼团价、券后价。常见做法是抓包手机端 H5 页面，把接口参数照搬到 requests 里。这比开模拟器轻量得多，但需要你熟悉抓包工具。

2.3 为什么不能“一份模板跑五个平台”

有新手问能不能写一个统一模板，传个平台名进去就跑。我的回答是：不要这么做。五个平台的登录链路、页面结构、字段命名、分页规则各不相同，强行统一解析层，代码会塞满if platform == 'xxx'这种分支，最后没人敢改。

通用的只有两层：数据模型和调度框架。数据模型保证入库结构一致，调度框架处理限速、重试、去重；具体平台的解析逻辑各自封装成一个适配器，对外暴露统一接口。这样加一个新平台就是新增一个类，不用动主流程。

from abc import ABC, abstractmethod class BaseSpider(ABC): platform = "base" def __init__(self, session, timeout=10): self.session = session self.timeout = timeout @abstractmethod def parse_search(self, keyword: str, page: int): """返回 Product 实例列表""" @abstractmethod def fetch_detail(self, product): """补全商品详情字段"""

这段代码的核心是把“请求是什么样子”和“数据怎么解析”拆开。parse_search返回的不再是字典，而是统一的商品对象，这样入库代码只认一种结构。爬虫最怕的不是慢，是改需求——今天新增一个字段，结果五个解析器各写各的，改到想哭。

2.4 先定表结构再写爬虫：商品、价格历史、任务日志三类表

我吃过亏才总结出这个顺序：先建表，再写解析。表结构一旦定下来，字段名就是解析器的“接口契约”，每层各管各的，能少踩一半坑。

products表必须加(platform, item_id)唯一联合索引，这是去重的第一道保险。price_history只做追加，不做更新，这样历史数据天然不可变，后面画价格趋势、算涨跌幅都靠它。日志表很多人觉得没用，等到线上任务挂了你不知道挂在哪一步的时候，就会后悔没多写这一行。

3. 商品数据模型与 SQLAlchemy 存储：用 ORM 管理爬虫数据的三层架构

3.1 用 dataclass 定义统一商品模型，告别裸字典

爬虫代码最容易写烂的地方，是解析函数里直接返回嵌套字典。键值对了还好，键名一改，下游全部报错，而且报错信息是“KeyError: 'price'”，你还得去找是哪个解析器漏了字段。我的习惯是先定义一个统一商品模型，所有平台解析结果都落进这个结构。

from dataclasses import dataclass, field from datetime import datetime @dataclass class Product: platform: str item_id: str title: str price: float sale: int shop_name: str url: str extra: dict = field(default_factory=dict) captured_at: datetime = field(default_factory=datetime.now)

为什么用 dataclass 而不是字典？因为字段是显式的，解析器少返回一个字段，Python 直接报TypeError而不是跑到入库阶段才崩。extra字典留给平台特有字段，比如淘宝的“发货地”、京东的“自营标识”，这样通用字段保持稳定，特殊字段又不会丢。

价格字段我统一成float，销量统一成int。不要在模型里留“原价”“到手价”两个浮点，除非你的业务真的需要——多一个字段多一重清洗工作。采集阶段保证字段类型干净，比把压力抛给数据库强。

3.2 用 SQLAlchemy 储存爬虫数据的落地姿势：会话、唯一索引与幂等写入

数据量级在一两百万条以内，SQLite 完全撑得住。SQLAlchemy 的好处是以后量大了换 MySQL，只需要改create_engine的连接串，模型代码一行不用动。

from sqlalchemy import create_engine, Column, String, Float, Integer, DateTime, UniqueConstraint from sqlalchemy.orm import declarative_base, sessionmaker from datetime import datetime engine = create_engine("sqlite:///products.db", echo=False) Base = declarative_base() class ProductORM(Base): __tablename__ = "products" id = Column(Integer, primary_key=True) platform = Column(String(16), index=True) item_id = Column(String(64)) title = Column(String(255)) price = Column(Float) sale = Column(Integer) url = Column(String(512)) updated_at = Column(DateTime, default=datetime.now) __table_args__ = ( UniqueConstraint("platform", "item_id", name="uniq_platform_item"), ) Base.metadata.create_all(engine) Session = sessionmaker(bind=engine)

echo=False是必须的，别开着，不然控制台被 SQL 日志刷爆。唯一约束告诉数据库：同一个平台同一个商品只能有一行。这比在代码里先查后插更可靠，因为并发写的时候代码里的“先查”会漏。

写入要做成幂等操作：查到就更新，查不到就插入。价格变化不能直接改products.price旧值，那是更新，不是留痕。

def upsert_product(db_session, product: Product): orm = db_session.query(ProductORM).filter_by( platform=product.platform, item_id=product.item_id ).first() if orm: orm.title = product.title orm.price = product.price orm.sale = product.sale orm.url = product.url else: db_session.add(ProductORM( platform=product.platform, item_id=product.item_id, title=product.title, price=product.price, sale=product.sale, url=product.url, )) db_session.commit()

注意这里只更新了products表。价格历史是另一码事，每抓到一次价格就往price_history里插一行。这个函数是五平台共用的，任何平台的适配器解析完商品对象，都走同一个入库入口。

3.3 调度骨架：线程数、限速与重试的保守参数

调度模块的职责是控制采集节奏，不是拼命跑。单 IP 并发过高，平台的风控马上会盯上你。我第一次做淘宝采集时，开了 8 个线程没限速，半小时后搜索页直接返回验证码，整批任务全废。后来学乖了：线程数 2 到 4，单次任务间隔 1 到 3 秒随机。

import time import random from queue import Queue from threading import Thread task_queue = Queue() def worker(spider_cls, session): while True: task = task_queue.get() if task is None: break keyword, page = task try: spider = spider_cls(session) products = spider.parse_search(keyword, page) with Session() as db: for p in products: upsert_product(db, p) except Exception as exc: print(f"task failed: {keyword} page:{page} -> {exc}") finally: task_queue.task_done() time.sleep(random.uniform(1, 3)) threads = [Thread(target=worker, args=(TaobaoSpider, session)) for _ in range(3)] for t in threads: t.start()

random.uniform(1, 3)是必须的，固定间隔很容易被识别出机器节奏。重试不能放在这个循环里无脑重试，正确的做法是记录失败任务，下一轮调度再处理。失败的 keyword 和 page 写进crawl_tasks表，标记为 failed，后面单独补跑。

3.4 会话生命周期：一个线程一个连接，别共享

SQLite 有个著名的坑：多线程共用连接，报sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread。这不是玄学，是 SQLite 默认线程模式的限制。

解法是每个线程各自创建Session。上面代码里with Session() as db每次任务都拿新会话，用完即关，虽然会多一点点开销，但安全。如果你一定要共享引擎，给create_engine加参数：

engine = create_engine( "sqlite:///products.db", connect_args={"check_same_thread": False} )

这个参数的意思是允许跨线程使用同一个连接。但我建议只在调试时用，生产环境还是“一线程一会话”更稳。另一个提升并发写性能的办法是开 WAL 模式，一条 SQL 搞定：

PRAGMA journal_mode=WAL;

WAL 模式下读和写可以并行，采集线程写库的时候，查询线程不会被锁住，对后面做数据校验非常有用。

4. 基于 Python 的淘宝商品爬虫完整实现：搜索页解析、Selenium 兜底与字段清洗

4.1 数据定位：淘宝搜索页的商品 JSON 到底在哪

淘宝是个很好的示范平台，因为它把难题都集齐了：需要登录态、页面动态渲染、部分接口带加密参数。以搜索页为例，直接requests.get拿到的 HTML 里，商品列表并不在标准 HTML 标签里，而是藏在页面内嵌的初始化数据中。

常见做法是搜索页面源码里的window.__INITIAL_DATA__或者类似命名的全局变量。这一段文本很大，直接用正则切出来，再用json.loads解析。如果哪天这个变量名变了，说明淘宝改了前端结构，你的解析器也要跟着升级。

import requests import re import json def load_search_page(keyword: str, page: int, cookie: str): headers = { "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", "Referer": "https://www.taobao.com/", "Cookie": cookie, } params = {"q": keyword, "s": (page - 1) * 60} resp = requests.get( "https://s.taobao.com/search", params=params, headers=headers, timeout=10, ) resp.encoding = "utf-8" match = re.search(r"window\.__INITIAL_DATA__\s*=\s*(\{.*?\});", resp.text, re.S) if not match: raise RuntimeError("未找到初始化数据，可能被重定向到登录页") return json.loads(match.group(1))

淘宝搜索的分页参数是s，不是page或pn。第一页s=0，第二页s=60，一页 60 个商品。注意timeout=10不能省，不设超时，遇到慢响应时线程会一直挂着，越积越多。Cookie 从浏览器复制，登录一次通常能管几小时到一天。

4.2 从 JSON 里提取商品字段：字段映射表是解析器的核心

拿到初始化数据之后，接下来就是定位商品列表的数组路径。不同时期的淘宝页面，数组路径不一样，常见的位置是在data下的某个列表字段里。我的做法是先打印 JSON 的 top-level key，再用递归搜索包含item_id和title的节点。

def extract_products(page_data: dict): def find_items(node): if isinstance(node, dict): if "item_id" in node and "title" in node: yield node for value in node.values(): yield from find_items(value) elif isinstance(node, list): for item in node: yield from find_items(item) for raw in find_items(page_data): try: yield Product( platform="taobao", item_id=str(raw["item_id"]), title=raw["title"], price=float(raw["price"]), sale=int(raw.get("sale", 0)), shop_name=raw.get("nick", ""), url=f"https://item.taobao.com/item.htm?id={raw['item_id']}", ) except (KeyError, ValueError, TypeError): continue

递归找字段有个好处：淘宝哪天把商品列表挪了个层级，只要字段名没变，代码仍然能跑。try...except必须加，因为列表里偶尔混着广告位或推荐位，缺字段直接跳过，别让一颗老鼠屎坏了一锅汤。

4.3 Python Selenium 兜底：什么时候才值得动用浏览器

如果初始化数据找不到，或者页面改版把 JSON 内嵌改成了纯异步渲染，这时候才上 Selenium。很多人以为 Python Selenium 就是为了对付反爬虫，实际上它解决的是“数据根本不进 HTML”的极端情况。

from selenium import webdriver from selenium.webdriver.common.by import By from selenium.webdriver.chrome.options import Options def fetch_with_selenium(keyword: str, page: int): opts = Options() opts.add_argument("--headless=new") opts.add_argument("--disable-gpu") opts.add_argument("--lang=zh-CN") driver = webdriver.Chrome(options=opts) try: url = f"https://s.taobao.com/search?q={keyword}" driver.get(url) driver.implicitly_wait(10) cards = driver.find_elements(By.CSS_SELECTOR, "[data-nid]") print(f"拿到 {len(cards)} 个商品卡片") finally: driver.quit()

--headless=new是 Chrome 109 之后的新无头模式，旧写法--headless在新版本里会被警告。implicitly_wait(10)是全局等待，每个find_element都会等，如果你的页面有异步加载，建议改成显式等待，只等关键元素出现。Selenium 模式只建议在 requests 方案失效时启用，不要默认走这条路。

4.4 字段清洗：价格、销量与“到手价”口径问题

平台返回的原始字段基本都不是干净的。价格可能带“¥”、带逗号，销量可能是“10万+”，这些字符串直接入库，后面做数据分析会很难受。清洗函数要独立出来，单元测试也好写。

import re def clean_price(value) -> float: if isinstance(value, (int, float)): return float(value) text = str(value).replace("¥", "").replace(",", "") match = re.search(r"\d+(\.\d+)?", text) return float(match.group()) if match else 0.0 def clean_sale(value) -> int: if isinstance(value, int): return value text = str(value) if "万+" in text: return int(float(text.replace("万+", "")) * 10000) if "万" in text: return int(float(text.replace("万", "")) * 10000) digits = re.sub(r"\D", "", text) return int(digits) if digits else 0

这两个函数的边界情况很多：价格传None时返回0.0还是抛异常？销量出现“1.2万”时乘法会得到浮点，要先转int。我在项目里会把清洗规则统一成这样：清洗失败返回默认值，同时在日志里记一条 warning，而不是让异常中断整个采集任务。

另外一个隐蔽的坑是“到手价”和“原价”。同一件商品，列表页显示的是券后到手价，详情页显示的是原价，两个数不一样。如果你把两种口径混着存，价格历史曲线会莫名其妙跳动。我的做法是products表只存列表页的统一口径字段，详情页补全的价格单独放在extra里，不参与历史曲线计算。

5. 多平台商品爬虫的避坑清单：登录态、限频、页面改版与入库异常

5.1 请求突然返回登录页或验证码

现象：爬虫跑了一个小时正常，突然返回的 HTML 不再是商品列表，而是登录跳转页或验证码页，解析器报“未找到初始化数据”。

原因：请求频率触发了平台的风控策略，或者浏览器里复制的 Cookie 过期了。电商平台对搜索接口的频控非常敏感，同一 IP 短时间大量请求必然被盯上。

解决：先别急着换代码。降低任务频率，把线程数降到 2，单次请求间隔从 1 秒提高到 3 到 5 秒。然后重新打开浏览器登录一次，手动操作到搜索页，复制新的 Cookie，更新到配置里。如果仍然触发验证，就让任务休息半小时再继续。爬虫不是赛车，是马拉松。

注意：不要试图用任何手段绕过验证码。合规的做法是降速、错峰、人工介入。

5.2 CSS 选择器全部失效，页面结构改版

现象：凌晨定时任务跑完，数据库里商品数量为 0，日志显示解析器一个元素都没匹配到。

原因：平台前端发版，class 名带了新的随机后缀，或者 DOM 层级调整。你写死的div.product-list > div.item已经不存在了。

解决：选择器优先用稳定的属性，比如>def save_product_with_history(db, product: Product): price_row = PriceHistory( platform=product.platform, item_id=product.item_id, title=product.title, price=product.price, captured_at=product.captured_at, ) db.add(price_row) upsert_product(db, product) db.commit()

判断商品表是否真的发生了更新，可以对比price字段。如果价格没变，没必要写一条历史记录，否则一张表里全是噪音。加一行判断即可：

if orm and abs(orm.price - product.price) < 0.01: return

6.2 用 SQL 验证数据质量的几个查询

采集完先别急着做可视化界面，先用 SQL 自检。查出价格历史曲线，看有没有异常跳变：

SELECT captured_at, price FROM price_history WHERE platform = 'taobao' AND item_id = '123456789' ORDER BY captured_at;

正常的曲线应该平滑，如果前一天 199，后一天 49，先怀疑是自己把“到手价”和“原价”混存了。另一个自检是“重复采集率”：

SELECT COUNT(DISTINCT platform || item_id) AS unique_items, COUNT(*) AS total_rows FROM products;

两个数差太多，说明商品表里有脏数据。一般采集正常的话，total_rows会大于等于unique_items，因为同一商品可能被关键词重复命中。

6.3 分片键设计：为分布式爬虫留好路

日志表里有task_id，商品表里有item_id，这两个字段天然适合做分片。如果以后要扩展成分布式爬虫，最简单的方式是用item_id哈希对节点数取模：

shard = abs(hash(item_id)) % 4

这样同一个商品永远落在同一个节点处理，不会出现两台机器重复抓同一件商品。分布式爬虫不是跑得快，而是单机出问题时有冗余。配合crawl_tasks表，哪个节点挂了，任务重新入队就行。

爬虫数据流里还有一个被低估的环节是可视化。给products和price_history套一个简单的 Flask 页面，展示价格曲线和商品更新状态，这套采集体系才算真正闭环。Python 爬虫可视化界面不需要多复杂，一个图表组件加定时刷新就够用。

我做这个方向最深的教训是：第一版先把全部精力花在解析规则上，表结构只随手建了两张，结果一个月后数据重复、价格口径混乱，等于白跑。后来才悟到，表结构定的越早，返工越少。先定模型，再写解析器，最后调并发，这个顺序不能反。希望帮到你。

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