ユースケース
一括エクスポート、レビューの自動化、分析レポート、掲載情報の監査、一括投稿など、一連の処理を実装する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()は1回の呼び出しで、最近のレビューを取得し、評価と返信状況で絞り込み、テンプレートに基づく返信を投稿します。まず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のプロフィール分析、掲載情報の同期状況を1つのレポートオブジェクトにまとめます。
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()は、多数の拠点で1つの投稿を公開し、デフォルトでは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')}")サイトごとに本文をカスタマイズするには、messageにdictを渡してください。
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配列が返され、「一致するものが存在しない」場合と区別できないため、この実装例を確実に正しく動作させるにはポーリングが必要です。ステータス値については、拠点の照合を開始を参照してください。
各レコードのmatchedDataDatabaseIdとaccountTypeから構成した照合レコードIDをconnected_accounts.confirm_matches()に渡して照合結果を確定し(接続の照合結果を確定を参照)、listings.connect()で特定の掲載情報を拠点に関連付けてください。