DeepSeek 的 API 已经进入很多开发者的日常开发链路VSCode 接入、Codex 接入这类用法越来越普遍价格一变成本和选型都会跟着受影响。如果官方在某个时间段推出全天降价活动比如周末价格下调靠手动刷新开放平台页面很难保证第一时间发现。这里写一个可运行的 DeepSeek 价格监控与邮件通知脚本用 Python 定时抓取价格信息发现变化后自动发送邮件。脚本跑通后不只适用于 DeepSeek也能改成其他云产品价格或文档变化的监控器。整个方案的重点不是写几行爬虫代码而是把“监控”这件事做成一条完整链路定时抓取、快照保存、差异对比、邮件通知、日志记录。后面会从需求拆解开始逐步给出完整代码、本地验证方法、定时调度方式以及常见问题的排查路径。1. 这个监控脚本要解决什么问题1.1 使用场景DeepSeek 价格调整与周末活动价DeepSeek 开放平台的模型价格会随着活动、计费策略调整而发生变化。周末全天降价这类活动价格通常会在某个时间点被刷新随后一段时间内维持新价格。对个人开发者和团队来说价格变化影响两个地方一是成本预算二是模型选型。举例来说平时使用 deepseek-chat如果价格下调可以考虑加大调用量或者把更多非实时任务切过来。如果价格上调则要考虑切换其他模型、增加缓存策略或者调整调用频率。如果周末活动价只在特定时段生效最好在价格变化的第一时间收到邮件而不是等活动快结束才看到。这里要明确一个设计选择脚本不判断“降价还是涨价”它只发现“价格相对于上次快照发生了变化”然后发邮件。变化方向由收到邮件的人自行判断。这样设计会让监控逻辑更简单也更通用。今天可能是周末降价明天可能是计费规则调整脚本只需要忠实记录变化不需要懂得业务含义。1.2 直接人工盯价格的三个问题人工刷新价格页有几个明显问题。第一无法保证及时性。价格调整不会主动通知你等看到时可能已经过了几个小时甚至更久。对于短时活动错过意味着整个活动周期内都无法利用新价格。第二难以对比前后差异。只凭肉眼很难记住上一次的准确价格尤其是多个模型、多个计费维度同时变化时。即使截图保存事后也很难快速找出“哪个模型变了、从多少变到多少”。第三人工操作无法形成记录。今天看过的价格、上周看过的价格、活动期间的价格如果没有持久化事后无法追溯。一旦团队里有人问起“上次调价是什么时候”只能靠回忆。所以这个脚本的核心不是“爬虫”而是一条完整的检测链路定期抓取 - 保存快照 - 对比变化 - 触发通知 - 记录日志。有了快照和历史记录才能回答“什么时候变的”和“变了几次”这两个问题。1.3 核心设计抓取、对比、通知三件事整个脚本可以拆成三层。抓取层负责访问 DeepSeek 价格页面或价格接口拿到价格文本。对比层负责把当前价格与上一次快照做比较判断价格是否变化。通知层负责在变化发生时发送 SMTP 邮件。这里有一个容易被忽略的设计点价格监控脚本必须能容忍“解析失败”。价格页面的 HTML 结构一旦调整原有选择器可能立刻失效。如果脚本在解析失败时直接崩溃监控就断了如果错误地认为“价格没变”又会漏报。比较稳妥的策略是解析失败时记录一条 warning同时使用页面指纹检测页面是否变化。如果页面指纹变了说明页面内容确实有变动发送一封“页面结构可能变化请人工确认”的邮件。这个三层结构是后面所有代码的主线。接下来先看环境与项目结构再进入代码实现。2. 技术选型与项目结构2.1 为什么用 Python 而不是更重的框架这个监控任务的特点是轻量、定时、IO 密集。Python 的优势在于标准库自带 smtplib 和 email 模块发邮件不需要额外依赖。requests 和 BeautifulSoup 生态成熟抓取和解析都很方便。脚本可以放在任何有 Python 3 的服务器上直接执行不需要编译和构建。不需要引入 Celery 这类任务队列。场景只有“每 30 分钟执行一次”定时调度交给系统级 cron 或 systemd timer 更合适比在 Python 内部实现定时循环更稳定。定时任务挂了由系统负责拉起脚本本身不需要常驻内存。2.2 目录结构与文件职责推荐使用一个独立目录方便定时任务切换工作目录。deepseek-price-monitor/ ├── config.yaml ├── requirements.txt ├── monitor.py ├── data/ │ └── snapshot.json └── logs/ └── monitor.logconfig.yaml配置页面地址、解析规则、SMTP 信息、快照路径。monitor.py主脚本包含抓取、对比、通知、快照保存逻辑。requirements.txt依赖清单。data/snapshot.json上一次价格快照由脚本自动生成。logs/monitor.log运行日志用于排查问题。data 目录和 logs 目录不需要手动创建脚本会在启动时自动创建。部署时只要保留 config.yaml、monitor.py 和 requirements.txt 三个文件即可。2.3 requirements.txt 与 config.yamlrequirements.txt 内容requests beautifulsoup4 PyYAML安装命令pip install -r requirements.txtconfig.yaml 是核心配置文件。下面是一份示例里面所有价格数据都只是用来演示数据结构不代表 DeepSeek 官方价格source: # 示例地址实际以官方开放平台最新价格页为准 url: https://api-docs.deepseek.com/zh-cn/quick_start/pricing mode: html selector: container: table row: tbody tr smtp: host: smtp.example.com port: 465 user: notifyexample.com password_env: SMTP_PASSWORD use_ssl: true from_name: Price Monitor to: - devexample.com store: snapshot_file: data/snapshot.json log: level: INFO file: logs/monitor.log关键点password_env 不直接写密码而是指定环境变量名避免把密码提交到代码仓库。url 指向官方价格页。实际落地时以官方最新页面结构为准selector 需要按页面微调。mode 支持 html也可以扩展为 json应对官方提供结构化价格接口的情况。to 是列表可以配置多个收件人。配置项比较多整理成一张表方便速查配置项说明示例注意事项source.url价格页面或接口地址官方价格页页面迁移后需要更新source.mode解析方式html官方提供接口后可改 jsonsource.selector.container表格或容器的 CSS 选择器table页面改版后需要同步修改source.selector.row行选择器tbody tr只解析数据行避免表头混入smtp.hostSMTP 服务地址smtp.example.com使用真实邮箱的 SMTP 域名smtp.portSMTP 端口465465 通常配 SSL587 配 STARTTLSsmtp.user发件邮箱notifyexample.com需要开启 SMTP 服务smtp.password_env授权码环境变量名SMTP_PASSWORD不要写在配置文件里store.snapshot_file快照文件路径data/snapshot.json需要保证可写log.file日志文件路径logs/monitor.log目录不存在会自动创建3. 核心代码实现monitor.py 不拆成多个文件保持单文件脚本方便拷贝到服务器。下面按函数分段给出代码按顺序拼接保存为 monitor.py 即可运行。3.1 配置加载与日志初始化脚本首先完成两件事读取 config.yaml初始化日志。日志很关键因为定时任务不会像手动执行那样把错误打到终端上必须落到文件里。#!/usr/bin/env python3 DeepSeek 价格监控与邮件通知脚本。 定时抓取价格页面解析模型价格与上次快照对比。 发生变化时发送 SMTP 邮件。 import argparse import hashlib import json import logging import os import smtplib from datetime import datetime from email.header import Header from email.mime.text import MIMEText from email.utils import formataddr from pathlib import Path import requests import yaml from bs4 import BeautifulSoup def load_config(path): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def setup_logging(cfg): level getattr(logging, str(cfg.get(log, {}).get(level, INFO)).upper(), logging.INFO) handlers [logging.StreamHandler()] log_file cfg.get(log, {}).get(file) if log_file: Path(log_file).parent.mkdir(parentsTrue, exist_okTrue) handlers.append(logging.FileHandler(log_file, encodingutf-8)) logging.basicConfig( levellevel, format%(asctime)s %(levelname)s %(name)s %(message)s, handlershandlers, ) return logging.getLogger(monitor)这里有两个重点。第一日志目录通过Path(log_file).parent.mkdir(parentsTrue, exist_okTrue)自动创建避免手动建目录。第二日志使用FileHandler写入文件同时保留终端输出手动运行时能直接看到结果。3.2 价格抓取与页面解析抓取部分使用 requests 的 get 方法设置 User-Agent 是为了避免部分页面拒绝默认客户端。解析部分用 BeautifulSoup根据 config 中的 selector 找到价格表格提取每一行的文本。def fetch_page(url, timeout30): headers {User-Agent: Mozilla/5.0 (compatible; PriceMonitor/1.0)} resp requests.get(url, headersheaders, timeouttimeout) resp.raise_for_status() return resp.text def parse_prices_from_html(page_text, source_cfg): soup BeautifulSoup(page_text, html.parser) selector_cfg source_cfg.get(selector, {}) container soup.select_one(selector_cfg.get(container, table)) if not container: raise ValueError(container selector not found: %s % selector_cfg.get(container)) row_selector selector_cfg.get(row, tbody tr) rows container.select(row_selector) prices {} for row in rows: cells [c.get_text( , stripTrue) for c in row.find_all([td, th])] if len(cells) 2: continue model_name cells[0].lower() # 过滤表头和无关行 if any(key in model_name for key in (输入, 输出, 模型, 价格, 计费, unit, context)): continue prices[model_name] { cells: cells, raw: | .join(cells), } if not prices: raise ValueError(no price rows parsed, page structure may have changed) return pricesparse_prices_from_html 返回一个字典key 是模型名value 里存了原始文本raw。后续对比时不关心价格具体是多少只关心raw是否和上次一致。这样做的好处是即使页面增加了备注列、说明文字只要整行内容有变化就会被识别出来。这里要解释一个设计取舍为什么不精确提取数字价格因为价格页面的排版和列数可能变化精确解析到“输入价格、输出价格”两个字段需要针对特定页面写大量兼容逻辑。用整行原文作为比较基准虽然粒度粗但实现简单且足够满足“发现变化”这一核心目标。解析失败时抛出异常而不是返回空字典。原因很简单返回空字典会导致脚本误判为“价格清零”从而产生误报。抛出异常后主流程会走页面指纹回退逻辑更安全。3.3 快照存储与变化检测快照文件保存三类数据prices、fingerprint、last_notified_at。写入时先写临时文件再用os.replace原子替换避免写入中途进程被杀产生半截 JSON。def load_snapshot(path): if Path(path).exists(): with open(path, r, encodingutf-8) as f: return json.load(f) return {prices: {}, fingerprint: , last_notified_at: None} def save_snapshot(snapshot, path): Path(path).parent.mkdir(parentsTrue, exist_okTrue) tmp path .tmp with open(tmp, w, encodingutf-8) as f: json.dump(snapshot, f, ensure_asciiFalse, indent2) os.replace(tmp, path) def build_report(old_prices, new_prices): changed False lines [] for model, current in new_prices.items(): previous old_prices.get(model) if previous is None or previous.get(raw) ! current.get(raw): changed True lines.append({ model: model, previous_raw: previous.get(raw) if previous else N/A, current_raw: current.get(raw), }) return changed, lines def detect_fallback(page_text, snapshot_fingerprint): fp hashlib.sha256(page_text.encode(utf-8)).hexdigest() return fp ! snapshot_fingerprint, fpbuild_report是核心逻辑。它遍历当前解析出的所有模型与旧快照逐行对比。只要有一个模型的raw不一致就认为价格发生变化。此时生成一份变化明细包含模型名、上次内容、当前内容用于拼装邮件正文。detect_fallback是兜底逻辑。解析失败时对整页文本做 SHA256 指纹。如果指纹和上次不一样说明页面确实有内容变动需要人工确认。这种方式不精确但能避免漏报。3.4 邮件通知与 HTML 正文邮件通知功能使用 smtplib。发送前从环境变量读取授权码不在配置文件中保存密码。邮件正文采用 HTML 表格比纯文本更直观。def send_email(cfg, subject, html_body): smtp_cfg cfg[smtp] password os.environ.get(smtp_cfg.get(password_env, SMTP_PASSWORD)) if not password: raise RuntimeError(%s env not set % smtp_cfg.get(password_env, SMTP_PASSWORD)) msg MIMEText(html_body, html, utf-8) msg[Subject] Header(subject, utf-8) msg[From] formataddr((str(smtp_cfg.get(from_name, Price Monitor)), smtp_cfg[user])) to_addrs smtp_cfg[to] if isinstance(smtp_cfg[to], list) else [smtp_cfg[to]] msg[To] , .join(to_addrs) if smtp_cfg.get(use_ssl): server smtplib.SMTP_SSL(smtp_cfg[host], smtp_cfg[port], timeout30) else: server smtplib.SMTP(smtp_cfg[host], smtp_cfg[port], timeout30) server.ehlo() if smtp_cfg.get(use_starttls, True): server.starttls() server.ehlo() try: server.login(smtp_cfg[user], password) server.sendmail(smtp_cfg[user], to_addrs, msg.as_string()) finally: server.quit() return True def build_html_body(report_lines, is_fallbackFalse): if is_fallback: return p价格页面结构发生变化或解析失败请人工确认。/p rows .join( ftrtd{line[model]}/td ftd{line[previous_raw]}/td ftd{line[current_raw]}/td/tr for line in report_lines ) return ( table border\1\ cellspacing\0\ cellpadding\6\ trth模型/thth上次内容/thth当前内容/th/tr f{rows}/table )这里需要特别注意 SMTP 的端口和加密方式匹配。常见的组合有两种465 端口使用SMTP_SSL全程加密。587 端口使用SMTP加starttls()先建立普通连接再升级为加密连接。很多邮箱要求使用“授权码”而不是登录密码常见错误是把邮箱登录密码填进去导致认证失败。这个坑会在后面的排查章节详细说明。3.5 主流程串联主流程把抓取、解析、对比、通知、保存快照串起来。为了便于测试增加一个--send-anyway参数可以强制发送一封测试邮件。def main(): parser argparse.ArgumentParser(descriptionDeepSeek price monitor) parser.add_argument(--config, defaultconfig.yaml) parser.add_argument(--send-anyway, actionstore_true, helpforce send a test email) args parser.parse_args() cfg load_config(args.config) logger setup_logging(cfg) logger.info(start monitor, source%s, cfg[source][url]) snapshot_path cfg[store][snapshot_file] snapshot load_snapshot(snapshot_path) page_text fetch_page(cfg[source][url]) changed False report_lines None is_fallback False if args.send_anyway: changed True else: try: mode cfg.get(source, {}).get(mode, html) if mode html: new_prices parse_prices_from_html(page_text, cfg[source]) elif mode json: new_prices parse_prices_from_json(page_text, cfg[source]) else: raise ValueError(unsupported source mode: %s % mode) old_prices snapshot.get(prices, {}) changed, report_lines build_report(old_prices, new_prices) snapshot[prices] new_prices snapshot[fingerprint] hashlib.sha256(page_text.encode(utf-8)).hexdigest() except Exception as exc: logger.warning(parse failed, fallback to fingerprint: %s, exc) is_fallback True changed, snapshot[fingerprint] detect_fallback( page_text, snapshot.get(fingerprint, ) ) if changed: subject [PriceMonitor] DeepSeek 价格发生变化 html_body build_html_body(report_lines, is_fallback) send_email(cfg, subject, html_body) snapshot[last_notified_at] datetime.now().isoformat(timespecseconds) logger.info(price changed, notification sent) else: logger.info(price not changed, skip notification) save_snapshot(snapshot, snapshot_path) logger.info(done) def parse_prices_from_json(page_text, source_cfg): data json.loads(page_text) models data.get(models, data.get(data, [])) if isinstance(models, dict): models list(models.items()) prices {} for item in models: if isinstance(item, dict): name item.get(model, item.get(name, )) raw json.dumps(item, ensure_asciiFalse) else: name, raw item[ Python实现DeepSeek价格监控与邮件通知脚本实战
DeepSeek 的 API 已经进入很多开发者的日常开发链路VSCode 接入、Codex 接入这类用法越来越普遍价格一变成本和选型都会跟着受影响。如果官方在某个时间段推出全天降价活动比如周末价格下调靠手动刷新开放平台页面很难保证第一时间发现。这里写一个可运行的 DeepSeek 价格监控与邮件通知脚本用 Python 定时抓取价格信息发现变化后自动发送邮件。脚本跑通后不只适用于 DeepSeek也能改成其他云产品价格或文档变化的监控器。整个方案的重点不是写几行爬虫代码而是把“监控”这件事做成一条完整链路定时抓取、快照保存、差异对比、邮件通知、日志记录。后面会从需求拆解开始逐步给出完整代码、本地验证方法、定时调度方式以及常见问题的排查路径。1. 这个监控脚本要解决什么问题1.1 使用场景DeepSeek 价格调整与周末活动价DeepSeek 开放平台的模型价格会随着活动、计费策略调整而发生变化。周末全天降价这类活动价格通常会在某个时间点被刷新随后一段时间内维持新价格。对个人开发者和团队来说价格变化影响两个地方一是成本预算二是模型选型。举例来说平时使用 deepseek-chat如果价格下调可以考虑加大调用量或者把更多非实时任务切过来。如果价格上调则要考虑切换其他模型、增加缓存策略或者调整调用频率。如果周末活动价只在特定时段生效最好在价格变化的第一时间收到邮件而不是等活动快结束才看到。这里要明确一个设计选择脚本不判断“降价还是涨价”它只发现“价格相对于上次快照发生了变化”然后发邮件。变化方向由收到邮件的人自行判断。这样设计会让监控逻辑更简单也更通用。今天可能是周末降价明天可能是计费规则调整脚本只需要忠实记录变化不需要懂得业务含义。1.2 直接人工盯价格的三个问题人工刷新价格页有几个明显问题。第一无法保证及时性。价格调整不会主动通知你等看到时可能已经过了几个小时甚至更久。对于短时活动错过意味着整个活动周期内都无法利用新价格。第二难以对比前后差异。只凭肉眼很难记住上一次的准确价格尤其是多个模型、多个计费维度同时变化时。即使截图保存事后也很难快速找出“哪个模型变了、从多少变到多少”。第三人工操作无法形成记录。今天看过的价格、上周看过的价格、活动期间的价格如果没有持久化事后无法追溯。一旦团队里有人问起“上次调价是什么时候”只能靠回忆。所以这个脚本的核心不是“爬虫”而是一条完整的检测链路定期抓取 - 保存快照 - 对比变化 - 触发通知 - 记录日志。有了快照和历史记录才能回答“什么时候变的”和“变了几次”这两个问题。1.3 核心设计抓取、对比、通知三件事整个脚本可以拆成三层。抓取层负责访问 DeepSeek 价格页面或价格接口拿到价格文本。对比层负责把当前价格与上一次快照做比较判断价格是否变化。通知层负责在变化发生时发送 SMTP 邮件。这里有一个容易被忽略的设计点价格监控脚本必须能容忍“解析失败”。价格页面的 HTML 结构一旦调整原有选择器可能立刻失效。如果脚本在解析失败时直接崩溃监控就断了如果错误地认为“价格没变”又会漏报。比较稳妥的策略是解析失败时记录一条 warning同时使用页面指纹检测页面是否变化。如果页面指纹变了说明页面内容确实有变动发送一封“页面结构可能变化请人工确认”的邮件。这个三层结构是后面所有代码的主线。接下来先看环境与项目结构再进入代码实现。2. 技术选型与项目结构2.1 为什么用 Python 而不是更重的框架这个监控任务的特点是轻量、定时、IO 密集。Python 的优势在于标准库自带 smtplib 和 email 模块发邮件不需要额外依赖。requests 和 BeautifulSoup 生态成熟抓取和解析都很方便。脚本可以放在任何有 Python 3 的服务器上直接执行不需要编译和构建。不需要引入 Celery 这类任务队列。场景只有“每 30 分钟执行一次”定时调度交给系统级 cron 或 systemd timer 更合适比在 Python 内部实现定时循环更稳定。定时任务挂了由系统负责拉起脚本本身不需要常驻内存。2.2 目录结构与文件职责推荐使用一个独立目录方便定时任务切换工作目录。deepseek-price-monitor/ ├── config.yaml ├── requirements.txt ├── monitor.py ├── data/ │ └── snapshot.json └── logs/ └── monitor.logconfig.yaml配置页面地址、解析规则、SMTP 信息、快照路径。monitor.py主脚本包含抓取、对比、通知、快照保存逻辑。requirements.txt依赖清单。data/snapshot.json上一次价格快照由脚本自动生成。logs/monitor.log运行日志用于排查问题。data 目录和 logs 目录不需要手动创建脚本会在启动时自动创建。部署时只要保留 config.yaml、monitor.py 和 requirements.txt 三个文件即可。2.3 requirements.txt 与 config.yamlrequirements.txt 内容requests beautifulsoup4 PyYAML安装命令pip install -r requirements.txtconfig.yaml 是核心配置文件。下面是一份示例里面所有价格数据都只是用来演示数据结构不代表 DeepSeek 官方价格source: # 示例地址实际以官方开放平台最新价格页为准 url: https://api-docs.deepseek.com/zh-cn/quick_start/pricing mode: html selector: container: table row: tbody tr smtp: host: smtp.example.com port: 465 user: notifyexample.com password_env: SMTP_PASSWORD use_ssl: true from_name: Price Monitor to: - devexample.com store: snapshot_file: data/snapshot.json log: level: INFO file: logs/monitor.log关键点password_env 不直接写密码而是指定环境变量名避免把密码提交到代码仓库。url 指向官方价格页。实际落地时以官方最新页面结构为准selector 需要按页面微调。mode 支持 html也可以扩展为 json应对官方提供结构化价格接口的情况。to 是列表可以配置多个收件人。配置项比较多整理成一张表方便速查配置项说明示例注意事项source.url价格页面或接口地址官方价格页页面迁移后需要更新source.mode解析方式html官方提供接口后可改 jsonsource.selector.container表格或容器的 CSS 选择器table页面改版后需要同步修改source.selector.row行选择器tbody tr只解析数据行避免表头混入smtp.hostSMTP 服务地址smtp.example.com使用真实邮箱的 SMTP 域名smtp.portSMTP 端口465465 通常配 SSL587 配 STARTTLSsmtp.user发件邮箱notifyexample.com需要开启 SMTP 服务smtp.password_env授权码环境变量名SMTP_PASSWORD不要写在配置文件里store.snapshot_file快照文件路径data/snapshot.json需要保证可写log.file日志文件路径logs/monitor.log目录不存在会自动创建3. 核心代码实现monitor.py 不拆成多个文件保持单文件脚本方便拷贝到服务器。下面按函数分段给出代码按顺序拼接保存为 monitor.py 即可运行。3.1 配置加载与日志初始化脚本首先完成两件事读取 config.yaml初始化日志。日志很关键因为定时任务不会像手动执行那样把错误打到终端上必须落到文件里。#!/usr/bin/env python3 DeepSeek 价格监控与邮件通知脚本。 定时抓取价格页面解析模型价格与上次快照对比。 发生变化时发送 SMTP 邮件。 import argparse import hashlib import json import logging import os import smtplib from datetime import datetime from email.header import Header from email.mime.text import MIMEText from email.utils import formataddr from pathlib import Path import requests import yaml from bs4 import BeautifulSoup def load_config(path): with open(path, r, encodingutf-8) as f: return yaml.safe_load(f) def setup_logging(cfg): level getattr(logging, str(cfg.get(log, {}).get(level, INFO)).upper(), logging.INFO) handlers [logging.StreamHandler()] log_file cfg.get(log, {}).get(file) if log_file: Path(log_file).parent.mkdir(parentsTrue, exist_okTrue) handlers.append(logging.FileHandler(log_file, encodingutf-8)) logging.basicConfig( levellevel, format%(asctime)s %(levelname)s %(name)s %(message)s, handlershandlers, ) return logging.getLogger(monitor)这里有两个重点。第一日志目录通过Path(log_file).parent.mkdir(parentsTrue, exist_okTrue)自动创建避免手动建目录。第二日志使用FileHandler写入文件同时保留终端输出手动运行时能直接看到结果。3.2 价格抓取与页面解析抓取部分使用 requests 的 get 方法设置 User-Agent 是为了避免部分页面拒绝默认客户端。解析部分用 BeautifulSoup根据 config 中的 selector 找到价格表格提取每一行的文本。def fetch_page(url, timeout30): headers {User-Agent: Mozilla/5.0 (compatible; PriceMonitor/1.0)} resp requests.get(url, headersheaders, timeouttimeout) resp.raise_for_status() return resp.text def parse_prices_from_html(page_text, source_cfg): soup BeautifulSoup(page_text, html.parser) selector_cfg source_cfg.get(selector, {}) container soup.select_one(selector_cfg.get(container, table)) if not container: raise ValueError(container selector not found: %s % selector_cfg.get(container)) row_selector selector_cfg.get(row, tbody tr) rows container.select(row_selector) prices {} for row in rows: cells [c.get_text( , stripTrue) for c in row.find_all([td, th])] if len(cells) 2: continue model_name cells[0].lower() # 过滤表头和无关行 if any(key in model_name for key in (输入, 输出, 模型, 价格, 计费, unit, context)): continue prices[model_name] { cells: cells, raw: | .join(cells), } if not prices: raise ValueError(no price rows parsed, page structure may have changed) return pricesparse_prices_from_html 返回一个字典key 是模型名value 里存了原始文本raw。后续对比时不关心价格具体是多少只关心raw是否和上次一致。这样做的好处是即使页面增加了备注列、说明文字只要整行内容有变化就会被识别出来。这里要解释一个设计取舍为什么不精确提取数字价格因为价格页面的排版和列数可能变化精确解析到“输入价格、输出价格”两个字段需要针对特定页面写大量兼容逻辑。用整行原文作为比较基准虽然粒度粗但实现简单且足够满足“发现变化”这一核心目标。解析失败时抛出异常而不是返回空字典。原因很简单返回空字典会导致脚本误判为“价格清零”从而产生误报。抛出异常后主流程会走页面指纹回退逻辑更安全。3.3 快照存储与变化检测快照文件保存三类数据prices、fingerprint、last_notified_at。写入时先写临时文件再用os.replace原子替换避免写入中途进程被杀产生半截 JSON。def load_snapshot(path): if Path(path).exists(): with open(path, r, encodingutf-8) as f: return json.load(f) return {prices: {}, fingerprint: , last_notified_at: None} def save_snapshot(snapshot, path): Path(path).parent.mkdir(parentsTrue, exist_okTrue) tmp path .tmp with open(tmp, w, encodingutf-8) as f: json.dump(snapshot, f, ensure_asciiFalse, indent2) os.replace(tmp, path) def build_report(old_prices, new_prices): changed False lines [] for model, current in new_prices.items(): previous old_prices.get(model) if previous is None or previous.get(raw) ! current.get(raw): changed True lines.append({ model: model, previous_raw: previous.get(raw) if previous else N/A, current_raw: current.get(raw), }) return changed, lines def detect_fallback(page_text, snapshot_fingerprint): fp hashlib.sha256(page_text.encode(utf-8)).hexdigest() return fp ! snapshot_fingerprint, fpbuild_report是核心逻辑。它遍历当前解析出的所有模型与旧快照逐行对比。只要有一个模型的raw不一致就认为价格发生变化。此时生成一份变化明细包含模型名、上次内容、当前内容用于拼装邮件正文。detect_fallback是兜底逻辑。解析失败时对整页文本做 SHA256 指纹。如果指纹和上次不一样说明页面确实有内容变动需要人工确认。这种方式不精确但能避免漏报。3.4 邮件通知与 HTML 正文邮件通知功能使用 smtplib。发送前从环境变量读取授权码不在配置文件中保存密码。邮件正文采用 HTML 表格比纯文本更直观。def send_email(cfg, subject, html_body): smtp_cfg cfg[smtp] password os.environ.get(smtp_cfg.get(password_env, SMTP_PASSWORD)) if not password: raise RuntimeError(%s env not set % smtp_cfg.get(password_env, SMTP_PASSWORD)) msg MIMEText(html_body, html, utf-8) msg[Subject] Header(subject, utf-8) msg[From] formataddr((str(smtp_cfg.get(from_name, Price Monitor)), smtp_cfg[user])) to_addrs smtp_cfg[to] if isinstance(smtp_cfg[to], list) else [smtp_cfg[to]] msg[To] , .join(to_addrs) if smtp_cfg.get(use_ssl): server smtplib.SMTP_SSL(smtp_cfg[host], smtp_cfg[port], timeout30) else: server smtplib.SMTP(smtp_cfg[host], smtp_cfg[port], timeout30) server.ehlo() if smtp_cfg.get(use_starttls, True): server.starttls() server.ehlo() try: server.login(smtp_cfg[user], password) server.sendmail(smtp_cfg[user], to_addrs, msg.as_string()) finally: server.quit() return True def build_html_body(report_lines, is_fallbackFalse): if is_fallback: return p价格页面结构发生变化或解析失败请人工确认。/p rows .join( ftrtd{line[model]}/td ftd{line[previous_raw]}/td ftd{line[current_raw]}/td/tr for line in report_lines ) return ( table border\1\ cellspacing\0\ cellpadding\6\ trth模型/thth上次内容/thth当前内容/th/tr f{rows}/table )这里需要特别注意 SMTP 的端口和加密方式匹配。常见的组合有两种465 端口使用SMTP_SSL全程加密。587 端口使用SMTP加starttls()先建立普通连接再升级为加密连接。很多邮箱要求使用“授权码”而不是登录密码常见错误是把邮箱登录密码填进去导致认证失败。这个坑会在后面的排查章节详细说明。3.5 主流程串联主流程把抓取、解析、对比、通知、保存快照串起来。为了便于测试增加一个--send-anyway参数可以强制发送一封测试邮件。def main(): parser argparse.ArgumentParser(descriptionDeepSeek price monitor) parser.add_argument(--config, defaultconfig.yaml) parser.add_argument(--send-anyway, actionstore_true, helpforce send a test email) args parser.parse_args() cfg load_config(args.config) logger setup_logging(cfg) logger.info(start monitor, source%s, cfg[source][url]) snapshot_path cfg[store][snapshot_file] snapshot load_snapshot(snapshot_path) page_text fetch_page(cfg[source][url]) changed False report_lines None is_fallback False if args.send_anyway: changed True else: try: mode cfg.get(source, {}).get(mode, html) if mode html: new_prices parse_prices_from_html(page_text, cfg[source]) elif mode json: new_prices parse_prices_from_json(page_text, cfg[source]) else: raise ValueError(unsupported source mode: %s % mode) old_prices snapshot.get(prices, {}) changed, report_lines build_report(old_prices, new_prices) snapshot[prices] new_prices snapshot[fingerprint] hashlib.sha256(page_text.encode(utf-8)).hexdigest() except Exception as exc: logger.warning(parse failed, fallback to fingerprint: %s, exc) is_fallback True changed, snapshot[fingerprint] detect_fallback( page_text, snapshot.get(fingerprint, ) ) if changed: subject [PriceMonitor] DeepSeek 价格发生变化 html_body build_html_body(report_lines, is_fallback) send_email(cfg, subject, html_body) snapshot[last_notified_at] datetime.now().isoformat(timespecseconds) logger.info(price changed, notification sent) else: logger.info(price not changed, skip notification) save_snapshot(snapshot, snapshot_path) logger.info(done) def parse_prices_from_json(page_text, source_cfg): data json.loads(page_text) models data.get(models, data.get(data, [])) if isinstance(models, dict): models list(models.items()) prices {} for item in models: if isinstance(item, dict): name item.get(model, item.get(name, )) raw json.dumps(item, ensure_asciiFalse) else: name, raw item[