使用场景

export LISTINGSAPI_KEY="your-api-key"

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}")

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")

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}")

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}")

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")

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}")

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()

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')})")

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()

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')}")

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"],)

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")