Find reputation-management prospects with unanswered Google reviews

python3 -m venv .venv
source .venv/bin/activate
pip install "tapline==0.2.19"
export TAPLINE_API_KEY="your-api-key"

import os

from tapline import SyncTaplineClient

with SyncTaplineClient(api_key=os.environ["TAPLINE_API_KEY"]) as tapline:
    page = tapline.google_maps.search(query="dentists in Austin, TX", limit=3)
    for place in page.results:
        print(place.name, place.feature_id, place.rating, place.review_count)

ATX Family Dental 0x8644b4eb0b01986f:0x6b241d4769ed491e 4.9 915
South Austin Dentist 0x865b4b5e859f967f:0x38deff12e0bdefcb 5.0 181
Austin Lifetime Dental 0x8644cc20724ee6f5:0x2ae9fa2c2bf15149 4.9 846

"""Build a dated manual account-review queue from Google Maps review evidence."""

from __future__ import annotations

import argparse
import csv
import os
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from pathlib import Path

from tapline import SyncTaplineClient

SEARCH_LIMIT = 10
MAX_REVIEWED_BUSINESSES = 5
REVIEWS_PER_BUSINESS = 20
MAX_REVIEW_RATING = 3
RECENT_DAYS = 180
SEARCH_CREDITS = 2
REVIEWS_CREDITS = 2
OUTPUT_PREFIX = "google-maps-account-review-queue"


@dataclass(frozen=True)
class AccountReviewEvidence:
    business_name: str
    category: str
    address: str
    phone: str
    website: str
    google_maps_url: str
    place_rating: float | None
    place_review_count: int | None
    review_id: str
    review_rating: int
    review_created_at: str
    review_text: str
    review_url: str
    owner_response: str
    example_filter: str
    captured_at_utc: str


def parse_created_at(value: str) -> datetime:
    return datetime.fromisoformat(value.replace("Z", "+00:00")).astimezone(timezone.utc)


def public_phone(place: object) -> str:
    phone = getattr(place, "phone", None)
    if not phone:
        return ""
    return phone.international or phone.number or ""


def public_website(place: object) -> str:
    website = getattr(place, "website", None)
    return website.url if website and website.url else ""


def maps_url(place: object) -> str:
    if getattr(place, "url", None):
        return place.url
    if getattr(place, "cid", None):
        return f"https://www.google.com/maps?cid={place.cid}"
    return ""


def write_csv(rows: list[AccountReviewEvidence], output_dir: Path, run_at: datetime) -> Path:
    output_dir.mkdir(parents=True, exist_ok=True)
    timestamp = run_at.strftime("%Y%m%dT%H%M%SZ")
    output_path = output_dir / f"{OUTPUT_PREFIX}-{timestamp}.csv"
    with output_path.open("w", newline="", encoding="utf-8") as output:
        writer = csv.DictWriter(output, fieldnames=list(AccountReviewEvidence.__dataclass_fields__))
        writer.writeheader()
        writer.writerows(asdict(row) for row in rows)
    return output_path


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("category", help='Business category, such as "dentists"')
    parser.add_argument("location", help='City and region, such as "Austin, TX"')
    parser.add_argument(
        "--output-dir",
        type=Path,
        default=Path("."),
        help="Directory for the timestamped CSV (default: current directory)",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    run_at = datetime.now(timezone.utc)
    cutoff = run_at - timedelta(days=RECENT_DAYS)
    query = f"{args.category} in {args.location}"
    rows: list[AccountReviewEvidence] = []
    reviewed_businesses = 0

    with SyncTaplineClient(api_key=os.environ["TAPLINE_API_KEY"]) as tapline:
        places = tapline.google_maps.search(query=query, gl="us", limit=SEARCH_LIMIT).results

        for place in places:
            if reviewed_businesses >= MAX_REVIEWED_BUSINESSES:
                break

            phone = public_phone(place)
            website = public_website(place)
            source_url = maps_url(place)
            if (
                not place.name
                or place.is_operating_business is False
                or not place.feature_id
                or not source_url
                or not (phone or website)
            ):
                continue

            reviewed_businesses += 1
            reviews = tapline.google_maps.list_reviews(
                place.feature_id,
                sort="newest",
                limit=REVIEWS_PER_BUSINESS,
                gl="us",
            ).reviews
            matched_for_business = 0

            for review in reviews:
                if (
                    review.rating is None
                    or review.rating > MAX_REVIEW_RATING
                    or not review.created_at
                    or not review.text
                    or not review.url
                    or review.owner_response is not None
                ):
                    continue
                if parse_created_at(review.created_at) < cutoff:
                    continue

                matched_for_business += 1
                rows.append(
                    AccountReviewEvidence(
                        business_name=place.name,
                        category=place.categories[0] if place.categories else args.category,
                        address=place.address or "",
                        phone=phone,
                        website=website,
                        google_maps_url=source_url,
                        place_rating=place.rating,
                        place_review_count=place.review_count,
                        review_id=review.review_id or "",
                        review_rating=review.rating,
                        review_created_at=review.created_at,
                        review_text=review.text,
                        review_url=review.url,
                        owner_response="absent in captured response",
                        example_filter=(
                            f"rating <= {MAX_REVIEW_RATING}; created within {RECENT_DAYS} days; "
                            "owner_response absent"
                        ),
                        captured_at_utc=run_at.isoformat(),
                    )
                )

            print(
                f"REVIEWED | {place.name} | {len(reviews)} newest reviews | "
                f"{matched_for_business} matched the example filter"
            )

    output_path = write_csv(rows, args.output_dir, run_at)
    flagged_accounts = len({row.business_name for row in rows})
    documented_credits = SEARCH_CREDITS + reviewed_businesses * REVIEWS_CREDITS
    print(f"Google Maps returned {len(places)} businesses for {query!r}")
    print(
        f"Reviewed {reviewed_businesses} operating businesses with public contact and source fields"
    )
    print(
        f"Flagged {flagged_accounts} accounts with {len(rows)} exact review evidence rows"
    )
    print(
        f"API calls: 1 search + {reviewed_businesses} review pages; "
        f"documented cost: {documented_credits} credits"
    )
    print(f"Wrote the dated manual review queue to {output_path}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

python find_unanswered_reviews.py dentists "Austin, TX" --output-dir final-run

REVIEWED | ATX Family Dental | 20 newest reviews | 0 matched the example filter
REVIEWED | Austin Lifetime Dental | 20 newest reviews | 0 matched the example filter
REVIEWED | South Austin Dentist | 20 newest reviews | 0 matched the example filter
REVIEWED | Swish Dental South | 20 newest reviews | 1 matched the example filter
REVIEWED | 38th Street Dental | 20 newest reviews | 1 matched the example filter
Google Maps returned 10 businesses for 'dentists in Austin, TX'
Reviewed 5 operating businesses with public contact and source fields
Flagged 2 accounts with 2 exact review evidence rows
API calls: 1 search + 5 review pages; documented cost: 12 credits
Wrote the dated manual review queue to final-run/google-maps-account-review-queue-20261103T225511Z.csv