使用案例
10 個完整的 Python 實作範例,涵蓋批次匯出、評論自動化、分析報告、商家資訊稽核及批次發布貼文。
執行任何範例前,請先設定一次您的 API 金鑰:
export LISTINGSAPI_KEY="your-api-key"1. 列出您的地點
建立連線並印出第一頁地點資料。
import listingsapi client = listingsapi.ListingsAPI() page = client.locations.list(first=5)for loc in page: print(loc.name, "--", getattr(loc, "city", "N/A"), getattr(loc, "stateIso", "N/A")) print(f"Locations on page: {len(page)}, has_more: {page.has_more}")2. 批次匯出為 CSV
使用 auto_paging_iter() 將所有地點下載為 CSV。
import csvimport listingsapi client = listingsapi.ListingsAPI() locations = list(client.locations.list(first=100).auto_paging_iter())print(f"Fetched {len(locations)} locations") fields = ["id", "name", "storeId", "street", "city", "stateIso", "postalCode", "countryIso", "phone"] with open("locations_export.csv", "w", newline="") as f: writer = csv.DictWriter(f, fieldnames=fields, extrasaction="ignore") writer.writeheader() for loc in locations: writer.writerow(loc.to_dict()) print("Exported to locations_export.csv")auto_paging_iter() 會自動處理所有游標分頁。完整的方法參考資料請見 地點。
3. 建立地點
locations.add() 以關鍵字引數接收所有必填欄位,並在用戶端驗證酬載,因此無效的請求會在發出任何網路呼叫前立即失敗,並提供清楚的訊息。請透過 client.subcategories() 查詢有效的 sub_category_id(使用 databaseId)。
import listingsapifrom listingsapi import ValidationError client = listingsapi.ListingsAPI() subcategories = client.subcategories()dental = [s for s in subcategories if "dentist" in (s.get("name") or "").lower()]sub_category_id = dental[0].databaseId if dental else 1432 try: result = client.locations.add( name="Acme Dental", description=( "Acme Dental is a family-owned dental practice serving downtown " "New York for over fifteen years. Our team offers preventive care, " "cosmetic dentistry, orthodontics, and emergency appointments, with " "weekend hours and a patient-first approach that keeps every visit " "comfortable and affordable for the whole family." ), sub_category_id=sub_category_id, country_iso="US", city="New York", street="123 Jump Street", state_iso="NY", postal_code="10013", phone="6443859313", store_id="NYC-001", )except ValidationError as e: print(f"Fix the payload: {e}")else: print(f"Created location: {result.location.id}")說明至少須有 200 個字元;發布平台會將其作為商家資訊的主要文案。client.countries() 列出支援的國家與州 ISO 代碼,而 client.plan_sites() 顯示您的方案可發布至哪些目錄。
4. 評論監控
掃描近期評論,並標記需要回覆的負面評論。
import listingsapi client = listingsapi.ListingsAPI() page = client.locations.list(first=5) for loc in page: print(f"\n--- {loc.name} ---") reviews = client.reviews.list(loc.id, first=10) if not reviews: print(" No recent reviews") continue for review in reviews: rating = getattr(review, "rating", "N/A") author = getattr(review, "authorName", "Anonymous") site = getattr(review, "siteName", "Unknown") responded = bool(getattr(review, "responses", None)) flag = " ** NEEDS ATTENTION **" if isinstance(rating, (int, float)) and rating <= 2 else "" status = "Responded" if responded else "No reply" print(f" [{rating}] {author} on {site} ({status}){flag}")篩選選項請見 評論,包括 rating_filters、site_urls 和日期範圍。
5. 自動回覆評論
workflows.auto_reply_to_reviews() 可在一次呼叫中擷取近期評論、依評分與回覆狀態篩選,並使用您的範本發布回覆。請先使用 dry_run=True 預覽。{rating} 預留位置會替換為每則評論的星級評分。
import listingsapi client = listingsapi.ListingsAPI() LOCATION_ID = 16808 # Replace with your location IDTEMPLATE = "Thanks for the {rating}-star review! We appreciate you taking the time." # Preview what would be sentpreview = client.workflows.auto_reply_to_reviews( LOCATION_ID, template=TEMPLATE, min_rating=4, dry_run=True,)for entry in preview: print(f" Would reply to {entry['id']} ({entry['rating']} stars)") # Post the replies for realresults = client.workflows.auto_reply_to_reviews( LOCATION_ID, template=TEMPLATE, min_rating=4,)sent = [r for r in results if r["status"] == "sent"]print(f"Replied to {len(sent)} of {len(results)} matching reviews")預設的 min_rating=4 會刻意略過負面評論:這些評論應個別撰寫回覆。請使用 reviews.respond() 逐一處理:
import listingsapifrom listingsapi import APIError client = listingsapi.ListingsAPI() negative = client.reviews.list(16808, rating_filters=[1, 2], first=20)for review in negative: if getattr(review, "responses", None): continue try: client.reviews.respond( review.interactionId, "We are sorry to hear about your experience. Please reach out " "directly so we can make it right.", ) print(f"Replied to {getattr(review, 'authorName', 'Anonymous')}") except APIError as e: print(f"Failed: {e}")6. 分析報告
擷取每個地點的 Google 商家檔案分析資料與評論統計資料。
import listingsapi client = listingsapi.ListingsAPI() FROM_DATE = "2026-01-01"TO_DATE = "2026-06-30" all_locations = list(client.locations.list(first=100).auto_paging_iter())print(f"Generating report for {len(all_locations)} locations ({FROM_DATE} to {TO_DATE})\n") for loc in all_locations[:10]: print(f"--- {loc.name} ---") google = client.analytics.google(loc.id, from_date=FROM_DATE, to_date=TO_DATE) if google: print(f" Google: {google.to_dict()}") review_stats = client.reviews.analytics.overview(loc.id, start_date=FROM_DATE, end_date=TO_DATE) if review_stats: print(f" Reviews: {review_stats.to_dict()}") sites = client.reviews.analytics.sites_stats(loc.id, start_date=FROM_DATE, end_date=TO_DATE) if sites: print(f" Sites: {sites.to_dict()}") print()bing() 和 facebook() 的商家檔案指標請見 分析。
7. 每週聲譽報告
workflows.weekly_reputation_report() 將評論、評論分析、Google 和 Bing 商家檔案分析,以及商家資訊同步狀態整合為單一報告物件。
import listingsapi client = listingsapi.ListingsAPI() report = client.workflows.weekly_reputation_report( 16808, start_date="2026-06-29", end_date="2026-07-05",) summary = report.review_summaryprint(f"Average rating: {summary.get('averageRating')}")print(f"Recent reviews: {len(report.recent_reviews)}") analytics = report.analyticsprint(f"Google: {analytics.get('google')}")print(f"Bing: {analytics.get('bing')}") health = report.listings_healthprint(f"Listings synced: {health.get('synced')}/{health.get('total')} ({health.get('sync_rate')})")8. 商家資訊健康狀態稽核
workflows.listings_health_audit() 檢查某個地點的進階商家資訊、語音商家資訊與重複商家資訊,並計算 0 至 100 的健康分數。
import listingsapi client = listingsapi.ListingsAPI() page = client.locations.list(first=10)print(f"Auditing listings for {len(page)} locations\n") for loc in page: audit = client.workflows.listings_health_audit(loc.id) print(f"--- {loc.name} ---") print(f" Health score: {audit.health_score}%") print(f" Synced: {audit.synced_count}, issues: {audit.issue_count}") print(f" Voice listings: {len(audit.voice)}, duplicates: {len(audit.duplicates)}") for issue in audit.issues: print(f" [{issue.get('syncStatus')}] {issue.get('site')}") print()底層的 premium、voice、duplicates 及標記為重複項目的方法,請見 商家資訊。
9. 批次發布貼文
posts.bulk_publish() 可將一則貼文發布至多個地點,預設會同時發布至 Google 和 Facebook。地點 ID 會自動編碼,且酬載會在發出任何網路呼叫前於用戶端完成驗證。
import listingsapi client = listingsapi.ListingsAPI() result = client.posts.bulk_publish( name="Summer hours 2026", location_ids=[16808, 16809, 16810], message="We are open late all summer! Come see us until 9pm, Monday through Saturday.", media_url="https://cdn.example.com/summer-hours.jpg", cta_type="LEARN_MORE", cta_url="https://www.acmedental.com/hours",) post = result.get("socialPost") or {}print(f"Created bulk post: {post.get('id')}")將 dict 傳入 message,即可為各網站自訂文案:
result = client.posts.bulk_publish( name="July whitening offer", location_ids=[16808], message={ "GOOGLE": "20% off teeth whitening this July. Book online today!", "FACEBOOK": "July special: 20% off teeth whitening. Tap to book your visit!", }, sites=["GOOGLE", "FACEBOOK"],)若要為單一網站發布具型別的貼文,請使用 posts.create_announcement()、posts.create_event() 或 posts.create_offer()。
10. Google 連線流程
產生 OAuth URL、列出已連線的帳戶,並檢視配對建議。
import time import listingsapi client = listingsapi.ListingsAPI() # Step 1: Generate an OAuth URL (valid 24 hours)result = client.connected_accounts.connect_google( success_url="https://yourapp.com/connect/success", error_url="https://yourapp.com/connect/error",)print(f"Redirect user to: {result.get('url', 'N/A')}") # Step 2: List connected Google accounts. The payload is an offset page:# records[] plus pageInfo, not a bare list.accounts = client.connected_accounts.list(publisher="GoogleAccount", page=1, per_page=20)for acc in accounts.get("records") or []: account_id = acc["connectedAccountId"] print(f"{acc['email']} -- status: {acc['status']}") # Step 3: Kick off the fetch-and-match job for this account. client.connected_accounts.trigger_matches([account_id]) # Step 4: Matching is asynchronous. Poll the account's details until it # reports MATCH_COMPLETED, with a bounded number of attempts so a stuck # job cannot hang the script. status = None for _ in range(20): details = client.connected_accounts.details(account_id) status = (details.get("details") or {}).get("requestMatchesStatus") if status == "MATCH_COMPLETED": break time.sleep(15) if status != "MATCH_COMPLETED": print(f" matching still {status!r}, skipping for now") continue # Step 5: Only now are suggestions populated. Same offset-page shape. suggestions = client.connected_accounts.suggestions(account_id, page=1, per_page=10) records = suggestions.get("records") or [] print(f" Suggestions: {len(records)} matches found")在 requestMatchesStatus 達到 MATCH_COMPLETED 之前讀取建議,會傳回空的 records 陣列,這與「沒有任何配對」的結果無法區分,因此輪詢才能確保此範例正確運作,而非只是碰巧成功。狀態值請見 觸發地點配對。
使用 connected_accounts.confirm_matches() 確認配對,並傳入由每筆記錄的 matchedDataDatabaseId 與 accountType 組成的配對記錄 ID(請見 確認連線配對),再使用 listings.connect() 將特定商家資訊連結至地點。