企业官网建设流程全解析

把蹲点抢购这事儿交给Python，是我最近半年做得最值的一个决定。

事情是这样的：我一直在关注几款数码产品和家电的价格，想等一个合适的入手时机。但手动刷新页面太累了，而且很多商品是半夜改价，等你早上看到，便宜的时候早就过去了。后来我干脆写了一个Python脚本，通过淘宝关键词API定期搜索这些商品的实时价格，把价格数据存下来，再设置一个心理阈值，价格一旦低于预期，就自动推送提醒到我的钉钉。这套逻辑跑通之后，我基本不用再盯页面了，该干嘛干嘛，等到手机弹通知，再去看一眼下单就行。

这篇文章就把这套方案的完整思路和代码实现拆开讲一遍。核心就三件事：怎么通过淘宝关键词API去搜商品、怎么把价格数据存成历史记录、怎么根据价格变化触发预警通知。不管是想学Python接口调用的新手，还是想做个竞品价格监控的电商运营，都可以直接照着抄。

1. 为什么我决定用Python做价格监控

1.1 手动盯价的痛点，想必你也遇到过

先说一个很常见的场景。你购物车里躺着一件商品，比如一台两千多的显示器。你知道它历史低价在1800左右，但你没时间天天看。于是你每天中午打开App看一眼，结果连续一周都是原价。某天晚上你加班到十一点，回家刷手机，发现这件显示器搞限时秒杀，只要1799，但活动只剩半小时——你一边付款一边骂自己为什么没早看。

这种经历我反复遇到，而且我发现电商平台上"先涨价再降价"的玩法太普遍了。页面显示"限时特惠"四个字，你以为捡了大便宜，实际上比上个月的基本盘还贵。这时候最需要的就是历史价格数据：这个商品过去三个月到底卖多少钱，当前价格处于什么位置。没有数据，你所谓的"便宜"和"贵"都是凭感觉。

所以我的需求变得很明确：第一，自动搜索指定关键词下的商品；第二，定期记录每个商品的价格、销量、店铺信息；第三，当价格满足某个条件（低于心理价、低于历史最低价、降价幅度超过一定比例）时，给我发一个通知。注意，这个方案不是为了薅羊毛抢限时补贴，而是建立一个长期观察商品价格的自动化机制，弥补人工盯价的各种盲区。

1.2 爬虫和API，我为什么选了后者

市面上的技术方案其实就两条路：用爬虫直接抓淘宝页面，或者调用淘宝开放API。

爬虫这条路，我最早也试过。直接用requests请求商品搜索页，能拿到HTML源码，但问题一堆。淘宝的反爬机制升级很频繁，初次访问可能正常，翻几页之后就要求滑块验证，再往后IP会被临时限制。你就算用上Selenium模拟浏览器，也扛不住登录态和验证码的连环折腾。就算你费老大劲把数据抓下来了，页面结构一改版，解析代码当场作废，这种方案太脆了。

API方案就稳得多。淘宝开放平台以及第三方数据服务商提供了标准化的接口，比如通过关键词搜索商品、获取商品详情、解析优惠信息。你只要带上合法的app_key和app_secret，按照文档拼接参数、生成签名，发起HTTP请求就能拿到结构化的JSON数据。接口返回的字段是固定的，不会动不动就变，也没有验证码的烦恼。稳定性和开发效率都远超爬虫。

这两种方案的对比如下表：

我最终选择的是第三方数据服务商提供的淘宝关键词搜索API。这里需要说明一下：淘宝开放平台对个人开发者申请商品搜索类接口的门槛比较高，而市面上一些正规的数据服务商把这类能力做成了标准化的付费API，个人开发者可以比较方便地申请使用。这类服务本质上还是围绕淘宝商品数据做整合，前提是你自己要有合法的应用场景。下面所有代码我以这类通用API为例来写。

2. 动手前的准备：环境、依赖与API申请

2.1 Python环境与依赖

这个项目对Python版本没有太苛刻的要求，3.8以上都行。我自己用的是3.10，Windows上开发的，后来部署到一台Linux小主机上跑，代码完全没改。

如果你是刚接触Python，建议先装好Python解释器，再把pip的国内镜像配置一下，不然装包速度很痛苦。Windows用户在命令行执行：

python -m pip install -i https://pypi.tuna.tsinghua.edu.cn/simple requests apscheduler pandas

Linux下如果系统自带的Python版本偏低，可以先升级或者用apt安装新版，再执行同样的命令。依赖方面，核心就三个库：requests（发HTTP请求）、apscheduler（定时调度）、pandas（可选，做价格数据分析时方便）。数据库我直接用Python内置的sqlite3，连第三方库都不用装。

另外强烈建议用虚拟环境，尤其是你机器上还有别的Python项目时。我用的是venv：

python -m venv price_env # Windows激活 price_env\Scripts\activate # Linux激活 source price_env/bin/activate

跑在独立环境里，避免依赖冲突，这是最基本的工程习惯。别省这一步，以后有你省心的时候。

2.2 淘宝关键词API的申请与鉴权

市面上提供淘宝关键词API的服务商有好几家，申请流程大同小异。一般是：注册账号、完成企业或者个人实名认证、在控制台创建一个应用，然后系统会给你一对密钥——app_key和app_secret。

app_key相当于你的身份标识，app_secret是签名密钥，绝对不能泄露到前端或者公开仓库里。如果你打算把代码发布到GitHub，请一定把密钥放到config文件里，然后加入.gitignore。

有了这对密钥之后，调用API时还需要传入一个重点参数：方法名。以关键词搜索商品为例，方法名一般是item_search。接口的通用请求参数包括：

• app_key：应用标识
• method：方法名，这里就是item_search
• keyword：搜索关键词，比如"无线机械键盘"
• page：页码
• page_size：每页返回条数，有些服务商限制单页最多40
• sign：签名串，用于身份校验

签名算法是所有第三方API的通用逻辑：把除sign以外的所有请求参数按参数名字母升序排列，拼成key1value1key2value2的形式，然后拼接上app_secret，最后做MD5，得到的字符串就是签名。以"q=机械键盘"和"page=1"两个参数为例，拼接逻辑就是先把参数按字母排好，再和密钥一起做MD5摘要。

给你一段通用的签名函数代码，几乎可以直接移植到任何一家API上：

import hashlib import requests APP_KEY = "你的app_key" APP_SECRET = "你的app_secret" API_URL = "https://api.xxx.com/api" # 以实际服务商为准 def generate_sign(params: dict) -> str: # 1. 过滤掉空值 filtered = {k: v for k, v in params.items() if v != "" and v is not None} # 2. 按key字母升序 sorted_keys = sorted(filtered.keys()) # 3. 拼接成key1value1key2value2 plain_text = "".join(f"{k}{filtered[k]}" for k in sorted_keys) # 4. 拼接app_secret后再做MD5 plain_text = plain_text + APP_SECRET return hashlib.md5(plain_text.encode("utf-8")).hexdigest().upper()

注意签名里最后做了个大写处理，这是因为有些服务商要求签名全部大写。如果实际对接过程中被告知"sign error"，大概率是大小写问题或者参数漏传了。另外，app_secret串到待签名的字符串里时不需要URL编码，这一点比较容易踩坑。

2.3 请求封装与通用返回结构

所有接口的调用方式都是一样的，无非是换方法名和参数。所以我习惯把请求逻辑封装成一个统一函数，这样后面不管是搜索商品，还是获取商品详情，都直接复用同一套签名和请求逻辑。

def call_taobao_api(method: str, params: dict) -> dict: request_params = { "app_key": APP_KEY, "method": method, **params } request_params["sign"] = generate_sign(request_params) resp = requests.get(API_URL, params=request_params, timeout=10) resp.raise_for_status() result = resp.json() if result.get("code") != 200: raise Exception(f"API error: {result.get('code')} {result.get('msg')}") return result["data"] if isinstance(result, dict) else result

注意这里有一个比较实用的细节：有些数据服务商的返回字段名有差异，有的叫price，有的叫current_price，有的带了原始促销价。你在初期联调时，最好先把返回的JSON完整打印一遍，看清楚字段再写解析逻辑，不要想当然。

3. 核心代码：搜索商品、解析价格与入库

3.1 数据模型设计：既要存当前价，也要存历史价

有了API能力之后，下一步就是设计数据怎么存。我的方案是两张表：商品表和价格历史表。

商品表存的是商品的相对静态信息，比如标题、商品ID、店铺名称、商品链接，这些字段不会频繁变化，重复搜索同一商品时直接做去重。价格历史表存的是每次抓取到的价格、原价、销量和时间点，一个商品在历史表中会有多行记录，代表它在不同时间点的价格快照。

这两张表拆开是很有必要的。如果只建一张表，每插入一条新价格记录就要重复存一遍商品标题和链接，浪费空间不说，以后想分析"某个价格区间的商品中销量最高的有哪些"这种问题，SQL写起来也会很繁琐。

对应的建表语句如下：

CREATE TABLE IF NOT EXISTS product ( item_id TEXT PRIMARY KEY, title TEXT, shop_name TEXT, url TEXT, first_seen_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS price_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, item_id TEXT, price REAL, orginal_price REAL, sales INTEGER, fetched_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP );

price字段之所以用REAL，严格来说应该用DECIMAL避免浮点误差，但SQLite对DECIMAL的支持比较弱，所以我在Python层统一用Decimal做计算，入库时转成float，查询做比较时也走Decimal，这样双保险。

3.2 调用关键词API拉取商品列表

接下来是核心的拉取逻辑。我定义一个函数，入参是关键词，循环拉取前几页数据，把每件商品的详情写入数据库。

import time import sqlite3 def scrape_keyword(keyword: str, max_pages: int = 3): conn = sqlite3.connect("price_monitor.db") for page in range(1, max_pages + 1): data = call_taobao_api("item_search", { "keyword": keyword, "page": page, "page_size": 40 }) items = data.get("items", []) if not items: break for item in items: item_id = item.get("num_iid") or item.get("item_id") if not item_id: continue title = item.get("title", "") price = item.get("price") or item.get("current_price") orginal_price = item.get("orginal_price") or item.get("original_price") or price sales = int(item.get("sales") or 0) shop_name = item.get("shop_name", "") url = item.get("detail_url") or f"https://item.taobao.com/item.htm?id={item_id}" conn.execute( "INSERT OR IGNORE INTO product (item_id, title, shop_name, url) VALUES (?, ?, ?, ?)", (item_id, title, shop_name, url) ) conn.execute( "INSERT INTO price_history (item_id, price, orginal_price, sales) VALUES (?, ?, ?, ?)", (item_id, float(price), float(orginal_price), sales) ) conn.commit() print(f"[{keyword}] 第{page}页完成，累计{len(items)}条") time.sleep(1) # 控制请求频率，防止限流 conn.close()

有几个很关键的处理细节我必须说明。

第一，price和orginal_price字段名在每个服务商那里不一样，我用or逐个尝试。这是长期踩坑踩出来的经验，不要假设一个字段一定存在，也别假设它是字符串还是数字。

第二，sales直接int转换前先判断是否为None，有些非热销商品可能没有销量字段。同理，如果接口返回的是"已售100+"这种带文字的量级，需要自己写正则把数字提取出来。

第三，INSERT OR IGNORE是SQLite的去重写法，第一次见到这个商品就写入product表，price_history表则每轮都插入一条。这样同一件商品在历史表里有多条价格快照，画趋势图、算最低价都靠它。

3.3 解析返回数据与价格清洗

你以为拿到price字段就完事了？天真。实际工作中我发现几个很坑的问题。

第一个坑是价格类型不统一。有的API返回的是字符串"299.00"，有的直接是数字299，有的更过分，返回一个列表，比如"299.00-399.00"，表示这个商品有多 SKU，价格是个区间。遇到区间价格，如果直接float转换会直接报错。我的处理方式是：如果返回的是字符串且包含"-"或者"~"，取最低价，因为做价格监控关心的是入手门槛，取最低区间更符合场景。

第二个坑是"到手价"和"标价"的区别。接口返回的price通常是页面标价，但实际成交时可能有店铺优惠券、跨店满减、淘金币抵扣，最终到手价可能比标价低很多。有些API服务商会提供coupon_price或者promotion_price字段，如果有，优先级要高于price。我在代码里这样处理：

def parse_price(item: dict) -> tuple: raw_price = item.get("coupon_price") or item.get("promotion_price") or item.get("price") raw_orginal = item.get("orginal_price") or item.get("original_price") or raw_price def to_float(raw): if raw is None: return 0.0 if isinstance(raw, (int, float)): return float(raw) if isinstance(raw, str): raw = raw.strip().replace("元", "").replace("¥", "") if "-" in raw or "~" in raw: raw = re.split(r"[-~至]", raw)[0] try: return float(raw) except ValueError: return 0.0 return 0.0 return to_float(raw_price), to_float(raw_orginal)

配合上3.2节里面的字段兜底逻辑，这套解析函数到目前为止还没在哪个商品上报过错。

第三个坑是商品ID的类型。淘宝的商品ID（num_iid）是纯数字，但位数很多，用int类型存没问题，但如果用JSON传输后再解析，可能会被某些JSON库弄成科学计数法字符串。所以存product表时，item_id字段我用的是TEXT类型，而不是INTEGER。写SQL建表时一定要写成TEXT，不然你会看到很多商品ID变成了"1.23457e+18"这种鬼东西，后悔都来不及。

4. 价格预警机制：阈值、定时任务与通知推送

4.1 基于历史价格的动态阈值策略

数据存下来了，接下来最关键的就是怎么判断"该出手了"。我这里实现了两套预警策略，你可以根据自己的习惯选择或者叠加。

第一套是绝对阈值，最简单暴力。比如我给自己设定，无线键盘降到399以下就提醒。这个阈值写死在配置里，价格一旦低于阈值就直接触发通知。适合对目标商品的心理价位比较明确的人。

第二套是动态阈值，基于历史价格判断。核心逻辑：如果当前价格低于该商品历史最低价的3%以上，或者低于近30天平均价格的8%以上，就说明现在的价格已经处于一个相对低位，可以考虑入手。这个策略的好处是不用你自己定具体数值，系统帮你算。

动态阈值的本质是识别价格的相对位置。我举个例子：一件商品旺季卖699，淡季卖529，历史最低是499。你如果设绝对阈值500，可能永远等不到，因为499只出现过一次，且是秒杀价。但动态阈值会告诉你，当前529相对历史均值已经下降了20%，这时候就值得关注了。用一段SQL就能算出来：

SELECT item_id, MIN(price) AS min_price, AVG(price) AS avg_price_30d FROM price_history WHERE fetched_at >= datetime('now', '-30 days') GROUP BY item_id;

拿到这个结果后，在Python里和当前价做比较，低于阈值就进入预警流程。

4.2 APScheduler定时调度：每30分钟跑一轮

做完拉取逻辑和判断逻辑之后，需要让它定时自动化跑起来。这里我用了APScheduler，而不是自己写while True + sleep。原因很简单：APScheduler支持cron表达式，可以精确控制每天的运行时间，而且自带异常处理，单个任务崩了不会拖垮整个调度器。

我设置了两个任务：一个是整点拉取任务，每30分钟执行一次；另一个是每天早上9点整合前一天的监控数据，生成一份汇总报告。

from apscheduler.schedulers.blocking import BlockingScheduler def job_fetch_and_check(): try: for keyword in WATCH_KEYWORDS: scrape_keyword(keyword, max_pages=2) check_and_notify() except Exception as e: print(f"[任务失败] {e}") # 这里应该接入日志，别只print scheduler = BlockingScheduler() scheduler.add_job(job_fetch_and_check, "cron", minute="*/30", id="fetch_job") scheduler.start()

注意cron表达式里minute="*/30"表示每30分钟执行一次，但这里的实现是从凌晨0点开始每30分钟一次，也就是0分和30分。如果你希望是每小时的第15分和第45分跑，就写成minute="15,45"，这样更符合实际业务节奏。

还有一点要强调：BlockingScheduler会阻塞主线程，所以你需要把它放在单独的进程里运行，或者索性像我一样，放到Linux的systemd服务里。我是用nohup python main.py &直接挂着跑的，简单粗暴，日志重定向到文件就行。如果你用的是Windows，可以考虑创建计划任务。

4.3 预警通知接入钉钉/邮件

预警信息要能及时触达到你手上，否则系统跑得再勤也没用。我这里提供了两种通知方式：钉钉机器人和SMTP邮件。

钉钉机器人的成本最低，配置也最简单。你在钉钉群里添加一个自定义机器人，拿到webhook地址后，用requests直接POST一段JSON就能推送消息：

def send_dingtalk(message: str): webhook = "你的钉钉机器人webhook地址" headers = {"Content-Type": "application/json"} payload = { "msgtype": "text", "text": {"content": message} } requests.post(webhook, json=payload, headers=headers, timeout=5)

我实际运行下来，钉钉机器人的延迟基本在1秒以内，消息到达率很高。唯一要注意的是，钉钉机器人对每分钟消息条数有限制（大概20条），价格监控这个场景完全够用了。

邮件的话，用smtplib发即可。我选择的是QQ邮箱的SMTP服务，开启授权码后填到配置里。邮件的好处是排版可以做得更丰富，可以把当天所有降价商品列成表格一起发。但如果只为了即时性，钉钉或者企业微信的webhook体验更好。

import smtplib from email.mime.text import MIMEText def send_email(subject: str, html_content: str): sender = "你的邮箱" password = "你的SMTP授权码" receiver = "接收邮箱" msg = MIMEText(html_content, "html", "utf-8") msg["Subject"] = subject msg["From"] = sender msg["To"] = receiver with smtplib.SMTP_SSL("smtp.qq.com", 465, timeout=10) as server: server.login(sender, password) server.sendmail(sender, [receiver], msg.as_string())

我个人现在的做法是：实时预警走钉钉，每天早上9点再汇总一封邮件，把昨日价格波动明显的商品列表发出来。流量不大成本也低，一个月的API调用费用控制在几块钱到几十块钱之内，信息密度却比堆在消息群里高很多。

5. 常见问题与排查技巧实录

5.1 高频报错排查速查表

这套方案跑起来之后，你大概率会遇到下面几个问题。我做了个速查表，按我踩坑的频率从高到低排的。

5.2 那些API文档里不会写的事

第一件事，别在整点和大促节点跑定时任务。我一开始设置的是每天0点、6点、12点、18点拉取价格，结果发现整点的时候，接口延迟从平时的300毫秒飙到两三秒，偶发超时。后来我把拉取时间调整到每小时的第23分和第53分，避开整点高峰，稳定了很多。促销节点比如双11当天也一样，十点开始，十点整点绝对是最卡的。

第二件事，一个大词下面会有很多不相关的商品。比如你搜"键盘"，出来的结果除了电脑键盘，还有门锁的键盘、钢琴的键盘。所以关键词要尽量具体到型号，比如"MX Keys机械版"肯定比"键盘"准确。另外，同一个型号可能有官方店、第三方店、个人闲置转卖等多个商品，最好在判断逻辑里加一层白名单筛选，只监控你信任的店铺。我是把这些白名单店铺名放在config里，爬完数据后再过滤。

第三件事，别只盯价格这一个维度。有些商品降价是清仓或者过期品，比如食品的临期价格会暴跌，数码产品的"官翻机"价格也会低很多。所以我在商品表里加了一个condition字段，用来标记是全新还是二手还是官翻，不然你看着价格触底很兴奋，点进去发现是二手，白高兴一场。

第四件事，API的价格字段和页面成交价可能差出一截。最初我把接口返回的price直接当成真实成交价来做预警，结果发现预警之后点进链接，价格比预警价格高了不少。后来仔细对比才发现，接口返回的price是商品的基础标价，实际成交价要结合SKU（比如小容量版便宜，大容量版贵）、优惠券、活动折扣来计算。所以预警的时候，我建议宁可漏报不要错报，把阈值压低一点，留出缓冲空间。

6. 我的实操体会与后续扩展思路

这套监控系统从写第一行代码到现在，已经稳定跑了几个月，我个人最大的体会就是：掌握了历史数据之后再回来看商品价格，你会有一种完全不同的洞察力。以前看到"限时6折"会冲动下单，现在会先查一下这商品过去30天的价格分布，很多所谓的大促其实就是把价格先抬高再砍下来一点，根本没到我的动态阈值线。

整个项目真正写代码的时间其实也就是一个周末，大部分时间都耗在接口联调和字段梳理上。如果你也想做一套类似的工具，我的建议是先锁定一个商品、一个平台，把端到端的流程跑通，再逐步增加关键词和预警策略。别一开始就想做一个功能完整的子系统，那样会很累，而且容易失去动手的乐趣。

下一步我打算在现有代码基础上做两件小事：一是把价格历史数据导出成图表，用matplotlib画每个商品的价格趋势线，直观看到价格波动；二是把触发预警后的历史记录单独存一张表，等一个购买周期结束后回看，看哪些预警真正促成了"低价下单"，哪些优化了购物决策。说白了这个工具不复杂，但真正跑起来之后，节省的是时间，改变的是决策方式。