这个教程将带你一步步搭建一个能够对抗Cloudflare拦截、支持开机自启、稳定可靠的库存监控系统。
终极库存监控脚本搭建教程（从入门到服务器部署）
本教程旨在创建一个强大的商品库存和价格监控脚本。它使用 Python、通过模拟真实浏览器（Selenium）来访问网页以避免被屏蔽，利用 Telegram 发送实时通知，并最终部署为可在服务器上7×24小时无人值守运行的系统服务。
第一部分：准备工作 (Prerequisites)
- 一台Linux服务器：推荐使用 Ubuntu 20.04 或更高版本的系统。本教程所有命令均基于Ubuntu。
- 一个Telegram账号：用于接收库存和价格变动通知。
- 基础的SSH工具：如 Termius, Xshell, PuTTY 或系统自带的终端，用于连接你的服务器。
第二部分：服务器环境准备 (Environment Setup)
连接上你的服务器后，依次执行以下命令来安装所有必需的软件和依赖。
步骤 2.1: 更新系统软件包列表
这是一个好习惯，可以确保你安装的软件都是最新的。
sudo apt update
sudo apt upgrade -y
步骤 2.2: 安装 Python 和 pip
系统通常自带Python，但我们确保 pip（Python的包管理器）也已安装。
sudo apt install python3 python3-pip -y
步骤 2.3: 安装 Google Chrome 浏览器
我们的脚本需要驱动一个真实的浏览器，所以必须在服务器上安装它。
# 下载Chrome的官方安装包
wget https://dl.google.com/linux/direct/google-chrome-stable_current_amd64.deb
# 安装Chrome，apt会自动处理所有依赖
sudo apt install ./google-chrome-stable_current_amd64.deb -y
步骤 2.4: 安装 Python 依赖库
安装 selenium（浏览器“遥控器”）和 undetected-chromedriver（能避免被网站检测到的“智能钥匙”）。
pip3 install selenium undetected-chromedriver
第三部分：创建 Telegram 机器人 (Telegram Bot Setup)
你需要一个机器人来给你发消息。
- 在Telegram里搜索
BotFather并开始对话。 - 发送
/newbot命令。 - 按照提示，为你的机器人设置一个名称 (Name)，例如
我的库存监控助手。 - 再设置一个用户名 (Username)，它必须以
bot结尾，例如my_stock_monitor_bot。 - 创建成功后，
BotFather会给你一长串 Token，格式类似123456:ABC-DEF123456。请务必复制并保存好这个Token。 - 接着，找到你的机器人，给它发送一条任意消息（比如
/start）。 - 获取你的 Chat ID：在浏览器中访问
https://api.telegram.org/bot你的Token/getUpdates(把你的Token换成你刚刚获取到的Token)。在返回的页面中，找到"chat":{"id":一串数字}，这串数字就是你的Chat ID。
第四部分：编写与配置监控脚本 (The Script)
现在，我们来创建并配置核心的监控脚本。
步骤 4.1: 创建项目目录并进入
mkdir ~/stock_monitor
cd ~/stock_monitor
步骤 4.2: 创建脚本文件
使用 nano 编辑器创建一个新的Python文件。
nano stock_monitor.py
步骤 4.3: 粘贴最终脚本代码
将下面这个完整、最终优化版的代码，全部复制并粘贴到 nano 编辑器中。
Python：复制下面的：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
终极库存监控脚本 (浏览器模式-防Cloudflare)
功能：监控商品库存和价格变化，通过Telegram发送通知，并作为系统服务稳定运行。
"""
import re
import time
import json
import os
from datetime import datetime
import requests # 保留，因为发送Telegram消息时仍在使用
import undetected_chromedriver as uc # 导入新库以模拟浏览器
# ======================================================================
# 配置区域
# ======================================================================
# --- Telegram Bot 配置 (必须修改) ---
TELEGRAM_BOT_TOKEN = "在这里替换为你的Bot Token"
TELEGRAM_CHAT_ID = "在这里替换为你的电报ID"
# --- 监控商品列表 (请根据需求修改) ---
MONITOR_PRODUCTS = [
{
"name": "商品1", # 可自行更改名称
"url": "https://example.com/product1",
"stock_pattern": r'库存\((\d+)\)', # 修正为正确的库存匹配规则
"price_pattern": r'¥\s*(\d+\.?\d*)' # 修正为正确的价格匹配规则
},
# 下面两个商品，如果网站结构和第一个相同，也需要用新的匹配规则
# 如果不同，则需要为它们单独查找规则
{
"name": "商品2",
"url": "https://example.com/product2",
"stock_pattern": r'库存\((\d+)\)',
"price_pattern": r'¥\s*(\d+\.?\d*)'
},
{
"name": "商品3",
"url": "https://example.com/product1",
"stock_pattern": r'库存\((\d+)\)',
"price_pattern": r'¥\s*(\d+\.?\d*)'
}
]
# --- 全局设置 ---
CHECK_INTERVAL = 3600 # 检查间隔（秒）, 3600 = 1小时
DATA_FILE = "monitor_data.json" # 数据存储文件
# ======================================================================
# 核心函数区域
# ======================================================================
def load_history():
"""加载历史数据"""
if os.path.exists(DATA_FILE):
try:
with open(DATA_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except:
return {}
return {}
def save_history(data):
"""保存历史数据"""
with open(DATA_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=4)
def get_page_content(url):
"""获取网页内容 (使用 Selenium + undetected-chromedriver 版本，防Cloudflare)"""
driver = None # 先声明driver变量
try:
print(" (使用浏览器模式获取网页...)")
options = uc.ChromeOptions()
options.add_argument('--headless')
options.add_argument('--no-sandbox')
options.add_argument('--disable-dev-shm-usage')
options.add_argument('--disable-gpu') # 在某些服务器上需要
driver = uc.Chrome(options=options)
driver.get(url)
# 等待Cloudflare的JS验证完成，如果网络慢或验证复杂可适当增加
time.sleep(8)
html_content = driver.page_source
return html_content
except Exception as e:
print(f"❌ 使用浏览器获取网页失败: {e}")
return None
finally:
# 确保浏览器进程被关闭以释放资源
if driver:
driver.quit()
def extract_data(html_content, stock_pattern, price_pattern):
"""从HTML中提取库存和价格"""
stock, price = None, None
if stock_pattern and (match := re.search(stock_pattern, html_content, re.IGNORECASE)):
stock = int(match.group(1))
if price_pattern and (match := re.search(price_pattern, html_content, re.IGNORECASE)):
price = float(match.group(1))
return stock, price
def send_telegram_message(message):
"""发送Telegram消息"""
url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
data = {'chat_id': TELEGRAM_CHAT_ID, 'text': message, 'parse_mode': 'Markdown'}
try:
response = requests.post(url, data=data, timeout=10)
if response.status_code == 200:
print("✅ 消息发送成功")
else:
print(f"❌ 消息发送失败: {response.text}")
except Exception as e:
print(f"❌ 发送消息异常: {e}")
def format_change_message(product_name, old_stock, new_stock, old_price, new_price, url):
"""格式化变化消息"""
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
message = f"📦 *{product_name}* 监控提醒\n\n"
if new_stock is not None:
if old_stock is not None and old_stock != new_stock:
emoji = "📈" if new_stock > old_stock else "📉"
message += f"{emoji} 库存变化: {old_stock} → *{new_stock}*\n"
else:
message += f"📦 当前库存: *{new_stock}*\n"
if new_price is not None:
if old_price is not None and old_price != new_price:
emoji = "💰" if new_price < old_price else "💸"
message += f"{emoji} 价格变化: ¥{old_price} → *¥{new_price}*\n"
else:
message += f"💰 当前价格: *¥{new_price}*\n"
message += f"\n🕐 检查时间: {timestamp}\n🔗 [点击查看商品]({url})"
return message
def monitor_product(product, history):
"""监控单个商品"""
name = product['name']
print(f"🔍 检查商品: {name}")
html_content = get_page_content(product['url'])
if not html_content:
return
current_stock, current_price = extract_data(html_content, product['stock_pattern'], product['price_pattern'])
product_history = history.get(name, {})
old_stock, old_price = product_history.get('stock'), product_history.get('price')
print(f" 库存: {old_stock} → {current_stock}")
print(f" 价格: {old_price} → {current_price}")
if not product_history or old_stock != current_stock or old_price != current_price:
print(f" ✨ 发现状态变化，准备发送通知...")
message = format_change_message(name, old_stock, current_stock, old_price, current_price, product['url'])
send_telegram_message(message)
if current_stock is not None or current_price is not None:
history[name] = {'stock': current_stock, 'price': current_price}
def main():
"""主函数"""
print("🚀 库存监控脚本启动 (浏览器模式)")
print(f"📊 监控商品数量: {len(MONITOR_PRODUCTS)}")
print(f"⏰ 检查间隔: {CHECK_INTERVAL}秒")
print("-" * 50)
if "在这里替换" in TELEGRAM_BOT_TOKEN or "在这里替换" in TELEGRAM_CHAT_ID:
print("❌ 致命错误: 请先修改脚本文件，配置好你的 Telegram Bot Token 和 Chat ID!")
return
history = load_history()
while True:
print(f"\n🔄 开始检查 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
for product in MONITOR_PRODUCTS:
try:
monitor_product(product, history)
except Exception as e:
print(f"❌ 监控 {product.get('name', '未知商品')} 失败: {e}")
save_history(history)
print(f"✅ 本轮检查完成，等待 {CHECK_INTERVAL} 秒...")
time.sleep(CHECK_INTERVAL)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\n👋 监控脚本已由用户手动停止。")
except Exception as e:
print(f"\n💥 程序因无法恢复的严重错误而终止: {e}")
步骤 4.4: 配置脚本
- 在
nano编辑器里，找到配置区域。 - 将
TELEGRAM_BOT_TOKEN的值替换成你自己的Bot Token。 - 将
TELEGRAM_CHAT_ID的值替换成你自己的Chat ID。 - （可选）在
MONITOR_PRODUCTS列表中，修改商品名称，或添加/删除你想监控的商品。 - 配置完成后，按
Ctrl + X，然后按Y，最后按回车键，保存并退出。
第五部分：测试脚本 (Manual Test)
在正式部署前，先手动运行一次，确保一切正常。
python3 stock_monitor.py
如果你的配置都正确，你应该能看到脚本开始检查商品，并成功收到第一批Telegram通知。确认无误后，按 Ctrl + C 停止脚本。
第六部分：部署为系统服务 (Deployment as a Service)
为了让脚本能开机自启、无人值守地运行，我们将其创建为一个系统服务。
步骤 6.1: 创建服务文件
sudo nano /etc/systemd/system/stock-monitor.service
步骤 6.2: 粘贴服务配置
将下面的配置内容完整地复制并粘贴到编辑器中。
[Unit]
Description=Stock Monitor Service by Gemini
After=network.target
[Service]
Type=simple
User=ubuntu
WorkingDirectory=/home/ubuntu/stock_monitor
ExecStart=/usr/bin/python3 /home/ubuntu/stock_monitor/stock_monitor.py
Restart=always
RestartSec=15
[Install]
WantedBy=multi-user.target
注意：这里的
User和路径WorkingDirectory/ExecStart都是基于你的用户名是ubuntu。如果不是，请务必修改。
粘贴后，按 Ctrl + X -> Y -> 回车，保存并退出。
步骤 6.3: 启动并设置开机自启
依次运行以下命令：
# 重新加载systemd配置，让新服务生效
sudo systemctl daemon-reload
# 立即启动服务
sudo systemctl start stock-monitor
# 设置服务为开机自启
sudo systemctl enable stock-monitor
第七部分：服务管理与日志查看 (Management)
现在，你的脚本已经是一个真正的后台服务了。
- 查看服务状态:
sudo systemctl status stock-monitor.service看到绿色的
active (running)就代表一切正常。 - 查看实时日志:
journalctl -u stock-monitor.service -f - 停止服务:
sudo systemctl stop stock-monitor.service - 重启服务 (例如，在你修改了
stock_monitor.py的配置后):sudo systemctl restart stock-monitor.service
至此，你已成功搭建了一个强大、稳定、全自动的库存监控系统。恭喜！
0