Python定向爬虫实现商品比价：同款识别与价格归一化实战

简介在网购比价需求日益增长的背景下基于Python与定向爬虫的商品比价系统毕业设计源码面向高校毕设选题、课程实训以及爬虫爱好者。压缩包共16个文件其中10个py源码文件构成主体覆盖GUI界面、爬虫抓取、数据库存取等模块另有3个pyc编译缓存、1个txt说明文件、1个db数据库文件以及1个md项目文档。py文件可直接阅读db为SQLite数据库md提供使用指引整体仅26KB轻量易读。该资源已有146人浏览学习。项目采用requests、BeautifulSoup等库完成定向抓取与页面解析数据入库后通过GUI呈现比价结果模块划分清晰从数据获取、存储、处理到展示形成完整闭环直接运行或二次修改均可行是一套可运行的完整参考。md文档帮助快速理解项目结构适合作为毕业设计参考或爬虫入门实践样例。1. 定向爬虫比价系统为什么说抓价格只是“入场券”很多毕设选题一眼望去是爬虫实际上挂在“比价”上。把商品价格从三五个站点抓下来一晚上就能跑通但答辩时老师问的往往是“你这几个平台的同一款商品是怎么匹配到一起的”“价格含运费吗”“昨天的价格和今天对比怎么算”——这些才是比价系统真正的工作量。基于 Python 和定向爬虫实现商品比价系统难点不在 requests 和 BeautifulSoup而在数据模型、同款归一化和价格口径统一。这套源码适合做电商数据方向毕设的同学也适合想快速搭一个价格监控脚本的从业者思路可以复用代码可以直接改。2. 先立数据模型再写爬虫比价系统的四张核心表与商品归一化2.1 为什么表结构决定比价效果我见过不少初版实现把抓下来的商品直接丢进一个大列表内存里比价程序一关数据全没。这种写法演示可以答辩一定被追问。比价系统最容易被问倒的两个问题一个是“怎么把不同平台同一商品对上”另一个是“最低价这个结论是怎么算出来的”。这两个问题都不属于爬虫而属于数据建模。正确的顺序是先把商品、平台、价格、快照之间的关系画清楚再回头写爬虫。比价系统的核心概念是 SPUStandard Product Unit标准商品单元一个 SPU 代表“iPhone 15 Pro Max 256GB 原色钛金属”这一个商品实体不同平台上的具体链接叫 SKU 或平台商品它们都指向同一个 SPU。价格不是商品的一个固定属性而是商品在不同时刻、不同平台上的一个观测值。把价格独立成快照表后续才能做历史最低价、涨价提醒、价格曲线。数据库选型上毕设场景我强烈建议 SQLite。理由很直接零配置、单文件、答辩现场不用连服务器把.db文件拷过去就能跑。MySQL 可以作为加分项写到文档里但主存储用 SQLite 不会减分反而显得你懂取舍。2.2 四张表设计与建表 SQL四张表分别是平台表、标准商品表、平台商品表、价格快照表。平台表最简单存平台名和首页 URL标准商品表存归一化之后的商品实体带品牌、型号、规格和标题指纹平台商品表存每个平台实际抓到的商品链接和当时的标题价格快照表只干一件事记录某平台商品在某时刻的价格。CREATE TABLE platform ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE, base_url TEXT NOT NULL ); CREATE TABLE product ( id INTEGER PRIMARY KEY AUTOINCREMENT, brand TEXT NOT NULL, model TEXT NOT NULL, spec TEXT DEFAULT , title_fingerprint TEXT NOT NULL UNIQUE, created_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE platform_item ( id INTEGER PRIMARY KEY AUTOINCREMENT, product_id INTEGER NOT NULL REFERENCES product(id), platform_id INTEGER NOT NULL REFERENCES platform(id), platform_item_id TEXT NOT NULL UNIQUE, title TEXT NOT NULL, url TEXT NOT NULL, fetch_count INTEGER DEFAULT 1, last_fetched_at TEXT DEFAULT (datetime(now, localtime)) ); CREATE TABLE price_snapshot ( id INTEGER PRIMARY KEY AUTOINCREMENT, platform_item_id INTEGER NOT NULL REFERENCES platform_item(id), price REAL NOT NULL, original_price REAL, shipping_fee REAL DEFAULT 0, stock_status TEXT DEFAULT unknown, promo_type TEXT DEFAULT normal, captured_at TEXT DEFAULT (datetime(now, localtime)) );这段 SQL 里有几个字段是反复踩坑后才意识到必须加的。title_fingerprint是归一化之后的标题哈希用来快速判断这个商品是否已经存在shipping_fee必须单独存很多平台价很低但运费高排序时要算到手价stock_status用来标记有货、无货、预售缺货的商品不应该参与最低价排序promo_type记录 “秒杀”“券后价”“普通价”同一平台同一商品在不同促销状态下价格可能差很多不标记就没法解释价格突变。2.3 商品标题清洗品牌库、型号归一化与规格抽取爬虫抓回来的原始标题非常脏比如 “Apple/苹果 iPhone 15 Pro Max (A3106) 256GB 原色钛金属 支持全网通5G”。同一个商品在另一个平台写成 “苹果iPhone15ProMax 256G 原色 5G手机”。人工一眼能看出是同一个程序不能。所以要做标题清洗把品牌、型号、规格拆开转成统一格式。import re import hashlib BRAND_ALIASES { apple: [apple, 苹果, iphone], huawei: [huawei, 华为], xiaomi: [xiaomi, 小米, 红米, redmi], } def clean_title(raw_title): text raw_title.lower() brand unknown for canonical, aliases in BRAND_ALIASES.items(): if any(alias in text for alias in aliases): brand canonical break model_match re.search(r(iphone\s?\d[\s\w]*?|mate\s?\d|xiaomi\s?\d), text) model model_match.group(1).replace( , ).lower() if model_match else unknown ram_match re.search(r(\d)\s?g\b, text) spec_parts [] if ram_match: spec_parts.append(ram_match.group(1) g) color_match re.search(r(黑色|白色|蓝色|原色|钛金属|暗夜紫|星光色), text) if color_match: spec_parts.append(color_match.group(1)) spec _.join(spec_parts) fingerprint hashlib.md5(f{brand}|{model}|{spec}.encode()).hexdigest() return { brand: brand, model: model, spec: spec, fingerprint: fingerprint, }清洗函数的输出不只是给人看的fingerprint直接作为product.title_fingerprint的唯一键入库前先查一遍存在就说明这个 SPU 已经建过直接复用product_id。注意正则里的\s?处理 “iphone15” 和 “iphone 15” 两种写法replace( , )再把型号内部空格去掉。颜色词用了|正则后续加颜色直接往列表里加词就行。这里有个不足要提前说品牌别名表是手写的扩展到多品牌时维护成本高可以用外部词典文件但毕设规模手写完全够。2.4 价格快照让“最低价”从瞬时值变成可计算指标如果没有快照表“当前最低价”只能是当前这批抓取结果里的最小值老师一问“昨天最低多少”就答不上来。有了price_snapshot每次采集都追加一行比价系统就能回答三个关键问题当前谁最低、历史最低是多少、某个商品最近七天价格走势。写入快照时要注意幂等。同一个平台商品同一轮采集可能因为重试被抓两次要按platform_item_id captured_at做去重或者简单点同一轮采集只写一次。增量采集时会自然形成多条记录查询最新价用WHERE captured_at (SELECT MAX(captured_at) ...)或者直接按captured_at DESC LIMIT 1。这个查法慢但数据量在万级以下时无所谓毕设不追求性能追求逻辑正确。3. 定向爬虫的最小可运行实现requests BeautifulSoup 从列表页到详情页3.1 定向爬虫的“定向”到底锁定什么通用爬虫对着全网爬定向爬虫只针对目标站点和商品类型。定向体现在三个维度域名白名单、URL 模式、采集字段。域名白名单最简单只允许example.com和example.net两个域名的请求URL 模式指商品详情页长什么样比如路径里包含/item/或/product/才解析列表页包含/search/才解析采集字段固定为标题、价格、运费、库存、促销标记不碰用户信息和无关页面。限定的好处是解析逻辑可以写死。列表页的选择器在同一个站点内基本一致详情页的价格字段通常也一致坏处是目标站点改版后解析会失效所以定向爬虫必须把解析器集中放在一个模块里改版时只改一处。3.2 列表页爬虫构建请求与商品链接提取下面用requests加BeautifulSoup抓列表页提取商品详情页链接。以某 3C 垂直电商的搜索页为例实际使用时把LIST_URL和选择器换成目标站点的值。import requests import time import random from bs4 import BeautifulSoup HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36, Accept: text/html,application/xhtmlxml,application/xml;q0.9,*/*;q0.8, Accept-Language: zh-CN,zh;q0.9, } LIST_URL https://www.example.com/search?qiphone15promaxpage{page} ITEM_PATH_PATTERN /item/ def fetch_page(session, url, retries3): for attempt in range(retries): try: resp session.get(url, headersHEADERS, timeout10) if resp.status_code 200: return resp.text except requests.RequestException as exc: print(fattempt {attempt 1} failed: {exc}) time.sleep(2 * attempt 1) return None def parse_list(html): soup BeautifulSoup(html, html.parser) links [] for a in soup.select(a[href]): href a[href] if ITEM_PATH_PATTERN in href and href not in links: links.append(href) return links session requests.Session() for page in range(1, 4): html fetch_page(session, LIST_URL.format(pagepage)) if not html: continue for link in parse_list(html): print(link) time.sleep(random.uniform(1.5, 3.0))fetch_page里的重试是必须的网络超时和 5xx 是常态不重试爬虫跑一半就停了。重试间隔用2 * attempt 1第一次失败等 1 秒第二次等 3 秒第三次等 5 秒避免失败后立刻高频重试被服务端留意。解析列表页不要用正则抠 hrefBeautifulSoup的select(a[href])更可靠过滤条件ITEM_PATH_PATTERN是定向的体现把非商品页面的链接全部挡掉。time.sleep(random.uniform(1.5, 3.0))是给页面间隔的随机延时单位是秒列表页之间稍微松一点。3.3 详情页价格解析优先挖 JSON 而不是 HTML商品详情页的现代实现普遍把价格放在内嵌的 JSON 里比如window.__INITIAL_STATE__或window.__NUXT__HTML 标签里的价格反而可能是渲染前占位符。常见做法是先用正则把 JSON 块抠出来再用json.loads解析最后才用 BeautifulSoup 找价格标签。import json import re def extract_json_state(html, patternrwindow\.__INITIAL_STATE__\s*\s*(\{.*?\});): match re.search(pattern, html, re.DOTALL) if not match: return {} try: return json.loads(match.group(1)) except json.JSONDecodeError: return {} def parse_price_from_state(state): price None try: price state[goods][price] except (KeyError, TypeError): pass if price is None: try: price state[product][promotion][price] except (KeyError, TypeError): pass return price html fetch_page(session, https://www.example.com/item/123456.html) state extract_json_state(html) price parse_price_from_state(state) if price is None: soup BeautifulSoup(html, html.parser) price_node soup.select_one(.current-price) price price_node.text.strip() if price_node else None print(fprice: {price})JSON 解析有两个坑。第一正则里的\{.*?\}非贪婪匹配在 JSON 内部有嵌套花括号时会截断大概率解析失败所以extract_json_state返回空字典后要兜底走 HTML 解析不要一条路走到黑。第二价格字段在不同站点命名完全不同有的叫price有的叫salePrice还有的叫promotionPriceparse_price_from_state里用try...except KeyError逐个尝试是土办法但有效。HTML 兜底时选择器尽量选带current或price语义的 class直接抓页面里第一个价格数字很容易抓到原价而不是到手价。3.4 限速、UA、Cookie 与失败重试必须当成功能写新写的爬虫第一版容易跑得太猛并发请求一串接一串结果列表页抓了十几页就开始出现验证码或者 403。这些反爬机制只需要用工程手段就能大概率绕开前提是别把 UA、Cookie、限速当成可有可无的参数。class RateLimitedSession: def __init__(self, min_interval1.2, max_interval2.8): self.session requests.Session() self.min_interval min_interval self.max_interval max_interval self.last_request_at 0 def get(self, url, **kwargs): elapsed time.time() - self.last_request_at if elapsed self.min_interval: time.sleep(random.uniform(self.min_interval, self.max_interval)) resp self.session.get(url, headersHEADERS, timeout10, **kwargs) self.last_request_at time.time() return respRateLimitedSession把限速逻辑收口在请求入口爬虫业务代码里不需要到处写sleep。min_interval1.2表示两次请求之间至少间隔 1.2 秒max_interval2.8是随机上限模拟人工操作节奏。Cookie 的处理方式是先用浏览器登录一次目标站点把Cookie字符串放进HEADERS里或者用session.cookies.set注入。验证码出现时不要硬刚记录日志后跳过该商品下次增量采集再补。把 UA、Cookie、限速、重试统一封装成 session是定向爬虫工程化的第一步。3.5 增量采集用数据库去重而不是爬一遍全量商品比价系统不是跑一次就结束的至少需要每天跑一次。全量重复爬意味着每次都要重新解析所有商品浪费带宽和时间增量采集则只抓新的商品链接和价格变化。关键是每次详情页采集完后先查platform_item.platform_item_id是否已存在存在就只追加price_snapshot不存在才新建platform_item。def upsert_platform_item(cursor, product_id, platform_id, item_id, title, url): cursor.execute( SELECT id FROM platform_item WHERE platform_item_id ?, (item_id,), ) row cursor.fetchone() if row: cursor.execute( UPDATE platform_item SET fetch_count fetch_count 1, last_fetched_at datetime(now, localtime) WHERE id ?, (row[0],), ) return row[0] cursor.execute( INSERT INTO platform_item (product_id, platform_id, platform_item_id, title, url) VALUES (?, ?, ?, ?, ?), (product_id, platform_id, item_id, title, url), ) return cursor.lastrowid增量采集有一个反直觉的收益它顺便解决了“比价结果过期”的问题。只要每天定时跑price_snapshot里就会积累多条记录比价结果里的“历史最低价”字段就有数据可算。上面函数里fetch_count 1看起来不起眼但它记录了每个商品被采集的次数用来判断哪些商品持续抓不到有效价格比如连续三次price is None就应该把stock_status标记为unknown并从比价结果里降权。4. 比价核心逻辑同款识别、价格口径统一与最低价排序4.1 比价系统真正的“核心”不在爬虫而在匹配层能爬到价格只证明你写了采集代码能把不同平台的“iPhone 15 Pro Max 256GB 原色”识别成同一个商品才证明你理解了比价。原始标题匹配在真实数据上几乎不可用因为平台 A 写“苹果 iPhone15ProMax 256G”平台 B 写“Apple iPhone 15 Pro Max (A3106) 256GB 原色钛金属”字符串比较直接失败。所以同款识别的第一步是复用第 2 章的clean_title把标题转成品牌、型号、规格三段再用三段拼接的 fingerprint 去匹配product表。4.2 三级匹配品牌 型号 规格给出可落地的匹配函数这里用一个三级打分函数把匹配从“是或否”变成“得分排序”结果更可靠。def match_product_score(raw_title, product): cleaned clean_title(raw_title) score 0 if cleaned[brand] product[brand]: score 50 if cleaned[model] product[model]: score 40 elif cleaned[model] ! unknown and product[model] ! unknown: score 10 # 型号都识别出来了但不相等保留低分 if cleaned[spec] and cleaned[spec] product[spec]: score 10 return score三级匹配的逻辑是品牌 50 分、型号 40 分、规格 10 分。品牌错匹配一定是不同商品直接排除品牌对、型号对得 90 分基本可以判定同款品牌对、型号没识别出来但规格一致得 60 分需要人工确认。实际使用中建议设一个阈值 85 分才自动匹配低于阈值进人工复核队列。这里故意不直接用字符串哈希是因为标题清洗后的 model 和 spec 可能有多种写法比如 “15promax” 和 “15pro max”正则清洗已经保证了一部分但人工复核队列仍然要做因为总有平台会写“iPhone 15 Pro Max 钛金属”漏掉“256GB”。4.3 价格归一化到手价、运费、缺货状态参与排序价格不是抓到一个数字就结束。同一个商品平台 A 标价 8999 元包邮平台 B 标价 8899 元加 30 元运费平台 C 标价 8799 元但显示“无货”。只看价格数字平台 C 最低算到手价平台 B 最低。所以排序比的一定是“到手价”price shipping_fee而且缺货商品默认排到最后。def normalize_price(snapshot): if snapshot[stock_status] in (out_of_stock, unknown): return None final_price snapshot[price] original_price snapshot.get(original_price) or snapshot[price] if snapshot[promo_type] coupon: final_price min(final_price, snapshot.get(promo_price, final_price)) final_price round(final_price snapshot.get(shipping_fee, 0), 2) return { final_price: final_price, original_price: original_price, discount_ratio: round(final_price / original_price, 2) if original_price else 1.0, }normalize_price做了三件事缺货商品返回None不参与排序促销价取券后价和标价里的较小值运费合并进最终价格并算出折扣率。排序时用final_price ASCNone值排最后然后取第一条就是当前最优价格。这里有一个参数值得注意shipping_fee如果每次都从详情页抓不到默认填 0那就不叫“到手价”了。常见做法是详情页解析时尽量抓“运费”模块实在抓不到就按平台默认运费配置填配置写在config.ini里default_shipping_fee_per_platform按平台分别设置。4.4 用 Flask 把比价结果交出去JSON API 与最小页面爬虫、清洗、比价逻辑跑通后需要一个出口给答辩老师看。比价系统常见做法是接一个轻量 Web 层Flask 足够。核心接口就一个查询某关键词的比价结果。from flask import Flask, request, jsonify import sqlite3 app Flask(__name__) DB_PATH compare.db def query_compare(brand_model_keyword): conn sqlite3.connect(DB_PATH) conn.row_factory sqlite3.Row rows conn.execute( SELECT p.brand, p.model, p.spec, pi.title, pi.url, ps.price, ps.shipping_fee, ps.stock_status, ps.captured_at FROM price_snapshot ps JOIN platform_item pi ON ps.platform_item_id pi.id JOIN product p ON pi.product_id p.id WHERE p.model LIKE ? OR p.title_fingerprint IN (SELECT title_fingerprint FROM product WHERE model LIKE ?) ORDER BY ps.captured_at DESC , (f%{brand_model_keyword}%, f%{brand_model_keyword}%), ) # 这里只保留每个 product_id 最新一条快照模拟实时比价 latest {} for row in rows: key row[title] latest.setdefault(key, dict(row)) conn.close() return list(latest.values()) app.route(/compare) def compare(): q request.args.get(q, ) if not q: return jsonify({error: missing q}), 400 results query_compare(q) return jsonify({query: q, count: len(results), items: results}) if __name__ __main__: app.run(host127.0.0.1, port8000, debugTrue)query_compare里的 SQL 先按model LIKE过滤再按快照时间倒序最后在 Python 里用setdefault按标题取最新一条。这个实现不追求性能但逻辑直观同一商品多平台多时间点直接展示最新一轮结果。app.run(debugTrue)在答辩演示时可以开着出错能看到堆栈正式部署时一定要换成debugFalse。这个 JSON 接口可以直接对接一个极简的 HTML 页面用fetch(/compare?qiphone)展示表格代码量很小但“从爬虫到比价到展示”的闭环就完整了。5. 源码包交付避坑环境依赖、路径问题与爬虫翻车的五个高发点5.1 源码.zip 解压后跑不起来环境与依赖是翻车第一来源现象答辩老师在你电脑上双击run.py直接报ModuleNotFoundError: No module named bs4或者requests找不到。原因源码包里没有requirements.txt或者写了但没固定版本号。换一台电脑环境依赖版本不一致就崩。解决在源码根目录放一个requirements.txt固定主要依赖版本。Flask3.0.0、requests2.31.0、beautifulsoup44.12.2。然后在README.md开头写清楚安装步骤python -m venv venv、venv\Scripts\activateWindows或source venv/bin/activatemacOS/Linux、pip install -r requirements.txt。另外把数据库初始化的脚本和爬虫入口分开init_db.py负责建表run.py只负责启动爬虫和 Web 服务避免跑一次全流程就要重新建表。这个点是源码交付的第一优先级比功能完整还重要。5.2 采集一会儿就被拦截UA、频率与 Cookie 三件套现象爬虫前 50 个商品正常第 51 个开始返回 403 或验证码页面解析器拿不到价格。原因目标站点对同一会话的请求频率做了限制。只设置了基础 UA没有 Cookie或者限速间隔设为固定 0.5 秒被识别为脚本行为。解决第 3 章的RateLimitedSession直接用上间隔调大到min_interval1.5, max_interval3.5在HEADERS里补上Cookie和RefererReferer 填列表页地址能显著降低被拦概率。如果遇到验证码唯一的可靠处理是跳过当前商品并记录日志不要写自动识别验证码的逻辑验证码识别本身就是一个大坑投入产出比极低。采集被拦这件事频率是第一大主因Cookie 是第二大主因都排查完还翻车再考虑是否有更复杂的风控逻辑。5.3 同款识别把不同商品算成同一款现象搜索“iPhone 15 Pro”时比价结果里出现了 iPhone 15 Pro 和 iPhone 15 Pro Max 混在一起或者 128GB 和 256GB 版本被当成同一款。原因clean_title里的型号正则没有提取到完整的“Max”后缀或者规格提取只抓了 RAM 没抓存储容量导致 fingerprint 相同。解决在clean_title里增加规格字段的提取强度。规格不能只抓一个\dg要拆成 RAM 和 存储两部分re.search(r(\d)\s?g(?:b|内存), text)识别 8GB、12GB 内存另外单独匹配256g、512g作为存储。最终 fingerprint 参与拼接的字段从brand|model|spec改成brand|model|ram|storage。还有一个土办法但血泪经验值得用匹配完成后把同款商品标题两两显示在控制台里人工扫一眼这个动作叫“抽检”每次批量导入后必须做。5.4 比价结果过期增量逻辑缺失现象演示时对比的平台价格是上周的页面展示“历史最低价 7999”但当前所有平台价格已经是 8299。原因采集没有定时任务也没有增量更新。跑一次就完事的数据时间一长必然失真。解决给爬虫加一个调度入口用apscheduler写一个每天凌晨 2 点执行的定时任务。采集前先查platform_item.last_fetched_at超过 24 小时的才重新抓没超过的直接复用旧数据。比价查询时强制加时间条件只在最近 24 小时的快照里比价这样即使定时任务没跑也不会拿七天前的价格出来当真。这个时间窗口参数放config.ini里max_snapshot_age_hours 24答辩时只要改这一个参数就能演示“历史价格失效”的处理逻辑。5.5 答辩前必做的自测清单清单里第一项是“删库重跑”。把.db文件删掉从init_db.py开始完整跑一遍流程确认不依赖任何残留数据。第二项是“断网重试”拔网线后跑爬虫确认重试逻辑不会死循环而是打印日志后优雅退出。第三项是“空关键词查询”Web 接口输入空参数返回 400 而不是 500。第四项是“价格乱序”手动把数据库里某个价格改成异常值确认排序和折扣率不会出现负数或超低折扣。第五项是“换一台电脑跑”用同学的电脑或虚拟机按README.md从头部署一次。全部通过源码包才算交付合格。6. 把比价系统做厚定时采集、历史价格曲线与结果可验证最后一公里是让比价系统自己跑起来。写一个scheduler.py用apscheduler的BlockingScheduler挂两个任务每 6 小时采集一次目标商品每小时重新计算一次比价缓存。采集任务先查config.ini里的关键词列表逐个调用第 3 章爬虫Web 接口改成读缓存表不再实时查快照表这样页面响应快也不会在演示时触发新的网络请求。历史价格曲线建议在 Web 层加一个/history?product_id1接口返回该商品最近 30 天的price_snapshot序列前端用轻量图表库画折线。这个功能本身代码量不大但对答辩效果提升很大图中价格曲线一出来比价系统才真正有“数据系统”的样子而不是一个有界面的爬虫脚本。验证方法很直接手动修改数据库里某条price_snapshot的价格刷新历史曲线立刻看到变化再改回原价曲线出现一个“毛刺”这就验证了链路是通的。我当初做这类系统时吃过一个教训把采集、清洗、比价、展示全写在一个main.py里看着跑通了后面加定时任务时到处改函数签名改到崩溃。后来拆成spider/、core/、web/、db/四个包每个包只干一件事所有问题都好查了。源码.zip 交付的不只是跑得起来的代码更是这种“可拆分、可维护”的工程习惯——把这个习惯守住比多抓几个网站更有价值。希望帮到你。本文还有配套的精品资源点击获取