reveyes 0.1.1

Python SDK for the Reveyes Amazon Review Scraping API

reveyes

English | 中文

---

Reveyes Python SDK — Amazon Review Scraping API

Collect Amazon product reviews at scale with a simple Python API.
Submit a list of ASINs → get back structured review data (rating, title, content, reviewer, verified purchase, images, etc.).

Supports 20 marketplaces across 5 regions:

Region	Marketplaces
North America	US · CA · MX
Europe	UK · DE · FR · IT · ES · NL · SE · PL · BE · IE
Asia	JP · IN · SG
Middle East	AE · SA
Others	AU · BR

Sign up free at www.reveyes.cn to get your API key.

---

Table of Contents

• Installation
• Quick Start
• Authentication
• Client Configuration
• API Reference
  • fetch_reviews
  • get_task_result
  • wait_for_task
  • iter_all_reviews
  • list_tasks
• Data Models
• Error Handling
• Credits & Pricing
• Full Examples

---

Installation

pip install reveyes

Requirements: Python 3.8 or higher. Only dependency: requests.

---

Quick Start

from reveyes import ReveyesClient

client = ReveyesClient(api_key="your_api_key_here")

# 1. Submit a scraping task
task = client.fetch_reviews([
    {"asin": "B08N5WRWNW", "marketplace": "US", "pages": 2},
    {"asin": "B09G9FPHY6", "marketplace": "DE", "pages": 1, "filter_star": "critical"},
])
print(f"Task submitted: {task.task_id}  |  credits reserved: {task.pre_deduct}")

# 2. Wait for completion (blocks, polls every 5 seconds)
result = client.wait_for_task(task.task_id, timeout=300)
print(f"Done! {result.reviews.total} reviews collected, {result.actual_deduct} credits used")

# 3. Iterate reviews
for review in result.reviews.data:
    print(f"[{review.marketplace}] {review.asin}  ★{review.rating}  {review.title}")
    print(f"  {review.review_content[:120]}...")

---

Authentication

Every request requires an X-API-Key header, handled automatically by the client.

1. Register at www.reveyes.cn
2. Go to Dashboard → API Keys → Create Key
3. Copy the key and pass it to ReveyesClient

client = ReveyesClient(api_key="rv_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")

Keep your API key secret. Do not hard-code it in source code.
Use environment variables instead:

import os
client = ReveyesClient(api_key=os.environ["REVEYES_API_KEY"])

---

Client Configuration

from reveyes import ReveyesClient

client = ReveyesClient(
    api_key="your_api_key",                            # Required
    base_url="https://server.reveyes.cn/api/openapi",  # Optional, default shown
    timeout=30,                                         # Optional, seconds
)

Parameter	Type	Default	Description
api_key	str	—	Your Reveyes API key (required)
base_url	str	https://server.reveyes.cn/api/openapi	API base URL
timeout	int	30	HTTP request timeout in seconds

---

API Reference

fetch_reviews

task: FetchTaskResult = client.fetch_reviews(asins)

Submit a review scraping task. Credits are pre-deducted upfront and any unused amount is refunded when the task finishes.

Parameters

Field	Type	Required	Default	Description
asin	str	✅	—	Amazon ASIN (e.g. "B08N5WRWNW")
marketplace	str		"US"	Marketplace code (see full list below). Default "US"
pages	int		1	Number of review pages to scrape (1–10). 1 page ≈ 10 reviews. Counts as 1 credit per page.
filter_star	str		"all_stars"	Star filter. Values: all_stars one_star two_star three_star four_star five_star positive critical
filter_sort_by	str		"recent"	Sort order. Values: recent helpful
filter_reviewer_type	str		"all_reviews"	Reviewer type. Values: all_reviews avp_only_reviews (verified purchases only)
filter_media_type	str		"all_contents"	Media filter. Values: all_contents media_reviews_only
filter_variant	str		"all_formats"	Variant filter. Values: all_formats current_format

Example

task = client.fetch_reviews([
    # Minimum — only ASIN required
    {"asin": "B08N5WRWNW"},

    # Full options
    {
        "asin": "B09G9FPHY6",
        "marketplace": "DE",
        "pages": 5,
        "filter_star": "critical",
        "filter_sort_by": "helpful",
        "filter_reviewer_type": "avp_only_reviews",
        "filter_media_type": "all_contents",
        "filter_variant": "all_formats",
    },

    # Using AsinFetchItem dataclass
    AsinFetchItem(asin="B07XJ8C8F7", marketplace="JP", pages=2, filter_star="five_star"),
])

Supported marketplace codes

Region	Code	Domain
🇺🇸 United States	US	amazon.com
🇨🇦 Canada	CA	amazon.ca
🇲🇽 Mexico	MX	amazon.com.mx
🇬🇧 United Kingdom	UK	amazon.co.uk
🇩🇪 Germany	DE	amazon.de
🇫🇷 France	FR	amazon.fr
🇮🇹 Italy	IT	amazon.it
🇪🇸 Spain	ES	amazon.es
🇳🇱 Netherlands	NL	amazon.nl
🇸🇪 Sweden	SE	amazon.se
🇵🇱 Poland	PL	amazon.pl
🇧🇪 Belgium	BE	amazon.com.be
🇮🇪 Ireland	IE	amazon.ie
🇯🇵 Japan	JP	amazon.co.jp
🇮🇳 India	IN	amazon.in
🇸🇬 Singapore	SG	amazon.sg
🇦🇪 UAE	AE	amazon.ae
🇸🇦 Saudi Arabia	SA	amazon.sa
🇦🇺 Australia	AU	amazon.com.au
🇧🇷 Brazil	BR	amazon.com.br

print(task.task_id)    # "7198234098234"  — use to poll
print(task.status)     # "pending"
print(task.pre_deduct) # 8  — credits reserved (1+5+2)
print(task.created_at) # datetime object

Returns: FetchTaskResult

Field	Type	Description
task_id	str	Unique task ID for polling
status	str	Always "pending" on creation
total_asins	int	Number of ASINs in the task
pre_deduct	int	Credits reserved upfront
created_at	datetime	Task creation time

---

get_task_result

result: TaskResult = client.get_task_result(task_id, page=1, page_size=50)

Query the current status and paginated review results of a task.

Parameters

Parameter	Type	Default	Description
task_id	str	—	Task ID from fetch_reviews
page	int	1	Page number for review pagination
page_size	int	50	Reviews per page (max 200)

Example

result = client.get_task_result(task.task_id, page=1, page_size=100)

# Task info
print(result.task_id)         # "7198234098234"
print(result.status)          # "pending" / "running" / "done" / "failed"
print(result.is_done)         # True if status == "done"
print(result.is_running)      # True if status in ("pending", "running")
print(result.total_asins)     # 3
print(result.finished_asins)  # 2  — how many ASINs completed so far
print(result.pre_deduct)      # 8  — credits originally reserved
print(result.actual_deduct)   # 7  — credits actually consumed (after refund)
print(result.created_at)      # datetime
print(result.finished_at)     # datetime or None

# Per-ASIN progress
for item in result.items_summary:
    print(item.asin, item.marketplace, item.status)
    print(f"  requested {item.pages} pages, scraped {item.actual_pages}, got {item.review_count} reviews")

# Reviews (paginated)
print(result.reviews.total)    # total across all pages
print(result.reviews.page)     # current page
print(result.reviews.page_size)

for review in result.reviews.data:
    print(review.asin)              # "B08N5WRWNW"
    print(review.marketplace)       # "US"
    print(review.review_id)         # "R3K2P8QXXX"
    print(review.rating)            # 1–5
    print(review.title)             # "Great product!"
    print(review.review_date)       # "2024-03-15"
    print(review.review_content)    # full review text
    print(review.user_name)         # "John D."
    print(review.profile_url)       # Amazon reviewer profile link
    print(review.verified_purchase) # 1 = verified, 0 = not verified
    print(review.helpful_votes)     # number of helpful votes
    print(review.product_variant)   # e.g. "Color: Black"
    print(review.images)            # list of image URLs or None
    print(review.videos)            # list of video data or None
    print(review.page)              # which scrape page this came from

Task status values

Status	Meaning
pending	Queued, not started yet
running	Actively scraping
done	All ASINs finished, reviews available
failed	Task failed (partial results may exist)

---

wait_for_task

result: TaskResult = client.wait_for_task(task_id, poll_interval=5.0, timeout=300.0)

Convenience method that polls until the task is done (or raises TimeoutError).

Parameters

Parameter	Type	Default	Description
task_id	str	—	Task ID to poll
poll_interval	float	5.0	Seconds between each poll request
timeout	float	300.0	Max seconds to wait before raising TimeoutError

Example

try:
    result = client.wait_for_task(task.task_id, poll_interval=5, timeout=600)
    print(f"Collected {result.reviews.total} reviews in {result.actual_deduct} credits")
except TimeoutError:
    print("Task took too long — check status manually later")

Tip: For large tasks (many pages), set timeout higher. Each ASIN page takes roughly 2–5 seconds.

---

iter_all_reviews

reviews: Iterator[ReviewItem] = client.iter_all_reviews(task_id, page_size=100)

Automatically paginate through all reviews of a completed task. Yields one ReviewItem at a time without loading everything into memory.

Parameters

Parameter	Type	Default	Description
task_id	str	—	Task ID of a completed task
page_size	int	100	Reviews fetched per API call (max 200)

Example

# Count by rating
from collections import Counter
ratings = Counter()
for review in client.iter_all_reviews(task.task_id):
    ratings[review.rating] += 1
print(ratings)  # Counter({5: 320, 4: 95, 3: 28, 2: 12, 1: 45})

# Save to file
with open("reviews.jsonl", "w") as f:
    import json, dataclasses
    for review in client.iter_all_reviews(task.task_id):
        f.write(json.dumps(dataclasses.asdict(review), ensure_ascii=False, default=str) + "\n")

---

list_tasks

tasks: TaskList = client.list_tasks(page=1, page_size=20)

List your historical scraping tasks, newest first.

Parameters

Parameter	Type	Default	Description
page	int	1	Page number
page_size	int	20	Tasks per page (max 100)

Example

tasks = client.list_tasks(page=1, page_size=20)
print(f"Total tasks: {tasks.total}")

for t in tasks.items:
    print(f"{t.task_id}  {t.status:<10}  ASINs={t.total_asins}  credits={t.actual_deduct}  {t.created_at:%Y-%m-%d %H:%M}")

---

Data Models

ReviewItem

Field	Type	Description
asin	str | None	Product ASIN
marketplace	str | None	Marketplace code (US/UK/...)
review_id	str | None	Amazon review ID
rating	int | None	Star rating (1–5)
title	str | None	Review headline
review_content	str | None	Full review body
review_date	str | None	Date string, e.g. "2024-03-15"
user_name	str | None	Reviewer display name
profile_url	str | None	Full URL to reviewer's Amazon profile
verified_purchase	int	1 = verified, 0 = not verified
helpful_votes	int	Number of helpful votes
product_variant	str | None	e.g. "Size: Large, Color: Blue"
images	Any | None	List of image URLs attached to review
videos	Any | None	List of video data attached to review
page	int	Which scrape page this review came from

TaskResult

Field	Type	Description
task_id	str	Task ID
status	str	pending / running / done / failed
total_asins	int	Number of ASINs submitted
finished_asins	int	ASINs completed (done or failed)
pre_deduct	int	Credits reserved upfront
actual_deduct	int	Credits actually consumed
created_at	datetime	Task creation time
finished_at	datetime | None	Task completion time
items_summary	list[TaskItemSummary]	Per-ASIN scraping summary
reviews	ReviewsPage	Paginated reviews
is_done	bool	Shortcut: status == "done"
is_running	bool	Shortcut: status in ("pending", "running")

---

Error Handling

All exceptions inherit from reveyes.ReveyesError.

from reveyes.exceptions import (
    AuthenticationError,       # code 1001 — API key invalid or disabled
    InsufficientCreditsError,  # code 1002 — not enough credits
    BadParamsError,            # code 1003 — invalid request parameters
    NotFoundError,             # code 1004 — task not found
    ForbiddenError,            # code 1005 — access denied to another user's task
    APIError,                  # other non-zero response codes
    ReveyesError,              # base class for all above
)

try:
    task = client.fetch_reviews([
        {"asin": "B08N5WRWNW", "marketplace": "XX"}  # invalid marketplace
    ])
except BadParamsError as e:
    print(f"Bad params: {e}")
except InsufficientCreditsError as e:
    print(f"Need more credits — top up at https://www.reveyes.cn")
except AuthenticationError:
    print("API key is invalid. Check your key at https://www.reveyes.cn")
except APIError as e:
    print(f"API error {e.code}: {e}")

---

Credits & Pricing

Unit	Cost
1 page scraped	1 credit
1 page ≈	~10 reviews

• Credits are pre-deducted when a task is submitted
• Unused credits (e.g. if a page has 0 results) are automatically refunded on task completion
• View your balance and recharge at www.reveyes.cn

---

Full Examples

Export all critical reviews to CSV

import csv
import os
from reveyes import ReveyesClient

client = ReveyesClient(api_key=os.environ["REVEYES_API_KEY"])

asins = ["B08N5WRWNW", "B09G9FPHY6", "B07XJ8C8F7"]

task = client.fetch_reviews([
    {"asin": asin, "marketplace": "US", "pages": 5, "filter_star": "critical"}
    for asin in asins
])
print(f"Task {task.task_id} submitted, {task.pre_deduct} credits reserved")

result = client.wait_for_task(task.task_id, timeout=600)
print(f"Done — {result.reviews.total} reviews, {result.actual_deduct} credits used")

with open("critical_reviews.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.writer(f)
    writer.writerow(["asin", "marketplace", "rating", "date", "title", "content",
                     "verified", "helpful_votes", "reviewer", "profile_url"])
    for review in client.iter_all_reviews(task.task_id, page_size=200):
        writer.writerow([
            review.asin, review.marketplace, review.rating, review.review_date,
            review.title, review.review_content, review.verified_purchase,
            review.helpful_votes, review.user_name, review.profile_url,
        ])
print("Saved to critical_reviews.csv")

Multi-marketplace comparison

from collections import defaultdict
from reveyes import ReveyesClient

client = ReveyesClient(api_key="your_key")

asin = "B08N5WRWNW"
task = client.fetch_reviews([
    {"asin": asin, "marketplace": m, "pages": 3}
    for m in ["US", "UK", "DE", "JP", "CA", "AU"]
])

result = client.wait_for_task(task.task_id)

stats = defaultdict(lambda: {"count": 0, "total_rating": 0})
for review in client.iter_all_reviews(task.task_id):
    s = stats[review.marketplace]
    s["count"] += 1
    s["total_rating"] += review.rating or 0

for market, s in sorted(stats.items()):
    avg = s["total_rating"] / s["count"] if s["count"] else 0
    print(f"{market}: {s['count']} reviews, avg ★{avg:.2f}")

---

License

MIT © Reveyes

---

---

Reveyes Python SDK — 亚马逊评论抓取 API

用最简单的方式批量抓取亚马逊商品评论。提交 ASIN 列表，获取结构化评论数据（星级、标题、内容、评论者、已验证购买、图片等）。

支持 20 个站点，覆盖 5 大区域：

区域	站点
北美	US（美国）· CA（加拿大）· MX（墨西哥）
欧洲	UK · DE · FR · IT · ES · NL · SE · PL · BE · IE
亚洲	JP（日本）· IN（印度）· SG（新加坡）
中东	AE（阿联酋）· SA（沙特）
其他	AU（澳大利亚）· BR（巴西）

在 www.reveyes.cn 免费注册 获取 API Key。

---

目录

• 安装
• 快速开始
• 鉴权说明
• 客户端配置
• API 参考
  • fetch_reviews — 提交抓取任务
  • get_task_result — 查询任务结果
  • wait_for_task — 等待任务完成
  • iter_all_reviews — 迭代全部评论
  • list_tasks — 历史任务列表
• 数据模型
• 异常处理
• 积分与计费
• 完整示例

---

安装

pip install reveyes

要求：Python 3.8+，仅依赖 requests，无其他重型依赖。

---

快速开始

from reveyes import ReveyesClient

client = ReveyesClient(api_key="你的API Key")

# 1. 提交抓取任务
task = client.fetch_reviews([
    {"asin": "B08N5WRWNW", "marketplace": "US", "pages": 2},
    {"asin": "B09G9FPHY6", "marketplace": "DE", "pages": 1, "filter_star": "critical"},
])
print(f"任务已提交：{task.task_id}，预扣积分：{task.pre_deduct}")

# 2. 等待完成（自动轮询，每 5 秒一次）
result = client.wait_for_task(task.task_id, timeout=300)
print(f"完成！共抓取 {result.reviews.total} 条评论，实际消耗 {result.actual_deduct} 积分")

# 3. 遍历评论
for review in result.reviews.data:
    print(f"[{review.marketplace}] {review.asin}  ★{review.rating}  {review.title}")
    print(f"  {review.review_content[:120]}...")

---

鉴权说明

每个请求都需要在 Header 中携带 X-API-Key，SDK 会自动处理。

1. 前往 www.reveyes.cn 注册账号
2. 进入 控制台 → API Keys → 创建 Key
3. 复制 Key，传入 ReveyesClient

client = ReveyesClient(api_key="rv_xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx")

⚠️ 请妥善保管 API Key，不要硬编码在代码里，建议用环境变量：

import os
client = ReveyesClient(api_key=os.environ["REVEYES_API_KEY"])

---

客户端配置

from reveyes import ReveyesClient

client = ReveyesClient(
    api_key="你的API Key",                             # 必填
    base_url="https://server.reveyes.cn/api/openapi",  # 可选，默认值如左
    timeout=30,                                         # 可选，单位秒
)

参数	类型	默认值	说明
api_key	str	—	API Key（必填）
base_url	str	https://server.reveyes.cn/api/openapi	API 地址
timeout	int	30	HTTP 请求超时秒数

---

API 参考

fetch_reviews — 提交抓取任务

task: FetchTaskResult = client.fetch_reviews(asins)

提交评论抓取任务。积分按页数预扣，任务完成后多余部分自动退还。

每个 ASIN 的参数

字段	类型	必填	默认值	说明
asin	str	✅	—	亚马逊 ASIN，如 "B08N5WRWNW"
marketplace	str		"US"	站点代码，见下方完整列表
pages	int		1	抓取页数（1–10）。1 页 ≈ 10 条评论，消耗 1 积分/页
filter_star	str		"all_stars"	星级筛选：all_stars one_star two_star three_star four_star five_star positive critical
filter_sort_by	str		"recent"	排序方式：recent（最新）helpful（最有用）
filter_reviewer_type	str		"all_reviews"	评论类型：all_reviews avp_only_reviews（仅已验证购买）
filter_media_type	str		"all_contents"	媒体类型：all_contents media_reviews_only（含图片/视频）
filter_variant	str		"all_formats"	变体筛选：all_formats current_format

示例

task = client.fetch_reviews([
    # 最简写法，只填 ASIN
    {"asin": "B08N5WRWNW"},

    # 完整参数
    {
        "asin": "B09G9FPHY6",
        "marketplace": "DE",
        "pages": 5,
        "filter_star": "critical",         # 只抓差评
        "filter_sort_by": "helpful",       # 按有用性排序
        "filter_reviewer_type": "avp_only_reviews",  # 仅已验证购买
    },
])

print(task.task_id)     # "7198234098234" — 用于后续查询
print(task.status)      # "pending"
print(task.pre_deduct)  # 6 — 预扣积分（1+5）

支持的站点代码（20 个）

地区	代码	域名
🇺🇸 美国	US	amazon.com
🇨🇦 加拿大	CA	amazon.ca
🇲🇽 墨西哥	MX	amazon.com.mx
🇬🇧 英国	UK	amazon.co.uk
🇩🇪 德国	DE	amazon.de
🇫🇷 法国	FR	amazon.fr
🇮🇹 意大利	IT	amazon.it
🇪🇸 西班牙	ES	amazon.es
🇳🇱 荷兰	NL	amazon.nl
🇸🇪 瑞典	SE	amazon.se
🇵🇱 波兰	PL	amazon.pl
🇧🇪 比利时	BE	amazon.com.be
🇮🇪 爱尔兰	IE	amazon.ie
🇯🇵 日本	JP	amazon.co.jp
🇮🇳 印度	IN	amazon.in
🇸🇬 新加坡	SG	amazon.sg
🇦🇪 阿联酋	AE	amazon.ae
🇸🇦 沙特	SA	amazon.sa
🇦🇺 澳大利亚	AU	amazon.com.au
🇧🇷 巴西	BR	amazon.com.br

---

get_task_result — 查询任务结果

result: TaskResult = client.get_task_result(task_id, page=1, page_size=50)

查询任务当前状态与分页评论数据。

参数

参数	类型	默认值	说明
task_id	str	—	任务 ID
page	int	1	评论分页页码
page_size	int	50	每页评论数（最大 200）

任务状态说明

状态	含义
pending	已排队，等待执行
running	正在抓取中
done	全部 ASIN 完成，评论可取
failed	任务失败（可能有部分结果）

示例

result = client.get_task_result(task.task_id, page=1, page_size=100)

print(result.status)          # "done"
print(result.is_done)         # True
print(result.total_asins)     # 2
print(result.finished_asins)  # 2
print(result.pre_deduct)      # 6 — 预扣积分
print(result.actual_deduct)   # 5 — 实际消耗（差额已退还）
print(result.reviews.total)   # 总评论数

for review in result.reviews.data:
    print(f"★{review.rating} [{review.marketplace}] {review.asin}")
    print(f"  标题：{review.title}")
    print(f"  内容：{review.review_content[:100]}...")
    print(f"  评论者：{review.user_name}  已验证购买：{'是' if review.verified_purchase else '否'}")
    print(f"  有用数：{review.helpful_votes}  日期：{review.review_date}")

---

wait_for_task — 等待任务完成

result: TaskResult = client.wait_for_task(task_id, poll_interval=5.0, timeout=300.0)

阻塞等待直到任务完成，内部自动轮询。

参数

参数	类型	默认值	说明
task_id	str	—	任务 ID
poll_interval	float	5.0	每次轮询间隔（秒）
timeout	float	300.0	最长等待时间（秒），超时抛出 TimeoutError

try:
    result = client.wait_for_task(task.task_id, poll_interval=5, timeout=600)
except TimeoutError:
    print("任务超时，稍后手动查询")

提示：任务时长取决于 ASIN 数量和页数，通常每页需要 2–5 秒。大批量任务建议适当调高 timeout。

---

iter_all_reviews — 迭代全部评论

for review in client.iter_all_reviews(task_id, page_size=100):
    ...

自动处理分页，逐条 yield 评论，不一次性加载全部数据到内存，适合大量评论场景。

参数

参数	类型	默认值	说明
task_id	str	—	已完成任务的 ID
page_size	int	100	每次 API 请求拉取的评论数（最大 200）

# 统计各星级数量
from collections import Counter
counter = Counter(r.rating for r in client.iter_all_reviews(task.task_id))
for star in range(5, 0, -1):
    print(f"{'★' * star}  {counter[star]} 条")

---

list_tasks — 历史任务列表

tasks: TaskList = client.list_tasks(page=1, page_size=20)

分页查询历史任务，按创建时间倒序排列。

tasks = client.list_tasks()
print(f"共 {tasks.total} 个历史任务")
for t in tasks.items:
    status_cn = {"pending": "排队中", "running": "进行中", "done": "已完成", "failed": "失败"}.get(t.status, t.status)
    print(f"{t.task_id}  {status_cn}  ASIN数={t.total_asins}  积分={t.actual_deduct}  {t.created_at:%Y-%m-%d %H:%M}")

---

数据模型

ReviewItem — 单条评论

字段	类型	说明
asin	str | None	商品 ASIN
marketplace	str | None	站点（US/UK/DE 等）
review_id	str | None	亚马逊评论 ID
rating	int | None	星级（1–5）
title	str | None	评论标题
review_content	str | None	评论正文
review_date	str | None	评论日期，如 "2024-03-15"
user_name	str | None	评论者昵称
profile_url	str | None	评论者主页完整链接
verified_purchase	int	1=已验证购买，0=未验证
helpful_votes	int	有用票数
product_variant	str | None	商品变体，如 "颜色：黑色"
images	Any | None	评论附图 URL 列表
videos	Any | None	评论附视频数据
page	int	来自第几页抓取结果

---

异常处理

所有异常均继承自 reveyes.ReveyesError。

from reveyes.exceptions import (
    AuthenticationError,       # 1001 — API Key 无效或已禁用
    InsufficientCreditsError,  # 1002 — 积分不足
    BadParamsError,            # 1003 — 参数错误（如不支持的站点代码）
    NotFoundError,             # 1004 — 任务不存在
    ForbiddenError,            # 1005 — 无权访问他人任务
    APIError,                  # 其他 API 错误
    ReveyesError,              # 所有异常的基类
)

try:
    task = client.fetch_reviews([{"asin": "B08N5WRWNW", "pages": 10}])
except InsufficientCreditsError:
    print("积分不足，请前往 https://www.reveyes.cn 充值")
except AuthenticationError:
    print("API Key 无效，请检查")
except BadParamsError as e:
    print(f"参数错误：{e}")
except APIError as e:
    print(f"接口异常 code={e.code}：{e}")

---

积分与计费

单位	消耗
每页抓取	1 积分
1 页 ≈	约 10 条评论

• 提交任务时按页数预扣积分
• 任务完成后，未实际抓取到数据的页数自动退还积分
• 余额查询和充值：www.reveyes.cn

---

完整示例

导出差评到 CSV

import csv
import os
from reveyes import ReveyesClient

client = ReveyesClient(api_key=os.environ["REVEYES_API_KEY"])

asins = ["B08N5WRWNW", "B09G9FPHY6", "B07XJ8C8F7"]

task = client.fetch_reviews([
    {"asin": asin, "marketplace": "US", "pages": 5, "filter_star": "critical"}
    for asin in asins
])
print(f"任务 {task.task_id} 已提交，预扣 {task.pre_deduct} 积分，等待完成...")

result = client.wait_for_task(task.task_id, timeout=600)
print(f"完成！共 {result.reviews.total} 条评论，实耗 {result.actual_deduct} 积分")

with open("差评汇总.csv", "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.writer(f)
    writer.writerow(["ASIN", "站点", "星级", "日期", "标题", "内容", "已验证购买", "有用数", "评论者"])
    for review in client.iter_all_reviews(task.task_id, page_size=200):
        writer.writerow([
            review.asin, review.marketplace, review.rating, review.review_date,
            review.title, review.review_content,
            "是" if review.verified_purchase else "否",
            review.helpful_votes, review.user_name,
        ])
print("已保存到 差评汇总.csv")

多站点评分对比

from collections import defaultdict
from reveyes import ReveyesClient

client = ReveyesClient(api_key="your_key")

asin = "B08N5WRWNW"
task = client.fetch_reviews([
    {"asin": asin, "marketplace": m, "pages": 3}
    for m in ["US", "UK", "DE", "JP", "CA", "AU", "IN"]
])

result = client.wait_for_task(task.task_id)
stats = defaultdict(lambda: {"count": 0, "sum": 0})
for review in client.iter_all_reviews(task.task_id):
    stats[review.marketplace]["count"] += 1
    stats[review.marketplace]["sum"] += review.rating or 0

print(f"\n{'站点':<8} {'评论数':>8} {'平均分':>8}")
print("-" * 28)
for market, s in sorted(stats.items()):
    avg = s["sum"] / s["count"] if s["count"] else 0
    bar = "★" * round(avg)
    print(f"{market:<8} {s['count']:>8} {avg:>7.2f}  {bar}")

---

许可证

MIT © Reveyes