BBB - Business Search by Category + Location, Ratings, Accreditation and Complaint Counts

import json, os, re, tempfile, time
 
def bbb_state():
    # The page deletes window.__PRELOADED_STATE__ after hydrating, so parse the script text instead.
    t = js("(() => { const s = [...document.scripts].find(s => /__PRELOADED_STATE__\\s*=/.test(s.textContent)); return s ? s.textContent : null })()")
    if not t:
        return None
    # Seen both as `__PRELOADED_STATE__ = {...};` and minified `__PRELOADED_STATE__={...}`.
    return json.loads(t.split("=", 1)[1].strip().rstrip(";"))
 
def bbb_open(url, settle=2.0):
    """Open url. None for BBB's 404 page; raises on a challenge or if the tab is not on url."""
    goto_url(url)
    wait_for_load()
    time.sleep(settle)
    href, title = js("location.href"), js("document.title") or ""
    challenge = js("!!document.querySelector('#challenge-form, iframe[src*=\"challenges.cloudflare.com\"]')")
    if "Just a moment" in title or challenge:
        raise RuntimeError(f"Cloudflare challenge on {url}: stop and tell the user; do not retry around it")
    if "Page not found" in title:
        return None
    if href.rstrip("/") != url.rstrip("/"):
        raise RuntimeError(f"asked for {url} but the tab is on {href}; not reading it")
    return href
 
SLUG = re.compile(r"^[a-z0-9-]+$")
PAGE_SIZE = 15
 
def bbb_category(country, state, city, category, accredited=False):
    """Page 1 of a query-free listing. country 'us' or 'ca'; city=None for the state-level page.
    Returns None for unknown slugs."""
    parts = [country, state] + ([city] if city else []) + ["category", category]
    if not all(SLUG.match(p) for p in parts):
        raise ValueError("slugs must be lowercase-hyphenated, e.g. 'san-francisco', 'plumber'")
    url = "https://www.bbb.org/" + "/".join(parts) + ("/accredited" if accredited else "")
    if not bbb_open(url):
        return None
    sr = (bbb_state() or {}).get("searchResult")
    if not isinstance(sr, dict):
        raise RuntimeError(f"{url}: no searchResult in the page state; refusing to call it an empty listing")
    loc = (sr.get("location") or {}).get("displayText")
    if city and not loc:
        return None  # unknown city slug: BBB silently serves the state-level list instead
    results, total = sr.get("results"), sr.get("totalResults")
    if not isinstance(results, list) or not isinstance(total, int) or len(results) != min(total, PAGE_SIZE):
        got = len(results) if isinstance(results, list) else repr(results)
        raise RuntimeError(f"{url}: {got} rows for totalResults={total!r}, expected min(total, {PAGE_SIZE})")
    rows = []
    for r in results:
        if not (r.get("id") and r.get("reportUrl") and (r.get("businessName") or "").strip()):
            raise RuntimeError(f"{url}: a listing row lacks id, reportUrl or businessName")
        rows.append({
            "id": r["id"],
            "name": r["businessName"].strip(),
            "rating": r.get("rating"),
            "accredited_flag": r.get("bbbMember"),
            "street": r.get("address") or None,
            "city": (r.get("city") or "").strip(),
            "state": r.get("state"),
            "postal_code": r.get("postalcode"),
            "phone": (r.get("phone") or [None])[0],
            "primary_category": r.get("tobText"),
            "categories": [c["name"] for c in r.get("categories") or []],
            "profile_url": "https://www.bbb.org" + r["reportUrl"],
        })
    return {
        "url": url,
        "location": loc,
        "total_results": total,
        "truncated": total > len(rows),  # rows 16+ are on ?page=2+, which robots.txt disallows
        "related_categories": [c["url"].rsplit("/", 1)[1] for c in sr.get("relatedCategories") or []],
        "rows": rows,
    }
 
new_tab("about:blank")
cities = ["san-francisco", "oakland", "san-jose", "berkeley", "daly-city", "not-a-real-city"]
found, listings, last = {}, [], None
for city in cities:
    p = bbb_category("us", "ca", city, "plumber")
    time.sleep(3)
    if p is None:
        print(f"{city:15} -> no listing")
        continue
    last = p
    listings.append({k: p[k] for k in ("url", "total_results", "truncated")} | {"rows_read": len(p["rows"])})
    for row in p["rows"]:
        found.setdefault(row["profile_url"], row)  # one profile can appear twice in a listing
    print(f'{city:15} total={p["total_results"]:<5} rows={len(p["rows"])} truncated={p["truncated"]} unique_so_far={len(found)}')
partial = any(l["truncated"] for l in listings)
print(len(listings), "listings ->", len(found), "unique businesses;",
      "PARTIAL: page 1 only of each listing (robots.txt disallows ?page=2+)" if partial else "complete")
print(json.dumps(next(iter(found.values()), None), indent=1))
print("related:", last["related_categories"] if last else None)
out = os.path.join(tempfile.gettempdir(), "bbb_plumbers.json")
with open(out, "w") as f:
    json.dump({"complete": not partial, "listings": listings, "rows": list(found.values())}, f, indent=1)
print("saved", out)

import json, os, re, tempfile, time
 
def bbb_state():
    t = js("(() => { const s = [...document.scripts].find(s => /__PRELOADED_STATE__\\s*=/.test(s.textContent)); return s ? s.textContent : null })()")
    if not t:
        return None
    # Seen both as `__PRELOADED_STATE__ = {...};` and minified `__PRELOADED_STATE__={...}`.
    return json.loads(t.split("=", 1)[1].strip().rstrip(";"))
 
def bbb_open(url, settle=2.0, same=None):
    """Open url. None for BBB's 404 page; raises on a challenge or if the tab is not on url
    (a different landing URL is read only when same(href) says it is the same page)."""
    goto_url(url)
    wait_for_load()
    time.sleep(settle)
    href, title = js("location.href"), js("document.title") or ""
    challenge = js("!!document.querySelector('#challenge-form, iframe[src*=\"challenges.cloudflare.com\"]')")
    if "Just a moment" in title or challenge:
        raise RuntimeError(f"Cloudflare challenge on {url}: stop and tell the user; do not retry around it")
    if "Page not found" in title:
        return None
    if href.rstrip("/") != url.rstrip("/") and not (same and same(href)):
        raise RuntimeError(f"asked for {url} but the tab is on {href}; not reading it")
    return href
 
# Profile URLs end in <bbbId>-<businessId>; BBB redirects stale name/city/category slugs to the canonical URL.
PROFILE = re.compile(r"^https://www\.bbb\.org/(?:us|ca)/[a-z]{2}/[a-z0-9-]+/profile/[a-z0-9-]+/[a-z0-9-]*-(\d{4}-\d+)/?$")
 
def profile_id(url):
    m = PROFILE.match(url or "")
    return m.group(1) if m else None
 
def bbb_profile(profile_url):
    pid = profile_id(profile_url)
    if not pid:
        raise ValueError(f"not a BBB profile URL: {profile_url}")
    href = bbb_open(profile_url, same=lambda h: profile_id(h) == pid)  # query-free, same id, or raise
    if not href:
        return None
    bp = (bbb_state() or {}).get("businessProfile")
    if not isinstance(bp, dict):
        raise RuntimeError(f"{profile_url}: no businessProfile in the page state; refusing to call it a missing profile")
    s = bp.get("reviewsComplaintsSummary")
    name = (bp.get("names") or {}).get("primary")
    rating = (bp.get("rating") or {}).get("bbbRating")
    if not (name and rating and isinstance(s, dict) and isinstance(s.get("totalClosedComplaintsPastThreeYears"), int)
            and isinstance(s.get("totalClosedComplaintsPastTwelveMonths"), int)):
        raise RuntimeError(f"{profile_url}: profile state lacks name, rating or complaint counts")
    loc = bp.get("location") or {}
    ld = {}
    for raw in js("[...document.querySelectorAll('script[type=\"application/ld+json\"]')].map(s => s.textContent)") or []:
        try:
            data = json.loads(raw)
        except ValueError:
            continue
        for x in data if isinstance(data, list) else [data]:
            if x.get("@type") == "LocalBusiness":
                ld = {p["name"]: p["value"] for p in x.get("additionalProperty") or []}
    return {
        "requested_url": profile_url,
        "profile_url": href,
        "name": name,
        "rating": rating,
        "accredited": (bp.get("accreditationInformation") or {}).get("isAccredited"),
        "complaints_3y": s["totalClosedComplaintsPastThreeYears"],
        "complaints_12m": s["totalClosedComplaintsPastTwelveMonths"],
        "reviews": s.get("reviewsTotal"),
        "avg_stars": s.get("averageOfReviewStarRatings"),
        "website": (bp.get("urls") or {}).get("primary"),
        "phone": (bp.get("contactInformation") or {}).get("phoneNumber"),
        "address": (loc.get("formattedAddress") or "").lstrip(", "),
        "jsonld_complaints_3y": ld.get("BBB Complaints Filed in Last 3 Years"),
        "jsonld_accreditation": ld.get("BBB Accreditation"),
    }
 
LIMIT = 5
src = os.path.join(tempfile.gettempdir(), "bbb_plumbers.json")
rows = json.load(open(src))["rows"] if os.path.exists(src) else []
for row in rows[:LIMIT]:
    prof = bbb_profile(row["profile_url"])
    time.sleep(3)
    if prof is None:
        print("no profile (404):", row["profile_url"])
        continue
    print(f'{row["name"][:34]:34} {prof["rating"]:3} acc={prof["accredited"]!s:5} '
          f'complaints_3y={prof["complaints_3y"]} 12m={prof["complaints_12m"]} reviews={prof["reviews"]} site={prof["website"]}')
    if prof["profile_url"].rstrip("/") != row["profile_url"].rstrip("/"):
        print("  redirected to", prof["profile_url"])
if len(rows) > LIMIT:
    print(f"PARTIAL: enriched {LIMIT} of {len(rows)} rows (LIMIT={LIMIT})")
 
print(json.dumps(bbb_profile("https://www.bbb.org/us/ny/new-york/profile/telephones/verizon-communications-0011-1568"), indent=1))
time.sleep(3)
moved = bbb_profile("https://www.bbb.org/us/ca/oakland/profile/plumber/atlas-plumbing-rooter-1116-116593")
print("redirected:", moved["requested_url"], "->", moved["profile_url"])

import http.client, re, time, urllib.error, urllib.request
 
UA = "bbb-sitemap-reader/1.0 (python-urllib)"
BUDGET_S, WAIT_S = 110, 10
 
def fetch(url, tries=3):
    deadline = time.monotonic() + BUDGET_S
    delay, err, attempt = 5, None, 0
    while True:
        attempt += 1
        try:
            req = urllib.request.Request(url, headers={"User-Agent": UA})
            with urllib.request.urlopen(req, timeout=min(WAIT_S, deadline - time.monotonic())) as r:
                body = bytearray()
                while chunk := r.read1(65536):
                    body += chunk
                    if time.monotonic() > deadline:
                        raise TimeoutError("deadline hit mid-download")
                return body.decode("utf-8-sig")
        except urllib.error.HTTPError as e:
            if e.code in (403, 429):
                raise RuntimeError(f"{url}: HTTP {e.code}, blocked or rate limited; stopping, not retrying") from e
            if e.code < 500:
                raise RuntimeError(f"{url}: HTTP {e.code}; not retrying") from e
            err = e  # 5xx: retry
        except (OSError, http.client.HTTPException) as e:
            err = e  # network error or timeout: retry; decode and other errors propagate at once
        if attempt >= tries or deadline - time.monotonic() <= delay:
            break  # last attempt, or no budget left after a backoff: raise now, no sleep
        time.sleep(delay)  # 5 s, 10 s, ...
        delay *= 2
    raise RuntimeError(f"{url}: failed after {attempt} attempt(s) within the {BUDGET_S} s budget: {err}")
 
def state_category_slugs(state):
    idx = fetch("https://www.bbb.org/sitemap-us-state-categories-index.xml")
    files = re.findall(r"<loc>([^<]+)</loc>", idx)
    if not files:
        raise RuntimeError("sitemap index has no <loc> entries")
    slugs, seen_state = [], False
    for sm in files:
        locs = re.findall(r"<loc>https://www\.bbb\.org/us/([a-z]{2})/category/([^<]+)</loc>", fetch(sm))
        if not locs:
            raise RuntimeError(f"{sm}: no state category URLs")
        mine = [c for s, c in locs if s == state]
        slugs += mine
        if mine:
            seen_state = True
        elif seen_state:
            break  # files are sorted by state, so we are past it
        time.sleep(1)
    if not slugs:
        raise RuntimeError(f"no category slugs found for state {state!r}")
    return slugs
 
slugs = state_category_slugs("ca")
print(len(slugs), "category slugs for CA")
print([s for s in slugs if "plumb" in s])