> ## Documentation Index
> Fetch the complete documentation index at: https://docs.brightdata.com/llms.txt
> Use this file to discover all available pages before exploring further.
# 构建每日亚马逊价格监控器
> 使用Bright Data亚马逊爬虫API安排每日抓取亚马逊SKU列表，直接传输到S3存储桶，由GitHub Actions定时工作流进行编排。
你可以跟踪一份 Amazon SKU 列表中的价格和库存，并在每天早晨团队开始工作前获取最新数据。你不需要仅为此运行服务器，也不需要在笔记本电脑上监护 cron 作业。
在本教程中，我们将构建完全相同的管道。你将向 GitHub 仓库提交 SKU 列表，编写一个小的 Python 脚本来触发 Bright Data Amazon Scraper API，将脚本包装在每日运行的 GitHub Actions 工作流中，并配置 Bright Data 将结果直接交付到 S3 桶。每天早晨，一个新的 JSON 文件会被放入 S3，以快照 ID 为键，准备供你的 BI 管道使用。
无需服务器，无需 webhook 处理程序，无需胶合代码。只需一个工作流文件、一个脚本和一个交付配置。
## 你将构建什么
一个包含以下内容的 GitHub 仓库：
1. 一个 `skus.json` 文件，列出要监控的 Amazon 产品 URL
2. 一个 Python 脚本，将 SKU 列表 POST 到 Bright Data Amazon Scraper API
3. 一个 GitHub Actions 工作流，按每日计划运行脚本
4. Bright Data 配置为将每个快照交付到你的 S3 桶
最后，你每天都会在 S3 中看到一个新的 JSON 文件，每个文件都包含列表中每个 SKU 的最新价格、评级和可用性数据。
\*\*预计耗时：\*\*30 分钟。
## 前提条件
* 一个 [Bright Data 账户](https://www.bright.cn/cp/start)，包括 API 密钥（[获取你的密钥](https://www.bright.cn/cp/setting/users)）
* 一个已配置 Bright Data 交付的 S3 桶。请按照 [Amazon 到 S3 交付](/products/scrapers/scrapers-library/data-delivery) 完成一次，然后返回。本教程假设交付目标已保存在你的 Amazon Scraper 设置中。
* 一个 [GitHub 账户](https://github.com/signup)和一个新的（空）仓库
* 本地安装的 Python 3.9+
* 本地安装的 Git
## 第 1 部分：创建 SKU 列表
在本地克隆你的空 GitHub 仓库，并在仓库根目录创建 `skus.json` 文件：
```json skus.json theme={null}
[
"https://www.amazon.com/dp/B0D1XD1ZV3",
"https://www.amazon.com/dp/B0863TXGM3",
"https://www.amazon.com/dp/B09V3KXJPB"
]
```
这些是三个真实的产品 URL（AirPods Pro 2、Sony WH-1000XM4 耳机和 iPad Air M1）。稍后将其替换为你自己的 SKU。
将 SKU 列表保存在仓库中意味着每次编辑都会被版本控制，每次更改都会通过常规拉取请求审查过程。大型 SKU 列表可以存储在脚本加载的 CSV 文件中——我们将在"后续步骤"中提到这一点。
## 第 2 部分：编写触发脚本
在仓库根目录创建 `trigger_scrape.py`：
```python trigger_scrape.py theme={null}
import json
import os
import sys
import requests
DATASET_ID = "gd_l7q7dkf244hwjntr0" # Amazon products by URL
TRIGGER_URL = "https://api.brightdata.com/datasets/v3/trigger"
def main() -> int:
api_key = os.environ.get("BRIGHT_DATA_API_KEY")
if not api_key:
print("BRIGHT_DATA_API_KEY environment variable is not set.")
return 1
with open("skus.json") as f:
urls = json.load(f)
payload = [{"url": url} for url in urls]
response = requests.post(
TRIGGER_URL,
params={"dataset_id": DATASET_ID, "format": "json"},
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json=payload,
timeout=30,
)
response.raise_for_status()
snapshot_id = response.json().get("snapshot_id")
print(f"Triggered scrape for {len(urls)} SKUs. Snapshot: {snapshot_id}")
return 0
if __name__ == "__main__":
sys.exit(main())
```
然后创建 `requirements.txt`：
```
requests==2.32.3
```
有两点值得注意：
* \*\*脚本不等待结果。\*\*它触发脚本然后退出。Bright Data 异步运行爬取，并通过你在 Scraper 设置中保存的交付配置直接将结果交付到 S3。这是整个要点：脚本是一个廉价的、无状态的触发器。
* \*\*API 密钥来自环境变量。\*\*永远不要将密钥提交到仓库。我们将在第 4 部分将其关联到 GitHub Actions Secrets。
## 第 3 部分：在本地运行
安装依赖项并使用你的密钥运行脚本：
```bash theme={null}
pip install -r requirements.txt
export BRIGHT_DATA_API_KEY=your_actual_key_here
python trigger_scrape.py
```
你应该看到类似的输出：
```
Triggered scrape for 3 SKUs. Snapshot: sd_mntfmunq1yy7gi201q
```
等待 60 到 90 秒，然后检查你的 S3 桶：
```bash theme={null}
aws s3 ls s3://your-bucket-name/amazon/products/
```
你应该看到一个以快照 ID 命名的新文件：
```
2026-04-10 09:14:22 14382 sd_mntfmunq1yy7gi201q.json
```
下载并检查一条记录：
```bash theme={null}
aws s3 cp s3://your-bucket-name/amazon/products/sd_mntfmunq1yy7gi201q.json ./latest.json
python -m json.tool latest.json | head -30
```
你应该看到每个 SKU 的结构化产品数据：
```json theme={null}
[
{
"title": "Sony WH-1000XM4 Wireless Premium Noise Canceling Overhead Headphones",
"asin": "B0863TXGM3",
"brand": "Sony",
"final_price": 209.99,
"currency": "USD",
"rating": 4.6,
"reviews_count": 62492,
"availability": "Only 1 left in stock - order soon.",
"url": "https://www.amazon.com/dp/B0863TXGM3"
}
]
```
价格字段是 `final_price`，对于缺货或通货不明确的商品，它可以是 `null`。你的 BI 管道应该显式处理这种情况，而不是在缺少键时崩溃。
注意该文件按 `snapshot_id` 而非日期来命名。这是有意为之的：每个快照都是不可变的，你可以按创建时间戳或启用版本控制来按时间顺序遍历桶。我们将在"后续步骤"中讨论命名约定。
## 第 4 部分：在 GitHub Actions 上调度工作流
现在让我们将触发器从你的笔记本电脑移到每日计划。
创建 `.github/workflows/daily-scrape.yml`：
```yaml .github/workflows/daily-scrape.yml theme={null}
name: Daily Amazon price scrape
on:
schedule:
- cron: "0 6 * * *" # 06:00 UTC every day
workflow_dispatch: # Allows manual runs from the Actions tab
jobs:
trigger:
runs-on: ubuntu-latest
steps:
- name: Check out the repo
uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: "3.11"
- name: Install dependencies
run: pip install -r requirements.txt
- name: Trigger Bright Data scrape
env:
BRIGHT_DATA_API_KEY: ${{ secrets.BRIGHT_DATA_API_KEY }}
run: python trigger_scrape.py
```
两个关键要素：
* **`schedule: cron: "0 6 * * *"`** 每天在 06:00 UTC 运行作业。调整 cron 表达式以适应你的时区。GitHub 的计划工作流没有保证的精度，但每日运行通常在计划时间的几分钟内触发。
* **`workflow_dispatch`** 在 Actions 选项卡中添加一个 **Run workflow** 按钮，以便你无需等待计划即可按需启动作业。
现在将你的 Bright Data 密钥添加为仓库密钥：
1. 在你的 GitHub 仓库中，转到 **Settings** > **Secrets and variables** > **Actions**
2. 点击 **New repository secret**
3. 将其命名为 `BRIGHT_DATA_API_KEY` 并粘贴你的密钥
4. 点击 **Add secret**
对待你的 Bright Data API 密钥如同对待密码。永远不要将其提交到仓库，永远不要将其粘贴到工作流日志中，如果你怀疑它已被泄露，请轮转它。
## 第 5 部分：推送并验证
提交所有内容并推送：
```bash theme={null}
git add skus.json trigger_scrape.py requirements.txt .github/workflows/daily-scrape.yml
git commit -m "Add daily Amazon price monitor"
git push
```
在 GitHub 上打开你的仓库，转到 **Actions** 选项卡。你应该看到 **Daily Amazon price scrape** 工作流列出。
点击 **Run workflow** > **Run workflow** 手动启动。在几秒钟内会出现一个新的运行。点击进入它并观看步骤执行。最后一步应该记录：
```
Triggered scrape for 3 SKUs. Snapshot: sd_mntfn4abcdefghij
```
等待一分钟，然后再检查 S3：
```bash theme={null}
aws s3 ls s3://your-bucket-name/amazon/products/
```
你现在应该看到两个快照文件——来自第 3 部分的文件和来自工作流运行的文件。
从这里开始，GitHub Actions 将在计划时间每天运行工作流，每天早晨会有一个新文件出现在 S3 中。无需服务器，无需 cron 作业，无需监护。
## 恭喜
你已经构建了一个完全自动化的每日价格监控器：
* 在 GitHub 中版本控制的 **SKU 列表**
* 触发 Bright Data 爬取然后退出的 **Python 触发脚本**
* 按每日 cron 运行并通过仓库密钥进行身份验证的 **GitHub Actions 工作流**
* 将每个快照异步放入你的桶中的 **Bright Data S3 交付**
每个活动部分都是声明式的并且在仓库中。编辑 SKU 列表、cron 计划或目标是一个单行拉取请求。
## 后续步骤
使用 `stream_max_lines` 在第一条记录准备好后立即开始接收批次。
异步触发端点的完整参数列表，包括 `include_errors` 和 `limit_per_input`。
以编程方式从工作流内部检查快照状态和交付结果。
使用相同的触发调用将 S3 替换为 GCS、Azure、Snowflake 或 SFTP。