工作流

通过 client.workflows 提供的组合辅助函数,可执行多个 SDK 调用并返回一个合并结果。

所有工作流方法都位于 client.workflows 下。每个方法都替代了一系列原本需要您自行编写的常规 SDK 调用,并返回一个在客户端汇总的结果。工作流不对应单一的 REST 端点,因此结果中的键使用 SDK 自己定义的名称,而不是 REST 响应中传输的字段名。

由于工作流会执行多次调用,运行工作流所用的 API 密钥需要具备访问工作流涉及的每个资源的权限,而且一次工作流调用会按底层请求的次数计入您的速率限制。各方法接受一个 location_id,它支持数字 ID 或 base64 编码的 ID,详见地点 ID。


workflows.auto_reply_to_reviews

通过一次调用,使用模板回复某个地点的评价。您可以用它保持对正面评价的回复率,同时手动处理其余评价。

Python
client.workflows.auto_reply_to_reviews(    location_id: str | int,    *,    template: str = "Thank you for your feedback!",    min_rating: int = 4,    only_unanswered: bool = True,    dry_run: bool = False,) -> list[dict[str, Any]]
参数类型说明
location_idstr | int要处理的地点。
templatestr用于回复匹配评价的文本。{rating} 会替换为该评价的评分。
min_ratingint仅回复评分达到或高于此值的评价。默认值为 4。
only_unansweredbool跳过已有回复的评价。默认值为 True。
dry_runbool返回将要回复的评价,但不发布回复。默认值为 False。

它替代了以下调用过程:使用 auto_paging_iter() 分页调用 reviews.list,然后针对每条同时满足两个筛选条件的评价调用一次 reviews.respond。

Python
# See what would be sent, without postingplan = client.workflows.auto_reply_to_reviews(  16808,  template="Thanks for the {rating}-star review!",  min_rating=4,  dry_run=True,)for entry in plan:  print(entry["id"], entry["rating"], entry["status"]) # Then post for realresults = client.workflows.auto_reply_to_reviews(16808, min_rating=4)print(f"Replied to {len(results)} reviews")

返回一个普通的字典列表,每条处理过的评价对应一个字典,每个字典都包含 id、rating、reply 和 status。请先使用 dry_run=True 运行:返回条目的 status 会设为 dry_run,以便您在发布任何回复之前检查选中的评价。


workflows.listings_health_audit

通过一次调用获取某个地点的商家信息概况:高级商家信息、语音商家信息、重复商家信息和评分。您可以用它构建单个地点的健康状况面板,替代 3 次单独的调用。

Python
client.workflows.listings_health_audit(location_id: str | int) -> APIObject

它替代了针对同一地点的 listings.premium、listings.voice 和 listings.duplicates 调用,并按 syncStatus 对高级商家信息结果进行计数。

Python
audit = client.workflows.listings_health_audit(16808)print(f"Health score: {audit.health_score}%")print(f"Synced: {audit.synced_count} of {len(audit.premium)}")print(f"Duplicates found: {len(audit.duplicates)}")for issue in audit.issues:  print(issue.get("site"), issue.get("syncStatus"))

返回一个 APIObject,包含 location_id、premium、voice、duplicates、synced_count、issue_count、issues 和 health_score。health_score 表示 syncStatus 为 SYNCED 的高级商家信息所占的比例,以整数百分比表示;没有高级商家信息时,其值为 0。issues 包含 syncStatus 存在且不为 SYNCED 的高级商家信息。


workflows.weekly_reputation_report

获取某个地点在指定日期范围内的评价、评价分析、资料分析和商家信息同步状态。由于它会执行 5 次调用,建议将其用于定期报告,而非实时控制台。

Python
client.workflows.weekly_reputation_report(    location_id: str | int,    *,    start_date: str | None = None,    end_date: str | None = None,) -> APIObject

start_date 和 end_date 都是 YYYY-MM-DD 字符串,且均为可选项。它替代了 reviews.analytics.overview、带有 first=50 的 reviews.list、analytics.google、analytics.bing 和 listings.premium 调用,并将日期范围传递给每个接受该参数的调用。

Python
report = client.workflows.weekly_reputation_report(  16808,  start_date="2026-09-01",  end_date="2026-09-07",)print(report.review_summary.get("averageRating"))print(report.analytics.get("google", {}).get("views"))print(report.listings_health.get("sync_rate"))

返回一个 APIObject,包含 location_id、review_summary(评价分析概览)、recent_reviews(该日期范围内最多 50 条评价)、analytics(包含 google 和 bing 的字典)以及 listings_health(包含 total、synced 和 sync_rate 的字典)。sync_rate 是格式化的百分比字符串,当该地点没有高级商家信息时,其值为 N/A。


另请参阅