Google Meu Negócio: sync + coleta de avaliações

1#!/usr/bin/env python3
2"""Cap. 20 - Standalone offline demo of the GBP review pipeline (ZERO credentials).
3
4Runs the whole review-collection pipeline with FICTIONAL data:
5
6  1. Creates an in-memory SQLite DB via ``reviews_collector``.
7  2. Seeds ~6 fictional reviews (no API, no network, no credentials).
8  3. Prints a per-review star breakdown and the computed ``aggregateRating``.
9  4. Prints the schema.org AggregateRating / Review JSON-LD.
10
11Run it standalone::
12
13    python3 demo.py
14
15Nothing here touches Google Business Profile - the reviews are made up so you
16can see the aggregate rating and JSON-LD working before wiring real OAuth
17credentials into ``gmn_sync.py`` / ``reviews_collector.fetch_reviews``.
18
19Part of the "living code" companion repository for the book
20"Aparecer ou Sumir" / "Be Seen or Be Forgotten".
21"""
22
23from __future__ import annotations
24
25import json
26
27import reviews_collector as rc
28
29BUSINESS_NAME = "Padaria do Bairro"
30
31
32def main() -> None:
33    conn = rc.connect(":memory:")
34    rc.init_db(conn)
35
36    written = rc.collect(conn, rc.SAMPLE_REVIEWS)
37
38    print("=" * 60)
39    print("  COLETA DE AVALIAÇÕES — dados fictícios (sem credenciais)")
40    print("  Review collection — fictional data (no credentials)")
41    print("=" * 60)
42    print(f"\n[demo] {written} avaliações fictícias armazenadas / fictional reviews stored")
43
44    # Per-review breakdown (stars + author + short text).
45    print("\n  Avaliações (reviews)")
46    print("  " + "-" * 52)
47    for r in rc.all_reviews(conn):
48        stars = "★" * int(r["star_rating"]) + "☆" * (5 - int(r["star_rating"]))
49        author = r["author"] or "Anônimo"
50        text = r["text"] or "(sem comentário / no comment)"
51        print(f"  {stars}  {author:<14} {text}")
52
53    # Aggregate rating.
54    agg = rc.aggregate_rating(conn)
55    print("\n  Nota agregada (aggregate rating)")
56    print("  " + "-" * 52)
57    print(
58        f"  ratingValue = {agg['ratingValue']}  "
59        f"reviewCount = {agg['reviewCount']}  "
60        f"(best {agg['bestRating']} / worst {agg['worstRating']})"
61    )
62
63    # JSON-LD.
64    jsonld = rc.build_jsonld(conn, business_name=BUSINESS_NAME)
65    print("\n  AggregateRating / Review JSON-LD")
66    print("  " + "-" * 52)
67    print(json.dumps(jsonld, ensure_ascii=False, indent=2))
68    print("=" * 60)
69
70    conn.close()
71
72
73if __name__ == "__main__":
74    main()

1#!/usr/bin/env python3
2"""Cap. 20 - Automatic review collection for Google Business Profile.
3
4Fetches reviews for a business location, normalizes them into a common shape,
5stores them in SQLite (stdlib ``sqlite3``), computes an ``aggregateRating`` and
6emits schema.org ``Review`` + ``AggregateRating`` JSON-LD you can drop into a
7page to show star ratings in search results.
8
9Two ways to run:
10
11  * ``--demo``  : loads FICTIONAL sample reviews (inline), stores them in an
12                  in-memory DB, prints the aggregate rating and the JSON-LD.
13                  No credentials, no network.
14  * (live)      : ``fetch_reviews()`` hits the GBP API. That path is a
15                  credential-gated stub — see the REQUER CREDENCIAL block.
16
17Usage::
18
19    python3 reviews_collector.py --demo
20    python3 reviews_collector.py --demo --db reviews.db --pretty
21
22============================================================================
23 REQUER CREDENCIAL / REQUIRES CREDENTIALS
24 ---------------------------------------------------------------------------
25 Only ``fetch_reviews()`` (the live GBP call) needs OAuth credentials, read
26 from ``os.environ`` via ``gmn_sync``. Everything else - normalization,
27 storage, aggregate rating and JSON-LD - runs with zero credentials, which is
28 exactly what ``--demo`` exercises.
29============================================================================
30
31Part of the "living code" companion repository for the book
32"Aparecer ou Sumir" / "Be Seen or Be Forgotten".
33"""
34
35from __future__ import annotations
36
37import argparse
38import json
39import sqlite3
40from typing import Dict, List, Optional
41
42# ---------------------------------------------------------------------------
43# Fictional sample reviews used by --demo (and by demo.py). Zero credentials.
44# star_rating is 1..5; text is optional. Dates are ISO-8601.
45# ---------------------------------------------------------------------------
46SAMPLE_REVIEWS: List[Dict[str, object]] = [
47    {
48        "review_id": "demo-1",
49        "author": "Marina S.",
50        "star_rating": 5,
51        "text": "Atendimento impecável e entrega no prazo. Recomendo demais!",
52        "create_time": "2026-06-28T14:03:00Z",
53    },
54    {
55        "review_id": "demo-2",
56        "author": "João P.",
57        "star_rating": 4,
58        "text": "Ótima experiência, só achei o estacionamento apertado.",
59        "create_time": "2026-06-25T09:41:00Z",
60    },
61    {
62        "review_id": "demo-3",
63        "author": "Ana Beatriz",
64        "star_rating": 5,
65        "text": "Equipe muito atenciosa, resolveram tudo rapidinho.",
66        "create_time": "2026-06-20T18:22:00Z",
67    },
68    {
69        "review_id": "demo-4",
70        "author": "Carlos M.",
71        "star_rating": 3,
72        "text": "Bom, mas a fila estava grande no horário de pico.",
73        "create_time": "2026-06-15T12:10:00Z",
74    },
75    {
76        "review_id": "demo-5",
77        "author": "Fernanda L.",
78        "star_rating": 5,
79        "text": "Melhor da região, sem dúvida. Voltarei sempre.",
80        "create_time": "2026-06-10T20:05:00Z",
81    },
82    {
83        "review_id": "demo-6",
84        "author": "Roberto A.",
85        "star_rating": 4,
86        "text": None,  # a review with a star rating but no written comment
87        "create_time": "2026-06-05T08:30:00Z",
88    },
89]
90
91# Map the GBP API star-rating enum to an integer 1..5.
92_STAR_ENUM = {"ONE": 1, "TWO": 2, "THREE": 3, "FOUR": 4, "FIVE": 5}
93
94
95# ---------------------------------------------------------------------------
96# Storage (SQLite, stdlib) — fully functional, no credentials.
97# ---------------------------------------------------------------------------
98def connect(db_path: str = ":memory:") -> sqlite3.Connection:
99    """Open a SQLite connection with row access by column name."""
100    conn = sqlite3.connect(db_path)
101    conn.row_factory = sqlite3.Row
102    return conn
103
104
105def init_db(conn: sqlite3.Connection) -> None:
106    """Create the reviews table if missing.
107
108    UNIQUE(review_id) makes ingestion idempotent: re-collecting the same review
109    updates it in place instead of duplicating rows.
110    """
111    conn.execute(
112        """
113        CREATE TABLE IF NOT EXISTS reviews (
114            review_id    TEXT PRIMARY KEY,
115            author       TEXT,
116            star_rating  INTEGER NOT NULL,
117            text         TEXT,
118            create_time  TEXT
119        )
120        """
121    )
122    conn.commit()
123
124
125def store_reviews(conn: sqlite3.Connection, reviews: List[Dict[str, object]]) -> int:
126    """Upsert normalized reviews keyed by review_id. Returns rows processed."""
127    prepared = [
128        (
129            r["review_id"],
130            r.get("author"),
131            int(r["star_rating"]),
132            r.get("text"),
133            r.get("create_time"),
134        )
135        for r in reviews
136    ]
137    conn.executemany(
138        """
139        INSERT INTO reviews (review_id, author, star_rating, text, create_time)
140        VALUES (?, ?, ?, ?, ?)
141        ON CONFLICT(review_id) DO UPDATE SET
142            author = excluded.author,
143            star_rating = excluded.star_rating,
144            text = excluded.text,
145            create_time = excluded.create_time
146        """,
147        prepared,
148    )
149    conn.commit()
150    return len(prepared)
151
152
153def all_reviews(conn: sqlite3.Connection) -> List[sqlite3.Row]:
154    """Return every stored review, newest first."""
155    cur = conn.execute(
156        "SELECT review_id, author, star_rating, text, create_time "
157        "FROM reviews ORDER BY create_time DESC"
158    )
159    return cur.fetchall()
160
161
162# ---------------------------------------------------------------------------
163# Normalization + aggregate rating + JSON-LD — no credentials.
164# ---------------------------------------------------------------------------
165def normalize_review(raw: Dict[str, object]) -> Dict[str, object]:
166    """Normalize a raw GBP review payload into our common shape.
167
168    Accepts both the GBP API shape (``starRating`` enum, ``reviewId``,
169    ``comment``, ``reviewer.displayName``) and our already-normalized shape,
170    so the demo rows pass through untouched.
171    """
172    if "star_rating" in raw:  # already normalized (e.g. SAMPLE_REVIEWS)
173        return {
174            "review_id": str(raw.get("review_id", "")),
175            "author": raw.get("author"),
176            "star_rating": int(raw["star_rating"]),
177            "text": raw.get("text"),
178            "create_time": raw.get("create_time"),
179        }
180    # GBP API shape.
181    reviewer = raw.get("reviewer") or {}
182    star = raw.get("starRating", "")
183    return {
184        "review_id": str(raw.get("reviewId", "")),
185        "author": reviewer.get("displayName") if isinstance(reviewer, dict) else None,
186        "star_rating": _STAR_ENUM.get(str(star), 0),
187        "text": raw.get("comment"),
188        "create_time": raw.get("createTime"),
189    }
190
191
192def aggregate_rating(conn: sqlite3.Connection) -> Dict[str, object]:
193    """Compute the aggregate rating from all stored reviews.
194
195    Returns a dict with ``ratingValue`` (mean, 1 decimal), ``reviewCount`` and
196    the ``bestRating`` / ``worstRating`` bounds.
197    """
198    cur = conn.execute("SELECT COUNT(*) AS n, AVG(star_rating) AS avg FROM reviews")
199    row = cur.fetchone()
200    count = int(row["n"] or 0)
201    mean = round(float(row["avg"]), 1) if count else 0.0
202    return {
203        "ratingValue": mean,
204        "reviewCount": count,
205        "bestRating": 5,
206        "worstRating": 1,
207    }
208
209
210def build_jsonld(
211    conn: sqlite3.Connection,
212    business_name: str = "Seu Negócio",
213) -> Dict[str, object]:
214    """Build a schema.org LocalBusiness graph with AggregateRating + Reviews."""
215    agg = aggregate_rating(conn)
216    reviews_ld = []
217    for r in all_reviews(conn):
218        review_ld: Dict[str, object] = {
219            "@type": "Review",
220            "author": {"@type": "Person", "name": r["author"] or "Anônimo"},
221            "reviewRating": {
222                "@type": "Rating",
223                "ratingValue": r["star_rating"],
224                "bestRating": 5,
225                "worstRating": 1,
226            },
227        }
228        if r["create_time"]:
229            review_ld["datePublished"] = r["create_time"]
230        if r["text"]:
231            review_ld["reviewBody"] = r["text"]
232        reviews_ld.append(review_ld)
233
234    return {
235        "@context": "https://schema.org",
236        "@type": "LocalBusiness",
237        "name": business_name,
238        "aggregateRating": {
239            "@type": "AggregateRating",
240            "ratingValue": agg["ratingValue"],
241            "reviewCount": agg["reviewCount"],
242            "bestRating": agg["bestRating"],
243            "worstRating": agg["worstRating"],
244        },
245        "review": reviews_ld,
246    }
247
248
249# ---------------------------------------------------------------------------
250# Live GBP fetch — REQUER CREDENCIAL / REQUIRES CREDENTIALS.
251# ---------------------------------------------------------------------------
252def fetch_reviews(location_id: str, service: Optional[object] = None) -> List[Dict[str, object]]:
253    """Fetch reviews for ``location_id`` from the GBP API (live call).
254
255    TODO: credential — with an authorized ``service`` from
256    ``gmn_sync.authenticate()``::
257
258        resp = service.accounts().locations().reviews().list(
259            parent=f"{GBP_ACCOUNT_ID}/{location_id}",
260        ).execute()
261        return [normalize_review(r) for r in resp.get("reviews", [])]
262
263    Reviews come paginated; follow ``nextPageToken`` until exhausted. Without
264    credentials this raises ``NotImplementedError`` — use ``--demo`` to run the
265    rest of the pipeline offline.
266    """
267    import gmn_sync  # local import so --demo never requires credentials
268
269    if service is None and not gmn_sync._have_oauth():
270        raise NotImplementedError(
271            "fetch_reviews() needs an authenticated GBP session. "
272            "Set the GBP_OAUTH_* env vars, or run with --demo."
273        )
274    # TODO: credential — replace with the real reviews().list() call.
275    raise NotImplementedError("GBP reviews.list not implemented yet.")
276
277
278# ---------------------------------------------------------------------------
279# Pipeline + CLI
280# ---------------------------------------------------------------------------
281def collect(
282    conn: sqlite3.Connection,
283    reviews: List[Dict[str, object]],
284) -> int:
285    """Normalize + store a batch of reviews. Returns rows written."""
286    normalized = [normalize_review(r) for r in reviews]
287    return store_reviews(conn, normalized)
288
289
290def print_summary(conn: sqlite3.Connection, jsonld: Dict[str, object], pretty: bool) -> None:
291    """Print a small text summary of the aggregate rating + the JSON-LD."""
292    agg = aggregate_rating(conn)
293    stars = "★" * int(round(agg["ratingValue"]))
294    print(
295        f"Aggregate rating: {agg['ratingValue']} {stars} "
296        f"({agg['reviewCount']} reviews)"
297    )
298    print("\nReview / AggregateRating JSON-LD:")
299    print(json.dumps(jsonld, ensure_ascii=False, indent=2 if pretty else None))
300
301
302def main(argv: Optional[List[str]] = None) -> int:
303    parser = argparse.ArgumentParser(description="GBP review collector.")
304    parser.add_argument(
305        "--demo",
306        action="store_true",
307        help="Load fictional sample reviews (no credentials, no network).",
308    )
309    parser.add_argument(
310        "--location-id",
311        default="locations/DEMO",
312        help="GBP location resource name (used only for the live path).",
313    )
314    parser.add_argument(
315        "--db",
316        default=":memory:",
317        help="SQLite path (default: in-memory).",
318    )
319    parser.add_argument(
320        "--business-name",
321        default="Padaria do Bairro",
322        help="Business name to embed in the JSON-LD.",
323    )
324    parser.add_argument(
325        "--pretty",
326        action="store_true",
327        help="Pretty-print the JSON-LD.",
328    )
329    args = parser.parse_args(argv)
330
331    conn = connect(args.db)
332    init_db(conn)
333
334    if args.demo:
335        written = collect(conn, SAMPLE_REVIEWS)
336        print(f"[demo] {written} avaliações fictícias armazenadas / fictional reviews stored\n")
337    else:
338        reviews = fetch_reviews(args.location_id)  # credential-gated
339        written = collect(conn, reviews)
340        print(f"[live] {written} reviews stored from {args.location_id}\n")
341
342    jsonld = build_jsonld(conn, business_name=args.business_name)
343    print_summary(conn, jsonld, pretty=args.pretty)
344    conn.close()
345    return 0
346
347
348if __name__ == "__main__":
349    raise SystemExit(main())

1#!/usr/bin/env python3
2"""Cap. 20 - Google Meu Negócio / Google Business Profile sync - STRUCTURE.
3
4Pushes and keeps your business location(s) in sync with the Google Business
5Profile (GBP, a.k.a. "Google Meu Negócio") API: name, categories, opening
6hours, phone and website. The flow is:
7
8    authenticate()  ->  list_locations()  ->  update_location(id, patch)
9
10============================================================================
11 REQUER CREDENCIAL / REQUIRES CREDENTIALS
12 ---------------------------------------------------------------------------
13 The GBP API only returns/accepts live data through an authenticated OAuth
14 session tied to the Google account that manages the business. Every value
15 below is read ONLY from ``os.environ`` - no secret is ever hardcoded. Until
16 the credentials are set, ``authenticate()`` / ``list_locations()`` /
17 ``update_location()`` raise ``NotImplementedError`` with setup guidance, but
18 the module stays importable and ``py_compile``-clean.
19
20 There is NO offline sample path here on purpose: writing to GBP is a real,
21 side-effecting operation. To see the companion pipeline run offline with
22 fictional data, use ``reviews_collector.py --demo`` or ``demo.py``.
23============================================================================
24
25Part of the "living code" companion repository for the book
26"Aparecer ou Sumir" / "Be Seen or Be Forgotten".
27"""
28
29from __future__ import annotations
30
31import os
32from typing import Dict, List, Optional
33
34# ---------------------------------------------------------------------------
35# TODO: credential — all values come from the environment. Nothing is hardcoded.
36# Set these before any real GBP call can succeed.
37# ---------------------------------------------------------------------------
38# TODO: credential — OAuth client id/secret for a Google Cloud project with the
39# "Business Profile API" (mybusinessbusinessinformation) enabled.
40GBP_OAUTH_CLIENT_ID = os.environ.get("GBP_OAUTH_CLIENT_ID", "")
41GBP_OAUTH_CLIENT_SECRET = os.environ.get("GBP_OAUTH_CLIENT_SECRET", "")
42# TODO: credential — a long-lived OAuth refresh token for the managing account.
43GBP_OAUTH_REFRESH_TOKEN = os.environ.get("GBP_OAUTH_REFRESH_TOKEN", "")
44# TODO: credential — the account resource name, e.g. "accounts/1234567890".
45GBP_ACCOUNT_ID = os.environ.get("GBP_ACCOUNT_ID", "")
46
47# OAuth scope needed to read and write business information.
48GBP_SCOPE = "https://www.googleapis.com/auth/business.manage"
49
50LocationPatch = Dict[str, object]
51
52
53def _have_oauth() -> bool:
54    """True only when every OAuth credential is present in the environment."""
55    return bool(
56        GBP_OAUTH_CLIENT_ID
57        and GBP_OAUTH_CLIENT_SECRET
58        and GBP_OAUTH_REFRESH_TOKEN
59        and GBP_ACCOUNT_ID
60    )
61
62
63def authenticate() -> object:
64    """Build an authorized GBP API session from the OAuth credentials.
65
66    TODO: credential — steps once you have an OAuth client + refresh token:
67      1. pip install google-api-python-client google-auth google-auth-oauthlib
68      2. Exchange the refresh token for an access token::
69
70             from google.oauth2.credentials import Credentials
71             creds = Credentials(
72                 None,
73                 refresh_token=GBP_OAUTH_REFRESH_TOKEN,
74                 client_id=GBP_OAUTH_CLIENT_ID,
75                 client_secret=GBP_OAUTH_CLIENT_SECRET,
76                 token_uri="https://oauth2.googleapis.com/token",
77                 scopes=[GBP_SCOPE],
78             )
79      3. Build the service::
80
81             from googleapiclient.discovery import build
82             service = build("mybusinessbusinessinformation", "v1", credentials=creds)
83
84         and return it for ``list_locations`` / ``update_location`` to use.
85    """
86    if not _have_oauth():
87        print(
88            "[gmn_sync] Missing OAuth credentials. Set GBP_OAUTH_CLIENT_ID, "
89            "GBP_OAUTH_CLIENT_SECRET, GBP_OAUTH_REFRESH_TOKEN and GBP_ACCOUNT_ID "
90            "in your environment. See the authenticate() docstring for the "
91            "full setup steps."
92        )
93        raise NotImplementedError(
94            "authenticate() needs the four GBP_OAUTH_* / GBP_ACCOUNT_ID env vars."
95        )
96    # TODO: credential — replace with the real Credentials + build() call.
97    raise NotImplementedError("GBP OAuth session build not implemented yet.")
98
99
100def list_locations(service: Optional[object] = None) -> List[Dict[str, object]]:
101    """List the business locations managed under ``GBP_ACCOUNT_ID``.
102
103    TODO: credential — with an authorized ``service`` from ``authenticate()``::
104
105        resp = service.accounts().locations().list(
106            parent=GBP_ACCOUNT_ID,
107            readMask="name,title,categories,phoneNumbers,websiteUri,regularHours",
108        ).execute()
109        return resp.get("locations", [])
110
111    Each returned location has a ``name`` like "locations/0987654321" that you
112    pass to :func:`update_location`.
113    """
114    if service is None and not _have_oauth():
115        print(
116            "[gmn_sync] Cannot list locations without an authenticated session. "
117            "Call authenticate() once the GBP_OAUTH_* env vars are set."
118        )
119        raise NotImplementedError(
120            "list_locations() needs an authenticated GBP session."
121        )
122    # TODO: credential — replace with the real accounts().locations().list() call.
123    raise NotImplementedError("GBP locations.list not implemented yet.")
124
125
126def update_location(
127    location_id: str,
128    patch: LocationPatch,
129    service: Optional[object] = None,
130) -> Dict[str, object]:
131    """Push a partial update (name, categories, hours, phone, website) to GBP.
132
133    Args:
134        location_id: resource name of the location, e.g. "locations/0987654321".
135        patch: the fields to change. Only keys present are written, e.g.::
136
137            {
138                "title": "Padaria do Bairro",
139                "phoneNumbers": {"primaryPhone": "+55 51 99999-0000"},
140                "websiteUri": "https://padariadobairro.com.br",
141                "categories": {"primaryCategory": {"name": "categories/bakery"}},
142                "regularHours": {...},
143            }
144
145        service: an authorized session from :func:`authenticate` (optional; if
146            omitted the function will require the OAuth env vars to be present).
147
148    TODO: credential — the real write is a PATCH with an updateMask listing only
149    the fields you changed::
150
151        update_mask = ",".join(patch.keys())
152        return service.locations().patch(
153            name=location_id,
154            updateMask=update_mask,
155            body=patch,
156        ).execute()
157
158    GBP validates edits and some categories/hours changes may be reviewed by
159    Google before going live.
160    """
161    if not location_id:
162        raise ValueError("location_id is required, e.g. 'locations/0987654321'.")
163    if not isinstance(patch, dict) or not patch:
164        raise ValueError("patch must be a non-empty dict of fields to update.")
165    if service is None and not _have_oauth():
166        print(
167            "[gmn_sync] Cannot update location without an authenticated session. "
168            f"Would PATCH {location_id} with fields: {', '.join(sorted(patch))}. "
169            "Set the GBP_OAUTH_* env vars and call authenticate() first."
170        )
171        raise NotImplementedError(
172            "update_location() needs an authenticated GBP session."
173        )
174    # TODO: credential — replace with the real locations().patch() call.
175    raise NotImplementedError("GBP locations.patch not implemented yet.")
176
177
178def main() -> None:
179    """CLI entry point. Runs the credential-gated sync flow end to end.
180
181    Without credentials it prints guidance and exits cleanly (no traceback), so
182    the module is safe to run as a smoke test.
183    """
184    print("Google Meu Negócio / Google Business Profile sync")
185    print("-" * 52)
186    if not _have_oauth():
187        print(
188            "No GBP credentials found in the environment.\n"
189            "Set GBP_OAUTH_CLIENT_ID, GBP_OAUTH_CLIENT_SECRET, "
190            "GBP_OAUTH_REFRESH_TOKEN and GBP_ACCOUNT_ID, then re-run.\n"
191            "To try the companion review pipeline offline instead, run:\n"
192            "    python3 reviews_collector.py --demo\n"
193            "    python3 demo.py"
194        )
195        return
196    # With credentials present, wire the real flow (still stubbed until the API
197    # calls above are implemented).
198    service = authenticate()
199    for location in list_locations(service):
200        print("location:", location.get("name"), "-", location.get("title"))
201
202
203if __name__ == "__main__":
204    main()