"""Build a qualified local-business lead list from public business data.""" from __future__ import annotations import argparse import csv import os import re import time from collections import Counter from dataclasses import asdict, dataclass from datetime import datetime, timezone from pathlib import Path from statistics import median from urllib.parse import urlparse import requests API_ORIGIN = "https://api.tapline.sh" REQUEST_INTERVAL_SECONDS = 1.05 MAPS_ZOOM = 13 MAPS_PAGE_SIZE = 20 REVIEWS_PER_BUSINESS = 20 GRID_LATITUDE_STEP = 0.035 GRID_LONGITUDE_STEP = 0.04 LOW_REVIEW_RATIO = 0.5 LOW_RATING = 4.0 GENERIC_NAME_WORDS = { "and", "center", "clinic", "dental", "dentist", "dentistry", "of", "practice", "the", } COMPLAINT_PATTERNS = { "phones": re.compile(r"\b(phone|phones|voicemail)\b", re.I), "callbacks": re.compile(r"\b(call(?:ed|ing)? back|callback|callbacks|return(?:ed)? my call)\b", re.I), "wait times": re.compile(r"\b(wait|waited|waiting|late|delay|delayed)\b", re.I), "booking": re.compile(r"\b(book|booking|appointment|schedule|scheduling|reschedule)\b", re.I), } ROLE_PATTERN = re.compile(r"\b(owner|founder|practice manager|office manager|operations manager)\b", re.I) CREDIT_COSTS = { "maps_search": 2, "maps_reviews": 2, "brave_search": 5, "linkedin_company": 1, "linkedin_profile": 2, } class TaplineApiError(RuntimeError): def __init__(self, status_code: int, path: str, message: str): super().__init__(f"{path} returned HTTP {status_code}: {message}") self.status_code = status_code self.path = path class TaplineClient: def __init__(self, api_key: str): self.session = requests.Session() self.session.headers["X-API-Key"] = api_key self.last_request_at = 0.0 self.calls = Counter() self.credits = Counter() def get(self, path: str, credit_key: str, **params): elapsed = time.monotonic() - self.last_request_at if elapsed < REQUEST_INTERVAL_SECONDS: time.sleep(REQUEST_INTERVAL_SECONDS - elapsed) response = self.session.get( f"{API_ORIGIN}{path}", params=params, timeout=90, ) self.last_request_at = time.monotonic() self.calls[credit_key] += 1 if not response.ok: try: message = response.json().get("message", response.text[:200]) except requests.exceptions.JSONDecodeError: message = response.text[:200] raise TaplineApiError(response.status_code, path, message) self.credits[credit_key] += CREDIT_COSTS[credit_key] return response.json() @dataclass class Lead: feature_id: str name: str phone: str website: str rating: float | None review_count: int | None is_claimed: bool categories: str city: str postal_code: str qualification_gaps: str area_median_review_count: int recent_reviews_checked: int owner_response_count: int complaint_topics: str complaint_excerpt: str linkedin_company_status: str linkedin_company_url: str linkedin_employee_count: int | None linkedin_company_website: str linkedin_person_status: str linkedin_person_name: str linkedin_person_headline: str linkedin_person_url: str score: int evidence_summary: str captured_at_utc: str def normalized_name(value: str | None) -> str: return re.sub(r"[^a-z0-9]", "", (value or "").lower()) def name_words(value: str | None) -> set[str]: return { word for word in re.findall(r"[a-z0-9]+", (value or "").lower()) if len(word) > 2 and word not in GENERIC_NAME_WORDS } def domain(value: str | None) -> str: if not value: return "" parsed = urlparse(value if "://" in value else f"https://{value}") return parsed.netloc.lower().split(":", 1)[0].removeprefix("www.") def same_domain(left: str | None, right: str | None) -> bool: left_domain = domain(left) right_domain = domain(right) if not left_domain or not right_domain: return False return ( left_domain == right_domain or left_domain.endswith(f".{right_domain}") or right_domain.endswith(f".{left_domain}") ) def website_url(place: dict) -> str: return (place.get("website") or {}).get("url") or "" def phone_number(place: dict) -> str: phone = place.get("phone") or {} return phone.get("international") or phone.get("number") or "" def grid_points(latitude: float, longitude: float, grid_size: int): half = grid_size // 2 for row in range(-half, half + 1): for column in range(-half, half + 1): yield ( latitude + row * GRID_LATITUDE_STEP, longitude + column * GRID_LONGITUDE_STEP, ) def fetch_places( client: TaplineClient, query: str, latitude: float, longitude: float, grid_size: int, max_pages_per_tile: int, max_businesses: int, ) -> list[dict]: places_by_id = {} for point_latitude, point_longitude in grid_points(latitude, longitude, grid_size): cursor = "" for _ in range(max_pages_per_tile): page = client.get( "/v1/google-maps/search", "maps_search", query=query, latitude=point_latitude, longitude=point_longitude, zoom=MAPS_ZOOM, limit=MAPS_PAGE_SIZE, cursor=cursor, gl="us", hl="en", ) for place in page["results"]: feature_id = place.get("feature_id") if feature_id: places_by_id[feature_id] = place if len(places_by_id) >= max_businesses: return list(places_by_id.values()) cursor = page["pagination"].get("next_cursor") or "" if not cursor: break return list(places_by_id.values()) def initial_gaps(place: dict, review_median: float) -> list[str]: gaps = [] review_count = place.get("review_count") rating = place.get("rating") if not website_url(place): gaps.append("no website") if place.get("is_claimed") is False: gaps.append("unclaimed profile") if review_count is not None and review_median and review_count < review_median * LOW_REVIEW_RATIO: gaps.append("review count under half the area median") if rating is not None and rating < LOW_RATING: gaps.append("rating under 4.0") return gaps def review_evidence(client: TaplineClient, place: dict) -> dict: try: page = client.get( f"/v1/google-maps/places/{place['feature_id']}/reviews", "maps_reviews", sort="newest", limit=REVIEWS_PER_BUSINESS, gl="us", hl="en", ) except TaplineApiError as error: return { "checked": 0, "owner_responses": 0, "topics": [], "excerpt": "", "status": f"reviews error: HTTP {error.status_code}", } topics = [] excerpt = "" owner_responses = 0 for review in page["reviews"]: if review.get("owner_response"): owner_responses += 1 text = review.get("text") or review.get("preview_text") or "" matched_topics = [] if review.get("rating") is not None and review["rating"] <= 3: matched_topics = [ topic for topic, pattern in COMPLAINT_PATTERNS.items() if pattern.search(text) ] if matched_topics: topics.extend(matched_topics) if not excerpt: excerpt = re.sub(r"\s+", " ", text).strip()[:180] return { "checked": len(page["reviews"]), "owner_responses": owner_responses, "topics": sorted(set(topics)), "excerpt": excerpt, "status": "checked", } def brave_results(client: TaplineClient, query: str) -> tuple[list[dict], str]: try: page = client.get( "/v1/brave-search/search", "brave_search", query=query, country="us", ) except TaplineApiError as error: return [], f"search error: HTTP {error.status_code}" return page["results"], "searched" def linkedin_urls(results: list[dict], path: str) -> list[dict]: needle = f"linkedin.com/{path}/" return [result for result in results if needle in result.get("url", "").lower()] def best_company_result(results: list[dict], business_name: str) -> dict | None: words = name_words(business_name) scored = [] for result in linkedin_urls(results, "company"): text = f"{result.get('title', '')} {result.get('description', '')}".lower() score = sum(word in text for word in words) if normalized_name(business_name) in normalized_name(result.get("title")): score += 3 scored.append((score, -result.get("position", 999), result)) if not scored: return None return max(scored, key=lambda item: (item[0], item[1]))[2] def linkedin_company_evidence(client: TaplineClient, place: dict) -> dict: business_name = place["name"] city = (place.get("location") or {}).get("city") or "" results, search_status = brave_results( client, f'site:linkedin.com/company "{business_name}" {city}'.strip(), ) candidate = best_company_result(results, business_name) if not candidate: return { "status": f"{search_status}; no company result", "url": "", "employee_count": None, "website": "", "employees": [], } try: company = client.get( "/v1/linkedin/company", "linkedin_company", url=candidate["url"], ) except TaplineApiError as error: return { "status": f"company error: HTTP {error.status_code}", "url": candidate["url"], "employee_count": None, "website": "", "employees": [], } maps_website = website_url(place) company_website = company.get("website") or "" if not maps_website: status = "not trusted: Maps has no website to compare" elif not same_domain(maps_website, company_website): status = "not trusted: website mismatch" else: status = "trusted: website match" return { "status": status, "url": company.get("url") or candidate["url"], "employee_count": company.get("employeeCount"), "website": company_website, "employees": company.get("employees") or [], } def best_profile_result( results: list[dict], business_name: str, listed_employees: list[dict], ) -> dict | None: words = name_words(business_name) employee_names = [normalized_name(employee.get("name")) for employee in listed_employees] scored = [] for result in linkedin_urls(results, "in"): text = f"{result.get('title', '')} {result.get('description', '')}".lower() compact_title = normalized_name(result.get("title")) score = 3 if ROLE_PATTERN.search(text) else 0 score += sum(word in text for word in words) if any(name and name in compact_title for name in employee_names): score += 2 scored.append((score, -result.get("position", 999), result)) if not scored: return None score, _, result = max(scored, key=lambda item: (item[0], item[1])) return result if score >= 4 else None def profile_has_business(profile: dict, business_name: str) -> bool: words = name_words(business_name) if not words: return False company_names = [organization.get("name", "").lower() for organization in profile.get("experience", [])] return any(sum(word in name for word in words) >= min(2, len(words)) for name in company_names) def linkedin_person_evidence( client: TaplineClient, place: dict, company_evidence: dict, ) -> dict: business_name = place["name"] results, search_status = brave_results( client, f'site:www.linkedin.com/in "{business_name}" owner OR "practice manager"', ) candidate = best_profile_result(results, business_name, company_evidence["employees"]) if not candidate: return {"status": f"{search_status}; no reliable person result", "name": "", "headline": "", "url": ""} if company_evidence["status"] != "trusted: website match": return { "status": "not fetched: company website was not verified", "name": "", "headline": "", "url": candidate["url"], } try: profile = client.get( "/v1/linkedin/profile", "linkedin_profile", url=candidate["url"], ) except TaplineApiError as error: return { "status": f"profile error: HTTP {error.status_code}", "name": "", "headline": "", "url": candidate["url"], } if not profile_has_business(profile, business_name): status = "not trusted: experience did not match the business" else: status = "trusted: experience matched the business" return { "status": status, "name": profile.get("name") or "", "headline": profile.get("headline") or "", "url": profile.get("url") or candidate["url"], } def build_evidence_summary( place: dict, gaps: list[str], review_median: int, reviews: dict, company: dict, person: dict, ) -> str: facts = [] if "no website" in gaps: facts.append("Maps: no website listed") if "unclaimed profile" in gaps: facts.append("Maps: profile unclaimed") if "review count under half the area median" in gaps: facts.append( f"Maps: {place['review_count']} reviews; area median: {review_median}" ) if "rating under 4.0" in gaps: facts.append(f"Maps: {place['rating']} rating") if reviews["topics"]: facts.append(f"Recent low-rated review topics: {', '.join(reviews['topics'])}") if reviews["checked"]: facts.append( f"Owner responses: {reviews['owner_responses']} across " f"{reviews['checked']} newest reviews checked" ) if company["status"] == "trusted: website match": facts.append("LinkedIn company: website matched Maps") if person["status"] == "trusted: experience matched the business": facts.append("LinkedIn person: experience matched the business") return " | ".join(facts) def score_lead( gaps: list[str], reviews: dict, company: dict, person: dict, ) -> int: score = 2 * len(gaps) if reviews["checked"] and reviews["owner_responses"] == 0: score += 1 if reviews["topics"]: score += 3 if company["status"] == "trusted: website match": score += 1 if person["status"] == "trusted: experience matched the business": score += 1 return score def write_csv(rows: list[Lead], output_path: Path) -> None: with output_path.open("w", newline="", encoding="utf-8") as output: writer = csv.DictWriter(output, fieldnames=list(Lead.__dataclass_fields__)) writer.writeheader() writer.writerows(asdict(row) for row in rows) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("query", help='Business category, such as "dentist"') parser.add_argument("city", help='Target location, such as "Austin"') parser.add_argument("latitude", type=float, help="Target location center latitude") parser.add_argument("longitude", type=float, help="Target location center longitude") parser.add_argument("--output", type=Path, default=Path("leads.csv")) parser.add_argument("--grid-size", type=int, choices=(1, 3, 5), default=5) parser.add_argument("--max-pages-per-tile", type=int, default=3) parser.add_argument("--max-businesses", type=int, default=1000) parser.add_argument("--max-shortlist", type=int, default=200) args = parser.parse_args() if args.max_pages_per_tile < 1 or args.max_businesses < 1 or args.max_shortlist < 1: parser.error("page and business limits must be positive") return args def main() -> int: args = parse_args() client = TaplineClient(os.environ["TAPLINE_API_KEY"]) captured_at = datetime.now(timezone.utc).replace(microsecond=0).isoformat() places = fetch_places( client, args.query, args.latitude, args.longitude, args.grid_size, args.max_pages_per_tile, args.max_businesses, ) counts = [place["review_count"] for place in places if place.get("review_count") is not None] review_median = int(median(counts)) if counts else 0 candidates = [] for place in places: gaps = initial_gaps(place, review_median) if gaps and place.get("name") and place.get("feature_id"): candidates.append((place, gaps)) candidates.sort( key=lambda item: ( -len(item[1]), item[0].get("review_count") if item[0].get("review_count") is not None else -1, ) ) leads = [] for place, gaps in candidates[: args.max_shortlist]: reviews = review_evidence(client, place) company = linkedin_company_evidence(client, place) person = linkedin_person_evidence(client, place, company) location = place.get("location") or {} lead = Lead( feature_id=place["feature_id"], name=place["name"], phone=phone_number(place), website=website_url(place), rating=place.get("rating"), review_count=place.get("review_count"), is_claimed=place.get("is_claimed") is True, categories=" | ".join(place.get("categories") or []), city=location.get("city") or args.city, postal_code=location.get("postal_code") or "", qualification_gaps=" | ".join(gaps), area_median_review_count=review_median, recent_reviews_checked=reviews["checked"], owner_response_count=reviews["owner_responses"], complaint_topics=" | ".join(reviews["topics"]), complaint_excerpt=reviews["excerpt"], linkedin_company_status=company["status"], linkedin_company_url=company["url"], linkedin_employee_count=company["employee_count"], linkedin_company_website=company["website"], linkedin_person_status=person["status"], linkedin_person_name=person["name"], linkedin_person_headline=person["headline"], linkedin_person_url=person["url"], score=score_lead(gaps, reviews, company, person), evidence_summary=build_evidence_summary( place, gaps, review_median, reviews, company, person, ), captured_at_utc=captured_at, ) leads.append(lead) print(f"qualified lead {len(leads)}: score={lead.score}") leads.sort(key=lambda lead: (-lead.score, lead.name.lower())) write_csv(leads, args.output) print(f"wrote {len(leads)} leads from {len(places)} unique businesses to {args.output}") print(f"requests: {sum(client.calls.values())} {dict(client.calls)}") print(f"documented full-page credits: {sum(client.credits.values())} {dict(client.credits)}") return 0 if __name__ == "__main__": raise SystemExit(main())