AI编程(四十八):价格监控工具 — 会爬、会存、会提醒的完整小系统

作者:

在

阶段4的收官之战来了。这一篇不学新概念——因为你要用的每一样手艺都学过了:requests 抓网页(第37篇)、BeautifulSoup 抠数据(第41~42篇)、robots.txt + 限速讲礼貌(第47篇)、CSV 存历史(第25篇)、API 发通知(第38篇)。今天干一件更大的事:把散装手艺攒成一台会转的机器——一个真的、每天能替你干活的价格监控工具:定时去商城看一眼价格,历史记录存进 CSV,一旦跌破你心里的目标价,第一时间提醒你。做完这篇,”从网上拿数据”这件事你就正式出师了。

一、真实场景:你和”降价”之间差一个机器人

双十一前你看中一把机械键盘,标价 239 元,你猜它会降到 180 以下,于是每天早晚各刷一次商品页。刷到第五天忘了,第六天想起来一看——179 元,但活动已经结束,涨回去了。你气得拍大腿:我不是在等降价,我是在跟降价赌眼力。

其实这种活儿根本不该人干。人刷价格有两个致命弱点:一是记不住(昨天到底多少钱来着?),二是熬不起(总不能半夜三点定闹钟看一眼)。而这两件事恰恰是程序的强项:它记性好(历史全存着)、不用睡觉(定时自己跑)、还特别听话(说盯几个就盯几个)。

所以今天把价格监控拆成一条流水线,六道工序,每道都是学过的东西:

工序 干什么 手艺来自

|——|——–|———|

① 配置 盯哪几个商品、目标价多少 第28篇 配置管理
② 抓取 下载商品页 HTML 第37篇 HTTP请求
③ 解析 从 HTML 里抠出价格数字 第41~42篇 BeautifulSoup
④ 存储 价格写进 CSV 历史 第25篇 CSV
⑤ 判断 现价 ≤ 目标价吗 第5篇 条件判断
⑥ 通知 跌破了就提醒你 第38篇 API调用

全程守礼仪:先看 robots.txt、抓完歇一拍(第47篇)。看出来了吗?没有一行是”新知识”,全是”老手艺的排班表”——这就是”阶段项目”的意义:单项技能谁都会,把它们无缝串成一台机器,才是真本事。

写程序的高手和新手,差距往往不在”会多少招”,而在”能不能把招连成套路”。今天练的就是连招。

二、项目蓝图:先画图纸,再动工

盖房子之前先画图纸,写这种”多工序程序”也一样。跟AI聊需求时,先把流水线说清楚,AI 给你的代码就是”结构化”的,而不是一坨面条:

price_monitor.py
├── 配置区:商品清单 + 目标价(要改只改这里)
├── get_price(pid):抓一页 → 抠价格 → 返回数字
├── save_history(...):把这轮价格追加进 CSV
├── notify(...):跌破目标价就写"通知"
└── main():按清单挨个走一遍流水线

两个设计原则提前立好,等下写代码时你会感谢自己:

  1. 配置和逻辑分开:盯的商品、目标价、文件名全放顶部”配置区”。以后加商品、改目标价,只动配置区,不碰逻辑——这是第28篇的老规矩;
  2. 每道工序只干一件事:抓取的函数只管抓,存储的函数只管存。哪个环节坏了修哪个,不用在几百行里找问题。

图纸有了,开始动工。本篇所有代码都在本地跑:我们先搭一个”假商城”当靶子(和第47篇的假网站一个套路),监控工具抓它——不给任何真实网站添麻烦,练熟了再考虑盯真实商品(坑3里说法律边界)。

三、动手:从假商城到监控工具

第 1 步:先搭假商城(fake_shop.py)

# fake_shop.py — 本地假商城(先跑这个,再跑监控工具)
from flask import Flask, Response
import random

app = Flask(__name__)

ROBOTS = """User-agent: *
Allow: /product/
Crawl-delay: 1
"""

@app.route("/robots.txt")
def robots():
    return Response(ROBOTS, mimetype="text/plain")

@app.route("/product/")
def product(pid):
    # 每次刷新价格小幅波动,方便练习"降价触发通知"
    price = 199 + random.randint(-30, 30)
    html = f"""

商品{pid}

¥{price} """ return html if __name__ == "__main__": app.run(port=5004)

跑起来:python3.14 fake_shop.py,浏览器打开 http://127.0.0.1:5004/product/1,能看到一行 ¥169(数字每次刷新会变)就算成功。注意价格包在 里——监控工具就靠这个标记抠数字,这和第42篇”先看网页结构再写选择器”是同一个动作。

第 2 步:监控工具主体(price_monitor.py)

# price_monitor.py — 价格监控工具
import csv
import os
import re
import time
import random
import requests
from urllib.robotparser import RobotFileParser

# ---------- 配置区(要改的都在这里) ----------
BASE_URL = "http://127.0.0.1:5004"
PRODUCTS = {
    1: {"name": "机械键盘", "target": 180},
    2: {"name": "无线鼠标", "target": 90},
}
HISTORY_FILE = "price_history.csv"
ALERTS_FILE = "alerts.txt"
UA = "price-monitor/1.0 (个人比价用途; me@example.com)"


def allowed(url):
    """第47篇的护身符:进门先看告示牌"""
    rp = RobotFileParser()
    rp.set_url(BASE_URL + "/robots.txt")
    rp.read()
    return rp.can_fetch("*", url)


def get_price(pid):
    """工序②③:抓一页,抠出价格"""
    url = f"{BASE_URL}/product/{pid}"
    if not allowed(url):
        print(f"跳过(robots.txt 禁止): {url}")
        return None
    try:
        resp = requests.get(url, headers={"User-Agent": UA}, timeout=5)
        resp.raise_for_status()
    except requests.RequestException as e:
        print(f"抓取失败 {url}: {e}")
        return None
    time.sleep(random.uniform(1.0, 2.0))  # 礼貌限速(第47篇)
    match = re.search(r'class="price">¥([d.]+)', resp.text)
    if not match:
        print(f"页面里没找到价格: {url}")
        return None
    return float(match.group(1))


def save_history(pid, name, price):
    """工序④:追加一行进 CSV 历史"""
    is_new = not os.path.exists(HISTORY_FILE)
    with open(HISTORY_FILE, "a", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        if is_new:
            writer.writerow(["时间", "商品ID", "商品名", "价格"])
        writer.writerow([time.strftime("%Y-%m-%d %H:%M"), pid, name, price])


def notify(name, price, target):
    """工序⑥:跌破目标价,写通知"""
    msg = f"[通知] 降价啦!{name} 现价 {price} 元(目标价 {target} 元)"
    print(msg)
    with open(ALERTS_FILE, "a", encoding="utf-8") as f:
        f.write(time.strftime("%Y-%m-%d %H:%M") + " " + msg + "n")


def main():
    print("=== 价格监控启动 ===")
    for pid, info in PRODUCTS.items():
        price = get_price(pid)
        if price is None:
            continue
        print(f"{info['name']}: 当前 {price} 元(目标 {info['target']} 元)")
        save_history(pid, info["name"], price)          # 工序④
        if price <= info["target"]:                     # 工序⑤
            notify(info["name"], price, info["target"])  # 工序⑥
    print("=== 本轮结束 ===")


if __name__ == "__main__":
    main()

第 3 步:跑起来看效果

两个终端:一个跑 python3.14 fake_shop.py,另一个跑 python3.14 price_monitor.py,输出大概长这样:

=== 价格监控启动 ===
机械键盘: 当前 172.0 元(目标 180 元)
[通知] 降价啦!机械键盘 现价 172.0 元(目标价 180 元)
无线鼠标: 当前 211.0 元(目标 90 元)
=== 本轮结束 ===

看一眼多出来的 price_history.csv:

时间,商品ID,商品名,价格
2026-10-25 09:30,1,机械键盘,172.0
2026-10-25 09:30,2,无线鼠标,211.0

再看 alerts.txt,降价通知已经躺在里面了。跑三遍,CSV 就有六行历史——从今天起,”昨天到底多少钱”再也不是靠回忆了。

跟AI说需求的参考说法:

“帮我写一个 Python 价格监控工具:配置区放商品清单(ID、名称、目标价);抓取前先检查 robots.txt,禁止就跳过;从商品页的 span.price 里抠出价格;每轮结果追加写进 CSV(表头只写一次);价格低于目标价就打印通知并写进 alerts.txt;请求之间随机限速 1~2 秒;每个函数只干一件事,抓取失败要跳过当前商品而不是整个程序崩掉。”

四、改进:从”跑一次”到”真监控”

手动跑一次只是”查询器”,监控的意思是”它自己盯着”。三个升级,一个一个跟AI要:

升级1:定时循环——让它一直盯着。最简单的定时就是”跑完歇一会儿再跑”:

if __name__ == "__main__":
    while True:
        main()
        print("休息 30 分钟后再来一轮……")
        time.sleep(30 * 60)

跟AI说:”把主程序改成 while 循环,每 30 分钟跑一轮,Ctrl+C 能优雅退出(捕获 KeyboardInterrupt 并说再见)。”顺便记住一条纪律:轮询间隔宁长勿短——盯价格不是抢票,30 分钟一轮足够了;把间隔压到几秒,那就从”监控”退化成第47篇警告过的”骚扰”了。

升级2:防重复提醒——同一次降价只提醒一回。现在只要价格低于目标价,每一轮都会狂发通知——你将被自己的程序轰炸。修法是给程序加”记忆”:提醒过的(商品+价格)记下来,没变化不再提醒:

ALERTED_FILE = "alerted.txt"

def notify(name, price, target):
    key = f"{name}@{price}"
    alerted = set()
    if os.path.exists(ALERTED_FILE):
        with open(ALERTED_FILE, encoding="utf-8") as f:
            alerted = set(f.read().splitlines())
    if key in alerted:
        return  # 这个价格已经提醒过了,别轰炸
    msg = f"[通知] 降价啦!{name} 现价 {price} 元(目标价 {target} 元)"
    print(msg)
    with open(ALERTS_FILE, "a", encoding="utf-8") as f:
        f.write(time.strftime("%Y-%m-%d %H:%M") + " " + msg + "n")
    with open(ALERTED_FILE, "a", encoding="utf-8") as f:
        f.write(key + "n")

注意 key 里带了价格:降到 172 提醒一次,继续降到 168 会再提醒一次(新低价值得知道),停在 172 不动就闭嘴——这是”值得打扰你”和”骚扰”的分界线。

升级3:真推送——把通知送到你眼前。写进 alerts.txt 只是”留条”,真通知得送到手机或聊天软件。第38篇的 requests.post 这就派上用场了——大多数推送服务(Server酱、企业微信机器人、Telegram Bot)都是”往一个 URL POST 一段 JSON”:

WEBHOOK_URL = "https://你的推送服务地址"   # 换成你自己的

def notify(name, price, target):
    # ……防重复提醒的判断照旧……
    msg = f"降价啦!{name} 现价 {price} 元(目标价 {target} 元"
    print(msg)
    try:
        requests.post(WEBHOOK_URL, json={"text": msg}, timeout=5)
    except requests.RequestException as e:
        print(f"推送失败(不影响主程序): {e}")

跟AI说:”我想让降价通知推到手机,我用的是 XXX(写清你用的服务),帮我把 notify 函数改成调它的 API,推送失败不能影响主程序。”——AI 会给你带鉴权参数的具体写法。先把防重复提醒(升级2)做上,再开推送,顺序反了你的手机会被震到怀疑人生。

五、三个必踩的坑(先打疫苗)

坑1:抠出来的价格不是你以为的价格。最典型的翻车是”页面显示 ¥172,程序存了 239″——因为你抠到的是划线原价、或者是”¥199 券后 172″里的 199。真实商品页往往有好几个价格元素:原价、现价、会员价、分期价。疫苗:先看结构再写正则/选择器(第42篇的规矩),写完打印抠出来的值人肉核对几轮,和眼睛看到的对上了再进下一步。另一个小坑是数字格式:¥1,299 里的逗号会让 float 直接报错,¥199.00~299.00 的区间价会抠出 199.00~299——正则写成 [d.]+ 时想清楚你要的是哪个数。

坑2:CSV 越写越乱。追加模式 open(..., "a") 忘了判断文件存不存在,表头就每轮写一次,最后 CSV 里一半是表头;或者更隐蔽的:价格存进去是数字,读回来全是字符串,直接 if price <= target 就成了字符串比较("9" > "10"!)。疫苗:表头用 is_new 只写一次(代码里已处理);读回 CSV 做比较时显式 float(row["价格"])——第31篇读 CSV 时就该养成这个习惯。还有别忘了 newline=""(第25篇的老坑),不然 Windows 上行尾会多个空行。

坑3:拿练习代码直接怼真实网站。本地假商城随便折腾,真实网站就是另一回事了:一是礼仪,第47篇全套照旧——robots.txt、限速、UA 自报家门,一个都不能少;二是频率,”每 30 分钟看一次商品页”通常没问题,”每 5 秒抢一次”就是骚扰了;三是用途边界,自己比价、做个人记录是常见合理用途,把抓来的价格数据拿去商用、或者扒人家”不允许抓”的区域,性质就变了。疫苗:先拿本地靶子把程序练顺,盯真实商品前重读一遍第47篇,用途变了就重新问一遍”这么干合适吗”。

六、跟AI协作的正确姿势(项目特供版)

  1. 把流水线当需求文档:开工前把六道工序的表格发给AI:”按这个结构写,每道工序一个函数。”AI 产出的代码天然模块化——这比”帮我写个监控价格的程序”得到一坨面条强十倍。
  2. 一次只改一道工序:想换解析方式、想加新通知渠道,就明确说”只改 notify 函数,其他不动”。跟AI的协作纪律和改代码的纪律是同一条:小步走,每步可验证。
  3. 让AI帮你写”体检清单”:”这个价格监控工具上线前该测什么?给我一个 5 项检查清单。”它多半会答:降价触发、没降价不触发、抓取失败不崩、CSV 不重复表头、通知去重——照着清单逐项跑一遍,比”我感觉没问题”靠谱。
  4. 出了怪事,贴证据给AI:”CSV 里价格是 239 但网页显示 172″,把网页 HTML 片段、你的正则、实际输出三样贴全。AI 像老侦探,证据齐了破案飞快;只说”它坏了”,AI 只能猜。

七、今日小练习

练习1:跑通全流水线,亲眼看到”降价触发通知”。先跑 fake_shop.py 再跑 price_monitor.py,确认三件事:控制台打印了两个商品的现价、price_history.csv 里多了两行、降价的那个商品在 alerts.txt 里留了通知。然后把假商城的价格波动改小(比如 random.randint(-5, 5)),让它一直不跌破目标价,再跑——确认这次没有通知(没消息就是好消息)。

提示:如果控制台打印”抓取失败”,先确认假商城那个终端还活着;如果通知狂轰滥炸,说明还没做升级2的防重复提醒——正好把它补上当加练。

练习2:给工具加”第三只眼”——每日价格简报。写一个新函数 daily_report():读 price_history.csv,按商品分组,打印每个商品的”今日最低价 / 最高价 / 最新价”。用它替换一轮监控结束时的 === 本轮结束 ===。做完连跑两轮,简报里应该能看到”有历史数据”的商品多于”今天才出现”的商品。

提示:第25篇的 csv.DictReader + 第33篇的分组思想就够用——先把每行读成字典,按”商品名”归堆,每堆里比价格。注意 CSV 里价格是字符串,比较前 float()。

练习3(综合大题):把升级1~3全装上,做成”能放着不管”的完整工具。验收标准四条:① while 循环自动轮询,Ctrl+C 优雅退出;② 同一个价格只提醒一次,降价到新低再提醒;③ CSV 历史持续追加不乱表头;④ 推送函数失败时程序不崩(本地没配推送地址就先用”写 alerts.txt”顶着)。选做挑战:把配置区挪进 config.json(第28篇),商品清单和目标价从文件读——从此加商品不用改代码。

提示:升级2 的”记忆”用文件存最简单(alerted.txt),但注意 while True 里每轮都要重新读它(或者把集合提到循环外、提醒后同步更新)——文件才是持久记忆,变量只活在一次运行里。

下篇预告

下一篇(第49篇):数据库概念 — 为什么 CSV 用着用着就不够了。你的价格历史越存越长,某天你想问”上周三到今天键盘降了几次”,CSV 翻起来开始费劲——这时候该请”数据库”出场了。第5篇新阶段的开门课,不写代码,先掰开揉碎讲清楚:数据库到底比文件强在哪,什么时候该用它。价格监控工具会在数据库阶段迎来它的终极形态。

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注