حالات الاستخدام

عشرة أمثلة تطبيقية متكاملة بلغة Python، من التصدير الجماعي وأتمتة التقييمات إلى تقارير التحليلات وتدقيق إدراجات الأنشطة التجارية والنشر الجماعي.

اضبط مفتاح API مرة واحدة قبل تشغيل أي مثال:

bash
export LISTINGSAPI_KEY="your-api-key"

1. عرض قائمة مواقعك

اتصل واطبع الصفحة الأولى من مواقعك.

01_quickstart.py
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

نزّل جميع المواقع إلى ملف CSV باستخدام auto_paging_iter().

02_bulk_export_locations.py
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() كل حقل إلزامي كوسيطة مسماة ويتحقق من صحة حمولة الطلب من جانب العميل، لذا يفشل الطلب غير الصالح فورًا مع رسالة واضحة قبل إجراء أي اتصال بالشبكة. ابحث عن قيمة صالحة لـ sub_category_id باستخدام client.subcategories() (استخدم databaseId).

03_create_location.py
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. مراقبة التقييمات

افحص التقييمات الحديثة وعلّم التقييمات السلبية التي تحتاج إلى رد.

04_review_monitoring.py
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} بالتقييم بالنجوم لكل تقييم.

05_auto_reply_reviews.py
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():

05b_respond_to_negative.py
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 وإحصاءات التقييمات لكل موقع.

06_analytics_report.py
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 وحالة مزامنة إدراجات الأنشطة التجارية في كائن تقرير واحد.

07_reputation_report.py
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.

08_listings_audit.py
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 معًا افتراضيًا. تُرمَّز معرّفات المواقع تلقائيًا ويُتحقق من صحة حمولة الطلب من جانب العميل قبل إجراء أي اتصال بالشبكة.

09_bulk_publish_post.py
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 لتخصيص النص لكل منصة:

09b_per_site_message.py
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

أنشئ عنوان URL لـ OAuth، واعرض الحسابات المرتبطة، وراجع اقتراحات المطابقة.

10_google_connect_flow.py
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 في كل سجل (راجع تأكيد مطابقات الربط)، واربط إدراج نشاط تجاري محددًا بموقع باستخدام listings.connect().


الخطوات التالية